# Fan-out and join

> Fan a task out into one branch per item, optionally run the branches in parallel, and collect their typed outputs in one join node.

You end up with a task that runs one task node per item of a list, all at the same time, and collects every result in a single join node. The task types need the `swiftide-tasks` feature flag ([Tasks overview](/tasks/overview/#feature-flag)). The code comes from the workspace-architect showcase, which analyses each crate of a Cargo workspace with its own agent.

```rust title="examples/workspace-architect/src/graph.rs"
// One branch per crate, all evaluating the same analyst node with a different input.
task.register_transition(discover, move |members: WorkspaceMembers| {
    let mut crates = members.crates.into_iter();
    let Some(first) = crates.next() else {
        return Transition::error("the workspace has no crates to document");
    };

    crates
        .fold(Transition::fan_out(&analyse, first), |fan_out, member| {
            fan_out.and(&analyse, member)
        })
        .concurrency_model(ConcurrencyModel::Parallel)
        .join_with(join.join())
})?;
```

Source: [`examples/workspace-architect/src/graph.rs#L92-L105`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/workspace-architect/src/graph.rs#L92-L105)

```rust title="examples/workspace-architect/src/graph.rs"
// Each branch hands its `CrateSummary` to the join; the join runs once all have arrived.
task.register_transition(analyse, join.join())?;
task.register_transition(join, {
    let progress = progress.clone();
    move |prompt| {
        progress.step("render", "drafting ARCHITECTURE.md");
        render.transitions_with(prompt)
    }
})?;
```

Source: [`examples/workspace-architect/src/graph.rs#L109-L117`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/workspace-architect/src/graph.rs#L109-L117)

The `discover` node returns the workspace members, and its transition folds them into one fan-out with a branch per crate. Every branch evaluates the same `analyse` node with a different `WorkspaceCrate`, in parallel, and `join_with` names the node that waits for the whole group.

The `analyse` node registers one transition to the join target, `task.register_transition(analyse, join.join())`, and every branch uses it. Transitions belong to nodes, not branches, and each node has [exactly one](/tasks/nodes-and-transitions/#transitions). The join node is a closure over `&JoinInput<CrateSummary>`, registered with `task.register_node_fn` in the `nodes` region of `graph.rs`. It runs once every branch has delivered its `CrateSummary`, and its `Prompt` goes to the render node with an ordinary `transitions_with`. [Built-in nodes](/tasks/built-in-nodes/#prompt-nodes) covers the render node.

## Fan-out

`Transition::fan_out(&node, input)` starts a `FanOutTransition` with one branch. `.and(&node, input)` adds another. Branches can target the same node or different nodes. Each input has to convert into the target node's input type, or the code does not compile.

A `FanOutTransition` becomes a `Transition` only when you call `join_with`, so a fan-out without a join does not compile.

`fan_out` needs a first target, so handle an empty list first. The showcase returns `Transition::error(..)`, which fails the run with a `TaskError::NodeError`.

With branches known up front, write them out. This example starts two branches on two different nodes:

```rust
// The fan-out transition defines the branch group and the join that waits for that group.
task.register_transition(start, move |input| {
    Transition::fan_out(&increment, input)
        .and(&double, input)
        .join_with(join.join())
})?;

// Each branch still declares where its own output should go. Branches can do any amount of work
// before eventually transitioning to the join.
task.register_transition(increment, join.join())?;
task.register_transition(double, join.join())?;
task.register_transition(join, task.transitions_to_finish())?;
```

Source: [`examples/tasks_fanout.rs#L27-L38`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/tasks_fanout.rs#L27-L38)

A branch can pass through intermediate nodes before it joins. Only the last node on the branch registers the join target. If a branch joins a different node than the one its fan-out named, the run fails with `TaskError::InvalidState`.

## Join node

A join node takes `JoinInput<T>` as its input, where `T` is the payload every branch contributes. `NodeId::join()` exists only on nodes with that input type. `JoinInput` has three methods:

- `iter()` yields the payloads in the order the branches were created, not the order they finished.
- `len()` returns the number of payloads.
- `is_empty()` returns true when there are none.

The showcase relies on that order: the prompt lists the crates as `discover` found them, although `swiftide-core` finished last after a retry. A branch whose output does not convert into `T` does not compile.

## Payload mapping

A branch does not have to return the payload type itself. `JoinTarget::map` converts the branch output on its way into the join:

```rust
let mut task = Task::<String, String>::new();

let start = task.register_node_fn(|text: &String| -> Result<String, NodeError> {
    Ok(text.trim().to_string())
});
let count = task.register_node_fn(|text: &String| -> Result<usize, NodeError> {
    Ok(text.split_whitespace().count())
});
let length = task
    .register_node_fn(|text: &String| -> Result<usize, NodeError> { Ok(text.chars().count()) });
let shout = task
    .register_node_fn(|text: &String| -> Result<String, NodeError> { Ok(text.to_uppercase()) });
let join = task.register_node_fn(|parts: &JoinInput<String>| -> Result<String, NodeError> {
    Ok(parts.iter().cloned().collect::<Vec<_>>().join("; "))
});

task.starts_with(start);
task.register_transition(start, move |text: String| {
    Transition::fan_out(&count, text.clone())
        .and(&length, text.clone())
        .and(&shout, text)
        .join_with(join.join())
})?;

// `count` returns a usize; `map` turns it into the join's String payload.
task.register_transition(
    count,
    join.join().map(|words: usize| format!("{words} words")),
)?;
// An async mapping is registered with `register_transition_async`.
task.register_transition_async(
    length,
    join.join()
        .map_async(|chars: usize| async move { format!("{chars} characters") }),
)?;
// `shout` already returns a String, so it joins as is.
task.register_transition(shout, join.join())?;
task.register_transition(join, task.transitions_to_finish())?;

// "2 words; 11 characters; HELLO TASKS"
let TaskRunOutcome::Completed(summary) = task.run(" hello tasks ").await? else {
    anyhow::bail!("nothing in this task pauses");
};
```

Source: [`snippets/src/tasks/fan_out_and_join.rs#L8-L50`](https://github.com/bosun-ai/swiftide-website/blob/master/snippets/src/tasks/fan_out_and_join.rs#L8-L50)

`map` turns the `usize` from `count` into the `String` the join expects. `JoinTarget::map_async` does the same with an async closure and is registered with `Task::register_transition_async`. `shout` already returns a `String`. Any output that converts into the payload type with `Into` needs no mapping.

The task returns `"2 words; 11 characters; HELLO TASKS"`, in branch creation order.

## Mixed payloads

When branches return different types and the join needs all of them, give the join node an `AnyJoinInput`. Use `join_any()` in place of `join()`, both in the fan-out and for each branch:

```rust
let mut task = Task::<String, String>::new();

let start = task.register_node_fn(|text: &String| -> Result<String, NodeError> {
    Ok(text.trim().to_string())
});
let count = task.register_node_fn(|text: &String| -> Result<usize, NodeError> {
    Ok(text.split_whitespace().count())
});
let shout = task
    .register_node_fn(|text: &String| -> Result<String, NodeError> { Ok(text.to_uppercase()) });
let join = task.register_node_fn(|parts: &AnyJoinInput| -> Result<String, NodeError> {
    let words: usize = parts.iter::<usize>().sum();
    let shouted = parts.iter::<String>().cloned().collect::<Vec<_>>();
    Ok(format!("{} ({words} words)", shouted.join(" ")))
});

task.starts_with(start);
task.register_transition(start, move |text: String| {
    Transition::fan_out(&count, text.clone())
        .and(&shout, text)
        .join_with(join.join_any())
})?;
task.register_transition(count, join.join_any())?;
task.register_transition(shout, join.join_any())?;
task.register_transition(join, task.transitions_to_finish())?;

// "HELLO TASKS (2 words)"
let TaskRunOutcome::Completed(summary) = task.run(" hello tasks ").await? else {
    anyhow::bail!("nothing in this task pauses");
};
```

Source: [`snippets/src/tasks/fan_out_and_join.rs#L58-L87`](https://github.com/bosun-ai/swiftide-website/blob/master/snippets/src/tasks/fan_out_and_join.rs#L58-L87)

`AnyJoinInput::iter::<T>()` downcasts each payload and skips the ones of other types. `iter_any()` yields every payload as a type-erased value. The compiler no longer checks that a branch delivers a type the join reads, so prefer `JoinInput<T>` with `map` where you can.

## Concurrency model

`ConcurrencyModel` has two variants. `Sequential` is the default, and `Parallel` is opt-in. Set the default for the whole task with `with_default_concurrency_model`:

```rust
let mut task = Task::new().with_default_concurrency_model(ConcurrencyModel::Parallel);

let start =
    task.register_node_fn(|input: &i32| -> std::result::Result<i32, NodeError> { Ok(*input) });
let increment = task
    .register_node_fn(|input: &i32| -> std::result::Result<i32, NodeError> { Ok(*input + 1) });
let double = task
    .register_node_fn(|input: &i32| -> std::result::Result<i32, NodeError> { Ok(*input * 2) });
let join = task.register_node_fn(
    |input: &JoinInput<i32>| -> std::result::Result<i32, NodeError> {
        Ok(input.iter().copied().sum::<i32>())
    },
);

task.starts_with(start);
```

Source: [`examples/tasks_fanout.rs#L9-L23`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/tasks_fanout.rs#L9-L23)

To override the model from a given step on, call `concurrency_model` on a `Transition`, `MarkedTransition` or `FanOutTransition`. A branch keeps the override for every later step until another transition overrides it, and the branches of a fan-out inherit its model. The join node always runs with the task's default model.

Under `Sequential`, the runtime evaluates one node at a time. Fan-out branches take turns: after a node finishes, the branch's next step goes to the back of the queue.

Under `Parallel`, the branches run concurrently inside the future that `run` returns, as the [overview](/tasks/overview/#crate-contents) explains. Waiting on input/output (I/O), such as model calls, overlaps. Synchronous, compute-only closures do not.

This is the showcase's captured run, with `claude-sonnet-5-5` as the model for all ten analysts. The `model` and `discover` lines before it and the render steps after it are left out:

```text
[   0.0s] analyse  started  swiftide-core
[   0.0s] analyse  started  swiftide-langfuse
[   0.0s] analyse  started  swiftide-macros
[   0.0s] analyse  started  swiftide-indexing
[   0.0s] analyse  started  swiftide-integrations
[   0.0s] analyse  started  swiftide-query
[   0.0s] analyse  started  swiftide-tasks
[   0.0s] analyse  started  swiftide-agents
[   0.0s] analyse  started  swiftide-agents-contrib
[   0.0s] analyse  started  swiftide
[  14.0s] analyse  finished swiftide-langfuse in 14.0s (6 tool calls)
[  14.3s] analyse  finished swiftide-macros in 14.3s (6 tool calls)
[  14.7s] analyse  finished swiftide in 14.7s (5 tool calls)
[  19.2s] analyse  finished swiftide-query in 19.1s (10 tool calls)
[  19.7s] analyse  retry    swiftide-agents-contrib: the `stop` arguments do not match the schema: unknown field `arguments`, expected one of `purpose`, `key_types`, `depends_on`, `notes`
[  20.2s] analyse  finished swiftide-agents in 20.2s (10 tool calls)
[  24.4s] analyse  finished swiftide-indexing in 24.4s (10 tool calls)
[  24.9s] analyse  finished swiftide-tasks in 24.9s (10 tool calls)
[  27.1s] analyse  finished swiftide-agents-contrib in 27.1s (11 tool calls)
[  31.4s] analyse  finished swiftide-integrations in 31.4s (13 tool calls)
[  33.9s] analyse  retry    swiftide-core: the `stop` arguments do not match the schema: unknown field `arguments`, expected one of `purpose`, `key_types`, `depends_on`, `notes`
[  40.5s] analyse  finished swiftide-core in 40.5s (15 tool calls)
[  40.5s] join     10 summaries; 230s of agent work in 41s wall-clock (5.7x)
```

All ten branches start at 0.0s, and the fan-out takes as long as its slowest branch. The join line sums the agent time of every branch: 230 seconds of work in 41 seconds. `OUTPUT.md` estimates close to four minutes for the same analyses run one after another.

## Limits

- A branch cannot start a second fan-out while it belongs to an active join group. The run fails with `TaskError::InvalidState`. Fan out again from the join node instead.
- A branch cannot finish the task on its own. If a branch inside a group registers `task.transitions_to_finish()`, the task output counts as an arrival at the join, unmapped. When that type differs from the join's payload type, the join fails at runtime with `TaskError::InvalidState`. Route every branch to the join and finish from there.
- The first branch that fails fails the run. The other in-flight branches are dropped and never complete.
- All branches share one node instance; keep per-branch state inside `evaluate` ([Agents as task nodes](/tasks/agents-as-nodes/#custom-agent-node)).

## Reference

- [`Transition`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.Transition.html)
- [`Transition::error`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.Transition.html#method.error)
- [`FanOutTransition`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.FanOutTransition.html)
- [`MarkedTransition`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.MarkedTransition.html)
- [`JoinInput`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.JoinInput.html)
- [`JoinTarget`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.JoinTarget.html)
- [`MappedJoinTarget`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.MappedJoinTarget.html)
- [`AnyJoinInput`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.AnyJoinInput.html)
- [`AnyJoinTarget`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.AnyJoinTarget.html)
- [`ConcurrencyModel`](https://docs.rs/swiftide/0.34/swiftide/tasks/enum.ConcurrencyModel.html)
- [`NodeId`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.NodeId.html)
- [`NodeId::join_any`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.NodeId.html#method.join_any)
- [`Task`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.Task.html)
- [`Task::register_transition_async`](https://docs.rs/swiftide/0.34/swiftide/tasks/struct.Task.html#method.register_transition_async)
- [`TaskError`](https://docs.rs/swiftide/0.34/swiftide/tasks/enum.TaskError.html)

## Next

[Pause and resume](/tasks/pause-and-resume/)