workflow

package module
v0.17.2 Latest Latest
Warning

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

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

README

go-workflow

Standalone DAG workflow engine for Go. 15 step types, MCP server integration, pluggable persistence (file/SQLite/PostgreSQL), distributed execution, templates, approval flows, crash recovery.

Go 1.26 · Apache-2.0 · Releases

Installation

go get github.com/anatolykoptev/go-workflow

Features

  • DAG execution — steps run in parallel when dependencies allow
  • 15 step types — tool, llm, agent, a2a, message, condition, transform, approval, workflow, foreach, branchall, suspend, noop, image, and vision (multimodal LLM with image inputs via the optional VisionCapable interface)
  • MCP integration — WithMCPServers() connects to any MCP server, auto-discovers tools
  • Templates — parameterized workflow definitions with {{variable}} (string) and "@@int:NAME" / "@@bool:NAME" / "@@float:NAME" (typed) substitution, loaded from JSON files
  • Approval flow — pause workflow, await human/AI approval, resume or reject
  • Pluggable persistence — JSON files (default), SQLite, or PostgreSQL
  • Distributed execution — dispatch steps to remote workers via PostgreSQL SKIP LOCKED queue
  • Production retry — exponential backoff with ±25% jitter, Retry-After honoring, per-step timeout, conditional retry/skip, dead letter
  • Circuit breaking — per-endpoint circuit breaker on outbound tool / agent / MCP / A2A / vision calls; fails fast when an endpoint is down, recovers on a half-open probe
  • Rate limiting — opt-in per-provider QPS token-bucket on outbound calls via WithRateLimit(provider, rate, burst)
  • Crash recovery — RecoverAll() resumes workflows interrupted by process crash
  • Idempotency — IdempotencyKey prevents duplicate workflow runs
  • Security policies — step budgets, duration limits, tool allow/deny lists, secret masking
  • Watchdog — auto-detect stalled steps, auto-retry transient failures
  • n8n import — convert n8n workflow JSON to native templates
  • Scheduler + Cron — time-based workflow triggers
  • Metrics — atomic counters for workflows, steps, agents, hooks, triggers
  • Cost tracking — every LLM and image step contributes to Workflow.Cost (tokens, USD, image bytes). Optional WithBudget(maxUSD) aborts overrun workflows with ErrBudgetExceeded.
  • OpenTelemetry tracing — WithTracerProvider(tp) emits a workflow.run span per workflow and a step.<kind> span per step, with step.duration_ms, step.cache_hit, cost, and classified error attributes. Wires any OTel-compatible backend (Jaeger, Tempo, Honeycomb, Datadog, OTLP).
  • Webhook triggers — Engine.RegisterWebhooks(mux, runtime, []WebhookTrigger{...}) instantiates a template from an HTTP POST. Supports bearer-token and HMAC-SHA256 (GitHub-style) auth with constant-time comparison; body cap 10 MiB; default JSON VarMapper overridable per trigger.
  • Step caching — WithStepCache(cache) / WithStepCacheKinds(...) skips deterministic steps when the same (kind, config, depends_on) hash has been seen before. InMemoryCache and FileCache ship in-package; cache hits replay both Output and Cost so accounting stays accurate.

Quick Start

engine := workflow.NewEngine(store,
    workflow.WithMCPServers(map[string]string{
        "content-server": "http://127.0.0.1:8080/mcp",
        "search-service": "http://127.0.0.1:8081/mcp",
    }),
    workflow.WithLLMClient(llmClient),
)

engine.RecoverAll(ctx)

wf := workflow.NewWorkflow("wf-1", "Content Pipeline", "ai:claude", []workflow.Step{
    {ID: "research", Kind: workflow.StepTool, Config: map[string]any{
        "tool": "search_web",
        "args": map[string]any{"topic": "best viewpoints in the city", "count": 12},
    }},
    {ID: "select", Kind: workflow.StepApproval, Config: map[string]any{
        "message": "Review results and select top 12",
    }, DependsOn: []string{"research"}},
    {ID: "images", Kind: workflow.StepTool, Config: map[string]any{
        "tool": "get_images",
        "args": map[string]any{"action": "batch"},
    }, DependsOn: []string{"select"}},
})

_ = store.Save(wf)
_ = engine.StartAsync(ctx, "wf-1") // runs in background, pauses at approval

MCP Integration

Connect to any MCP server. Tools are auto-discovered via ListTools.

engine := workflow.NewEngine(store,
    workflow.WithMCPServers(map[string]string{
        "my-tools":       "http://127.0.0.1:8080/mcp",
        "search-service": "http://127.0.0.1:8081/mcp",
        "browser":        "http://127.0.0.1:8082/mcp",
    }),
)

// Steps reference tools by name — routing is automatic:
// "create_item"   → my-tools server
// "smart_search"  → search-service server
// "fetch_page"    → browser server

MCPToolRunner handles lazy connection, tool discovery, and multi-server routing. Combined with local ToolRunner via MultiToolRunner.

Templates

JSON files loaded by TemplateStore. IDE / Claude validation via JSON Schema:

{ "$schema": "https://raw.githubusercontent.com/anatolykoptev/go-workflow/main/docs/template.schema.json" }

See docs/template.schema.json for the full schema reference.

Substitution

  • {{key}} — string substitution. The surrounding JSON quotes are preserved; result is always a JSON string.

  • Typed ParamSpec (preferred) — declare "type": "int" (or bool, float) on the param; engine coerces the value before substitution and strips surrounding quotes automatically:

    "params": {
  "count":   {"type": "int",   "description": "Number of places", "default": 12},
  "verbose": {"type": "bool",  "description": "Verbose log flag",  "default": false}
}

    Then "count": "{{count}}" in a step config becomes "count": 12 after substitution (bare integer).

  • "@@int:KEY" / "@@bool:KEY" / "@@float:KEY" (deprecated since v0.13) — typed substitution via magic markers. Quotes are stripped after substitution. Emits a deprecation log per match pointing to the typed ParamSpec form above.

{
  "name": "Content pipeline: {{topic}}",
  "params": {"topic": "Article topic", "count": "Number of places", "verbose": "Verbose log flag"},
  "defaults": {"count": "12", "verbose": "false"},
  "steps": [
    {"id": "research", "kind": "tool", "config": {"tool": "search_web", "args": {"topic": "{{topic}}", "count": "@@int:count", "verbose": "@@bool:verbose"}}},
    {"id": "select",   "kind": "approval", "config": {"message": "Select items"}, "depends_on": ["research"]},
    {"id": "enrich",   "kind": "tool", "config": {"tool": "update_metadata"}, "depends_on": ["select"]},
    {"id": "compose",  "kind": "approval", "config": {"message": "Write content"}, "depends_on": ["enrich"]},
    {"id": "publish",  "kind": "tool", "config": {"tool": "create_post"}, "depends_on": ["compose"]}
  ]
}

ts := workflow.NewTemplateStore("/path/to/templates")
wf, _ := ts.Instantiate("create-collection", "wf-123", "ai:claude", map[string]any{
    "topic":   "city viewpoints",
    "count":   "15",     // typed @@int:count emits bare 15 in JSON
    "verbose": "true",   // typed @@bool:verbose emits bare true
})

After substitution args.count is the JSON integer 15 (not the string "15"); args.verbose is the JSON boolean true. Coercion errors (e.g. @@int: against a non-numeric value) surface from ResolveRefsErr; the legacy ResolveRefs logs and continues for backward compat.

Environment Variables

Variable Description
WORKFLOW_TOOL_API_URL Base URL of the tool API server. HTTP request nodes that call /tools/execute or /agent/run on this URL are detected and converted to native tool/agent steps by the n8n importer. Example: http://127.0.0.1:8080

Approval Flow

Steps with Kind: StepApproval pause the workflow. Resume via HandleApproval:

engine := workflow.NewEngine(store,
    workflow.WithApprovalNotifier(func(wf *workflow.Workflow, step *workflow.Step) {
        // Notify AI/human that approval is needed
        log.Printf("Workflow %s waiting at step %s", wf.ID, step.ID)
    }),
    workflow.WithCompletionNotifier(func(wf *workflow.Workflow) {
        log.Printf("Workflow %s completed", wf.ID)
    }),
)

// Later, when approved:
engine.HandleApproval("wf-123", true)  // or false to reject
engine.ResumeAsync(ctx, "wf-123")

Storage Backends

Backend Constructor Use case
JSON files store.NewFileStore(dir) Development, single-process
SQLite store.NewSQLiteStore(path) Tests, single-binary deployments
PostgreSQL store.NewPostgresStore(dsn) Production, multi-process

Step Types

Kind Executor Description
tool ToolExecutor Call a named tool (local or MCP)
llm LLMExecutor Send prompt to LLM, supports tool calling
agent AgentExecutor Delegate to a full agent loop
a2a A2AExecutor Call remote A2A agent
message MessageExecutor Send message to user channel
condition ConditionExecutor Branch on expression
transform TransformExecutor Transform data between steps
approval ApprovalExecutor Pause for human/AI approval
workflow SubWorkflowExecutor Run a sub-workflow
foreach ForEachExecutor Iterate over collection
branchall BranchAllExecutor Run all branches in parallel
suspend SuspendExecutor Suspend with TTL
noop NoopExecutor Join point, completes immediately
image ImageExecutor Render HTML to PNG/JPEG/WebP/SVG via pluggable ImageRenderer
vision VisionExecutor Multimodal LLM call with image attachments (VisionCapable provider)

Cost tracking

Every LLM call (StepLLM, StepVision) and every image render (StepImage) updates Workflow.Cost:

wf, _ := engine.RunToCompletion(ctx, wf)
fmt.Printf("Total: %d in / %d out tokens, $%.4f, %d images\n",
    wf.Cost.InputTokens, wf.Cost.OutputTokens, wf.Cost.USDEstimate, wf.Cost.ImagesRendered)

Optional ceiling: workflow.WithBudget(0.50) aborts when the running total exceeds $0.50, surfacing ErrBudgetExceeded. Pricing comes from DefaultCostModel; override with WithCostModel(custom).

Retry & Timeout

step := workflow.Step{
    ID:   "flaky-api",
    Kind: workflow.StepTool,
    Config: map[string]any{
        "tool":       "http_request",
        "timeout_ms": 10000,
        "retry": map[string]any{
            "max": 3, "delay_ms": 1000, "backoff_multiplier": 2.0,
            "retry_on": []any{"timeout", "503"},
            "skip_on":  []any{"401", "403"},
        },
    },
}

After exhausting retries, steps enter StepDeadLettered state — the watchdog will not re-retry them.

Interfaces

All external dependencies are injected via interfaces:

Interface Purpose
StoreBackend Persist and load workflows
ToolRunner Execute named tools
LLMProvider Send prompts to LLM
AgentRunner Delegate to agent loop
A2ACaller Call remote A2A agents
MessagePublisher Deliver messages to users
HookPublisher Fire lifecycle events
SkillResolver Load skill prompts by name
StepDispatcher Route steps to local/remote execution
StepWorkerQueue Distributed work queue
StepReaper Reclaim steps from dead workers
StepExecutor Execute a single workflow step
StepCache Cache deterministic step results
StepEnqueuer Enqueue steps for distributed execution
ImageRenderer Render HTML to image formats
SubWorkflowRunner Run nested sub-workflows
VisionCapable Multimodal LLM capability probe

Distributed Execution

Dispatch steps to remote workers via PostgreSQL SKIP LOCKED queue. Features: heartbeat protocol, concurrency limits, LISTEN/NOTIFY, graceful shutdown. Default LocalDispatcher preserves in-process execution (zero config change).

Documentation

Overview

Package workflow — Engine ties everything together: it owns the executor map, the persistence store, lifecycle notifiers, cost accounting, and the dispatcher. The Engine type itself lives here, alongside the small set of types it depends on. See companion files for the rest of the engine's surface area:

  • engine_options.go — WithX functional options (configuration DSL)
  • engine_setters.go — Set* methods for post-NewEngine wiring
  • engine_validate.go — ValidateWorkflow + ValidateTemplate pre-flight checks
  • step_cache.go — WithStepCache / WithStepCacheKinds
  • tracing.go — WithTracerProvider
  • dispatcher.go — WithDispatcher

Index

Constants

View Source 
const (
	EventWorkflowStarted        = "workflow_started"
	EventWorkflowCompleted      = "workflow_completed"
	EventWorkflowFailed         = "workflow_failed"
	EventWorkflowCancelled      = "workflow_cancelled"
	EventWorkflowStepStarted    = "workflow_step_started"
	EventWorkflowStepCompleted  = "workflow_step_completed"
	EventWorkflowStepFailed     = "workflow_step_failed"
	EventWorkflowApprovalNeeded = "workflow_approval_needed"
)

Hook event constants.

View Source 
const (
	ParamTypeString = "string"
	ParamTypeInt    = "int"
	ParamTypeBool   = "bool"
	ParamTypeFloat  = "float"
	ParamTypeObject = "object"
	ParamTypeArray  = "array"
)

ParamSpec type constants — used in ParamSpec.Type and coercion switches.

View Source 
const (
	OnErrorFail = "fail"
	OnErrorSkip = "skip"
	// OnErrorContinue is an alias for OnErrorSkip: step is marked skipped and
	// execution continues with downstream steps. Useful in n8n-style definitions
	// where "continue" is a natural value for continueOnFail.
	OnErrorContinue = "continue"
)

OnError strategy constants for step error handling.

View Source 
const (
	ToolHTTPRequest = "http_request"
)

Built-in tool name constants used as step kind aliases.

Variables

View Source 
var DefaultCostModel = map[string]ModelPrice{
	"claude-haiku-4-5":          {InputUSDPer1K: 0.001, OutputUSDPer1K: 0.005},
	"claude-haiku-4-5-20251001": {InputUSDPer1K: 0.001, OutputUSDPer1K: 0.005},
	"claude-sonnet-4-6":         {InputUSDPer1K: 0.003, OutputUSDPer1K: 0.015},
	"claude-opus-4-7":           {InputUSDPer1K: 0.015, OutputUSDPer1K: 0.075},
	"claude-opus-4-6":           {InputUSDPer1K: 0.015, OutputUSDPer1K: 0.075},
	"gemini-2.5-flash":          {InputUSDPer1K: 0.0001, OutputUSDPer1K: 0.0004},
	"gemini-2.5-flash-lite":     {InputUSDPer1K: 0.00005, OutputUSDPer1K: 0.0002},
	"gemini-2.5-pro":            {InputUSDPer1K: 0.00125, OutputUSDPer1K: 0.005},
}

DefaultCostModel maps known model IDs to per-1k-token prices. Override via WithCostModel(map[string]ModelPrice{...}). Unknown models cost $0 — surfaces no error, just doesn't accumulate USD.

View Source 
var ErrBudgetExceeded = errors.New("workflow budget exceeded")

ErrBudgetExceeded is returned by cost-bearing executors when a workflow's running USD total exceeds the engine's configured budget. Workflows that trigger this error fail; partial cost is preserved on Workflow.Cost.

View Source 
var GlobalMetrics = NewMetrics()

Deprecated: GlobalMetrics is a package-level singleton. Use NewMetrics() + WithMetrics() instead.

Functions

func EstimateUSD 

func EstimateUSD(model string, inputTokens, outputTokens int64, costModel map[string]ModelPrice) float64

EstimateUSD computes the dollar cost of a single LLM call given a cost model. Returns 0 for unknown models (no error — surfaces in metrics as zero rather than crashing the workflow).

func IsTransientError 

func IsTransientError(errMsg string) bool

IsTransientError checks if an error message matches known transient patterns.

func IsValidStepKind 

func IsValidStepKind(kind StepKind) bool

IsValidStepKind returns true if the kind is a known canonical or alias step kind.

func MaskSecrets 

func MaskSecrets(text string) string

MaskSecrets replaces detected secrets in text with [REDACTED].

func MaskSecretsInMap 

func MaskSecretsInMap(m map[string]any) map[string]any

MaskSecretsInMap recursively masks secrets in a map's string values.

func MatchesFilter 

func MatchesFilter(filter map[string]string, data map[string]any) bool

MatchesFilter returns true if all filter key-value pairs match the data. Empty filter always matches. Values are compared case-insensitively.

func NextCronRun 

func NextCronRun(expr, tz string, nowMS int64) (int64, error)

NextCronRun computes the next run time for a 5-field cron expression. Fields: minute hour day-of-month month day-of-week. Returns the next run time in milliseconds since epoch.

func ParseOwner 

func ParseOwner(owner string) (channel, chatID string)

ParseOwner splits "channel:chatID" into parts.

func PrometheusHandler 

func PrometheusHandler(m *Metrics) http.Handler

PrometheusHandler returns an http.Handler that renders the given Metrics in Prometheus text exposition format. No external dependency required.

func RegisterMCPTools 

func RegisterMCPTools(server *mcp.Server, deps MCPDeps) int

RegisterMCPTools registers all workflow MCP tools on the given server and returns the number of tools registered.

func ResolveRefs 

func ResolveRefs(s string, wf *Workflow) string

ResolveRefs replaces {{stepID}} placeholders in a string with workflow context values. Also resolves ${VAR} patterns with environment variables (n8n compat). Preserves the original string-only signature for backward compatibility. Callers that need typed @@int/@@bool/@@float substitution should migrate to ResolveRefsErr.

TODO(#typed-markers): switch engine.go call sites to ResolveRefsErr so typed-marker errors surface as workflow step failures instead of log warns.

func ResolveRefsErr 

func ResolveRefsErr(s string, wf *Workflow) (string, error)

ResolveRefsErr is like ResolveRefs but also handles typed markers @@int:NAME, @@bool:NAME, @@float:NAME. Typed markers are written with surrounding quotes in the template (so the JSON file stays valid pre- substitution). After substitution the quotes are stripped and the value is emitted as a bare typed literal. Returns an error when a marker references a missing key or when the value cannot be coerced to the requested type.

Classic {{var}} substitution is unchanged and backward compatible.

func ResolveSecretRef 

func ResolveSecretRef(s string) string

ResolveSecretRef resolves a secret reference like $SECRET{name} from environment. Returns the original string if the reference is not found.

Types

type A2ACaller 

type A2ACaller interface {
	Call(ctx context.Context, agentID, message string) (string, error)
}

A2ACaller is the interface for calling remote A2A agents. Satisfied by *a2a.ClientManager (which has Call(ctx, agentID, message)).

type A2AExecutor 

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

A2AExecutor delegates a step to a remote A2A agent.

func NewA2AExecutor 

func NewA2AExecutor(caller A2ACaller, metrics *Metrics) *A2AExecutor

func (*A2AExecutor) Execute 

func (e *A2AExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

type AgentExecutor 

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

AgentExecutor delegates a task to the full agent loop (with tools, memory, skills).

func NewAgentExecutor 

func NewAgentExecutor(runner AgentRunner, metrics *Metrics) *AgentExecutor

func (*AgentExecutor) Execute 

func (e *AgentExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

type AgentRunOpts 

type AgentRunOpts struct {
	Model          string
	TimeoutSeconds int
	MaxIterations  int
	SkipContext    bool // skip MemDB context fetch (default true for workflow steps)
}

AgentRunOpts configures an agent step execution.

type AgentRunner 

type AgentRunner interface {
	RunTask(ctx context.Context, task string, sessionKey string, opts AgentRunOpts) (string, error)
}

AgentRunner is the interface for delegating tasks to the full agent loop. Satisfied by agent.WorkflowAgentAdapter.

type ApprovalExecutor 

type ApprovalExecutor struct{}

ApprovalExecutor transitions the workflow to waiting_approval state. Actual approval handling is done externally via Engine.HandleApproval.

func NewApprovalExecutor 

func NewApprovalExecutor() *ApprovalExecutor

func (*ApprovalExecutor) Execute 

func (e *ApprovalExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

type ApprovalNotifier 

type ApprovalNotifier func(wf *Workflow, step *Step)

ApprovalNotifier is called when a workflow needs user approval.

type BranchAllExecutor 

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

BranchAllExecutor fans out N branches in parallel. Config: 

{
  "branches": [
    {"id": "b1", "kind": "tool", "config": {"tool": "fetch_news"}},
    {"id": "b2", "kind": "llm",  "config": {"prompt": "..."}}
  ]
}

Each branch becomes an independent child step. All run in parallel. Results are collected into `wf.Context[parentID]` as `map[branchID]result`.

func NewBranchAllExecutor 

func NewBranchAllExecutor(engine *Engine) *BranchAllExecutor

NewBranchAllExecutor creates a BranchAll executor.

func (*BranchAllExecutor) Execute 

func (e *BranchAllExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

type CompletionNotifier 

type CompletionNotifier func(wf *Workflow)

CompletionNotifier is called when a workflow reaches a terminal state.

type ConcurrencyLimit 

type ConcurrencyLimit struct {
	Key           string `db:"key"`
	MaxConcurrent int    `db:"max_concurrent"`
	CurrentCount  int    `db:"current_count"`
}

ConcurrencyLimit defines a concurrency constraint.

type ConditionExecutor 

type ConditionExecutor struct{}

ConditionExecutor evaluates a simple expression and skips or proceeds. Config: {"check": "stepID", "contains": "keyword"}

Operators: "contains", "equals", "not_empty" (bool). 

"not_empty" treats "", "<nil>", "[]", "[ ]", "{}", and "null" as empty.

If "stop_workflow" is true and the condition evaluates to false, the step returns an error which stops the workflow immediately instead of silently passing "false" to downstream steps.

func NewConditionExecutor 

func NewConditionExecutor() *ConditionExecutor

func (*ConditionExecutor) Execute 

func (e *ConditionExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

type Engine 

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

Engine orchestrates workflow execution: DAG resolution, step dispatch, persistence.

func NewEngine 

func NewEngine(store *WorkflowStore, opts ...EngineOption) *Engine

NewEngine creates a workflow engine with functional options. The default executor set covers the dependency-free step kinds (condition, approval, transform). Other kinds are registered by the corresponding WithX option: WithLLMProvider for LLM/vision, WithToolRunner / WithMCPServers for tool, WithImageRenderer for image, etc. Sub-workflow / foreach / branchall / suspend / noop executors are wired automatically here since they need a pointer back to the engine.

func (*Engine) Advance 

func (e *Engine) Advance(ctx context.Context, workflowID string) (bool, error)

Advance finds all runnable steps (all deps completed) and executes them. Independent steps run in parallel. Returns true if any step was executed.

Special-case: when the workflow is in StateFailed, Advance still drains any pending AlwaysRun steps whose deps have reached a terminal state. Cancelled, Completed, WaitingApproval, and Paused short-circuit unconditionally.

func (*Engine) AutoRetryFailed 

func (e *Engine) AutoRetryFailed(maxAge time.Duration) int

AutoRetryFailed checks recently failed workflows for transient errors and retries them. Returns the number of workflows retried. Only retries workflows that failed within maxAge.

func (*Engine) Cancel 

func (e *Engine) Cancel(workflowID string) error

Cancel cancels a running or paused workflow.

func (*Engine) DetectStalled 

func (e *Engine) DetectStalled(threshold time.Duration) []StalledWorkflow

DetectStalled finds workflows with steps that have been running longer than the threshold. Default threshold: 15 minutes. Returns stalled workflow info for monitoring/alerting.

func (*Engine) GracefulShutdown 

func (e *Engine) GracefulShutdown(timeout time.Duration) int

GracefulShutdown pauses all workflows and stops the watchdog.

func (*Engine) HandleApproval 

func (e *Engine) HandleApproval(workflowID string, approved bool) error

HandleApproval resumes or rejects a workflow waiting for approval.

func (*Engine) HandleApprovalWithData 

func (e *Engine) HandleApprovalWithData(workflowID string, approved bool, data map[string]any) error

HandleApprovalWithData resumes a workflow with structured data from the approver. Data is stored in wf.Context[stepID] for downstream steps to consume via $steps.{id}.result. If data is nil, falls back to approvalResult string (same as HandleApproval).

func (*Engine) InjectStepsAndRewriteDeps 

func (e *Engine) InjectStepsAndRewriteDeps(workflowID string, steps []Step, afterStepID, newDepID string) error

InjectStepsAndRewriteDeps atomically adds child steps after a parent step and rewrites dependencies.

func (*Engine) ListenForResults 

func (e *Engine) ListenForResults(ctx context.Context, listener *StepListener)

ListenForResults listens for step_done notifications and advances workflows. Blocks until ctx is cancelled. Run in a goroutine.

func (*Engine) PauseAll 

func (e *Engine) PauseAll() int

PauseAll pauses all running workflows. Used for graceful shutdown.

func (*Engine) RecoverAll 

func (e *Engine) RecoverAll(ctx context.Context) []string

RecoverAll finds workflows stuck in StateRunning at startup (sign of a crash) and resumes them. Running steps are reset to pending.

func (*Engine) RecoverStalled 

func (e *Engine) RecoverStalled(threshold time.Duration) int

RecoverStalled attempts to recover stalled workflows by failing the hung step and letting the engine's error handling (retry/on_error) take over. Returns the number of workflows recovered.

func (*Engine) RegisterWebhooks 

func (e *Engine) RegisterWebhooks(mux *http.ServeMux, runtime *TemplateRuntime, triggers []WebhookTrigger) error

RegisterWebhooks attaches every trigger to mux as a POST handler at its Path. Returns an error on misconfiguration (duplicate path, missing secret for an auth mode that needs it, etc.). Workflows start asynchronously — the HTTP response returns 202 with the workflow ID before the workflow finishes.

func (*Engine) Reopen added in v0.17.0 

func (e *Engine) Reopen(workflowID string) error

Reopen transitions a cancelled workflow back to waiting_approval, provided it is safely resumable — i.e. its CurrentStep points at a pending approval step (the gate it was waiting on when cancelled; Cancel only touches workflow-level State/Error, so that step is still StepPending). This is the inverse of Cancel for the "stop parking this, needs a human decision" cleanup pattern: once the human resolves the blocking issue, the caller can Reopen and then proceed with a normal HandleApproval/HandleApprovalWithData on the SAME workflow.

It does NOT touch step states — the pending approval step is already correctly StepPending, ready for HandleApproval right after. Step selection uses the workflow's own CurrentStep field (the authoritative "which step am I actually blocked on" signal) rather than counting pending approval steps: every not-yet-reached approval step defaults to StepPending from creation, so a count-based scan cannot distinguish "the one it's blocked on" from "future placeholder pendings" and would refuse any workflow cancelled before its final approval gate. If CurrentStep is empty, or points at a step that is missing / not a pending approval step, there is nothing to reopen INTO and an error is returned.

func (*Engine) ResumeAll 

func (e *Engine) ResumeAll(ctx context.Context) []string

ResumeAll resumes all paused workflows. Used after restart.

func (*Engine) ResumeAsync 

func (e *Engine) ResumeAsync(_ context.Context, workflowID string)

ResumeAsync resumes a running workflow in a background goroutine. Used after approval or any state change that allows continued execution. The caller's context is intentionally ignored — the workflow must outlive the HTTP handler or MCP tool call that triggered the resume.

func (*Engine) RunStep 

func (e *Engine) RunStep(ctx context.Context, workflowID, stepID string) error

RunStep executes a single step, updating state and persisting. All state mutations go through store.Modify to be concurrent-safe.

func (*Engine) RunToCompletion 

func (e *Engine) RunToCompletion(ctx context.Context, workflowID string) error

RunToCompletion runs Advance in a loop until the workflow stops (completed, failed, approval, paused).

When Advance returns an error AND the workflow has entered StateFailed but still has runnable AlwaysRun steps, we keep iterating so cleanup steps drain. The original error is preserved and returned once draining finishes.

func (*Engine) SetA2ACaller 

func (e *Engine) SetA2ACaller(caller A2ACaller)

func (*Engine) SetAgentRunner 

func (e *Engine) SetAgentRunner(runner AgentRunner)

func (*Engine) SetApprovalNotifier 

func (e *Engine) SetApprovalNotifier(fn ApprovalNotifier)

func (*Engine) SetCompletionNotifier 

func (e *Engine) SetCompletionNotifier(fn CompletionNotifier)

func (*Engine) SetHooks 

func (e *Engine) SetHooks(h HookPublisher)

func (*Engine) SetSkills 

func (e *Engine) SetSkills(sr SkillResolver)

func (*Engine) Start 

func (e *Engine) Start(ctx context.Context, workflowID string) error

Start sets a pending workflow to running and begins synchronous execution. Blocks until the workflow reaches a terminal or paused state.

func (*Engine) StartAsync 

func (e *Engine) StartAsync(ctx context.Context, workflowID string) error

StartAsync sets a pending workflow to running and begins execution in a background goroutine. Returns immediately after starting. The CompletionNotifier is called when the workflow finishes.

func (*Engine) StartWatchdog 

func (e *Engine) StartWatchdog(interval time.Duration)

StartWatchdog begins a background goroutine that periodically: 1. Recovers stalled workflows (running > 15min) 2. Auto-retries recently failed workflows with transient errors Call StopWatchdog to shut it down gracefully.

func (*Engine) StopWatchdog 

func (e *Engine) StopWatchdog()

StopWatchdog stops the background watchdog goroutine.

func (*Engine) Store 

func (e *Engine) Store() *WorkflowStore

func (*Engine) ValidateTemplate 

func (e *Engine) ValidateTemplate(t *Template) error

ValidateTemplate reports an error if the template references a step kind for which no executor is registered on this engine. The error message names the first missing kind and hints at the With*Provider option that would register it. Use this at template-load or workflow-create time so authoring + wiring gaps surface BEFORE the workflow ever runs.

Returns nil for nil templates (caller's responsibility to nil-check).

Note: ValidateWorkflow above validates an already-instantiated Workflow (DAG cycles, unknown tool refs, retry config). ValidateTemplate is the one-step-earlier check — does the engine even have executors for the kinds this template will need?

func (*Engine) ValidateWorkflow 

func (e *Engine) ValidateWorkflow(wf *Workflow) []ValidationError

ValidateWorkflow performs a dry-run validation of a workflow without executing it. Catches: missing deps, invalid step kinds, unknown tools, DAG cycles, empty configs. Returns nil if the workflow is valid.

type EngineOption 

type EngineOption func(*Engine)

EngineOption configures an Engine. Options apply during NewEngine; some options also have post-creation Set* twins on Engine for late binding (see engine_setters.go).

func WithA2ACaller 

func WithA2ACaller(c A2ACaller) EngineOption

func WithAgentRunner 

func WithAgentRunner(a AgentRunner) EngineOption

func WithApprovalNotifier 

func WithApprovalNotifier(fn ApprovalNotifier) EngineOption

func WithBudget 

func WithBudget(maxUSD float64) EngineOption

WithBudget sets a maximum USD ceiling per workflow. When exceeded, the next cost-bearing step returns ErrBudgetExceeded and the workflow fails. 0 (default) means unlimited.

func WithCompletionNotifier 

func WithCompletionNotifier(fn CompletionNotifier) EngineOption

func WithCostModel 

func WithCostModel(model map[string]ModelPrice) EngineOption

WithCostModel overrides the default per-model USD pricing table. Useful to add new models or apply discounted contract pricing. Map is shallow-copied — caller can mutate after.

func WithDispatcher 

func WithDispatcher(d StepDispatcher) EngineOption

WithDispatcher sets a custom step dispatcher on the engine.

func WithEventLog 

func WithEventLog(el *EventLog) EngineOption

func WithHookPublisher 

func WithHookPublisher(h HookPublisher) EngineOption

func WithImageRenderer 

func WithImageRenderer(r ImageRenderer) EngineOption

WithImageRenderer registers a backend renderer for the StepImage primitive. When set, the engine accepts steps of kind "image". Without it, image steps are rejected at validation time with an "unknown step kind" error.

Apply this option BEFORE WithImageWorkspace — the workspace option mutates the executor created here, so it must already exist.

func WithImageWorkspace 

func WithImageWorkspace(dir string) EngineOption

WithImageWorkspace makes the image executor persist rendered bytes to disk under the given directory. Each rendered step writes <workspaceDir>/<workflow_id>/<step_id>.<ext> and the path appears in the step's result map for downstream reference.

Must be applied AFTER WithImageRenderer; it is a no-op when the image executor has not been registered.

func WithLLMClient 

func WithLLMClient(c *llm.Client) EngineOption

func WithLLMProvider 

func WithLLMProvider(p LLMProvider) EngineOption

func WithLogger 

func WithLogger(l *slog.Logger) EngineOption

func WithMCPServerHeaders 

func WithMCPServerHeaders(serverID string, headers map[string]string) EngineOption

WithMCPServerHeaders sets HTTP headers (e.g. Authorization) for a specific MCP server. Must be called after WithMCPServers.

func WithMCPServers 

func WithMCPServers(servers map[string]string) EngineOption

func WithMessagePublisher 

func WithMessagePublisher(m MessagePublisher) EngineOption

func WithMetrics 

func WithMetrics(m *Metrics) EngineOption

func WithRateLimit 

func WithRateLimit(provider string, rate float64, burst int) EngineOption

WithRateLimit registers a token-bucket QPS limiter for the named provider/tool. rate is the sustained token rate (tokens per second); burst is the bucket size. Calls that would exceed the limit return a transient error so existing step retry machinery can back off and retry. Default is unlimited (no WithRateLimit).

Example: 

engine := NewEngine(store,
    WithRateLimit("claude-3-5-sonnet-20241022", 5, 10),
    WithRateLimit("my-tool", 2, 4),
)

func WithScheduler 

func WithScheduler(s *Scheduler) EngineOption

func WithSkillResolver 

func WithSkillResolver(s SkillResolver) EngineOption

func WithStepCache 

func WithStepCache(cache StepCache) EngineOption

WithStepCache plugs in a cache backend. When unset, the engine never looks up or writes cache entries.

func WithStepCacheKinds 

func WithStepCacheKinds(kinds ...StepKind) EngineOption

WithStepCacheKinds restricts caching to a specific set of step kinds. Defaults to StepTransform when WithStepCache is set without this option — transforms are pure data shaping and always safe to cache.

Caller is responsible for confirming determinism of any kind they enable (e.g. StepLLM with temperature=0).

func WithStepListener 

func WithStepListener(l *StepListener) EngineOption

func WithStreamCallback 

func WithStreamCallback(cb StreamCallback) EngineOption

func WithToolRunner 

func WithToolRunner(t ToolRunner) EngineOption

func WithTracerProvider 

func WithTracerProvider(tp trace.TracerProvider) EngineOption

WithTracerProvider opts the engine into OpenTelemetry tracing. Every workflow gets a parent span; every step gets a child span with cost, duration, and outcome attributes. When unset, all tracing helpers no-op and observable behaviour is identical to v0.10.x.

Operators wire any OTel-compatible exporter (Jaeger, Tempo, Honeycomb, Datadog, OTLP) via the standard SDK and pass the resulting trace.TracerProvider here.

func WithTriggers 

func WithTriggers(ts *TriggerService) EngineOption

func WithVisionProvider 

func WithVisionProvider(p LLMProvider) EngineOption

WithVisionProvider registers a multimodal LLM provider for the StepVision primitive. Use this when a separate vision-capable provider is desired (e.g. Claude Opus for vision, Sonnet for general LLM steps), or when the same provider should serve both LLM and vision steps. When the same provider is passed to both WithLLMProvider and WithVisionProvider, the later option wins for the vision executor.

type Event 

type Event struct {
	Type       EventType      `json:"type"`
	WorkflowID string         `json:"workflow_id"`
	StepID     string         `json:"step_id,omitempty"`
	StepKind   string         `json:"step_kind,omitempty"`
	Timestamp  int64          `json:"ts"`
	DurationMS int64          `json:"duration_ms,omitempty"`
	Error      string         `json:"error,omitempty"`
	Data       map[string]any `json:"data,omitempty"`
}

Event is a single entry in the structured event log.

func LoadEventLog 

func LoadEventLog(path string) ([]Event, error)

LoadEventLog reads a JSONL event log file and returns all events.

type EventLog 

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

EventLog is an append-only JSONL writer for workflow execution events.

func NewEventLog 

func NewEventLog(dir string) (*EventLog, error)

NewEventLog creates an event log backed by JSONL files in the given directory.

func (*EventLog) Append 

func (el *EventLog) Append(e Event) error

Append writes an event to the workflow's JSONL file.

func (*EventLog) Path 

func (el *EventLog) Path(workflowID string) string

Path returns the JSONL file path for a given workflow.

type EventTrigger 

type EventTrigger struct {
	ID        string            `json:"id"`
	Name      string            `json:"name"`
	Enabled   bool              `json:"enabled"`
	Event     string            `json:"event"`
	Filter    map[string]string `json:"filter"`
	Action    TriggerAction     `json:"action"`
	CreatedAt int64             `json:"created_at_ms"`
}

EventTrigger fires a TriggerAction when a matching hook event occurs.

type EventType 

type EventType string

EventType identifies the kind of event in the log. 

const (
	EventStepStarted  EventType = "step_started"
	EventStepFinished EventType = "step_completed"
	EventStepFailed   EventType = "step_failed"
	EventStepRetried  EventType = "step_retried"
	EventWFStarted    EventType = "workflow_started"
	EventWFCompleted  EventType = "workflow_completed"
	EventWFFailed     EventType = "workflow_failed"
)

type ExecutionTrace 

type ExecutionTrace struct {
	WorkflowID string      `json:"workflow_id"`
	Steps      []StepTrace `json:"steps"`
	TotalMS    int64       `json:"total_ms"`
	Error      string      `json:"error,omitempty"`
	StartedAt  int64       `json:"started_at"`
	EndedAt    int64       `json:"ended_at"`
}

ExecutionTrace is a reconstructed execution from event log.

func ReplayTrace 

func ReplayTrace(events []Event) *ExecutionTrace

ReplayTrace reconstructs an execution trace from a sequence of events.

type FileCache 

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

FileCache stores entries as JSON files under a directory keyed by hash. Persists across processes; safe to share between sibling workers but does not provide locking — use InMemoryCache + dispatcher coordination if you need single-flight semantics.

func NewFileCache 

func NewFileCache(dir string) (*FileCache, error)

NewFileCache returns a file-backed cache rooted at dir. Creates the dir lazily on first Put. Returns an error only if dir is empty.

func (*FileCache) Get 

func (c *FileCache) Get(_ context.Context, key string) (StepCacheEntry, bool, error)

func (*FileCache) Put 

func (c *FileCache) Put(_ context.Context, key string, entry StepCacheEntry) error

type ForEachExecutor 

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

ForEachExecutor iterates over a list and injects child steps for each item. Config: 

{
  "items": "step_id_that_produced_list",
  "step_kind": "tool",
  "step_config": {"tool": "process_item"},
  "sequential": false,
  "concurrency": 0
}

Each child step gets `{"item": <value>, "index": <int>}` merged into its config. Results are collected into `wf.Context[parentID]` as `[]any`.

func NewForEachExecutor 

func NewForEachExecutor(engine *Engine) *ForEachExecutor

NewForEachExecutor creates a ForEach executor.

func (*ForEachExecutor) Execute 

func (e *ForEachExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

type HookPublisher 

type HookPublisher interface {
	Fire(event string, data map[string]any) int
}

HookPublisher fires lifecycle events. Satisfied by hooks.Registry.

type ImageExecutor 

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

ImageExecutor implements StepExecutor for StepImage. It validates step config, resolves HTML/font references against the workflow context, calls a pluggable ImageRenderer, optionally persists the bytes to disk, and exposes a result map downstream via wf.Context[stepID].

func NewImageExecutor 

func NewImageExecutor(r ImageRenderer, metrics *Metrics) *ImageExecutor

NewImageExecutor builds an ImageExecutor wired to the given renderer + metrics. metrics may be nil; counters are guarded.

func (*ImageExecutor) Execute 

func (e *ImageExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

Execute renders an image for the given step.

type ImageRenderRequest 

type ImageRenderRequest struct {
	HTML    string   // required
	Width   int      // pixels, default 1200
	Height  int      // pixels, default 630
	Format  string   // "png" | "jpeg" | "webp" | "svg"; default "png"
	Density int      // 1 | 2 | 3; default 1
	Fonts   []string // optional, names of pre-bundled or URL-loaded fonts
}

ImageRenderRequest captures everything an image step config can carry.

type ImageRenderResult 

type ImageRenderResult struct {
	Bytes      []byte
	MIMEType   string
	SizeBytes  int64
	DurationMS int64
	// Path is set by the executor when the image is persisted to disk;
	// useful so downstream steps can reference the file. Empty when caller
	// wants in-memory bytes only.
	Path string
}

ImageRenderResult is what the renderer returns.

type ImageRenderer 

type ImageRenderer interface {
	Render(ctx context.Context, req ImageRenderRequest) (ImageRenderResult, error)
}

ImageRenderer is the pluggable backend that turns HTML+geometry into image bytes. Implementations may include a satori-render HTTP adapter, an in-memory mock for tests, or a wkhtmltoimage shell wrapper. The engine never speaks to the renderer directly except through this interface.

type InMemoryCache 

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

InMemoryCache is a process-local map-backed StepCache. Suitable for unit tests and single-process deployments where re-running a workflow inside the same process is the cache-hit pathway.

func NewInMemoryCache 

func NewInMemoryCache() *InMemoryCache

NewInMemoryCache returns a fresh in-memory cache.

func (*InMemoryCache) Get 

func (c *InMemoryCache) Get(_ context.Context, key string) (StepCacheEntry, bool, error)

func (*InMemoryCache) Put 

func (c *InMemoryCache) Put(_ context.Context, key string, entry StepCacheEntry) error

type Job 

type Job struct {
	ID             string     `json:"id"`
	Name           string     `json:"name"`
	Enabled        bool       `json:"enabled"`
	Schedule       Schedule   `json:"schedule"`
	Payload        JobPayload `json:"payload"`
	State          JobState   `json:"state"`
	CreatedAtMS    int64      `json:"created_at_ms"`
	UpdatedAtMS    int64      `json:"updated_at_ms"`
	DeleteAfterRun bool       `json:"delete_after_run"`
}

Job is a scheduled task.

type JobHandler 

type JobHandler func(job *Job) (string, error)

JobHandler is called when a scheduled job is due.

type JobPayload 

type JobPayload struct {
	Kind       string         `json:"kind"` // "workflow", "message", or custom
	WorkflowID string         `json:"workflow_id,omitempty"`
	TemplateID string         `json:"template_id,omitempty"`
	Message    string         `json:"message,omitempty"`
	Params     map[string]any `json:"params,omitempty"`
}

JobPayload defines what happens when a job fires.

type JobState 

type JobState struct {
	NextRunAtMS *int64 `json:"next_run_at_ms,omitempty"`
	LastRunAtMS *int64 `json:"last_run_at_ms,omitempty"`
	LastStatus  string `json:"last_status,omitempty"`
	LastError   string `json:"last_error,omitempty"`
}

JobState tracks execution state of a scheduled job.

type LLMExecutor 

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

LLMExecutor sends a prompt to the LLM provider and stores the response. Supports both legacy LLMProvider interface and go-kit *llm.Client (preferred). Supports skill references: {"skill": "name", "input": "..."}. Supports multi-turn tool calling when tools are configured and a ToolRunner is set.

func NewLLMExecutor 

func NewLLMExecutor(provider LLMProvider, metrics *Metrics) *LLMExecutor

NewLLMExecutor creates an LLMExecutor using the legacy LLMProvider interface.

func NewLLMExecutorWithClient 

func NewLLMExecutorWithClient(client *llm.Client, metrics *Metrics) *LLMExecutor

NewLLMExecutorWithClient creates an LLMExecutor using the go-kit LLM client.

func (*LLMExecutor) Execute 

func (e *LLMExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

func (*LLMExecutor) SetSkills 

func (e *LLMExecutor) SetSkills(sr SkillResolver)

SetSkills sets the skill resolver for skill-aware LLM steps. Nil-safe.

func (*LLMExecutor) SetStreamCallback 

func (e *LLMExecutor) SetStreamCallback(cb StreamCallback)

SetStreamCallback sets the callback for streaming LLM chunks.

func (*LLMExecutor) SetToolRunner 

func (e *LLMExecutor) SetToolRunner(tr ToolRunner)

SetToolRunner sets the tool runner for multi-turn tool calling.

type LLMImageContent 

type LLMImageContent struct {
	Bytes    []byte
	MIMEType string
}

LLMImageContent is an inline image payload attached to an LLMMessage. MIMEType examples: "image/png", "image/jpeg", "image/webp".

type LLMMessage 

type LLMMessage struct {
	Role    string
	Content string
	Images  []LLMImageContent
}

LLMMessage is a single message in a conversation.

Images is an optional list of inline image attachments for multimodal providers. Text-only providers ignore this field; multimodal providers (Anthropic Claude, OpenAI GPT-4V, etc.) consume the bytes alongside the Content text. The field is additive — existing callers that leave it nil continue to behave as before.

type LLMProvider 

type LLMProvider interface {
	Chat(ctx context.Context, messages []LLMMessage, model string) (*LLMResponse, error)
	GetDefaultModel() string
}

LLMProvider sends prompts to a language model (replaces providers.LLMProvider).

type LLMResponse 

type LLMResponse struct {
	Content      string
	Model        string
	InputTokens  int
	OutputTokens int
}

LLMResponse is the model's reply with optional token usage data.

type LocalDispatcher 

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

LocalDispatcher runs steps directly on the embedded engine (in-process). This preserves the original single-node behavior.

func NewLocalDispatcher 

func NewLocalDispatcher(engine *Engine) *LocalDispatcher

NewLocalDispatcher creates a dispatcher that calls engine.RunStep directly.

func (*LocalDispatcher) Dispatch 

func (d *LocalDispatcher) Dispatch(ctx context.Context, workflowID, stepID string, _ StepKind) error

Dispatch executes a single step synchronously via the engine.

func (*LocalDispatcher) DispatchBatch 

func (d *LocalDispatcher) DispatchBatch(ctx context.Context, workflowID string, stepIDs []string, _ []StepKind) error

DispatchBatch executes multiple steps concurrently via engine.runParallel. Returns the first error encountered.

type MCPDeps 

type MCPDeps struct {
	Engine    *Engine
	Templates *TemplateStore
}

MCPDeps holds the dependencies required by MCP tool handlers.

type MCPToolRunner 

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

MCPToolRunner implements ToolRunner for remote MCP servers. Supports multiple servers with lazy connect and auto-discovery of tool names.

func NewMCPToolRunner 

func NewMCPToolRunner(servers map[string]string) *MCPToolRunner

NewMCPToolRunner creates a runner for the given MCP servers. Connections are established lazily on first use or via Connect.

func (*MCPToolRunner) Close 

func (r *MCPToolRunner) Close() error

Close closes all MCP sessions.

func (*MCPToolRunner) Connect 

func (r *MCPToolRunner) Connect(ctx context.Context) error

Connect explicitly connects to all servers and discovers their tools.

func (*MCPToolRunner) Execute 

func (r *MCPToolRunner) Execute(ctx context.Context, name string, args map[string]any) (string, error)

Execute calls a tool on the appropriate MCP server.

func (*MCPToolRunner) SetHeaders 

func (r *MCPToolRunner) SetHeaders(serverID string, headers map[string]string)

SetHeaders sets HTTP headers for a specific server (e.g. Authorization).

func (*MCPToolRunner) Tools 

func (r *MCPToolRunner) Tools() map[string][]string

Tools returns discovered tool names grouped by server ID.

type MessageExecutor 

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

MessageExecutor publishes a message to the bus for delivery to the user.

func NewMessageExecutor 

func NewMessageExecutor(bus MessagePublisher) *MessageExecutor

func (*MessageExecutor) Execute 

func (e *MessageExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

type MessagePublisher 

type MessagePublisher interface {
	PublishOutbound(msg OutboundMessage)
}

MessagePublisher delivers messages to users (replaces *bus.MessageBus).

type Metrics 

type Metrics struct {
	WorkflowsCreated      atomic.Int64
	WorkflowsCompleted    atomic.Int64
	WorkflowsFailed       atomic.Int64
	WorkflowsCancelled    atomic.Int64
	StepsExecuted         atomic.Int64
	StepsRetried          atomic.Int64
	StepsSkipped          atomic.Int64
	StepsDeadLettered     atomic.Int64
	ApprovalsPending      atomic.Int64
	AgentStepsExecuted    atomic.Int64
	AgentStepsFailed      atomic.Int64
	A2AStepsExecuted      atomic.Int64
	A2AStepsFailed        atomic.Int64
	HooksFired            atomic.Int64
	TriggersEvaluated     atomic.Int64
	TriggersFired         atomic.Int64
	SchedulerJobsExecuted atomic.Int64
	SchedulerJobsFailed   atomic.Int64
	LLMTokensInput        atomic.Int64
	LLMTokensOutput       atomic.Int64
	ImageRendersSuccess   atomic.Int64
	ImageRendersFailed    atomic.Int64
	ImageBytesTotal       atomic.Int64
	ImageDurationMSTotal  atomic.Int64
	VisionCallsSuccess    atomic.Int64
	VisionCallsFailed     atomic.Int64
	VisionTokensInput     atomic.Int64
	VisionTokensOutput    atomic.Int64
	// WorkflowCostUSDTotal stores USD as micro-cents (USD * 1,000,000,
	// rounded) because atomic.Float64 does not exist. Microcents preserve
	// sub-cent precision so cheap-model calls (Haiku/Flash) still move
	// the metric. Divide by 1,000,000 for USD.
	WorkflowCostUSDTotal        atomic.Uint64
	WorkflowTokensInputTotal    atomic.Int64
	WorkflowTokensOutputTotal   atomic.Int64
	WorkflowImagesRenderedTotal atomic.Int64
	WorkflowBudgetExceededTotal atomic.Int64
	WebhooksReceived            atomic.Int64
	WebhooksRejected            atomic.Int64
	StepCacheHits               atomic.Int64
	StepCacheMisses             atomic.Int64
}

Metrics tracks workflow execution counters.

func NewMetrics 

func NewMetrics() *Metrics

NewMetrics creates a fresh Metrics instance for dependency injection.

func (*Metrics) Reset 

func (m *Metrics) Reset()

Reset zeroes all counters.

func (*Metrics) Summary 

func (m *Metrics) Summary() string

Summary returns a formatted string of all metrics.

type ModelPrice 

type ModelPrice struct {
	InputUSDPer1K  float64
	OutputUSDPer1K float64
}

ModelPrice is per-1k-tokens USD pricing for an LLM model.

type MultiToolRunner 

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

MultiToolRunner routes tool calls to registered runners. Runners are checked in registration order; the first match wins.

func NewMultiToolRunner 

func NewMultiToolRunner(runners ...ToolRunner) *MultiToolRunner

NewMultiToolRunner creates a runner with the given runners registered as fallbacks. Fallback runners accept any tool name not matched by earlier runners.

func (*MultiToolRunner) Execute 

func (m *MultiToolRunner) Execute(ctx context.Context, name string, args map[string]any) (string, error)

Execute routes the tool call to the first matching runner.

func (*MultiToolRunner) Register 

func (m *MultiToolRunner) Register(runner ToolRunner, tools ...string)

Register adds a runner that handles only the specified tools. If no tools are specified, the runner is registered as a fallback.

type N8nConnectionTarget 

type N8nConnectionTarget struct {
	Node  string `json:"node"`
	Type  string `json:"type"`
	Index int    `json:"index"`
}

N8nConnectionTarget identifies a downstream node.

type N8nNode 

type N8nNode struct {
	ID               string         `json:"id"`
	Name             string         `json:"name"`
	Type             string         `json:"type"`
	TypeVersion      int            `json:"typeVersion,omitempty"`
	Parameters       map[string]any `json:"parameters"`
	Position         [2]int         `json:"position,omitempty"`
	RetryOnFail      bool           `json:"retryOnFail,omitempty"`
	MaxTries         int            `json:"maxTries,omitempty"`
	WaitBetweenTries int            `json:"waitBetweenTries,omitempty"`
	ContinueOnFail   bool           `json:"continueOnFail,omitempty"`
}

N8nNode represents a single node in an n8n workflow.

type N8nNodeConnection 

type N8nNodeConnection struct {
	Main [][]N8nConnectionTarget `json:"main"`
}

N8nNodeConnection holds the output connections for a node. Format: {"main": [[{"node": "Name", "type": "main", "index": 0}]]}

type N8nTag 

type N8nTag struct {
	Name string `json:"name"`
}

N8nTag is a workflow tag.

type N8nWorkflow 

type N8nWorkflow struct {
	Name        string                       `json:"name"`
	Nodes       []N8nNode                    `json:"nodes"`
	Connections map[string]N8nNodeConnection `json:"connections"`
	Settings    map[string]any               `json:"settings,omitempty"`
	Tags        []N8nTag                     `json:"tags,omitempty"`
}

N8nWorkflow represents a complete n8n workflow JSON export.

type NoopExecutor 

type NoopExecutor struct{}

NoopExecutor completes immediately with 'ok' result. Useful as a join step.

func (*NoopExecutor) Execute 

func (e *NoopExecutor) Execute(_ context.Context, step *Step, _ *Workflow) error

type OutboundMessage 

type OutboundMessage struct {
	Channel string
	ChatID  string
	Content string
}

OutboundMessage is a message to be delivered to a user channel.

type ParamSpec 

type ParamSpec struct {
	Type        string `json:"type"` // ParamTypeString|ParamTypeInt|ParamTypeBool|ParamTypeFloat|ParamTypeObject|ParamTypeArray
	Description string `json:"description,omitempty"`
	Default     any    `json:"default,omitempty"`
	Required    bool   `json:"required,omitempty"`
	Enum        []any  `json:"enum,omitempty"`
}

ParamSpec declares the type, default, and validation rules for a template parameter. Replaces the legacy free-text-description form where Params was map[string]string.

JSON forms accepted: 

Legacy:  "name": "Description text"          → {Type:"string", Description:"..."}
Typed:   "name": {"type":"int", "default":6}  → {Type:"int", Default:6}

type ParamsMap 

type ParamsMap map[string]ParamSpec

ParamsMap is a map from param name to ParamSpec. It accepts both the legacy string form ("name": "description") and the new typed object form ("name": {"type": "int", "default": 6}).

func (ParamsMap) MarshalJSON 

func (p ParamsMap) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler. It preserves the legacy wire shape for trivially-legacy entries (Type=string, no default/required/enum) by emitting a bare string (the Description). This keeps backward-compat for consumers that read params as map[string]string.

Non-trivial entries (typed, required, enumerated, or with a default) are marshaled as the full {"type":"...","description":"...",...} object form.

func (*ParamsMap) UnmarshalJSON 

func (p *ParamsMap) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler. It handles both:

  • Legacy string value: "paramName": "description text" → ParamSpec{Type: "string", Description: "description text"}
  • Typed object value: "paramName": {"type": "int", "default": 6} → parsed directly into ParamSpec

type PostgresDispatcher 

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

PostgresDispatcher implements StepDispatcher by enqueuing steps to the step_queue table for distributed worker execution.

func NewPostgresDispatcher 

func NewPostgresDispatcher(queue StepEnqueuer, closers ...io.Closer) *PostgresDispatcher

NewPostgresDispatcher creates a dispatcher from a StepEnqueuer. Additional closers (e.g. ConcurrencyLimiter) can be passed for lifecycle management.

func (*PostgresDispatcher) Close 

func (d *PostgresDispatcher) Close() error

Close releases the queue and any additional closers.

func (*PostgresDispatcher) Dispatch 

func (d *PostgresDispatcher) Dispatch(_ context.Context, workflowID, stepID string, kind StepKind) error

Dispatch enqueues a single step to the step_queue table.

func (*PostgresDispatcher) DispatchBatch 

func (d *PostgresDispatcher) DispatchBatch(
	_ context.Context, workflowID string, stepIDs []string, kinds []StepKind,
) error

DispatchBatch enqueues multiple steps to the step_queue table.

type QueueItem 

type QueueItem struct {
	ID             int64          `db:"id"`
	WorkflowID     string         `db:"workflow_id"`
	StepID         string         `db:"step_id"`
	StepKind       string         `db:"step_kind"`
	ConcurrencyKey string         `db:"concurrency_key"`
	Priority       int            `db:"priority"`
	State          QueueItemState `db:"state"`
	WorkerID       string         `db:"worker_id"`
	ClaimedAt      *time.Time     `db:"claimed_at"`
	HeartbeatAt    *time.Time     `db:"heartbeat_at"`
	Result         []byte         `db:"result"`
	Error          string         `db:"error"`
	CreatedAt      time.Time      `db:"created_at"`
}

QueueItem represents a step waiting for or being executed by a worker.

type QueueItemState 

type QueueItemState string

QueueItemState represents the lifecycle of a queued step. 

const (
	QueuePending  QueueItemState = "pending"
	QueueClaimed  QueueItemState = "claimed"
	QueueDone     QueueItemState = "done"
	QueueFailed   QueueItemState = "failed"
	QueueTimedOut QueueItemState = "timed_out"
)

type Reaper 

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

Reaper periodically reclaims steps from dead workers.

func NewReaper 

func NewReaper(reaper StepReaper, heartbeatTimeout, checkInterval time.Duration) *Reaper

NewReaper creates a reaper that checks for stale items every interval.

func (*Reaper) Logger 

func (r *Reaper) Logger() *slog.Logger

Logger returns the reaper's logger.

func (*Reaper) Run 

func (r *Reaper) Run(ctx context.Context)

Run starts the reaper loop. Blocks until ctx is cancelled.

type ReducerKind 

type ReducerKind string

ReducerKind defines how a context key is updated when a step writes to it. 

const (
	ReducerReplace ReducerKind = "replace"
	ReducerAppend  ReducerKind = "append"
	ReducerSum     ReducerKind = "sum"
	ReducerMerge   ReducerKind = "merge"
)

type RetryAfterError 

type RetryAfterError interface {
	error
	RetryAfter() time.Duration
}

RetryAfterError is implemented by errors that carry a Retry-After hint. When the step machinery encounters this interface on a step error, it uses the returned duration as a floor for the retry delay (overriding the computed exponential backoff when larger).

type Schedule 

type Schedule struct {
	Kind    string `json:"kind"`               // "at", "every", "cron"
	AtMS    *int64 `json:"at_ms,omitempty"`    // for kind="at": one-shot unix ms
	EveryMS *int64 `json:"every_ms,omitempty"` // for kind="every": interval ms
	Expr    string `json:"expr,omitempty"`     // for kind="cron": 5-field expression
	TZ      string `json:"tz,omitempty"`       // timezone for cron (e.g. "Europe/Moscow")
}

Schedule defines when a job runs.

type Scheduler 

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

Scheduler manages time-based job scheduling with JSON file persistence.

func NewScheduler 

func NewScheduler(storePath string, handler JobHandler, opts ...SchedulerOption) *Scheduler

NewScheduler creates a scheduler backed by the given JSON file path.

func (*Scheduler) AddJob 

func (s *Scheduler) AddJob(name string, schedule Schedule, payload JobPayload) (*Job, error)

AddJob adds a new scheduled job.

func (*Scheduler) EnableJob 

func (s *Scheduler) EnableJob(id string, enabled bool) *Job

EnableJob enables or disables a job.

func (*Scheduler) ListJobs 

func (s *Scheduler) ListJobs(includeDisabled bool) []Job

ListJobs returns all jobs. If includeDisabled is false, only enabled jobs.

func (*Scheduler) RemoveJob 

func (s *Scheduler) RemoveJob(id string) bool

RemoveJob removes a job by ID.

func (*Scheduler) Start 

func (s *Scheduler) Start() error

Start begins the scheduler tick loop.

func (*Scheduler) Status 

func (s *Scheduler) Status() map[string]any

Status returns scheduler status summary.

func (*Scheduler) Stop 

func (s *Scheduler) Stop()

Stop halts the scheduler.

type SchedulerOption 

type SchedulerOption func(*Scheduler)

SchedulerOption configures a Scheduler.

func WithSchedulerLogger 

func WithSchedulerLogger(l *slog.Logger) SchedulerOption

WithSchedulerLogger sets the logger for the scheduler.

func WithSchedulerMetrics 

func WithSchedulerMetrics(m *Metrics) SchedulerOption

WithSchedulerMetrics sets the metrics instance for the scheduler.

type SecurityPolicy 

type SecurityPolicy struct {
	// MaxSteps is the maximum total steps a workflow can execute (including retries).
	// 0 = unlimited (default: 100).
	MaxSteps int `json:"max_steps,omitempty"`

	// MaxDuration is the maximum wall-clock time for the entire workflow.
	// 0 = unlimited (default: 30 minutes).
	MaxDuration time.Duration `json:"max_duration_ms,omitempty"`

	// MaxStepDuration is the maximum time for a single step execution.
	// 0 = unlimited (default: 10 minutes).
	MaxStepDuration time.Duration `json:"max_step_duration_ms,omitempty"`

	// AllowedTools restricts which tools can be called. Empty = all allowed.
	AllowedTools []string `json:"allowed_tools,omitempty"`

	// DeniedTools explicitly blocks specific tools. Takes precedence over AllowedTools.
	DeniedTools []string `json:"denied_tools,omitempty"`

	// AllowShell controls whether the exec/shell tool can be used.
	AllowShell bool `json:"allow_shell,omitempty"`

	// AllowNetwork controls whether HTTP request tools can be used.
	AllowNetwork bool `json:"allow_network,omitempty"`

	// SecretPatterns are regex patterns for values that should be masked in logs/results.
	SecretPatterns []string `json:"secret_patterns,omitempty"`
}

SecurityPolicy defines execution limits and constraints for a workflow.

func DefaultSecurityPolicy 

func DefaultSecurityPolicy() SecurityPolicy

DefaultSecurityPolicy returns a reasonable default policy.

func (SecurityPolicy) IsToolAllowed 

func (p SecurityPolicy) IsToolAllowed(toolName string) bool

IsToolAllowed checks if a tool name is permitted by this policy.

func (SecurityPolicy) Validate 

func (p SecurityPolicy) Validate() error

Validate checks the policy for consistency.

type SkillResolver 

type SkillResolver interface {
	LoadSkill(name string) (string, bool)
}

SkillResolver loads a skill prompt by name. Satisfied by skills.SkillsLoader.

type StalledWorkflow 

type StalledWorkflow struct {
	WorkflowID   string        `json:"workflow_id"`
	WorkflowName string        `json:"workflow_name"`
	StepID       string        `json:"step_id"`
	StepKind     StepKind      `json:"step_kind"`
	RunningFor   time.Duration `json:"running_for_ms"`
}

StalledWorkflow describes a workflow with a step that has been running too long.

type Step 

type Step struct {
	ID        string         `json:"id"`
	Kind      StepKind       `json:"kind"`
	Config    map[string]any `json:"config"`
	DependsOn []string       `json:"depends_on,omitempty"`
	State     StepState      `json:"state"`
	Result    any            `json:"result,omitempty"`
	Error     string         `json:"error,omitempty"`
	Retries   int            `json:"retries,omitempty"`
	StartedAt int64          `json:"started_at_ms,omitempty"`
	EndedAt   int64          `json:"ended_at_ms,omitempty"`
	// AlwaysRun marks the step for teardown-on-failure semantics. When true,
	// the step runs even after the workflow has entered StateFailed, provided
	// every direct dependency has reached a terminal state (completed, skipped,
	// failed, or dead-lettered). Dependencies still in StepPending block it
	// (so cleanup never runs with unmet inputs).
	AlwaysRun bool `json:"always_run,omitempty"`
}

Step is a single unit of work within a workflow.

func (*Step) GetBackoffMultiplier 

func (s *Step) GetBackoffMultiplier() float64

GetBackoffMultiplier returns the backoff multiplier from retry config, default 1.0 (no backoff).

func (*Step) GetMaxDelayMS 

func (s *Step) GetMaxDelayMS() int64

GetMaxDelayMS returns the max delay cap from retry config, default 0 (no cap).

func (*Step) GetOnError 

func (s *Step) GetOnError() string

GetOnError returns the error handling strategy: "fail" (default) or "skip".

func (*Step) GetRetryDelayMS 

func (s *Step) GetRetryDelayMS() int64

GetRetryDelayMS returns the retry delay from step config, default 1000ms.

func (*Step) GetRetryMax 

func (s *Step) GetRetryMax() int

GetRetryMax returns the max retries from step config, default 0.

func (*Step) GetRetryOn 

func (s *Step) GetRetryOn() []string

GetRetryOn returns patterns that must match for retry to happen. Empty = retry on any error.

func (*Step) GetSkipOn 

func (s *Step) GetSkipOn() []string

GetSkipOn returns patterns that skip retry if matched. Empty = never skip.

func (*Step) GetTimeoutMS 

func (s *Step) GetTimeoutMS() int64

GetTimeoutMS returns the per-step timeout from config, default 0 (no timeout).

type StepCache 

type StepCache interface {
	Get(ctx context.Context, key string) (StepCacheEntry, bool, error)
	Put(ctx context.Context, key string, entry StepCacheEntry) error
}

StepCache is a content-addressed store for step outputs. Implementations are pluggable; built-in InMemoryCache (process-local) and FileCache (cross-process via filesystem) cover the common cases.

Cache keys are SHA-256 of canonicalized JSON of (kind, config, depends_on) — see stepCacheKey. The cache is a side-channel: hits short-circuit executor.Execute and replay both Output and Cost, so Workflow.Cost still reflects the cached cost as if the step had run.

type StepCacheEntry 

type StepCacheEntry struct {
	Output   any
	Context  map[string]any
	Cost     *StepCost
	StoredAt time.Time
	TTL      time.Duration // 0 = no expiration
}

StepCacheEntry is one cached step result. Cost is preserved verbatim so downstream budget enforcement and reporting see the same dollars as a fresh execution would.

type StepCost 

type StepCost struct {
	StepID       string   `json:"step_id"`
	Kind         StepKind `json:"kind"`
	Model        string   `json:"model,omitempty"` // empty for non-LLM steps
	InputTokens  int64    `json:"input_tokens,omitempty"`
	OutputTokens int64    `json:"output_tokens,omitempty"`
	USDEstimate  float64  `json:"usd_estimate,omitempty"`
	Bytes        int64    `json:"bytes,omitempty"` // for image steps
	DurationMS   int64    `json:"duration_ms,omitempty"`
}

StepCost is the resource consumption of a single step execution.

type StepDispatcher 

type StepDispatcher interface {
	Dispatch(ctx context.Context, workflowID, stepID string, kind StepKind) error
	DispatchBatch(ctx context.Context, workflowID string, stepIDs []string, kinds []StepKind) error
}

StepDispatcher decides how steps are executed. LocalDispatcher runs them in-process; PostgresDispatcher will enqueue to a step_queue table.

type StepDoneEvent 

type StepDoneEvent struct {
	WorkflowID string
	StepID     string
}

StepDoneEvent is emitted when a step completes via pg_notify.

type StepEnqueuer 

type StepEnqueuer interface {
	Enqueue(item QueueItem) error
	io.Closer
}

StepEnqueuer enqueues steps for distributed execution. Implemented by store.StepQueue.

type StepExecutor 

type StepExecutor interface {
	Execute(ctx context.Context, step *Step, wf *Workflow) error
}

StepExecutor runs a single step within a workflow.

type StepKind 

type StepKind string

StepKind defines the type of work a step performs. 

const (
	StepTool      StepKind = "tool"
	StepLLM       StepKind = "llm"
	StepApproval  StepKind = "approval"
	StepCondition StepKind = "condition"
	StepMessage   StepKind = "message"
	StepWorkflow  StepKind = "workflow"
	StepAgent     StepKind = "agent"
	StepTransform StepKind = "transform"
	StepA2A       StepKind = "a2a"
	StepForEach   StepKind = "foreach"
	StepBranchAll StepKind = "branchall"
	StepSuspend   StepKind = "suspend"
	StepNoop      StepKind = "noop"
	StepImage     StepKind = "image"
	StepVision    StepKind = "vision"
)

func NormalizeStepKind 

func NormalizeStepKind(kind StepKind) StepKind

NormalizeStepKind resolves a step kind alias to the canonical step kind. Returns the input unchanged if it's already canonical or unrecognized.

type StepListener 

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

StepListener listens for PostgreSQL LISTEN/NOTIFY events on the "step_done" channel. Each notification carries a "workflow_id:step_id" payload that the Engine uses to advance the DAG.

func NewStepListener 

func NewStepListener(dsn string) (*StepListener, error)

NewStepListener validates the DSN by connecting once, then returns a listener.

func (*StepListener) Close 

func (l *StepListener) Close() error

Close is a no-op — each Listen call manages its own connection.

func (*StepListener) Listen 

func (l *StepListener) Listen(ctx context.Context) <-chan StepDoneEvent

Listen starts a goroutine that subscribes to the "step_done" channel and sends parsed events to the returned channel. The channel is closed when ctx is cancelled or an unrecoverable error occurs.

type StepReaper 

type StepReaper interface {
	ReapStale(timeout time.Duration) (int, error)
}

StepReaper is satisfied by store.StepQueue.ReapStale.

type StepState 

type StepState string

StepState represents the lifecycle state of a single step. 

const (
	StepPending      StepState = "pending"
	StepRunning      StepState = "running"
	StepCompleted    StepState = "completed"
	StepFailed       StepState = "failed"
	StepSkipped      StepState = "skipped"
	StepDeadLettered StepState = "dead_lettered"
)

type StepTrace 

type StepTrace struct {
	StepID     string `json:"step_id"`
	StepKind   string `json:"step_kind"`
	StartedAt  int64  `json:"started_at"`
	EndedAt    int64  `json:"ended_at"`
	DurationMS int64  `json:"duration_ms"`
	Error      string `json:"error,omitempty"`
	Retries    int    `json:"retries,omitempty"`
}

StepTrace is a single step in the execution trace.

type StepWorkerQueue 

type StepWorkerQueue interface {
	Dequeue(workerID string, kinds []string) (*QueueItem, bool)
	Complete(itemID int64, result []byte, errMsg string) error
	Fail(itemID int64, errMsg string) error
	Heartbeat(itemID int64) error
	io.Closer
}

StepWorkerQueue is the queue interface consumed by WorkerNode. Implemented by store.StepQueue.

type StoreBackend 

type StoreBackend interface {
	Save(w *Workflow) error
	Load(id string) (*Workflow, bool)
	Delete(id string) error
	List(state WorkflowState) []*Workflow
	ListByOwner(owner string) []*Workflow
	FindByIdempotencyKey(key string) *Workflow
	Modify(id string, fn func(w *Workflow)) error
	Close() error
}

StoreBackend is the storage interface that concrete backends must implement. Backends handle persistence; they do NOT clone workflows — the WorkflowStore wrapper handles that.

type StreamCallback 

type StreamCallback func(workflowID, stepID, delta string)

StreamCallback receives streaming LLM chunks during execution.

type SubWorkflowExecutor 

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

SubWorkflowExecutor runs another workflow as a child step. Config: {"workflow_id": "child-id"} The child workflow must already exist in the store. The parent step blocks until the child completes. Child results are available in the parent context under the step ID.

func NewSubWorkflowExecutor 

func NewSubWorkflowExecutor(runner SubWorkflowRunner) *SubWorkflowExecutor

func (*SubWorkflowExecutor) Execute 

func (e *SubWorkflowExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

type SubWorkflowRunner 

type SubWorkflowRunner interface {
	Store() *WorkflowStore
	Start(ctx context.Context, workflowID string) error
	RunToCompletion(ctx context.Context, workflowID string) error
}

SubWorkflowRunner is the interface for running sub-workflows (satisfied by Engine).

type SuspendExecutor 

type SuspendExecutor struct{}

SuspendExecutor pauses the workflow until a specified time. Config: {"suspend_until_ms": <unix_ms>} The watchdog auto-resumes workflows past their deadline.

func NewSuspendExecutor 

func NewSuspendExecutor() *SuspendExecutor

NewSuspendExecutor creates a new SuspendExecutor.

func (*SuspendExecutor) Execute 

func (e *SuspendExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

type Template 

type Template struct {
	Name        string         `json:"name"`
	Description string         `json:"description,omitempty"`
	Params      ParamsMap      `json:"params,omitempty"` // param name → ParamSpec (or legacy description string)
	Steps       []TemplateStep `json:"steps"`
	Defaults    map[string]any `json:"defaults,omitempty"` // default param values
}

Template is a parameterized workflow definition loaded from a JSON file. Variables in the form {{key}} in step configs are replaced at instantiation time.

func ConvertN8nFile 

func ConvertN8nFile(path string) (*Template, error)

ConvertN8nFile reads an n8n JSON file and converts it to a workflow Template.

func ConvertN8nToTemplate 

func ConvertN8nToTemplate(data []byte) (*Template, error)

ConvertN8nToTemplate parses an n8n workflow JSON and converts it to a workflow Template.

func ParseTemplate 

func ParseTemplate(data []byte) (*Template, error)

ParseTemplate parses a native go-workflow template from raw JSON bytes. It rewrites known step-field aliases (e.g. n8n-style "depends" → "depends_on") and then unmarshals strictly — unknown fields cause a clear error instead of being silently dropped (which previously masked author typos).

Load-time validation: every ParamSpec.Default is coercion-checked against its declared Type. A mismatched default (e.g. type="int", default="hello") returns an error immediately so the bug is caught at startup, not at runtime. Multiple default mismatches are collected and returned as a single error.

type TemplateMeta 

type TemplateMeta struct {
	Name        string    `json:"name"`
	Description string    `json:"description,omitempty"`
	Params      ParamsMap `json:"params,omitempty"`
	Source      string    `json:"source"`
}

TemplateMeta is lightweight metadata for listing templates.

type TemplateRuntime 

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

TemplateRuntime loads workflow templates from a single directory.

func NewTemplateRuntime 

func NewTemplateRuntime(dir string) *TemplateRuntime

NewTemplateRuntime creates a template runtime backed by a single directory.

func (*TemplateRuntime) Dir 

func (tr *TemplateRuntime) Dir() string

Dir returns the template directory path.

func (*TemplateRuntime) Instantiate 

func (tr *TemplateRuntime) Instantiate(templateName, workflowID, owner string, params map[string]any) (*Workflow, error)

Instantiate creates a workflow from a template.

func (*TemplateRuntime) List 

func (tr *TemplateRuntime) List() []TemplateMeta

List returns all available templates sorted by name.

func (*TemplateRuntime) Reload 

func (tr *TemplateRuntime) Reload()

Reload reloads templates from disk.

type TemplateStep 

type TemplateStep struct {
	ID        string          `json:"id"`
	Kind      StepKind        `json:"kind"`
	Config    json.RawMessage `json:"config"`
	DependsOn []string        `json:"depends_on,omitempty"`
	Retry     json.RawMessage `json:"retry,omitempty"`
	OnError   string          `json:"on_error,omitempty"`
}

TemplateStep mirrors Step but holds raw JSON config for variable substitution.

type TemplateStore 

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

TemplateStore loads and caches workflow templates from a directory.

func NewTemplateStore 

func NewTemplateStore(dir string) *TemplateStore

NewTemplateStore creates a store that loads templates from the given directory.

func (*TemplateStore) Get 

func (ts *TemplateStore) Get(name string) (*Template, bool)

Get returns a template by name (filename without .json extension).

func (*TemplateStore) Instantiate 

func (ts *TemplateStore) Instantiate(templateName, workflowID, owner string, params map[string]any) (*Workflow, error)

Instantiate creates a Workflow from a template with the given parameters. Parameters replace {{key}} placeholders in step configs.

Before substitution:

  1. Required params are validated — missing required param → error.
  2. Non-string typed params are coerced to their declared type — a string "6000" for an int-typed param becomes the integer 6000, so the subsequent "{{wait_ms}}" substitution emits a bare JSON integer.

func (*TemplateStore) List 

func (ts *TemplateStore) List() []string

List returns all template names.

func (*TemplateStore) Reload 

func (ts *TemplateStore) Reload()

Reload reloads all templates from disk.

type ToolExecutor 

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

ToolExecutor calls a registered tool and stores the result in workflow context.

func NewToolExecutor 

func NewToolExecutor(runner ToolRunner) *ToolExecutor

func (*ToolExecutor) Execute 

func (e *ToolExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

type ToolRunner 

type ToolRunner interface {
	Execute(ctx context.Context, name string, args map[string]any) (string, error)
}

ToolRunner is the interface for executing tools (satisfied by tools.ToolRegistry).

type TransformExecutor 

type TransformExecutor struct{}

TransformExecutor performs lightweight JSON data transformations. Config operations:

  • "set": map of key->value pairs to set (supports {{stepID}} refs)
  • "pick": array of keys to keep from input
  • "omit": array of keys to remove from input
  • "rename": map of old_key->new_key
  • "input": step ID to use as input data (default: previous step)
  • "merge": array of step IDs whose results are merged into one object

This is much cheaper than agent delegation for simple data routing.

func NewTransformExecutor 

func NewTransformExecutor() *TransformExecutor

func (*TransformExecutor) Execute 

func (e *TransformExecutor) Execute(_ context.Context, step *Step, wf *Workflow) error

type TriggerAction 

type TriggerAction struct {
	Kind       string `json:"kind"`
	WorkflowID string `json:"workflow_id,omitempty"`
	TemplateID string `json:"template_id,omitempty"`
	Message    string `json:"message,omitempty"`
	Channel    string `json:"channel,omitempty"`
	To         string `json:"to,omitempty"`
}

TriggerAction defines what happens when a trigger fires.

type TriggerExecutor 

type TriggerExecutor func(trigger *EventTrigger) error

TriggerExecutor is called when a trigger fires.

type TriggerOption 

type TriggerOption func(*TriggerService)

TriggerOption configures a TriggerService.

func WithTriggerLogger 

func WithTriggerLogger(l *slog.Logger) TriggerOption

WithTriggerLogger sets the logger for the trigger service.

func WithTriggerMetrics 

func WithTriggerMetrics(m *Metrics) TriggerOption

WithTriggerMetrics sets the metrics instance for the trigger service.

type TriggerService 

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

TriggerService manages event-driven triggers with JSON persistence.

func NewTriggerService 

func NewTriggerService(storePath string, opts ...TriggerOption) *TriggerService

NewTriggerService creates a trigger service backed by the given file path.

func (*TriggerService) AddTrigger 

func (ts *TriggerService) AddTrigger(name, event string, filter map[string]string, action TriggerAction) (*EventTrigger, error)

AddTrigger adds a new event trigger.

func (*TriggerService) EnableTrigger 

func (ts *TriggerService) EnableTrigger(id string, enabled bool) bool

EnableTrigger enables or disables a trigger.

func (*TriggerService) Evaluate 

func (ts *TriggerService) Evaluate(event string, data map[string]any) []EventTrigger

Evaluate returns all matching triggers for the given event and data.

func (*TriggerService) HookHandler 

func (ts *TriggerService) HookHandler(event string) func(data map[string]any) error

HookHandler returns a function suitable for hooks.Registry.On.

func (*TriggerService) ListTriggers 

func (ts *TriggerService) ListTriggers() []EventTrigger

ListTriggers returns a copy of all triggers.

func (*TriggerService) RegisterHooks 

func (ts *TriggerService) RegisterHooks(hookRegistrar func(event string, fn func(map[string]any) error))

RegisterHooks registers hook handlers for all unique events in the store.

func (*TriggerService) RemoveTrigger 

func (ts *TriggerService) RemoveTrigger(id string) bool

RemoveTrigger removes a trigger by ID.

func (*TriggerService) SetExecutor 

func (ts *TriggerService) SetExecutor(fn TriggerExecutor)

SetExecutor sets the callback for executing trigger actions.

type ValidationError 

type ValidationError struct {
	StepID  string `json:"step_id,omitempty"`
	Field   string `json:"field,omitempty"`
	Message string `json:"message"`
}

ValidationError describes a single issue found during workflow validation.

func (ValidationError) Error 

func (ve ValidationError) Error() string

type VisionCapable 

type VisionCapable interface {
	SupportsVision() bool
}

VisionCapable is implemented by LLMProvider implementations that support multimodal input (text + images). The StepVision executor probes for this interface; providers that do not implement it (or return false) cause the vision step to log a warning and fall back to a text-only call with the prompt alone.

type VisionExecutor 

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

func NewVisionExecutor 

func NewVisionExecutor(p LLMProvider, metrics *Metrics) *VisionExecutor

NewVisionExecutor builds a VisionExecutor wired to the given provider + metrics. metrics may be nil; counter writes are guarded.

func (*VisionExecutor) Execute 

func (e *VisionExecutor) Execute(ctx context.Context, step *Step, wf *Workflow) error

Execute resolves image refs, builds a multimodal LLMMessage, calls the provider, and stores the response in step.Result + wf.Context.

type WebhookAuth 

type WebhookAuth int

WebhookAuth selects the authentication strategy for a webhook trigger. 

const (
	// WebhookAuthNone leaves the endpoint open. Intended for development —
	// the engine logs a warning at registration when this mode is selected.
	WebhookAuthNone WebhookAuth = iota
	// WebhookAuthBearer requires "Authorization: Bearer <secret>" with a
	// constant-time comparison.
	WebhookAuthBearer
	// WebhookAuthHMAC requires SignHeader to contain the hex-encoded
	// HMAC-SHA256 of the raw request body, computed with Secret as the key.
	// Mirrors GitHub's X-Hub-Signature-256 shape.
	WebhookAuthHMAC
)

type WebhookTrigger 

type WebhookTrigger struct {
	Path       string
	Template   string
	VarMapper  func(*http.Request, []byte) (map[string]any, error)
	AuthMode   WebhookAuth
	Secret     string
	SignHeader string // for WebhookAuthHMAC, e.g. "X-Hub-Signature-256"
	Owner      string // owner string assigned to instantiated workflows
}

WebhookTrigger fires a named template instantiation when an HTTP request hits Path. Path must be unique per registered trigger. Symmetric with EventTrigger (hook events) and the Scheduler (cron) — three families that all instantiate templates and start workflows.

type WorkerConfig 

type WorkerConfig struct {
	ID                string          // unique worker identifier
	Queue             StepWorkerQueue // queue to dequeue from
	StepKinds         []string        // which step kinds this worker handles
	Engine            *Engine         // engine with executors for step execution
	HeartbeatInterval time.Duration   // default 30s
	PollInterval      time.Duration   // default 1s
	Logger            *slog.Logger
}

WorkerConfig configures a WorkerNode.

type WorkerNode 

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

WorkerNode dequeues steps from step_queue and executes them via the Engine.

func NewWorkerNode 

func NewWorkerNode(cfg WorkerConfig) (*WorkerNode, error)

NewWorkerNode creates a worker that polls the step_queue and executes steps.

func (*WorkerNode) DrainAndStop 

func (w *WorkerNode) DrainAndStop(timeout time.Duration)

DrainAndStop gracefully shuts down: stops accepting new work, waits for the current step to finish, then stops.

func (*WorkerNode) ProcessOne 

func (w *WorkerNode) ProcessOne(ctx context.Context) bool

ProcessOne dequeues one item, executes it, and completes/fails in the queue. Returns false if the queue was empty.

func (*WorkerNode) Run 

func (w *WorkerNode) Run(ctx context.Context)

Run starts the main dequeue-execute loop and a heartbeat goroutine. Blocks until ctx is cancelled or Stop is called.

func (*WorkerNode) Stop 

func (w *WorkerNode) Stop()

Stop signals the worker to shut down and closes the queue connection.

type Workflow 

type Workflow struct {
	ID              string                 `json:"id"`
	Name            string                 `json:"name"`
	TemplateName    string                 `json:"template_name,omitempty"` // source template name (for concurrency guards)
	Description     string                 `json:"description,omitempty"`
	IdempotencyKey  string                 `json:"idempotency_key,omitempty"` // dedup key — only one active workflow per key
	Steps           []Step                 `json:"steps"`
	State           WorkflowState          `json:"state"`
	CurrentStep     string                 `json:"current_step,omitempty"`
	Context         map[string]any         `json:"context"`
	Owner           string                 `json:"owner"`
	AllowedTools    []string               `json:"allowed_tools,omitempty"` // restrict tool steps to these tools; empty = all allowed
	Security        *SecurityPolicy        `json:"security,omitempty"`      // execution limits and constraints
	Error           string                 `json:"error,omitempty"`
	StepsExecuted   int                    `json:"steps_executed,omitempty"`   // total steps executed (including retries)
	Reducers        map[string]ReducerKind `json:"reducers,omitempty"`         // per-key context merge strategy
	InterruptBefore []string               `json:"interrupt_before,omitempty"` // pause before these step IDs
	InterruptAfter  []string               `json:"interrupt_after,omitempty"`  // pause after these step IDs
	CreatedAt       int64                  `json:"created_at_ms"`
	UpdatedAt       int64                  `json:"updated_at_ms"`
	// Cost is the running aggregate of resource consumption (tokens, USD, image
	// bytes) for cost-bearing steps in this workflow. Nil when no cost-bearing
	// step has executed yet. Updated by recordStepCost after every successful
	// LLM, vision, or image step.
	Cost *WorkflowCost `json:"cost,omitempty"`
}

Workflow is a multi-step execution plan with DAG dependencies and persistence.

func NewWorkflow 

func NewWorkflow(id, name, owner string, steps []Step) *Workflow

NewWorkflow creates a workflow with sensible defaults.

func (*Workflow) AddCost 

func (w *Workflow) AddCost(c StepCost)

AddCost merges a single step's cost into the workflow aggregate, creating the aggregate map on first call. Safe to call from inside step executors after they record their own StepCost.

func (*Workflow) BlockingStep added in v0.17.2 

func (w *Workflow) BlockingStep() *Step

BlockingStep returns the pending approval gate the workflow is currently halted on when State==StateWaitingApproval, else nil. It discriminates three cases for a workflow in the waiting-approval state:

  1. Primary/authoritative: CurrentStep names a pending approval step — that is the gate the workflow is actually blocked on; return it. CurrentStep is written at step-start and preserved across the approval pause, so it is the authoritative "which gate am I blocked on" signal in this case.

  2. Interrupt-pause/no-op: CurrentStep names a NON-approval step that is an active interrupt_before/interrupt_after pause point (the step an interrupt checkpoint paused on — see handleInterrupt / the interrupt_after block in completeStep). There is NO approval gate pending resolution here at all; the pause is just a checkpoint. Return nil so HandleApproval only clears the interrupt list and flips state, touching nothing else. Falling through to the scan here would grab an UNRELATED downstream approval gate that hasn't even been reached yet and silently bypass it (see issue #23 round 4).

  3. Race-fallback/scan: CurrentStep is empty, dangling, or raced to an unrelated non-interrupt non-approval sibling under parallel dispatch (runParallel/DispatchBatch siblings each write CurrentStep inside store.Modify; last-write-wins can leave it pointing at a non-approval sibling after the real gate already paused the workflow). Scan the committed Steps state (race-immune — it reads the step slice, not the racy CurrentStep field) for the first pending approval step. This also covers workflows persisted before CurrentStep was reliable.

func (*Workflow) Clone 

func (w *Workflow) Clone() *Workflow

Clone returns a deep copy of the workflow.

func (*Workflow) GetStep 

func (w *Workflow) GetStep(stepID string) *Step

GetStep returns a pointer to the step with the given ID, or nil.

func (*Workflow) IsTerminal 

func (w *Workflow) IsTerminal() bool

IsTerminal returns true if the workflow is in a final state.

type WorkflowCost 

type WorkflowCost struct {
	InputTokens    int64               `json:"input_tokens"`
	OutputTokens   int64               `json:"output_tokens"`
	USDEstimate    float64             `json:"usd_estimate"`
	ImagesRendered int64               `json:"images_rendered"`
	BytesRendered  int64               `json:"bytes_rendered"`
	BySteps        map[string]StepCost `json:"by_steps,omitempty"` // step ID → cost
	UpdatedAt      time.Time           `json:"updated_at"`
}

WorkflowCost is the running aggregate of an execution's resource consumption. Updated after every cost-bearing step (LLM calls, vision calls, image renders). Returned with the workflow result so consumers can budget, log, and bill.

type WorkflowState 

type WorkflowState string

WorkflowState represents the lifecycle state of a workflow. 

const (
	StatePending         WorkflowState = "pending"
	StateRunning         WorkflowState = "running"
	StateWaitingApproval WorkflowState = "waiting_approval"
	StatePaused          WorkflowState = "paused"
	StateCompleted       WorkflowState = "completed"
	StateFailed          WorkflowState = "failed"
	StateCancelled       WorkflowState = "cancelled"
)

type WorkflowStore 

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

WorkflowStore wraps a StoreBackend with clone-on-entry/exit semantics. All public methods are safe for concurrent use (thread safety is the backend's responsibility).

func NewWorkflowStore 

func NewWorkflowStore(backend StoreBackend) *WorkflowStore

NewWorkflowStore creates a WorkflowStore that delegates to the given backend.

func (*WorkflowStore) Close 

func (s *WorkflowStore) Close() error

Close releases resources held by the backend.

func (*WorkflowStore) Delete 

func (s *WorkflowStore) Delete(id string) error

Delete removes a workflow from the backend.

func (*WorkflowStore) FindByIdempotencyKey 

func (s *WorkflowStore) FindByIdempotencyKey(key string) *Workflow

FindByIdempotencyKey returns a deep copy of the first non-terminal workflow with the given key, or nil.

func (*WorkflowStore) List 

func (s *WorkflowStore) List(state WorkflowState) []*Workflow

List returns deep copies of all workflows, optionally filtered by state.

func (*WorkflowStore) ListByOwner 

func (s *WorkflowStore) ListByOwner(owner string) []*Workflow

ListByOwner returns deep copies of workflows owned by the given session key.

func (*WorkflowStore) Load 

func (s *WorkflowStore) Load(id string) (*Workflow, bool)

Load returns a deep copy of the workflow with the given ID.

func (*WorkflowStore) Modify 

func (s *WorkflowStore) Modify(id string, fn func(w *Workflow)) error

Modify atomically loads a workflow, applies fn, and saves it back. Delegates directly to backend — the backend ensures atomicity.

func (*WorkflowStore) Save 

func (s *WorkflowStore) Save(w *Workflow) error

Save persists a deep copy of the workflow to the backend.

func (*WorkflowStore) String 

func (s *WorkflowStore) String() string

String returns a human-readable description of the store.

Source Files

View all Source files

Directories

Path Synopsis

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.