workflow

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 1, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package workflow is a graph-execution engine for multi-agent orchestration.

A Builder wires ExecutorBindings into a graph — sequential, concurrent, conditional, fan-out/fan-in, and subworkflows — and execution advances via a TurnToken. Per-executor private scopes and named shared scopes hold state and underpin checkpointing. This engine is distinct from single-agent runs in the agent package.

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func PortableValueAs 

func PortableValueAs[T any](v PortableValue) (T, bool)

PortableValueAs attempts to convert the supplied PortableValue to the requested type T.

Types

type AddHandlerOption 

type AddHandlerOption func(*addHandlerOptions)

AddHandlerOption configures handler registration for RouteBuilder.AddHandlerRaw and RouteBuilder.AddCatchAll.

func WithHandlerOverwrite 

func WithHandlerOverwrite(overwrite bool) AddHandlerOption

WithHandlerOverwrite controls whether handler registration replaces an existing handler.

For RouteBuilder.AddHandlerRaw, overwrite applies to the handler registered for the same message type. For RouteBuilder.AddCatchAll, overwrite applies to the single catch-all handler.

type AttrSendsMessage 

type AttrSendsMessage[T any] struct {
	// contains filtered or unexported fields
}

AttrSendsMessage marks a struct-based executor as declaring an additional sent message type T. Use it as a field on a struct passed to NewExecutor. The field name is not significant; use _ when the field is only a marker.

type AttrYieldsOutput 

type AttrYieldsOutput[T any] struct {
	// contains filtered or unexported fields
}

AttrYieldsOutput marks a struct-based executor as declaring an additional workflow output type T. Use it as a field on a struct passed to NewExecutor. The field name is not significant; use _ when the field is only a marker.

type Builder 

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

Builder assembles a Workflow graph from executors, edges, and request/response ports, then produces an immutable Workflow via Build.

func NewBuilder 

func NewBuilder(start ExecutorBinding) *Builder

NewBuilder returns a Builder rooted at the given start executor binding, which becomes the workflow entry point.

func (*Builder) AddChain 

func (wb *Builder) AddChain(source ExecutorBinding, executors []ExecutorBinding, allowRepetition bool) *Builder

AddChain connects source to each binding in executors in order, producing a linear pipeline source → executors[0] → executors[1] → ...

If allowRepetition is false, the same binding may not appear twice in the chain (including the source). Adding the same edge twice is idempotent.

func (*Builder) AddEdge 

func (wb *Builder) AddEdge(source ExecutorBinding, target ExecutorBinding, opts ...EdgeOption) *Builder

AddEdge adds a direct edge from source to target. Pass WithEdgeCondition to make the edge conditional. Adding a duplicate conditionless edge records an error unless IdempotentEdge is supplied. Only conditionless edges participate in the duplicate-edge check.

func (*Builder) AddFanInBarrierEdge 

func (wb *Builder) AddFanInBarrierEdge(sources []ExecutorBinding, target ExecutorBinding, opts ...EdgeOption) *Builder

AddFanInBarrierEdge adds a fan-in edge from sources to target, waiting for all sources before dispatching to the target.

func (*Builder) AddFanOutEdge 

func (wb *Builder) AddFanOutEdge(source ExecutorBinding, targets []ExecutorBinding, opts ...EdgeOption) *Builder

AddFanOutEdge adds a fan-out edge from source to one or more targets.

By default the message is delivered to every target. Pass WithEdgeAssigner to choose a subset of targets per message. See Edge.Assigner.

func (*Builder) AddSwitch 

func (wb *Builder) AddSwitch(source ExecutorBinding) *SwitchBuilder

AddSwitch starts building a switch-style fan-out from source. Configure cases via SwitchBuilder.AddCase and an optional default via SwitchBuilder.WithDefault, then commit by calling SwitchBuilder.AddToBuilder.

func (*Builder) BindExecutor 

func (wb *Builder) BindExecutor(binding ExecutorBinding) *Builder

BindExecutor registers an executor binding without adding any edge. It records an error if binding is a placeholder registration.

func (*Builder) Build 

func (wb *Builder) Build() (*Workflow, error)

Build validates the assembled graph, including orphan-executor checks, and returns the immutable *Workflow. It returns the first accumulated error if any builder step failed or validation did not pass.

func (*Builder) WithDescription 

func (wb *Builder) WithDescription(description string) *Builder

WithDescription sets the workflow's human-readable description. It is a no-op if the builder already holds an error.

func (*Builder) WithIntermediateOutputFrom 

func (wb *Builder) WithIntermediateOutputFrom(bindings ...ExecutorBinding) *Builder

WithIntermediateOutputFrom registers bindings as workflow output sources carrying OutputTagIntermediate.

func (*Builder) WithName 

func (wb *Builder) WithName(name string) *Builder

WithName sets the workflow's human-readable name. It is a no-op if the builder already holds an error.

func (*Builder) WithOutputFrom 

func (wb *Builder) WithOutputFrom(bindings ...ExecutorBinding) *Builder

WithOutputFrom registers the given bindings as terminal (untagged) workflow output sources. It mirrors Builder.WithIntermediateOutputFrom but without the OutputTagIntermediate tag.

func (*Builder) WithTelemetry 

func (wb *Builder) WithTelemetry(tracer workflowobservability.Tracer, options TelemetryOptions) *Builder

WithTelemetry enables telemetry instrumentation for workflows built by this builder.

type CatchAllFunc 

type CatchAllFunc func(*Context, PortableValue) (any, error)

CatchAllFunc handles messages that do not match a typed route.

The message is supplied as a PortableValue so catch-all handlers can forward or inspect messages even when their concrete Go type is not known to the current executor.

type CheckpointInfo 

type CheckpointInfo struct {
	// SessionID is the workflow session that owns the checkpoint.
	SessionID string `json:"sessionId"`

	// CheckpointID is unique within the session and identifies one stored
	// checkpoint record.
	CheckpointID string `json:"checkpointId"`
}

CheckpointInfo identifies a persisted workflow checkpoint.

Checkpoint managers use this value to store, retrieve, index, and resume checkpoints for a workflow session.

func NewCheckpointInfo 

func NewCheckpointInfo(sessionID string) CheckpointInfo

NewCheckpointInfo creates checkpoint metadata for sessionID.

The generated checkpoint ID is a UUID string. NewCheckpointInfo panics if sessionID is empty.

type Context 

type Context struct {
	// Context carries cancellation, deadlines, and request-scoped values for the
	// current executor invocation.
	context.Context

	// AddEvent adds an event to the workflow's output queue. These events will be raised to the caller of the workflow at the
	// end of the current [SuperStep].
	AddEvent func(event Event) error

	// SendMessage queues a message to be sent to connected executors. The message will be sent during the next [SuperStep].
	// targetID is an optional identifier of the target executor. If empty, the message is sent to all connected
	// executors. If the target executor is not connected from this executor via an edge, it will still not receive the
	// message
	SendMessage func(targetID string, message any) error

	// YieldOutput adds an output value to the workflow's output queue.
	// These outputs will be bubbled out of the workflow using the [SuperStep].
	//
	// The type of the output message must match one of the output types declared by the [Executor]. By default, the return
	// types of registered message handlers are considered output types, unless otherwise specified using [Executor].
	YieldOutput func(output any) error

	// RequestHalt adds a request to "halt" workflow execution at the end of the current [SuperStep].
	RequestHalt func() error

	// PostRequest raises an [ExternalRequest] from the current executor.
	// The request becomes a [RequestInfoEvent] in the workflow event stream,
	// and the matching [ExternalResponse] (sent later by the caller via the
	// run handle) is delivered back to this executor as a regular message of
	// type *[ExternalResponse]. The executor is responsible for registering
	// a handler for *[ExternalResponse] via its [RouteBuilder] and extracting
	// the typed payload via [ExternalResponse.Data] and [PortableValue.As].
	PostRequest func(request *ExternalRequest) error

	// ReadState reads a state value from the workflow's state store. If no scope is provided, the executor's
	// default scope is used.
	ReadState func(key string, scope string) (any, error)

	// ReadOrInitState reads or initializes a state value from the workflow's state store.
	// If no scope is provided, the executor's default scope is used.
	ReadOrInitState func(key string, scope string, initFunc func(ctx context.Context, key string, scope string) (any, error)) (any, error)

	// ReadStateKeys reads all state keys within the specified scope.
	// If no scope is provided, the executor's default scope is used.
	ReadStateKeys func(scope string) iter.Seq2[string, error]

	// QueueStateUpdate updates the state of a queue entry identified by the specified key and optional scope.
	// If no scope is provided, the executor's default scope is used.
	QueueStateUpdate func(key string, scope string, value any) error

	// QueueClearScope clears all state entries within the specified scope.
	// If no scope is provided, the executor's default scope is used.
	QueueClearScope func(scope string) error

	// TraceContext returns the trace context associated with the current message about to be processed by the executor, if any.
	TraceContext func() map[string]any

	// ConcurrentRunsEnabled reports whether the current execution environment
	// supports concurrent runs against the same workflow instance.
	ConcurrentRunsEnabled bool
}

Context provides services for an Executor during the execution of a workflow.

type Edge 

type Edge struct {
	Index int

	Connection EdgeConnection

	// Label is an optional label for the edge, allowing arbitrary metadata
	// to be associated with it (e.g. for visualization purposes).
	Label string

	// Assigner is an optional function that maps an incoming message to a subset of the target
	// executor nodes (or optionally all of them).
	// If nil, all destination nodes are selected.
	// Only used for fan out edges.
	Assigner func(int, any) iter.Seq[int]

	// Condition is an optional predicate determining whether the edge is active for a given message.
	// If nil, the edge is always active when a message is generated by the source.
	// Only used for direct edges.
	Condition func(any) bool
}

Edge represents a connection or relationship between nodes.

func (Edge) Equal 

func (e Edge) Equal(other Edge) bool

Equal reports whether this Edge is equal to another Edge without considering the Index field.

type EdgeConnection 

type EdgeConnection struct {
	SourceIDs []string
	SinkIDs   []string
}

EdgeConnection is a representation for the connection structure of an edge of any multiplicity, defined by an ordered list of sources and sinks connected by this edge.

func (EdgeConnection) Equal 

func (c EdgeConnection) Equal(other EdgeConnection) bool

Equal reports whether c and other connect the same ordered SourceIDs and SinkIDs.

type EdgeInfo 

type EdgeInfo struct {
	// Connection describes the edge endpoints and connection shape.
	Connection EdgeConnection

	// Label is the optional label associated with the edge.
	Label string

	// HasCondition reports whether the edge has a condition callback.
	HasCondition bool

	// HasAssigner reports whether the edge has a fan-out assigner callback.
	HasAssigner bool
}

EdgeInfo is a serializable description of a workflow edge.

It records the edge connection and metadata that can be reflected or serialized, while representing condition and assigner callbacks only by their presence.

func (*EdgeInfo) Match 

func (e *EdgeInfo) Match(other Edge) bool

Match reports whether other has the same reflected edge metadata.

Callback functions are compared by presence only; the function values themselves are not comparable and are not represented in EdgeInfo.

type EdgeOption 

type EdgeOption func(*edgeOptions)

EdgeOption configures an Edge when it is added to a workflow via Builder.

func IdempotentEdge 

func IdempotentEdge() EdgeOption

IdempotentEdge makes adding a duplicate conditionless direct edge a no-op instead of a build error. It has no effect on conditional, fan-out, or fan-in edges.

func WithEdgeAssigner 

func WithEdgeAssigner[T any](assigner func(targetCount int, message T) iter.Seq[int]) EdgeOption

WithEdgeAssigner attaches an Edge.Assigner callback to a fan-out edge. The assigner is invoked for each message and must return the indexes of the targets that should receive it. PortableValue messages are decoded as T before the assigner is invoked. Messages that cannot be assigned to T are passed as the zero value of T. When T is any, the original message is passed unchanged. Has no effect on direct or fan-in edges.

func WithEdgeCondition 

func WithEdgeCondition[T any](condition func(T) bool) EdgeOption

WithEdgeCondition attaches a condition that receives messages as T to a direct edge. PortableValue messages are decoded as T before the condition is invoked. Messages that cannot be assigned to T are passed as the zero value of T. When T is any, the original message is passed unchanged.

func WithEdgeCondition0 

func WithEdgeCondition0(condition func() bool) EdgeOption

WithEdgeCondition0 attaches a message-independent condition to a direct edge. The condition is invoked once for each incoming message.

func WithEdgeLabel 

func WithEdgeLabel(label string) EdgeOption

WithEdgeLabel sets an optional label on the edge. Labels can be used by visualizers to annotate edges.

type ErrorEvent 

type ErrorEvent struct {
	Error error

	// Optional SubWorkflowID indicates the sub-workflow where the error occurred.
	SubWorkflowID string
}

ErrorEvent is an event triggered when an error occurs in the workflow.

func (ErrorEvent) Data 

func (e ErrorEvent) Data() any

type Event 

type Event interface {
	Data() any
}

Event is implemented by every workflow run event. Data returns the event payload.

type Executor 

type Executor struct {
	// ID is the executor's workflow-unique identifier.
	ID string

	// ImplementationID identifies the implementation or semantic source for this executor.
	ImplementationID string

	// If true, the result of a message handler that returns a value will be sent
	// as a message from the executor. The default is true.
	AutoSendMessageHandlerResultObject *bool

	// If true, the result of a message handler that returns a value will be
	// yielded as workflow output from the executor. The default is true.
	AutoYieldOutputHandlerResultObject *bool

	// CrossRunShareable reports whether this executor instance can be shared
	// safely by concurrent workflow runs. [Executor.Bind] copies this value to
	// [ExecutorBinding.SupportsConcurrentSharedExecution].
	CrossRunShareable bool

	// ConfigureProtocol configures message handlers and declared send/yield
	// message types on the supplied builder.
	// It may return the same builder or a replacement builder.
	ConfigureProtocol func(builder *ProtocolBuilder) (*ProtocolBuilder, error)

	// Initialize is called when the executor instance is created for a run.
	InitializeFunc func(ctx *Context) error

	// AttachRuntimeFunc is called by an execution environment to attach
	// runtime-specific capabilities to this executor instance before Initialize.
	AttachRuntimeFunc func(runtime any) error

	// Reset clears any executor-local cached state before an instance is reused.
	ResetFunc func() error

	// Close releases executor-local resources when a workflow run ends.
	CloseFunc func(ctx context.Context) error

	// OnCheckpoint is called before workflow state is checkpointed.
	OnCheckpointFunc func(ctx *Context) error

	// OnCheckpointRestored is called after workflow state is restored.
	OnCheckpointRestoredFunc func(ctx *Context) error

	// OnMessageDeliveryStarting is invoked once per superstep, before any
	// messages are delivered to the executor. It is given a context bound to
	// this executor with no per-message trace context.
	OnMessageDeliveryStartingFunc func(ctx *Context) error

	// OnMessageDeliveryFinished is invoked once per superstep, after all
	// messages have been delivered to the executor (regardless of whether
	// individual deliveries succeeded). It is given a context bound to this
	// executor with no per-message trace context.
	OnMessageDeliveryFinishedFunc func(ctx *Context) error
	// contains filtered or unexported fields
}

Executor is the runnable behavior for a workflow node. It owns the node's message protocol, lifecycle hooks, routing cache, and any executor-local state for one executor instance.

An Executor is not the graph registration used by builders. Use Executor.Bind or BindNewExecutorFunc to produce an ExecutorBinding. Bindings carry workflow identity and instance creation/reuse policy; runners call a binding to obtain an Executor for a workflow session.

The zero value has no behavior. An Executor must have at least one non-nil route or lifecycle callback before it can build routes. Use Executor.Extend to add behavior from reusable helpers.

Executors cache their route table after first use.

func NewAggregatingExecutor 

func NewAggregatingExecutor[TInput, TAggregate any](id string, aggregator func(*TAggregate, TInput) *TAggregate) *Executor

NewAggregatingExecutor creates an executor that incrementally aggregates input messages. Aggregate state is persisted in workflow checkpoints. The aggregator receives nil when no state is present; returning nil clears the state. Non-nil aggregates are sent and yielded normally.

func NewExecutor 

func NewExecutor(id string, v any) *Executor

NewExecutor converts v into an Executor. The id parameter is the workflow executor ID to assign or validate. If id is empty for an existing executor, the existing executor ID is used.

NewExecutor accepts these values:

  • *Executor, which is returned as an executor instance after checking that a non-empty id matches the executor's ID.
  • RequestPort or *RequestPort, which create the executor used by RequestPort.Bind.
  • ExecutorBinding, which is instantiated through the binding and must produce an executor with the binding ID. A non-empty id must match the binding ID.
  • Struct values or pointers to structs with a Handle method. Handle must be a valid function handler shape accepted by NewExecutor.
  • Function handlers must have exactly one typed non-context input parameter and may return zero or one typed output parameter.
  • A *Context input is optional; when present, it must be the first input parameter.
  • A final error output is treated as the handler error; a non-final error output is allowed only as the single ordinary output value.
  • Variadic functions are not supported; define an explicit slice or struct input type instead.
  • Function executors register the non-context input type as the accepted message type and the non-handler-error output type, when present, as an auto-sent and auto-yielded output type.
  • Named functions use their runtime function name as implementation identity; anonymous functions use id. Struct-based executors use their struct type name.
  • Struct-based executors may declare additional protocol types using AttrSendsMessage and AttrYieldsOutput fields. Field names are not significant; _ is recommended for marker-only fields.

NewExecutor panics for nil values, nil functions, unsupported function signatures, factory failures, or mismatched IDs.

Example (Context) 
package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/microsoft/agent-framework-go/workflow"
)

func main() {
	executor := workflow.NewExecutor("normalize", func(ctx *workflow.Context, input string) (string, error) {
		if err := ctx.Err(); err != nil {
			return "", err
		}
		return strings.ToLower(input), nil
	})

	result, err := executor.Execute(exampleContext(nil, nil), "HELLO")
	if err != nil {
		panic(err)
	}

	fmt.Println(result)

}

func exampleContext(sent *any, yielded *any) *workflow.Context {
	return &workflow.Context{
		Context: context.Background(),
		AddEvent: func(workflow.Event) error {
			return nil
		},
		SendMessage: func(_ string, message any) error {
			if sent != nil {
				*sent = message
			}
			return nil
		},
		YieldOutput: func(output any) error {
			if yielded != nil {
				*yielded = output
			}
			return nil
		},
	}
}

Output:
hello
Example (Function) 
package main

import (
	"context"
	"fmt"

	"github.com/microsoft/agent-framework-go/workflow"
)

func main() {
	executor := workflow.NewExecutor("length", func(input string) int {
		return len(input)
	})

	var sent any
	var yielded any
	result, err := executor.Execute(exampleContext(&sent, &yielded), "hello")
	if err != nil {
		panic(err)
	}

	fmt.Println(result)
	fmt.Println(sent)
	fmt.Println(yielded)

}

func exampleContext(sent *any, yielded *any) *workflow.Context {
	return &workflow.Context{
		Context: context.Background(),
		AddEvent: func(workflow.Event) error {
			return nil
		},
		SendMessage: func(_ string, message any) error {
			if sent != nil {
				*sent = message
			}
			return nil
		},
		YieldOutput: func(output any) error {
			if yielded != nil {
				*yielded = output
			}
			return nil
		},
	}
}

Output:
5
5
5
Example (Struct) 
package main

import (
	"fmt"
	"reflect"
	"slices"
	"strings"

	"github.com/microsoft/agent-framework-go/workflow"
)

func main() {
	executor := workflow.NewExecutor("reviewer", exampleReviewer{})
	descriptor := executor.DescribeProtocol()

	fmt.Println(slices.Contains(descriptor.Accepts, reflect.TypeFor[exampleDraft]()))
	fmt.Println(slices.Contains(descriptor.Sends, reflect.TypeFor[exampleReviewRequest]()))
	fmt.Println(slices.Contains(descriptor.Yields, reflect.TypeFor[exampleReviewReport]()))

}

type exampleDraft struct {
	Text string
}

type exampleReviewRequest struct {
	Text string
}

type exampleReviewReport struct {
	Approved bool
}

type exampleReviewer struct {
	_ workflow.AttrSendsMessage[exampleReviewRequest]
	_ workflow.AttrYieldsOutput[exampleReviewReport]
}

func (exampleReviewer) Handle(ctx *workflow.Context, draft exampleDraft) error {
	return ctx.SendMessage("", exampleReviewRequest{Text: strings.TrimSpace(draft.Text)})
}

Output:
true
true
true

func (*Executor) AttachRuntime 

func (e *Executor) AttachRuntime(runtime any) error

func (*Executor) Bind 

func (e *Executor) Bind() ExecutorBinding

Bind returns an ExecutorBinding for e.

Calling Bind multiple times on the same executor is fine: each returned binding has the same ID and implementation identity and returns the same executor instance. Bind does not clone e.

func (*Executor) Close 

func (e *Executor) Close(ctx context.Context) error

func (*Executor) DescribeProtocol 

func (e *Executor) DescribeProtocol() ProtocolDescriptor

DescribeProtocol returns the executor's accepted, sent, and yielded message types. It panics if the executor's protocol configuration is invalid.

func (*Executor) Execute 

func (e *Executor) Execute(ctx *Context, message any) (result any, err error)

func (*Executor) Extend 

func (e *Executor) Extend(executor *Executor) *Executor

Extend adds behavior from executor to e and returns e.

Existing protocol configuration and lifecycle hooks run before hooks from executor. Most hooks stop on the first error. OnMessageDeliveryFinished runs every hook and returns the first error encountered.

Runtime policy fields are combined conservatively: automatic send or yield remains enabled only when both executors enable it.

func (*Executor) Initialize 

func (e *Executor) Initialize(ctx *Context) error

func (*Executor) OnCheckpoint 

func (e *Executor) OnCheckpoint(ctx *Context) error

func (*Executor) OnCheckpointRestored 

func (e *Executor) OnCheckpointRestored(ctx *Context) error

func (*Executor) OnMessageDeliveryFinished 

func (e *Executor) OnMessageDeliveryFinished(ctx *Context) error

OnMessageDeliveryFinished invokes all configured OnMessageDeliveryFinished hooks. All hooks are run; the first non-nil error encountered is returned after all have been invoked.

func (*Executor) OnMessageDeliveryStarting 

func (e *Executor) OnMessageDeliveryStarting(ctx *Context) error

OnMessageDeliveryStarting invokes all configured OnMessageDeliveryStarting hooks. Returns the first error from any hook.

func (*Executor) Reset 

func (e *Executor) Reset() error

func (*Executor) SetCrossRunShareable 

func (e *Executor) SetCrossRunShareable(v bool) *Executor

SetCrossRunShareable sets whether e can be shared safely by concurrent workflow runs and returns e.

type ExecutorBinding 

type ExecutorBinding struct {
	// ID is the workflow-unique identifier for the executor. It is also the
	// address used by edges and messages inside the workflow graph.
	ID string

	// ImplementationID identifies the binding implementation or semantic executor source for
	// diagnostics, validation, and workflow metadata.
	ImplementationID string

	// RawValue optionally carries the comparable source value behind this binding.
	// When the builder sees another binding with the same [ID] and [ImplementationID],
	// it compares RawValue to catch accidental reuse of the [ID] for a different
	// source value. RawValue must be nil or comparable; the builder rejects
	// non-comparable values. Leave it nil for sources such as function values.
	RawValue any

	// SharedInstance reports whether [NewExecutorFunc] returns a shared executor
	// instance rather than creating an independent instance for each session.
	// Shared instances participate in workflow reset checks through [Reset]. Keep
	// this value consistent with [NewExecutorFunc]; setting it to false for a
	// shared executor opts out of reset checks.
	SharedInstance bool

	// SupportsConcurrentSharedExecution reports whether this binding may be used
	// by concurrent workflow runs. Bindings produced by [Executor.Bind] copy this
	// value from [Executor.CrossRunShareable]. Factory bindings set it directly.
	SupportsConcurrentSharedExecution bool

	// Ports lists [RequestPort]s that this binding exposes, either as the
	// workflow boundary port created by [RequestPort.Bind] or as additional ports
	// an executor uses to raise [ExternalRequest]s via [Context.PostRequest]. The
	// builder registers them so they appear in [Workflow.ReflectPorts] metadata.
	Ports []RequestPort

	// NewExecutorFunc creates the executor instance for a workflow session. The
	// returned executor must have the same ID as this binding; [CreateInstance]
	// validates that contract and records this binding's implementation ID on the
	// instance when it does not already have one.
	NewExecutorFunc func(sessionID string) (*Executor, error)

	// ResetFunc restores shared resources to their initial state. It is only called
	// for bindings marked as [SharedInstance]. A false return value means the
	// workflow could not be reset safely.
	ResetFunc func() bool
}

ExecutorBinding is the graph and session registration for an executor ID. It records the workflow address, implementation identity, and how a runner obtains an executable Executor instance for a workflow session.

An Executor contains executable behavior: routes, lifecycle hooks, and local state. An ExecutorBinding is the stable handle that builders store on graph edges and runners use to create or reuse an Executor. A binding may wrap a shared Executor instance, create a fresh one per session, or act as a placeholder while a graph is being built.

func BindNewExecutorFunc 

func BindNewExecutorFunc(id string, fn func(sessionID string, executorID string) (*Executor, error)) ExecutorBinding

BindNewExecutorFunc returns an ExecutorBinding that creates a new Executor for each workflow session by calling fn with the session ID and executor ID. Created executors are validated against id and stamped with the binding's implementation identity.

The returned binding is non-shared, so executor-local state does not require a reset hook between runs. It is not marked safe for concurrent workflow runs by default; set ExecutorBinding.SupportsConcurrentSharedExecution to true only when the factory and the executors it creates are safe for that use.

func (ExecutorBinding) CreateInstance 

func (eb ExecutorBinding) CreateInstance(sessionID string) (*Executor, error)

CreateInstance creates the executor for sessionID and validates that the returned executor matches this binding. It returns an error for placeholder bindings, nil executors, factory errors, or executor ID mismatches.

func (ExecutorBinding) String 

func (eb ExecutorBinding) String() string

String returns a compact representation of the binding for diagnostics.

func (ExecutorBinding) TryReset 

func (eb ExecutorBinding) TryReset() bool

TryReset resets this binding if it wraps a shared executor instance. Non-shared bindings are already isolated per session and therefore report success without invoking [ExecutorBinding.Reset].

type ExecutorCompletedEvent 

type ExecutorCompletedEvent struct {
	ExecutorID string
	Result     any
}

ExecutorCompletedEvent is an event triggered when an executor handler completes.

func (ExecutorCompletedEvent) Data 

func (e ExecutorCompletedEvent) Data() any

type ExecutorFailedEvent 

type ExecutorFailedEvent struct {
	ExecutorID string
	Error      error
}

ExecutorFailedEvent is an event triggered when an executor handler fails.

func (ExecutorFailedEvent) Data 

func (e ExecutorFailedEvent) Data() any

type ExecutorInvokedEvent 

type ExecutorInvokedEvent struct {
	ExecutorID string
	Message    any
}

ExecutorInvokedEvent is an event triggered when an executor handler is invoked.

func (ExecutorInvokedEvent) Data 

func (e ExecutorInvokedEvent) Data() any

type ExternalRequest 

type ExternalRequest struct {
	// PortInfo is the port to invoke.
	PortInfo RequestPortInfo

	// RequestID is a unique identifier for this request instance.
	RequestID string

	// Data is the data contained in the request.
	Data PortableValue
}

ExternalRequest represents a request to an external input port.

func NewExternalRequest 

func NewExternalRequest(id string, port RequestPort, data any) (*ExternalRequest, error)

NewExternalRequest creates a new ExternalRequest for the specified input port and data payload.

id is an optional unique identifier for this request instance. If id is empty, a UUID will be generated.

NewExternalRequest returns an error when the port is invalid or data does not match the expected request type.

func (*ExternalRequest) CreateResponse 

func (r *ExternalRequest) CreateResponse(data any) (*ExternalResponse, error)

CreateResponse creates a new ExternalResponse corresponding to r, with the specified data payload.

CreateResponse returns an error when data does not match the expected response type.

type ExternalResponse 

type ExternalResponse struct {
	// PortInfo is the port invoked.
	PortInfo RequestPortInfo

	// RequestID is the unique identifier of the corresponding request.
	RequestID string

	// Data is the data contained in the response.
	Data PortableValue
}

ExternalResponse represents a response from an external input port.

type MessageHandlerFunc 

type MessageHandlerFunc func(*Context, any) (any, error)

MessageHandlerFunc handles a routed workflow message.

The handler receives the concrete message value registered with RouteBuilder.AddHandlerRaw. Its return value may be forwarded as a message or yielded as output depending on the executor's Executor.

type OutputEvent 

type OutputEvent struct {
	// ExecutorID is the unique identifier of the executor that yielded this output.
	ExecutorID string
	Output     any
	Tags       []OutputTag
}

OutputEvent is an event triggered when the workflow produces an output.

func (OutputEvent) Data 

func (e OutputEvent) Data() any

func (OutputEvent) HasTag 

func (e OutputEvent) HasTag(tag OutputTag) bool

HasTag reports whether e carries tag.

func (OutputEvent) IsIntermediate 

func (e OutputEvent) IsIntermediate() bool

IsIntermediate reports whether e carries OutputTagIntermediate.

type OutputTag 

type OutputTag string

OutputTag identifies the kind of output represented by an OutputEvent. Terminal outputs are untagged; intermediate outputs carry OutputTagIntermediate. 

const OutputTagIntermediate OutputTag = "intermediate"

OutputTagIntermediate marks an output as intermediate rather than terminal.

func (OutputTag) MarshalJSON 

func (t OutputTag) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler.

func (OutputTag) String 

func (t OutputTag) String() string

func (*OutputTag) UnmarshalJSON 

func (t *OutputTag) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler.

func (OutputTag) Value 

func (t OutputTag) Value() string

Value returns the string identifier of the tag.

type PortableValue 

type PortableValue struct {
	TypeID TypeID
	// contains filtered or unexported fields
}

PortableValue represents a value that can be exported / imported to a workflow, e.g. through an external request/response, or through checkpointing. The zero PortableValue is invalid.

func AnyPortableValue 

func AnyPortableValue(v any) PortableValue

AnyPortableValue returns a PortableValue for the supplied value.

If the supplied value is of type PortableValue, it is returned unmodified.

func (*PortableValue) Any 

func (v *PortableValue) Any() any

Any returns v's value as an any.

func (*PortableValue) As 

func (v *PortableValue) As(typ reflect.Type) (any, bool)

As returns the contained value and true when it is assignable to typ (or can be delayed-deserialized to a value assignable to typ); otherwise it returns nil and false. No conversion is performed: when typ is an interface, the returned value is the concrete value assignable to that interface. It is the comma-ok extraction counterpart to PortableValue.Is.

func (*PortableValue) Delayed 

func (v *PortableValue) Delayed() bool

Delayed reports whether the value is stored in a delayed deserialized form.

func (*PortableValue) Is 

func (v *PortableValue) Is(typ reflect.Type) bool

Is reports whether the value is of the specified type. If the value is stored in a delayed deserialized form, it will attempt to deserialize it to the requested type.

func (PortableValue) MarshalJSON 

func (v PortableValue) MarshalJSON() ([]byte, error)

func (*PortableValue) UnmarshalJSON 

func (v *PortableValue) UnmarshalJSON(data []byte) error

type ProtocolBuilder 

type ProtocolBuilder struct {
	RouteBuilder RouteBuilder
	// contains filtered or unexported fields
}

ProtocolBuilder configures the routes and declared message protocol for an executor.

A ProtocolBuilder owns a RouteBuilder for message handlers and tracks the message types that may be sent with Context.SendMessage or yielded with Context.YieldOutput. The zero value is ready to use.

func (*ProtocolBuilder) ConfigureRoutes 

func (pb *ProtocolBuilder) ConfigureRoutes(configure func(*RouteBuilder) (*RouteBuilder, error)) *ProtocolBuilder

ConfigureRoutes fluently configures message handlers on this builder's route builder.

Any error returned from configure is recorded and returned later when the protocol is built.

func (*ProtocolBuilder) SendsMessageType 

func (pb *ProtocolBuilder) SendsMessageType(messageTypes ...reflect.Type) *ProtocolBuilder

SendsMessageType adds messageTypes to the set of declared message types this executor may send with Context.SendMessage. Nil and duplicate types are ignored.

func (*ProtocolBuilder) YieldsOutputType 

func (pb *ProtocolBuilder) YieldsOutputType(outputTypes ...reflect.Type) *ProtocolBuilder

YieldsOutputType adds outputTypes to the set of declared output types this executor may yield with Context.YieldOutput. Nil and duplicate types are ignored.

type ProtocolDescriptor 

type ProtocolDescriptor struct {
	// Accepts lists the message types accepted by the described protocol.
	Accepts []reflect.Type

	// Yields lists the message types the described protocol can yield as output.
	Yields []reflect.Type

	// Sends lists the message types the described protocol can send to connected
	// executors. Workflow descriptors leave this empty.
	Sends []reflect.Type

	// AcceptsAll reports whether the described protocol has a catch-all handler.
	AcceptsAll bool
}

ProtocolDescriptor describes the message protocol accepted, yielded, and sent by an executor or workflow.

type RequestHaltEvent 

type RequestHaltEvent struct {
	Result any
}

RequestHaltEvent signals that the workflow requested a halt, carrying an optional Result payload.

func (RequestHaltEvent) Data 

func (e RequestHaltEvent) Data() any

type RequestInfoEvent 

type RequestInfoEvent struct {
	Request *ExternalRequest
}

RequestInfoEvent is an event containing request information.

func (RequestInfoEvent) Data 

func (e RequestInfoEvent) Data() any

type RequestPort 

type RequestPort struct {
	// ID is the unique identifier for the input port.
	ID string

	// Request is the type of request messages that the input port will accept.
	Request reflect.Type

	// Response is the type of response messages that the input port will produce.
	Response reflect.Type
}

RequestPort is an external request port for a Workflow with the specified request and response types.

func (RequestPort) Bind 

func (p RequestPort) Bind() ExecutorBinding

Bind returns an ExecutorBinding that exposes p at the workflow boundary. The resulting executor accepts messages of p.Request type, raises them as [ExternalRequest]s via Context.PostRequest, and forwards the matching ExternalResponse data to downstream executors.

Calling Bind multiple times on the same port is fine: each returned binding has the same ID and port metadata, and creates request-port executor instances from that port. Bind panics if the port ID is empty or either message type is nil.

type RequestPortInfo 

type RequestPortInfo struct {
	// PortID is the request port's workflow-unique identifier.
	PortID string

	// RequestType is the type accepted by the request port.
	RequestType TypeID

	// ResponseType is the type returned through the request port.
	ResponseType TypeID
}

RequestPortInfo describes a workflow request port.

Request and response types are stored as TypeID values so external requests and responses can be validated after serialization or across package boundaries.

func NewRequestPortInfo 

func NewRequestPortInfo(port RequestPort) RequestPortInfo

NewRequestPortInfo creates a serializable descriptor for port.

type RouteBuilder 

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

RouteBuilder configures the message routes for an executor.

A RouteBuilder is normally used from Executor.ConfigureProtocol to register typed handlers with RouteBuilder.AddHandlerRaw and an optional fallback handler with RouteBuilder.AddCatchAll. The zero value is ready to use, and registration methods return the builder so calls can be chained. Invalid registrations are recorded on the builder and returned when the executor builds its router.

func (*RouteBuilder) AddCatchAll 

func (rb *RouteBuilder) AddCatchAll(handler func(*Context, PortableValue) (any, error), options ...AddHandlerOption) *RouteBuilder

AddCatchAll registers a fallback handler for messages without a typed handler.

The catch-all handler receives the message as a PortableValue. If the message was not already portable, the router wraps it with AnyPortableValue before invoking the handler.

By default, registering a second catch-all handler is an error; use WithHandlerOverwrite to replace an existing catch-all handler.

func (*RouteBuilder) AddHandlerRaw 

func (rb *RouteBuilder) AddHandlerRaw(messageType reflect.Type, outputType reflect.Type, handler MessageHandlerFunc, options ...AddHandlerOption) *RouteBuilder

AddHandlerRaw registers a typed handler using explicit runtime types.

messageType is the concrete message type accepted by handler. outputType, when non-nil, declares the handler result type used for workflow protocol discovery and automatic send/yield of handler return values. A nil outputType registers an action-style handler: it may use Context to send messages or yield outputs manually, but its handler return value is not auto-forwarded.

By default, registering a duplicate message type is an error; use WithHandlerOverwrite to replace an existing handler.

type ScopeID 

type ScopeID struct {

	// ScopeName identifies a shared scope when set.
	ScopeName string

	// ExecutorID identifies the executor that owns a private default scope.
	ExecutorID string
	// contains filtered or unexported fields
}

ScopeID is a unique identifier for a workflow state scope. If a scope name is not provided, it references the default scope private to the executor. Otherwise, regardless of the executor ID, it references a shared scope with the specified name.

func (ScopeID) Equal 

func (s ScopeID) Equal(other ScopeID) bool

Equal reports whether s and other refer to the same workflow state scope.

func (ScopeID) Hash 

func (s ScopeID) Hash(h *maphash.Hash)

Hash writes a stable hash representation of s into h.

func (ScopeID) MarshalJSON 

func (s ScopeID) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler for ScopeID.

func (*ScopeID) UnmarshalJSON 

func (s *ScopeID) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler for ScopeID.

type ScopeKey 

type ScopeKey struct {
	// ID identifies the state scope that owns the key.
	ID ScopeID

	// Key identifies the value within the scope.
	Key string
}

ScopeKey identifies a single state value within a workflow state scope.

func (ScopeKey) Equal 

func (s ScopeKey) Equal(other ScopeKey) bool

Equal reports whether s and other identify the same state value.

func (ScopeKey) Hash 

func (s ScopeKey) Hash(h *maphash.Hash)

Hash writes a stable hash representation of s into h.

type StartedEvent 

type StartedEvent struct {
	Message any
}

StartedEvent is emitted at the beginning of each input → processing → halt cycle, immediately before the first [SuperStep] of that cycle runs. It fires once per cycle in which there is actual work to process — typically once at the start of a run and again whenever new messages or external responses arrive after a halt. No event is emitted on cycles that complete without work (e.g. timeout-only loop iterations).

func (StartedEvent) Data 

func (e StartedEvent) Data() any

type StatefulExecutorCache 

type StatefulExecutorCache[T any] struct {
	// StateKey is the workflow state key used for reads and queued updates.
	StateKey string
	// InitialStateFactory creates the state value used when no state has been
	// stored yet, or when a stored pointer/slice/map-like value is nil.
	InitialStateFactory func() T

	// ScopeName is the optional workflow state scope name.
	ScopeName string
	// contains filtered or unexported fields
}

StatefulExecutorCache helps an executor read, update, and cache typed state.

The cache reads from Context.ReadOrInitState when available, falling back to Context.ReadState. It caches state within an executor instance unless concurrent runs are enabled or the caller asks to skip the cache.

func (*StatefulExecutorCache[T]) InvokeWithState 

func (s *StatefulExecutorCache[T]) InvokeWithState(ctx *Context, skipCache bool, fn func(ctx *Context, state T) (T, error)) error

InvokeWithState reads the current state, calls fn, and queues the returned state as an update.

func (*StatefulExecutorCache[T]) OnCheckpointRestored 

func (s *StatefulExecutorCache[T]) OnCheckpointRestored(*Context) error

OnCheckpointRestored invalidates the executor-local cache so the next read observes the state imported from the checkpoint rather than a stale cached value. Its signature matches Executor.OnCheckpointRestoredFunc, so a shared stateful executor should wire it there to stay consistent after an in-place checkpoint restore.

func (*StatefulExecutorCache[T]) QueueStateUpdate 

func (s *StatefulExecutorCache[T]) QueueStateUpdate(ctx *Context, state T) error

QueueStateUpdate normalizes and queues a typed state update on the context.

When concurrent runs are disabled, the in-memory cache is updated immediately so later reads in the same executor instance observe the new value.

func (*StatefulExecutorCache[T]) ReadState 

func (s *StatefulExecutorCache[T]) ReadState(ctx *Context, skipCache bool) (T, error)

ReadState returns the current typed state value.

When skipCache is false and concurrent runs are disabled, ReadState uses an executor-local cached value after the first read. When skipCache is true, or concurrent runs are enabled, it reads from workflow state each time.

func (*StatefulExecutorCache[T]) Reset 

func (s *StatefulExecutorCache[T]) Reset() error

Reset clears the executor-local cached state.

type SuperStepCompletedEvent 

type SuperStepCompletedEvent struct {
	StepNumber     int
	CompletionInfo *SuperStepCompletionInfo
}

SuperStepCompletedEvent is an event triggered when a super step completes.

func (SuperStepCompletedEvent) Data 

func (e SuperStepCompletedEvent) Data() any

type SuperStepCompletionInfo 

type SuperStepCompletionInfo struct {
	ActivatedExecutors    []string
	InstantiatedExecutors []string
	HasPendingMessages    bool
	HasPendingRequests    bool
	StateUpdated          bool
	CheckpointInfo        *CheckpointInfo
}

SuperStepCompletionInfo contains information about a completed super step.

type SuperStepStartInfo 

type SuperStepStartInfo struct {
	// The unique identifiers of [Executor] instances that sent messages
	// during the previous [SuperStep].
	SendingExecutors []string

	HasExternalMessages bool
}

SuperStepStartInfo contains debug information about the [SuperStep] starting to run.

type SuperStepStartedEvent 

type SuperStepStartedEvent struct {
	StepNumber int
	StartInfo  *SuperStepStartInfo
}

SuperStepStartedEvent is an event triggered when a super step starts.

func (SuperStepStartedEvent) Data 

func (e SuperStepStartedEvent) Data() any

type SwitchBuilder 

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

SwitchBuilder constructs a case-based fan-out edge. Cases are evaluated in the order they were added; matching cases route the message to that case's targets. If no case matches, the default targets are used.

func (*SwitchBuilder) AddCase 

func (s *SwitchBuilder) AddCase(predicate func(msg any) bool, targets ...ExecutorBinding) *SwitchBuilder

AddCase adds a case branch matching messages of type T satisfying the predicate. The matched message is routed to all bindings in targets.

func (*SwitchBuilder) AddToBuilder 

func (s *SwitchBuilder) AddToBuilder(wb *Builder) *Builder

AddToBuilder commits the configured switch onto wb as a fan-out edge with an assigner that picks targets based on the registered cases.

func (*SwitchBuilder) WithDefault 

func (s *SwitchBuilder) WithDefault(targets ...ExecutorBinding) *SwitchBuilder

WithDefault sets the targets to dispatch to when no case matches.

type TelemetryOptions 

type TelemetryOptions struct {
	// EnableSensitiveData includes serialized message inputs, outputs, and
	// message contents in span attributes. It is disabled by default.
	EnableSensitiveData bool

	DisableWorkflowBuild    bool
	DisableWorkflowRun      bool
	DisableExecutorProcess  bool
	DisableEdgeGroupProcess bool
	DisableMessageSend      bool
}

TelemetryOptions configures workflow telemetry instrumentation.

type TurnToken 

type TurnToken struct {
	// EmitEvents overrides the workflow's default event-emission behavior for
	// this turn when set.
	EmitEvents *bool
}

TurnToken is a control message that advances a workflow turn.

func (TurnToken) EmitEventsOr 

func (t TurnToken) EmitEventsOr(defaultValue bool) bool

EmitEventsOr returns the token's event-emission override, or defaultValue if the token does not specify one.

type TypeID 

type TypeID struct {
	// PackageName is the package path for named Go types. It is empty for
	// built-in and unnamed types.
	PackageName string

	// TypeName is the type name for named Go types, or the string form for
	// unnamed types such as pointers, slices, and maps.
	TypeName string
}

TypeID is a representation of a type's identity, including its package and type names. Pointer types are identified by their element type.

func NewTypeID 

func NewTypeID(typ reflect.Type) TypeID

NewTypeID creates a TypeID for typ and caches typ as a runtime type that can be resolved from that identity. A nil type returns the zero TypeID. Pointer types are identified by their element type.

func (TypeID) Match 

func (t TypeID) Match(typ reflect.Type) bool

Match reports whether typ has the same package and type names as t. The zero TypeID represents an unknown type and does not match any type.

func (TypeID) MatchPolymorphic 

func (t TypeID) MatchPolymorphic(typ reflect.Type) bool

MatchPolymorphic reports whether typ either matches this type identity exactly or can be assigned to the runtime type cached or discovered for this identity. Concrete types can match interface response ports when the interface type has been cached through NewTypeID or discovered from runtime type metadata.

func (TypeID) String 

func (t TypeID) String() string

String returns the type identity in a compact diagnostic form.

type Workflow 

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

Workflow is an executable graph of bound executors, edges, outputs, and request ports. Workflows are created by Builder.

func (*Workflow) AllowConcurrent 

func (w *Workflow) AllowConcurrent() bool

AllowConcurrent reports whether every bound executor in the workflow supports cross-run shared execution.

func (*Workflow) CheckOwnership 

func (w *Workflow) CheckOwnership(token any) bool

CheckOwnership reports whether token currently owns the workflow. A nil token matches only an unowned workflow. Non-nil tokens are compared by pointer-like identity, matching .NET's reference-token ownership model.

func (*Workflow) ContextWithTelemetry 

func (w *Workflow) ContextWithTelemetry(ctx context.Context) context.Context

ContextWithTelemetry returns ctx annotated with the workflow's telemetry context.

func (*Workflow) DescribeProtocol 

func (w *Workflow) DescribeProtocol() (ProtocolDescriptor, error)

DescribeProtocol returns the protocol accepted by the workflow's start executor and yielded by its output executors.

func (*Workflow) Description 

func (w *Workflow) Description() string

Description returns optional human-readable workflow detail.

func (*Workflow) Edges 

func (w *Workflow) Edges() map[string][]Edge

Edges returns a copy of the workflow edges keyed by source executor ID.

func (*Workflow) ExecutorBinding 

func (w *Workflow) ExecutorBinding(id string) (ExecutorBinding, bool)

ExecutorBinding returns the executor binding for id.

func (*Workflow) HasOutputExecutor 

func (w *Workflow) HasOutputExecutor(executorID string) bool

HasOutputExecutor reports whether executorID is exposed as a workflow output.

func (*Workflow) HasResettableExecutors 

func (w *Workflow) HasResettableExecutors() bool

HasResettableExecutors reports whether any executor binding can reset shared resources between workflow runs.

func (*Workflow) Name 

func (w *Workflow) Name() string

Name returns the optional human-readable workflow name.

func (*Workflow) OutgoingEdges 

func (w *Workflow) OutgoingEdges(executorID string) []Edge

OutgoingEdges returns the edges whose source is executorID.

func (*Workflow) OutputExecutorIDs 

func (w *Workflow) OutputExecutorIDs() []string

OutputExecutorIDs returns the executor IDs exposed as workflow outputs.

func (*Workflow) OutputExecutors 

func (w *Workflow) OutputExecutors() map[string][]OutputTag

OutputExecutors returns workflow output sources with their tag metadata.

func (*Workflow) ReflectEdges 

func (w *Workflow) ReflectEdges() map[string][]EdgeInfo

ReflectEdges returns workflow edge metadata keyed by source executor ID.

func (*Workflow) ReflectExecutors 

func (w *Workflow) ReflectExecutors() map[string]ExecutorBinding

ReflectExecutors returns a copy of the workflow's executor bindings keyed by ID. Modifying the returned map does not affect the workflow.

func (*Workflow) ReflectPorts 

func (w *Workflow) ReflectPorts() map[string]RequestPortInfo

ReflectPorts returns workflow request port metadata keyed by port ID.

func (*Workflow) ReleaseOwnership 

func (w *Workflow) ReleaseOwnership(token any) error

ReleaseOwnership releases workflow ownership held by token and attempts to reset shared executor state. token is compared by pointer-like identity.

func (*Workflow) ReleaseOwnershipTo 

func (w *Workflow) ReleaseOwnershipTo(token any, targetToken any) error

ReleaseOwnershipTo releases workflow ownership held by token, restores targetToken as the owner when non-nil, and attempts to reset shared executor state. Non-nil tokens are compared by pointer-like identity.

func (*Workflow) RequestPort 

func (w *Workflow) RequestPort(id string) (RequestPort, bool)

RequestPort returns the workflow request port with id.

func (*Workflow) RequestPorts 

func (w *Workflow) RequestPorts() map[string]RequestPort

RequestPorts returns a copy of the workflow request ports keyed by port ID.

func (*Workflow) StartExecutorID 

func (w *Workflow) StartExecutorID() string

StartExecutorID returns the executor ID that receives the initial input.

func (*Workflow) TakeOwnership 

func (w *Workflow) TakeOwnership(token any, newToken any, subworkflow bool) error

TakeOwnership transfers workflow ownership from token to newToken. The subworkflow flag records whether the workflow is being owned by a parent workflow rather than by a direct runner. newToken must be a non-nil pointer-like value so ownership can be compared by identity.

func (*Workflow) TryReset 

func (w *Workflow) TryReset() bool

TryReset attempts to reset all shared executor bindings that require reset support before workflow reuse.

Source Files

View all Source files

Directories

Path Synopsis
This file hosts an agent.Agent as a workflow workflow.Executor, so the agent can participate in graphs alongside regular executors and other hosted agents.
 This file hosts an agent.Agent as a workflow workflow.Executor, so the agent can participate in graphs alongside regular executors and other hosted agents.
Package checkpoint provides checkpoint storage and management for workflow runs, enabling restartability and time-travel resume.
 Package checkpoint provides checkpoint storage and management for workflow runs, enabling restartability and time-travel resume.
Package inproc provides the in-process workflow runtime: it instantiates a workflow's executors and drives execution, streaming or to completion, with optional checkpointing.
 Package inproc provides the in-process workflow runtime: it instantiates a workflow's executors and drives execution, streaming or to completion, with optional checkpointing.
internal
checkpoint
execution
observability
workflowtest
Package observability defines the tracing abstraction (Tracer, spans, and attributes) used to instrument workflow execution; the opentelemetry subpackage provides an OpenTelemetry implementation.
 Package observability defines the tracing abstraction (Tracer, spans, and attributes) used to instrument workflow execution; the opentelemetry subpackage provides an OpenTelemetry implementation.
opentelemetry
Package opentelemetry implements the workflow observability Tracer using OpenTelemetry, emitting spans for workflow and executor activity.
 Package opentelemetry implements the workflow observability Tracer using OpenTelemetry, emitting spans for workflow and executor activity.

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.