workflow

type Context ¶

type Context struct {
	context.Context

	SessionID    string
	InvocationID string
	State        map[string]any
}

Context is passed to workflow nodes during graph execution.

type Edge ¶

type Edge struct {
	From  Node
	To    Node
	Route Route
}

Edge connects two workflow nodes. A nil Route is an unconditional edge.

func Chain(nodes ...Node) []Edge

func Concat(groups ...[]Edge) []Edge

type EdgeBuilder ¶

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

EdgeBuilder provides a fluent way to assemble graph edges.

func NewEdgeBuilder() *EdgeBuilder

func (b *EdgeBuilder) Add(from, to Node) *EdgeBuilder

func (b *EdgeBuilder) AddFanIn(sources []Node, target Node) *EdgeBuilder

func (b *EdgeBuilder) AddFanOut(from Node, targets ...Node) *EdgeBuilder

func (b *EdgeBuilder) AddRoute(from, to Node, route Route) *EdgeBuilder

func (b *EdgeBuilder) Build() []Edge

type EmitFunc ¶

type EmitFunc func(*Event) error

EmitFunc lets a node publish one or more explicit workflow events.

type Event ¶

type Event struct {
	Output   any
	Message  string
	Routes   []any
	Metadata map[string]string
}

Event carries node output and optional route selections to successor edges.

type FileRunStore ¶ added in v1.13.5

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

FileRunStore stores each workflow run as an atomically replaced JSON file. It is useful for local deployments and can be replaced with a database-backed RunStore in production.

func NewFileRunStore(dir string) (*FileRunStore, error)

func (s *FileRunStore) Create(ctx context.Context, state *RunState) error

func (s *FileRunStore) Delete(ctx context.Context, runID string) error

func (s *FileRunStore) Load(ctx context.Context, runID string) (*RunState, error)

func (s *FileRunStore) Save(ctx context.Context, state *RunState) error

type Graph ¶

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

Graph executes workflow nodes according to explicit edges and routes.

func NewGraph(edges []Edge, config GraphConfig) (*Graph, error)

func (g *Graph) Edges() []Edge

func (g *Graph) ResumeRun(ctx context.Context, store RunStore, runID string) (any, error)

func (g *Graph) Run(ctx context.Context, sessionID string, input any) (any, error)

func (g *Graph) StartRun(ctx context.Context, store RunStore, runID, sessionID string, input any) (any, error)

type GraphConfig ¶

type GraphConfig struct {
	MaxSteps    int
	JoinTimeout time.Duration
}

GraphConfig configures workflow graph execution.

type InMemoryRunStore ¶ added in v1.13.5

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

InMemoryRunStore is a JSON-round-tripping RunStore suited to tests and single-process development. It enforces the same serialization requirements as FileRunStore so that a run which works in tests can be persisted later.

func NewInMemoryRunStore() *InMemoryRunStore

func (s *InMemoryRunStore) Create(ctx context.Context, state *RunState) error

func (s *InMemoryRunStore) Delete(ctx context.Context, runID string) error

func (s *InMemoryRunStore) Load(ctx context.Context, runID string) (*RunState, error)

func (s *InMemoryRunStore) Save(ctx context.Context, state *RunState) error

type JoinFunc ¶ added in v1.13.2

type JoinFunc func(Context, map[string]any) (any, error)

JoinFunc combines the outputs of every direct predecessor of a JoinNode. The map is keyed by the predecessor node name.

type JoinState ¶ added in v1.13.5

type JoinState struct {
	Values       map[string]any `json:"values"`
	FirstArrival time.Time      `json:"first_arrival"`
}

JoinState holds the accumulated predecessor outputs for a join node.

type Node ¶

type Node interface {
	Name() string
	Description() string
	// contains filtered or unexported methods
}

Node is an executable step in a workflow graph.

var Start Node = startNode{}

func NewAgentNode(name string, runner SessionGenerator, config NodeConfig) Node

func NewEmittingFunctionNode[I, O any](name string, fn func(Context, I, EmitFunc) (O, error), config NodeConfig) Node

func NewFunctionNode[I, O any](name string, fn func(Context, I) (O, error), config NodeConfig) Node

func NewJoinNode(name string, fn JoinFunc, config NodeConfig) Node

func NewToolNode(name string, tool goagent.Tool, config NodeConfig) Node

type NodeConfig ¶

type NodeConfig struct {
	Description string
}

NodeConfig carries metadata for workflow nodes.

type QueuedNode ¶ added in v1.13.5

type QueuedNode struct {
	From  string `json:"from"`
	Node  string `json:"node"`
	Input any    `json:"input"`
}

QueuedNode is a node invocation waiting to be executed as part of a durable workflow run. Node names are persisted instead of Node implementations so a run can be resumed after a process restart.

type Route ¶

type Route interface {
	Match(Event) bool
	// contains filtered or unexported methods
}

Route selects successor edges based on emitted Event.Routes.

var Default Route = routeFunc{
	// contains filtered or unexported fields
}

func BoolRoute(value bool) Route

func IntRoute(value int) Route

func MultiRoute[T comparable](values ...T) Route

func StringRoute(value string) Route

type RunState ¶ added in v1.13.5

type RunState struct {
	ID           string               `json:"id"`
	SessionID    string               `json:"session_id"`
	InvocationID string               `json:"invocation_id"`
	Status       RunStatus            `json:"status"`
	Queue        []QueuedNode         `json:"queue"`
	Finals       []any                `json:"finals,omitempty"`
	JoinStates   map[string]JoinState `json:"join_states,omitempty"`
	ContextState map[string]any       `json:"context_state,omitempty"`
	Steps        int                  `json:"steps"`
	Result       any                  `json:"result,omitempty"`
	LastError    string               `json:"last_error,omitempty"`
	CreatedAt    time.Time            `json:"created_at"`
	UpdatedAt    time.Time            `json:"updated_at"`
}

RunState is the serializable execution state for a durable workflow run.

Inputs, terminal outputs, join values, and ContextState must be JSON marshalable when using the supplied stores. Runs have at-least-once node execution semantics: if a process stops after a node runs but before its transition is saved, resuming the run invokes that node again. Side-effecting nodes should therefore be idempotent.

type RunStatus ¶ added in v1.13.5

type RunStatus string

RunStatus describes the lifecycle state of a durable workflow run.

const (
	// RunStatusRunning identifies a run with pending work. A run remains running
	// after a node error so it can be resumed.
	RunStatusRunning RunStatus = "running"
	// RunStatusCompleted identifies a run that reached a terminal graph output.
	RunStatusCompleted RunStatus = "completed"
	// RunStatusFailed identifies a run that cannot make progress, such as an
	// incomplete join or an invalid persisted node reference.
	RunStatusFailed RunStatus = "failed"
)

type RunStore ¶ added in v1.13.5

type RunStore interface {
	Create(ctx context.Context, state *RunState) error
	Load(ctx context.Context, runID string) (*RunState, error)
	Save(ctx context.Context, state *RunState) error
	Delete(ctx context.Context, runID string) error
}

RunStore persists workflow execution state. Implementations must return an isolated snapshot from Load and retain an isolated snapshot passed to Create or Save, because workflow execution mutates RunState between checkpoints.

type SessionGenerator ¶

type SessionGenerator interface {
	Generate(ctx context.Context, sessionID, input string) (any, error)
}

SessionGenerator is the minimal surface needed to wrap an agent node.