workflow

Jul 21, 2026

Changes in this version

+ func ApplyInputDefaults(declared []types.WorkflowInputDef, input types.KV) types.KV

+ func ExportYAML(wf *types.WorkflowMetadata) ([]byte, error)

+ func InputsToMaps(inputs []types.WorkflowInputDef) []map[string]any

+ func MetadataFromRows(rows WorkflowRows) (*types.WorkflowMetadata, error)

+ func ReloadTriggers(ctx context.Context) error

+ func RetryToMap(r *types.RetryConfig) map[string]any

+ func SetReloadService(svc *Service)

+ func ValidateInputs(declared []types.WorkflowInputDef, input types.KV) error

+ type Catalog interface

+ ApplyDefinition func(ctx context.Context, meta *types.WorkflowMetadata) (*gen.Workflow, error)

+ DeleteDefinitionByName func(ctx context.Context, name string) error

+ ListDefinitions func(ctx context.Context) ([]*gen.Workflow, error)

+ ListRunsByName func(ctx context.Context, name string) ([]*gen.WorkflowRun, error)

+ type DefinitionStore interface

+ GetMetadata func(ctx context.Context, name string) (*types.WorkflowMetadata, error)

type Runner

+ func (r *Runner) WithDefinitionStore(ds DefinitionStore) *Runner

+ func (r *Runner) WithExistingRun(run *gen.WorkflowRun) *Runner

+ func (r *Runner) WithWorkflowID(id int64) *Runner

+ type Service struct

+ func ActiveService() *Service

+ func NewService(catalog Catalog, runs WorkflowRunStore, auditor audit.Auditor, ...) *Service

+ func (s *Service) ApplyYAML(ctx context.Context, data []byte) (*gen.Workflow, error)

+ func (s *Service) Delete(ctx context.Context, name string) error

+ func (s *Service) Export(ctx context.Context, name string) ([]byte, error)

+ func (s *Service) Get(ctx context.Context, name string) (*types.WorkflowMetadata, error)

+ func (s *Service) List(ctx context.Context) ([]*gen.Workflow, error)

+ func (s *Service) ListRuns(ctx context.Context, name string) ([]*gen.WorkflowRun, error)

+ func (s *Service) LookupWebhook(path string) (*WebhookEndpoint, bool)

+ func (s *Service) ReloadTriggers(ctx context.Context) error

+ func (s *Service) StartRunAsync(ctx context.Context, name, triggerType string, input types.KV) (int64, error)

+ func (s *Service) Stop()

+ func (s *Service) WebhookConfigs() map[string]*WebhookEndpoint

+ type WebhookEndpoint struct

+ Config *pipeline.WebhookConfig

+ WorkflowName string

+ type WorkflowRows struct

+ Tasks []*gen.WorkflowTask

+ Triggers []*gen.WorkflowTrigger

+ Workflow *gen.Workflow

Jun 6, 2026

Changes in this version

+ func DetermineRuntimeType(t *types.Task) string

+ func LoadFile(path string) (*types.WorkflowMetadata, error)

+ func ParseYAML(data []byte) (*types.WorkflowMetadata, error)

+ func ValidateDAG(tasks []types.WorkflowTask) error

+ func WorkflowTaskToTask(wt types.WorkflowTask) (*types.Task, error)

+ type ActionInfo struct

+ CapType string

+ Details string

+ IsCapability bool

+ Operation string

+ Type string

+ func ParseAction(action string) ActionInfo

+ type CheckpointData struct

+ CompletedTasks map[string]bool

+ HeartbeatAt time.Time

+ Input types.KV

+ StepIndex int

+ StepResults map[string]string

+ type Runner struct

+ func NewRunner() *Runner

+ func NewRunnerWithStore(store WorkflowRunStore, auditor audit.Auditor, wc *metrics.WorkflowCollector, ...) *Runner

+ func (r *Runner) Close() error

+ func (r *Runner) Execute(ctx context.Context, wf types.WorkflowMetadata, input types.KV, file string) error

+ func (r *Runner) ResumeWorkflow(ctx context.Context, runID int64) error

+ func (r *Runner) Run(ctx context.Context, t *types.Task) error

+ type WorkflowRunStore interface

+ CreateRun func(ctx context.Context, workflowName, workflowFile, triggerType string, ...) (*gen.WorkflowRun, error)

+ CreateStepRun func(ctx context.Context, runID int64, stepID, stepName, action, actionType string, ...) (*gen.WorkflowStepRun, error)

+ GetCheckpoint func(ctx context.Context, runID int64, target any) error

+ GetIncompleteRuns func(ctx context.Context) ([]*gen.WorkflowRun, error)

+ GetRun func(ctx context.Context, runID int64) (*gen.WorkflowRun, error)

+ SaveCheckpoint func(ctx context.Context, runID int64, data any) error

+ UpdateRunHeartbeat func(ctx context.Context, runID int64) error

+ UpdateRunStatus func(ctx context.Context, runID int64, status int, errMsg string) error

+ UpdateStepRun func(ctx context.Context, stepRunID int64, status int, result map[string]any, ...) error