feat(nodes): per-item fan-out — run an array of work concurrently

[senamakel]

What

Lets a single node multiply an array of input into N concurrent units of work — array in, array out. Where the existing graph fan-out fixes its width when the graph is authored (you hand-write N sibling nodes), this width is data-driven.

// N parallel agent turns, at most 8 in flight
{ "kind": "agent", "config": {
    "execution": "per_item",
    "concurrency": 8,
    "agent_ref": "researcher",
    "prompt": "Research =item.name"
} }

// ...or N parallel runs of a whole child workflow — the multiplier
{ "kind": "sub_workflow", "config": {
    "execution": "per_item", "concurrency": 4, "workflow_id": "deep_dive"
} }

execution: per_item already existed on agent / tool_call / http_request / memory, but every implementation was a sequential for … .await loop. This adds the concurrency dial and applies it uniformly, plus a new per_item mode on sub_workflow.

Key	Values	Meaning
execution	once | per_item	Map over the input array, or not. Unchanged per-kind defaults.
concurrency	integer | "all"	1 (default) sequential, n bounded, 0/"all" unbounded. Clamped to 64.
on_item_error	collect | fail_fast | skip	What a failing item does to the batch.

Results always come back in input order with paired_item set, so a fan-out never reorders data.

Design notes

concurrency defaults to 1, so nothing existing changes. Fan-out is strictly opt-in; an e2e test pins that execution: per_item alone stays sequential, so a future regression that makes parallelism the default fails loudly.

on_item_error's default follows the execution shape — collect when fanning out, fail_fast when sequential. This is the one subtle decision in the PR. Collecting unconditionally looked reasonable, but tool_call, http_request, and memory are per_item by default, so it would have silently disabled on_error, retry, and the error port for the most ordinary nodes in the engine — a graph that never asked for a fan-out would quietly stop failing. It was caught by the existing on_error suite going red. Opting into concurrency is opting into batch semantics; an explicit on_item_error overrides either way.

A fan-out widens a run without deepening it. sub_workflow's MAX_SUB_WORKFLOW_DEPTH guard is per child run, so N siblings sit at depth d+1, never d+N.

No new dependency. futures-util and tokio were already deps, so the bounded concurrency is buffer_unordered with index-carrying futures re-slotted into input order. tinyagents::graph::parallel::map_reduce was the alternative but would have dragged TinyAgentsError conversion across the seam for no gain.

Validation rejects a no-op knob. concurrency on a node that runs once is an error, not a silent ignore — otherwise an author asks for parallelism and gets none with no signal at all.

Surface

• src/nodes/map.rs — the one new primitive (map_options, map_items).
• src/nodes/integration/{agent,tool_call,http_request,memory}.rs — the four sequential loops now call it.
• src/nodes/integration/sub_workflow.rs — new per_item mode; body extracted to run_child.
• src/validate.rs, src/catalog.rs — author-time rejection and machine-readable contracts (the catalog is what an authoring agent reads, so a feature absent from it is unreachable).

Validation

• cargo test — 449 lib tests + all integration suites green
• cargo clippy --all-targets --all-features -- -D warnings — clean
• cargo fmt --check — clean

tests/per_item_fanout_e2e.rs covers the properties that make the feature usable rather than merely present: a probe records peak in-flight concurrency (so a regression to a sequential loop fails), input-order results, the sequential default, and both item-error policies.

Per-item observability

RunObserver gains on_item_start / on_item_finish, both defaulting to no-ops so existing observers are untouched.

A fanned-out node is one step running N units of work, so the step callbacks fire once for the whole node and a host can only render it as one long step. These report the individual items, keyed on the input index so a host can match each to the output item that lands in that slot, and carrying the batch size so it need keep no per-node map of its own.

The calls live inside the item future rather than where the stream is built: buffer_unordered only polls up to concurrency at a time, so announcing eagerly would show every worker as live the moment the batch began. A test pins that started-but-unfinished never exceeds the bound.

Threading the observer to the nodes meant adding a field to NodeContext (~50 construction sites, mechanical — executor unit tests pass NoopObserver).

[greptile-apps]

senamakel has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.

[coderabbitai]

Warning

Review limit reached

You’ve reached a temporary PR review limit under our Fair Usage Limits Policy.

Your recent review volume is higher than typical usage, so adaptive limits are currently applied.

Next review available in: 8 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 002151e5-87d7-46f8-995b-66e3c19ebb46

📥 Commits

Reviewing files that changed from the base of the PR and between ead2615 and 4817cd5.

📒 Files selected for processing (12)

• README.md
• src/catalog.rs
• src/nodes/integration/agent.rs
• src/nodes/integration/http_request.rs
• src/nodes/integration/memory.rs
• src/nodes/integration/sub_workflow.rs
• src/nodes/integration/tool_call.rs
• src/nodes/map.rs
• src/nodes/mod.rs
• src/validate.rs
• tests/per_item_fanout_e2e.rs
• wiki/Node-Catalog.md

---

_{Comment @coderabbitai help to get the list of available commands.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 4817cd51b6

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Keep approval failures outside item collection

When a per-item sub-workflow uses fan-out concurrency, map_options defaults to collect, so an error returned by run_child because a child paused at a requires_approval gate is converted into a normal error item. The parent then reports success and schedules downstream nodes, bypassing the existing fail-closed behavior that deliberately halts the parent when cross-boundary approval resume is unsupported. Lifecycle errors for pending approvals must bypass item collection and remain node failures.

Useful? React with 👍 / 👎.