Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

engine

package

Go to main page

Versions in this module

v1

v1.3.1

Jun 11, 2026

v1.3.0

Jun 10, 2026

v1.2.0

May 30, 2026

v1.1.6

May 26, 2026

Changes in this version

+ const DefaultCancelTTL

+ const DefaultQueueSize

+ var ErrChannelFull = errors.New("engine: channel full (use Emit for blocking semantics)")

+ var ErrConcurrentFetchByRun = errors.New("engine: only one FetchByRun in flight per run")

+ var ErrCycleDetected = errors.New("engine: cycle detected without AllowCycle")

+ var ErrDeadlineExceeded = errors.New("engine: envelope deadline exceeded")

+ var ErrDuplicateNodeName = errors.New("engine: duplicate node name")

+ var ErrEmptyAdjacencies = errors.New("engine: adjacencies must be non-empty")

+ var ErrEmptyRunID = errors.New("engine: EmitChunk requires a non-empty RunID on the originating envelope")

+ var ErrEngineStopped = errors.New("engine: stopped")

+ var ErrIdentityRequired = errors.New("engine: Emit requires non-empty identity triple")

+ var ErrInvalidQueueSize = errors.New("engine: queue size must be > 0")

+ var ErrNodeNotFound = errors.New("engine: node not found")

+ var ErrNotImplemented = errors.New("engine: not implemented in this phase")

+ var ErrRunCancelled = errors.New("engine: run cancelled")

+ var ErrSeqProvided = errors.New("engine: caller pre-filled StreamFrame.Seq; engine owns sequencing")

+ var ErrStreamClosed = errors.New("engine: stream closed (Done frame already drained)")

+ var ErrSubflowFactoryFailed = errors.New("engine: subflow factory failed")

+ type Adjacency struct

+ From Node

+ To []Node

+ type EmitOption func(*emitOptions)

+ func WithRunCapacity(n int) EmitOption

+ type Engine interface

+ Cancel func(ctx context.Context, runID string) (bool, error)

+ Emit func(ctx context.Context, env messages.Envelope, opts ...EmitOption) error

+ EmitTo func(ctx context.Context, env messages.Envelope, target NodeRef) error

+ Fetch func(ctx context.Context, opts ...FetchOption) (messages.Envelope, error)

+ FetchByRun func(ctx context.Context, runID string) (messages.Envelope, error)

+ Run func(ctx context.Context) error

+ Stop func(ctx context.Context) error

+ Topology func(ctx context.Context) (types.TopologyProjection, error)

+ func New(adjacencies []Adjacency, opts ...Option) (Engine, error)

+ type FetchOption func(*fetchOptions)

+ type Node struct

+ AllowCycle bool

+ Func NodeFunc

+ Name string

+ Policy NodePolicy

+ func (n Node) Ref() NodeRef

+ type NodeContext struct

+ func (nctx *NodeContext) CallSubflow(ctx context.Context, factory SubflowFactory, parentEnv messages.Envelope) (messages.Envelope, error)

+ func (nctx *NodeContext) Emit(ctx context.Context, env messages.Envelope) error

+ func (nctx *NodeContext) EmitChunk(ctx context.Context, frame StreamFrame) error

+ func (nctx *NodeContext) EmitNoWait(ctx context.Context, env messages.Envelope) error

+ type NodeFunc func(ctx context.Context, in messages.Envelope, nctx *NodeContext) (messages.Envelope, error)

+ type NodePolicy struct

+ BackoffBase time.Duration

+ BackoffMult float64

+ MaxBackoff time.Duration

+ MaxRetries int

+ RunCapacity int

+ TimeoutMS int

+ Validate ValidateMode

+ ValidateFunc func(messages.Envelope) error

+ type NodeRef struct

+ Name string

+ type Option func(*engineConfig)

+ func WithCancelTTL(d time.Duration) Option

+ func WithChannelOverride(from, to NodeRef, n int) Option

+ func WithErrorEmissionToEgress(enabled bool) Option

+ func WithEventBus(b events.EventBus) Option

+ func WithQueueSize(n int) Option

+ func WithRunCancelledHandler(h RunCancelledHandler) Option

+ func WithRunErrorHandler(h RunErrorHandler) Option

+ type RunCancelledHandler func(ctx context.Context, n RunCancelledNotice)

+ type RunCancelledNotice struct

+ CancelledAt time.Time

+ DroppedEnvelopeCount int64

+ RunID string

+ type RunError struct

+ Cause error

+ Code RunErrorCode

+ Message string

+ Metadata map[string]any

+ NodeID string

+ NodeName string

+ RunID string

+ SessionID string

+ TenantID string

+ UserID string

+ func (e *RunError) Error() string

+ func (e *RunError) ToPayload() map[string]any

+ func (e *RunError) Unwrap() error

+ type RunErrorCode string

+ const CodeDeadlineExceeded

+ const CodeNodeException

+ const CodeNodeTimeout

+ const CodeRunCancelled

+ const CodeValidationFailed

+ type RunErrorHandler func(ctx context.Context, re *RunError)

+ type StreamFrame struct

+ Done bool

+ Meta map[string]any

+ Seq int

+ StreamID string

+ Text string

+ type SubflowFactory func() (Engine, error)

+ type ValidateMode string

+ const ValidateBoth

+ const ValidateIn

+ const ValidateNone

+ const ValidateOut

v1.1.5

May 26, 2026

v1.1.0

May 25, 2026

v1.0.0

May 22, 2026