workflow

package module

Go to main page

Versions in this module

v0

v0.0.5

Apr 13, 2026

v0.0.4

Apr 13, 2026

v0.0.3

Apr 12, 2026

v0.0.2

Apr 12, 2026

Changes in this version

+ const CheckpointSchemaVersion

+ var ErrAlreadyStarted = errors.New("workflow: execution already started")

+ var ErrBranchNotFound = errors.New("workflow: branch not found")

+ var ErrDuplicateActivity = errors.New("workflow: duplicate activity registration")

+ var ErrDuplicateBranchName = errors.New("workflow: duplicate branch name")

+ var ErrDuplicateStepName = errors.New("workflow: duplicate step name")

+ var ErrEmptyStepName = errors.New("workflow: empty step name")

+ var ErrFenceViolation = errors.New("fence violation: lease lost")

+ var ErrInvalidExpression = errors.New("workflow: invalid expression")

+ var ErrInvalidHeartbeatInterval = errors.New("workflow: heartbeat interval must be positive")

+ var ErrInvalidModifier = errors.New("workflow: modifier not allowed on step kind")

+ var ErrInvalidRetryConfig = errors.New("workflow: invalid retry config")

+ var ErrInvalidSleepConfig = errors.New("workflow: invalid sleep config")

+ var ErrInvalidStepKind = errors.New("workflow: conflicting step kinds")

+ var ErrInvalidStorePath = errors.New("workflow: store field must be a bare variable name")

+ var ErrInvalidTemplate = errors.New("workflow: invalid template")

+ var ErrInvalidWaitConfig = errors.New("workflow: invalid wait_signal config")

+ var ErrNilExecution = errors.New("workflow: execution must not be nil")

+ var ErrNilHeartbeatFunc = errors.New("workflow: heartbeat func must not be nil")

+ var ErrNoCheckpoint = errors.New("workflow: no checkpoint found")

+ var ErrReservedBranchName = errors.New("workflow: branch name 'main' is reserved")

+ var ErrUnknownActivity = errors.New("workflow: activity not registered")

+ var ErrUnknownCatchTarget = errors.New("workflow: catch destination not found")

+ var ErrUnknownEdgeTarget = errors.New("workflow: edge destination not found")

+ var ErrUnknownJoinBranch = errors.New("workflow: join branch not found")

+ var ErrUnknownStartStep = errors.New("workflow: start step not found")

+ var ErrWaitTimeout = errors.New("workflow: wait timed out")

+ func DefaultScriptCompiler() script.Compiler

+ func OutputAs[T any](r *ExecutionResult, key string) (T, bool)

+ func PauseBranchInCheckpoint(ctx context.Context, cp Checkpointer, executionID, branchID, reason string) error

+ func UnpauseBranchInCheckpoint(ctx context.Context, cp Checkpointer, executionID, branchID string) error

type Activity

+ func ActivityFunc(name string, fn ExecuteActivityFunc) Activity

+ func TypedActivityFunc[TParams, TResult any](name string, fn func(ctx Context, params TParams) (TResult, error)) Activity

type ActivityExecutionEvent

+ BranchID string

type ActivityLogEntry

+ BranchID string

type ActivityRegistry

+ func NewActivityRegistry() *ActivityRegistry

+ func (r *ActivityRegistry) Get(name string) (Activity, bool)

+ func (r *ActivityRegistry) MustRegister(a Activity) *ActivityRegistry

+ func (r *ActivityRegistry) Names() []string

+ func (r *ActivityRegistry) Register(a Activity) error

+ type AtomicCheckpointer interface

+ AtomicUpdate func(ctx context.Context, executionID string, fn func(*Checkpoint) error) error

type BaseExecutionCallbacks

+ func (n *BaseExecutionCallbacks) AfterBranchExecution(ctx context.Context, event *BranchExecutionEvent)

+ func (n *BaseExecutionCallbacks) BeforeBranchExecution(ctx context.Context, event *BranchExecutionEvent)

+ type BranchExecutionEvent struct

+ BranchID string

+ CurrentStep string

+ Duration time.Duration

+ EndTime time.Time

+ Error error

+ ExecutionID string

+ StartTime time.Time

+ Status ExecutionStatus

+ StepOutputs map[string]any

+ WorkflowName string

+ type BranchLocalState struct

+ func NewBranchLocalState(inputs, variables map[string]any) *BranchLocalState

+ func (s *BranchLocalState) Delete(key string)

+ func (s *BranchLocalState) Get(key string) (any, bool)

+ func (s *BranchLocalState) Keys() []string

+ func (s *BranchLocalState) Set(key string, value any)

+ type BranchState struct

+ ActivityHistory map[string]any

+ ActivityHistoryStep string

+ CurrentStep string

+ EndTime time.Time

+ ErrorMessage string

+ ID string

+ PauseReason string

+ PauseRequested bool

+ StartTime time.Time

+ Status ExecutionStatus

+ StepOutputs map[string]any

+ Variables map[string]any

+ Wait *WaitState

+ func (p *BranchState) Copy() *BranchState

type CallbackChain

+ func (c *CallbackChain) AfterBranchExecution(ctx context.Context, event *BranchExecutionEvent)

+ func (c *CallbackChain) BeforeBranchExecution(ctx context.Context, event *BranchExecutionEvent)

type Checkpoint

+ BranchCounter int

+ BranchStates map[string]*BranchState

+ SchemaVersion int

type Checkpointer

+ func WithFencing(inner Checkpointer, fenceCheck FenceFunc) Checkpointer

type ChildWorkflowExecutorOptions

+ CleanupTimeout time.Duration

+ ScriptCompiler script.Compiler

+ type CompletionHook func(ctx context.Context, result *ExecutionResult) ([]FollowUpSpec, error)

type Context

+ BranchID func() string

+ Compiler func() script.Compiler

+ Delete func(key string)

+ Get func(key string) (value any, exists bool)

+ History func() *History

+ Inputs func() Inputs

+ Keys func() []string

+ Logger func() *slog.Logger

+ ReportProgress func(detail ProgressDetail)

+ Set func(key string, value any)

+ StepName func() string

+ Wait func(topic string, timeout time.Duration) (any, error)

type Edge

+ BranchName string

+ type ExecuteOption func(*executeConfig)

+ func ResumeFrom(priorID string) ExecuteOption

type Execution

+ func (e *Execution) Execute(ctx context.Context, opts ...ExecuteOption) (*ExecutionResult, error)

+ func (e *Execution) PauseBranch(branchID, reason string) error

+ func (e *Execution) UnpauseBranch(branchID string) error

type ExecutionCallbacks

+ AfterBranchExecution func(ctx context.Context, event *BranchExecutionEvent)

+ BeforeBranchExecution func(ctx context.Context, event *BranchExecutionEvent)

type ExecutionContextOptions

+ ActivityHistory *History

+ BranchID string

+ BranchLocalState *BranchLocalState

+ ExecutionID string

+ PendingWait *WaitState

+ SignalStore SignalStore

+ type ExecutionOption func(*executionConfig)

+ func WithActivityLogger(al ActivityLogger) ExecutionOption

+ func WithCheckpointer(cp Checkpointer) ExecutionOption

+ func WithExecutionCallbacks(cb ExecutionCallbacks) ExecutionOption

+ func WithExecutionID(id string) ExecutionOption

+ func WithInputs(m map[string]any) ExecutionOption

+ func WithLogger(l *slog.Logger) ExecutionOption

+ func WithScriptCompiler(sc script.Compiler) ExecutionOption

+ func WithSignalStore(ss SignalStore) ExecutionOption

+ func WithStepProgressStore(s StepProgressStore) ExecutionOption

+ type ExecutionResult struct

+ Error *WorkflowError

+ FollowUps []FollowUpSpec

+ Outputs map[string]any

+ Status ExecutionStatus

+ Suspension *SuspensionInfo

+ Timing ExecutionTiming

+ WorkflowName string

+ func (r *ExecutionResult) Completed() bool

+ func (r *ExecutionResult) Failed() bool

+ func (r *ExecutionResult) NeedsResume() bool

+ func (r *ExecutionResult) NextWakeAt() (time.Time, bool)

+ func (r *ExecutionResult) Output(key string) (any, bool)

+ func (r *ExecutionResult) OutputBool(key string) (bool, bool)

+ func (r *ExecutionResult) OutputInt(key string) (int, bool)

+ func (r *ExecutionResult) OutputString(key string) (string, bool)

+ func (r *ExecutionResult) Paused() bool

+ func (r *ExecutionResult) Suspended() bool

+ func (r *ExecutionResult) Topics() []string

+ func (r *ExecutionResult) WaitReason() SuspensionReason

type ExecutionStatus

+ const ExecutionStatusPaused

+ const ExecutionStatusSuspended

+ type ExecutionTiming struct

+ Duration time.Duration

+ FinishedAt time.Time

+ StartedAt time.Time

+ type FenceFunc func(ctx context.Context) error

+ type FollowUpSpec struct

+ Inputs map[string]any

+ Metadata map[string]any

+ WorkflowName string

+ type HeartbeatConfig struct

+ Func HeartbeatFunc

+ Interval time.Duration

+ type HeartbeatFunc func(ctx context.Context) error

+ type History struct

+ func NewHistoryForTest() *History

+ func (h *History) Get(key string) (any, bool)

+ func (h *History) Len() int

+ func (h *History) RecordOrReplay(key string, fn func() (any, error)) (any, error)

+ type Inputs struct

+ func NewInputsForTest(m map[string]any) Inputs

+ func (i Inputs) Get(key string) (any, bool)

+ func (i Inputs) Keys() []string

+ func (i Inputs) Len() int

+ func (i Inputs) ToMap() map[string]any

type JoinConfig

+ BranchMappings map[string]string

+ Branches []string

type JoinState

+ WaitingBranchID string

+ type MemorySignalStore struct

+ func NewMemorySignalStore() *MemorySignalStore

+ func (m *MemorySignalStore) Receive(ctx context.Context, executionID, topic string) (*Signal, error)

+ func (m *MemorySignalStore) Send(ctx context.Context, executionID, topic string, payload any) error

type Options

+ StartAt string

type Output

+ Branch string

+ type PauseConfig struct

+ Reason string

+ type ProgressDetail struct

+ Data map[string]any

+ Message string

+ type RunOption func(*runConfig)

+ func WithCompletionHook(h CompletionHook) RunOption

+ func WithHeartbeat(h *HeartbeatConfig) RunOption

+ func WithResumeFrom(priorExecutionID string) RunOption

+ func WithRunTimeout(d time.Duration) RunOption

+ type Runner struct

+ func NewRunner(opts ...RunnerOption) *Runner

+ func (r *Runner) Run(ctx context.Context, exec *Execution, opts ...RunOption) (*ExecutionResult, error)

+ type RunnerOption func(*runnerConfig)

+ func WithDefaultTimeout(d time.Duration) RunnerOption

+ func WithRunnerLogger(l *slog.Logger) RunnerOption

+ type Signal struct

+ ExecutionID string

+ Payload any

+ Topic string

+ type SignalStore interface

+ Receive func(ctx context.Context, executionID, topic string) (*Signal, error)

+ Send func(ctx context.Context, executionID, topic string, payload any) error

+ type SleepConfig struct

+ Duration time.Duration

type Step

+ Pause *PauseConfig

+ Sleep *SleepConfig

+ WaitSignal *WaitSignalConfig

+ type StepProgress struct

+ ActivityName string

+ Attempt int

+ BranchID string

+ Detail *ProgressDetail

+ Error string

+ FinishedAt time.Time

+ StartedAt time.Time

+ Status StepStatus

+ StepName string

+ type StepProgressStore interface

+ UpdateStepProgress func(ctx context.Context, executionID string, progress StepProgress) error

+ type StepStatus string

+ const StepStatusCompleted

+ const StepStatusFailed

+ const StepStatusPending

+ const StepStatusRunning

+ const StepStatusSkipped

+ type SuspendedBranch struct

+ BranchID string

+ PauseReason string

+ Reason SuspensionReason

+ StepName string

+ Topic string

+ WakeAt time.Time

+ type SuspensionInfo struct

+ Reason SuspensionReason

+ SuspendedBranches []SuspendedBranch

+ Topics []string

+ WakeAt time.Time

+ type SuspensionReason string

+ const SuspensionReasonPaused

+ const SuspensionReasonSleeping

+ const SuspensionReasonWaitingSignal

+ type ValidationError struct

+ Problems []ValidationProblem

+ func (e *ValidationError) Error() string

+ func (e *ValidationError) Is(target error) bool

+ type ValidationProblem struct

+ Err error

+ Message string

+ Step string

+ func (p ValidationProblem) String() string

+ type WaitKind string

+ const WaitKindSignal

+ const WaitKindSleep

+ func (k *WaitKind) UnmarshalJSON(data []byte) error

+ type WaitSignalConfig struct

+ OnTimeout string

+ Store string

+ Timeout time.Duration

+ Topic string

+ type WaitState struct

+ Kind WaitKind

+ Remaining time.Duration

+ Timeout time.Duration

+ Topic string

+ WakeAt time.Time

type Workflow

+ func (w *Workflow) Validate() error

v0.0.1

Jul 23, 2025

Changes in this version

+ const ErrorTypeActivityFailed

+ const ErrorTypeAll

+ const ErrorTypeFatal

+ const ErrorTypeTimeout

+ func ApplyPatches(container VariableContainer, patches []Patch)

+ func InputsFromContext(ctx Context) map[string]any

+ func MatchesErrorType(err error, errorType string) bool

+ func NewContext(ctx context.Context, opts ExecutionContextOptions) *executionContext

+ func NewExecutionID() string

+ func NewJSONLogger() *slog.Logger

+ func NewLogger() *slog.Logger

+ func VariablesFromContext(ctx Context) map[string]any

+ type Activity interface

+ Execute func(ctx Context, parameters map[string]any) (any, error)

+ Name func() string

+ func NewActivityFunction(name string, fn ExecuteActivityFunc) Activity

+ func NewTypedActivityFunction[TParams, TResult any](name string, fn func(ctx Context, params TParams) (TResult, error)) Activity

+ func NewTypedActivity[TParams, TResult any](activity TypedActivity[TParams, TResult]) Activity

+ type ActivityExecutionEvent struct

+ ActivityName string

+ Duration time.Duration

+ EndTime time.Time

+ Error error

+ ExecutionID string

+ Parameters map[string]any

+ PathID string

+ Result any

+ StartTime time.Time

+ StepName string

+ WorkflowName string

+ type ActivityExecutor interface

+ ExecuteActivity func(ctx context.Context, stepName, pathID string, activity Activity, ...) (result interface{}, err error)

+ type ActivityFunction struct

+ func (a *ActivityFunction) Execute(ctx Context, parameters map[string]any) (any, error)

+ func (a *ActivityFunction) Name() string

+ type ActivityLogEntry struct

+ Activity string

+ Duration float64

+ Error string

+ ExecutionID string

+ ID string

+ Parameters map[string]interface{}

+ PathID string

+ Result interface{}

+ StartTime time.Time

+ StepName string

+ type ActivityLogger interface

+ GetActivityHistory func(ctx context.Context, executionID string) ([]*ActivityLogEntry, error)

+ LogActivity func(ctx context.Context, entry *ActivityLogEntry) error

+ type ActivityRegistry map[string]Activity

+ type BaseExecutionCallbacks struct

+ func (n *BaseExecutionCallbacks) AfterActivityExecution(ctx context.Context, event *ActivityExecutionEvent)

+ func (n *BaseExecutionCallbacks) AfterPathExecution(ctx context.Context, event *PathExecutionEvent)

+ func (n *BaseExecutionCallbacks) AfterWorkflowExecution(ctx context.Context, event *WorkflowExecutionEvent)

+ func (n *BaseExecutionCallbacks) BeforeActivityExecution(ctx context.Context, event *ActivityExecutionEvent)

+ func (n *BaseExecutionCallbacks) BeforePathExecution(ctx context.Context, event *PathExecutionEvent)

+ func (n *BaseExecutionCallbacks) BeforeWorkflowExecution(ctx context.Context, event *WorkflowExecutionEvent)

+ type CallbackChain struct

+ func NewCallbackChain(callbacks ...ExecutionCallbacks) *CallbackChain

+ func (c *CallbackChain) Add(callback ExecutionCallbacks)

+ func (c *CallbackChain) AfterActivityExecution(ctx context.Context, event *ActivityExecutionEvent)

+ func (c *CallbackChain) AfterPathExecution(ctx context.Context, event *PathExecutionEvent)

+ func (c *CallbackChain) AfterWorkflowExecution(ctx context.Context, event *WorkflowExecutionEvent)

+ func (c *CallbackChain) BeforeActivityExecution(ctx context.Context, event *ActivityExecutionEvent)

+ func (c *CallbackChain) BeforePathExecution(ctx context.Context, event *PathExecutionEvent)

+ func (c *CallbackChain) BeforeWorkflowExecution(ctx context.Context, event *WorkflowExecutionEvent)

+ type CatchConfig struct

+ ErrorEquals []string

+ Next string

+ Store string

+ type Checkpoint struct

+ CheckpointAt time.Time

+ EndTime time.Time

+ Error string

+ ExecutionID string

+ ID string

+ Inputs map[string]interface{}

+ JoinStates map[string]*JoinState

+ Outputs map[string]interface{}

+ PathCounter int

+ PathStates map[string]*PathState

+ StartTime time.Time

+ Status string

+ Variables map[string]interface{}

+ WorkflowName string

+ type Checkpointer interface

+ DeleteCheckpoint func(ctx context.Context, executionID string) error

+ LoadCheckpoint func(ctx context.Context, executionID string) (*Checkpoint, error)

+ SaveCheckpoint func(ctx context.Context, checkpoint *Checkpoint) error

+ type ChildWorkflowExecutor interface

+ ExecuteAsync func(ctx context.Context, spec *ChildWorkflowSpec) (*ChildWorkflowHandle, error)

+ ExecuteSync func(ctx context.Context, spec *ChildWorkflowSpec) (*ChildWorkflowResult, error)

+ GetResult func(ctx context.Context, handle *ChildWorkflowHandle) (*ChildWorkflowResult, error)

+ type ChildWorkflowExecutorOptions struct

+ Activities []Activity

+ ActivityLogger ActivityLogger

+ Checkpointer Checkpointer

+ Logger *slog.Logger

+ WorkflowRegistry WorkflowRegistry

+ type ChildWorkflowHandle struct

+ ExecutionID string

+ WorkflowName string

+ type ChildWorkflowResult struct

+ Duration time.Duration

+ Error error

+ ExecutionID string

+ Outputs map[string]interface{}

+ Status ExecutionStatus

+ type ChildWorkflowSpec struct

+ Inputs map[string]interface{}

+ ParentID string

+ Sync bool

+ Timeout time.Duration

+ WorkflowName string

+ type Context interface

+ GetCompiler func() script.Compiler

+ GetInput func(key string) (value any, exists bool)

+ GetLogger func() *slog.Logger

+ GetPathID func() string

+ GetStepName func() string

+ ListInputs func() []string

+ func WithCancel(parent Context) (Context, context.CancelFunc)

+ func WithTimeout(parent Context, timeout time.Duration) (Context, context.CancelFunc)

+ type DefaultChildWorkflowExecutor struct

+ func NewDefaultChildWorkflowExecutor(opts ChildWorkflowExecutorOptions) (*DefaultChildWorkflowExecutor, error)

+ func (e *DefaultChildWorkflowExecutor) ExecuteAsync(ctx context.Context, spec *ChildWorkflowSpec) (*ChildWorkflowHandle, error)

+ func (e *DefaultChildWorkflowExecutor) ExecuteSync(ctx context.Context, spec *ChildWorkflowSpec) (*ChildWorkflowResult, error)

+ func (e *DefaultChildWorkflowExecutor) GetResult(ctx context.Context, handle *ChildWorkflowHandle) (*ChildWorkflowResult, error)

+ type Each struct

+ As string

+ Items any

+ type Edge struct

+ Condition string

+ Path string

+ Step string

+ type EdgeMatchingStrategy string

+ const EdgeMatchingAll

+ const EdgeMatchingFirst

+ type ErrorOutput struct

+ Cause string

+ Details interface{}

+ Error string

+ type ExecuteActivityFunc func(ctx Context, parameters map[string]any) (any, error)

+ type Execution struct

+ func NewExecution(opts ExecutionOptions) (*Execution, error)

+ func (e *Execution) GetOutputs() map[string]any

+ func (e *Execution) ID() string

+ func (e *Execution) Resume(ctx context.Context, priorExecutionID string) error

+ func (e *Execution) Run(ctx context.Context) error

+ func (e *Execution) Status() ExecutionStatus

+ type ExecutionAdapter struct

+ func (e *ExecutionAdapter) ExecuteActivity(ctx context.Context, stepName string, pathID string, activity Activity, ...) (any, error)

+ type ExecutionCallbacks interface

+ AfterActivityExecution func(ctx context.Context, event *ActivityExecutionEvent)

+ AfterPathExecution func(ctx context.Context, event *PathExecutionEvent)

+ AfterWorkflowExecution func(ctx context.Context, event *WorkflowExecutionEvent)

+ BeforeActivityExecution func(ctx context.Context, event *ActivityExecutionEvent)

+ BeforePathExecution func(ctx context.Context, event *PathExecutionEvent)

+ BeforeWorkflowExecution func(ctx context.Context, event *WorkflowExecutionEvent)

+ func NewBaseExecutionCallbacks() ExecutionCallbacks

+ type ExecutionContextOptions struct

+ Compiler script.Compiler

+ Logger *slog.Logger

+ PathID string

+ PathLocalState *PathLocalState

+ StepName string

+ type ExecutionOptions struct

+ Activities []Activity

+ ActivityLogger ActivityLogger

+ Checkpointer Checkpointer

+ ExecutionCallbacks ExecutionCallbacks

+ ExecutionID string

+ Formatter WorkflowFormatter

+ Inputs map[string]any

+ Logger *slog.Logger

+ ScriptCompiler script.Compiler

+ Workflow *Workflow

+ type ExecutionState struct

+ func (s *ExecutionState) AddPathToJoin(stepName, pathID string, config *JoinConfig, ...)

+ func (s *ExecutionState) FromCheckpoint(checkpoint *Checkpoint)

+ func (s *ExecutionState) GeneratePathID(parentID, pathName string) (string, error)

+ func (s *ExecutionState) GetAllJoinStates() map[string]*JoinState

+ func (s *ExecutionState) GetError() error

+ func (s *ExecutionState) GetFailedPathIDs() []string

+ func (s *ExecutionState) GetInputs() map[string]any

+ func (s *ExecutionState) GetJoinState(stepName string) *JoinState

+ func (s *ExecutionState) GetOutputs() map[string]any

+ func (s *ExecutionState) GetPathStates() map[string]*PathState

+ func (s *ExecutionState) GetStartTime() time.Time

+ func (s *ExecutionState) GetStatus() ExecutionStatus

+ func (s *ExecutionState) GetWaitingPathIDs() []string

+ func (s *ExecutionState) ID() string

+ func (s *ExecutionState) IsJoinReady(stepName string) bool

+ func (s *ExecutionState) NextPathID(baseID string) string

+ func (s *ExecutionState) RemoveJoinState(stepName string)

+ func (s *ExecutionState) SetError(err error)

+ func (s *ExecutionState) SetFinished(status ExecutionStatus, endTime time.Time, err error)

+ func (s *ExecutionState) SetID(id string)

+ func (s *ExecutionState) SetOutput(key string, value any)

+ func (s *ExecutionState) SetPathState(pathID string, state *PathState)

+ func (s *ExecutionState) SetStatus(status ExecutionStatus)

+ func (s *ExecutionState) SetTiming(startTime, endTime time.Time)

+ func (s *ExecutionState) ToCheckpoint() *Checkpoint

+ func (s *ExecutionState) UpdatePathState(pathID string, updateFn func(*PathState))

+ type ExecutionStatus string

+ const ExecutionStatusCompleted

+ const ExecutionStatusFailed

+ const ExecutionStatusPending

+ const ExecutionStatusRunning

+ const ExecutionStatusWaiting

+ type ExecutionSummary struct

+ Duration time.Duration

+ EndTime time.Time

+ Error string

+ ExecutionID string

+ StartTime time.Time

+ Status string

+ WorkflowName string

+ type FileActivityLogger struct

+ func NewFileActivityLogger(directory string) *FileActivityLogger

+ func (l *FileActivityLogger) GetActivityHistory(ctx context.Context, executionID string) ([]*ActivityLogEntry, error)

+ func (l *FileActivityLogger) LogActivity(ctx context.Context, entry *ActivityLogEntry) error

+ type FileCheckpointer struct

+ func NewFileCheckpointer(dataDir string) (*FileCheckpointer, error)

+ func (c *FileCheckpointer) DeleteCheckpoint(ctx context.Context, executionID string) error

+ func (c *FileCheckpointer) ListExecutions(ctx context.Context) ([]*ExecutionSummary, error)

+ func (c *FileCheckpointer) LoadCheckpoint(ctx context.Context, executionID string) (*Checkpoint, error)

+ func (c *FileCheckpointer) SaveCheckpoint(ctx context.Context, checkpoint *Checkpoint) error

+ type Input struct

+ Default interface{}

+ Description string

+ Name string

+ Type string

+ func (i *Input) IsRequired() bool

+ type JitterStrategy string

+ const JitterFull

+ const JitterNone

+ type JoinConfig struct

+ Count int

+ PathMappings map[string]string

+ Paths []string

+ type JoinRequest struct

+ Config *JoinConfig

+ StepName string

+ StepOutputs map[string]any

+ Variables map[string]any

+ type JoinState struct

+ Config *JoinConfig

+ CreatedAt time.Time

+ StepName string

+ WaitingPathID string

+ type MemoryWorkflowRegistry struct

+ func NewMemoryWorkflowRegistry() *MemoryWorkflowRegistry

+ func (r *MemoryWorkflowRegistry) Get(name string) (*Workflow, bool)

+ func (r *MemoryWorkflowRegistry) List() []string

+ func (r *MemoryWorkflowRegistry) Register(workflow *Workflow) error

+ type NullActivityLogger struct

+ func NewNullActivityLogger() *NullActivityLogger

+ func (l *NullActivityLogger) GetActivityHistory(ctx context.Context, executionID string) ([]*ActivityLogEntry, error)

+ func (l *NullActivityLogger) LogActivity(ctx context.Context, entry *ActivityLogEntry) error

+ type NullCheckpointer struct

+ func NewNullCheckpointer() *NullCheckpointer

+ func (c *NullCheckpointer) DeleteCheckpoint(ctx context.Context, executionID string) error

+ func (c *NullCheckpointer) LoadCheckpoint(ctx context.Context, executionID string) (*Checkpoint, error)

+ func (c *NullCheckpointer) SaveCheckpoint(ctx context.Context, checkpoint *Checkpoint) error

+ type Options struct

+ Description string

+ Inputs []*Input

+ Name string

+ Outputs []*Output

+ Path string

+ State map[string]any

+ Steps []*Step

+ type Output struct

+ Description string

+ Name string

+ Path string

+ Variable string

+ type Patch struct

+ func GeneratePatches(original, modified map[string]any) []Patch

+ func NewPatch(opts PatchOptions) Patch

+ func (p Patch) Delete() bool

+ func (p Patch) Value() any

+ func (p Patch) Variable() string

+ type PatchOptions struct

+ Delete bool

+ Value any

+ Variable string

+ type Path struct

+ func NewPath(id string, step *Step, opts PathOptions) *Path

+ func (p *Path) CurrentStep() *Step

+ func (p *Path) ID() string

+ func (p *Path) Run(ctx context.Context) error

+ func (p *Path) Variables() map[string]any

+ type PathExecutionEvent struct

+ CurrentStep string

+ Duration time.Duration

+ EndTime time.Time

+ Error error

+ ExecutionID string

+ PathID string

+ StartTime time.Time

+ Status ExecutionStatus

+ StepOutputs map[string]any

+ WorkflowName string

+ type PathLocalState struct

+ func NewPathLocalState(inputs, variables map[string]any) *PathLocalState

+ func (s *PathLocalState) DeleteVariable(key string)

+ func (s *PathLocalState) GetInput(key string) (any, bool)

+ func (s *PathLocalState) GetVariable(key string) (any, bool)

+ func (s *PathLocalState) ListInputs() []string

+ func (s *PathLocalState) ListVariables() []string

+ func (s *PathLocalState) SetVariable(key string, value any)

+ type PathOptions struct

+ ActivityExecutor ActivityExecutor

+ ActivityRegistry map[string]Activity

+ ExecutionCallbacks ExecutionCallbacks

+ Formatter WorkflowFormatter

+ Inputs map[string]any

+ Logger *slog.Logger

+ ScriptCompiler script.Compiler

+ UpdatesChannel chan<- PathSnapshot

+ Variables map[string]any

+ Workflow *Workflow

+ type PathSnapshot struct

+ EndTime time.Time

+ Error error

+ JoinRequest *JoinRequest

+ NewPaths []PathSpec

+ PathID string

+ StartTime time.Time

+ Status ExecutionStatus

+ StepName string

+ StepOutput any

+ Timestamp time.Time

+ type PathSpec struct

+ Name string

+ Step *Step

+ Variables map[string]any

+ type PathState struct

+ CurrentStep string

+ EndTime time.Time

+ ErrorMessage string

+ ID string

+ StartTime time.Time

+ Status ExecutionStatus

+ StepOutputs map[string]any

+ Variables map[string]any

+ func (p *PathState) Copy() *PathState

+ type RetryConfig struct

+ BackoffRate float64

+ BaseDelay time.Duration

+ ErrorEquals []string

+ JitterStrategy JitterStrategy

+ MaxDelay time.Duration

+ MaxRetries int

+ Timeout time.Duration

+ type Step struct

+ Activity string

+ Catch []*CatchConfig

+ Description string

+ Each *Each

+ EdgeMatchingStrategy EdgeMatchingStrategy

+ Join *JoinConfig

+ Name string

+ Next []*Edge

+ Parameters map[string]any

+ Retry []*RetryConfig

+ Store string

+ func (s *Step) GetEdgeMatchingStrategy() EdgeMatchingStrategy

+ type TypedActivity interface

+ Execute func(ctx Context, parameters TParams) (TResult, error)

+ Name func() string

+ type TypedActivityAdapter struct

+ func (a *TypedActivityAdapter[TParams, TResult]) Activity() TypedActivity[TParams, TResult]

+ func (a *TypedActivityAdapter[TParams, TResult]) Execute(ctx Context, parameters map[string]any) (any, error)

+ func (a *TypedActivityAdapter[TParams, TResult]) Name() string

+ type TypedActivityFunction struct

+ func (t *TypedActivityFunction[TParams, TResult]) Execute(ctx Context, params TParams) (TResult, error)

+ func (t *TypedActivityFunction[TParams, TResult]) Name() string

+ func (t *TypedActivityFunction[TParams, TResult]) ParametersType() reflect.Type

+ func (t *TypedActivityFunction[TParams, TResult]) ResultType() reflect.Type

+ type VariableContainer interface

+ DeleteVariable func(key string)

+ GetVariable func(key string) (value any, exists bool)

+ ListVariables func() []string

+ SetVariable func(key string, value any)

+ type Workflow struct

+ func LoadFile(path string) (*Workflow, error)

+ func LoadString(data string) (*Workflow, error)

+ func New(opts Options) (*Workflow, error)

+ func (w *Workflow) Description() string

+ func (w *Workflow) GetStep(name string) (*Step, bool)

+ func (w *Workflow) InitialState() map[string]any

+ func (w *Workflow) Inputs() []*Input

+ func (w *Workflow) Name() string

+ func (w *Workflow) Outputs() []*Output

+ func (w *Workflow) Path() string

+ func (w *Workflow) Start() *Step

+ func (w *Workflow) StepNames() []string

+ func (w *Workflow) Steps() []*Step

+ type WorkflowError struct

+ Cause string

+ Details interface{}

+ Type string

+ Wrapped error

+ func ClassifyError(err error) *WorkflowError

+ func NewWorkflowError(errorType, cause string) *WorkflowError

+ func (e *WorkflowError) Error() string

+ func (e *WorkflowError) ToErrorOutput() ErrorOutput

+ func (e *WorkflowError) Unwrap() error

+ type WorkflowExecutionEvent struct

+ Duration time.Duration

+ EndTime time.Time

+ Error error

+ ExecutionID string

+ Inputs map[string]any

+ Outputs map[string]any

+ PathCount int

+ StartTime time.Time

+ Status ExecutionStatus

+ WorkflowName string

+ type WorkflowFormatter interface

+ PrintStepError func(stepName string, err error)

+ PrintStepOutput func(stepName string, content any)

+ PrintStepStart func(stepName string, activityName string)

+ type WorkflowRegistry interface

+ Get func(name string) (*Workflow, bool)

+ List func() []string

+ Register func(workflow *Workflow) error

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL