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

README.md

@@ -30,6 +30,31 @@ Rust 2024 · MSRV 1.85 · `#![forbid(unsafe_code)]` · GPL-3.0-or-later.
 - Linear execution, conditional routing on output ports, **parallel fan-out**
   (concurrent successors sharing a port), and a **merge fan-in barrier** (a node
   runs only once all its predecessors finish).
+- **Per-item fan-out** — a single node multiplying an array of input into N
+  concurrent units of work, array in and array out. Where graph fan-out fixes
+  the width when the graph is authored, this width is data-driven:
+
+  ```jsonc
+  // one agent turn per topic, at most 8 at a time
+  { "kind": "agent", "config": {
+      "execution": "per_item",   // map over the input array
+      "concurrency": 8,          // 1 = sequential (default), n = bounded, 0/"all" = unbounded
+      "prompt": "Research =item.name"
+  } }
+
+  // ...or one whole child workflow per item — the multiplier
+  { "kind": "sub_workflow", "config": {
+      "execution": "per_item", "concurrency": 4, "workflow_id": "deep_dive"
+  } }
+  ```
+
+  Results always come back in **input order** with `paired_item` set, so a
+  fan-out never reorders data. `on_item_error` decides what a failing item does
+  to the batch — `collect` (the default when fanning out) marks that item
+  `{ error, failed: true }` and keeps the rest, `fail_fast` (the default when
+  sequential) hands the error to the node's `on_error` / retry policy, and
+  `skip` drops it. Supported on `agent`, `tool_call`, `http_request`, `memory`,
+  and `sub_workflow`.
 
 **Nodes**
 

src/catalog.rs

@@ -588,7 +588,79 @@ pub fn contract_for(kind: &str) -> Option<NodeKindContract> {
         },
         _ => return None,
     };
-    Some(c)
+    Some(with_fan_out_fields(c))
+}
+
+/// The node kinds that map over their input, and whether they do so by default.
+///
+/// `true` means the kind is `per_item` unless told otherwise, so its fan-out
+/// knobs apply without an explicit `execution`.
+const FAN_OUT_KINDS: [(&str, bool); 5] = [
+    ("agent", false),
+    ("tool_call", true),
+    ("http_request", true),
+    ("memory", true),
+    ("sub_workflow", false),
+];
+
+/// Appends the shared per-item fan-out contract (`execution`, `concurrency`,
+/// `on_item_error`) to the kinds that support it.
+///
+/// These three keys behave identically on every mapping kind, so they are
+/// described once here rather than copied into five contracts that would then
+/// drift. Kinds that cannot map over their input are returned untouched — and
+/// [`crate::validate`] rejects the keys there, so the contract and the validator
+/// agree on exactly which kinds fan out.
+fn with_fan_out_fields(mut c: NodeKindContract) -> NodeKindContract {
+    let Some((_, per_item_by_default)) = FAN_OUT_KINDS.iter().find(|(k, _)| *k == c.kind) else {
+        return c;
+    };
+    let default_mode = if *per_item_by_default {
+        "per_item"
+    } else {
+        "once"
+    };
+
+    c.config_fields.push(
+        ConfigField::optional(
+            "execution",
+            "enum",
+            &format!(
+                "Whether this node runs once for the whole input array or once per input item. \
+                 Defaults to \"{default_mode}\" for this kind."
+            ),
+        )
+        .with_enum(&["once", "per_item"]),
+    );
+    c.config_fields.push(ConfigField::optional(
+        "concurrency",
+        "integer | \"all\"",
+        "With execution \"per_item\", how many items run at a time: 1 (the default) is strictly \
+         sequential, n runs at most n at once, and 0 or \"all\" runs every item at once. This is \
+         the fan-out dial — use it to turn an array of work into parallel work. Ignored (and \
+         rejected by validation) unless the node runs per item.",
+    ));
+    c.config_fields.push(
+        ConfigField::optional(
+            "on_item_error",
+            "enum",
+            "What a failing item does to the batch. Defaults to \"collect\" when the node fans \
+             out (concurrency other than 1) and \"fail_fast\" when it runs sequentially. \
+             \"collect\" emits an error item — {json:{error,failed:true}} — in that item's slot so \
+             the node still returns one output per input and a downstream condition can branch on \
+             =item.json.failed. \"fail_fast\" fails the node on the first error in input order, \
+             handing it to the node's on_error/retry policy. \"skip\" drops failed items, so the \
+             output array may be shorter than the input.",
+        )
+        .with_enum(&["collect", "fail_fast", "skip"]),
+    );
+
+    c.notes.push(
+        "Output items are always returned in INPUT order with paired_item set, however the \
+         concurrency is set — a fan-out never reorders data."
+            .to_string(),
+    );
+    c
 }
 
 #[cfg(test)]
@@ -727,3 +799,62 @@ mod tests {
         }
     }
 }
+
+#[cfg(test)]
+mod fan_out_contract_tests {
+    use super::*;
+
+    #[test]
+    fn every_mapping_kind_advertises_the_fan_out_knobs() {
+        for (kind, _) in FAN_OUT_KINDS {
+            let c = contract_for(kind).expect("contract");
+            for field in ["execution", "concurrency", "on_item_error"] {
+                assert!(
+                    c.config_fields.iter().any(|f| f.name == field),
+                    "{kind} should advertise `{field}`"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn kinds_that_cannot_map_do_not_advertise_them() {
+        // The contract and the validator must agree on which kinds fan out;
+        // advertising a key that validation rejects would be worse than silence.
+        for kind in [
+            "trigger",
+            "condition",
+            "switch",
+            "merge",
+            "transform",
+            "code",
+        ] {
+            let c = contract_for(kind).expect("contract");
+            assert!(
+                !c.config_fields.iter().any(|f| f.name == "concurrency"),
+                "{kind} must not advertise `concurrency`"
+            );
+        }
+    }
+
+    #[test]
+    fn the_execution_default_is_stated_per_kind() {
+        let doc = |kind: &str| {
+            contract_for(kind)
+                .expect("contract")
+                .config_fields
+                .iter()
+                .find(|f| f.name == "execution")
+                .expect("execution field")
+                .description
+                .clone()
+        };
+        // An author needs to know that `agent` must opt in but `tool_call` need not.
+        assert!(doc("agent").contains("\"once\""), "{}", doc("agent"));
+        assert!(
+            doc("tool_call").contains("\"per_item\""),
+            "{}",
+            doc("tool_call")
+        );
+    }
+}

src/nodes/integration/agent.rs

@@ -59,15 +59,21 @@ impl NodeExecutor for AgentNode {
                 && !ctx.input.is_empty();
 
         if per_item {
-            let mut items = Vec::with_capacity(ctx.input.len());
-            let mut diagnostics = Vec::new();
-            for (index, input_item) in ctx.input.iter().enumerate() {
-                let (cfg, diags) =
-                    crate::nodes::resolve_config_traced_for_item(&ctx, input_item.json.clone());
-                let item = run_turn(&ctx, &cfg).await?;
-                items.push(item.paired_with(index));
-                diagnostics.extend(diags);
-            }
+            // Fan out: `config.concurrency` decides how many turns run at once
+            // (default 1 — sequential, as this node has always behaved), and
+            // `config.on_item_error` what a failing turn does to the batch.
+            let opts = crate::nodes::map::map_options(&ctx.node.config, &ctx.node.id);
+            let ctx = &ctx;
+            let (items, diagnostics) =
+                crate::nodes::map::map_items(ctx.input.len(), opts, move |index| async move {
+                    let (cfg, diags) = crate::nodes::resolve_config_traced_for_item(
+                        ctx,
+                        ctx.input[index].json.clone(),
+                    );
+                    let item = run_turn(ctx, &cfg).await?;
+                    Ok((item, diags))
+                })
+                .await?;
             return Ok(NodeOutput::main(items).with_diagnostics(diagnostics));
         }
 

src/nodes/integration/http_request.rs

@@ -34,15 +34,20 @@ impl NodeExecutor for HttpRequestNode {
             && !ctx.input.is_empty();
 
         if per_item {
-            let mut items = Vec::with_capacity(ctx.input.len());
-            let mut diagnostics = Vec::new();
-            for (index, input_item) in ctx.input.iter().enumerate() {
-                let (cfg, diags) =
-                    crate::nodes::resolve_config_traced_for_item(&ctx, input_item.json.clone());
-                let response = request(&ctx, &cfg).await?;
-                items.push(Item::new(envelope::wrap(response)).paired_with(index));
-                diagnostics.extend(diags);
-            }
+            // `config.concurrency` decides how many requests are in flight at
+            // once (default 1 — sequential, as before).
+            let opts = crate::nodes::map::map_options(&ctx.node.config, &ctx.node.id);
+            let ctx = &ctx;
+            let (items, diagnostics) =
+                crate::nodes::map::map_items(ctx.input.len(), opts, move |index| async move {
+                    let (cfg, diags) = crate::nodes::resolve_config_traced_for_item(
+                        ctx,
+                        ctx.input[index].json.clone(),
+                    );
+                    let response = request(ctx, &cfg).await?;
+                    Ok((Item::new(envelope::wrap(response)), diags))
+                })
+                .await?;
             Ok(NodeOutput::main(items).with_diagnostics(diagnostics))
         } else {
             let (cfg, diagnostics) = crate::nodes::resolve_config_traced(&ctx);

src/nodes/integration/memory.rs

@@ -224,15 +224,20 @@ impl NodeExecutor for MemoryNode {
         );
 
         if per_item {
-            let mut items = Vec::with_capacity(ctx.input.len());
-            let mut diagnostics = Vec::new();
-            for (index, input_item) in ctx.input.iter().enumerate() {
-                let (cfg, diags) =
-                    crate::nodes::resolve_config_traced_for_item(&ctx, input_item.json.clone());
-                let result = call_provider(&ctx, &cfg).await?;
-                items.push(Item::new(envelope::wrap(result)).paired_with(index));
-                diagnostics.extend(diags);
-            }
+            // `config.concurrency` decides how many provider calls are in flight
+            // at once (default 1 — sequential, as before).
+            let opts = crate::nodes::map::map_options(&ctx.node.config, &ctx.node.id);
+            let ctx = &ctx;
+            let (items, diagnostics) =
+                crate::nodes::map::map_items(ctx.input.len(), opts, move |index| async move {
+                    let (cfg, diags) = crate::nodes::resolve_config_traced_for_item(
+                        ctx,
+                        ctx.input[index].json.clone(),
+                    );
+                    let result = call_provider(ctx, &cfg).await?;
+                    Ok((Item::new(envelope::wrap(result)), diags))
+                })
+                .await?;
             tracing::debug!(
                 node = %ctx.node.id,
                 emitted = items.len(),