Reducers

Reducers control how parallel branches in a workflow merge their state when they fan in. When two or more branches write to the same field, the Foreman needs to know whether to add the values, append them to a list, take their union, or just keep the last write. Reducers answer that question.

In Microbus, reducers are wired explicitly at graph-build time with graph.SetReducer(field, reducer). A field name carries no hidden meaning — messages, total, and tags are all ordinary fields until a SetReducer call attaches a merge strategy. Fields with no registered reducer use Replace (last write wins).

The Fan-In Problem

When a graph fans out into parallel branches and those branches converge on a shared successor, every branch writes its piece of state. If two branches write to the same field, the framework has to combine those writes somehow.

Consider this graph:

graph.SetFanIn("decide")

graph.AddTransition("submit", "verifyCredit")
graph.AddTransition("submit", "verifyIdentity")
graph.AddTransition("submit", "verifyEmployment")

graph.AddTransition("verifyCredit", "decide")
graph.AddTransition("verifyIdentity", "decide")
graph.AddTransition("verifyEmployment", "decide")

SetFanIn("decide") marks decide as the node where the three parallel branches fan back in. The Foreman parks each branch at decide until its siblings arrive, then runs the reducers below to merge their writes.

Each verification task writes to a failures field if it failed:

func (svc *Service) VerifyCredit(ctx context.Context, flow *workflow.Flow, score int) (failures []string, err error) {
    if score < 600 {
        return []string{"low credit score"}, nil
    }
    return nil, nil
}

When decide runs, it expects to see all the failures from all three verifications. But three branches are about to write to the same failures field. Without a reducer, fan-in would just pick one (last write wins) and silently lose the others. Wiring the field with the append reducer fixes that:

graph.SetReducer("failures", workflow.ReducerAppend)

No task change, no per-task configuration — the merge strategy lives next to the rest of the graph definition.

The Reducer Vocabulary

graph.SetReducer(field, reducer) accepts any of these strategies:

ReducerReplace is the default, so you only call SetReducer for the fields that need to accumulate. Wire each non-default fan-in field at graph-build time:

graph.SetReducer("messages", workflow.ReducerAppend)  // accumulate per-branch deltas
graph.SetReducer("total",    workflow.ReducerAdd)     // sum numeric contributions
graph.SetReducer("seen",     workflow.ReducerUnion)   // dedupe across branches
graph.SetReducer("attrs",    workflow.ReducerMerge)   // merge per-branch objects
graph.SetReducer("approved", workflow.ReducerAnd)     // all branches must approve
graph.SetReducer("flagged",  workflow.ReducerOr)      // any branch flags
graph.SetReducer("notes",    workflow.ReducerConcat)  // join string deltas
graph.SetReducer("lowScore", workflow.ReducerMin)     // smallest contribution wins
graph.SetReducer("topScore", workflow.ReducerMax)     // largest contribution wins

Making the reducer explicit costs one line per accumulating field, but it keeps state field names free of hidden execution semantics — a field is just a field, and the merge behavior is declared where the rest of the graph is.

The Delta Rule

Reducers operate on the delta each branch produces, not on the full accumulated value. This is the rule that bites in practice. Assume failures is wired with ReducerAppend.

Wrong:

func (svc *Service) VerifyCredit(ctx context.Context, flow *workflow.Flow, failures []string, score int) (failuresOut []string, err error) {
    if score < 600 {
        return append(failures, "low credit score"), nil
    }
    return failures, nil
}

This task reads the existing failures, appends to it, and returns the appended array. Across three parallel branches, fan-in receives three arrays that already contain each other’s contents. The append reducer then concatenates them all — producing duplicates of every entry up to nine times.

Right:

func (svc *Service) VerifyCredit(ctx context.Context, flow *workflow.Flow, score int) (failures []string, err error) {
    if score < 600 {
        return []string{"low credit score"}, nil
    }
    return nil, nil
}

Each branch writes only its own contribution. The reducer combines them. The branches do not need to know what the other branches produced.

The same rule applies to ReducerAdd (return the increment, not the running total), ReducerUnion (return the elements you want added, not the existing set), and ReducerConcat (return your branch’s fragment, not the joined string).

ReducerUnion and ReducerMerge Semantics

The two structural reducers are strict about their value type — unlike the old prefix-driven set*, which dispatched on the value’s JSON kind. ReducerUnion is for arrays; ReducerMerge is for objects.

ReducerUnion: Array Element Union

graph.SetReducer("tags", workflow.ReducerUnion)

// Branch A returns: tags = ["red", "blue"]
// Branch B returns: tags = ["blue", "green"]
// Fan-in produces: tags = ["red", "blue", "green"]

The de-duplication is by JSON value. For arrays of primitives (strings, numbers, booleans) this is exact-match dedup. For arrays of objects, two objects are equal only if their serialized JSON is equal — so dedup is sensitive to field ordering and number formatting. Values produced through Go’s encoding/json are deterministic; hand-built json.RawMessage is the danger zone. ReducerUnion rejects object values.

ReducerMerge: Object Field-by-Field Merge

graph.SetReducer("users", workflow.ReducerMerge)

// Branch A returns: users = {"alice": {...}, "bob":   {...}}
// Branch B returns: users = {"bob":   {...}, "carol": {...}}
// Fan-in produces: users = {"alice": {...}, "bob":   {...}, "carol": {...}}

If both branches write the same key, the new key wins for that key. Inner objects are not recursively merged — merge is one level deep. ReducerMerge rejects array values.

Type Strictness

Reducers are strict on their input type. ReducerUnion rejects object values, ReducerMerge rejects array values, ReducerAdd / ReducerMin / ReducerMax reject strings, and so on. A type mismatch fails the step rather than silently coercing.

The one exception is a cleared slot. A branch that returns Go nil (or whose field was removed via flow.Clear / flow.Delete) short-circuits to the reducer’s identity rather than failing the type check, so that branch contributes nothing to the merge.

Edge Cases

Empty Branches and Empty Cohorts

A forEach transition over an empty array spawns zero branches. The Foreman still fires the fan-in step directly, and the field takes the reducer’s identity — the Go zero value for its type (0 for Add, [] for Append / Union, {} for Merge, false for And / Or, "" for Concat).

Both ReducerAnd and ReducerOr return false on an empty cohort — there is no “vacuously true” mode. If you need an empty cohort to be treated as approval, branch on the cohort size downstream rather than relying on the reducer.

ReducerMin and ReducerMax have no algebraic identity (0 is a legitimate value, not a neutral one), so they fold cleared slots asymmetrically: a cleared side defers to the other side rather than collapsing to 0. An all-cleared empty cohort therefore leaves the field absent on the fan-in step. A workflow that needs a defined Min / Max value when the cohort might be empty must seed the field upstream of the fan-out.

nil and Cleared Writes

A branch that returns nil for a reducer-managed field is a no-op for that field — it folds to the reducer’s identity and is ignored. This is the clean way to express “this branch has nothing to contribute”:

func (svc *Service) VerifyCredit(ctx context.Context, flow *workflow.Flow, score int) (failures []string, err error) {
    if score < 600 {
        return []string{"low credit score"}, nil
    }
    return nil, nil  // no-op for fan-in
}

flow.Clear / flow.Delete on a reducer-managed field has the same effect — the cleared contribution folds to the identity.

Ordering in ReducerAppend and ReducerConcat

ReducerAppend (arrays) and ReducerConcat (strings) preserve the cohort merge order, which is updated_at, step_id — the order the branches’ steps last changed, not the order they were created under shard scatter, and not branch-completion order. The result is deterministic, but if you need a specific order — alphabetical, by score, fan-out index — sort the result downstream of the fan-in:

func (svc *Service) Decide(ctx context.Context, flow *workflow.Flow, failures []string) (decision string, err error) {
    sort.Strings(failures) // re-order from cohort order to alphabetical
    if len(failures) > 0 {
        return "rejected", nil
    }
    return "approved", nil
}

ReducerAdd, ReducerMin, ReducerMax, ReducerUnion, and ReducerMerge are order-insensitive in their result (commutative, or last-write-wins per key), so ordering never matters there.

Subgraphs

A subgraph is a function call: the calling task invokes flow.Subgraph(url, in, &out) and receives the child’s final state in out on re-entry. The child’s state is not auto-merged into the parent — the calling task adopts the fields it wants and writes them back. So whether a subgraph’s output participates in a parent fan-in is up to the calling task: if it writes the adopted field, that write is one branch’s contribution and goes through the parent’s reducer like any other. The subgraph can have its own reducer fan-ins internally; they resolve before the child terminates and returns.

Reducer Selection Cannot Vary by Branch

The reducer for a field is fixed by SetReducer for the lifetime of the flow. Two branches writing the same field always go through the same reducer — there is no “this branch should sum, that branch should append” mode.

See Also

• State - the broader treatment: how state splits at fan-out (where reducers come into play), forEach element aliasing with injected index and count, flow.Delete / Clear / Keep / Transform for housekeeping, and the function-call subgraph boundary.
• Building Agentic Workflows - the fan-out and fan-in patterns where reducers come into play.
• Credit Flow example - the employmentFailures field is wired with graph.SetReducer("employmentFailures", workflow.ReducerAdd) and reduced across every forEach branch of VerifyEmployment; tasks return per-branch deltas, never the running total.
• Workflows overview - conceptual background on tasks, graphs, and state.
• Package workflow - the Reducer type and the constants used by SetReducer.
• Package coreservices/foreman - the orchestration engine that applies reducers at fan-in time.