Documentation
¶
Overview ¶
Package workflowkit provides host-side workflow orchestration contracts for composing ordered steps, approvals, retries, resumable state, and references to host-owned artifacts or audit records.
The package is intentionally not an agent runtime. Agent execution, OCR, context compression, durable payload storage, queues, HTTP APIs, and UI layers belong in host applications or sibling modules.
Index ¶
- Variables
- func IsTransient(err error) bool
- type Approval
- type Executor
- func (e *Executor) Approve(ctx context.Context, id string, approval Approval) (WorkflowRun, error)
- func (e *Executor) Cancel(ctx context.Context, id string, reason string) (WorkflowRun, error)
- func (e *Executor) Continue(ctx context.Context, id string) (WorkflowRun, error)
- func (e *Executor) Run(ctx context.Context, run WorkflowRun) (WorkflowRun, error)
- type InvalidStatusError
- type InvalidTransitionError
- type MemoryStore
- func (s *MemoryStore) ClaimRunnable(ctx context.Context, workerID string, lease time.Duration) (WorkflowRun, error)
- func (s *MemoryStore) ExtendLease(ctx context.Context, id string, workerID string, lease time.Duration) (WorkflowRun, error)
- func (s *MemoryStore) FailExpiredRunning(ctx context.Context, reason string) ([]WorkflowRun, error)
- func (s *MemoryStore) Get(ctx context.Context, id string) (WorkflowRun, error)
- func (s *MemoryStore) ListWorkflows(ctx context.Context, query WorkflowQuery) ([]WorkflowRun, error)
- func (s *MemoryStore) ReleaseLease(ctx context.Context, id string, workerID string) (WorkflowRun, error)
- func (s *MemoryStore) Save(ctx context.Context, run WorkflowRun) error
- func (s *MemoryStore) Update(ctx context.Context, id string, mutate func(WorkflowRun) (WorkflowRun, error)) (WorkflowRun, error)
- type Option
- type QueueLeaseStore
- type QueueRecoveryStore
- type QueueStore
- type RetryPolicy
- type Status
- type Step
- type StepRecord
- type StepResult
- type Store
- type TransientError
- type WorkflowOrder
- type WorkflowQuery
- type WorkflowQueryStore
- type WorkflowRun
Constants ¶
This section is empty.
Variables ¶
var ( ErrInvalidTransition = errors.New("invalid workflow status transition") ErrInvalidStatus = errors.New("invalid workflow status") // ErrRunPreempted reports that a running workflow was cancelled, failed, or // claimed by another lease owner, so the executor stopped writing to it. ErrRunPreempted = errors.New("workflow run preempted") )
var ErrNoRunnableWorkflow = errors.New("no runnable workflow")
var ErrRunNotFound = errors.New("workflow run not found")
var ErrWorkflowLeaseNotOwned = errors.New("workflow lease not owned")
Functions ¶
func IsTransient ¶
Types ¶
type Executor ¶
type Executor struct {
// contains filtered or unexported fields
}
Executor runs workflow steps in order and persists progress after each one. Every write is guarded: it only applies while the stored run is still running under the executor's lease owner, so a cancelled, failed, or reclaimed run returns ErrRunPreempted instead of being overwritten.
With a RetryPolicy, a run claimed from a queue does not hold its worker during a backoff: Run returns it in StatusPending with NextAttemptAt set, the worker releases its lease as usual, and ClaimRunnable hands it out again once the retry is due.
If ctx is cancelled, the executor stops without recording a terminal state; the caller owns finalization. Queued runs left running this way are failed by QueueRecoveryStore.FailExpiredRunning once their lease expires.
func (*Executor) Cancel ¶
Cancel marks a non-terminal run cancelled. An executor still running it stops at its next state write instead of overwriting the cancellation.
func (*Executor) Run ¶
func (e *Executor) Run(ctx context.Context, run WorkflowRun) (WorkflowRun, error)
type InvalidStatusError ¶
func (InvalidStatusError) Error ¶
func (e InvalidStatusError) Error() string
func (InvalidStatusError) Unwrap ¶
func (e InvalidStatusError) Unwrap() error
type InvalidTransitionError ¶
func (InvalidTransitionError) Error ¶
func (e InvalidTransitionError) Error() string
func (InvalidTransitionError) Unwrap ¶
func (e InvalidTransitionError) Unwrap() error
type MemoryStore ¶
type MemoryStore struct {
// contains filtered or unexported fields
}
func NewMemoryStore ¶
func NewMemoryStore() *MemoryStore
func (*MemoryStore) ClaimRunnable ¶
func (s *MemoryStore) ClaimRunnable(ctx context.Context, workerID string, lease time.Duration) (WorkflowRun, error)
func (*MemoryStore) ExtendLease ¶
func (s *MemoryStore) ExtendLease(ctx context.Context, id string, workerID string, lease time.Duration) (WorkflowRun, error)
func (*MemoryStore) FailExpiredRunning ¶
func (s *MemoryStore) FailExpiredRunning(ctx context.Context, reason string) ([]WorkflowRun, error)
func (*MemoryStore) Get ¶
func (s *MemoryStore) Get(ctx context.Context, id string) (WorkflowRun, error)
func (*MemoryStore) ListWorkflows ¶
func (s *MemoryStore) ListWorkflows(ctx context.Context, query WorkflowQuery) ([]WorkflowRun, error)
func (*MemoryStore) ReleaseLease ¶
func (s *MemoryStore) ReleaseLease(ctx context.Context, id string, workerID string) (WorkflowRun, error)
func (*MemoryStore) Save ¶
func (s *MemoryStore) Save(ctx context.Context, run WorkflowRun) error
func (*MemoryStore) Update ¶
func (s *MemoryStore) Update(ctx context.Context, id string, mutate func(WorkflowRun) (WorkflowRun, error)) (WorkflowRun, error)
type Option ¶
type Option func(*Executor)
func WithRetryPolicy ¶
func WithRetryPolicy(policy RetryPolicy) Option
type QueueLeaseStore ¶
type QueueLeaseStore interface {
QueueStore
ExtendLease(ctx context.Context, id string, workerID string, lease time.Duration) (WorkflowRun, error)
ReleaseLease(ctx context.Context, id string, workerID string) (WorkflowRun, error)
}
type QueueRecoveryStore ¶
type QueueRecoveryStore interface {
QueueLeaseStore
FailExpiredRunning(ctx context.Context, reason string) ([]WorkflowRun, error)
}
QueueRecoveryStore fails running workflows whose lease expired, which means their worker stopped heartbeating. Recovery is fail-closed: the interrupted step may already have had side effects, so it is never re-executed automatically. A host can requeue the failed run after inspecting it. Runs without a lease owner are never touched.
type QueueStore ¶
type RetryPolicy ¶
type RetryPolicy struct {
// MaxAttempts is the total number of attempts per step, counting the
// first; <= 1 disables retries.
MaxAttempts int
BaseDelay time.Duration
// MaxDelay caps the backoff; it defaults to 32 x BaseDelay.
MaxDelay time.Duration
}
RetryPolicy retries a step that fails with a TransientError. Attempt n waits a random delay in [0, min(MaxDelay, BaseDelay*2^(n-1))] ("full jitter"), so retries back off exponentially without retrying in lockstep.
type Step ¶
type Step interface {
Name() string
Run(context.Context, WorkflowRun) (StepResult, error)
}
type StepRecord ¶
type StepResult ¶
type Store ¶
type Store interface {
Save(ctx context.Context, run WorkflowRun) error
Get(ctx context.Context, id string) (WorkflowRun, error)
Update(ctx context.Context, id string, mutate func(WorkflowRun) (WorkflowRun, error)) (WorkflowRun, error)
}
type TransientError ¶
type TransientError struct {
Err error
}
func (TransientError) Error ¶
func (e TransientError) Error() string
func (TransientError) Unwrap ¶
func (e TransientError) Unwrap() error
type WorkflowOrder ¶
type WorkflowOrder string
const ( WorkflowOrderAsc WorkflowOrder = "asc" WorkflowOrderDesc WorkflowOrder = "desc" )
func (WorkflowOrder) IsValid ¶
func (o WorkflowOrder) IsValid() bool
type WorkflowQuery ¶
type WorkflowQuery struct {
Status Status
MetadataEquals map[string]string
Order WorkflowOrder
Limit int
}
type WorkflowQueryStore ¶
type WorkflowQueryStore interface {
Store
ListWorkflows(ctx context.Context, query WorkflowQuery) ([]WorkflowRun, error)
}
type WorkflowRun ¶
type WorkflowRun struct {
ID string
Status Status
InputRef string
OutputRef string
AgentRunID string
AuditRef string
Error string
ApprovalRef string
WaitingReason string
CurrentStep string
CompletedSteps []string
StepAttempts map[string]int
StepRecords []StepRecord
// StepRetries counts consecutive transient failures of CurrentStep. It
// carries the retry budget across claims and resets when the step
// succeeds or the run fails, so a requeued run starts a fresh budget.
StepRetries int
// NextAttemptAt holds a pending run back from ClaimRunnable until a
// scheduled retry is due.
NextAttemptAt time.Time
Metadata map[string]any
LeaseOwner string
LeaseUntil time.Time
CreatedAt time.Time
UpdatedAt time.Time
}
Directories
¶
| Path | Synopsis |
|---|---|
|
Package agentstep adapts goagent RunDetailed calls into workflowkit steps.
|
Package agentstep adapts goagent RunDetailed calls into workflowkit steps. |
|
examples
|
|
|
agent-approval
command
|
|
|
basic
command
|
|
|
ocr-review
command
|
|
|
sqlite-resume
command
|
|
|
Package sqlitestore implements workflowkit.Store with SQLite persistence.
|
Package sqlitestore implements workflowkit.Store with SQLite persistence. |
|
Package storetest provides conformance tests for workflowkit.Store implementations.
|
Package storetest provides conformance tests for workflowkit.Store implementations. |