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 workflow’s identity, state, and control signals.

Identity Fields (read-only)

• FlowID - unique identifier for this flow execution
• WorkflowName - the workflow graph’s name
• TaskName - the current task being executed
• StepNum - the current step number in the flow
• CreatedAt() / UpdatedAt() - wall-clock timestamps of the flow row’s creation and its last status transition, populated by the orchestrator on dispatch. Useful for tasks that want their own elapsed-time guard (e.g. if time.Since(flow.CreatedAt()) > 24*time.Hour { ... }).

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.Set("approved", true)

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

// Struct-based access (parse/diff pattern)
snap := flow.Snapshot() // returns map[string]any
var state MyState
flow.ParseState(&state)
// ... modify state ...
flow.SetChanges(state, snap) // only records fields that differ from snap

State-Mutation Primitives

Four methods drop fields from state in shapes that mirror Go’s map builtins. Each writes JSON null into the step’s changes for the dropped fields and removes them from the in-memory state map so subsequent reads within the same task see them as absent. At fan-in, a dropped field contributes the reducer’s identity (Replace = field absent; ReducerAdd = 0; ReducerAppend / ReducerUnion = empty array; and so on).

flow.Delete("blob", "scratch")          // drop the listed fields
flow.Clear()                             // drop every field
flow.Keep("ssn", "address")              // drop everything except those listed
flow.Transform("subInput", "parentVar",  // clear all, then re-introduce listed
              "subInput2", "parentVar2") //   fields under new names (pairs)

flow.Transform panics on an odd argument count. Old keys that were absent or already null are skipped, so the new key isn’t introduced as null. These primitives are handy for reshaping a derived copy of parent state when building a flow.Subgraph in map, which crosses the function-call subgraph boundary.

Control Signals

Tasks can influence execution flow:

flow.Goto("Charge")                                           // jump to a graph node by its AddTask 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 OnError / OnTimeout self-transitions (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 is always nil. 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.

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 reducers for fan-in state merging.

graph := workflow.NewGraph("CreateOrder")
graph.SetEndpoint("Validate", myserviceapi.Validate.URL())
graph.SetEndpoint("Charge", myserviceapi.Charge.URL())
graph.SetEndpoint("Reject", myserviceapi.Reject.URL())
graph.SetEndpoint("Fulfill", myserviceapi.Fulfill.URL())
graph.AddTransition("Validate", "Charge")
graph.AddTransitionWhen("Validate", "Reject", "valid != true")
graph.AddTransition("Charge", "Fulfill")
graph.AddTransition("Fulfill", workflow.END)

Tasks are registered with SetEndpoint(name, url) and then referenced by name in transitions, keeping graph definitions readable and decoupled from where each task is implemented.

Transition Types

Fan-Out and Fan-In

Any node with two or more non-goto/non-error outgoing transitions, 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.SetEndpoint("Critique", myserviceapi.Critique.URL())
graph.SetEndpoint("TranslateFrench", myserviceapi.TranslateFrench.URL())
graph.SetEndpoint("TranslateSpanish", myserviceapi.TranslateSpanish.URL())
graph.SetEndpoint("Finalize", myserviceapi.Finalize.URL())
graph.SetFanIn("Finalize")
graph.AddTransition("Critique", "TranslateFrench")
graph.AddTransition("Critique", "TranslateSpanish")
graph.AddTransition("TranslateFrench", "Finalize")
graph.AddTransition("TranslateSpanish", "Finalize")

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 SetEndpoint(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 state fields are merged at fan-in points. Every field that needs anything other than last-write-wins is wired explicitly with graph.SetReducer(field, reducer) at graph-build time; fields with no registered reducer use ReducerReplace. Field names carry no hidden semantics.

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: ReducerUnion rejects object values, ReducerMerge rejects array values, ReducerAdd / ReducerMin / ReducerMax reject strings, and so on. A cleared slot (Go nil or JSON null, e.g. from flow.Clear) short-circuits to the reducer’s identity rather than failing the type check, so that branch contributes nothing. The full reducer semantics cover identities and the delta convention.

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 := graph.Mermaid() // generate a Mermaid flowchart

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)
    When       string // boolean expression evaluated against state
    WithGoto   bool   // only taken on explicit flow.Goto()
    ForEach    string // state field to iterate over (dynamic fan-out)
    As         string // alias for current element during forEach
    OnError    bool   // taken when the source task returns an error
    StatusCode int    // when non-zero (e.g. 408), match only that error status
}

Node

Node describes a task node registered in a graph. Every node is a plain task — there is no static subgraph node type; a child workflow is invoked at runtime via flow.Subgraph(url, input) from a task body.

type Node struct {
    Name string // local name used in transitions
    URL  string // task endpoint URL
}

Reducer

Reducer is a string constant defining a merge strategy:

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
    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            map[string]any // accumulated state at the moment in time
    Error            string         // populated when Status == StatusFailed
    InterruptPayload map[string]any // populated when Status == StatusInterrupted
    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. Use workflow.MergeState(out.State, out.InterruptPayload, graph.Reducers()) if you need the pre-merge behavior.

Status Constants

const (
    StatusCreated     = "created"     // Flow/step exists but has not been started
    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
    StatusRetried     = "retried"     // Step was failed but has been retried (replaced by a new step)
    StatusCancelled   = "cancelled"   // Flow was cancelled by the user
)

FlowOptions

FlowOptions sets flow-level scheduling properties at create time. It is passed as the last argument to the foreman’s Create / Run when submitting a flow. A nil *FlowOptions, or any zero field, uses the foreman’s defaults — the workflow package never interprets these values; the foreman owns all defaulting.

type FlowOptions struct {
    Priority       int           // >= 1, lower runs first; 0 means to use the foreman's DefaultPriority
    FairnessKey    string        // groups flows for fair scheduling; empty derives from the tenant claim
    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 engine default
    ThreadKey      string        // join an existing thread; empty starts a fresh one
}

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 raw state/config/changes access, control signal readers, and mutation methods needed for orchestration bookkeeping.

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