workflow

The workflow package provides the core types for defining and executing agentic workflows in Microbus. It is used by both task authors (who interact with Flow) and workflow authors (who build Graph definitions).

Flow

Flow is the carrier object passed to every task endpoint. It provides access to the step’s identity, the workflow’s state, and control signals.

A Flow is not safe for concurrent use. A task that fans work out to goroutines collects the results and writes them to the flow from a single goroutine; two goroutines writing one flow trip Go’s concurrent-map-write detector, which cannot be recovered and takes down the whole process.

Identity and Timing (Read-Only)

The orchestrator populates these on dispatch. Outside a dispatched task they return the zero value.

• FlowKey() - the external key of the flow this task is executing in, for correlating logs and traces or calling back into the foreman (History, Snapshot) with the flow’s own key
• StepKey() - the external key of the step this task is executing
• Attempt() - the zero-based retry counter of the current step: 0 on the first execution, incremented on each Retry
• CreatedAt() - when the flow was created. Useful for a task’s own elapsed-time guard (e.g. if time.Since(flow.CreatedAt()) > 24*time.Hour { ... })
• UpdatedAt() - the time of the flow’s last status change, such as its start or its latest Resume; it does not move as steps run
• StepCreatedAt() - when this step was first created, preserved across retries. It anchors Retry’s giveUpAfter horizon

State Access

Tasks read input from state and write output back:

// Field-based access (one-off reads/writes)
name := flow.GetString("name")
score := flow.GetFloat("score")
flow.SetBool("approved", true)

// Unmarshal one field into a typed target; ok is false for an absent or cleared field
var applicant MyApplicant
ok, err := flow.Get("applicant", &applicant)

// Check for existence (returns false for cleared/null slots)
if flow.Has("delay") {
    delay := flow.GetFloat("delay")
    // ...
}

// Struct-based access
var state MyState
err = flow.Parse(&state) // unmarshal the whole state into a struct, matched by JSON tag
// ... modify state ...
err = flow.SetChanges(&state) // Set each field the struct marshals

The typed getters (GetString, GetStrings, GetInt, GetFloat, GetBool, GetDuration) return the zero value for an absent or cleared field, and panic on a field that holds a value of the wrong type, which fails the step with the stack trace attached. Use Get where a type mismatch should be handled rather than fail the step: it returns the mismatch as an error. An absent field leaves Get’s target untouched, so a default set beforehand survives.

SetChanges(source) sets each field of source, typically a pointer to a struct, exactly as Set would, under the names encoding/json emits. It takes no snapshot and compares nothing: it sets every field the source marshals, and skips only what encoding/json omits, which includes an empty omitempty or omitzero field. A field left at its zero value under omitzero therefore leaves the state unchanged rather than resetting it. And because every marshaled field is set, a struct filled by Parse writes back every field it read; on a reduced field that is a contribution of the whole value, so set a reduced field on its own instead.

The State Type

workflow.State is the type every state value travels in: flow.Snapshot(), FlowOutcome.State and InterruptPayload, FlowStep.State and Changes, and workflow.BaggageFrom(ctx) all return one. It holds each field as raw JSON and decodes only what is read, so carrying a large field through a task that never touches it costs no unmarshaling.

It carries the same accessors as the flow itself, which is what removes the per-read type assertion:

total := outcome.State.GetInt("count")     // no float64 assertion, no conversion

var messages []llmapi.Item
ok, err := outcome.State.Get("messages", &messages)  // unmarshal one field into a typed target

var summary MySummary
err = outcome.State.Parse(&summary)        // unmarshal the whole state into a struct

GetString, GetStrings, GetInt, GetFloat, GetBool and GetDuration return the zero value for an absent field and panic on a mistyped one, as the flow’s do; Get returns the mismatch as an error instead. Has reports presence, Len counts fields, IsZero reports an unpopulated State, and Names() iterates the field names. Because it is a struct rather than a map, indexing, len(), range and comparison against nil do not apply — use the accessors.

State-Mutation Primitives

Every write goes into both the step’s changes, which the orchestrator persists, and the task’s view of the state, so a later read in the same task sees the field’s combined value.

flow.Set("decision", MyDecision{Approved: true}) // write a field; nil deletes a field with no reducer
flow.SetString("status", "approved")             // typed setters: SetString, SetStrings, SetInt, SetFloat, SetBool, SetDuration
flow.Overwrite("messages", trimmed)              // replace a field whatever its reducer; nil deletes it
flow.Del("blob", "scratch")                      // delete the listed fields: Overwrite with nil
flow.Clear()                                     // delete every field

On a field with no reducer, Set replaces the value and a nil deletes the field. On a field the graph wires with SetReducer, Set is a contribution, combined through the reducer with what the field already holds, and two Sets in one task both count. A nil contributes nothing there. Overwrite replaces a field whatever its reducer, and later Sets in the same task combine onto the new value, which is how a task trims or resets a reduced field such as a conversation history. Del is Overwrite with nil, and Clear is Del on every field, so a dropped field is removed from the state, reduced or not.

Inside a fan-out branch, at any depth, a reduced field cannot be replaced: the branches are parallel, and a reset in one of them has no defined meaning for the others. There Overwrite returns an error and records nothing, and a Del or Clear of a reduced field fails the step. Reset a reduced field before the fan-out or after its fan-in. A field with no reducer can be overwritten or deleted anywhere.

Set returns an error if the value cannot be marshaled to JSON (a NaN, an infinity, a channel), if the field’s reducer cannot combine it (a string under ReducerAdd), or if the field name ends with ! or +, which the engine reserves for its change records. The typed setters panic on the same errors, which fails the step.

Typed Arguments and Returned Values

A task’s typed input arguments are read from the state fields of the same names, and its typed outputs are written back by the generated wrapper with one flow.Set per output after the task body returns. Every value a task returns without an error is applied, zero values included, through the field’s reducer or the default replace; an error return applies nothing, neither the outputs nor the task’s own writes in that run. Because the outputs land after the body, do not both flow.Set a field and return it. The state guide covers the Out-suffix convention and the argument types genservice accepts.

Renaming fields across a contract boundary — reshaping a derived copy of parent state when building a flow.Subgraph in map, which crosses the function-call subgraph boundary — is written as a snapshot, a clear, and explicit re-sets:

snap := flow.Snapshot()
var messages []llmapi.Item
snap.Get("messages", &messages)
flow.Clear()
flow.Set("conversation", messages)

Control Signals

Tasks can influence execution flow:

flow.Goto("Charge")                                                 // jump to a task by its SetURL name
flow.Retry(initialDelay, multiplier, maxIntervalDelay, giveUpAfter) // re-execute this task with backoff
flow.Sleep(5 * time.Second)                                         // delay before next execution
flow.Interrupt(payload, &resume)                                    // park for external input; resume data -> &resume
flow.Subgraph(workflowURL, in, &out)                                // park and run a child workflow; result -> &out

flow.Retry is the single retry primitive. It returns true while the step is still within its retry horizon and false once the horizon is spent. The bound is wall-clock, not a count: giveUpAfter is the elapsed time since the step was first created (flow.StepCreatedAt()), so it includes execution time and survives parks and sleeps — the honest “fail by” deadline. The engine consumes only the backoff shape (initialDelay, multiplier, maxIntervalDelay) to compute the next re-dispatch delay; the give-up decision is made in-task, so the bool tracks return nil vs return err. flow.Retry carries no condition of its own, so the task writes the retryable predicate explicitly in the surrounding if — a condition baked into the method name would make it too easy to retry faults that should never be retried.

err := svc.venv.Call(ctx, "train", in, &out)
if err != nil {
    // Retry only on a timeout (step time-budget or handler deadline, both HTTP 408),
    // backing off 1s→…→1m and giving up 1h after the step was first created.
    if errors.StatusCode(err) == http.StatusRequestTimeout &&
        flow.Retry(time.Second, 2.0, time.Minute, time.Hour) {
        return nil // retry scheduled; suppress the error
    }
    return errors.Trace(err) // non-timeout (or horizon-exceeded) error surfaces to the workflow
}

giveUpAfter <= 0 means retry forever — reserve it for a genuinely unbounded poll, since a permanent failure the predicate misclassifies will loop indefinitely. For count-based bounding instead, gate on flow.Attempt() (the zero-based retry counter): if flow.Attempt() < 5 && flow.Retry(...).

Graph validation rejects an error transition that targets its own source (from == to) because they would loop indefinitely with no backoff budget — flow.Retry is the in-task primitive for re-running the same task with a bounded horizon. Immediate retry is zero delays (flow.Retry(0, 0, 0, giveUpAfter)); unbounded is giveUpAfter <= 0.

Park-and-Resume Signals

flow.Interrupt(payload, &out) and flow.Subgraph(url, in, &out) both park the step and unmarshal their result into the out pointer on re-entry, rather than merging it into state by field name:

var out otherapi.SomeWorkflowOut
yield, err := flow.Subgraph(otherapi.SomeWorkflow.URL(), otherapi.SomeWorkflowIn{Field: value}, &out)
if yield {
    return nil // first pass: parked, child workflow running
}
if err != nil {
    return errors.Trace(err) // child failed; the task may retry, route, or propagate
}
result := out.ResultField // re-entry: out is the child's typed final state

Subgraph always invokes a whole workflow — a task is only ever a node in a graph, so to call a single task on its own you declare a one-node workflow around it and invoke that. There is no primitive that runs a bare task as a standalone child flow.

flow.Interrupt is symmetric: yield, err := flow.Interrupt(payload, &resume) unmarshals the data passed to foremanapi.Resume into resume. Its err reports only a payload that cannot be marshaled, resume data that cannot be unmarshaled into resume, or a second park in the same dispatch. A step parks at most once per dispatch (interrupt XOR subgraph), and a task must set at most one of Goto, Retry, Interrupt, or Subgraph — the full pattern for building agentic workflows walks through each.

A step’s runs compose. After a Retry, a Resume of an Interrupt, or a Subgraph returning, the task runs again from the top, and each run’s writes are kept and combined with the earlier runs’. On a reduced field, a contribution made before the park is therefore made twice: contribute after the park, and return a value that adds nothing (such as nil) on the yield path. A write to a field with no reducer simply lands again, so state written before the park survives it and is visible on re-entry, which is what lets a task record an external resource’s identifier before parking and skip re-creating it afterwards.

From another service’s typed client, prefer otherapi.NewSubgraph(flow).SomeWorkflow(ctx, args...) (→ Subgraph) over the raw primitive — it marshals the typed In/Out for you. The Executor client is test-only; do not call it (or foremanapi) from a task body.

Graph

Graph defines the structure of a workflow: tasks, transitions between them, and the reducers that combine writes to state fields.

graph := workflow.NewGraph("CreateOrder")
graph.SetURL("Validate", myserviceapi.Validate.URL())
graph.SetURL("Charge", myserviceapi.Charge.URL())
graph.SetURL("Reject", myserviceapi.Reject.URL())
graph.SetURL("Fulfill", myserviceapi.Fulfill.URL())
graph.AddTransitionSwitch("Validate", "Charge", "valid == true")
graph.AddTransitionSwitch("Validate", "Reject", "true") // default branch
graph.AddTransition("Charge", "Fulfill")
graph.AddTransition("Fulfill", workflow.End)
graph.AddTransition("Reject", workflow.End)

Tasks are registered with SetURL(name, url) and then referenced by name in transitions, keeping graph definitions readable and decoupled from where each task is implemented. A task named in a transition but never bound with SetURL is registered with its name as its URL, so a forgotten SetURL is not caught by Validate and surfaces only when the task is first dispatched. The first task registered is the entry point unless SetEntryPoint(name) names another. Tasks() lists the registered tasks as Task values, Transitions() the edges, Reducers() the field reducers, and URLOf(name) resolves a task name to its URL.

Transition Types

An error transition fires on any error; the engine does not inspect the status code. A handler that should treat a timeout differently reads the status from onErr (both a step’s time budget and a handler’s own deadline surface as 408). An error voids the changes the failing run made, so the handler sees the state as it was before that run, including what the step’s earlier runs wrote before a Retry or a park.

Fan-Out and Fan-In

Any task with two or more ordinary outgoing transitions (not switch, goto or error), or any forEach outgoing transition, is a fan-out source. SetFanIn(name) marks the node where the parallel branches fan back in. Every fan-out must be fanned back in: Validate tracks each fan-out as a frame pushed onto a stack and requires the matching SetFanIn node to pop it before the branches reach workflow.End.

graph.SetURL("Critique", myserviceapi.Critique.URL())
graph.SetURL("TranslateFrench", myserviceapi.TranslateFrench.URL())
graph.SetURL("TranslateSpanish", myserviceapi.TranslateSpanish.URL())
graph.SetURL("Finalize", myserviceapi.Finalize.URL())
graph.SetFanIn("Finalize")
graph.AddTransition("Critique", "TranslateFrench")
graph.AddTransition("Critique", "TranslateSpanish")
graph.AddTransition("TranslateFrench", "Finalize")
graph.AddTransition("TranslateSpanish", "Finalize")
graph.AddTransition("Finalize", workflow.End)

Validate returns an error if a fan-out source has no fan-in node downstream, if a branch reaches workflow.End with unpopped fan-out frames, or if a node is reached with two different lineage stacks. The last case means the same node sits inside two distinct fan-out scopes - register a second alias with SetURL(otherName, sameURL) so each scope has its own node.

flow.Goto and error transitions do not push a frame; they stay in the source’s scope.

Reducers

Reducers control how writes to a state field combine. On a field wired with graph.SetReducer(field, reducer), every write is a contribution, folded through the reducer wherever one step hands its state to the next: the next step, a fan-in (in branch order), the final state, and Continue. A read returns the combined value. Fields with no registered reducer use ReducerReplace (last write wins). Field names carry no hidden semantics, and the field must be a plain identifier (an ASCII letter or underscore, then ASCII letters, digits or underscores).

graph.SetReducer("total",    workflow.ReducerAdd)     // sum numeric contributions
graph.SetReducer("messages", workflow.ReducerAppend)  // concatenate arrays, duplicates kept
graph.SetReducer("seen",     workflow.ReducerUnion)   // merge arrays, deduplicated
graph.SetReducer("attrs",    workflow.ReducerMerge)   // merge objects, new key wins
graph.SetReducer("approved", workflow.ReducerAnd)     // logical AND across branches

Reducers are strict on their input type: ReducerMerge rejects array values, ReducerAdd / ReducerMin / ReducerMax reject strings, ReducerConcat rejects non-strings, and so on; ReducerAppend and ReducerUnion spread an array and add any other value as a single element. A nil (JSON null) is the only value that contributes nothing under every reducer; a zero value is folded like any other. Because a write is a contribution, a task writes what it adds, never the value it read, and uses flow.Overwrite to reset the field. The full reducer semantics cover the contribution rules, Out pairs, parks and subgraphs.

Time Budgets

A task’s execution timeout is a property of the task endpoint, not the graph — declare it on the endpoint’s subscription with sub.TimeBudget. The foreman’s TimeBudget config is the hard ceiling applied to every dispatch; the effective deadline is the smaller of the two. The graph itself carries no timing.

Validation and Visualization

err := graph.Validate()                                // check for structural errors
mermaid := workflow.NewGraphRenderer(graph).Render() // render a Mermaid flowchart

Validate runs when the foreman fetches a graph, so an invalid graph compiles and fails only when a flow is created. GraphRenderer renders any graph, including one that would fail Validate, and is configured with WithTopDown() / WithLeftRight(), WithPrimaryColors(fill, text), WithSecondaryColors(fill, text), and WithLinks(paramName), which makes each task node a link to ?<paramName>=<taskName>.

Transition

Transition describes a single edge in the graph:

type Transition struct {
    From    string // source task name
    To      string // target task name (or workflow.End)
    Kind    string // "" (ordinary), TransitionSwitch, TransitionGoto or TransitionError
    When    string // boolean expression over state; conditions an ordinary or switch transition
    ForEach string // state array to fan out over, one branch per element (ordinary transitions only)
    As      string // name of the element in each forEach branch's state
}

Each transition is exactly one kind. An ordinary transition (Kind empty) is taken when the source succeeds and its When, if any, holds, and every matching ordinary transition is taken. TransitionSwitch is taken by the first switch transition whose When holds and never fans out. TransitionGoto is taken only when the task calls flow.Goto with its destination, and TransitionError only when the task returns an error. Goto and error transitions take no When, and only an ordinary transition can carry ForEach.

Task

Task describes a task registered in a graph, as listed by Graph.Tasks(). Every task is a plain task — there is no static subgraph type; a child workflow is invoked at runtime via flow.Subgraph(url, in, &out) from a task body.

type Task struct {
    Name string // the task's name in the graph, used by transitions
    URL  string // the task endpoint's dispatch URL
}

Reducer

Reducer is a string constant naming how writes to a field combine:

const (
    ReducerReplace Reducer = "replace" // Last write wins (default)
    ReducerAppend  Reducer = "append"  // Concatenate arrays
    ReducerAdd     Reducer = "add"     // Sum numeric values
    ReducerMin     Reducer = "min"     // Smaller of two numeric values
    ReducerMax     Reducer = "max"     // Larger of two numeric values
    ReducerUnion   Reducer = "union"   // Merge arrays, deduplicate (numbers compare by value)
    ReducerMerge   Reducer = "merge"   // Merge objects, new key wins
    ReducerAnd     Reducer = "and"     // Logical AND of booleans
    ReducerOr      Reducer = "or"      // Logical OR of booleans
    ReducerConcat  Reducer = "concat"  // Concatenate strings
)

FlowOutcome

FlowOutcome is the single struct returned by the foreman’s Snapshot, Await, Poll, and Run. Side-channel fields are populated only for the matching status:

type FlowOutcome struct {
    Status           string // lifecycle status (see Status constants below)
    State            State  // accumulated state at the moment in time
    Error            string // populated when Status == StatusFailed
    InterruptPayload State  // populated when Status == StatusInterrupted
    TerminateReason  string // populated when Status == StatusTerminated
    CancelReason     string // populated when Status == StatusCancelled
}

The Go-level error return from Snapshot/Await/Run is reserved for transport, foreman, and timeout failures. A workflow failure is signalled by outcome.Status == workflow.StatusFailed with outcome.Error populated - callers branch on the outcome’s Status field rather than on a non-nil Go error.

Snapshot of an interrupted flow returns State as the merged step snapshot at the time of the interrupt and InterruptPayload as the raw flow.Interrupt(payload) argument - the two are kept separate so callers can tell which fields are workflow state and which are the resume request. A caller that wants them merged does it explicitly, folding the payload onto a copy of the state as contributions through the graph’s reducers:

merged := outcome.State.Clone()
err := merged.Apply(workflow.ChangesOf(outcome.InterruptPayload), graph.Reducers())

Status Constants

const (
    StatusCreated     = "created"     // Step exists but has not been started; a flow is never reported as created
    StatusPending     = "pending"     // Step is awaiting execution
    StatusRunning     = "running"     // Flow is actively executing a task
    StatusInterrupted = "interrupted" // Flow is paused, waiting for external input
    StatusCompleted   = "completed"   // Flow has finished successfully
    StatusFailed      = "failed"      // Flow has failed with an error
    StatusTerminated  = "terminated"  // Flow was forcefully stopped by Terminate; in-flight work was abandoned
    StatusCancelled   = "cancelled"   // Flow was gracefully stopped by Cancel
)

IsTerminalStatus(s) reports whether a status is one of the terminal four, completed, failed, terminated and cancelled, and IsValidStatus(s) whether it is any of the eight. IsCancelled(err) reports whether an error is a cancellation delivered by a graceful Cancel, so an onError handler can tell it apart from a real failure.

FlowOptions

FlowOptions sets a flow’s policy at genesis. It is passed as the last argument to the foreman’s Create / Run when submitting a flow. Derived operations take no options and inherit instead: Continue inherits the thread’s policy, Fork the origin’s, and a subgraph its parent’s. A nil *FlowOptions, or any zero field, uses the defaults — the workflow package never interprets these values; the foreman owns all defaulting.

type FlowOptions struct {
    Priority           int           // 1 to the foreman's MaxPriority (5 by default), lower runs first; 0 takes the midpoint (3 by default)
    FairnessKey        string        // groups flows for fair scheduling, typically a tenant; empty defaults to the caller's tenant
    FairnessWeight     float64       // relative dispatch share of the fairness key; 0 means weight 1
    TimeBudget         time.Duration // per-task time budget for this flow; 0 uses the foreman's TimeBudget
    DeleteOnCompletion bool          // delete the flow and its subgraphs shortly after it completes successfully
    Baggage            any           // opaque context carried with the flow; the foreman sets it to the caller's actor claims
    ThreadKey          string        // join an existing thread; empty starts a fresh one
}

Create and Run reject a priority above the foreman’s MaxPriority. When FairnessKey is empty, the foreman uses the caller’s tenant ID from its actor claims, so tenants share capacity fairly without the caller naming them; a caller with no tenant lands in the "" bucket. The foreman also rejects a TimeBudget over 15 minutes.

DeleteOnCompletion is for fire-and-forget jobs whose output is not retained. Deletion is deferred: for a short grace window Await and Snapshot still return the completed outcome, then the flow is removed. Failed, terminated and cancelled flows are kept.

Baggage is opaque to the engine: it is stored with the flow, inherited by subgraphs and Continue, and handed to the host on every dispatch, where workflow.BaggageFrom(ctx) reads it back as a State. In Microbus the foreman is that host and uses it for the actor: it sets Baggage to the caller’s actor claims at Create, replacing any value the caller set when the call carries an actor, and mints each task dispatch an access token from it, so every task runs as the actor that created the flow. Leave it unset.

ThreadKey places the new flow into an existing thread instead of starting its own — pass any flow key already in the thread, and the engine routes the new flow to the thread’s shard. It is the explicit-policy counterpart to Continue, which joins a thread by inheriting its policy wholesale.

The priority and fairness guide covers the concepts and usage in full.

RawFlow

RawFlow wraps Flow with additional methods used by the foreman orchestrator. Task authors do not interact with RawFlow directly. It exposes the raw state as a State (RawState, SetRawState) and the step’s raw changes as workflow.Changes (RawChanges, SetRawChanges), which record each written field as a contribution, an overwrite, or an addition. It also holds the readers of the control signals a task set (GotoRequested, RetryRequested, SleepRequested, InterruptRequested, SubgraphRequested) and the setters orchestration bookkeeping needs (SetReducers, SetInBranch, SetAttempt, SetFlowKey, SetStepKey and the timestamps).

Seeding a carrier’s state is a raw-orchestration operation, which is why the test-only Executor stub’s WithInputFlow takes a *workflow.RawFlow — a test that needs a task to start from populated state builds the carrier at this level.

Further Reading

• Agentic workflows - conceptual overview of tasks, workflows and flows
• Building agentic workflows - step-by-step guide to implementing workflows
• Priority and Fairness - flow priority and fairness via FlowOptions
• Package coreservices/foreman - Foreman configuration and endpoints