Reducers

Reducers control how writes to a workflow’s state fields combine. Without one, a field holds whatever was written to it last. With one, each write is a contribution that the Foreman folds into the field’s value: adding numbers, appending to a list, taking a union, and so on.

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 reducer. Fields with no registered reducer use Replace (last write wins).

Contributions at Every Hand-Off

A write to a field with a reducer is a contribution, folded through the reducer wherever one step hands its state to the next:

• the next step in a straight line,
• the fan-in, where parallel branches converge,
• the flow’s final state,
• and Continue, which folds the new input onto the thread’s last state.

A read always returns the combined value. So the result of a reduced field does not depend on the graph’s shape: the same rules hold in a straight line, inside a fan-out branch and at a fan-in.

Consider a conversation history wired with ReducerAppend:

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

A task that adds a reply writes only the reply:

func (svc *Service) Reply(ctx context.Context, flow *workflow.Flow, messages []llmapi.Item) (err error) {
    reply, err := svc.generateReply(ctx, messages) // messages is the whole history
    if err != nil {
        return errors.Trace(err)
    }
    return flow.Set("messages", []llmapi.Item{reply}) // contributes one item
}

The task reads the whole history, contributes one item, and the next step sees the history with the reply at the end. Two Sets of a reduced field in one task both count.

The Reducer Vocabulary

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

ReducerReplace is the default, so you only call SetReducer for the fields that need to combine:

graph.SetReducer("messages", workflow.ReducerAppend)  // accumulate each write's elements
graph.SetReducer("total",    workflow.ReducerAdd)     // sum numeric contributions
graph.SetReducer("seen",     workflow.ReducerUnion)   // dedupe across writes
graph.SetReducer("attrs",    workflow.ReducerMerge)   // merge objects
graph.SetReducer("approved", workflow.ReducerAnd)     // every contribution must approve
graph.SetReducer("flagged",  workflow.ReducerOr)      // any contribution flags
graph.SetReducer("notes",    workflow.ReducerConcat)  // join string contributions
graph.SetReducer("lowScore", workflow.ReducerMin)     // smallest contribution wins
graph.SetReducer("topScore", workflow.ReducerMax)     // largest contribution wins

Making the reducer explicit costs one line per combining field, but it keeps state field names free of hidden execution semantics — a field is just a field, and the combining behavior is declared where the rest of the graph is. The field must be a plain identifier (an ASCII letter or underscore, then ASCII letters, digits or underscores), and the reducer for a field is fixed for the lifetime of the flow: every write to it goes through the same reducer.

nil and Zero Values

A nil (JSON null) is the only value that contributes nothing under every reducer, which makes it the clean way to say “this write adds nothing”. A zero value is folded like any other: 0 changes nothing under Add but competes like any other number under Min, false changes nothing under Or but makes And false, and "" changes nothing under Concat.

Type Strictness

Reducers are strict on their input type. ReducerMerge rejects array values, ReducerAdd / ReducerMin / ReducerMax reject strings, ReducerConcat rejects anything but a string, and ReducerAnd / ReducerOr anything but a boolean. ReducerAppend and ReducerUnion spread an array’s elements and add any other value, an object included, as a single element; they reject a field that already holds something other than an array. A type mismatch fails the write rather than silently coercing: flow.Set returns the error, and a typed setter or a task’s returned value fails the step.

Fan-In

When a graph fans out into parallel branches and those branches converge on a shared successor, each branch contributes to the same fields. 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")

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

SetFanIn("Decide") marks Decide as the task where the three parallel branches converge. The Foreman waits for every branch to arrive, then folds their contributions into Decide’s state. Each verification contributes its own failures:

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 // contributes nothing
}

Decide sees every branch’s failures appended to whatever failures held at the fan-out. Without the reducer, the branches’ writes would replace one another and only one branch’s value would survive.

A forEach fan-out follows the same rule, one contribution per element. The Credit Flow example wires employmentFailures with ReducerAdd and has VerifyEmployment return 0 or 1 for its own employer, so the fan-in sees the total across employers.

Write Your Contribution, Not the Value You Read

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
}

The task reads the existing failures, appends to it, and returns the result. That value is a contribution too, so the history is appended again. In a straight line every entry is duplicated by every step that does this; across three parallel branches each branch adds the whole history once more.

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 write carries only what it adds, and the reducer does the combining. The task does not need to know what other steps or branches produced. The same rule applies to ReducerAdd (return the increment, not the running total), ReducerUnion (return the elements to add, not the existing set), and ReducerConcat (return your fragment, not the joined string).

Out Pairs on a Reduced Field

A task argument with an Out suffix maps to the same state field as its input: items in, itemsOut out. On a field with no reducer the pair is a plain read-modify-write. On a reduced field it invites exactly the mistake above, so return only the contribution:

func (svc *Service) AddTurn(ctx context.Context, flow *workflow.Flow, items []llmapi.Item) (itemsOut []llmapi.Item, err error) {
    if nothingToSay(items) {
        return nil, nil // adds nothing; returning items would add the history again
    }
    reply, err := svc.generateReply(ctx, items)
    if err != nil {
        return nil, errors.Trace(err)
    }
    return []llmapi.Item{reply}, nil // only the new item
}

A pass-through branch such as if nothingToDo { return items, nil } adds the history again, so on every path that should add nothing, return a value that changes nothing under the field’s reducer: nil for a list, 0 under Add, false under Or. The CallLLM task of the LLM service’s ChatLoop follows this shape: it reads the whole items conversation and returns only the turn it adds, or nil when it schedules a retry.

Every Returned Value Is Applied

Any value a task returns without an error is applied to the state, through the field’s reducer or the default Replace, on every successful return. Zero values are applied like any other: under Replace a zero value replaces the field and a nil deletes it, while under a reducer a zero value folds as described above. An error return applies nothing.

The generated task wrapper applies the outputs after the task body returns, with one flow.Set per declared output, so a returned value lands after anything the body wrote itself. Do not both flow.Set a field and return it as an output: on a field with no reducer the returned value overwrites the Set, and on a reduced field both count. Use one or the other.

Resetting a Reduced Field

Because every Set is a contribution, trimming or resetting a reduced field takes flow.Overwrite, which replaces the field whatever its reducer:

func (svc *Service) TrimHistory(ctx context.Context, flow *workflow.Flow, messages []llmapi.Item) (err error) {
    if len(messages) <= 20 {
        return nil
    }
    return flow.Overwrite("messages", messages[len(messages)-20:]) // keep the last 20
}

Later Sets in the same task combine onto the new value. flow.Del(name) is Overwrite with nil: it removes the field, reduced or not. flow.Clear() deletes every field the same way. The CallLLM task deletes its Append-reduced toolResults field once it has folded a round’s results into the conversation, so the next round’s fan-in starts from empty.

Inside a fan-out branch a reduced field cannot be replaced, because the branches run in 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, including inside a branch.

Parks, Retries and Subgraphs

A step’s runs compose. After a flow.Retry, a Resume of a flow.Interrupt, or a flow.Subgraph returning, the task runs again from the top, and each run’s contributions are added to the earlier runs’. A contribution made before the park is therefore made twice. Contribute after the park, and return nothing on the yield path:

func (svc *Service) CollectReview(ctx context.Context, flow *workflow.Flow, draft string) (notes []string, err error) {
    var resume struct {
        Note string `json:"note"`
    }
    yield, err := flow.Interrupt(map[string]any{"draft": draft}, &resume)
    if err != nil || yield {
        return nil, err // parked: contribute nothing
    }
    return []string{resume.Note}, nil // contribute once, after the Resume
}

A task that schedules a Retry returns before contributing for the same reason. On a field with no reducer a write simply lands again, so the yield path of an Out pair returns the input unchanged rather than a zero value that would overwrite the field.

A subgraph returns only what it adds. A subgraph is a function call: its final state comes back to the calling task in out and is not merged into the parent. If the calling task writes a child’s result into a reduced field of the parent, that write is a contribution like any other, so have the child return its new elements, not the history it was given. A child’s own reducers fold inside the child before it returns.

ChatLoop is the common exception to plan for: it returns the whole conversation, not just what it added. A parent normally gives its items field no reducer, so the run task’s write replaces it, which is correct. If the parent wires items with ReducerAppend and passed its history in, the run task must contribute only the items past the prompt it sent.

ReducerUnion and ReducerMerge Semantics

The two structural reducers are strict about their value type: ReducerUnion is for arrays, ReducerMerge is for objects.

ReducerUnion: Array Element Union

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

// tags holds:          ["red", "blue"]
// A task contributes:  ["blue", "green"]
// The next step sees:  ["red", "blue", "green"]

The field’s existing elements are kept as they are, and each contributed element is added only if an equal one is not already present. Elements compare by decoded value, so the key order of an object does not matter, and numbers compare by value however they are spelled (1, 1.0 and 1e0 match). The number 1 never matches the string "1".

ReducerMerge: Object Key-by-Key Merge

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

// users holds:         {"alice": {...}, "bob":   {...}}
// A task contributes:  {"bob":   {...}, "carol": {...}}
// The next step sees:  {"alice": {...}, "bob":   {...}, "carol": {...}}

When the contribution has a key the field already holds, the contribution’s value wins for that key. Inner objects are not recursively merged — merge is one level deep. ReducerMerge rejects array values.

Edge Cases

Empty Cohorts

A forEach transition over an empty array spawns zero branches. The Foreman still creates the fan-in step, which runs on the source task’s own state with no branch contributions, so a reduced field keeps its incoming value (absent if nothing wrote it).

There is no “vacuously true” ReducerAnd: an empty cohort contributes nothing, so approved keeps whatever it held before the fan-out. If an empty cohort must count as approval, seed the field upstream or branch on the cohort size downstream rather than relying on the reducer.

Ordering in ReducerAppend and ReducerConcat

ReducerAppend (arrays) and ReducerConcat (strings) combine contributions in order. Within a task, that is the order of the writes. At a fan-in, it is branch order — the index of the element in the forEach array, or the declaration order of static branches — not the order in which the branches complete, so the result is deterministic. If you need another order — alphabetical, by score — sort downstream of the fan-in:

func (svc *Service) Decide(ctx context.Context, flow *workflow.Flow, failures []string) (decision string, err error) {
    if len(failures) > 0 {
        slices.Sort(failures) // a local copy, for display; sorting does not write the field
        return "rejected: " + strings.Join(failures, ", "), nil
    }
    return "approved", nil
}

ReducerAdd, ReducerMin, ReducerMax, ReducerAnd and ReducerOr give the same result in any order. ReducerUnion and ReducerMerge give the same set of elements or keys in any order, though ReducerUnion keeps the first occurrence’s position and ReducerMerge lets the later contribution win a colliding key.

See Also

• State - the broader treatment: how state splits at fan-out, forEach element aliasing with injected index and count, flow.Del / Clear 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; each branch returns its own count, never the running total.
• Workflows overview - conceptual background on tasks, graphs, and state.
• Package workflow - the Reducer type, SetReducer, and the Set / Overwrite / Del primitives.
• Package coreservices/foreman - the orchestration engine that folds contributions at every hand-off.