Skip to content
Documentation
Reference
Go SDK
dag
API

dag

import "github.com/danmestas/dagnats/dag"

WorkflowBuilder provides a fluent DSL for constructing WorkflowDefs. Centralizing construction here lets callers express graph topology naturally without touching StepDef internals — the builder enforces invariants and delegates final structural validation to Validate.

dag/duration_json.go Duration fields decode from either integer nanoseconds (what encoding/json writes for time.Duration, so every stored definition) or a Go duration string such as “30s” (what people write in workflow JSON); see internal/durationjson. A type that embeds one of these inherits its UnmarshalJSON and needs its own to decode extra fields.

Labels let external triggers stamp arbitrary key/value metadata on a WorkflowRun at start time -- an event source starts a run *about* something (an order ID, a tenant, a region) and needs to find or bulk cancel those runs later without standing up a shadow lookup table. The bounds here keep labels cheap to carry on every snapshot and cheap to scan when filtering.

dag/priority.go Priority resolution for workflow run ordering.

Index

Constants

Warning kinds. Stable strings — consumers (registration response, CLI surfaces) may switch on these.

const (
    WarnMissingRespond   = "missing_respond"
    WarnDuplicateRespond = "duplicate_respond"
    WarnMissingSchemas   = "missing_schemas"
)

LabelKeyLengthMax bounds the length of a single label key.

const LabelKeyLengthMax = 64

LabelValueLengthMax bounds the length of a single label value.

const LabelValueLengthMax = 256

LabelsCountMax bounds how many labels a single run may carry.

const LabelsCountMax = 16

RetryAttemptCountMax bounds MaxAttempts on any retry policy. Validate rejects larger values at definition time so the engine’s retry scheduler can assert the bound as a true unreachable invariant rather than a config-reachable one (see internal/engine scheduleRetryBackoff).

const RetryAttemptCountMax = 100_000

RetryMultiplierMax bounds an explicit exponential multiplier. math.Pow with an unbounded multiplier can overflow to +Inf, and converting Inf to time.Duration is implementation-defined; bounding at definition time keeps CalculateDelay’s arithmetic finite without relying on callers to clamp.

const RetryMultiplierMax = 100

func CalculateDelay 

func CalculateDelay(policy RetryPolicy, attempt int) time.Duration

CalculateDelay returns the delay before the next retry attempt. Attempt is 1-based (first retry = attempt 1).

func CronSleepLocation 

func CronSleepLocation(cfg SleepConfig, runInput json.RawMessage) (*time.Location, error)

CronSleepLocation returns the zone a Cron sleep is evaluated in: the static CronTimezone, the zone named at CronTimezoneInputPath in the run input, or UTC when neither is set. The location is never nil. When a named zone cannot be resolved it is UTC and the error says why, so the caller can log the fallback: a recipient’s bad timezone should shift their delivery, not fail their run.

func DefHash 

func DefHash(def WorkflowDef) string

DefHash returns a deterministic content hash for a WorkflowDef: the hex-encoded SHA-256 of its canonical JSON encoding. It exists so a caller that re-registers the same definition on every trigger (#630) can compare hashes and skip the POST /workflows round-trip when the definition is unchanged, instead of re-registering unconditionally.

Determinism relies on two Go/encoding/json guarantees rather than any custom canonicalization: encoding/json sorts map keys before emitting them, and struct fields are always marshaled in declaration order. So two WorkflowDef values that are field-for-field equal -- including maps populated in a different insertion order -- always marshal to byte-identical JSON and therefore hash identically. This does NOT extend to the json.RawMessage fields (WorkflowDef.InputSchema, WorkflowDef.OutputSchema, StepDef.Config): those are hashed verbatim as whatever bytes they hold, so two schemas that are semantically equal but differ in whitespace or key order (e.g. `{“a”:1,“b”:2}` vs `{“b”:2,“a”:1}`) hash differently.

Panics on an empty def.Name (a WorkflowDef without a name is a programmer error, not a runtime condition to handle) and if marshaling fails (WorkflowDef holds only JSON-safe field types, so a marshal failure indicates a violated invariant elsewhere, not bad input).

func DefHashFromVersionKey 

func DefHashFromVersionKey(key, name string) (hash string, ok bool)

DefHashFromVersionKey extracts the hash suffix from a version key produced by DefVersionKey(name, hash). ok is false when key is not EXACTLY a version key belonging to name.

Anchored on both length and content, not a prefix match: a naive strings.HasPrefix(key, name+".v.") check is fooled by a workflow literally named “orders.v” -- its own version keys (“orders.v” + “.v.” + hash) also start with “orders.v.”, so a prefix-only check would mistake them for version keys of a DIFFERENT workflow “orders”, corrupting that name’s retention accounting and enabling cross-workflow eviction (#637 review fix). Requiring the remainder after name’s prefix to be exactly 64 hex characters -- no more, no less -- makes that impossible: “orders.v.v.<hash>” has “v.<hash>” (66 chars) left over under name “orders”, which fails the length/hex check and returns false.

func DefVersionKey 

func DefVersionKey(name, hash string) string

DefVersionKey returns the workflow_defs KV key under which the immutable, content-addressed snapshot of the def named name at content hash hash is stored, alongside the mutable name -> latest pointer at plain key name. hash is expected to be a DefHash output (64 lowercase hex chars); DefVersionKey does not itself validate that shape -- callers pass DefHash(def) directly.

func DefVersionKeyPrefix 

func DefVersionKeyPrefix(name string) string

DefVersionKeyPrefix returns the prefix every DefVersionKey(name, _) output starts with -- name + “.v.”. A prefix match ALONE is not enough to identify name’s own version keys: a workflow literally named e.g. “orders.v” has its own version keys sharing the prefix “orders.v.” that “orders” would compute too. Callers scanning workflow_defs for name’s retained versions should use DefHashFromVersionKey(key, name), which anchors on the full key shape and rejects that collision, rather than testing this prefix directly.

func ExtractDotPath 

func ExtractDotPath(path string, data []byte) (any, error)

ExtractDotPath extracts a value from JSON data using a dot-separated path. The path walks nested objects: “data.user.id” accesses data[“user”][“id”]. Returns the raw value (string, number, bool, map, array, or nil). Panics if path is empty (programmer error); returns error for missing keys.

func IsComplete 

func IsComplete(def WorkflowDef, completed map[string]bool) bool

IsComplete returns true when every step in the definition has been completed or skipped. Auxiliary steps (OnFailure/Compensate targets) that were never triggered don’t block completion — they are expected to remain Pending in the happy path.

func IsDefVersionKey 

func IsDefVersionKey(key string) bool

IsDefVersionKey reports whether key has the shape a DefVersionKey call produces: some prefix followed by “.v.” and 64 lowercase hex characters. RegisterWorkflow rejects workflow names of this shape (see registerWorkflowInner) so a legitimate name -> latest pointer key can never collide with a version key -- any name that would collide is refused at registration time instead.

func LabelsMatch 

func LabelsMatch(want, have map[string]string) bool

LabelsMatch reports whether every key/value in want is present in have with an equal value (AND semantics) -- a nil/empty want matches anything. This is the single deep entry point for label-filter matching: RunsFilter (GET /runs, CountRuns) and BulkCancelRequest (POST /runs/cancel) both narrow by label and must apply the same semantics rather than maintaining two copies of this loop.

want must already be a validated filter (len(want) <= LabelsCountMax) -- callers run it through ValidateLabels before matching, the same way a run’s own Labels are validated before being stored. have is a stored run’s Labels and carries the same guarantee from run-creation time. Both are true invariants by the time a run is filterable, not user input reachable from here, so violations panic rather than silently mismatching.

func MarshalConfig 

func MarshalConfig(cfg interface{}) json.RawMessage

MarshalConfig serializes a config struct into raw JSON for StepDef.Config. Panics on nil or marshal failure — both are programmer errors that should be caught at build time.

func ResolveInput 

func ResolveInput(step StepDef, steps map[string]StepState, runInput ...json.RawMessage) ([]byte, error)

ResolveInput builds the input payload for a step from its upstream outputs. No deps → forward the run-level input so root steps receive caller data. Single dep → pass that step’s output through unchanged. Fan-in → map of dep ID → raw output, so the task can address each upstream.

func ResolvePriority 

func ResolvePriority(cfg *PriorityConfig, input json.RawMessage) int

ResolvePriority computes the priority offset from input data.

func ResolveSleepDuration 

func ResolveSleepDuration(cfg SleepConfig, runInput json.RawMessage, now time.Time) (time.Duration, error)

ResolveSleepDuration turns a sleep config into the concrete delay to wait, given the run input and the dispatch instant. `now` is a parameter so callers and tests share one clock.

The returned duration is never negative. Only a past RFC3339 instant clamps to zero — a negative millisecond count errors instead, because it is far more likely a producer bug than an intentional zero sleep, and clamping would hide it. Every form is bound-checked against maxSleepDuration here, cron included — relying on the cron package’s internal scan horizon would tie this invariant to an unexported constant elsewhere that could be raised without any compiler error or failing test here.

func RunStatusNames 

func RunStatusNames() []string

RunStatusNames returns the canonical lowercase string names for every RunStatus value, in numeric order. Callers (CLI help text, API docs, error messages) reuse this rather than maintaining their own copy of the slice.

func ValidTaskType 

func ValidTaskType(s string) error

ValidTaskType reports whether s is safe to publish verbatim as one or more NATS subject tokens. StepSubject builds “task.{Task}.{runID}” (optionally “task.{Task}.{WorkerGroup}.{runID}”) from a StepDef’s Task with no sanitization — unlike a step ID, which passes through natsutil.SubjectToken first — because a worker’s registered task_types must match the value byte-for-byte to poll it. consumername.FilterFor anchors a worker’s consumer filter on the same value. An unsafe Task (whitespace, a NATS wildcard, a stray leading, trailing, or empty dotted token) would register cleanly today and only break at dispatch time, either minting a malformed subject or silently never getting picked up — see issue #674.

Dots ARE allowed within a token: “dagger.call” is a production task type, and rejecting dots outright would break it. What keeps a dotted task type from leaking into another worker’s poll is FilterFor’s exact token-count anchor, not a charset restriction here — see internal/consumername.FilterFor’s doc comment.

func ValidWorkerGroup 

func ValidWorkerGroup(s string) error

ValidWorkerGroup reports whether s is safe to use as a StepDef.WorkerGroup. StepSubject appends WorkerGroup as its OWN subject token, prefixed with the “=” sentinel (“task.{Task}.={WorkerGroup}.{runID}”, #704), so it must satisfy ValidTaskType’s charset/length/leading-trailing-dot rule AND additionally contain no dots at all — unlike Task, WorkerGroup is never a dotted namespace.

Dots are banned here (not merely anchored around, the way FilterFor isolates a dotted Task) because a dot in WorkerGroup is indistinguishable from the separator FilterFor/StepSubject place between Task and WorkerGroup: FilterFor(“render”, “gpu.fast”) and FilterFor(“render.gpu”, “fast”) both derive “task.render.=gpu.fast.*”, and NameFor collapses both to “workers-render-=gpu-fast” — a dotted group would silently collide with an unrelated dotted-task/group split. Banning dots in WorkerGroup outright closes that regardless of what the step’s Task looks like.

func Validate 

func Validate(def WorkflowDef) error

Validate checks a WorkflowDef for structural correctness before any run is created. Catching these errors at definition time defines them out of existence at runtime — the engine can safely assume every WorkflowDef it receives has already passed Validate.

func ValidateFragment 

func ValidateFragment(fragment []StepDef, cfg PlannerConfig, existingIDs map[string]bool) error

ValidateFragment checks a planner-generated DAG fragment against bounds. All IDs must be unique and not collide with existing steps. Tasks must be non-empty and in AllowedTasks if configured. Dependencies must reference only within-fragment steps.

func ValidateLabels 

func ValidateLabels(labels map[string]string) error

ValidateLabels reports whether labels is a legal set of run labels. A nil or empty map is valid -- labels are optional. On the first violation it returns a descriptive error naming the offending key (or the label count, when the count itself is the violation). Keys are checked in sorted order so the reported “first violation” is deterministic across calls -- map iteration order is not, and two invalid keys in the same input must always name the same one.

func ValidateRunOutput 

func ValidateRunOutput(def WorkflowDef, steps map[string]StepState) (string, error)

ValidateRunOutput checks the run’s final output against the workflow’s OutputSchema. The final output is the output of every terminal step (one no other step depends on) that completed with a non-empty output; each must match. Returns the offending step ID with the violation (path + reason). A def without a schema, or terminal steps that produced no output (skipped, or void), pass.

func ValidateSchema 

func ValidateSchema(schema json.RawMessage, data json.RawMessage) error

ValidateSchema validates data against a JSON Schema subset. Supports: type (string, number, boolean, object, array), required, properties (nested). Returns nil if schema is nil.

type AgentLoopConfig

AgentLoopConfig bounds the iterative behavior of an agent-loop step. Both limits are enforced: whichever fires first terminates the loop. BudgetWarnAt and BudgetForceAt are advisory — the engine does not enforce them. Workers read these to inject budget warnings into LLM conversations (e.g., “you have N iterations left”).

type AgentLoopConfig struct {
    MaxIterations int           `json:"max_iterations"`
    MaxDuration   time.Duration `json:"max_duration,omitempty"`
    LoopDelay     time.Duration `json:"loop_delay,omitempty"`
    BudgetWarnAt  int           `json:"budget_warn_at,omitempty"`
    BudgetForceAt int           `json:"budget_force_at,omitempty"`
}

func ParseAgentLoopConfig 

func ParseAgentLoopConfig(step StepDef) (AgentLoopConfig, error)

ParseAgentLoopConfig extracts AgentLoopConfig from a StepDef’s Config field. Returns an error if the step type is wrong, Config is nil, or the JSON is malformed.

func (*AgentLoopConfig) UnmarshalJSON 

func (cfg *AgentLoopConfig) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type ApprovalConfig

ApprovalConfig holds configuration for approval gate steps. Subject is the NATS subject where a notification is published when the approval is requested. External systems subscribe to this subject and present approve/reject actions to humans.

type ApprovalConfig struct {
    Timeout     time.Duration     `json:"timeout"`
    Subject     string            `json:"subject"`
    Description string            `json:"description,omitempty"`
    Metadata    map[string]string `json:"metadata,omitempty"`
}

func ParseApprovalConfig 

func ParseApprovalConfig(step StepDef) (ApprovalConfig, error)

ParseApprovalConfig extracts ApprovalConfig from a StepDef’s Config field. Returns an error if the step type is wrong, Config is nil, or the JSON is malformed.

func (*ApprovalConfig) UnmarshalJSON 

func (cfg *ApprovalConfig) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type CancelOn

CancelOn specifies an event that cancels a running workflow.

type CancelOn struct {
    Event   string        `json:"event"`
    Match   Match         `json:"match"`
    Timeout time.Duration `json:"timeout,omitempty"`
}

func (*CancelOn) UnmarshalJSON 

func (cancel *CancelOn) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type ConcurrencyLimit

ConcurrencyLimit controls parallel execution at workflow and step level.

type ConcurrencyLimit struct {
    MaxRuns  int `json:"max_runs,omitempty"`
    MaxSteps int `json:"max_steps,omitempty"`
}

type KeyedRateLimit

KeyedRateLimit configures per-key rate limiting using a dot-path expression.

type KeyedRateLimit struct {
    Key    string        `json:"key"`
    Limit  int           `json:"limit"`
    Period time.Duration `json:"period"`
    Units  int           `json:"units"`
}

func (*KeyedRateLimit) UnmarshalJSON 

func (limit *KeyedRateLimit) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type MapConfig

MapConfig controls parallel execution for map steps that fan out. MaxItems caps the array size to prevent unbounded parallelism.

type MapConfig struct {
    MaxItems int `json:"max_items"`
}

func ParseMapConfig 

func ParseMapConfig(step StepDef) (MapConfig, error)

ParseMapConfig extracts MapConfig from a StepDef’s Config field.

type MapInstanceState

MapInstanceState tracks runtime state for one map item execution. Each instance represents one parallel task invocation for a map step.

type MapInstanceState struct {
    Status StepStatus      `json:"status"`
    Output json.RawMessage `json:"output,omitempty"`
    Error  string          `json:"error,omitempty"`
}

type Match

Match is the builder-time type. Both sides are dot-path strings.

type Match struct {
    Left  string  `json:"left"`
    Op    MatchOp `json:"op"`
    Right string  `json:"right"`
}

func (Match) Resolve 

func (m Match) Resolve(stepOutputs map[string][]byte, workflowInput []byte) (ResolvedMatch, error)

Resolve converts a builder-time Match to a runtime ResolvedMatch.

type MatchOp

MatchOp defines comparison operators for event matching.

type MatchOp string

const MatchOpEq MatchOp = "eq"

type ParentCond

ParentCond evaluates a simple comparison on a parent step’s JSON output. Field is a top-level key in the output JSON object. Op is one of the six comparison operators: ==, !=, <, >, <=, >=. Value is the comparison target. Evaluated purely — no I/O, no side effects.

type ParentCond struct {
    StepID string `json:"step_id"`
    Field  string `json:"field"`
    Op     string `json:"op"`
    Value  any    `json:"value"`
}

func SkipIfOutput 

func SkipIfOutput(parent StepRef, field, op string, value any) *ParentCond

SkipIfOutput creates a ParentCond for use with StepRef.SkipIf. The parent must be in the step’s DependsOn list (enforced by Validate).

func (*ParentCond) Evaluate 

func (c *ParentCond) Evaluate(steps map[string]StepState) bool

Evaluate returns true when the condition is satisfied against the given step states. Returns false if the parent step has no output, the field is missing, or the types are not comparable.

type PlannerConfig

PlannerConfig holds configuration for planner steps that generate DAG fragments at runtime. MaxSteps bounds the number of steps the planner can emit; MaxDepth bounds the longest dependency chain. AllowedTasks restricts which task types the fragment may reference.

type PlannerConfig struct {
    MaxSteps     int      `json:"max_steps"`
    MaxDepth     int      `json:"max_depth,omitempty"`
    AllowedTasks []string `json:"allowed_tasks,omitempty"`
}

func ParsePlannerConfig 

func ParsePlannerConfig(step StepDef) (PlannerConfig, error)

ParsePlannerConfig extracts PlannerConfig from a StepDef’s Config field. Returns an error if the step type is wrong, Config is nil, or the JSON is malformed.

type PriorityConfig

PriorityConfig controls run ordering when concurrency limits create backlogs. Key is a dot-path into input data.

type PriorityConfig struct {
    Key           string         `json:"key"`
    Rules         map[string]int `json:"rules"`
    DefaultOffset int            `json:"default_offset"`
}

type RateLimit

RateLimit configures global per-task-type rate limiting.

type RateLimit struct {
    Limit  int           `json:"limit"`
    Period time.Duration `json:"period"`
}

func (*RateLimit) UnmarshalJSON 

func (limit *RateLimit) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type ResolvedMatch

ResolvedMatch is the runtime type stored in KV waiter entries. Right is resolved to a concrete value when the waiter is created.

type ResolvedMatch struct {
    Left  string  `json:"left"`
    Op    MatchOp `json:"op"`
    Right any     `json:"right"`
}

func (ResolvedMatch) Evaluate 

func (m ResolvedMatch) Evaluate(eventData []byte) (bool, error)

Evaluate checks if the match condition holds against event JSON data.

type RespondConfig

RespondConfig is the configuration for a StepTypeRespond step per ADR-013. A respond step publishes the workflow’s outward-facing HTTP response on dagnats.http.response.<run_id>. The step is a side effect, not a return: subsequent steps in the DAG continue to run after the response has been dispatched.

Status defaults to 200 and ContentType defaults to “application/json” via Defaulted() — the engine applies defaults at execution time so authors can omit the boilerplate. BodyFrom is a dotpath into prior step output; empty means “use the immediate upstream step’s output”.

type RespondConfig struct {
    Status      int               `json:"status,omitempty"`
    Headers     map[string]string `json:"headers,omitempty"`
    BodyFrom    string            `json:"body_from,omitempty"`
    ContentType string            `json:"content_type,omitempty"`
}

func ParseRespondConfig 

func ParseRespondConfig(step StepDef) (RespondConfig, error)

ParseRespondConfig extracts RespondConfig from a StepDef’s Config field. Mirrors the other Parse*Config helpers in this package. Programmer-error invariants (empty step ID, mismatched type) trigger a panic — the workflow validator catches these one frame up; reaching this function with a malformed step means a registration path skipped Validate(), which is itself a bug to fix.

func (RespondConfig) Defaulted 

func (c RespondConfig) Defaulted() RespondConfig

Defaulted returns a copy of c with zero-valued fields filled from the documented defaults. Callers should use this exactly once at execute time; storing the defaulted copy back into the workflow definition would obscure the author’s intent and lose the “default” signal on subsequent re-reads.

type RetryPolicy

RetryPolicy configures retry behavior for a step or as a workflow default. MaxAttempts=0 means no retries.

type RetryPolicy struct {
    MaxAttempts  int           `json:"max_attempts"`
    Strategy     RetryStrategy `json:"strategy"`
    InitialDelay time.Duration `json:"initial_delay"`
    MaxDelay     time.Duration `json:"max_delay"`
    Multiplier   float64       `json:"multiplier,omitempty"`
}

func ResolveRetryPolicy 

func ResolveRetryPolicy(wfDef WorkflowDef, stepDef StepDef) *RetryPolicy

ResolveRetryPolicy returns the effective retry policy for a step. Resolution order: step Retry → workflow DefaultRetry → legacy Retries field → nil (no retries).

func (*RetryPolicy) UnmarshalJSON 

func (policy *RetryPolicy) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type RetryStrategy

RetryStrategy selects the backoff algorithm for step retries.

type RetryStrategy int

const (
    RetryFixed       RetryStrategy = iota // Same delay every attempt
    RetryLinear                           // delay * attempt
    RetryExponential                      // delay * multiplier^(attempt-1)
)

func (RetryStrategy) MarshalJSON 

func (s RetryStrategy) MarshalJSON() ([]byte, error)

func (RetryStrategy) String 

func (s RetryStrategy) String() string

func (*RetryStrategy) UnmarshalJSON 

func (s *RetryStrategy) UnmarshalJSON(data []byte) error

type RunStatus

RunStatus tracks the lifecycle of a workflow run. The zero value (pending) is a safe default — a newly created run has not yet been claimed by the engine.

type RunStatus int

const (
    RunStatusPending RunStatus = iota
    RunStatusRunning
    RunStatusCompleted
    RunStatusFailed
    RunStatusCancelled
    RunStatusCompensated
    RunStatusCompensateFailed
)

func ParseRunStatus 

func ParseRunStatus(s string) (RunStatus, error)

ParseRunStatus parses an operator-supplied string into a RunStatus. Case-insensitive, trims surrounding whitespace. On no match it returns a descriptive error listing the valid set, so callers never need to maintain their own copy of the status names.

Single deep entry point: every CLI flag, API parameter, or other surface that accepts a run-status string MUST funnel through here.

func (RunStatus) IsTerminal 

func (r RunStatus) IsTerminal() bool

IsTerminal returns true for statuses that represent a finished run.

func (RunStatus) MarshalJSON 

func (r RunStatus) MarshalJSON() ([]byte, error)

func (RunStatus) String 

func (r RunStatus) String() string

func (*RunStatus) UnmarshalJSON 

func (r *RunStatus) UnmarshalJSON(data []byte) error

type SingletonConfig

SingletonConfig constrains runs to one-at-a-time per key.

type SingletonConfig struct {
    Mode SingletonMode `json:"mode"`
    Key  string        `json:"key,omitempty"`
}

type SingletonMode

SingletonMode determines behavior on duplicate detection. String type for safe JSON serialization in KV storage.

type SingletonMode string

const (
    SingletonModeSkip   SingletonMode = "skip"
    SingletonModeCancel SingletonMode = "cancel"
)

type SleepConfig

SleepConfig holds configuration for sleep steps. Exactly one of the three forms must be set; ParseSleepConfig enforces that so no downstream caller can observe an ambiguous config.

Duration fixes the delay at def-registration time. Cron and UntilInputPath defer it to dispatch: Cron sleeps until the next occurrence strictly after dispatch, UntilInputPath reads a deadline from the run input (an RFC3339 instant or integer milliseconds).

CronTimezone and CronTimezoneInputPath name the zone a Cron sleep is evaluated in, statically or from the run input. They modify the Cron form rather than being forms of their own, so at most one may be set and only alongside Cron. With neither, the zone is UTC: a host’s local zone would make the same def fire at different instants on different machines.

type SleepConfig struct {
    Duration              time.Duration `json:"duration,omitempty"`
    Cron                  string        `json:"cron,omitempty"`
    UntilInputPath        string        `json:"until_input_path,omitempty"`
    CronTimezone          string        `json:"cron_timezone,omitempty"`
    CronTimezoneInputPath string        `json:"cron_timezone_input_path,omitempty"`
}

func ParseSleepConfig 

func ParseSleepConfig(step StepDef) (SleepConfig, error)

ParseSleepConfig extracts SleepConfig from a StepDef’s Config field.

func (*SleepConfig) UnmarshalJSON 

func (cfg *SleepConfig) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type SnapshotRevision

SnapshotRevision records the stored revision a WorkflowRun was read at, so the next write of that run can be a compare-and-set against it. Several engines (one per cluster node) advance the same run; a write that does not prove it saw the latest snapshot would erase whatever another engine wrote in between.

One SnapshotRevision belongs to one load: copies of the run share it by pointer, while a separate load gets its own. That is what lets one handler save the same run twice while two handlers that loaded the same revision still conflict.

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

func NewSnapshotRevision 

func NewSnapshotRevision(seq uint64) *SnapshotRevision

NewSnapshotRevision returns a handle at seq. seq 0 means the run’s key does not exist yet.

func (*SnapshotRevision) Advance 

func (r *SnapshotRevision) Advance(seq uint64)

Advance records a successful write at seq. Revisions only grow, so a seq at or below the current one is a programmer error.

func (*SnapshotRevision) Seq 

func (r *SnapshotRevision) Seq() uint64

Seq returns the revision the next write must expect.

type StepDef

StepDef is the immutable declaration of a single step within a WorkflowDef. DependsOn lists step IDs that must complete before this step is queued. Config holds type-specific configuration as raw JSON — use ParseXxxConfig helpers to extract typed structs.

type StepDef struct {
    ID                 string            `json:"id"`
    Task               string            `json:"task"`
    DependsOn          []string          `json:"depends_on,omitempty"`
    Retries            int               `json:"retries,omitempty"`
    Timeout            time.Duration     `json:"timeout"`
    Type               StepType          `json:"type"`
    Config             json.RawMessage   `json:"config,omitempty"`
    SkipIf             *ParentCond       `json:"skip_if,omitempty"`
    Metadata           map[string]string `json:"metadata,omitempty"`
    Retry              *RetryPolicy      `json:"retry,omitempty"`
    WorkerGroup        string            `json:"worker_group,omitempty"`
    OnFailure          string            `json:"on_failure,omitempty"`
    Compensate         string            `json:"compensate,omitempty"`
    RateLimit          *RateLimit        `json:"rate_limit,omitempty"`
    KeyedRateLimit     *KeyedRateLimit   `json:"keyed_rate_limit,omitempty"`
    MaxTaskConcurrency int               `json:"max_task_concurrency,omitempty"`
    Singleton          bool              `json:"singleton,omitempty"`
    // RequiredCapabilities names runtime capabilities the step's handler
    // requires (e.g. "control-plane"). The orchestrator copies these into
    // the TaskPayload at enqueue time; the worker uses them to decide
    // whether to grant a gated handle. Zero value (nil) = no capabilities
    // declared, which is today's behavior. Additive — omitempty keeps old
    // defs byte-identical on the wire.
    RequiredCapabilities []string `json:"required_capabilities,omitempty"`
}

func EffectiveSteps 

func EffectiveSteps(def WorkflowDef, run WorkflowRun) []StepDef

EffectiveSteps returns the combined static + dynamic steps for a running workflow. When no dynamic steps exist, the original slice is returned unchanged to avoid allocation.

func NamespaceFragment 

func NamespaceFragment(plannerID string, fragment []StepDef) []StepDef

NamespaceFragment prefixes all step IDs and DependsOn references with the planner step ID to prevent collisions. Also forces all steps to StepTypeNormal — planners cannot spawn nested planners.

func ResolveCompensateChain 

func ResolveCompensateChain(def WorkflowDef, completed map[string]bool, failedStepID string) []StepDef

ResolveCompensateChain returns compensate steps for completed steps in reverse topological order. Each step (except the first) gets a DependsOn pointing to the previous — this lets the engine enforce sequential execution using existing resolution logic.

func ResolveReady 

func ResolveReady(def WorkflowDef, completed map[string]bool, queued map[string]bool) []StepDef

ResolveReady returns steps whose dependencies are fully satisfied (completed or skipped) and that have not yet been queued or completed. Both completed and queued are checked to avoid double-dispatching a step already in flight.

func ResolveSkipped 

func ResolveSkipped(def WorkflowDef, completed map[string]bool, queued map[string]bool, steps map[string]StepState) []StepDef

ResolveSkipped returns steps whose dependencies are satisfied AND whose SkipIf condition evaluates to true. These should be marked Skipped by the orchestrator instead of being enqueued. Steps without SkipIf are never returned here.

func (*StepDef) UnmarshalJSON 

func (step *StepDef) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type StepRef

StepRef is a compile-time-safe handle to a step within a WorkflowBuilder. Returned by Task() and AgentLoop(), it replaces string-based DependsOn with typed references that cannot silently miswire dependencies. The zero value is unusable — only the builder constructs valid StepRefs.

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

func (StepRef) After 

func (r StepRef) After(refs ...StepRef) StepRef

After declares that this step depends on the given steps. Compile-time safe: passing a StepRef from a different builder panics immediately rather than producing a corrupt WorkflowDef discovered at validation time.

func (StepRef) Compensate 

func (r StepRef) Compensate(target StepRef) StepRef

Compensate designates a step to reverse this step’s side effects during saga compensation. Runs in reverse topo order when a downstream step fails permanently. Panics on zero-value StepRef, cross-builder refs, or self-reference.

func (StepRef) ID 

func (r StepRef) ID() string

ID returns the step’s string identifier. Useful for bridge code that still needs the raw ID (e.g. serialization boundaries).

func (StepRef) OnFailure 

func (r StepRef) OnFailure(target StepRef) StepRef

OnFailure designates a step to run when this step fails permanently (retries exhausted). The target step receives error context as input. Panics on zero-value StepRef, cross-builder refs, or self-reference.

func (StepRef) SkipIf 

func (r StepRef) SkipIf(cond *ParentCond) StepRef

SkipIf sets a condition that, when true, causes this step to be skipped instead of executed. The condition’s StepID must be in DependsOn (enforced by Validate). Skipped steps are treated as “satisfied” for downstream deps.

func (StepRef) WithDetach 

func (r StepRef) WithDetach() StepRef

WithDetach marks a SubWorkflow step as detached — the parent step completes immediately after spawning the child, without waiting for the child to finish. Panics if called on a non-SubWorkflow step.

func (StepRef) WithKeyedRateLimit 

func (r StepRef) WithKeyedRateLimit(krl KeyedRateLimit) StepRef

WithKeyedRateLimit sets per-key rate limiting on this step.

func (StepRef) WithLoopDelay 

func (r StepRef) WithLoopDelay(d time.Duration) StepRef

WithLoopDelay configures the delay between agent loop iterations. The orchestrator waits this duration before re-enqueuing the step. Useful for rate-limited APIs where you need spacing between calls.

func (StepRef) WithMaxDuration 

func (r StepRef) WithMaxDuration(d time.Duration) StepRef

WithMaxDuration configures the wall-clock bound on an AgentLoop step.

func (StepRef) WithMaxItems 

func (r StepRef) WithMaxItems(n int) StepRef

WithMaxItems configures the maximum number of items to process for a Map step. Calling this on a non-Map step or with n <= 0 panics — these are programmer errors that should be caught immediately.

func (StepRef) WithMaxIterations 

func (r StepRef) WithMaxIterations(n int) StepRef

WithMaxIterations configures the iteration bound on an AgentLoop step. Panics if the step is not an AgentLoop — calling this on a Task step indicates a logic error in the caller.

func (StepRef) WithRateLimit 

func (r StepRef) WithRateLimit(rl RateLimit) StepRef

WithRateLimit sets global per-task-type rate limiting on this step.

func (StepRef) WithRetries 

func (r StepRef) WithRetries(n int) StepRef

WithRetries sets the maximum number of retry attempts for this step. Zero means no retries — the step fails permanently on first error.

func (StepRef) WithTaskConcurrency 

func (r StepRef) WithTaskConcurrency(max int) StepRef

WithTaskConcurrency sets the global per-task-type concurrency limit on this step. At most max tasks of this type will execute concurrently across all workflow runs.

func (StepRef) WithTimeout 

func (r StepRef) WithTimeout(d time.Duration) StepRef

WithTimeout sets the per-attempt timeout on this step.

type StepState

StepState captures mutable runtime state for one step in a run. Output is kept as raw bytes to remain payload-agnostic. Iterations tracks how many agent-loop Continue cycles have completed; used to generate unique dedup IDs for each re-enqueue. LoopStartedAt records when the first iteration began, for MaxDuration enforcement. MapInstances tracks state for each parallel map item when Type == StepTypeMap. WakeAt records when a sleep step should complete, for engine scheduling. ChildRunID links to the spawned child run for SubWorkflow steps.

type StepState struct {
    Status        StepStatus         `json:"status"`
    Attempts      int                `json:"attempts"`
    Iterations    int                `json:"iterations,omitempty"`
    LoopStartedAt time.Time          `json:"loop_started_at,omitempty"`
    Output        []byte             `json:"output,omitempty"`
    Error         string             `json:"error,omitempty"`
    MapInstances  []MapInstanceState `json:"map_instances,omitempty"`
    WakeAt        *time.Time         `json:"wake_at,omitempty"`
    ChildRunID    string             `json:"child_run_id,omitempty"`
    // DispatchNonce is a fresh random token stamped each time this step is
    // dispatched to a worker (#380, ADR-021 Phase A). It rides this snapshot
    // write — no extra KV write. The same value is placed on the
    // TaskPayload; the worker echoes it back on control-plane requests. The
    // server verifies the echoed nonce equals this value, proving the caller
    // is the worker that received this exact dispatch — a sibling-run worker
    // cannot forge another run's nonce. A retry re-stamps a fresh nonce.
    // Additive: legacy snapshots deserialize to "".
    DispatchNonce string `json:"dispatch_nonce,omitempty"`
    // StartedAt is the engine's dispatch decision time for this step (#626)
    // — when the orchestrator marked it Queued and stamped DispatchNonce,
    // not when the worker later picked it up. A retry re-stamps StartedAt
    // alongside the fresh nonce, so it reflects the current attempt only.
    // Worker-reported DurationMs remains the execution-only figure; this
    // field is for waterfall/UI rendering of end-to-end step timing.
    // Additive: legacy snapshots deserialize to nil.
    StartedAt *time.Time `json:"started_at,omitempty"`
    // CompletedAt is when the step reached a terminal status (Completed or
    // Failed), stamped on the same snapshot write that persists the
    // terminal status — no extra KV write. Additive: legacy snapshots
    // deserialize to nil.
    CompletedAt *time.Time `json:"completed_at,omitempty"`
    // AttemptStartedAt is when the current attempt first reported
    // step.started; the step's timeout runs from it (#743). A NATS
    // redelivery of the same attempt (a worker died or stopped extending
    // its claim, or a pause resumed) keeps it, so redeliveries cannot
    // restart the clock. Cleared when the engine dispatches the step
    // afresh or schedules a retry, so a new attempt gets the full
    // timeout. Additive: legacy snapshots deserialize to nil, which
    // starts the clock at the next step.started.
    AttemptStartedAt *time.Time `json:"attempt_started_at,omitempty"`
    // RetryDispatchAt is when a scheduled retry's task is published. A
    // step.started before it is a leftover redelivery of the failed
    // attempt and does not start the retry's clock. Cleared once the
    // retry's clock starts. Additive, like AttemptStartedAt.
    RetryDispatchAt *time.Time `json:"retry_dispatch_at,omitempty"`
    // DispatchUnsent is set by the save that dispatches the step and
    // cleared once its task, timer or waiter has been sent. A step
    // still marked after that save is one whose send may never have
    // happened (the engine died in between), so the engine sends it
    // again. Additive: legacy snapshots deserialize to false.
    DispatchUnsent bool `json:"dispatch_unsent,omitempty"`
    // LoopState is the state an agent-loop step committed with its last
    // accepted step.continue (#801), written in the same save that
    // advanced Iterations. Every later dispatch of the step (the next
    // iteration, a retry, a resend after an engine crash) carries it in
    // the task payload, so a retry resumes from the last committed
    // iteration. Opaque to the engine; bounded by the engine's loop
    // state cap. Additive: legacy snapshots deserialize to nil.
    LoopState []byte `json:"loop_state,omitempty"`
    // WaitUntil and WaitInput are the durable wait the step's last
    // step.continue asked for (ADR-027): the next iteration is sent at
    // WaitUntil, or as soon as an input is queued when WaitInput is set,
    // whichever comes first. Both are cleared by the save that sends
    // the iteration; a timer or input event that finds them cleared
    // is stale. Additive: legacy snapshots deserialize to none.
    WaitUntil *time.Time `json:"wait_until,omitempty"`
    WaitInput bool       `json:"wait_input,omitempty"`
    // HasInputs records that inputs were submitted to the step while it
    // was unfinished, so the save that ends the step settles whatever
    // is still queued as unanswered (ADR-027) and clears it. A step
    // nobody wrote to costs no lookup on completion.
    HasInputs bool `json:"has_inputs,omitempty"`
}

type StepStatus

StepStatus tracks the lifecycle of a single step within a run. Queued means the step has been dispatched to NATS but not yet claimed by a worker.

type StepStatus int

const (
    StepStatusPending StepStatus = iota
    StepStatusQueued
    StepStatusRunning
    StepStatusCompleted
    StepStatusFailed
    StepStatusSkipped
    StepStatusCancelled
    StepStatusRecovered
)

func (StepStatus) MarshalJSON 

func (s StepStatus) MarshalJSON() ([]byte, error)

func (StepStatus) String 

func (s StepStatus) String() string

func (*StepStatus) UnmarshalJSON 

func (s *StepStatus) UnmarshalJSON(data []byte) error

type StepType

StepType distinguishes execution semantics — normal tasks run once, agent loops iterate until a termination signal, and sub-workflows delegate to a nested DAG.

type StepType int

const (
    StepTypeNormal StepType = iota
    StepTypeAgentLoop
    StepTypeSubWorkflow
    StepTypeAgent
    StepTypeMap
    StepTypeSleep
    StepTypeWaitForEvent
    StepTypeApproval
    StepTypePlanner
    StepTypeRespond
)

func (StepType) MarshalJSON 

func (s StepType) MarshalJSON() ([]byte, error)

func (StepType) String 

func (s StepType) String() string

func (*StepType) UnmarshalJSON 

func (s *StepType) UnmarshalJSON(data []byte) error

type StickyStrategy

StickyStrategy controls worker affinity for workflow runs.

type StickyStrategy string

const (
    StickyNone StickyStrategy = ""
    StickySoft StickyStrategy = "soft"
    StickyHard StickyStrategy = "hard"
)

type SubWorkflowConfig

SubWorkflowConfig holds configuration for sub-workflow steps. Workflow names the child workflow definition to spawn. Detach controls whether the parent waits for the child to complete: when true, the parent step completes immediately after spawn.

type SubWorkflowConfig struct {
    Workflow string `json:"workflow"`
    Detach   bool   `json:"detach,omitempty"`
}

func ParseSubWorkflowConfig 

func ParseSubWorkflowConfig(step StepDef) (SubWorkflowConfig, error)

ParseSubWorkflowConfig extracts SubWorkflowConfig from a StepDef’s Config field.

type WaitForEventOpts

WaitForEventOpts configures a wait-for-event step.

type WaitForEventOpts struct {
    Event   string        `json:"event"`
    Match   Match         `json:"match"`
    Timeout time.Duration `json:"timeout"`
}

func ParseWaitForEventConfig 

func ParseWaitForEventConfig(step StepDef) (WaitForEventOpts, error)

ParseWaitForEventConfig extracts WaitForEventOpts from a StepDef’s Config field.

func (*WaitForEventOpts) UnmarshalJSON 

func (opts *WaitForEventOpts) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type Warning

Warning is the structured result of a graph-level validation check per ADR-013 Layer 1. Warnings are surfaced through the workflow registration response — they do NOT fail the registration, because legitimate branch-per-outcome patterns can produce false positives and a hard rejection would block real use cases. Fatal field rules live in the per-trigger Validate() methods.

type Warning struct {
    Kind    string `json:"kind"`
    Message string `json:"message"`
}

func ValidateRespondReachability 

func ValidateRespondReachability(def WorkflowDef, hasHTTPTrigger bool) []Warning

ValidateRespondReachability returns warnings (not errors) for the two ADR-013 graph problems unique to HTTP-triggered workflows:

  • missing_respond — workflow declares an HTTP trigger but no reachable terminal path contains a respond step. At runtime the caller hangs until the 504 timeout fires.
  • duplicate_respond — two respond steps are simultaneously reachable on the same execution. The second publish has no subscriber (subject already unsubscribed) and silently drops.

hasHTTPTrigger gates the missing_respond check — a workflow without an HTTP trigger has no caller to leave hanging. duplicate_respond is always emitted, since two responds on the same run is wrong regardless of trigger kind (the second publish always drops). missing_schemas is emitted only for HTTP-triggered workflows that lack input_schema and/or output_schema, since the OpenAPI generator falls back to free-form objects in that case and clients lose type safety.

Returns a nil slice when there are no problems. Callers should distinguish nil from non-empty rather than relying on len().

type WorkflowBuilder

WorkflowBuilder accumulates step definitions and wires them into a WorkflowDef on Build(). current tracks the most recently added step so that chained modifier calls (DependsOn, WithTimeout, etc.) always target the right step.

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

func NewWorkflow 

func NewWorkflow(name string) *WorkflowBuilder

NewWorkflow starts a new builder for a workflow with the given name. Version defaults to “1” — override via Version() if needed.

func (*WorkflowBuilder) Agent 

func (b *WorkflowBuilder) Agent(id, task string, metadata map[string]string) StepRef

Agent appends a Claude Agent SDK step. Metadata carries role and other agent-specific config — the core DAG package is ignorant of what it means.

func (*WorkflowBuilder) AgentLoop 

func (b *WorkflowBuilder) AgentLoop(id, task string) StepRef

AgentLoop appends an agent-loop step with an initialised (but unconfigured) AgentLoopConfig. Callers must configure bounds via WithMaxIterations / WithMaxDuration before Build() — Validate enforces MaxIterations > 0.

func (*WorkflowBuilder) Approval 

func (b *WorkflowBuilder) Approval(id string, cfg ApprovalConfig) StepRef

Approval adds a human approval gate step to the workflow. No worker is involved — the engine manages the token and timeout.

func (*WorkflowBuilder) Build 

func (b *WorkflowBuilder) Build() (WorkflowDef, error)

func (*WorkflowBuilder) CancelOn 

func (b *WorkflowBuilder) CancelOn(event string, match Match) *WorkflowBuilder

CancelOn registers an event that cancels the workflow.

func (*WorkflowBuilder) CancelOnWithTimeout 

func (b *WorkflowBuilder) CancelOnWithTimeout(event string, match Match, timeout time.Duration) *WorkflowBuilder

CancelOnWithTimeout registers a cancellation event with timeout.

func (*WorkflowBuilder) DependsOn 

func (b *WorkflowBuilder) DependsOn(ids ...string) *WorkflowBuilder

DependsOn declares that the active step must not start until all listed step IDs have completed. Kept for backward compatibility — prefer After(StepRef) for new code which provides compile-time safety.

func (*WorkflowBuilder) Map 

func (b *WorkflowBuilder) Map(id, taskType string) StepRef

Map appends a map step that fans out over an array from its dependency. The step will execute taskType once per item in the input array, up to MapConfig.MaxItems. Returns a StepRef for chaining dependency wiring and calling WithMaxItems to override the default bound of 1000.

func (*WorkflowBuilder) Name 

func (b *WorkflowBuilder) Name() string

Name returns the workflow name. Used by higher-level packages that need to derive task names from the workflow identity.

func (*WorkflowBuilder) Planner 

func (b *WorkflowBuilder) Planner(id, task string, cfg PlannerConfig) StepRef

Planner appends a planner step that generates a DAG fragment at runtime. The worker outputs JSON steps; the engine validates, namespaces, and materializes them into the running workflow.

func (*WorkflowBuilder) Sleep 

func (b *WorkflowBuilder) Sleep(id string, duration time.Duration) StepRef

Sleep adds a durable delay step to the workflow. No worker is involved — the engine handles the timer.

func (*WorkflowBuilder) SubWorkflow 

func (b *WorkflowBuilder) SubWorkflow(id, workflow string) StepRef

SubWorkflow appends a sub-workflow step that spawns a child workflow execution. The child workflow must be registered in the workflow_defs KV bucket. By default the parent step blocks until the child completes; use WithDetach() on the returned StepRef to fire-and-forget.

func (*WorkflowBuilder) Task 

func (b *WorkflowBuilder) Task(id, task string) StepRef

Task appends a normal (non-looping) step and returns a StepRef for compile-time-safe dependency wiring and modifier chaining.

func (*WorkflowBuilder) Version 

func (b *WorkflowBuilder) Version(v string) *WorkflowBuilder

Version overrides the default workflow version string.

func (*WorkflowBuilder) WaitForEvent 

func (b *WorkflowBuilder) WaitForEvent(id string, opts WaitForEventOpts) StepRef

WaitForEvent adds a step that waits for an external event to match a condition. No worker is involved — the engine handles event matching.

func (*WorkflowBuilder) WithConcurrency 

func (b *WorkflowBuilder) WithConcurrency(maxRuns, maxSteps int) *WorkflowBuilder

WithConcurrency sets workflow-level concurrency limits. MaxRuns bounds how many runs of this workflow execute in parallel; MaxSteps bounds how many steps execute concurrently within a single run.

func (*WorkflowBuilder) WithIdempotencyKey 

func (b *WorkflowBuilder) WithIdempotencyKey(dotPath string) *WorkflowBuilder

WithIdempotencyKey configures a dot-path expression evaluated against workflow input to produce a dedup key. Duplicate runs with the same key value return the existing run ID instead of creating a new one.

func (*WorkflowBuilder) WithMaxDuration 

func (b *WorkflowBuilder) WithMaxDuration(d time.Duration) *WorkflowBuilder

WithMaxDuration configures the wall-clock bound on the active AgentLoop step. Kept for backward compatibility — prefer StepRef.WithMaxDuration.

func (*WorkflowBuilder) WithMaxIterations 

func (b *WorkflowBuilder) WithMaxIterations(n int) *WorkflowBuilder

WithMaxIterations configures the iteration bound on the active AgentLoop step. Kept for backward compatibility — prefer StepRef.WithMaxIterations.

func (*WorkflowBuilder) WithPriority 

func (b *WorkflowBuilder) WithPriority(cfg PriorityConfig) *WorkflowBuilder

WithPriority configures run priority ordering.

func (*WorkflowBuilder) WithSingleton 

func (b *WorkflowBuilder) WithSingleton(mode SingletonMode) *WorkflowBuilder

WithSingleton configures global singleton constraint.

func (*WorkflowBuilder) WithSingletonKey 

func (b *WorkflowBuilder) WithSingletonKey(mode SingletonMode, key string) *WorkflowBuilder

WithSingletonKey configures per-entity singleton.

func (*WorkflowBuilder) WithSticky 

func (b *WorkflowBuilder) WithSticky(s StickyStrategy) *WorkflowBuilder

Build assembles the WorkflowDef and delegates to Validate. Any structural error (cycle, missing dep, etc.) is surfaced here so callers get a clean error value rather than a panic at execution time. WithSticky configures worker affinity for workflow runs. Soft prefers the same worker; Hard requires it.

func (*WorkflowBuilder) WithTimeout 

func (b *WorkflowBuilder) WithTimeout(d time.Duration) *WorkflowBuilder

WithTimeout sets the per-attempt timeout on the active step. Kept for backward compatibility — prefer StepRef.WithTimeout for new code.

type WorkflowDef

WorkflowDef is the immutable schema for a workflow. Stored once, referenced by many runs. Version allows schema evolution without breaking existing runs.

type WorkflowDef struct {
    Name           string            `json:"name"`
    Version        string            `json:"version"`
    Steps          []StepDef         `json:"steps"`
    DefaultRetry   *RetryPolicy      `json:"default_retry,omitempty"`
    Concurrency    *ConcurrencyLimit `json:"concurrency,omitempty"`
    Timeout        time.Duration     `json:"timeout,omitempty"`
    InputSchema    json.RawMessage   `json:"input_schema,omitempty"`
    OutputSchema   json.RawMessage   `json:"output_schema,omitempty"`
    AuxSteps       map[string]bool   `json:"aux_steps,omitempty"`
    IdempotencyKey string            `json:"idempotency_key,omitempty"`
    Sticky         StickyStrategy    `json:"sticky,omitempty"`
    Priority       *PriorityConfig   `json:"priority,omitempty"`
    CancelOn       []CancelOn        `json:"cancel_on,omitempty"`
    Singleton      *SingletonConfig  `json:"singleton,omitempty"`
}

func EffectiveDef 

func EffectiveDef(def WorkflowDef, run WorkflowRun) WorkflowDef

EffectiveDef returns a WorkflowDef augmented with dynamic steps from the run. The original def is not mutated — a shallow copy is returned with the combined step list and rebuilt AuxSteps.

func WithSchemas 

func WithSchemas[I, O any](def WorkflowDef) WorkflowDef

WithSchemas generates JSON schemas from Go types I (input) and O (output) and attaches them to the WorkflowDef. Applied after Build(). Supports flat structs with primitive fields, slices, and maps.

func (*WorkflowDef) UnmarshalJSON 

func (def *WorkflowDef) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts duration fields as Go duration strings (“30s”) or integer nanoseconds.

type WorkflowRun

WorkflowRun holds live state for a single execution of a WorkflowDef. Steps maps step ID to its current StepState; initialized to pending for all steps. Input preserves the original user-supplied payload so retries can reuse it.

type WorkflowRun struct {
    RunID        string               `json:"run_id"`
    WorkflowID   string               `json:"workflow_id"`
    Status       RunStatus            `json:"status"`
    Steps        map[string]StepState `json:"steps"`
    Input        json.RawMessage      `json:"input,omitempty"`
    CreatedAt    time.Time            `json:"created_at"`
    DynamicSteps []StepDef            `json:"dynamic_steps,omitempty"`
    ParentRunID  string               `json:"parent_run_id,omitempty"`
    // RootRunID is the run ID of the tree-root for this run's lineage —
    // the same value for every run in a spawn tree (#377). Server-derived,
    // never worker-supplied. Additive: legacy snapshots deserialize to ""
    // and a run with RootRunID=="" self-roots (see rootRunIDOf).
    RootRunID      string     `json:"root_run_id,omitempty"`
    ParentStepID   string     `json:"parent_step_id,omitempty"`
    Deadline       *time.Time `json:"deadline,omitempty"`
    PriorityOffset int        `json:"priority_offset,omitempty"`
    SingletonKey   string     `json:"singleton_key,omitempty"`
    TraceParent    string     `json:"trace_parent,omitempty"`
    CompletedAt    *time.Time `json:"completed_at,omitempty"`
    // Labels are caller-supplied key/value metadata set at start time
    // (#629). External triggers start runs *about* something (an order,
    // a tenant, a region) and need to find or bulk-cancel those runs
    // later without standing up a shadow lookup table. Additive: legacy
    // snapshots deserialize to nil. See ValidateLabels for bounds.
    Labels map[string]string `json:"labels,omitempty"`
    // TriggerDepth counts hops through run_terminal trigger chains
    // (#634): 0 for every manual/HTTP/cron-started run, source run's
    // TriggerDepth+1 for a run started by a run_terminal trigger
    // reacting to that source run's completion. The runtime half of
    // the loop guard for cross-workflow trigger cycles (A→B→A) — see
    // internal/trigger's TriggerDepthMax cap, enforced before a
    // chained run is ever started. Additive: legacy snapshots
    // deserialize to 0.
    TriggerDepth int `json:"trigger_depth,omitempty"`
    // ReleasePending marks a terminal run whose admission release
    // (singleton lock / concurrency slot / queue advance) did NOT
    // complete when the run was finalized (#648) — the terminal
    // snapshot and both terminal events are still authoritative, but
    // the release itself is owed. The reconciler's terminal-run sweep
    // looks for this flag and retries the release, clearing it on
    // success. Additive: legacy snapshots deserialize to false, which
    // is correct — a run finalized before this field existed either
    // released cleanly or is already outside any recovery window.
    ReleasePending bool `json:"release_pending,omitempty"`
    // ReleaseAttempts counts the reconciler's recovery attempts for a
    // ReleasePending run (#648 PR review round 3). Without a bound, a
    // run whose release keeps failing for a structural reason (not a
    // transient blip) would retry -- and WARN-log -- every reconciler
    // pass forever. The engine package's unexported releaseAttemptsMax
    // constant caps it; reaching the cap clears ReleasePending (the
    // sweep stops, an ERROR is logged once, and
    // engine.finalize.release_abandoned is incremented) rather than
    // retrying unboundedly. Additive: legacy snapshots deserialize to
    // 0, the correct starting count.
    ReleaseAttempts int `json:"release_attempts,omitempty"`
    // DefHash pins the run to the exact def content it started under
    // (#637): the SHA-256 hex digest DefHash(def) computed at run
    // creation, before any dynamic steps are added. loadRunAndDef uses
    // it to re-read that immutable version instead of the mutable
    // name -> latest-def pointer, so a POST /workflows re-register
    // mid-flight never changes what a running run sees on its next
    // advance. Additive: legacy snapshots deserialize to "" and fall
    // back to today's by-name read.
    DefHash string `json:"def_hash,omitempty"`
    // Revision is the snapshot store's optimistic-concurrency handle:
    // the KV revision this run was loaded at. Never serialized. It is
    // a pointer so every value copy made after one load shares it, and
    // a save through any copy advances what the others expect. nil
    // means the run was not loaded from the store. See SnapshotRevision.
    Revision *SnapshotRevision `json:"-"`
}

func NewWorkflowRun 

func NewWorkflowRun(def WorkflowDef, runID string) WorkflowRun

NewWorkflowRun constructs a WorkflowRun with all steps initialized to pending. runID must be non-empty — callers are responsible for providing a unique ID (e.g. nuid.Next()) before calling this constructor.

func (WorkflowRun) EffectiveTime 

func (r WorkflowRun) EffectiveTime() time.Time

EffectiveTime returns the priority-adjusted queue position.

Generated by gomarkdoc