Help us shape the future of Go! Take the 2026 Go Developer Survey

workflowkit

package module
v0.1.1 Latest Latest
Warning

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

Go to latest
Published: Jul 19, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

README

WorkflowKit

workflowkit is a host-side orchestration module for long-running application flows with workflow state, approval waits, audit references, artifacts, and external capabilities.

It is not an agent runtime. goagent owns a single ReAct-style agent run; workflowkit owns the application lifecycle around one or more runs.

Install

go get github.com/eruca/goagents/workflowkit

For local development before publishing:

go mod edit -replace github.com/eruca/goagents/workflowkit=/Users/nick/VibeCoding/goagents/workflowkit
go get github.com/eruca/goagents/workflowkit

Boundary

Keep the dependency direction at the application boundary:

application
  imports github.com/eruca/goagents/workflowkit
  imports github.com/eruca/goagents/goagent
  imports github.com/eruca/goagents/workflowkit/agentstep when needed
  imports capability modules such as github.com/eruca/goagents/ocrs

github.com/eruca/goagents/goagent does not import github.com/eruca/goagents/workflowkit
github.com/eruca/goagents/workflowkit does not import github.com/eruca/goagents/goagent

workflowkit currently provides:

  • WorkflowRun and stable workflow statuses
  • Store and MemoryStore
  • ordered Step execution through Executor
  • step progress through CurrentStep and CompletedSteps
  • step history through StepRecords
  • Continue for resuming a persisted run after waiting_approval
  • AuditRef, InputRef, OutputRef, AgentRunID, and ApprovalRef for host-owned references
  • WorkflowQueryStore for host-owned workflow list/query views
  • QueueStore for worker claim proofs and QueueLeaseStore for host-owned lease lifecycle proofs
  • lifecycle guards for Run, Continue, and Cancel
  • explicit retry policy for transient step errors
  • sqlitestore for SQLite-backed persistence
  • storetest conformance suite for Store implementations

It intentionally does not provide DAG execution, distributed worker loops, cron, durable databases, UI, or multi-agent team orchestration.

Basic Workflow

store := workflowkit.NewMemoryStore()
executor := workflowkit.NewExecutor(store, []workflowkit.Step{
	prepareStep{},
	finalizeStep{},
})

run, err := executor.Run(ctx, workflowkit.WorkflowRun{
	ID:       "wf-1",
	Status:   workflowkit.StatusPending,
	InputRef: "artifact:input",
})

Steps return StepResult values. Returning waiting_approval, failed, or cancelled stops the executor and persists the current run. Returning a Go error marks the workflow failed.

When a step returns waiting_approval, include host-owned references instead of raw payloads:

return workflowkit.StepResult{
	Status:        workflowkit.StatusWaitingApproval,
	ApprovalRef:   "approval:req-1",
	WaitingReason: "operator approval required",
}

After the host records approval, call:

run, err = executor.Approve(ctx, "wf-1", workflowkit.Approval{
	AuditRef: "audit:approval-recorded",
	Metadata: map[string]any{
		"approved_by": "operator-1",
	},
})

Approve stores host-owned audit refs and bounded metadata, then continues the workflow from the next step. Use Continue directly only when the host has already updated the run itself.

Lifecycle

Workflow status transitions are guarded:

pending -> running
running -> waiting_approval
running -> succeeded
running -> failed
running -> cancelled
waiting_approval -> running
waiting_approval -> cancelled

Run accepts only pending or an empty status. Continue and Approve accept only waiting_approval. Cancel accepts pending, running, or waiting_approval. succeeded, failed, and cancelled are terminal.

Invalid transitions return InvalidTransitionError and match ErrInvalidTransition with errors.Is.

Step results may use only running, waiting_approval, succeeded, failed, or cancelled; an empty status means the step succeeded and the workflow should continue. pending is only an initial workflow status. Unknown step result statuses return InvalidStatusError, match ErrInvalidStatus with errors.Is, mark the workflow failed, and record the attempt as failed.

Steps that return failed or cancelled stop the workflow but are not added to CompletedSteps. Steps that return waiting_approval are added to CompletedSteps so Continue and Approve resume from the next step.

Retry

Retries are opt-in. By default, even transient errors are attempted once.

Configure retry policy on the executor:

executor := workflowkit.NewExecutor(store, steps,
	workflowkit.WithRetryPolicy(workflowkit.RetryPolicy{MaxAttempts: 3}),
)

Only errors marked as transient are retried:

return workflowkit.StepResult{}, workflowkit.TransientError{Err: err}

Plain Go errors fail the workflow immediately. Retry is intentionally local to a step execution; it is not a distributed scheduler or delayed job queue.

Step History

StepRecords is the workflow observability trail. Each step attempt appends one record with the step name, attempt number, status, refs, bounded metadata, error, and timestamps. Retry leaves multiple records for the same step.

StepRecords should carry refs such as OutputRef, AuditRef, AgentRunID, and small metadata. It should not store raw prompts, raw tool inputs, full model messages, or full tool output.

SQLite Store

Use sqlitestore when a host needs workflow state to survive process restarts:

store, err := sqlitestore.Open("workflow.db")
if err != nil {
	panic(err)
}
defer store.Close()

The SQLite store implements the same Store contract as MemoryStore. It keeps queryable scalar fields in columns and stores step history, metadata, attempts, and completed steps as JSON fields. It still stores refs and bounded metadata, not raw prompts or full tool payloads. The current SQLite schema is versioned as sqlitestore.SchemaVersion.

Queue Lease

Use QueueStore when a host needs to claim a pending workflow for background execution. Use QueueLeaseStore when the host also needs to extend or release the claimed lease:

queue := store.(workflowkit.QueueLeaseStore)
run, err := queue.ClaimRunnable(ctx, "worker-1", 30*time.Second)
run, err = queue.ExtendLease(ctx, run.ID, "worker-1", 30*time.Second)
run, err = queue.ReleaseLease(ctx, run.ID, "worker-1")

ClaimRunnable selects the oldest pending workflow whose lease is empty or expired, writes LeaseOwner and LeaseUntil, and returns the claimed run. It does not run the workflow. ExtendLease refreshes only the current active owner; an expired owner cannot revive its lease. ReleaseLease clears the current owner's lease after execution stops at waiting_approval or terminal.

QueueLeaseStore still does not start worker goroutines, run heartbeat loops, recover stuck workers, or define multi-worker scheduling policy. Those remain host-owned execution concerns.

After claiming, hosts can pass the returned pending run to Executor.Run.

Workflow Query

Use WorkflowQueryStore when a host needs an operational list view over stored workflow runs:

query := store.(workflowkit.WorkflowQueryStore)
runs, err := query.ListWorkflows(ctx, workflowkit.WorkflowQuery{
	Status:         workflowkit.StatusPending,
	MetadataEquals: map[string]string{"run_mode": "queued"},
	Order:          workflowkit.WorkflowOrderDesc,
	Limit:          50,
})

ListWorkflows returns copies ordered by CreatedAt then workflow id. Status and MetadataEquals are optional filters. Order defaults to ascending and can be set to WorkflowOrderDesc. Limit is optional and means no store-level limit when zero.

Store Conformance

Use storetest when adding a new Store implementation:

func TestStoreConformance(t *testing.T) {
	storetest.RunStoreConformance(t, func(t *testing.T) workflowkit.Store {
		return newStoreForTest(t)
	})
}

The suite verifies copy-on-read/write behavior, Update, and not-found errors so store implementations keep the same semantics.

Agent Step

Use the optional github.com/eruca/goagents/workflowkit/agentstep module when a workflow step should run a goagent agent:

step := agentstep.New("agent", agent, func(run workflowkit.WorkflowRun) agentcore.RunRequest {
	return agentcore.RunRequest{Input: run.InputRef}
})

The adapter preserves the agent run ID and maps aborts into a failed workflow step. Host applications decide how to persist audit events and full artifacts.

Verify

./scripts/verify-e2e.sh

The script runs core tests, race tests, basic workflow examples, SQLite resume, the main-module dependency boundary check, the optional agentstep module, and the nested agent-approval and ocr-review examples.

See docs/contracts.md for API and persistence contracts, and docs/release.md and docs/release-readiness.md for release preparation.

Examples

  • examples/basic shows retry, approval wait, Approve, and resume.
  • examples/agent-approval shows agentstep wrapping a mock goagent run, mapping the agent result to waiting_approval, recording an audit ref, and continuing to finalization.
  • examples/sqlite-resume shows waiting approval, closing the process-owned store, reopening SQLite, updating audit refs, and continuing from persisted state.
  • examples/ocr-review is a nested example module that shows a host-owned composition across ocrs, contextkit, goagent, and workflowkit. OCR output is stored behind artifact refs, contextkit builds a bounded model-facing projection, agentstep maps the review to waiting_approval, and the workflow resumes after the host records an audit ref.

API Stability

The intended stable surface is:

  • WorkflowRun, Status, Step, StepResult, and StepRecord
  • Executor methods: Run, Continue, Approve, and Cancel
  • Store, WorkflowQueryStore, QueueStore, QueueLeaseStore, MemoryStore, RetryPolicy, TransientError
  • lifecycle errors: ErrRunNotFound, ErrInvalidTransition, and InvalidTransitionError
  • queue errors: ErrNoRunnableWorkflow and ErrWorkflowLeaseNotOwned
  • status errors: ErrInvalidStatus and InvalidStatusError

Extension packages are useful but still early:

  • agentstep is a separately versioned optional module. It depends on goagent; the main workflowkit module does not.
  • sqlitestore is suitable for local persistence and prototypes; schema compatibility should be treated as experimental until versioned migrations are introduced.
  • storetest is intended for future store implementations.

workflowkit/agentstep requires tagged goagent and workflowkit versions and contains no local replace. The root go.work supplies version-specific local mappings before those tags exist; nested example modules keep relative replaces because they are verification programs, not published libraries.

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

Constants

This section is empty.

Variables

View Source
var (
	ErrInvalidTransition = errors.New("invalid workflow status transition")
	ErrInvalidStatus     = errors.New("invalid workflow status")
)
View Source
var ErrNoRunnableWorkflow = errors.New("no runnable workflow")
View Source
var ErrRunNotFound = errors.New("workflow run not found")
View Source
var ErrWorkflowLeaseNotOwned = errors.New("workflow lease not owned")

Functions

func IsTransient

func IsTransient(err error) bool

Types

type Approval

type Approval struct {
	AuditRef string
	Metadata map[string]any
}

type Executor

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

func NewExecutor

func NewExecutor(store Store, steps []Step, options ...Option) *Executor

func (*Executor) Approve

func (e *Executor) Approve(ctx context.Context, id string, approval Approval) (WorkflowRun, error)

func (*Executor) Cancel

func (e *Executor) Cancel(ctx context.Context, id string, reason string) (WorkflowRun, error)

func (*Executor) Continue

func (e *Executor) Continue(ctx context.Context, id string) (WorkflowRun, error)

func (*Executor) Run

func (e *Executor) Run(ctx context.Context, run WorkflowRun) (WorkflowRun, error)

type InvalidStatusError

type InvalidStatusError struct {
	Status Status
	Field  string
}

func (InvalidStatusError) Error

func (e InvalidStatusError) Error() string

func (InvalidStatusError) Unwrap

func (e InvalidStatusError) Unwrap() error

type InvalidTransitionError

type InvalidTransitionError struct {
	From Status
	To   Status
	Op   string
}

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) 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 QueueStore

type QueueStore interface {
	Store
	ClaimRunnable(ctx context.Context, workerID string, lease time.Duration) (WorkflowRun, error)
}

type RetryPolicy

type RetryPolicy struct {
	MaxAttempts int
	Delay       time.Duration
}

type Status

type Status string
const (
	StatusPending         Status = "pending"
	StatusRunning         Status = "running"
	StatusWaitingApproval Status = "waiting_approval"
	StatusSucceeded       Status = "succeeded"
	StatusFailed          Status = "failed"
	StatusCancelled       Status = "cancelled"
)

func (Status) IsValid

func (s Status) IsValid() bool

type Step

type Step interface {
	Name() string
	Run(context.Context, WorkflowRun) (StepResult, error)
}

type StepRecord

type StepRecord struct {
	Name          string
	Status        Status
	Attempt       int
	OutputRef     string
	AgentRunID    string
	AuditRef      string
	Error         string
	ApprovalRef   string
	WaitingReason string
	StartedAt     time.Time
	EndedAt       time.Time
	Metadata      map[string]any
}

type StepResult

type StepResult struct {
	Status        Status
	OutputRef     string
	AgentRunID    string
	AuditRef      string
	Error         string
	ApprovalRef   string
	WaitingReason string
	Metadata      map[string]any
}

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
	Metadata       map[string]any
	LeaseOwner     string
	LeaseUntil     time.Time
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

Directories

Path Synopsis
examples
basic 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.

Jump to

Keyboard shortcuts

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