workflow

Documentation

Overview

Package workflow loads and runs declarative pipelines (workflow files) that chain scripts, tool calls, one-shot model completions, and subagent delegations without routing intermediate data through the main model.

Design: docs/guides/workflow.zh.md. The invariant is that the main conversation never touches large payloads: steps exchange data by reference (artifact files under work/workflow/<run>/) and only the rendered result template crosses back to the model.

Package workflow registers tool/workflow: expose each discovered declarative pipeline (script / tool / model / agent steps) as its own dynamic tool, so the model calls a workflow directly with a structured input schema while large payloads flow between steps without crossing the main model.

Index

• Constants
• Variables
• func NewWorkflow(cfg WorkflowConfig, deps WorkflowDeps) (agentkit.ToolProvider, error)
• type Engine
  • func (e *Engine) Run(ctx context.Context, wf *Workflow, args map[string]any) (string, error)
• type JSON
  • func (j JSON) String() string
• type Step
• type StructuredShell
• type Workflow
  • func Discover(ctx context.Context, ws workspace.Service, dirs []string) ([]*Workflow, []error)
  • func Parse(name string, data []byte) (*Workflow, error)
  • func (wf *Workflow) DefaultTimeoutSeconds() int
  • func (wf *Workflow) InputsJSONSchema() string
• type WorkflowConfig
• type WorkflowDeps

Constants

const (
	StepScript = "script"
	StepTool   = "tool"
	StepModel  = "model"
	StepAgent  = "agent"
)

Step types.

const (
	// TimeoutBufferSeconds is added to the summed step timeouts when computing
	// the workflow tool's default call timeout, leaving room for result
	// rendering after the final step.
	TimeoutBufferSeconds = 5
	// DefaultStepTimeoutSeconds estimates a script/tool/model step without an
	// explicit timeoutSeconds when computing the default call timeout (agent
	// steps use the subagent delegation default).
	DefaultStepTimeoutSeconds = 120
)

const DefaultModelMaxInputBytes = 16384

DefaultModelMaxInputBytes bounds the rendered model-step prompt. Exceeding it fails the step instead of silently truncating: routing a large payload back through a model is exactly what workflows exist to avoid.

const MaxViewBytes = 50 * 1024

MaxViewBytes caps the inline view a step state exposes to templates; larger outputs are written to an artifact file and referenced by path instead.

Variables

var DefaultDirs = []string{workspace.ScopeLocal + ":workflows", workspace.ScopeGlobal + ":workflows"}

DefaultDirs are scanned when the plugin config omits dirs; local wins.

Functions

func NewWorkflow

func NewWorkflow(cfg WorkflowConfig, deps WorkflowDeps) (agentkit.ToolProvider, error)

NewWorkflow registers tool/workflow: one dynamic tool per workflow file.

Best practices:

• Mount through tools/runtime deps.dynamicTools; with tools/deferred the workflow tools join MCP/OpenAPI behind the disclosure bridges.
• Point deps.runtime at a sibling tools/runtime that includes the tools your steps call, and never includes tool/workflow itself.
• For script steps, inject deps.shell with the same tool/shell-bash instance (StructuredShell) to avoid parsing bash text; otherwise the engine calls the bash tool through Runtime.
• Keep step data exchange by reference: scripts and tools read/write files; the model only sees the rendered result.
• The "workflow" slash command is contributed via CommandProvider on the same instance.

Types

type Engine

type Engine struct {
	// Tools executes tool steps through the policy/approval pipeline. Use a
	// sibling tools/runtime instance that does not include tool/workflow.
	Tools agentkit.ToolRuntime
	// Workspace resolves scoped dirs and the artifact root.
	Workspace workspace.Service
	// Shell runs script steps with raw output (no text parsing). When nil,
	// script steps call the bash tool through Tools and parse its output.
	Shell StructuredShell
	// LLM backs model steps (one-shot completion).
	LLM agentkit.LLMProvider
	// Subagent backs agent steps.
	Subagent subagent.Spawner
	// Telemetry receives workflow.run / workflow.step / workflow.script spans
	// and model-step generations (nested tool.workflow → run → step → …).
	Telemetry captelemetry.Toolkit
}

func (*Engine) Run

func (e *Engine) Run(ctx context.Context, wf *Workflow, args map[string]any) (string, error)

Run executes the workflow level by level (dependency DAG; serial when no step declares dependsOn) and returns the rendered result template.

type JSON

type JSON struct{ V any }

JSON wraps a structured step output so direct template actions ({{ .steps.x.json }}) render compact JSON while dig can still navigate it.

func (JSON) String

func (j JSON) String() string

type Step

type Step struct {
	ID   string `json:"id"`
	Type string `json:"type"`
	// ContinueOnError records the failure in step state and runs the next step.
	ContinueOnError bool `json:"continueOnError,omitempty"`
	TimeoutSeconds  int  `json:"timeoutSeconds,omitempty"`

	// When is a template rendered after the step's dependencies complete;
	// a falsy result ("", "0", "false") skips the step with status "skipped".
	// Built-in comparison funcs (eq, ne, lt, gt, and, or, not) cover most
	// conditions, e.g. {{ eq (dig .steps.check.json "ok") true }}.
	When string `json:"when,omitempty"`
	// DependsOn lists the step ids this step waits for. nil (absent) keeps the
	// serial default — wait for the previous step in file order; an explicit
	// array (even empty) replaces that edge, enabling DAG fan-out/join.
	DependsOn []string `json:"dependsOn,omitempty"`

	// script
	Run string `json:"run,omitempty"`
	Dir string `json:"dir,omitempty"`

	// tool
	Tool string         `json:"tool,omitempty"`
	Args map[string]any `json:"args,omitempty"`

	// model
	Model         string         `json:"model,omitempty"`
	Prompt        string         `json:"prompt,omitempty"`
	OutputSchema  map[string]any `json:"outputSchema,omitempty"`
	MaxInputBytes int            `json:"maxInputBytes,omitempty"`

	// agent
	Agent string `json:"agent,omitempty"`
	Task  string `json:"task,omitempty"`
}

Step is one pipeline stage. Fields are type-specific; see the design doc.

type StructuredShell

type StructuredShell interface {
	RunStructured(ctx context.Context, command string, timeoutSeconds *float64) (stdout, stderr string, exitCode int, err error)
}

Engine runs workflows against injected capabilities. All fields except Workspace are optional per step type; a step whose capability is missing fails with a clear error. StructuredShell runs a command and returns raw stdout/stderr/exit code (e.g. tool/shell-bash StructuredShell capability). Used for script steps when available instead of parsing bash tool text.

type Workflow

type Workflow struct {
	Name        string         `json:"-"`
	Description string         `json:"description"`
	Inputs      map[string]any `json:"inputs"`
	Steps       []Step         `json:"steps"`
	Result      string         `json:"result"`
	// Dir is the directory holding the workflow file; relative script paths
	// resolve against it.
	Dir string `json:"-"`
	// Path is the workflow file itself, for error messages.
	Path string `json:"-"`
}

Workflow is one parsed pipeline file. Name comes from the file name.

func Discover

func Discover(ctx context.Context, ws workspace.Service, dirs []string) ([]*Workflow, []error)

Discover re-reads every configured dir and returns the workflows found, sorted by name. Like subagent Definitions, re-reading per call makes edits take effect without a reload command. The first dir wins on duplicate names; files that fail to parse are collected into errs (and skipped) so one bad file does not block the others.

func Parse

func Parse(name string, data []byte) (*Workflow, error)

Parse decodes and validates one workflow file body.

func (*Workflow) DefaultTimeoutSeconds

func (wf *Workflow) DefaultTimeoutSeconds() int

DefaultTimeoutSeconds estimates the workflow's wall-clock upper bound as the critical path (longest dependency chain) of per-step timeouts plus a small buffer — serial workflows form one chain, so the estimate equals the old sum-of-all-steps, while parallel branches overlap instead of adding up. It backs the workflow tool's default per-call timeout (agentkit.CallTimeoutProvider), keeping the outer call deadline behind every inner step timeout.

func (*Workflow) InputsJSONSchema

func (wf *Workflow) InputsJSONSchema() string

InputsJSONSchema returns the declared inputs schema as a JSON string for the tool description; empty when no inputs are declared.

type WorkflowConfig

type WorkflowConfig struct {
	// Dirs are scoped workflow directories in precedence order (first dir wins
	// on duplicate names). Defaults to local:workflows then global:workflows.
	Dirs []string `json:"dirs,omitempty"`
}

type WorkflowDeps

type WorkflowDeps struct {
	// Runtime executes tool steps; use a sibling tools/runtime instance without
	// tool/workflow itself (cycle break, same pattern as tool/subagent).
	Runtime   agentkit.ToolRuntime `json:"runtime"`
	Workspace workspace.Service    `json:"workspace"`
	// LLM is required only when a workflow uses model steps.
	LLM agentkit.LLMProvider `json:"llm,omitempty"`
	// Subagent is required only when a workflow uses agent steps.
	Subagent subagent.Spawner `json:"subagent,omitempty"`
	// Telemetry receives workflow.run / workflow.step / workflow.script spans.
	Telemetry captelemetry.Toolkit `json:"telemetry,omitempty"`
	// Shell runs script steps with raw output (no model formatting). Inject the
	// same tool/shell-bash instance (StructuredShell); omit to fall back to
	// calling the bash tool through Runtime and parsing its text output.
	Shell StructuredShell `json:"shell,omitempty"`
}