From 96d24e943b518ccca997ca961c5e25e716534e7e Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 11 Nov 2026 17:16:24 +0530 Subject: [PATCH] fix(engine): a gather in a loop counts only the current pass's lanes Lane slots are never cleared between loop passes, and the gather counted every slot under its predecessors against the scatter's lane_count, taken from the first lane_count anywhere in state. A pass with 3 lanes followed by one with 2 released [20, 21, 12], where 12 is the first pass's third lane. The scatter now stamps each lane envelope with its activation step, every lane slot records `_scatter` and `_scatter_step`, and the gather counts only slots from its own scatter's current activation, reading lane_count from that scatter. Slots from checkpoints written before this carry no stamp and count as before while no stamped slot exists. Also corrects the reason given for refusing a gather with no scatter: it releases empty and silently drops its predecessors' output. --- CHANGELOG.md | 15 ++ crates/tinyflows/src/engine.rs | 1 + .../src/engine/build/activation/routing.rs | 11 +- crates/tinyflows/src/engine/build/outcome.rs | 10 +- crates/tinyflows/src/engine/state.rs | 33 +++- crates/tinyflows/src/lane_context_tests.rs | 20 +++ .../src/nodes/control_flow/gather.rs | 97 ++++++++---- .../src/nodes/control_flow/gather_tests.rs | 84 +++++++++++ crates/tinyflows/src/nodes/mod.rs | 6 + crates/tinyflows/src/validate/scatter.rs | 8 +- .../tests/loop_scatter_lanes_tests.rs | 141 ++++++++++++++++++ crates/tinyflows/tests/scatter_gather_e2e.rs | 4 +- 12 files changed, 393 insertions(+), 37 deletions(-) create mode 100644 crates/tinyflows/tests/loop_scatter_lanes_tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index f5f7c142..10495a94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Gather in a looping scatter + +- **Fixed** a `gather` inside a loop body collecting an earlier pass's lanes: a + narrower pass (3 lanes, then 2) released the stale third lane too. Lane slots + now record the scatter and the super-step of the activation that opened them + (`_scatter`, `_scatter_step`), and a gather counts only the current + activation's lanes against its own scatter's `lane_count`. Slots from a + checkpoint written before this change carry no stamp and are counted as before + when no stamped slot exists, so an in-flight run still completes. +- **Added** `LaneContext::scatter_step` (`Option`); code that builds a + `LaneContext` literal must set it. +- **Fixed** the validation reason for a gather with no upstream `scatter`: it + releases with no items, silently dropping its predecessors' output, rather + than waiting out its poll budget. + ### Run event outcomes - **Added** the engine output to `Runs.PollRun` parked and finished event diff --git a/crates/tinyflows/src/engine.rs b/crates/tinyflows/src/engine.rs index 6719ee17..eb377b22 100644 --- a/crates/tinyflows/src/engine.rs +++ b/crates/tinyflows/src/engine.rs @@ -267,6 +267,7 @@ mod state; pub(crate) use state::replace; #[cfg(test)] pub(crate) use state::{LANE_KEY, REPLACE}; +pub(crate) use state::{LANE_SCATTER_KEY, LANE_SCATTER_STEP_KEY}; use state::{ MergeReducer, collect_input, collect_input_since, lane_context, lane_envelope, lane_input, lane_items_update, merge, stamp_activation_step, diff --git a/crates/tinyflows/src/engine/build/activation/routing.rs b/crates/tinyflows/src/engine/build/activation/routing.rs index 5891a9c0..1b2500b4 100644 --- a/crates/tinyflows/src/engine/build/activation/routing.rs +++ b/crates/tinyflows/src/engine/build/activation/routing.rs @@ -37,9 +37,14 @@ if gather_nodes.contains(&target) { RouteTarget::Node(target.into()) } else { - let envelope = - lane_envelope(&lane.origin, lane.index, lane.count, routed_items) - .unwrap_or(Value::Null); + let envelope = lane_envelope( + &lane.origin, + lane.index, + lane.count, + lane.scatter_step, + routed_items, + ) + .unwrap_or(Value::Null); RouteTarget::Send(crate::graph::Send::new(target, envelope)) } }) diff --git a/crates/tinyflows/src/engine/build/outcome.rs b/crates/tinyflows/src/engine/build/outcome.rs index 6505a545..e0d34d94 100644 --- a/crates/tinyflows/src/engine/build/outcome.rs +++ b/crates/tinyflows/src/engine/build/outcome.rs @@ -118,8 +118,14 @@ where let mut routed: Vec = Vec::new(); for (index, lane_items) in lanes.iter().enumerate() { for target in plain_targets { - let envelope = lane_envelope(&node.id, index, count, lane_items) - .unwrap_or(Value::Null); + let envelope = lane_envelope( + &node.id, + index, + count, + Some(activation_step), + lane_items, + ) + .unwrap_or(Value::Null); routed.push(RouteTarget::Send(crate::graph::Send::new( target.clone(), envelope, diff --git a/crates/tinyflows/src/engine/state.rs b/crates/tinyflows/src/engine/state.rs index d1ce7315..e214f8db 100644 --- a/crates/tinyflows/src/engine/state.rs +++ b/crates/tinyflows/src/engine/state.rs @@ -96,19 +96,30 @@ pub(super) fn lane_items_update( "status": status, "index": lane.index, }); - if let (Some(Value::Object(extra)), Some(map)) = (meta, slot.as_object_mut()) { - for (key, value) in extra { - map.insert(key.clone(), value.clone()); + if let Some(map) = slot.as_object_mut() { + if let Some(Value::Object(extra)) = meta { + for (key, value) in extra { + map.insert(key.clone(), value.clone()); + } + } + // After `meta`, so a node's own meta cannot claim another activation. + map.insert(LANE_SCATTER_KEY.to_string(), json!(lane.origin)); + if let Some(step) = lane.scatter_step { + map.insert(LANE_SCATTER_STEP_KEY.to_string(), json!(step)); } } Ok(json!({ "nodes": { node_id: { "lanes": { lane.id.clone(): slot } } } })) } /// The lane envelope a fan-out schedules one activation with. +/// +/// `scatter_step` is the super-step of the scatter activation that opened the +/// lane; every node along the lane passes it on unchanged. pub(super) fn lane_envelope( origin: &str, index: usize, count: usize, + scatter_step: Option, items: &[Item], ) -> crate::graph::Result { Ok(json!({ @@ -117,6 +128,7 @@ pub(super) fn lane_envelope( "origin": origin, "index": index, "count": count, + "scatter_step": scatter_step, }, "items": serde_json::to_value(items)?, })) @@ -147,6 +159,15 @@ pub(super) fn lane_input(send_arg: Option<&Value>) -> Vec { /// and must never be mistaken for user data. pub(crate) const LANE_KEY: &str = "_lane"; +/// The lane-slot key naming the scatter that opened the lane. +pub(crate) const LANE_SCATTER_KEY: &str = "_scatter"; + +/// The lane-slot key recording the super-step of the scatter activation that +/// opened the lane. Lane ids repeat across loop passes (`#`) +/// and slots are never cleared, so this is what lets a gather count only the +/// current pass. Absent on slots written from a pre-stamp envelope. +pub(crate) const LANE_SCATTER_STEP_KEY: &str = "_scatter_step"; + /// Decodes the lane identity from an activation's `send_arg`. /// /// `None` for an ordinary activation — one scheduled by a plain route rather @@ -160,6 +181,12 @@ pub(super) fn lane_context(send_arg: Option<&Value>) -> Option.lanes.*` against the `lane_count` the scatter -//! recorded, and asks to be re-run until its release policy is satisfied. +//! recorded, and asks to be re-run until its release policy is satisfied. Only +//! lanes opened by the scatter's current activation count, so a loop pass never +//! gathers an earlier pass's lanes. //! //! That is also why it supports the same policies as a [`gate`]: once the wait //! is a decision rather than a topological fact, "proceed on a quorum" and @@ -19,6 +21,7 @@ use async_trait::async_trait; use serde_json::{Value, json}; use crate::data::Item; +use crate::engine::{LANE_SCATTER_KEY, LANE_SCATTER_STEP_KEY}; use crate::error::{EngineError, Result}; use crate::nodes::release::{Release, ReleasePolicy}; use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; @@ -64,23 +67,61 @@ struct Arrived { failed: Option, } -/// Reads every lane slot recorded by this gather's lane-terminal predecessors. +/// Every lane slot recorded by this gather's lane-terminal predecessors. /// /// Lanes are keyed by id under `nodes..lanes`, which is what keeps N /// concurrent activations of one node from clobbering each other — the reducer /// merges objects key-by-key, so distinct lane keys never collide. -fn arrivals(ctx: &NodeContext<'_>, predecessors: &[String]) -> Vec { - let mut arrived = Vec::new(); - for pred in predecessors { - let Some(lanes) = ctx - .nodes - .get(pred) - .and_then(|slot| slot.get("lanes")) - .and_then(Value::as_object) - else { - continue; - }; - for slot in lanes.values() { +fn lane_slots<'a>(nodes: &'a Value, predecessors: &[String]) -> Vec<&'a Value> { + predecessors + .iter() + .filter_map(|pred| nodes.get(pred)?.get("lanes")?.as_object()) + .flat_map(|lanes| lanes.values()) + .collect() +} + +/// The scatter that opened these lanes, as the lane slots record it. +fn scatter_of<'a>(slots: &[&'a Value]) -> Option<&'a str> { + slots + .iter() + .copied() + .find_map(|slot| slot.get(LANE_SCATTER_KEY)?.as_str()) +} + +/// Keeps only the slots opened by the scatter's **current** activation. +/// +/// Slots are never cleared, so in a loop whose body scatters, an earlier pass's +/// lanes are still in state when a later pass gathers; counting them would +/// release a narrower pass with a stale lane in it. The current activation is +/// the super-step the scatter's own slot was last stamped with. +/// +/// A slot without the stamp was written from an envelope that predates it (a +/// run checkpointed by an older binary). Such slots are all counted, as before, +/// when no stamped slot exists, so a run resumed across an upgrade still +/// completes; once any stamped slot exists, unstamped ones are stale and ignored. +fn current_activation<'a>( + nodes: &Value, + slots: Vec<&'a Value>, + scatter: Option<&str>, +) -> Vec<&'a Value> { + let step_of = |slot: &Value| slot.get(LANE_SCATTER_STEP_KEY).and_then(Value::as_u64); + if !slots.iter().any(|slot| step_of(slot).is_some()) { + return slots; + } + let current = scatter + .and_then(|id| nodes.get(id)?.get("_activation_step")?.as_u64()) + .or_else(|| slots.iter().filter_map(|slot| step_of(slot)).max()); + slots + .into_iter() + .filter(|slot| step_of(slot) == current) + .collect() +} + +/// Decodes the lane slots that count. +fn arrivals(slots: Vec<&Value>) -> Vec { + slots + .into_iter() + .map(|slot| { let index = slot .get("index") .and_then(Value::as_u64) @@ -106,27 +147,31 @@ fn arrivals(ctx: &NodeContext<'_>, predecessors: &[String]) -> Vec { .unwrap_or("lane failed") .to_string() }); - arrived.push(Arrived { + Arrived { index, items, failed, - }); - } - } - arrived + } + }) + .collect() } /// How many lanes the scatter upstream of this gather opened. /// /// Read from the scatter's own slot rather than inferred from arrivals: a /// gather that guessed "however many turned up" would release immediately on -/// the first one, which is the bug this exists to prevent. -fn expected_lanes(ctx: &NodeContext<'_>) -> Option { - ctx.nodes.as_object()?.values().find_map(|slot| { +/// the first one, which is the bug this exists to prevent. Slots that do not +/// name their scatter (pre-stamp) fall back to the first `lane_count` in state. +fn expected_lanes(nodes: &Value, scatter: Option<&str>) -> Option { + let lane_count = |slot: &Value| { slot.get("lane_count") .and_then(Value::as_u64) .and_then(|n| usize::try_from(n).ok()) - }) + }; + match scatter { + Some(id) => nodes.get(id).and_then(lane_count), + None => nodes.as_object()?.values().find_map(lane_count), + } } fn positive_u64(config: &Value, key: &str, default: u64) -> u64 { @@ -157,8 +202,10 @@ impl NodeExecutor for GatherNode { }) .unwrap_or_default(); - let arrived = arrivals(&ctx, &predecessors); - let expected = expected_lanes(&ctx).unwrap_or(arrived.len()); + let slots = lane_slots(ctx.nodes, &predecessors); + let scatter = scatter_of(&slots); + let arrived = arrivals(current_activation(ctx.nodes, slots, scatter)); + let expected = expected_lanes(ctx.nodes, scatter).unwrap_or(arrived.len()); let polls = ctx .nodes diff --git a/crates/tinyflows/src/nodes/control_flow/gather_tests.rs b/crates/tinyflows/src/nodes/control_flow/gather_tests.rs index f8ce08de..6c7184f4 100644 --- a/crates/tinyflows/src/nodes/control_flow/gather_tests.rs +++ b/crates/tinyflows/src/nodes/control_flow/gather_tests.rs @@ -65,3 +65,87 @@ fn on_lane_error_defaults_to_collect() { "an unrecognised policy must not silently fail the run" ); } + +/// The lane indices a gather over `evaluate` would count, and its expected total. +fn counted(nodes: &Value) -> (Vec, Option) { + let slots = lane_slots(nodes, &["evaluate".to_string()]); + let scatter = scatter_of(&slots); + let mut indices: Vec = arrivals(current_activation(nodes, slots, scatter)) + .iter() + .map(|lane| lane.index) + .collect(); + indices.sort_unstable(); + (indices, expected_lanes(nodes, scatter)) +} + +fn stamped(index: usize, step: u64) -> Value { + json!({ + "items": [], "status": "ok", "index": index, + "_scatter": "scatter", "_scatter_step": step, + }) +} + +/// A second loop pass opened 2 lanes at step 9; pass 1's lane #2 (step 3) is +/// still in state and must not be counted. +#[test] +fn lanes_from_an_earlier_scatter_activation_are_ignored() { + let nodes = json!({ + "scatter": { "lane_count": 2, "_activation_step": 9 }, + "evaluate": { "lanes": { + "scatter#0": stamped(0, 9), + "scatter#1": stamped(1, 9), + "scatter#2": stamped(2, 3), + } }, + }); + assert_eq!(counted(&nodes), (vec![0, 1], Some(2))); +} + +/// Before any of this pass's lanes land, the earlier pass's lanes count for +/// nothing — the gather waits rather than releasing stale results. +#[test] +fn an_activation_with_no_lanes_landed_yet_counts_none() { + let nodes = json!({ + "scatter": { "lane_count": 2, "_activation_step": 9 }, + "evaluate": { "lanes": { "scatter#0": stamped(0, 3), "scatter#1": stamped(1, 3) } }, + }); + assert_eq!(counted(&nodes), (vec![], Some(2))); +} + +/// The expected total is this gather's own scatter's, not the first +/// `lane_count` anywhere in state. +#[test] +fn the_expected_count_comes_from_the_scatter_that_opened_the_lanes() { + let nodes = json!({ + "an_earlier_scatter": { "lane_count": 7, "_activation_step": 1 }, + "scatter": { "lane_count": 1, "_activation_step": 4 }, + "evaluate": { "lanes": { "scatter#0": stamped(0, 4) } }, + }); + assert_eq!(counted(&nodes), (vec![0], Some(1))); +} + +/// Slots written from a pre-stamp envelope (a checkpoint from an older binary) +/// all count when nothing is stamped, so an in-flight run still completes. +#[test] +fn unstamped_slots_count_when_no_slot_is_stamped() { + let nodes = json!({ + "scatter": { "lane_count": 2, "_activation_step": 9 }, + "evaluate": { "lanes": { + "scatter#0": { "items": [], "status": "ok", "index": 0 }, + "scatter#1": { "items": [], "status": "ok", "index": 1 }, + } }, + }); + assert_eq!(counted(&nodes), (vec![0, 1], Some(2))); +} + +/// Once this pass's lanes are stamped, unstamped slots are an earlier pass's. +#[test] +fn unstamped_slots_are_stale_once_a_stamped_slot_exists() { + let nodes = json!({ + "scatter": { "lane_count": 1, "_activation_step": 9 }, + "evaluate": { "lanes": { + "scatter#0": stamped(0, 9), + "scatter#1": { "items": [], "status": "ok", "index": 1 }, + } }, + }); + assert_eq!(counted(&nodes), (vec![0], Some(1))); +} diff --git a/crates/tinyflows/src/nodes/mod.rs b/crates/tinyflows/src/nodes/mod.rs index cec94984..a243faf9 100644 --- a/crates/tinyflows/src/nodes/mod.rs +++ b/crates/tinyflows/src/nodes/mod.rs @@ -104,6 +104,12 @@ pub struct LaneContext { /// How many lanes the fan-out created, which is what a collector counts /// arrivals against. pub count: usize, + /// The super-step of the fan-out activation that opened this lane, which + /// tells this pass's lanes apart from an earlier pass's in a loop. + /// + /// `None` for a lane scheduled by an envelope that predates the field (a + /// checkpoint written by an older binary). + pub scatter_step: Option, } /// Builds the expression scope for a node from its runtime [`NodeContext`]. diff --git a/crates/tinyflows/src/validate/scatter.rs b/crates/tinyflows/src/validate/scatter.rs index a770cb4c..273198ca 100644 --- a/crates/tinyflows/src/validate/scatter.rs +++ b/crates/tinyflows/src/validate/scatter.rs @@ -29,7 +29,10 @@ pub(super) fn validate_scatter_regions(graph: &WorkflowGraph, errors: &mut Vec Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: vec![], + position: None, + } +} + +fn edge(from: &str, to: &str) -> Edge { + port_edge(from, "main", to) +} + +fn port_edge(from: &str, port: &str, to: &str) -> Edge { + Edge { + from_node: from.to_string(), + from_port: port.to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +/// Returns its arguments, so each lane's output is its candidate. +struct Echo; + +#[async_trait::async_trait] +impl ToolInvoker for Echo { + async fn invoke( + &self, + _slug: &str, + args: Value, + _conn: Option<&str>, + ) -> tinyflows::error::Result { + Ok(args) + } +} + +/// A refinement loop whose body scatters, with `candidates` proposed per pass. +/// +/// `candidates` is the expression `propose` sets per pass; each candidate is +/// evaluated in its own lane, gathered, and folded into the loop state. +fn refinement_loop(candidates: &str, loop_config: Value) -> WorkflowGraph { + WorkflowGraph { + name: "refinement_loop_scatter".to_string(), + nodes: vec![ + node( + "t", + NodeKind::Trigger, + json!({ "recursion_limit": 200, "max_node_visits": 100 }), + ), + node("refine", NodeKind::Loop, loop_config), + node( + "propose", + NodeKind::Transform, + json!({ "set": { "candidates": candidates } }), + ), + node( + "scatter", + NodeKind::Scatter, + json!({ "path": "candidates" }), + ), + node( + "evaluate", + NodeKind::ToolCall, + json!({ "slug": "refine.evaluate", "args": "=item" }), + ), + node( + "gather", + NodeKind::Gather, + json!({ "from": ["evaluate"], "poll_interval_ms": 1 }), + ), + node("out", NodeKind::OutputParser, Value::Null), + ], + edges: vec![ + edge("t", "refine"), + port_edge("refine", "body", "propose"), + edge("propose", "scatter"), + edge("scatter", "evaluate"), + edge("evaluate", "gather"), + edge("gather", "refine"), + port_edge("refine", "done", "out"), + ], + ..Default::default() + } +} + +fn gathered_candidates(output: &Value) -> Vec { + output["nodes"]["gather"]["items"] + .as_array() + .expect("gather items") + .iter() + .filter_map(|item| item["json"]["json"]["c"].as_i64()) + .collect() +} + +/// A narrower second pass must not gather the first pass's extra lane. +/// +/// Lane slots are never cleared between passes, so after pass 2 (lanes +/// `scatter#0`, `scatter#1`) `evaluate`'s `scatter#2` slot from pass 1 is still +/// in state. The gather counts only lanes stamped with the scatter's current +/// activation, so it releases `[20, 21]` rather than `[20, 21, 12]`. +#[tokio::test] +async fn a_narrower_pass_does_not_gather_stale_lanes() { + let graph = refinement_loop( + "=if .nodes.refine.iteration == 1 then [{c: 10}, {c: 11}, {c: 12}] \ + else [{c: 20}, {c: 21}] end", + json!({ "max_iterations": 2, "on_exceeded": "continue" }), + ); + let compiled = compile(&graph).expect("compile"); + let mut caps = mock_capabilities(); + caps.tools = Arc::new(Echo); + let outcome = tokio::time::timeout(GUARD, tinyflows::engine::run(&compiled, json!({}), &caps)) + .await + .expect("refinement loop hung") + .expect("refinement loop runs"); + assert_eq!(gathered_candidates(&outcome.output), vec![20, 21]); + assert_eq!( + outcome.output["nodes"]["evaluate"]["lanes"] + .as_object() + .map(serde_json::Map::len), + Some(3), + "the stale pass-1 slot is still in state; only the gather ignores it" + ); +} diff --git a/crates/tinyflows/tests/scatter_gather_e2e.rs b/crates/tinyflows/tests/scatter_gather_e2e.rs index eb4769ba..3b53c162 100644 --- a/crates/tinyflows/tests/scatter_gather_e2e.rs +++ b/crates/tinyflows/tests/scatter_gather_e2e.rs @@ -365,8 +365,8 @@ async fn a_scatter_without_a_gather_is_refused() { ); } -/// A gather with no scatter upstream is refused: it would wait on lanes nobody -/// opens until its poll budget ran out. +/// A gather with no scatter upstream is refused: no lane ever reaches it, so it +/// would release at once with nothing and drop its predecessors' output. #[tokio::test] async fn a_gather_without_a_scatter_is_refused() { let graph = WorkflowGraph {