From 413944284628f7a256f8d4712573e781770da325 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:23:02 +0300 Subject: [PATCH 1/7] feat(nodes): add bounded-concurrency fan-out helper for per-item execution --- src/nodes/map.rs | 551 +++++++++++++++++++++++++++++++++++++++++++++++ src/nodes/mod.rs | 6 + 2 files changed, 557 insertions(+) create mode 100644 src/nodes/map.rs diff --git a/src/nodes/map.rs b/src/nodes/map.rs new file mode 100644 index 0000000..33fe991 --- /dev/null +++ b/src/nodes/map.rs @@ -0,0 +1,551 @@ +//! Bounded-concurrency fan-out: mapping a node's work over its input items. +//! +//! A node in [`ExecutionMode::PerItem`](super::ExecutionMode::PerItem) runs its +//! body once per input item. *How many* of those run at a time is the dial this +//! module owns: +//! +//! | `config.concurrency` | Behaviour | +//! |---|---| +//! | unset (or `1`) | strictly sequential — one item at a time, in input order | +//! | `n > 1` | at most `n` items in flight | +//! | `0` or `"all"` | every item in flight at once | +//! +//! Results are always returned in **input order** regardless of completion +//! order, and each output item carries `paired_item` so a downstream node can +//! correlate it back to the input that produced it. That is what makes a +//! fan-out node array-in/array-out: `split_out` → `agent(concurrency: 8)` → +//! `merge` behaves like a bounded `Promise.all`. +//! +//! ## Failure +//! +//! [`ItemErrorPolicy`] decides what a failing item does to the batch. The +//! default is [`Collect`](ItemErrorPolicy::Collect): the batch never fails, and +//! the failed slot is filled with an error item so downstream nodes still see +//! one output per input. See that type for the other two policies. + +use std::future::Future; + +use futures_util::stream::StreamExt; +use serde_json::{Value, json}; + +use crate::data::Item; +use crate::error::Result; +use crate::expr::NullResolution; + +/// The largest `concurrency` a node may request. +/// +/// A graph is authored data — often by a model — so an absurd `concurrency` +/// (or a typo like `10000`) must not be able to open ten thousand simultaneous +/// agent turns against a host. Requests above this are clamped, not rejected, +/// so a workflow still runs; the clamp is `tracing::warn!`ed. Hosts layer their +/// own, usually lower, ceiling on top (e.g. a semaphore around agent runs). +pub(crate) const MAX_CONCURRENCY: usize = 64; + +/// What a failing item does to the rest of the batch. +/// +/// Read from `config.on_item_error`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub(crate) enum ItemErrorPolicy { + /// **Default.** The batch never fails: a failed item is replaced by an + /// error item (`{ json: { error, failed: true }, … }`) in its own slot, so + /// the node always emits exactly one item per input and a downstream node + /// can branch on `=item.json.failed`. + /// + /// Note this is deliberately *more* forgiving than a bare sequential loop, + /// which propagated the first error and failed the node. Graphs that want + /// the old behaviour set [`FailFast`](ItemErrorPolicy::FailFast). + #[default] + Collect, + /// The first failure **in input order** fails the whole node, which then + /// falls to the node's own `on_error` / retry policy. Remaining in-flight + /// items are cancelled once no earlier item can still fail. + FailFast, + /// Failed items are dropped: the node emits only the successes, so the + /// output array may be shorter than the input. + Skip, +} + +/// How a per-item node maps over its input: how many at a time, and what a +/// failure does. +#[derive(Debug, Clone, Copy)] +pub(crate) struct MapOptions { + /// Maximum items in flight; `0` means unbounded. + pub concurrency: usize, + /// What a failing item does to the batch. + pub on_item_error: ItemErrorPolicy, +} + +impl Default for MapOptions { + /// Sequential and collecting — the back-compatible defaults. + fn default() -> Self { + Self { + concurrency: 1, + on_item_error: ItemErrorPolicy::Collect, + } + } +} + +/// Reads `concurrency` and `on_item_error` off a node's **raw** config. +/// +/// Like [`execution_mode`](super::execution_mode), these select the execution +/// strategy itself rather than describing data, so they are read before (and +/// independently of) `=`-expression resolution — a concurrency bound that +/// depended on the current item would be meaningless, since the bound applies +/// to the batch. +/// +/// Unrecognized values fall back to the defaults rather than erroring; +/// [`crate::validate`] rejects them at author time, where the message can point +/// at the offending node. +#[must_use] +pub(crate) fn map_options(config: &Value, node_id: &str) -> MapOptions { + let concurrency = match config.get("concurrency") { + Some(Value::Number(n)) => n.as_u64().map_or(1, |n| usize::try_from(n).unwrap_or(usize::MAX)), + // `"all"` is the readable spelling of "no bound" — `Promise.all`. + Some(Value::String(s)) if s == "all" => 0, + _ => 1, + }; + let concurrency = if concurrency > MAX_CONCURRENCY { + tracing::warn!( + node = %node_id, + requested = concurrency, + max = MAX_CONCURRENCY, + "concurrency above the engine ceiling; clamping" + ); + MAX_CONCURRENCY + } else { + concurrency + }; + + let on_item_error = match config.get("on_item_error").and_then(Value::as_str) { + Some("fail_fast") => ItemErrorPolicy::FailFast, + Some("skip") => ItemErrorPolicy::Skip, + _ => ItemErrorPolicy::Collect, + }; + + MapOptions { + concurrency, + on_item_error, + } +} + +/// What one mapped item produced: its output item plus any null-resolution +/// diagnostics gathered while resolving that item's config. +pub(crate) type MappedItem = (Item, Vec); + +/// The error item substituted for a failed slot under +/// [`ItemErrorPolicy::Collect`]. +/// +/// Shaped as the standard capability +/// [envelope](crate::nodes::integration::envelope) so the accessors a graph +/// already uses keep working: `=item.json.failed` is the branch predicate and +/// `=item.json.error` the message, on every node kind that can fan out. +fn error_item(message: &str) -> Item { + Item::new(crate::nodes::integration::envelope::from_parts( + json!({ "error": message, "failed": true }), + Some(message.to_string()), + Value::Null, + )) +} + +/// Runs `f` over `input` with bounded concurrency, returning the output items +/// in **input order** with `paired_item` set, plus the union of every item's +/// diagnostics. +/// +/// `f` receives each item's input index and the item itself. Items complete out +/// of order; the results are re-sorted into input-order slots before returning, +/// so a fan-out never reorders a workflow's data. +/// +/// # Errors +/// +/// Only under [`ItemErrorPolicy::FailFast`], which returns the failure with the +/// **lowest input index** — not the first to complete, which would make the +/// error non-deterministic across runs. The other two policies never error. +pub(crate) async fn map_items<'a, F, Fut>( + input: &'a [Item], + opts: MapOptions, + f: F, +) -> Result<(Vec, Vec)> +where + F: Fn(usize, &'a Item) -> Fut, + Fut: Future> + 'a, +{ + let total = input.len(); + // `buffer_unordered(0)` would never poll anything, so "unbounded" is spelled + // as "as many as there are items". + let in_flight = if opts.concurrency == 0 { + total.max(1) + } else { + opts.concurrency + }; + + // Each future carries its input index so completions can be re-slotted. + let mut stream = futures_util::stream::iter(input.iter().enumerate().map(|(index, item)| { + let fut = f(index, item); + async move { (index, fut.await) } + })) + .buffer_unordered(in_flight); + + let mut slots: Vec>> = + (0..total).map(|_| None).collect(); + // Under `FailFast` the reported error must be the lowest-index one, but + // items finish out of order. Track the lowest index seen so far; once every + // *earlier* slot has also resolved, no smaller index can still fail, so that + // error is final and dropping the stream cancels the rest. + let mut fail_fast_error: Option<(usize, crate::error::EngineError)> = None; + + while let Some((index, result)) = stream.next().await { + match result { + Ok(mapped) => slots[index] = Some(Ok(mapped)), + Err(err) => { + if opts.on_item_error == ItemErrorPolicy::FailFast { + // Mark the slot resolved so the prefix check below can see it. + slots[index] = Some(Err(String::new())); + if fail_fast_error + .as_ref() + .is_none_or(|(seen, _)| index < *seen) + { + fail_fast_error = Some((index, err)); + } + let lowest = fail_fast_error.as_ref().map_or(index, |(i, _)| *i); + if slots[..lowest].iter().all(Option::is_some) { + break; // dropping the stream cancels the remaining work + } + continue; + } + slots[index] = Some(Err(err.to_string())); + } + } + } + + if let Some((_, err)) = fail_fast_error { + return Err(err); + } + + let mut items = Vec::with_capacity(total); + let mut diagnostics = Vec::new(); + for (index, slot) in slots.into_iter().enumerate() { + match slot { + Some(Ok((item, diags))) => { + items.push(item.paired_with(index)); + diagnostics.extend(diags); + } + Some(Err(message)) if opts.on_item_error == ItemErrorPolicy::Collect => { + items.push(error_item(&message).paired_with(index)); + } + // `Skip` drops the slot; `FailFast` already returned above. A `None` + // slot is only reachable on the cancelled tail of a fail-fast break. + _ => {} + } + } + + Ok((items, diagnostics)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + fn items(n: usize) -> Vec { + (0..n).map(|i| Item::new(json!({ "i": i }))).collect() + } + + fn opts(concurrency: usize, on_item_error: ItemErrorPolicy) -> MapOptions { + MapOptions { + concurrency, + on_item_error, + } + } + + /// Tracks how many mapped bodies are in flight simultaneously, so a test can + /// assert the bound was actually honoured (and that a "parallel" mode really + /// did run things at once). + #[derive(Default)] + struct Gauge { + live: AtomicUsize, + peak: AtomicUsize, + } + + impl Gauge { + fn enter(&self) { + let live = self.live.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(live, Ordering::SeqCst); + } + fn exit(&self) { + self.live.fetch_sub(1, Ordering::SeqCst); + } + fn peak(&self) -> usize { + self.peak.load(Ordering::SeqCst) + } + } + + /// A mapped body that stays "in flight" long enough for its peers to start, + /// so the gauge can observe real overlap rather than a lucky interleaving. + async fn tick() { + for _ in 0..8 { + tokio::task::yield_now().await; + } + } + + #[tokio::test] + async fn results_keep_input_order_even_when_completion_order_is_reversed() { + let input = items(5); + // Later items finish first: item 0 yields the most, item 4 the least. + let (out, _) = map_items(&input, opts(0, ItemErrorPolicy::Collect), |index, item| { + let json = item.json.clone(); + async move { + for _ in 0..(5 - index) * 4 { + tokio::task::yield_now().await; + } + Ok((Item::new(json), vec![])) + } + }) + .await + .expect("map"); + + assert_eq!(out.len(), 5); + for (index, item) in out.iter().enumerate() { + assert_eq!(item.json["i"], index, "output must be in input order"); + assert_eq!(item.paired_item, Some(index), "pairing tracks the input"); + } + } + + #[tokio::test] + async fn concurrency_one_is_strictly_sequential() { + let input = items(6); + let gauge = Arc::new(Gauge::default()); + let g = gauge.clone(); + let (out, _) = map_items(&input, opts(1, ItemErrorPolicy::Collect), move |_, item| { + let g = g.clone(); + let json = item.json.clone(); + async move { + g.enter(); + tick().await; + g.exit(); + Ok((Item::new(json), vec![])) + } + }) + .await + .expect("map"); + + assert_eq!(out.len(), 6); + assert_eq!(gauge.peak(), 1, "unset/1 concurrency must not overlap work"); + } + + #[tokio::test] + async fn bounded_concurrency_overlaps_but_respects_the_ceiling() { + let input = items(12); + let gauge = Arc::new(Gauge::default()); + let g = gauge.clone(); + let (out, _) = map_items(&input, opts(4, ItemErrorPolicy::Collect), move |_, item| { + let g = g.clone(); + let json = item.json.clone(); + async move { + g.enter(); + tick().await; + g.exit(); + Ok((Item::new(json), vec![])) + } + }) + .await + .expect("map"); + + assert_eq!(out.len(), 12); + assert!(gauge.peak() > 1, "bounded fan-out must actually overlap"); + assert!( + gauge.peak() <= 4, + "never more than `concurrency` in flight, saw {}", + gauge.peak() + ); + } + + #[tokio::test] + async fn zero_concurrency_runs_every_item_at_once() { + let input = items(7); + let gauge = Arc::new(Gauge::default()); + let g = gauge.clone(); + let (out, _) = map_items(&input, opts(0, ItemErrorPolicy::Collect), move |_, item| { + let g = g.clone(); + let json = item.json.clone(); + async move { + g.enter(); + tick().await; + g.exit(); + Ok((Item::new(json), vec![])) + } + }) + .await + .expect("map"); + + assert_eq!(out.len(), 7); + assert_eq!(gauge.peak(), 7, "`0`/`\"all\"` means unbounded"); + } + + #[tokio::test] + async fn collect_substitutes_an_error_item_and_keeps_the_array_length() { + let input = items(4); + let (out, _) = map_items(&input, opts(0, ItemErrorPolicy::Collect), |index, item| { + let json = item.json.clone(); + async move { + if index == 2 { + return Err(crate::error::EngineError::Capability("boom".into())); + } + Ok((Item::new(json), vec![])) + } + }) + .await + .expect("collect never fails the batch"); + + assert_eq!(out.len(), 4, "one output per input"); + assert_eq!(out[2].json["json"]["failed"], true); + assert!( + out[2].json["json"]["error"] + .as_str() + .expect("error message") + .contains("boom") + ); + assert_eq!(out[2].paired_item, Some(2)); + // Its neighbours are untouched successes. + assert_eq!(out[1].json["i"], 1); + assert_eq!(out[3].json["i"], 3); + } + + #[tokio::test] + async fn skip_drops_failures_and_shortens_the_array() { + let input = items(4); + let (out, _) = map_items(&input, opts(0, ItemErrorPolicy::Skip), |index, item| { + let json = item.json.clone(); + async move { + if index % 2 == 0 { + return Err(crate::error::EngineError::Capability("nope".into())); + } + Ok((Item::new(json), vec![])) + } + }) + .await + .expect("skip never fails the batch"); + + assert_eq!(out.len(), 2); + assert_eq!(out[0].json["i"], 1); + assert_eq!(out[1].json["i"], 3); + // Pairing still points at the original input index, not the compacted one. + assert_eq!(out[0].paired_item, Some(1)); + assert_eq!(out[1].paired_item, Some(3)); + } + + #[tokio::test] + async fn fail_fast_reports_the_lowest_index_error_not_the_first_to_finish() { + let input = items(6); + // Item 4 fails immediately; item 1 fails only after yielding. Input order + // must win, so the reported error is item 1's. + let err = map_items(&input, opts(0, ItemErrorPolicy::FailFast), |index, _| async move { + if index == 4 { + return Err(crate::error::EngineError::Capability("late-index".into())); + } + if index == 1 { + tick().await; + return Err(crate::error::EngineError::Capability("early-index".into())); + } + Ok((Item::new(json!({ "i": index })), vec![])) + }) + .await + .expect_err("fail_fast must surface an error"); + + assert!( + err.to_string().contains("early-index"), + "expected the lowest-index failure, got {err}" + ); + } + + #[tokio::test] + async fn empty_input_yields_no_items_and_no_error() { + let input: Vec = vec![]; + let (out, diags) = map_items(&input, opts(0, ItemErrorPolicy::Collect), |_, _| async { + unreachable!("no items to map") + }) + .await + .expect("map"); + assert!(out.is_empty()); + assert!(diags.is_empty()); + } + + #[tokio::test] + async fn diagnostics_from_every_item_are_unioned() { + let input = items(3); + let (_, diags) = map_items(&input, opts(0, ItemErrorPolicy::Collect), |index, _| async move { + Ok(( + Item::new(Value::Null), + vec![NullResolution { + location: format!("config.prompt[{index}]"), + expression: "=item.missing".to_string(), + }], + )) + }) + .await + .expect("map"); + assert_eq!(diags.len(), 3); + } + + // --- map_options --- + + #[test] + fn options_default_to_sequential_and_collect() { + let o = map_options(&json!({}), "n"); + assert_eq!(o.concurrency, 1, "unset concurrency stays sequential"); + assert_eq!(o.on_item_error, ItemErrorPolicy::Collect); + } + + #[test] + fn options_read_numeric_and_all_concurrency() { + assert_eq!(map_options(&json!({ "concurrency": 8 }), "n").concurrency, 8); + assert_eq!(map_options(&json!({ "concurrency": 0 }), "n").concurrency, 0); + assert_eq!( + map_options(&json!({ "concurrency": "all" }), "n").concurrency, + 0, + "`\"all\"` is the readable spelling of unbounded" + ); + } + + #[test] + fn options_clamp_an_absurd_concurrency_instead_of_failing_the_run() { + let o = map_options(&json!({ "concurrency": 10_000 }), "n"); + assert_eq!(o.concurrency, MAX_CONCURRENCY); + } + + #[test] + fn options_ignore_a_nonsense_concurrency_and_stay_sequential() { + // `validate` rejects these at author time; at run time they must not + // silently become unbounded. + assert_eq!( + map_options(&json!({ "concurrency": "lots" }), "n").concurrency, + 1 + ); + assert_eq!(map_options(&json!({ "concurrency": -3 }), "n").concurrency, 1); + assert_eq!( + map_options(&json!({ "concurrency": true }), "n").concurrency, + 1 + ); + } + + #[test] + fn options_read_every_item_error_policy() { + assert_eq!( + map_options(&json!({ "on_item_error": "fail_fast" }), "n").on_item_error, + ItemErrorPolicy::FailFast + ); + assert_eq!( + map_options(&json!({ "on_item_error": "skip" }), "n").on_item_error, + ItemErrorPolicy::Skip + ); + assert_eq!( + map_options(&json!({ "on_item_error": "collect" }), "n").on_item_error, + ItemErrorPolicy::Collect + ); + assert_eq!( + map_options(&json!({ "on_item_error": "bogus" }), "n").on_item_error, + ItemErrorPolicy::Collect, + "unknown policies fall back to the default" + ); + } +} diff --git a/src/nodes/mod.rs b/src/nodes/mod.rs index 0fb30cb..02c75a0 100644 --- a/src/nodes/mod.rs +++ b/src/nodes/mod.rs @@ -8,6 +8,7 @@ pub mod control_flow; pub mod integration; +pub(crate) mod map; use async_trait::async_trait; use serde_json::Value; @@ -91,6 +92,11 @@ pub(crate) fn expr_scope_for(ctx: &NodeContext, item: Value) -> Value { /// `paired_item`). This is the n8n-style default for `tool_call` / /// `http_request`, so a fan-out (`split_out` → node) actually runs per element /// instead of silently dropping all but the first. +/// +/// `PerItem` says *that* the node maps over its input; [`map`] says **how many +/// items run at a time** (`config.concurrency`) and what a failing item does to +/// the batch (`config.on_item_error`). Concurrency defaults to `1`, so a node +/// that does not opt in keeps the sequential ordering and timing it always had. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum ExecutionMode { /// Single invocation against the first item. From 200e6bf92e372cfede42594e97ae1f3ee45e9b38 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:28:12 +0300 Subject: [PATCH 2/7] feat(nodes): run per-item agent/tool_call/http_request/memory work concurrently `config.concurrency` bounds how many items a per-item node runs at once (unset/1 = sequential as before, n = bounded, 0/"all" = unbounded), and `config.on_item_error` decides what a failing item does to the batch. The policy default follows the execution shape: a fan-out collects (one bad item must not discard the batch) while a sequential run keeps failing fast, because tool_call/http_request/memory are per_item by default and collecting there would silently disable on_error, retry, and the error port for the most ordinary nodes in the engine. --- src/nodes/integration/agent.rs | 24 +- src/nodes/integration/http_request.rs | 23 +- src/nodes/integration/memory.rs | 23 +- src/nodes/integration/tool_call.rs | 23 +- src/nodes/map.rs | 342 +++++++++++++++++--------- 5 files changed, 281 insertions(+), 154 deletions(-) diff --git a/src/nodes/integration/agent.rs b/src/nodes/integration/agent.rs index d1f2974..860207e 100644 --- a/src/nodes/integration/agent.rs +++ b/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)); } diff --git a/src/nodes/integration/http_request.rs b/src/nodes/integration/http_request.rs index 63c8ce8..2096a17 100644 --- a/src/nodes/integration/http_request.rs +++ b/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); diff --git a/src/nodes/integration/memory.rs b/src/nodes/integration/memory.rs index 06511f8..63f35e7 100644 --- a/src/nodes/integration/memory.rs +++ b/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(), diff --git a/src/nodes/integration/tool_call.rs b/src/nodes/integration/tool_call.rs index 183c4af..a6d4258 100644 --- a/src/nodes/integration/tool_call.rs +++ b/src/nodes/integration/tool_call.rs @@ -43,15 +43,20 @@ impl NodeExecutor for ToolCallNode { if per_item { // Map over the input: re-resolve config against each item (so // `=item.x` binds to the current item) and invoke once 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 = invoke(&ctx, &cfg).await?; - items.push(Item::new(envelope::wrap(result)).paired_with(index)); - diagnostics.extend(diags); - } + // `config.concurrency` decides how many of those invocations 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 = invoke(ctx, &cfg).await?; + Ok((Item::new(envelope::wrap(result)), diags)) + }) + .await?; Ok(NodeOutput::main(items).with_diagnostics(diagnostics)) } else { // Single invocation against the first-item scope (or empty input). diff --git a/src/nodes/map.rs b/src/nodes/map.rs index 33fe991..9379e02 100644 --- a/src/nodes/map.rs +++ b/src/nodes/map.rs @@ -18,10 +18,24 @@ //! //! ## Failure //! -//! [`ItemErrorPolicy`] decides what a failing item does to the batch. The -//! default is [`Collect`](ItemErrorPolicy::Collect): the batch never fails, and -//! the failed slot is filled with an error item so downstream nodes still see -//! one output per input. See that type for the other two policies. +//! [`ItemErrorPolicy`] (`config.on_item_error`) decides what a failing item +//! does to the batch, and **its default follows the execution shape**: +//! +//! - **fanned out** (`concurrency` other than `1`) → +//! [`Collect`](ItemErrorPolicy::Collect). One bad item must not discard the +//! whole batch, so the failed slot is filled with an error item and the node +//! still emits one output per input. +//! - **sequential** (`concurrency` unset or `1`) → +//! [`FailFast`](ItemErrorPolicy::FailFast), exactly how per-item nodes +//! behaved before fan-out existed, so the node's `on_error` / retry policy +//! still sees the error. +//! +//! That split is deliberate. `tool_call`, `http_request`, and `memory` are +//! `per_item` *by default*, so collecting unconditionally would silently +//! disable `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. Opting into concurrency is also opting into batch semantics; an +//! explicit `on_item_error` overrides the default in either direction. use std::future::Future; @@ -44,21 +58,30 @@ pub(crate) const MAX_CONCURRENCY: usize = 64; /// What a failing item does to the rest of the batch. /// /// Read from `config.on_item_error`. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum ItemErrorPolicy { - /// **Default.** The batch never fails: a failed item is replaced by an - /// error item (`{ json: { error, failed: true }, … }`) in its own slot, so - /// the node always emits exactly one item per input and a downstream node - /// can branch on `=item.json.failed`. + /// **The default when the node fans out** (`concurrency` other than `1`). + /// The batch never fails: a failed item is replaced by an error item + /// (`{ json: { error, failed: true }, … }`) in its own slot, so the node + /// always emits exactly one item per input and a downstream node can branch + /// on `=item.json.failed`. /// - /// Note this is deliberately *more* forgiving than a bare sequential loop, - /// which propagated the first error and failed the node. Graphs that want - /// the old behaviour set [`FailFast`](ItemErrorPolicy::FailFast). - #[default] + /// A fan-out is a batch of independent work — losing 19 good results + /// because the 20th timed out is rarely what the author wanted, and with + /// items completing concurrently there is no single "the error" to hand to + /// `on_error` anyway. Collect, + /// **The default when the node runs sequentially** (`concurrency` unset or + /// `1`), which is how every per-item node behaved before fan-out existed. + /// /// The first failure **in input order** fails the whole node, which then /// falls to the node's own `on_error` / retry policy. Remaining in-flight /// items are cancelled once no earlier item can still fail. + /// + /// This default is load-bearing, not merely conservative: `tool_call`, + /// `http_request`, and `memory` are `per_item` *by default*, so collecting + /// here would silently disable `on_error` / retry / the `error` port for + /// the most ordinary nodes in the engine. FailFast, /// Failed items are dropped: the node emits only the successes, so the /// output array may be shorter than the input. @@ -76,11 +99,12 @@ pub(crate) struct MapOptions { } impl Default for MapOptions { - /// Sequential and collecting — the back-compatible defaults. + /// Sequential and fail-fast — exactly how per-item nodes behaved before + /// fan-out existed. fn default() -> Self { Self { concurrency: 1, - on_item_error: ItemErrorPolicy::Collect, + on_item_error: ItemErrorPolicy::FailFast, } } } @@ -99,7 +123,9 @@ impl Default for MapOptions { #[must_use] pub(crate) fn map_options(config: &Value, node_id: &str) -> MapOptions { let concurrency = match config.get("concurrency") { - Some(Value::Number(n)) => n.as_u64().map_or(1, |n| usize::try_from(n).unwrap_or(usize::MAX)), + Some(Value::Number(n)) => n + .as_u64() + .map_or(1, |n| usize::try_from(n).unwrap_or(usize::MAX)), // `"all"` is the readable spelling of "no bound" — `Promise.all`. Some(Value::String(s)) if s == "all" => 0, _ => 1, @@ -116,10 +142,20 @@ pub(crate) fn map_options(config: &Value, node_id: &str) -> MapOptions { concurrency }; + // The default follows the execution shape: a fan-out collects (one bad item + // must not discard the batch), while a sequential run keeps failing fast so + // the node's `on_error` / retry policy still sees the error. An explicit + // `on_item_error` overrides either way. + let default_policy = if concurrency == 1 { + ItemErrorPolicy::FailFast + } else { + ItemErrorPolicy::Collect + }; let on_item_error = match config.get("on_item_error").and_then(Value::as_str) { Some("fail_fast") => ItemErrorPolicy::FailFast, Some("skip") => ItemErrorPolicy::Skip, - _ => ItemErrorPolicy::Collect, + Some("collect") => ItemErrorPolicy::Collect, + _ => default_policy, }; MapOptions { @@ -147,29 +183,33 @@ fn error_item(message: &str) -> Item { )) } -/// Runs `f` over `input` with bounded concurrency, returning the output items -/// in **input order** with `paired_item` set, plus the union of every item's -/// diagnostics. +/// Runs `f` over the `total` input indices with bounded concurrency, returning +/// the output items in **input order** with `paired_item` set, plus the union +/// of every item's diagnostics. /// -/// `f` receives each item's input index and the item itself. Items complete out -/// of order; the results are re-sorted into input-order slots before returning, -/// so a fan-out never reorders a workflow's data. +/// `f` receives an input **index** rather than the item itself: the caller +/// already holds the input slice (on `ctx.input`), and passing a borrowed item +/// across this generic boundary forces a higher-ranked bound that rustc cannot +/// satisfy for an `async` body that also borrows the node context. An index +/// keeps every lifetime concrete at the call site. +/// +/// Items complete out of order; the results are re-sorted into input-order +/// slots before returning, so a fan-out never reorders a workflow's data. /// /// # Errors /// /// Only under [`ItemErrorPolicy::FailFast`], which returns the failure with the /// **lowest input index** — not the first to complete, which would make the /// error non-deterministic across runs. The other two policies never error. -pub(crate) async fn map_items<'a, F, Fut>( - input: &'a [Item], +pub(crate) async fn map_items( + total: usize, opts: MapOptions, f: F, ) -> Result<(Vec, Vec)> where - F: Fn(usize, &'a Item) -> Fut, - Fut: Future> + 'a, + F: Fn(usize) -> Fut, + Fut: Future>, { - let total = input.len(); // `buffer_unordered(0)` would never poll anything, so "unbounded" is spelled // as "as many as there are items". let in_flight = if opts.concurrency == 0 { @@ -179,8 +219,8 @@ where }; // Each future carries its input index so completions can be re-slotted. - let mut stream = futures_util::stream::iter(input.iter().enumerate().map(|(index, item)| { - let fut = f(index, item); + let mut stream = futures_util::stream::iter((0..total).map(|index| { + let fut = f(index); async move { (index, fut.await) } })) .buffer_unordered(in_flight); @@ -291,16 +331,18 @@ mod tests { #[tokio::test] async fn results_keep_input_order_even_when_completion_order_is_reversed() { let input = items(5); + let input = &input; // Later items finish first: item 0 yields the most, item 4 the least. - let (out, _) = map_items(&input, opts(0, ItemErrorPolicy::Collect), |index, item| { - let json = item.json.clone(); - async move { + let (out, _) = map_items( + input.len(), + opts(0, ItemErrorPolicy::Collect), + |index| async move { for _ in 0..(5 - index) * 4 { tokio::task::yield_now().await; } - Ok((Item::new(json), vec![])) - } - }) + Ok((Item::new(input[index].json.clone()), vec![])) + }, + ) .await .expect("map"); @@ -316,16 +358,20 @@ mod tests { let input = items(6); let gauge = Arc::new(Gauge::default()); let g = gauge.clone(); - let (out, _) = map_items(&input, opts(1, ItemErrorPolicy::Collect), move |_, item| { - let g = g.clone(); - let json = item.json.clone(); - async move { - g.enter(); - tick().await; - g.exit(); - Ok((Item::new(json), vec![])) - } - }) + let input = &input; + let (out, _) = map_items( + input.len(), + opts(1, ItemErrorPolicy::Collect), + move |index| { + let g = g.clone(); + async move { + g.enter(); + tick().await; + g.exit(); + Ok((Item::new(input[index].json.clone()), vec![])) + } + }, + ) .await .expect("map"); @@ -338,16 +384,20 @@ mod tests { let input = items(12); let gauge = Arc::new(Gauge::default()); let g = gauge.clone(); - let (out, _) = map_items(&input, opts(4, ItemErrorPolicy::Collect), move |_, item| { - let g = g.clone(); - let json = item.json.clone(); - async move { - g.enter(); - tick().await; - g.exit(); - Ok((Item::new(json), vec![])) - } - }) + let input = &input; + let (out, _) = map_items( + input.len(), + opts(4, ItemErrorPolicy::Collect), + move |index| { + let g = g.clone(); + async move { + g.enter(); + tick().await; + g.exit(); + Ok((Item::new(input[index].json.clone()), vec![])) + } + }, + ) .await .expect("map"); @@ -365,16 +415,20 @@ mod tests { let input = items(7); let gauge = Arc::new(Gauge::default()); let g = gauge.clone(); - let (out, _) = map_items(&input, opts(0, ItemErrorPolicy::Collect), move |_, item| { - let g = g.clone(); - let json = item.json.clone(); - async move { - g.enter(); - tick().await; - g.exit(); - Ok((Item::new(json), vec![])) - } - }) + let input = &input; + let (out, _) = map_items( + input.len(), + opts(0, ItemErrorPolicy::Collect), + move |index| { + let g = g.clone(); + async move { + g.enter(); + tick().await; + g.exit(); + Ok((Item::new(input[index].json.clone()), vec![])) + } + }, + ) .await .expect("map"); @@ -385,15 +439,17 @@ mod tests { #[tokio::test] async fn collect_substitutes_an_error_item_and_keeps_the_array_length() { let input = items(4); - let (out, _) = map_items(&input, opts(0, ItemErrorPolicy::Collect), |index, item| { - let json = item.json.clone(); - async move { + let input = &input; + let (out, _) = map_items( + input.len(), + opts(0, ItemErrorPolicy::Collect), + |index| async move { if index == 2 { return Err(crate::error::EngineError::Capability("boom".into())); } - Ok((Item::new(json), vec![])) - } - }) + Ok((Item::new(input[index].json.clone()), vec![])) + }, + ) .await .expect("collect never fails the batch"); @@ -414,15 +470,17 @@ mod tests { #[tokio::test] async fn skip_drops_failures_and_shortens_the_array() { let input = items(4); - let (out, _) = map_items(&input, opts(0, ItemErrorPolicy::Skip), |index, item| { - let json = item.json.clone(); - async move { + let input = &input; + let (out, _) = map_items( + input.len(), + opts(0, ItemErrorPolicy::Skip), + |index| async move { if index % 2 == 0 { return Err(crate::error::EngineError::Capability("nope".into())); } - Ok((Item::new(json), vec![])) - } - }) + Ok((Item::new(input[index].json.clone()), vec![])) + }, + ) .await .expect("skip never fails the batch"); @@ -439,16 +497,20 @@ mod tests { let input = items(6); // Item 4 fails immediately; item 1 fails only after yielding. Input order // must win, so the reported error is item 1's. - let err = map_items(&input, opts(0, ItemErrorPolicy::FailFast), |index, _| async move { - if index == 4 { - return Err(crate::error::EngineError::Capability("late-index".into())); - } - if index == 1 { - tick().await; - return Err(crate::error::EngineError::Capability("early-index".into())); - } - Ok((Item::new(json!({ "i": index })), vec![])) - }) + let err = map_items( + input.len(), + opts(0, ItemErrorPolicy::FailFast), + |index| async move { + if index == 4 { + return Err(crate::error::EngineError::Capability("late-index".into())); + } + if index == 1 { + tick().await; + return Err(crate::error::EngineError::Capability("early-index".into())); + } + Ok((Item::new(json!({ "i": index })), vec![])) + }, + ) .await .expect_err("fail_fast must surface an error"); @@ -460,8 +522,7 @@ mod tests { #[tokio::test] async fn empty_input_yields_no_items_and_no_error() { - let input: Vec = vec![]; - let (out, diags) = map_items(&input, opts(0, ItemErrorPolicy::Collect), |_, _| async { + let (out, diags) = map_items(0, opts(0, ItemErrorPolicy::Collect), |_| async { unreachable!("no items to map") }) .await @@ -473,15 +534,19 @@ mod tests { #[tokio::test] async fn diagnostics_from_every_item_are_unioned() { let input = items(3); - let (_, diags) = map_items(&input, opts(0, ItemErrorPolicy::Collect), |index, _| async move { - Ok(( - Item::new(Value::Null), - vec![NullResolution { - location: format!("config.prompt[{index}]"), - expression: "=item.missing".to_string(), - }], - )) - }) + let (_, diags) = map_items( + input.len(), + opts(0, ItemErrorPolicy::Collect), + |index| async move { + Ok(( + Item::new(Value::Null), + vec![NullResolution { + location: format!("config.prompt[{index}]"), + expression: "=item.missing".to_string(), + }], + )) + }, + ) .await .expect("map"); assert_eq!(diags.len(), 3); @@ -490,16 +555,61 @@ mod tests { // --- map_options --- #[test] - fn options_default_to_sequential_and_collect() { + fn options_default_to_sequential_and_fail_fast() { + // The pre-fan-out behaviour, unchanged: one at a time, and a failure + // reaches the node's own `on_error` / retry policy. let o = map_options(&json!({}), "n"); assert_eq!(o.concurrency, 1, "unset concurrency stays sequential"); - assert_eq!(o.on_item_error, ItemErrorPolicy::Collect); + assert_eq!(o.on_item_error, ItemErrorPolicy::FailFast); + } + + #[test] + fn fanning_out_flips_the_default_policy_to_collect() { + // Opting into concurrency opts into batch semantics: one bad item must + // not discard the other results. + for concurrency in [0, 2, 8] { + let o = map_options(&json!({ "concurrency": concurrency }), "n"); + assert_eq!( + o.on_item_error, + ItemErrorPolicy::Collect, + "concurrency {concurrency} should default to collect" + ); + } + // ...but an explicit `concurrency: 1` is not a fan-out. + assert_eq!( + map_options(&json!({ "concurrency": 1 }), "n").on_item_error, + ItemErrorPolicy::FailFast + ); + } + + #[test] + fn an_explicit_policy_overrides_the_shape_derived_default() { + assert_eq!( + map_options(&json!({ "on_item_error": "collect" }), "n").on_item_error, + ItemErrorPolicy::Collect, + "sequential can opt into collecting" + ); + assert_eq!( + map_options( + &json!({ "concurrency": 8, "on_item_error": "fail_fast" }), + "n" + ) + .on_item_error, + ItemErrorPolicy::FailFast, + "a fan-out can opt back into failing fast" + ); } #[test] fn options_read_numeric_and_all_concurrency() { - assert_eq!(map_options(&json!({ "concurrency": 8 }), "n").concurrency, 8); - assert_eq!(map_options(&json!({ "concurrency": 0 }), "n").concurrency, 0); + assert_eq!( + map_options(&json!({ "concurrency": 8 }), "n").concurrency, + 8 + ); + assert_eq!( + map_options(&json!({ "concurrency": 0 }), "n").concurrency, + 0 + ); assert_eq!( map_options(&json!({ "concurrency": "all" }), "n").concurrency, 0, @@ -521,7 +631,10 @@ mod tests { map_options(&json!({ "concurrency": "lots" }), "n").concurrency, 1 ); - assert_eq!(map_options(&json!({ "concurrency": -3 }), "n").concurrency, 1); + assert_eq!( + map_options(&json!({ "concurrency": -3 }), "n").concurrency, + 1 + ); assert_eq!( map_options(&json!({ "concurrency": true }), "n").concurrency, 1 @@ -530,22 +643,15 @@ mod tests { #[test] fn options_read_every_item_error_policy() { + let policy = + |v| map_options(&json!({ "concurrency": 4, "on_item_error": v }), "n").on_item_error; + assert_eq!(policy("fail_fast"), ItemErrorPolicy::FailFast); + assert_eq!(policy("skip"), ItemErrorPolicy::Skip); + assert_eq!(policy("collect"), ItemErrorPolicy::Collect); assert_eq!( - map_options(&json!({ "on_item_error": "fail_fast" }), "n").on_item_error, - ItemErrorPolicy::FailFast - ); - assert_eq!( - map_options(&json!({ "on_item_error": "skip" }), "n").on_item_error, - ItemErrorPolicy::Skip - ); - assert_eq!( - map_options(&json!({ "on_item_error": "collect" }), "n").on_item_error, - ItemErrorPolicy::Collect - ); - assert_eq!( - map_options(&json!({ "on_item_error": "bogus" }), "n").on_item_error, + policy("bogus"), ItemErrorPolicy::Collect, - "unknown policies fall back to the default" + "unknown policies fall back to the shape-derived default" ); } } From ff4e21d1b27f5c7d56491638d1633ddf95d2d0e0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:49:34 +0300 Subject: [PATCH 3/7] feat(sub_workflow): add per-item execution so one node runs N child workflows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `execution: per_item` turns sub_workflow into the multiplier: one full child run per input item, each seeded with just that item and resolving `workflow_id` against it, bounded by `concurrency`. Default stays `once`, so existing graphs are unchanged. The depth guard is per child run, so a fan-out widens the run without deepening it — N siblings at depth d+1, never d+N. --- src/nodes/integration/sub_workflow.rs | 465 ++++++++++++++++++++------ 1 file changed, 360 insertions(+), 105 deletions(-) diff --git a/src/nodes/integration/sub_workflow.rs b/src/nodes/integration/sub_workflow.rs index 596b280..43ada75 100644 --- a/src/nodes/integration/sub_workflow.rs +++ b/src/nodes/integration/sub_workflow.rs @@ -22,7 +22,27 @@ use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; /// /// The resolved child is compiled and run via [`crate::engine::run_sub_workflow`], /// sharing the host [`Capabilities`](crate::caps::Capabilities) with the parent -/// run, and its final run state is emitted as a single output item. +/// run, and its final run state is emitted as an output item. +/// +/// ## Execution: the multiplier +/// +/// `config.execution` (default `once`) decides how many child runs this node +/// performs: +/// +/// - `once` — one child run seeded with the node's whole input array. +/// - `per_item` — **one full child run per input item**, each seeded with just +/// that item and resolving `workflow_id` against it (so `=item.x` addresses +/// the element that run is for). `config.concurrency` bounds how many run at +/// a time and `config.on_item_error` what a failing child does to the batch +/// (see [`crate::nodes::map`]). This is how an array of work becomes N +/// parallel multi-step workflows. +/// +/// Only the fields *this* node reads are `=`-resolved; an inline `workflow` +/// graph always passes through untouched because its expressions belong to the +/// child run. +/// +/// The depth guard below is per child run, so a fan-out widens a run without +/// deepening it — N siblings at depth d+1, never d+N. /// /// ## Cycle / depth handling /// @@ -70,117 +90,158 @@ fn reject_self_reference(child: &WorkflowGraph, workflow_id: &str) -> Result<()> #[async_trait] impl NodeExecutor for SubWorkflowNode { async fn execute(&self, ctx: NodeContext<'_>) -> Result { - // The inline `workflow` graph carries its *own* `=`-expressions, scoped - // to the CHILD run — it must pass through untouched. Only the fields the - // sub_workflow node itself reads (here `workflow_id`) are resolved - // against this node's input scope, mirroring every other integration - // node (see `tool_call`). - let inline = ctx.node.config.get("workflow"); + // Execution mode (default `once`): `once` runs the child graph a single + // time with the node's whole input array as its payload. `per_item` + // makes this node the **multiplier** — one full child run per input + // item, each seeded with just that item, bounded by + // `config.concurrency`. That is what turns an array of work into N + // parallel multi-step workflows. + let per_item = + crate::nodes::execution_mode(&ctx.node.config, crate::nodes::ExecutionMode::Once) + == crate::nodes::ExecutionMode::PerItem + && !ctx.input.is_empty(); + + if per_item { + let opts = crate::nodes::map::map_options(&ctx.node.config, &ctx.node.id); + let ctx = &ctx; + let (items, _) = + crate::nodes::map::map_items(ctx.input.len(), opts, move |index| async move { + let item = &ctx.input[index]; + // Each child resolves `workflow_id` against *its own* item, + // so `=item.x` addresses the element this run is for, and + // receives that single item as its input. + let scope = crate::nodes::expr_scope_for(ctx, item.json.clone()); + let child = run_child(ctx, &scope, std::slice::from_ref(item)).await?; + Ok((child, vec![])) + }) + .await?; + return Ok(NodeOutput::main(items)); + } + let scope = crate::nodes::expr_scope(&ctx); - let resolved_workflow_id = ctx - .node - .config - .get("workflow_id") - .map(|v| crate::expr::resolve(v, &scope)); - let workflow_id = resolved_workflow_id - .as_ref() - .and_then(Value::as_str) - .filter(|s| !s.is_empty()); - - // Exactly one of `workflow` / `workflow_id` must be set. - let child: WorkflowGraph = match (inline, workflow_id) { - (Some(_), Some(_)) => { - return Err(EngineError::Capability( - "sub_workflow node: set exactly one of `workflow` (inline) or `workflow_id` \ - (reference), not both" - .to_string(), - )); - } - (None, None) => { - return Err(EngineError::Capability( - "sub_workflow node: missing `workflow` (inline) or `workflow_id` (reference) \ - in config" - .to_string(), - )); - } - (Some(inline_value), None) => { - tracing::debug!(node = %ctx.node.id, "sub_workflow: running inline child graph"); - inline_child(inline_value)? - } - (None, Some(id)) => { - tracing::debug!(node = %ctx.node.id, workflow_id = %id, "sub_workflow: resolving child graph by workflow_id"); - let resolved = ctx.caps.resolver.resolve(id).await?; - reject_self_reference(&resolved, id)?; - resolved - } - }; + let item = run_child(&ctx, &scope, ctx.input).await?; + Ok(NodeOutput::main(vec![item])) + } +} - // Depth / cycle guard: bound total nesting regardless of how a cycle is - // formed. The child runs one level deeper than the current run. - let child_depth = current_depth(ctx.run) + 1; - if child_depth > MAX_SUB_WORKFLOW_DEPTH { - return Err(EngineError::Capability(format!( - "sub_workflow node: maximum nesting depth {MAX_SUB_WORKFLOW_DEPTH} exceeded \ - (possible cycle)" - ))); +/// Resolves this node's child graph and runs it once, returning the child's +/// final run state as a single [`Item`](crate::data::Item). +/// +/// `scope` is the expression scope `workflow_id` is resolved against (the whole +/// input for `once`, the current element for `per_item`), and `child_input` is +/// the item array seeded into the child run. +async fn run_child( + ctx: &NodeContext<'_>, + scope: &Value, + child_input: &[crate::data::Item], +) -> Result { + // The inline `workflow` graph carries its *own* `=`-expressions, scoped + // to the CHILD run — it must pass through untouched. Only the fields the + // sub_workflow node itself reads (here `workflow_id`) are resolved + // against this node's input scope, mirroring every other integration + // node (see `tool_call`). + let inline = ctx.node.config.get("workflow"); + let resolved_workflow_id = ctx + .node + .config + .get("workflow_id") + .map(|v| crate::expr::resolve(v, scope)); + let workflow_id = resolved_workflow_id + .as_ref() + .and_then(Value::as_str) + .filter(|s| !s.is_empty()); + + // Exactly one of `workflow` / `workflow_id` must be set. + let child: WorkflowGraph = match (inline, workflow_id) { + (Some(_), Some(_)) => { + return Err(EngineError::Capability( + "sub_workflow node: set exactly one of `workflow` (inline) or `workflow_id` \ + (reference), not both" + .to_string(), + )); } - - let compiled = crate::compiler::compile(&child)?; - let input = - serde_json::to_value(ctx.input).map_err(|e| EngineError::Capability(e.to_string()))?; - // Box the recursive engine call so the async future type stays sized. - let outcome = Box::pin(crate::engine::run_sub_workflow( - &compiled, - input, - ctx.caps, - child_depth, - )) - .await?; - - // Enforce the child's lifecycle across the sub-workflow boundary (BUG-5). - // - // The child run is a *separate* engine invocation whose non-completion is - // reported on its [`RunOutcome`], not on the [`NodeOutput`] this node - // returns. A node executor has no channel to inject a tinyagents interrupt - // into the *parent* run (the parent's `pending_approvals` are collected - // solely from its own boundary interrupts), so we cannot yet transparently - // pause the parent and resume the child at its gate. What we MUST NOT do is - // keep only `outcome.output` and report success — that silently treats a - // child that paused at a `requires_approval` gate (or was cancelled) as if - // it had run to completion, making approval gating unenforceable across the - // boundary. - // - // Until full cross-boundary resume exists, fail loudly: a child that did - // not fully complete halts the parent with an error rather than letting it - // falsely complete. With the default `on_error: stop` policy this stops the - // parent run; with `continue`/`route` it becomes a routable error item — - // either way the gated child is never silently treated as completed. - // - // Follow-up for full cross-boundary resume: surface the child's - // `pending_approvals` (namespaced by this node's id) into the parent's - // pending set via a real interrupt at this node's boundary, and teach - // `engine::resume` to re-enter the child at its paused gate. That needs - // engine-level interrupt plumbing this node cannot express today. - if !outcome.pending_approvals.is_empty() { - return Err(EngineError::Capability(format!( - "sub_workflow node {:?}: child run paused awaiting approval at {:?}; \ - cross-boundary approval resume is not yet supported, so the parent run is \ - halted rather than falsely completed", - ctx.node.id, outcome.pending_approvals - ))); + (None, None) => { + return Err(EngineError::Capability( + "sub_workflow node: missing `workflow` (inline) or `workflow_id` (reference) \ + in config" + .to_string(), + )); + } + (Some(inline_value), None) => { + tracing::debug!(node = %ctx.node.id, "sub_workflow: running inline child graph"); + inline_child(inline_value)? } - if outcome.cancelled { - return Err(EngineError::Capability(format!( - "sub_workflow node {:?}: child run was cancelled before completing; the parent \ - run is halted rather than falsely completed", - ctx.node.id - ))); + (None, Some(id)) => { + tracing::debug!(node = %ctx.node.id, workflow_id = %id, "sub_workflow: resolving child graph by workflow_id"); + let resolved = ctx.caps.resolver.resolve(id).await?; + reject_self_reference(&resolved, id)?; + resolved } + }; - Ok(NodeOutput::main(vec![crate::data::Item::new( - outcome.output, - )])) + // Depth / cycle guard: bound total nesting regardless of how a cycle is + // formed. The child runs one level deeper than the current run. + let child_depth = current_depth(ctx.run) + 1; + if child_depth > MAX_SUB_WORKFLOW_DEPTH { + return Err(EngineError::Capability(format!( + "sub_workflow node: maximum nesting depth {MAX_SUB_WORKFLOW_DEPTH} exceeded \ + (possible cycle)" + ))); } + + let compiled = crate::compiler::compile(&child)?; + let input = + serde_json::to_value(child_input).map_err(|e| EngineError::Capability(e.to_string()))?; + // Box the recursive engine call so the async future type stays sized. + let outcome = Box::pin(crate::engine::run_sub_workflow( + &compiled, + input, + ctx.caps, + child_depth, + )) + .await?; + + // Enforce the child's lifecycle across the sub-workflow boundary (BUG-5). + // + // The child run is a *separate* engine invocation whose non-completion is + // reported on its [`RunOutcome`], not on the [`NodeOutput`] this node + // returns. A node executor has no channel to inject a tinyagents interrupt + // into the *parent* run (the parent's `pending_approvals` are collected + // solely from its own boundary interrupts), so we cannot yet transparently + // pause the parent and resume the child at its gate. What we MUST NOT do is + // keep only `outcome.output` and report success — that silently treats a + // child that paused at a `requires_approval` gate (or was cancelled) as if + // it had run to completion, making approval gating unenforceable across the + // boundary. + // + // Until full cross-boundary resume exists, fail loudly: a child that did + // not fully complete halts the parent with an error rather than letting it + // falsely complete. With the default `on_error: stop` policy this stops the + // parent run; with `continue`/`route` it becomes a routable error item — + // either way the gated child is never silently treated as completed. + // + // Follow-up for full cross-boundary resume: surface the child's + // `pending_approvals` (namespaced by this node's id) into the parent's + // pending set via a real interrupt at this node's boundary, and teach + // `engine::resume` to re-enter the child at its paused gate. That needs + // engine-level interrupt plumbing this node cannot express today. + if !outcome.pending_approvals.is_empty() { + return Err(EngineError::Capability(format!( + "sub_workflow node {:?}: child run paused awaiting approval at {:?}; \ + cross-boundary approval resume is not yet supported, so the parent run is \ + halted rather than falsely completed", + ctx.node.id, outcome.pending_approvals + ))); + } + if outcome.cancelled { + return Err(EngineError::Capability(format!( + "sub_workflow node {:?}: child run was cancelled before completing; the parent \ + run is halted rather than falsely completed", + ctx.node.id + ))); + } + + Ok(crate::data::Item::new(outcome.output)) } #[cfg(test)] @@ -229,6 +290,200 @@ mod tests { .expect_err("expected an error") } + /// Runs a `sub_workflow` node with the given config over `input_items`. + async fn execute_over( + config: Value, + input_items: Vec, + caps: &Capabilities, + ) -> NodeOutput { + let mut sw = node("sw", NodeKind::SubWorkflow); + sw.config = config; + let run_meta = Value::Null; + let ctx = NodeContext { + node: &sw, + input: &input_items, + run: &run_meta, + nodes: &Value::Null, + caps, + }; + SubWorkflowNode.execute(ctx).await.expect("execute") + } + + /// A child graph whose trigger simply carries the payload it was seeded with. + fn passthrough_child() -> WorkflowGraph { + WorkflowGraph { + nodes: vec![node("ct", NodeKind::Trigger)], + ..Default::default() + } + } + + #[tokio::test] + async fn per_item_runs_the_child_graph_once_per_input_item() { + // The multiplier: three items in, three complete child runs out. + let caps = mock_capabilities_with_resolver( + MockWorkflowResolver::default().with("child-1", passthrough_child()), + ); + let input = vec![ + crate::data::Item::new(json!({ "topic": "a" })), + crate::data::Item::new(json!({ "topic": "b" })), + crate::data::Item::new(json!({ "topic": "c" })), + ]; + let out = execute_over( + json!({ "workflow_id": "child-1", "execution": "per_item", "concurrency": 3 }), + input, + &caps, + ) + .await; + + assert_eq!(out.items.len(), 3, "one child run per input item"); + for (index, item) in out.items.iter().enumerate() { + assert_eq!(item.paired_item, Some(index), "output pairs to its input"); + } + // Each child was seeded with ONLY its own item: its trigger payload is a + // one-element item array, not the parent's whole input. + for item in &out.items { + assert_eq!( + item.json["run"]["trigger"] + .as_array() + .expect("trigger items") + .len(), + 1, + "each child sees exactly its own item" + ); + } + let topics: Vec<&str> = out + .items + .iter() + .map(|i| { + i.json["run"]["trigger"][0]["json"]["topic"] + .as_str() + .expect("topic") + }) + .collect(); + assert_eq!(topics, ["a", "b", "c"], "children keep input order"); + } + + #[tokio::test] + async fn once_is_still_the_default_and_seeds_the_whole_input_array() { + // Back-compat: without `execution` the node runs a single child seeded + // with every input item, exactly as before fan-out existed. + let caps = mock_capabilities_with_resolver( + MockWorkflowResolver::default().with("child-1", passthrough_child()), + ); + let input = vec![ + crate::data::Item::new(json!({ "topic": "a" })), + crate::data::Item::new(json!({ "topic": "b" })), + ]; + let out = execute_over(json!({ "workflow_id": "child-1" }), input, &caps).await; + + assert_eq!( + out.items.len(), + 1, + "one child run regardless of input count" + ); + let seeded = &out.items[0].json["run"]["trigger"]; + assert_eq!( + seeded.as_array().expect("trigger items").len(), + 2, + "the single child is seeded with the whole input array" + ); + } + + #[tokio::test] + async fn per_item_resolves_workflow_id_against_the_current_item() { + // `=item.x` in `workflow_id` addresses the element this child run is + // for, so one node can dispatch each item to a different child graph. + let mut alpha = passthrough_child(); + alpha.name = "alpha".to_string(); + let mut beta = passthrough_child(); + beta.name = "beta".to_string(); + let caps = mock_capabilities_with_resolver( + MockWorkflowResolver::default() + .with("wf-alpha", alpha) + .with("wf-beta", beta), + ); + let input = vec![ + crate::data::Item::new(json!({ "which": "wf-alpha" })), + crate::data::Item::new(json!({ "which": "wf-beta" })), + ]; + let out = execute_over( + json!({ "workflow_id": "=item.which", "execution": "per_item" }), + input, + &caps, + ) + .await; + assert_eq!(out.items.len(), 2); + // Both resolved (an unknown id would have errored the batch), and each + // child echoed its own seed. + assert_eq!( + out.items[0].json["run"]["trigger"][0]["json"]["which"], + "wf-alpha" + ); + assert_eq!( + out.items[1].json["run"]["trigger"][0]["json"]["which"], + "wf-beta" + ); + } + + #[tokio::test] + async fn a_fanned_out_child_failure_is_collected_not_fatal() { + // Only `wf-ok` resolves; the other item's child fails to resolve. Under + // a fan-out's collect default the batch still returns one item per + // input, with the failure marked for a downstream branch. + let caps = mock_capabilities_with_resolver( + MockWorkflowResolver::default().with("wf-ok", passthrough_child()), + ); + let input = vec![ + crate::data::Item::new(json!({ "which": "wf-ok" })), + crate::data::Item::new(json!({ "which": "wf-missing" })), + ]; + let out = execute_over( + json!({ "workflow_id": "=item.which", "execution": "per_item", "concurrency": 2 }), + input, + &caps, + ) + .await; + + assert_eq!(out.items.len(), 2, "one output per input even on failure"); + assert!( + out.items[0].json["nodes"]["ct"].is_object(), + "the good child ran" + ); + assert_eq!(out.items[1].json["json"]["failed"], true); + assert!( + out.items[1].json["json"]["error"] + .as_str() + .expect("error message") + .contains("wf-missing") + ); + } + + #[tokio::test] + async fn a_fan_out_widens_the_run_without_deepening_it() { + // Every sibling child runs at depth+1; a fan-out of N must not consume N + // levels of the nesting budget (which would make wide fan-outs of nested + // workflows spuriously trip the cycle guard). + let caps = mock_capabilities_with_resolver( + MockWorkflowResolver::default().with("child-1", passthrough_child()), + ); + let input: Vec<_> = (0..12) + .map(|i| crate::data::Item::new(json!({ "i": i }))) + .collect(); + let out = execute_over( + json!({ "workflow_id": "child-1", "execution": "per_item", "concurrency": "all" }), + input, + &caps, + ) + .await; + assert_eq!(out.items.len(), 12, "12 siblings all completed"); + for item in &out.items { + assert_eq!( + item.json["run"]["sub_workflow_depth"], 1, + "every sibling runs one level down, not cumulatively deeper" + ); + } + } + #[tokio::test] async fn missing_workflow_config_is_a_capability_error() { let err = execute_err(Value::Null).await; From 3378e5eb340a9ae072ca08fe2d291ded817bf005 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:51:55 +0300 Subject: [PATCH 4/7] feat(validate): reject malformed or no-op fan-out config at author time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `execution`, `concurrency`, and `on_item_error` select the execution strategy, so a bad value cannot be caught at run time without silently changing behaviour. Notably a fan-out knob on a node that runs once is rejected rather than ignored — otherwise an author asks for parallelism and gets none with no signal at all. --- src/validate.rs | 244 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 244 insertions(+) diff --git a/src/validate.rs b/src/validate.rs index 6428fd5..fc6012b 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -7,6 +7,16 @@ use serde_json::Value; use crate::error::ValidationError; use crate::model::{NodeKind, WorkflowGraph}; +/// The node kind's wire discriminator (`tool_call`, `sub_workflow`, …) for use +/// in error messages, so a validation error names the kind the way the graph +/// JSON spells it rather than in Rust's `PascalCase`. +fn kind_name(kind: &NodeKind) -> String { + serde_json::to_value(kind) + .ok() + .and_then(|v| v.as_str().map(str::to_string)) + .unwrap_or_else(|| format!("{kind:?}")) +} + /// Validates a workflow graph's structure. /// /// Currently checks: unique node ids, exactly one trigger node, that every edge @@ -137,6 +147,98 @@ pub fn validate_all(graph: &WorkflowGraph) -> Vec { } } + // Per-item fan-out config (`execution` / `concurrency` / `on_item_error`). + // These select the execution strategy, so an unrecognized value cannot be + // caught at run time without silently changing behaviour — a bad + // `concurrency` would quietly stay sequential and a bad `on_item_error` + // would quietly pick a default. Reject them here, where the message can name + // the node. + for node in &graph.nodes { + let fans_out = matches!( + node.kind, + NodeKind::Agent + | NodeKind::ToolCall + | NodeKind::HttpRequest + | NodeKind::Memory + | NodeKind::SubWorkflow + ); + + if let Some(execution) = node.config.get("execution") { + match execution.as_str() { + Some("once" | "per_item") if fans_out => {} + Some("once" | "per_item") => { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!( + "`execution` is not supported on a {} node (only agent, tool_call, \ + http_request, memory, and sub_workflow map over their input)", + kind_name(&node.kind) + ), + }); + } + _ => { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!( + "unknown `execution` value {execution} (expected \"once\" or \ + \"per_item\")" + ), + }); + } + } + } + + // Whether this node actually maps over its input, accounting for the + // per-kind default: `tool_call` / `http_request` / `memory` are per-item + // unless told otherwise; `agent` / `sub_workflow` are not. + let per_item = match node.config.get("execution").and_then(Value::as_str) { + Some("per_item") => true, + Some("once") => false, + _ => matches!( + node.kind, + NodeKind::ToolCall | NodeKind::HttpRequest | NodeKind::Memory + ), + }; + + for key in ["concurrency", "on_item_error"] { + let Some(value) = node.config.get(key) else { + continue; + }; + // A fan-out knob on a node that runs once is a no-op, and a silent + // no-op reads as "I asked for parallelism and got none". + if !per_item { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!( + "`{key}` has no effect without `execution: \"per_item\"` on a {} node", + kind_name(&node.kind) + ), + }); + continue; + } + let ok = match key { + "concurrency" => { + matches!( + value, + Value::Number(n) if n.as_u64().is_some(), + ) || value.as_str() == Some("all") + } + _ => matches!(value.as_str(), Some("collect" | "fail_fast" | "skip")), + }; + if !ok { + let expected = if key == "concurrency" { + "a non-negative integer or \"all\"" + } else { + "\"collect\", \"fail_fast\", or \"skip\"" + }; + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!("`{key}` must be {expected}, got {value}"), + }); + } + } + } + // `memory` node config checks, including THE hard security invariant: a // `remember`/`forget` operation may never target `scope: "user"` — the // caller's durable, cross-flow memory. Rejecting this structurally, at the @@ -1277,3 +1379,145 @@ mod tests { ); } } + +#[cfg(test)] +mod fanout_tests { + use super::{validate, validate_all}; + use crate::error::ValidationError; + use crate::model::{Node, NodeKind, WorkflowGraph}; + use serde_json::{Value, json}; + + /// A trigger plus one configured node of `kind` — the smallest graph that + /// exercises a per-kind config check. + fn graph(kind: NodeKind, config: Value) -> WorkflowGraph { + let mk = |id: &str, kind: NodeKind, config: Value| Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: Vec::new(), + position: None, + }; + WorkflowGraph { + nodes: vec![ + mk("t", NodeKind::Trigger, Value::Null), + mk("n", kind, config), + ], + ..Default::default() + } + } + + /// The `reason` of the single `InvalidNodeConfig` error, or a panic. + fn reason(kind: NodeKind, config: Value) -> String { + match validate_all(&graph(kind, config)) + .into_iter() + .find(|e| matches!(e, ValidationError::InvalidNodeConfig { .. })) + { + Some(ValidationError::InvalidNodeConfig { reason, .. }) => reason, + other => panic!("expected an InvalidNodeConfig error, got {other:?}"), + } + } + + #[test] + fn a_valid_fan_out_passes() { + assert_eq!( + validate(&graph( + NodeKind::Agent, + json!({ "execution": "per_item", "concurrency": 8, "on_item_error": "collect" }) + )), + Ok(()) + ); + // `"all"` and `0` are both legal spellings of unbounded. + for c in [json!("all"), json!(0)] { + assert_eq!( + validate(&graph( + NodeKind::ToolCall, + json!({ "execution": "per_item", "concurrency": c }) + )), + Ok(()) + ); + } + } + + #[test] + fn per_item_default_kinds_may_carry_fan_out_config_without_declaring_execution() { + // tool_call / http_request / memory are per-item by default, so the + // knobs apply without an explicit `execution`. + for kind in [NodeKind::ToolCall, NodeKind::HttpRequest] { + assert_eq!( + validate(&graph(kind.clone(), json!({ "concurrency": 4 }))), + Ok(()), + "{kind:?} is per-item by default" + ); + } + } + + #[test] + fn concurrency_on_a_once_node_is_rejected_rather_than_silently_ignored() { + // `agent` defaults to `once`, so this author asked for parallelism and + // would otherwise have got none, with no signal at all. + let reason = reason(NodeKind::Agent, json!({ "concurrency": 8 })); + assert!( + reason.contains("no effect") && reason.contains("per_item"), + "expected a no-effect explanation, got: {reason}" + ); + + // Explicitly opting out is the same story. + let reason = reason_of( + NodeKind::ToolCall, + json!({ "execution": "once", "concurrency": 8 }), + ); + assert!(reason.contains("no effect"), "got: {reason}"); + } + + fn reason_of(kind: NodeKind, config: Value) -> String { + reason(kind, config) + } + + #[test] + fn a_malformed_concurrency_is_rejected() { + for bad in [json!("lots"), json!(-1), json!(1.5), json!(true)] { + let reason = reason( + NodeKind::ToolCall, + json!({ "execution": "per_item", "concurrency": bad }), + ); + assert!( + reason.contains("concurrency"), + "expected a concurrency error for {bad}, got: {reason}" + ); + } + } + + #[test] + fn an_unknown_item_error_policy_is_rejected() { + let reason = reason( + NodeKind::ToolCall, + json!({ "execution": "per_item", "on_item_error": "explode" }), + ); + assert!( + reason.contains("on_item_error") && reason.contains("collect"), + "expected the allowed policies to be listed, got: {reason}" + ); + } + + #[test] + fn an_unknown_execution_value_is_rejected() { + let reason = reason(NodeKind::Agent, json!({ "execution": "parallel" })); + assert!( + reason.contains("execution") && reason.contains("per_item"), + "expected the allowed modes to be listed, got: {reason}" + ); + } + + #[test] + fn execution_on_a_kind_that_cannot_map_is_rejected() { + // A `transform` node does not map over its input; accepting `execution` + // there would imply a fan-out that never happens. + let reason = reason(NodeKind::Transform, json!({ "execution": "per_item" })); + assert!( + reason.contains("not supported") && reason.contains("transform"), + "expected the kind to be named in its wire spelling, got: {reason}" + ); + } +} From 8d0085b0732de5db5b3bf6ddbb11b9aa97cb2c2e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:53:28 +0300 Subject: [PATCH 5/7] feat(catalog): advertise the fan-out knobs on every mapping node kind The catalog is what an authoring agent reads to discover config, so a feature absent from it is unreachable. Described once and appended to the five mapping kinds rather than copied into contracts that would drift. --- src/catalog.rs | 133 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 132 insertions(+), 1 deletion(-) diff --git a/src/catalog.rs b/src/catalog.rs index f40c91e..78c0b46 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -588,7 +588,79 @@ pub fn contract_for(kind: &str) -> Option { }, _ => 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") + ); + } +} From c2940f74fdfcc43ca1cbaab3e4509c4532de8ead Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:56:17 +0300 Subject: [PATCH 6/7] test(e2e): prove per-item fan-out is concurrent, ordered, and failure-isolated Covers the properties that make the feature usable rather than merely present: real overlap (a probe records peak in-flight, so a regression to a sequential loop fails loudly), input-order results, the sequential default, and both item-error policies. --- tests/per_item_fanout_e2e.rs | 297 +++++++++++++++++++++++++++++++++++ 1 file changed, 297 insertions(+) create mode 100644 tests/per_item_fanout_e2e.rs diff --git a/tests/per_item_fanout_e2e.rs b/tests/per_item_fanout_e2e.rs new file mode 100644 index 0000000..50c25b3 --- /dev/null +++ b/tests/per_item_fanout_e2e.rs @@ -0,0 +1,297 @@ +#![cfg(feature = "mock")] +//! End-to-end tests for **per-item fan-out**: one node multiplying an array of +//! input into N concurrent units of work, array in and array out. +//! +//! This is distinct from the graph-shaped fan-out covered by `parallel_e2e.rs`. +//! There, concurrency comes from authoring N sibling nodes, so the width is +//! fixed when the graph is written. Here a *single* node maps over whatever +//! array reaches it, so the width is data-driven — `split_out` → `agent` → +//! `merge` runs one agent turn per element, bounded by `config.concurrency`. +//! +//! The tests assert the three properties that make the feature usable: +//! items really do run concurrently, results come back in input order, and one +//! failing item does not discard the batch. +//! +//! Gated behind the `mock` cargo feature. + +use async_trait::async_trait; +use serde_json::{Value, json}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::caps::{Capabilities, LlmProvider}; +use tinyflows::compiler::compile; +use tinyflows::engine::run; +use tinyflows::model::{Edge, Node, NodeKind, TriggerKind, WorkflowGraph}; + +/// Builds a node with the given id, kind, and config (no ports, no position). +fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: vec![], + position: None, + } +} + +/// Builds an edge from `from_node`'s `main` port into `to_node`'s `main` port. +fn edge(from_node: &str, to_node: &str) -> Edge { + Edge { + from_node: from_node.to_string(), + from_port: "main".to_string(), + to_node: to_node.to_string(), + to_port: "main".to_string(), + } +} + +/// An LLM stand-in that records how many completions overlap, so a test can +/// prove work actually ran concurrently rather than merely finishing. +/// +/// Each call registers itself, yields enough times for its peers to start, then +/// echoes the request's prompt. `peak` is the high-water mark of simultaneous +/// calls — 1 means the batch ran strictly sequentially. +#[derive(Default)] +struct ConcurrencyProbe { + live: AtomicUsize, + peak: AtomicUsize, + calls: AtomicUsize, + /// A prompt that must fail, to exercise the collect policy. + fail_on: Option, +} + +impl ConcurrencyProbe { + fn peak(&self) -> usize { + self.peak.load(Ordering::SeqCst) + } + fn calls(&self) -> usize { + self.calls.load(Ordering::SeqCst) + } +} + +#[async_trait] +impl LlmProvider for ConcurrencyProbe { + async fn complete( + &self, + request: Value, + _conn: Option<&str>, + ) -> tinyflows::error::Result { + let live = self.live.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(live, Ordering::SeqCst); + self.calls.fetch_add(1, Ordering::SeqCst); + for _ in 0..8 { + tokio::task::yield_now().await; + } + self.live.fetch_sub(1, Ordering::SeqCst); + + let prompt = request + .get("prompt") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + if self.fail_on.as_deref() == Some(prompt.as_str()) { + return Err(tinyflows::error::EngineError::Capability(format!( + "probe: refusing {prompt}" + ))); + } + Ok(json!({ "text": prompt })) + } +} + +fn caps_with(probe: Arc) -> Capabilities { + Capabilities { + llm: probe, + ..mock_capabilities() + } +} + +/// `trigger → split_out(topics) → agent(per_item) → merge`, with the agent's +/// prompt bound to the current item. This is the canonical fan-out shape. +fn fanout_graph(agent_config: Value) -> WorkflowGraph { + WorkflowGraph { + nodes: vec![ + node( + "t", + NodeKind::Trigger, + json!({ "kind": TriggerKind::Manual }), + ), + node("split", NodeKind::SplitOut, json!({ "path": "topics" })), + node("work", NodeKind::Agent, agent_config), + node("join", NodeKind::Merge, Value::Null), + ], + edges: vec![ + edge("t", "split"), + edge("split", "work"), + edge("work", "join"), + ], + ..Default::default() + } +} + +/// The trigger payload: five topics to fan out over. Each element is an object +/// so the agent prompt can bind a named field (`=item.name`), which is how a +/// real fan-out addresses its current element. +fn topics() -> Value { + json!({ + "topics": [ + { "name": "alpha" }, + { "name": "beta" }, + { "name": "gamma" }, + { "name": "delta" }, + { "name": "epsilon" }, + ] + }) +} + +/// The `text` of every item a node emitted, in emission order. +fn texts(output: &Value, node_id: &str) -> Vec { + output["nodes"][node_id]["items"] + .as_array() + .expect("items array") + .iter() + .map(|i| i["json"]["text"].as_str().unwrap_or_default().to_string()) + .collect() +} + +#[tokio::test] +async fn a_fanned_out_agent_runs_items_concurrently_and_returns_them_in_order() { + let probe = Arc::new(ConcurrencyProbe::default()); + let graph = fanout_graph(json!({ + "prompt": "=item.name", + "execution": "per_item", + "concurrency": 4, + })); + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, topics(), &caps_with(probe.clone())) + .await + .expect("run"); + + assert_eq!(probe.calls(), 5, "one agent turn per input item"); + assert!( + probe.peak() > 1, + "the whole point is concurrency; peak in-flight was {}", + probe.peak() + ); + assert!( + probe.peak() <= 4, + "must respect the concurrency bound; peak in-flight was {}", + probe.peak() + ); + // Array in, array out — in the original order despite finishing out of order. + assert_eq!( + texts(&out.output, "work"), + ["alpha", "beta", "gamma", "delta", "epsilon"] + ); + // ...and the merge downstream sees the whole array. + assert_eq!( + out.output["nodes"]["join"]["items"] + .as_array() + .expect("merged items") + .len(), + 5 + ); +} + +#[tokio::test] +async fn concurrency_all_runs_every_item_at_once() { + let probe = Arc::new(ConcurrencyProbe::default()); + let graph = fanout_graph(json!({ + "prompt": "=item.name", + "execution": "per_item", + "concurrency": "all", + })); + let compiled = compile(&graph).expect("compile"); + run(&compiled, topics(), &caps_with(probe.clone())) + .await + .expect("run"); + + assert_eq!(probe.peak(), 5, "`\"all\"` means every item at once"); +} + +#[tokio::test] +async fn without_concurrency_the_same_graph_stays_sequential() { + // Back-compat guard: opting into `per_item` alone must not change timing. + // If this ever reports a peak above 1, fan-out has become the default and + // every existing workflow silently changed its concurrency profile. + let probe = Arc::new(ConcurrencyProbe::default()); + let graph = fanout_graph(json!({ + "prompt": "=item.name", + "execution": "per_item", + })); + let compiled = compile(&graph).expect("compile"); + run(&compiled, topics(), &caps_with(probe.clone())) + .await + .expect("run"); + + assert_eq!(probe.calls(), 5); + assert_eq!(probe.peak(), 1, "unset concurrency must stay sequential"); +} + +#[tokio::test] +async fn one_failing_item_does_not_discard_the_rest_of_the_batch() { + let probe = Arc::new(ConcurrencyProbe { + fail_on: Some("gamma".to_string()), + ..Default::default() + }); + let graph = fanout_graph(json!({ + "prompt": "=item.name", + "execution": "per_item", + "concurrency": 4, + })); + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, topics(), &caps_with(probe.clone())) + .await + .expect("a fanned-out batch collects item failures instead of failing the run"); + + let items = out.output["nodes"]["work"]["items"] + .as_array() + .expect("items array"); + assert_eq!(items.len(), 5, "one output per input, failures included"); + + // The failed slot is marked in place, so the good results survive and a + // downstream `condition` can branch on `=item.json.failed`. Each stored + // entry is a serialized Item, so `["json"]` is the envelope and + // `["json"]["json"]` is what `=item.json` resolves to. + assert_eq!(items[2]["json"]["json"]["failed"], true); + assert!( + items[2]["json"]["json"]["error"] + .as_str() + .expect("error message") + .contains("gamma") + ); + for good in [0, 1, 3, 4] { + assert!( + items[good]["json"]["json"]["failed"].is_null(), + "item {good} should have succeeded" + ); + } + // The surviving results are still the right ones, in the right slots. + assert_eq!(items[0]["json"]["text"], "alpha"); + assert_eq!(items[4]["json"]["text"], "epsilon"); +} + +#[tokio::test] +async fn a_fanned_out_batch_can_opt_back_into_failing_the_node() { + // `on_item_error: fail_fast` restores the sequential contract: the error + // reaches the node, so the node's own `on_error` policy governs the run. + let probe = Arc::new(ConcurrencyProbe { + fail_on: Some("gamma".to_string()), + ..Default::default() + }); + let graph = fanout_graph(json!({ + "prompt": "=item.name", + "execution": "per_item", + "concurrency": 4, + "on_item_error": "fail_fast", + })); + let compiled = compile(&graph).expect("compile"); + let err = run(&compiled, topics(), &caps_with(probe)) + .await + .expect_err("fail_fast must surface the item error as a node failure"); + assert!( + err.to_string().contains("gamma"), + "expected the failing item's error, got: {err}" + ); +} From 4817cd51b6cc142a6ede931ff7ae781bd072eca1 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:57:13 +0300 Subject: [PATCH 7/7] docs: document per-item fan-out in the README and node catalog --- README.md | 25 +++++++++++++++++++++++++ wiki/Node-Catalog.md | 44 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+) diff --git a/README.md b/README.md index 20c472e..cfb376b 100644 --- a/README.md +++ b/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** diff --git a/wiki/Node-Catalog.md b/wiki/Node-Catalog.md index cbe8a36..684a3b6 100644 --- a/wiki/Node-Catalog.md +++ b/wiki/Node-Catalog.md @@ -52,5 +52,49 @@ Per-node error handling (`on_error` stop/continue/route, `retry`, an `error` port) and approval gating (`requires_approval`) are configured through the same free-form `config`. +### Per-item fan-out + +`agent`, `tool_call`, `http_request`, `memory`, and `sub_workflow` can map over +their input array instead of running once. Three config keys control it, and +they mean the same thing on every one of those kinds: + +| Key | Values | Meaning | +|-----|--------|---------| +| `execution` | `once` \| `per_item` | Run once for the whole input array, or once per item. Defaults to `per_item` for `tool_call` / `http_request` / `memory`, and `once` for `agent` / `sub_workflow`. | +| `concurrency` | integer \| `"all"` | How many items run at a time: `1` (default) sequential, `n` at most n in flight, `0` or `"all"` unbounded. Clamped to 64. | +| `on_item_error` | `collect` \| `fail_fast` \| `skip` | What a failing item does to the batch. | + +```jsonc +// one agent turn per topic, at most 8 concurrently +{ "id": "research", "kind": "agent", "name": "Research each", + "config": { + "execution": "per_item", + "concurrency": 8, + "agent_ref": "researcher", + "prompt": "Research =item.name" + } } +``` + +`sub_workflow` in `per_item` mode is the **multiplier**: one complete child run +per item, each seeded with just that item and resolving `workflow_id` against +it. The nesting-depth guard is per child run, so a fan-out widens a run without +deepening it — N siblings at depth d+1, never d+N. + +Output items always come back in **input order** with `paired_item` set, +whatever the concurrency, so a fan-out never reorders data. + +`on_item_error` defaults to `collect` when the node fans out (`concurrency` +other than `1`) and `fail_fast` when it runs sequentially. That split matters: +`tool_call`, `http_request`, and `memory` are `per_item` *by default*, so +collecting unconditionally would silently disable `on_error`, `retry`, and the +`error` port for the most ordinary nodes in the engine. Under `collect` a failed +item becomes `{ json: { error, failed: true } }` in its own slot, so the node +still emits one output per input and a downstream `condition` can branch on +`=item.json.failed`; under `skip` it is dropped, so the output array may be +shorter than the input. + +These keys are rejected at validation time on a node that does not map over its +input — a fan-out knob that silently does nothing is worse than an error. + Each node kind's config keys and ports, along with the available trigger kinds, are documented in the sections above.