Fan-out and join

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). The code comes from the workspace-architect showcase, which analyses each crate of a Cargo workspace with its own agent.

// 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

// 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

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. 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 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:

// 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

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:

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

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:

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

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:

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

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 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:

[   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).

Reference

Pause and resume