workflow

package module

Go to main page

Versions in this module

v0

v0.20.0

Jul 19, 2026

Changes in this version

+ func BuildWFStatusOutput(wf *Workflow) wfStatusOutput

type Workflow

+ func (w *Workflow) ResumableArtifact() (resumable bool, artifactPath string)

v0.19.2

Jul 17, 2026

v0.19.1

Jul 17, 2026

v0.19.0

Jul 17, 2026

v0.18.0

Jul 17, 2026

v0.17.2

Jul 17, 2026

Changes in this version

type Workflow

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

v0.17.1

Jul 16, 2026

v0.17.0

Jul 16, 2026

Changes in this version

type Engine

+ func (e *Engine) Reopen(workflowID string) error

v0.16.1

Jun 21, 2026

v0.16.0

Jun 21, 2026

Changes in this version

+ const EventWorkflowApprovalNeeded

+ const EventWorkflowCancelled

+ const EventWorkflowCompleted

+ const EventWorkflowFailed

+ const EventWorkflowStarted

+ const EventWorkflowStepCompleted

+ const EventWorkflowStepFailed

+ const EventWorkflowStepStarted

+ const OnErrorContinue

+ const OnErrorFail

+ const OnErrorSkip

+ const ParamTypeArray

+ const ParamTypeBool

+ const ParamTypeFloat

+ const ParamTypeInt

+ const ParamTypeObject

+ const ParamTypeString

+ const ToolHTTPRequest

+ var DefaultCostModel = map[string]ModelPrice

+ var ErrBudgetExceeded = errors.New("workflow budget exceeded")

+ var GlobalMetrics = NewMetrics()

+ func EstimateUSD(model string, inputTokens, outputTokens int64, costModel map[string]ModelPrice) float64

+ func IsTransientError(errMsg string) bool

+ func IsValidStepKind(kind StepKind) bool

+ func MaskSecrets(text string) string

+ func MaskSecretsInMap(m map[string]any) map[string]any

+ func MatchesFilter(filter map[string]string, data map[string]any) bool

+ func NextCronRun(expr, tz string, nowMS int64) (int64, error)

+ func ParseOwner(owner string) (channel, chatID string)

+ func PrometheusHandler(m *Metrics) http.Handler

+ func RegisterMCPTools(server *mcp.Server, deps MCPDeps) int

+ func ResolveRefs(s string, wf *Workflow) string

+ func ResolveRefsErr(s string, wf *Workflow) (string, error)

+ func ResolveSecretRef(s string) string

+ type A2ACaller interface

+ Call func(ctx context.Context, agentID, message string) (string, error)

+ type A2AExecutor struct

+ func NewA2AExecutor(caller A2ACaller, metrics *Metrics) *A2AExecutor

+ func (e *A2AExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

+ type AgentExecutor struct

+ func NewAgentExecutor(runner AgentRunner, metrics *Metrics) *AgentExecutor

+ func (e *AgentExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

+ type AgentRunOpts struct

+ MaxIterations int

+ Model string

+ SkipContext bool

+ TimeoutSeconds int

+ type AgentRunner interface

+ RunTask func(ctx context.Context, task string, sessionKey string, opts AgentRunOpts) (string, error)

+ type ApprovalExecutor struct

+ func NewApprovalExecutor() *ApprovalExecutor

+ func (e *ApprovalExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

+ type ApprovalNotifier func(wf *Workflow, step *Step)

+ type BranchAllExecutor struct

+ func NewBranchAllExecutor(engine *Engine) *BranchAllExecutor

+ func (e *BranchAllExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

+ type CompletionNotifier func(wf *Workflow)

+ type ConcurrencyLimit struct

+ CurrentCount int

+ Key string

+ MaxConcurrent int

+ type ConditionExecutor struct

+ func NewConditionExecutor() *ConditionExecutor

+ func (e *ConditionExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

+ type Engine struct

+ func NewEngine(store *WorkflowStore, opts ...EngineOption) *Engine

+ func (e *Engine) Advance(ctx context.Context, workflowID string) (bool, error)

+ func (e *Engine) AutoRetryFailed(maxAge time.Duration) int

+ func (e *Engine) Cancel(workflowID string) error

+ func (e *Engine) DetectStalled(threshold time.Duration) []StalledWorkflow

+ func (e *Engine) GracefulShutdown(timeout time.Duration) int

+ func (e *Engine) HandleApproval(workflowID string, approved bool) error

+ func (e *Engine) HandleApprovalWithData(workflowID string, approved bool, data map[string]any) error

+ func (e *Engine) InjectStepsAndRewriteDeps(workflowID string, steps []Step, afterStepID, newDepID string) error

+ func (e *Engine) ListenForResults(ctx context.Context, listener *StepListener)

+ func (e *Engine) PauseAll() int

+ func (e *Engine) RecoverAll(ctx context.Context) []string

+ func (e *Engine) RecoverStalled(threshold time.Duration) int

+ func (e *Engine) RegisterWebhooks(mux *http.ServeMux, runtime *TemplateRuntime, triggers []WebhookTrigger) error

+ func (e *Engine) ResumeAll(ctx context.Context) []string

+ func (e *Engine) ResumeAsync(_ context.Context, workflowID string)

+ func (e *Engine) RunStep(ctx context.Context, workflowID, stepID string) error

+ func (e *Engine) RunToCompletion(ctx context.Context, workflowID string) error

+ func (e *Engine) SetA2ACaller(caller A2ACaller)

+ func (e *Engine) SetAgentRunner(runner AgentRunner)

+ func (e *Engine) SetApprovalNotifier(fn ApprovalNotifier)

+ func (e *Engine) SetCompletionNotifier(fn CompletionNotifier)

+ func (e *Engine) SetHooks(h HookPublisher)

+ func (e *Engine) SetSkills(sr SkillResolver)

+ func (e *Engine) Start(ctx context.Context, workflowID string) error

+ func (e *Engine) StartAsync(ctx context.Context, workflowID string) error

+ func (e *Engine) StartWatchdog(interval time.Duration)

+ func (e *Engine) StopWatchdog()

+ func (e *Engine) Store() *WorkflowStore

+ func (e *Engine) ValidateTemplate(t *Template) error

+ func (e *Engine) ValidateWorkflow(wf *Workflow) []ValidationError

+ type EngineOption func(*Engine)

+ func WithA2ACaller(c A2ACaller) EngineOption

+ func WithAgentRunner(a AgentRunner) EngineOption

+ func WithApprovalNotifier(fn ApprovalNotifier) EngineOption

+ func WithBudget(maxUSD float64) EngineOption

+ func WithCompletionNotifier(fn CompletionNotifier) EngineOption

+ func WithCostModel(model map[string]ModelPrice) EngineOption

+ func WithDispatcher(d StepDispatcher) EngineOption

+ func WithEventLog(el *EventLog) EngineOption

+ func WithHookPublisher(h HookPublisher) EngineOption

+ func WithImageRenderer(r ImageRenderer) EngineOption

+ func WithImageWorkspace(dir string) EngineOption

+ func WithLLMClient(c *llm.Client) EngineOption

+ func WithLLMProvider(p LLMProvider) EngineOption

+ func WithLogger(l *slog.Logger) EngineOption

+ func WithMCPServerHeaders(serverID string, headers map[string]string) EngineOption

+ func WithMCPServers(servers map[string]string) EngineOption

+ func WithMessagePublisher(m MessagePublisher) EngineOption

+ func WithMetrics(m *Metrics) EngineOption

+ func WithRateLimit(provider string, rate float64, burst int) EngineOption

+ func WithScheduler(s *Scheduler) EngineOption

+ func WithSkillResolver(s SkillResolver) EngineOption

+ func WithStepCache(cache StepCache) EngineOption

+ func WithStepCacheKinds(kinds ...StepKind) EngineOption

+ func WithStepListener(l *StepListener) EngineOption

+ func WithStreamCallback(cb StreamCallback) EngineOption

+ func WithToolRunner(t ToolRunner) EngineOption

+ func WithTracerProvider(tp trace.TracerProvider) EngineOption

+ func WithTriggers(ts *TriggerService) EngineOption

+ func WithVisionProvider(p LLMProvider) EngineOption

+ type Event struct

+ Data map[string]any

+ DurationMS int64

+ Error string

+ StepID string

+ StepKind string

+ Timestamp int64

+ Type EventType

+ WorkflowID string

+ func LoadEventLog(path string) ([]Event, error)

+ type EventLog struct

+ func NewEventLog(dir string) (*EventLog, error)

+ func (el *EventLog) Append(e Event) error

+ func (el *EventLog) Path(workflowID string) string

+ type EventTrigger struct

+ Action TriggerAction

+ CreatedAt int64

+ Enabled bool

+ Event string

+ Filter map[string]string

+ ID string

+ Name string

+ type EventType string

+ const EventStepFailed

+ const EventStepFinished

+ const EventStepRetried

+ const EventStepStarted

+ const EventWFCompleted

+ const EventWFFailed

+ const EventWFStarted

+ type ExecutionTrace struct

+ EndedAt int64

+ Error string

+ StartedAt int64

+ Steps []StepTrace

+ TotalMS int64

+ WorkflowID string

+ func ReplayTrace(events []Event) *ExecutionTrace

+ type FileCache struct

+ func NewFileCache(dir string) (*FileCache, error)

+ func (c *FileCache) Get(_ context.Context, key string) (StepCacheEntry, bool, error)

+ func (c *FileCache) Put(_ context.Context, key string, entry StepCacheEntry) error

+ type ForEachExecutor struct

+ func NewForEachExecutor(engine *Engine) *ForEachExecutor

+ func (e *ForEachExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

+ type HookPublisher interface

+ Fire func(event string, data map[string]any) int

+ type ImageExecutor struct

+ func NewImageExecutor(r ImageRenderer, metrics *Metrics) *ImageExecutor

+ func (e *ImageExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

+ type ImageRenderRequest struct

+ Density int

+ Fonts []string

+ Format string

+ HTML string

+ Height int

+ Width int

+ type ImageRenderResult struct

+ Bytes []byte

+ DurationMS int64

+ MIMEType string

+ Path string

+ SizeBytes int64

+ type ImageRenderer interface

+ Render func(ctx context.Context, req ImageRenderRequest) (ImageRenderResult, error)

+ type InMemoryCache struct

+ func NewInMemoryCache() *InMemoryCache

+ func (c *InMemoryCache) Get(_ context.Context, key string) (StepCacheEntry, bool, error)

+ func (c *InMemoryCache) Put(_ context.Context, key string, entry StepCacheEntry) error

+ type Job struct

+ CreatedAtMS int64

+ DeleteAfterRun bool

+ Enabled bool

+ ID string

+ Name string

+ Payload JobPayload

+ Schedule Schedule

+ State JobState

+ UpdatedAtMS int64

+ type JobHandler func(job *Job) (string, error)

+ type JobPayload struct

+ Kind string

+ Message string

+ Params map[string]any

+ TemplateID string

+ WorkflowID string

+ type JobState struct

+ LastError string

+ LastRunAtMS *int64

+ LastStatus string

+ NextRunAtMS *int64

+ type LLMExecutor struct

+ func NewLLMExecutor(provider LLMProvider, metrics *Metrics) *LLMExecutor

+ func NewLLMExecutorWithClient(client *llm.Client, metrics *Metrics) *LLMExecutor

+ func (e *LLMExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

+ func (e *LLMExecutor) SetSkills(sr SkillResolver)

+ func (e *LLMExecutor) SetStreamCallback(cb StreamCallback)

+ func (e *LLMExecutor) SetToolRunner(tr ToolRunner)

+ type LLMImageContent struct

+ Bytes []byte

+ MIMEType string

+ type LLMMessage struct

+ Content string

+ Images []LLMImageContent

+ Role string

+ type LLMProvider interface

+ Chat func(ctx context.Context, messages []LLMMessage, model string) (*LLMResponse, error)

+ GetDefaultModel func() string

+ type LLMResponse struct

+ Content string

+ InputTokens int

+ Model string

+ OutputTokens int

+ type LocalDispatcher struct

+ func NewLocalDispatcher(engine *Engine) *LocalDispatcher

+ func (d *LocalDispatcher) Dispatch(ctx context.Context, workflowID, stepID string, _ StepKind) error

+ func (d *LocalDispatcher) DispatchBatch(ctx context.Context, workflowID string, stepIDs []string, _ []StepKind) error

+ type MCPDeps struct

+ Engine *Engine

+ Templates *TemplateStore

+ type MCPToolRunner struct

+ func NewMCPToolRunner(servers map[string]string) *MCPToolRunner

+ func (r *MCPToolRunner) Close() error

+ func (r *MCPToolRunner) Connect(ctx context.Context) error

+ func (r *MCPToolRunner) Execute(ctx context.Context, name string, args map[string]any) (string, error)

+ func (r *MCPToolRunner) SetHeaders(serverID string, headers map[string]string)

+ func (r *MCPToolRunner) Tools() map[string][]string

+ type MessageExecutor struct

+ func NewMessageExecutor(bus MessagePublisher) *MessageExecutor

+ func (e *MessageExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

+ type MessagePublisher interface

+ PublishOutbound func(msg OutboundMessage)

+ type Metrics struct

+ A2AStepsExecuted atomic.Int64

+ A2AStepsFailed atomic.Int64

+ AgentStepsExecuted atomic.Int64

+ AgentStepsFailed atomic.Int64

+ ApprovalsPending atomic.Int64

+ HooksFired atomic.Int64

+ ImageBytesTotal atomic.Int64

+ ImageDurationMSTotal atomic.Int64

+ ImageRendersFailed atomic.Int64

+ ImageRendersSuccess atomic.Int64

+ LLMTokensInput atomic.Int64

+ LLMTokensOutput atomic.Int64

+ SchedulerJobsExecuted atomic.Int64

+ SchedulerJobsFailed atomic.Int64

+ StepCacheHits atomic.Int64

+ StepCacheMisses atomic.Int64

+ StepsDeadLettered atomic.Int64

+ StepsExecuted atomic.Int64

+ StepsRetried atomic.Int64

+ StepsSkipped atomic.Int64

+ TriggersEvaluated atomic.Int64

+ TriggersFired atomic.Int64

+ VisionCallsFailed atomic.Int64

+ VisionCallsSuccess atomic.Int64

+ VisionTokensInput atomic.Int64

+ VisionTokensOutput atomic.Int64

+ WebhooksReceived atomic.Int64

+ WebhooksRejected atomic.Int64

+ WorkflowBudgetExceededTotal atomic.Int64

+ WorkflowCostUSDTotal atomic.Uint64

+ WorkflowImagesRenderedTotal atomic.Int64

+ WorkflowTokensInputTotal atomic.Int64

+ WorkflowTokensOutputTotal atomic.Int64

+ WorkflowsCancelled atomic.Int64

+ WorkflowsCompleted atomic.Int64

+ WorkflowsCreated atomic.Int64

+ WorkflowsFailed atomic.Int64

+ func NewMetrics() *Metrics

+ func (m *Metrics) Reset()

+ func (m *Metrics) Summary() string

+ type ModelPrice struct

+ InputUSDPer1K float64

+ OutputUSDPer1K float64

+ type MultiToolRunner struct

+ func NewMultiToolRunner(runners ...ToolRunner) *MultiToolRunner

+ func (m *MultiToolRunner) Execute(ctx context.Context, name string, args map[string]any) (string, error)

+ func (m *MultiToolRunner) Register(runner ToolRunner, tools ...string)

+ type N8nConnectionTarget struct

+ Index int

+ Node string

+ Type string

+ type N8nNode struct

+ ContinueOnFail bool

+ ID string

+ MaxTries int

+ Name string

+ Parameters map[string]any

+ Position [2]int

+ RetryOnFail bool

+ Type string

+ TypeVersion int

+ WaitBetweenTries int

+ type N8nNodeConnection struct

+ Main [][]N8nConnectionTarget

+ type N8nTag struct

+ Name string

+ type N8nWorkflow struct

+ Connections map[string]N8nNodeConnection

+ Name string

+ Nodes []N8nNode

+ Settings map[string]any

+ Tags []N8nTag

+ type NoopExecutor struct

+ func (e *NoopExecutor) Execute(_ context.Context, step *Step, _ *Workflow) error

+ type OutboundMessage struct

+ Channel string

+ ChatID string

+ Content string

+ type ParamSpec struct

+ Default any

+ Description string

+ Enum []any

+ Required bool

+ Type string

+ type ParamsMap map[string]ParamSpec

+ func (p *ParamsMap) UnmarshalJSON(data []byte) error

+ func (p ParamsMap) MarshalJSON() ([]byte, error)

+ type PostgresDispatcher struct

+ func NewPostgresDispatcher(queue StepEnqueuer, closers ...io.Closer) *PostgresDispatcher

+ func (d *PostgresDispatcher) Close() error

+ func (d *PostgresDispatcher) Dispatch(_ context.Context, workflowID, stepID string, kind StepKind) error

+ func (d *PostgresDispatcher) DispatchBatch(_ context.Context, workflowID string, stepIDs []string, kinds []StepKind) error

+ type QueueItem struct

+ ClaimedAt *time.Time

+ ConcurrencyKey string

+ CreatedAt time.Time

+ Error string

+ HeartbeatAt *time.Time

+ ID int64

+ Priority int

+ Result []byte

+ State QueueItemState

+ StepID string

+ StepKind string

+ WorkerID string

+ WorkflowID string

+ type QueueItemState string

+ const QueueClaimed

+ const QueueDone

+ const QueueFailed

+ const QueuePending

+ const QueueTimedOut

+ type Reaper struct

+ func NewReaper(reaper StepReaper, heartbeatTimeout, checkInterval time.Duration) *Reaper

+ func (r *Reaper) Logger() *slog.Logger

+ func (r *Reaper) Run(ctx context.Context)

+ type ReducerKind string

+ const ReducerAppend

+ const ReducerMerge

+ const ReducerReplace

+ const ReducerSum

+ type RetryAfterError interface

+ RetryAfter func() time.Duration

+ type Schedule struct

+ AtMS *int64

+ EveryMS *int64

+ Expr string

+ Kind string

+ TZ string

+ type Scheduler struct

+ func NewScheduler(storePath string, handler JobHandler, opts ...SchedulerOption) *Scheduler

+ func (s *Scheduler) AddJob(name string, schedule Schedule, payload JobPayload) (*Job, error)

+ func (s *Scheduler) EnableJob(id string, enabled bool) *Job

+ func (s *Scheduler) ListJobs(includeDisabled bool) []Job

+ func (s *Scheduler) RemoveJob(id string) bool

+ func (s *Scheduler) Start() error

+ func (s *Scheduler) Status() map[string]any

+ func (s *Scheduler) Stop()

+ type SchedulerOption func(*Scheduler)

+ func WithSchedulerLogger(l *slog.Logger) SchedulerOption

+ func WithSchedulerMetrics(m *Metrics) SchedulerOption

+ type SecurityPolicy struct

+ AllowNetwork bool

+ AllowShell bool

+ AllowedTools []string

+ DeniedTools []string

+ MaxDuration time.Duration

+ MaxStepDuration time.Duration

+ MaxSteps int

+ SecretPatterns []string

+ func DefaultSecurityPolicy() SecurityPolicy

+ func (p SecurityPolicy) IsToolAllowed(toolName string) bool

+ func (p SecurityPolicy) Validate() error

+ type SkillResolver interface

+ LoadSkill func(name string) (string, bool)

+ type StalledWorkflow struct

+ RunningFor time.Duration

+ StepID string

+ StepKind StepKind

+ WorkflowID string

+ WorkflowName string

+ type Step struct

+ AlwaysRun bool

+ Config map[string]any

+ DependsOn []string

+ EndedAt int64

+ Error string

+ ID string

+ Kind StepKind

+ Result any

+ Retries int

+ StartedAt int64

+ State StepState

+ func (s *Step) GetBackoffMultiplier() float64

+ func (s *Step) GetMaxDelayMS() int64

+ func (s *Step) GetOnError() string

+ func (s *Step) GetRetryDelayMS() int64

+ func (s *Step) GetRetryMax() int

+ func (s *Step) GetRetryOn() []string

+ func (s *Step) GetSkipOn() []string

+ func (s *Step) GetTimeoutMS() int64

+ type StepCache interface

+ Get func(ctx context.Context, key string) (StepCacheEntry, bool, error)

+ Put func(ctx context.Context, key string, entry StepCacheEntry) error

+ type StepCacheEntry struct

+ Context map[string]any

+ Cost *StepCost

+ Output any

+ StoredAt time.Time

+ TTL time.Duration

+ type StepCost struct

+ Bytes int64

+ DurationMS int64

+ InputTokens int64

+ Kind StepKind

+ Model string

+ OutputTokens int64

+ StepID string

+ USDEstimate float64

+ type StepDispatcher interface

+ Dispatch func(ctx context.Context, workflowID, stepID string, kind StepKind) error

+ DispatchBatch func(ctx context.Context, workflowID string, stepIDs []string, kinds []StepKind) error

+ type StepDoneEvent struct

+ StepID string

+ WorkflowID string

+ type StepEnqueuer interface

+ Enqueue func(item QueueItem) error

+ type StepExecutor interface

+ Execute func(ctx context.Context, step *Step, wf *Workflow) error

+ type StepKind string

+ const StepA2A

+ const StepAgent

+ const StepApproval

+ const StepBranchAll

+ const StepCondition

+ const StepForEach

+ const StepImage

+ const StepLLM

+ const StepMessage

+ const StepNoop

+ const StepSuspend

+ const StepTool

+ const StepTransform

+ const StepVision

+ const StepWorkflow

+ func NormalizeStepKind(kind StepKind) StepKind

+ type StepListener struct

+ func NewStepListener(dsn string) (*StepListener, error)

+ func (l *StepListener) Close() error

+ func (l *StepListener) Listen(ctx context.Context) <-chan StepDoneEvent

+ type StepReaper interface

+ ReapStale func(timeout time.Duration) (int, error)

+ type StepState string

+ const StepCompleted

+ const StepDeadLettered

+ const StepFailed

+ const StepPending

+ const StepRunning

+ const StepSkipped

+ type StepTrace struct

+ DurationMS int64

+ EndedAt int64

+ Error string

+ Retries int

+ StartedAt int64

+ StepID string

+ StepKind string

+ type StepWorkerQueue interface

+ Complete func(itemID int64, result []byte, errMsg string) error

+ Dequeue func(workerID string, kinds []string) (*QueueItem, bool)

+ Fail func(itemID int64, errMsg string) error

+ Heartbeat func(itemID int64) error

+ type StoreBackend interface

+ Close func() error

+ Delete func(id string) error

+ FindByIdempotencyKey func(key string) *Workflow

+ List func(state WorkflowState) []*Workflow

+ ListByOwner func(owner string) []*Workflow

+ Load func(id string) (*Workflow, bool)

+ Modify func(id string, fn func(w *Workflow)) error

+ Save func(w *Workflow) error

+ type StreamCallback func(workflowID, stepID, delta string)

+ type SubWorkflowExecutor struct

+ func NewSubWorkflowExecutor(runner SubWorkflowRunner) *SubWorkflowExecutor

+ func (e *SubWorkflowExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

+ type SubWorkflowRunner interface

+ RunToCompletion func(ctx context.Context, workflowID string) error

+ Start func(ctx context.Context, workflowID string) error

+ Store func() *WorkflowStore

+ type SuspendExecutor struct

+ func NewSuspendExecutor() *SuspendExecutor

+ func (e *SuspendExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

+ type Template struct

+ Defaults map[string]any

+ Description string

+ Name string

+ Params ParamsMap

+ Steps []TemplateStep

+ func ConvertN8nFile(path string) (*Template, error)

+ func ConvertN8nToTemplate(data []byte) (*Template, error)

+ func ParseTemplate(data []byte) (*Template, error)

+ type TemplateMeta struct

+ Description string

+ Name string

+ Params ParamsMap

+ Source string

+ type TemplateRuntime struct

+ func NewTemplateRuntime(dir string) *TemplateRuntime

+ func (tr *TemplateRuntime) Dir() string

+ func (tr *TemplateRuntime) Instantiate(templateName, workflowID, owner string, params map[string]any) (*Workflow, error)

+ func (tr *TemplateRuntime) List() []TemplateMeta

+ func (tr *TemplateRuntime) Reload()

+ type TemplateStep struct

+ Config json.RawMessage

+ DependsOn []string

+ ID string

+ Kind StepKind

+ OnError string

+ Retry json.RawMessage

+ type TemplateStore struct

+ func NewTemplateStore(dir string) *TemplateStore

+ func (ts *TemplateStore) Get(name string) (*Template, bool)

+ func (ts *TemplateStore) Instantiate(templateName, workflowID, owner string, params map[string]any) (*Workflow, error)

+ func (ts *TemplateStore) List() []string

+ func (ts *TemplateStore) Reload()

+ type ToolExecutor struct

+ func NewToolExecutor(runner ToolRunner) *ToolExecutor

+ func (e *ToolExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

+ type ToolRunner interface

+ Execute func(ctx context.Context, name string, args map[string]any) (string, error)

+ type TransformExecutor struct

+ func NewTransformExecutor() *TransformExecutor

+ func (e *TransformExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

+ type TriggerAction struct

+ Channel string

+ Kind string

+ Message string

+ TemplateID string

+ To string

+ WorkflowID string

+ type TriggerExecutor func(trigger *EventTrigger) error

+ type TriggerOption func(*TriggerService)

+ func WithTriggerLogger(l *slog.Logger) TriggerOption

+ func WithTriggerMetrics(m *Metrics) TriggerOption

+ type TriggerService struct

+ func NewTriggerService(storePath string, opts ...TriggerOption) *TriggerService

+ func (ts *TriggerService) AddTrigger(name, event string, filter map[string]string, action TriggerAction) (*EventTrigger, error)

+ func (ts *TriggerService) EnableTrigger(id string, enabled bool) bool

+ func (ts *TriggerService) Evaluate(event string, data map[string]any) []EventTrigger

+ func (ts *TriggerService) HookHandler(event string) func(data map[string]any) error

+ func (ts *TriggerService) ListTriggers() []EventTrigger

+ func (ts *TriggerService) RegisterHooks(hookRegistrar func(event string, fn func(map[string]any) error))

+ func (ts *TriggerService) RemoveTrigger(id string) bool

+ func (ts *TriggerService) SetExecutor(fn TriggerExecutor)

+ type ValidationError struct

+ Field string

+ Message string

+ StepID string

+ func (ve ValidationError) Error() string

+ type VisionCapable interface

+ SupportsVision func() bool

+ type VisionExecutor struct

+ func NewVisionExecutor(p LLMProvider, metrics *Metrics) *VisionExecutor

+ func (e *VisionExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

+ type WebhookAuth int

+ const WebhookAuthBearer

+ const WebhookAuthHMAC

+ const WebhookAuthNone

+ type WebhookTrigger struct

+ AuthMode WebhookAuth

+ Owner string

+ Path string

+ Secret string

+ SignHeader string

+ Template string

+ VarMapper func(*http.Request, []byte) (map[string]any, error)

+ type WorkerConfig struct

+ Engine *Engine

+ HeartbeatInterval time.Duration

+ ID string

+ Logger *slog.Logger

+ PollInterval time.Duration

+ Queue StepWorkerQueue

+ StepKinds []string

+ type WorkerNode struct

+ func NewWorkerNode(cfg WorkerConfig) (*WorkerNode, error)

+ func (w *WorkerNode) DrainAndStop(timeout time.Duration)

+ func (w *WorkerNode) ProcessOne(ctx context.Context) bool

+ func (w *WorkerNode) Run(ctx context.Context)

+ func (w *WorkerNode) Stop()

+ type Workflow struct

+ AllowedTools []string

+ Context map[string]any

+ Cost *WorkflowCost

+ CreatedAt int64

+ CurrentStep string

+ Description string

+ Error string

+ ID string

+ IdempotencyKey string

+ InterruptAfter []string

+ InterruptBefore []string

+ Name string

+ Owner string

+ Reducers map[string]ReducerKind

+ Security *SecurityPolicy

+ State WorkflowState

+ Steps []Step

+ StepsExecuted int

+ TemplateName string

+ UpdatedAt int64

+ func NewWorkflow(id, name, owner string, steps []Step) *Workflow

+ func (w *Workflow) AddCost(c StepCost)

+ func (w *Workflow) Clone() *Workflow

+ func (w *Workflow) GetStep(stepID string) *Step

+ func (w *Workflow) IsTerminal() bool

+ type WorkflowCost struct

+ BySteps map[string]StepCost

+ BytesRendered int64

+ ImagesRendered int64

+ InputTokens int64

+ OutputTokens int64

+ USDEstimate float64

+ UpdatedAt time.Time

+ type WorkflowState string

+ const StateCancelled

+ const StateCompleted

+ const StateFailed

+ const StatePaused

+ const StatePending

+ const StateRunning

+ const StateWaitingApproval

+ type WorkflowStore struct

+ func NewWorkflowStore(backend StoreBackend) *WorkflowStore

+ func (s *WorkflowStore) Close() error

+ func (s *WorkflowStore) Delete(id string) error

+ func (s *WorkflowStore) FindByIdempotencyKey(key string) *Workflow

+ func (s *WorkflowStore) List(state WorkflowState) []*Workflow

+ func (s *WorkflowStore) ListByOwner(owner string) []*Workflow

+ func (s *WorkflowStore) Load(id string) (*Workflow, bool)

+ func (s *WorkflowStore) Modify(id string, fn func(w *Workflow)) error

+ func (s *WorkflowStore) Save(w *Workflow) error

+ func (s *WorkflowStore) String() string

v0.15.0

May 31, 2026

v0.14.0

May 20, 2026

v0.13.1

May 6, 2026

v0.13.0

May 6, 2026

v0.12.0

Apr 26, 2026

v0.11.1

Apr 25, 2026

v0.11.0

Apr 25, 2026

v0.10.0

Apr 25, 2026

v0.9.5

Apr 25, 2026

v0.9.4

Apr 25, 2026

v0.9.3

Apr 24, 2026

v0.9.2

Mar 23, 2026

v0.9.1

Mar 23, 2026

v0.9.0

Mar 22, 2026

v0.8.1

Mar 3, 2026

v0.8.0

Mar 3, 2026

v0.7.0

Mar 3, 2026

v0.3.0

Mar 1, 2026

v0.2.0

Mar 1, 2026

v0.1.0

Mar 1, 2026

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