workflow

package
v1.4.0 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

The orchestrator controller. Reconcile is stateless and durable: all state lives in PostgreSQL, so restarts (and concurrent orchestrator instances) resume without losing dependency progress. Task creation is idempotent per (workflow, step, attempt) and every step/workflow transition is CAS-guarded, so racing instances converge with zero double dispatch.

Package workflow implements the durable orchestrator: WorkflowRun and Step state, dependency dispatch with conditions and joins, single-step retry, human approval, cancellation propagation and restart recovery. The orchestrator decides who executes when and creates ordinary Tasks; it never schedules (that stays with the scheduler) and agents never talk to each other directly — Agent A's result reaches Agent B through Fenced.

Index

Constants

This section is empty.

Variables

View Source 
var ErrOutputContract = errors.New("workflow output contract violation")

Functions

func DecodeSpec 

func DecodeSpec(raw []byte) ([]kernelstore.CreateWorkflowStepInput, error)

DecodeSpec strictly decodes and fully validates one workflow document, returning the step inputs (with default/step spec overlays merged) for durable storage.

Types

type ApprovalController 

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

ApprovalController owns the human-approval gate for steps that require it: it parks undecided pending steps at WAITING_APPROVAL and records a rejected decision as SKIPPED.

func NewApprovalController 

func NewApprovalController(workflows stepTransitioner, owner string) *ApprovalController

NewApprovalController builds an approval gate bound to a step store.

func (*ApprovalController) Park 

func (a *ApprovalController) Park(ctx context.Context, workflow kernelstore.Workflow, step kernelstore.WorkflowStep) (bool, error)

Park transitions an undecided pending approval-required step to WAITING_APPROVAL so it stays parked until a human decides. It reports whether the step was parked.

func (*ApprovalController) Reject 

func (a *ApprovalController) Reject(ctx context.Context, workflow kernelstore.Workflow, step kernelstore.WorkflowStep) (bool, error)

Reject records a decided rejection as SKIPPED. It reports whether the step was rejected (and therefore must not be dispatched).

type ClaimManager 

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

ClaimManager acquires the batch of workflows one reconcile round will process, either under an expiring per-instance claim or as every visible active workflow in single-instance mode.

func NewClaimManager 

func NewClaimManager(workflows workflowClaimer, owner string, batch int, lease time.Duration, tokenBudget int64) *ClaimManager

NewClaimManager builds a ClaimManager for one controller instance. A zero lease keeps single-instance behavior (reconcile everything visible).

func (*ClaimManager) Claim 

func (m *ClaimManager) Claim(ctx context.Context) ([]kernelstore.Workflow, error)

Claim returns the active workflows this instance should reconcile this round. In claiming mode the store leases each workflow under the manager's lease so a dead instance's batch is taken over by peers after expiry.

type Controller 

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

Controller reconciles active workflows.

func NewController 

func NewController(workflows kernelstore.WorkflowStore, tasks TaskPipeline, artifacts ResultReader, owner string, batch int) *Controller

NewController builds the orchestrator. batch bounds workflows per reconcile; parallel bounds concurrent per-workflow processing (default 4).

func (*Controller) Reconcile 

func (c *Controller) Reconcile(ctx context.Context) (int, error)

Reconcile drives active workflows one round: dispatching ready steps, observing task terminals, applying retries, propagating cancellation, enforcing workflow budgets and finalizing workflows. In claiming mode Each instance leases its batch under an expiring claim; otherwise it reconciles everything visible. Retryable transaction conflicts are retried with bounded backoff (ADR-002).

func (*Controller) WithClaiming 

func (c *Controller) WithClaiming(lease time.Duration, tokenBudget int64) *Controller

WithClaiming enables distributed work-claim sharding: each reconcile claims a batch of workflows under a lease that expires when the instance dies, so peers take over without double dispatch. A zero lease keeps the single-instance behavior (reconcile everything visible).

func (*Controller) WithMaxInFlightSteps 

func (c *Controller) WithMaxInFlightSteps(limit int) *Controller

WithMaxInFlightSteps bounds concurrently dispatched steps per workflow.

func (*Controller) WithParallelism 

func (c *Controller) WithParallelism(workers int) *Controller

WithParallelism bounds concurrent per-workflow reconciliation.

type DependencyResolver 

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

DependencyResolver gates one step on its declared dependencies and condition. It collects the succeeded upstream outputs for condition evaluation and downstream goals, marks the step SKIPPED when an upstream failed (or the condition is not met), and reports whether the step must keep waiting.

func NewDependencyResolver 

func NewDependencyResolver(workflows stepTransitioner, owner string) *DependencyResolver

NewDependencyResolver builds a DependencyResolver bound to a step store.

func (*DependencyResolver) Resolve 

func (r *DependencyResolver) Resolve(ctx context.Context, workflow kernelstore.Workflow, step kernelstore.WorkflowStep, declared *StepSpec, byName map[string]kernelstore.WorkflowStep) (map[string]string, bool, error)

Resolve evaluates one step's dependencies and condition against the current step set.

It returns (nil, false, nil) while an upstream is still non-terminal and (nil, true, err) when the step was transitioned to SKIPPED (failed upstream, unmet condition) — the caller must stop advancing the step. On success it returns the bounded dependency outputs and ready=true.

type FailureDrainer 

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

FailureDrainer cancels active step tasks, skips undispatched steps, and finalizes a failing or cancelled workflow once nothing is running. A step that already succeeded keeps its result.

func NewFailureDrainer 

func NewFailureDrainer(workflows drainStore, tasks TaskPipeline, owner string) *FailureDrainer

NewFailureDrainer builds a drainer bound to the workflow store and task pipeline of one controller.

func (*FailureDrainer) Cancel 

func (d *FailureDrainer) Cancel(ctx context.Context, workflow kernelstore.Workflow, steps []kernelstore.WorkflowStep) (bool, error)

Cancel propagates the durable cancel intent: active step tasks are cancelled through the kernel, undispatched steps are skipped, and the workflow finalizes CANCELLED once nothing is running.

func (*FailureDrainer) Drain 

func (d *FailureDrainer) Drain(ctx context.Context, workflow kernelstore.Workflow, steps []kernelstore.WorkflowStep, failureCode string) (bool, error)

Drain cancels active tasks, skips undispatched steps, and finalizes the workflow FAILED with the durable failureCode once every step is terminal.

func (*FailureDrainer) DrainBudgetExhausted 

func (d *FailureDrainer) DrainBudgetExhausted(ctx context.Context, workflow kernelstore.Workflow, steps []kernelstore.WorkflowStep) (bool, error)

DrainBudgetExhausted drains a budget-stopped workflow exactly like Drain with the durable budget-exhausted reason.

type LeaseManager 

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

LeaseManager keeps a claimed workflow's lease fresh while the controller reconciles it. If renewal fails the work context is cancelled so the instance stops processing a claim it no longer owns, allowing a peer to take over without waiting for full lease expiry.

func NewLeaseManager 

func NewLeaseManager(renewer workflowClaimRenewer, owner string, lease time.Duration) *LeaseManager

NewLeaseManager builds a LeaseManager bound to one controller instance's identity. A nil renewer or zero lease makes Guard a no-op.

func (*LeaseManager) Guard 

func (m *LeaseManager) Guard(ctx context.Context, workflow kernelstore.Workflow) (workCtx context.Context, release func() error)

Guard wraps ctx so the caller's work is cancelled when the claim lease can no longer be renewed. The returned release must be called when the work completes; its error is non-nil when renewal failed while the work was running, using the same error semantics as the original Controller:

  • If the work errored AND renewal failed → errors.Join(workErr, claimErr)
  • If the work succeeded AND renewal failed → fmt.Errorf("renew …")
  • If renewal succeeded → workErr (nil or not)

func (*LeaseManager) GuardedProcess 

func (m *LeaseManager) GuardedProcess(ctx context.Context, workflow kernelstore.Workflow, fn func(context.Context) (bool, error)) (bool, error)

GuardedProcess runs fn under lease guard and returns the combined error: work error + renewal error with the same join semantics as the original Controller.processClaimedWorkflow.

type ResultReader 

type ResultReader interface {
	Open(context.Context, string, kernelstore.ArtifactReference) (io.ReadCloser, error)
}

ResultReader opens task result artifacts to extract dependency outputs for condition evaluation and downstream goals. The artifact.Filesystem satisfies it.

type RetryController 

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

RetryController decides what happens to a step whose task failed: retry within the step retry budget (back to PENDING) or fail the step when the budget is exhausted. A retry denied by the workflow budget fails the step instead of retry-looping against a hard ceiling.

func NewRetryController 

func NewRetryController(workflows stepTransitioner, owner string) *RetryController

NewRetryController builds a retry gate bound to a step store.

func (*RetryController) HandleFailed 

func (r *RetryController) HandleFailed(ctx context.Context, workflow kernelstore.Workflow, step kernelstore.WorkflowStep, declared *StepSpec) (bool, error)

HandleFailed reacts to a FAILED step task. It reports whether the step state moved (retried or failed); nil error means the step is now terminal or parked for retry.

type StepCondition 

type StepCondition struct {
	Step           string `json:"step"`
	OutputContains string `json:"outputContains,omitempty"`
	OutputEquals   string `json:"outputEquals,omitempty"`
	// JSONPointer selects a typed value from the upstream output using RFC
	// 6901. EqualsJSON compares JSON values structurally (not textually).
	JSONPointer string          `json:"jsonPointer,omitempty"`
	EqualsJSON  json.RawMessage `json:"equalsJson,omitempty"`
}

StepCondition is the structured condition of one step: evaluated against the stored result summary of one dependency step. Exactly one predicate must be set; a false predicate skips the step (CONDITION_NOT_MET).

type StepDispatcher 

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

StepDispatcher creates the fenced Task for one ready step and transitions the step to RUNNING. Task creation is idempotent per (workflow, step, attempt), so racing orchestrators dispatch exactly once.

func NewStepDispatcher 

func NewStepDispatcher(tasks TaskPipeline, workflows stepTransitioner, newID func() uuid.UUID, owner string) *StepDispatcher

NewStepDispatcher builds a dispatcher bound to the task pipeline and step store of one controller.

func (*StepDispatcher) Dispatch 

func (d *StepDispatcher) Dispatch(ctx context.Context, workflow kernelstore.Workflow, step kernelstore.WorkflowStep, declared *StepSpec, defaultTaskSpec json.RawMessage, dependencyOutputs map[string]string, byName map[string]kernelstore.WorkflowStep) (bool, error)

Dispatch creates the step's task and records the RUNNING transition. dependencyOutputs are the resolved upstream results rendered into the goal.

type StepOutputContract 

type StepOutputContract struct {
	ContentType   string          `json:"contentType,omitempty"`
	SchemaVersion string          `json:"schemaVersion"`
	Schema        json.RawMessage `json:"schema"`
}

StepOutputContract defines the durable Agent A -> Agent B data contract. Schema is JSON Schema 2020-12 and validates the result document's output value before the step may become SUCCEEDED.

type StepRetry 

type StepRetry struct {
	MaxAttempts int `json:"maxAttempts"`
}

StepRetry bounds the single-step retry budget (total task attempts).

type StepSpec 

type StepSpec struct {
	Name             string              `json:"name"`
	AgentVersionRef  string              `json:"agentVersionRef"`
	Goal             string              `json:"goal"`
	Spec             json.RawMessage     `json:"spec,omitempty"`
	DependsOn        []string            `json:"dependsOn,omitempty"`
	Condition        *StepCondition      `json:"condition,omitempty"`
	Retry            *StepRetry          `json:"retry,omitempty"`
	RequiresApproval bool                `json:"requiresApproval"`
	Output           *StepOutputContract `json:"output,omitempty"`
}

StepSpec is one declared DAG node of the workflow document.

type TaskPipeline 

type TaskPipeline interface {
	CreateTask(context.Context, kernelstore.CreateTaskInput) (kernelstore.CreateTaskResult, error)
	GetTask(context.Context, string, uuid.UUID) (kernelstore.Task, error)
	RequestTaskCancellation(context.Context, string, uuid.UUID, int64) (kernelstore.Task, error)
}

TaskPipeline is the task surface the orchestrator drives. The kernel postgres Store satisfies it; every step becomes an ordinary fenced Task.

type WorkflowBudget 

type WorkflowBudget struct {
	MaxTasks        int64          `json:"maxTasks,omitempty"`
	MaxTokens       int64          `json:"maxTokens,omitempty"`
	MaxCostMicroUSD money.MicroUSD `json:"maxCostUsd,omitempty"`
}

WorkflowBudget is the workflow-level ceiling set of the workflow document Every dimension is optional; zero leaves it unbounded.

type WorkflowDynamicPolicy 

type WorkflowDynamicPolicy struct {
	Enabled              bool           `json:"enabled,omitempty"`
	MaxDynamicSteps      int64          `json:"maxDynamicSteps,omitempty"`
	MaxChildrenPerStep   int64          `json:"maxChildrenPerStep,omitempty"`
	MaxSpawnDepth        int            `json:"maxSpawnDepth,omitempty"`
	MaxWorkflowSteps     int64          `json:"maxWorkflowSteps,omitempty"`
	MaxSpawnTasks        int64          `json:"maxSpawnTasks,omitempty"`
	MaxSpawnTokens       int64          `json:"maxSpawnTokens,omitempty"`
	MaxSpawnCostMicroUSD money.MicroUSD `json:"maxSpawnCostUsd,omitempty"`
}

WorkflowDynamicPolicy bounds dynamic spawning.

type WorkflowRuntimePolicy 

type WorkflowRuntimePolicy struct {
	Dynamic WorkflowDynamicPolicy `json:"dynamic,omitempty"`
}

WorkflowRuntimePolicy is the dynamic-orchestration policy of the workflow document: the guards the spawning path enforces at runtime.

type WorkflowSpec 

type WorkflowSpec struct {
	DefaultTaskSpec json.RawMessage        `json:"defaultTaskSpec,omitempty"`
	Budget          *WorkflowBudget        `json:"budget,omitempty"`
	Runtime         *WorkflowRuntimePolicy `json:"runtime,omitempty"`
	Deadline        *time.Time             `json:"deadline,omitempty"`
	Steps           []StepSpec             `json:"steps"`
}

WorkflowSpec is the user-facing workflow document.

func DecodeWorkflowSpec 

func DecodeWorkflowSpec(raw []byte) (WorkflowSpec, error)

DecodeWorkflowSpec strictly decodes and fully validates one workflow document, preserving the budget and runtime policy alongside the step inputs.

func (WorkflowSpec) Budgets 

func (s WorkflowSpec) Budgets() (tasks, tokens int64, costMicroUSD money.MicroUSD)

Budgets returns the validated workflow budget ceilings.

func (WorkflowSpec) MergeTaskSpec 

func (s WorkflowSpec) MergeTaskSpec(overlay json.RawMessage) (json.RawMessage, error)

MergeTaskSpec applies the workflow default task specification to a dynamic-step overlay. Dynamic children therefore enter the ordinary Task pipeline with the same placement and budget defaults as declared steps.

func (WorkflowSpec) StepInputs 

func (s WorkflowSpec) StepInputs() []kernelstore.CreateWorkflowStepInput

StepInputs renders the durable step inputs of a validated spec (with the default/step overlay merged).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.