From 96d24e943b518ccca997ca961c5e25e716534e7e Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 11 Nov 2026 17:16:24 +0530 Subject: [PATCH 1/2] 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 { From f7f71f3b7e050d6e86a5db69c1e382e4b04411e6 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 11 Nov 2026 17:16:24 +0530 Subject: [PATCH 2/2] fix(engine): a lane activation never writes the node's top-level slot Four paths in a scatter lane wrote `nodes.` instead of the lane's slot: a cancel seen at the node boundary, a cancel during retry or backoff, a cancel while waiting to re-enter, and a Reenter (a gate or approval polling), which also re-activated with a bare id and so left its lane. A cancelled lane also kept routing down its static edge, so its successor ran outside the lane. A cancelled lane now marks its lane slot "cancelled" and stops; a re-entering lane writes "pending" notes to its lane slot and re-activates with its own lane envelope. The gather counts neither as arrived, and gate and approval read their poll count from the lane's slot. --- CHANGELOG.md | 13 + crates/tinyflows/src/engine.rs | 4 +- .../tinyflows/src/engine/build/activation.rs | 4 +- .../src/engine/build/activation/gates.rs | 8 +- crates/tinyflows/src/engine/build/outcome.rs | 49 ++- crates/tinyflows/src/engine/state.rs | 30 ++ .../src/nodes/control_flow/gather.rs | 9 + .../src/nodes/integration/approval.rs | 3 +- .../tinyflows/src/nodes/integration/gate.rs | 6 +- crates/tinyflows/src/nodes/mod.rs | 18 + .../tests/lane_top_level_writes_tests.rs | 412 ++++++++++++++++++ 11 files changed, 531 insertions(+), 25 deletions(-) create mode 100644 crates/tinyflows/tests/lane_top_level_writes_tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 10495a94..18c81065 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 releases with no items, silently dropping its predecessors' output, rather than waiting out its poll budget. +### Lane activations never write the top-level slot + +- **Fixed** three paths where an activation inside a `scatter` lane wrote the + node's top-level `nodes.` slot instead of its lane slot: a cancel seen at + the node boundary, a cancel seen during retry or backoff, and a `Reenter` + (a `gate` or `approval` polling). A cancelled lane now marks its lane slot + `"cancelled"` and schedules nothing further; a re-entering lane writes its + notes to its lane slot as `"pending"` and re-activates with its own lane + envelope, so it stays in its lane. A `gather` counts neither as an arrival. +- **Fixed** `gate` and `approval` inside a lane reading their poll count (and a + gate its `from` spawn's tickets) from the top-level slot; they now read this + lane's slot, falling back to the top level. + ### 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 eb377b22..29528df4 100644 --- a/crates/tinyflows/src/engine.rs +++ b/crates/tinyflows/src/engine.rs @@ -269,8 +269,8 @@ pub(crate) use state::replace; 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, + MergeReducer, cancelled_result, collect_input, collect_input_since, lane_context, + lane_envelope, lane_input, lane_items_update, merge, stamp_activation_step, }; mod routing; diff --git a/crates/tinyflows/src/engine/build/activation.rs b/crates/tinyflows/src/engine/build/activation.rs index 77c317f9..f2b75510 100644 --- a/crates/tinyflows/src/engine/build/activation.rs +++ b/crates/tinyflows/src/engine/build/activation.rs @@ -246,7 +246,7 @@ impl HandlerData { // winds down fast and reports cancelled. if token.is_cancelled() { tracing::info!(node = %node.id, "run cancelled during retry; skipping remaining attempts"); - return Ok(NodeResult::Update(items_update(&node.id, &[], None)?)); + return cancelled_result(&node.id, lane.as_ref(), None); } let ctx = NodeContext { node: &node, @@ -338,7 +338,7 @@ impl HandlerData { while remaining > 0 { if token.is_cancelled() { tracing::info!(node = %node.id, "run cancelled during backoff; skipping remaining attempts"); - return Ok(NodeResult::Update(items_update(&node.id, &[], None)?)); + return cancelled_result(&node.id, lane.as_ref(), None); } let step = remaining.min(BACKOFF_POLL_MS); // Yields even when the timer has already fired, so a retry diff --git a/crates/tinyflows/src/engine/build/activation/gates.rs b/crates/tinyflows/src/engine/build/activation/gates.rs index de455e2d..ff72bc93 100644 --- a/crates/tinyflows/src/engine/build/activation/gates.rs +++ b/crates/tinyflows/src/engine/build/activation/gates.rs @@ -9,11 +9,9 @@ // starting further node work. The engine reports it as cancelled. if token.is_cancelled() { tracing::info!(node = %node.id, "run cancelled; skipping node work"); - let mut update = items_update(&node.id, &[], None)?; - if lane.is_none() { - stamp_activation_step(&mut update, &node.id, ctx.step); - } - return Ok(NodeResult::Update(update)); + // A lane marks its own lane slot cancelled and ends there; it never + // owns `nodes.`. Only a non-lane activation is step-stamped. + return cancelled_result(&node.id, lane.as_ref(), Some(ctx.step)); } if is_trigger { diff --git a/crates/tinyflows/src/engine/build/outcome.rs b/crates/tinyflows/src/engine/build/outcome.rs index e0d34d94..292b8c94 100644 --- a/crates/tinyflows/src/engine/build/outcome.rs +++ b/crates/tinyflows/src/engine/build/outcome.rs @@ -76,13 +76,28 @@ where // one node visit, so the caller's own bounded poll // count is what stops this — not the run-level // backstop, which cannot say which node span. + // + // Inside a lane the notes go to the lane slot, marked + // `"pending"` so a gather does not count a lane that is + // still waiting as arrived, and the re-activation carries + // the same lane envelope so it stays in its lane. Some(crate::nodes::NodeControl::Reenter { after_ms }) => { - let update = items_update_with_meta( - &node.id, - &output.items, - port, - output.meta.as_ref(), - )?; + let update = match lane.as_ref() { + Some(lane) => lane_items_update( + &node.id, + lane, + &output.items, + port, + "pending", + output.meta.as_ref(), + )?, + None => items_update_with_meta( + &node.id, + &output.items, + port, + output.meta.as_ref(), + )?, + }; if after_ms > 0 { // Chopped into slices so a cancel during a // long poll interval is seen promptly, the @@ -91,7 +106,7 @@ where while remaining > 0 { if token.is_cancelled() { tracing::info!(node = %node.id, "run cancelled while waiting to re-enter"); - return Ok(NodeResult::Update(items_update(&node.id, &[], None)?)); + return cancelled_result(&node.id, lane.as_ref(), None); } let slice = remaining.min(BACKOFF_POLL_MS); // Always yields, so the tasks this node is polling @@ -101,9 +116,23 @@ where remaining -= slice; } } - return Ok(NodeResult::Command( - Command::goto(vec![node.id.clone()]).with_update(update), - )); + let command = match lane.as_ref() { + // `input` is this lane's own items (read from its + // envelope), so the rebuilt envelope is the one this + // activation was scheduled with. + Some(lane) => { + let envelope = lane_envelope( + &lane.origin, + lane.index, + lane.count, + lane.scatter_step, + input, + )?; + Command::send(vec![crate::graph::Send::new(node.id.clone(), envelope)]) + } + None => Command::goto(vec![node.id.clone()]), + }; + return Ok(NodeResult::Command(command.with_update(update))); } // Open a lane per entry: schedule every successor // once for each, each carrying its own work. diff --git a/crates/tinyflows/src/engine/state.rs b/crates/tinyflows/src/engine/state.rs index e214f8db..47e3c72c 100644 --- a/crates/tinyflows/src/engine/state.rs +++ b/crates/tinyflows/src/engine/state.rs @@ -111,6 +111,36 @@ pub(super) fn lane_items_update( Ok(json!({ "nodes": { node_id: { "lanes": { lane.id.clone(): slot } } } })) } +/// What an activation that observed cancellation returns: an empty result in +/// the slot it owns, and no further work. +/// +/// An ordinary activation empties the node's top-level slot (stamped with +/// `step` when given) and follows its static edge, so each successor no-ops +/// in turn — exactly as it always has. +/// +/// A lane activation marks only its own lane slot `"cancelled"`, which a +/// gather does not count as an arrival, and routes to `END`. Following the +/// static edge instead would schedule the successor as a *plain* activation, +/// outside the lane, and that activation would write the successor's +/// top-level slot. +pub(super) fn cancelled_result( + node_id: &str, + lane: Option<&crate::nodes::LaneContext>, + step: Option, +) -> crate::graph::Result> { + let Some(lane) = lane else { + let mut update = items_update(node_id, &[], None)?; + if let Some(step) = step { + stamp_activation_step(&mut update, node_id, step); + } + return Ok(NodeResult::Update(update)); + }; + let update = lane_items_update(node_id, lane, &[], None, "cancelled", None)?; + Ok(NodeResult::Command( + Command::goto(vec![END.to_string()]).with_update(update), + )) +} + /// The lane envelope a fan-out schedules one activation with. /// /// `scatter_step` is the super-step of the scatter activation that opened the diff --git a/crates/tinyflows/src/nodes/control_flow/gather.rs b/crates/tinyflows/src/nodes/control_flow/gather.rs index cc88516e..598aa148 100644 --- a/crates/tinyflows/src/nodes/control_flow/gather.rs +++ b/crates/tinyflows/src/nodes/control_flow/gather.rs @@ -121,6 +121,15 @@ fn current_activation<'a>( fn arrivals(slots: Vec<&Value>) -> Vec { slots .into_iter() + // A lane whose terminal is still re-entering (a gate polling) has + // written `"pending"` notes, and a cancelled one never finished. + // Neither has arrived. + .filter(|slot| { + !matches!( + slot.get("status").and_then(Value::as_str), + Some("pending" | "cancelled") + ) + }) .map(|slot| { let index = slot .get("index") diff --git a/crates/tinyflows/src/nodes/integration/approval.rs b/crates/tinyflows/src/nodes/integration/approval.rs index 2ba2180a..d248feaa 100644 --- a/crates/tinyflows/src/nodes/integration/approval.rs +++ b/crates/tinyflows/src/nodes/integration/approval.rs @@ -270,8 +270,7 @@ impl ApprovalNode { } let polls = ctx - .nodes - .get(&ctx.node.id) + .slot(&ctx.node.id) .and_then(|slot| slot.get(POLLS_KEY)) .and_then(Value::as_u64) .unwrap_or(0); diff --git a/crates/tinyflows/src/nodes/integration/gate.rs b/crates/tinyflows/src/nodes/integration/gate.rs index c14a7bb0..3826e374 100644 --- a/crates/tinyflows/src/nodes/integration/gate.rs +++ b/crates/tinyflows/src/nodes/integration/gate.rs @@ -109,8 +109,7 @@ fn awaited(ctx: &NodeContext<'_>) -> Result> { if let Some(from) = ctx.node.config.get("from").and_then(Value::as_array) { for source in from.iter().filter_map(Value::as_str) { let items = ctx - .nodes - .get(source) + .slot(source) .and_then(|slot| slot.get("items")) .and_then(Value::as_array); let Some(items) = items else { @@ -190,8 +189,7 @@ impl NodeExecutor for GateNode { // Poll count carried in this node's own slot, so it survives a // checkpoint — the same reason a `loop` node keeps its iteration there. let polls = ctx - .nodes - .get(&ctx.node.id) + .slot(&ctx.node.id) .and_then(|slot| slot.get(POLLS_KEY)) .and_then(Value::as_u64) .unwrap_or(0); diff --git a/crates/tinyflows/src/nodes/mod.rs b/crates/tinyflows/src/nodes/mod.rs index a243faf9..bfbbf011 100644 --- a/crates/tinyflows/src/nodes/mod.rs +++ b/crates/tinyflows/src/nodes/mod.rs @@ -87,6 +87,24 @@ pub struct NodeContext<'a> { pub step: usize, } +impl NodeContext<'_> { + /// The run-state slot this activation reads for node `id`: inside a lane, + /// that node's slot for this same lane when it has one, else its top-level + /// slot. + /// + /// A lane activation writes only `nodes..lanes.`, so a node + /// that re-enters inside a lane (a `gate`, an `approval`) finds its own + /// poll count there — and a gate finds the tickets of a `spawn` in the + /// same lane there — rather than at the top level. + pub(crate) fn slot(&self, id: &str) -> Option<&Value> { + let top = self.nodes.get(id)?; + self.lane + .as_ref() + .and_then(|lane| top.get("lanes").and_then(|lanes| lanes.get(&lane.id))) + .or(Some(top)) + } +} + /// Which lane of a fan-out an activation belongs to. /// /// Travels with the activation itself rather than through the run state, so N diff --git a/crates/tinyflows/tests/lane_top_level_writes_tests.rs b/crates/tinyflows/tests/lane_top_level_writes_tests.rs new file mode 100644 index 00000000..d7745b65 --- /dev/null +++ b/crates/tinyflows/tests/lane_top_level_writes_tests.rs @@ -0,0 +1,412 @@ +#![cfg(feature = "mock")] +//! A lane activation writes only `nodes..lanes.`, never the node's +//! top-level `nodes.` slot — on every path out of an activation, not just +//! the success path. +//! +//! Three paths used to break it: a cancel seen at the node boundary, a cancel +//! seen during retry or backoff, and a `Reenter` (a gate polling). The merge +//! reducer never deletes a key, so a top-level `items` absent from the final +//! state was absent at every boundary; the re-entry test checks every +//! checkpoint anyway, because there the question is also *when* the gather +//! released. + +use std::collections::{BTreeSet, HashMap}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use serde_json::{Value, json}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::caps::{TaskRunner, TaskSpec, TaskState, ToolInvoker}; +use tinyflows::compiler::compile; +use tinyflows::engine::{ + CancellationToken, Checkpointer, InMemoryCheckpointer, run_cancellable, run_with_checkpointer, +}; +use tinyflows::error::EngineError; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; + +const GUARD: Duration = Duration::from_secs(10); + +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, + } +} + +fn edge(from: &str, to: &str) -> Edge { + Edge { + from_node: from.to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +/// `trigger -> scatter -> body[0] -> … -> gather`, the gather collecting the +/// last body node. +fn lane_graph(body: Vec) -> WorkflowGraph { + let last = body.last().map(|n| n.id.clone()).unwrap_or_default(); + let mut nodes = vec![ + node( + "trigger", + NodeKind::Trigger, + json!({ "recursion_limit": 200 }), + ), + node("scatter", NodeKind::Scatter, json!({ "path": "rows" })), + ]; + let mut path = vec!["trigger".to_string(), "scatter".to_string()]; + path.extend(body.iter().map(|n| n.id.clone())); + path.push("gather".to_string()); + let edges = path + .windows(2) + .map(|pair| edge(&pair[0], &pair[1])) + .collect(); + nodes.extend(body); + nodes.push(node( + "gather", + NodeKind::Gather, + json!({ "from": [last], "poll_interval_ms": 1 }), + )); + WorkflowGraph { + name: "lane_top_level_writes".to_string(), + nodes, + edges, + ..Default::default() + } +} + +fn rows(count: usize) -> Value { + json!({ "rows": (0..count).map(|index| json!({ "index": index })).collect::>() }) +} + +/// Fails when any of `ids` has a top-level `items` — a write only a non-lane +/// activation may make, and every activation of these nodes is in a lane. +fn assert_no_top_level_items(state: &Value, ids: &[&str], when: &str) { + for id in ids { + assert!( + state["nodes"][id].get("items").is_none(), + "{when}: lane body node {id:?} has a top-level slot: {}", + state["nodes"][id] + ); + } +} + +fn lane_slots<'a>(state: &'a Value, id: &str) -> Vec<&'a Value> { + state["nodes"][id]["lanes"] + .as_object() + .map(|lanes| lanes.values().collect()) + .unwrap_or_default() +} + +fn statuses(state: &Value, id: &str) -> Vec { + lane_slots(state, id) + .iter() + .map(|slot| slot["status"].as_str().unwrap_or("").to_string()) + .collect() +} + +/// Starts work that never finishes and cancels the run on the first start. +struct CancellingRunner { + token: CancellationToken, + started: Mutex>, + cancelled: Mutex>, +} + +#[async_trait::async_trait] +impl TaskRunner for CancellingRunner { + async fn start(&self, _spec: TaskSpec) -> tinyflows::error::Result { + let mut started = self.started.lock().expect("started mutex poisoned"); + let ticket = format!("ticket-{}", started.len()); + started.push(ticket.clone()); + self.token.cancel(); + Ok(ticket) + } + + async fn poll(&self, _ticket: &str) -> tinyflows::error::Result { + Ok(TaskState::Running) + } + + async fn cancel(&self, ticket: &str) -> tinyflows::error::Result<()> { + self.cancelled + .lock() + .expect("cancelled mutex poisoned") + .push(ticket.to_string()); + Ok(()) + } +} + +/// (a) A cancel seen at a node boundary inside a lane. +/// +/// The first `spawn` cancels the run, so every `after` activation — each one +/// scheduled into its lane by a spawn that did start — no-ops at its boundary. +#[tokio::test] +async fn boundary_cancel_inside_a_lane_writes_only_the_lane_slot() { + let graph = lane_graph(vec![ + node( + "spawn", + NodeKind::Spawn, + json!({ "target": "tool", "slug": "lane.task" }), + ), + node( + "after", + NodeKind::Transform, + json!({ "set": { "seen": true } }), + ), + ]); + let compiled = compile(&graph).expect("compile"); + let token = CancellationToken::new(); + let runner = Arc::new(CancellingRunner { + token: token.clone(), + started: Mutex::new(Vec::new()), + cancelled: Mutex::new(Vec::new()), + }); + let mut caps = mock_capabilities(); + caps.tasks = Some(runner.clone()); + + let run = run_cancellable(&compiled, rows(2), &caps, token); + let outcome = tokio::time::timeout(GUARD, run) + .await + .expect("a cancelled lane run must settle") + .expect("cancellation returns a partial outcome"); + + assert!(outcome.cancelled); + let state = &outcome.output; + assert_no_top_level_items(state, &["spawn", "after"], "final state"); + let after = statuses(state, "after"); + assert!( + !after.is_empty() && after.iter().all(|status| status == "cancelled"), + "every lane that reached `after` was cancelled at its boundary: {after:?}" + ); + assert!( + lane_slots(state, "after") + .iter() + .all(|slot| slot["items"] == json!([])), + "a cancelled lane carries no items" + ); + let spawn = statuses(state, "spawn"); + assert!( + spawn.iter().all(|s| s == "ok" || s == "cancelled"), + "{spawn:?}" + ); + + // The cleanup still finds every ticket, now that cancelled lanes write + // their lane slot instead of the top level. + let started: BTreeSet = runner + .started + .lock() + .expect("started mutex poisoned") + .iter() + .cloned() + .collect(); + let cancelled: BTreeSet = runner + .cancelled + .lock() + .expect("cancelled mutex poisoned") + .iter() + .cloned() + .collect(); + assert!(!started.is_empty()); + assert_eq!(cancelled, started); +} + +/// Fails every call, cancelling the run as it does. +struct CancelOnCall { + token: CancellationToken, + calls: AtomicUsize, +} + +#[async_trait::async_trait] +impl ToolInvoker for CancelOnCall { + async fn invoke( + &self, + _slug: &str, + _args: Value, + _conn: Option<&str>, + ) -> tinyflows::error::Result { + self.calls.fetch_add(1, Ordering::SeqCst); + self.token.cancel(); + Err(EngineError::Capability("transient failure".to_string())) + } +} + +/// (b) A cancel seen during retry (`backoff_ms: 0`, the check before the next +/// attempt) or during backoff (`backoff_ms > 0`, the check inside the wait). +/// Either way the first attempt has already failed and cancelled the run, so +/// no time is spent waiting. +#[tokio::test] +async fn cancel_during_retry_or_backoff_inside_a_lane_writes_only_the_lane_slot() { + for backoff_ms in [0_u64, 40] { + let graph = lane_graph(vec![ + node( + "call", + NodeKind::ToolCall, + json!({ + "slug": "lane.flaky", + "retry": { "max_attempts": 3, "backoff_ms": backoff_ms }, + }), + ), + node( + "after", + NodeKind::Transform, + json!({ "set": { "seen": true } }), + ), + ]); + let compiled = compile(&graph).expect("compile"); + let token = CancellationToken::new(); + let tools = Arc::new(CancelOnCall { + token: token.clone(), + calls: AtomicUsize::new(0), + }); + let mut caps = mock_capabilities(); + caps.tools = tools.clone(); + + let run = run_cancellable(&compiled, rows(2), &caps, token); + let outcome = tokio::time::timeout(GUARD, run) + .await + .expect("a cancelled lane run must settle") + .expect("cancellation returns a partial outcome"); + + let when = format!("backoff_ms {backoff_ms}"); + assert!(outcome.cancelled, "{when}"); + assert!( + tools.calls.load(Ordering::SeqCst) >= 1, + "{when}: no attempt ran" + ); + let state = &outcome.output; + assert_no_top_level_items(state, &["call", "after"], &when); + // Both lanes end cancelled — the one whose attempt failed, at its retry + // check, and any other at its boundary. None is "failed": the retry + // budget was never spent. + assert_eq!( + statuses(state, "call"), + vec!["cancelled".to_string(); 2], + "{when}" + ); + assert!( + state["nodes"]["after"].is_null(), + "{when}: a cancelled lane schedules nothing: {}", + state["nodes"]["after"] + ); + } +} + +/// Starts tickets in order; `ticket-0` reports `Running` for its first two +/// polls, every other ticket settles on its first. Each result names its +/// ticket, which is how the test reads lane order back. +#[derive(Default)] +struct SlowFirstRunner { + started: Mutex, + polls: Mutex>, +} + +#[async_trait::async_trait] +impl TaskRunner for SlowFirstRunner { + async fn start(&self, _spec: TaskSpec) -> tinyflows::error::Result { + let mut started = self.started.lock().expect("started mutex poisoned"); + let ticket = format!("ticket-{started}"); + *started += 1; + Ok(ticket) + } + + async fn poll(&self, ticket: &str) -> tinyflows::error::Result { + let mut polls = self.polls.lock().expect("polls mutex poisoned"); + let count = polls.entry(ticket.to_string()).or_default(); + *count += 1; + Ok(if ticket == "ticket-0" && *count < 3 { + TaskState::Running + } else { + TaskState::Done(json!({ "ticket": ticket })) + }) + } + + async fn cancel(&self, _ticket: &str) -> tinyflows::error::Result<()> { + Ok(()) + } +} + +/// (c) A gate polling (`Reenter`) inside a lane body. +/// +/// One lane's gate re-enters twice while the other settles at once. The +/// re-entry must stay in its lane, write only its lane slot, and not count as +/// an arrival — so the gather releases once, with both lanes, in lane order. +/// Run for both ways a gate names its tickets. +#[tokio::test] +async fn reenter_inside_a_lane_stays_in_the_lane_and_gathers_in_lane_order() { + let expected_ok = vec!["ok".to_string(); 2]; + for gate in [ + json!({ "tickets": "=item.ticket", "poll_interval_ms": 1, "max_polls": 20 }), + json!({ "from": ["spawn"], "poll_interval_ms": 1, "max_polls": 20 }), + ] { + let when = format!("gate {gate}"); + let graph = lane_graph(vec![ + node( + "spawn", + NodeKind::Spawn, + json!({ "target": "tool", "slug": "lane.task" }), + ), + node("gate", NodeKind::Gate, gate), + ]); + let compiled = compile(&graph).expect("compile"); + let mut caps = mock_capabilities(); + caps.tasks = Some(Arc::new(SlowFirstRunner::default())); + let checkpointer = Arc::new(InMemoryCheckpointer::::new()); + let thread = "lane-reenter"; + + let run = run_with_checkpointer(&compiled, rows(2), &caps, checkpointer.clone(), thread); + let outcome = tokio::time::timeout(GUARD, run) + .await + .expect("the lane run must settle") + .expect("the lane run succeeds"); + + let state = &outcome.output; + assert_no_top_level_items(state, &["spawn", "gate"], &when); + assert_eq!(statuses(state, "gate"), expected_ok, "{when}"); + let max_polls = lane_slots(state, "gate") + .iter() + .filter_map(|slot| slot["polls"].as_u64()) + .max(); + assert_eq!( + max_polls, + Some(3), + "{when}: the slow lane's gate re-entered in its lane" + ); + + // Lane order, not completion order: the slow lane may be either index. + let gathered = state["nodes"]["gather"]["items"] + .as_array() + .expect("gather output"); + assert_eq!(gathered.len(), 2, "{when}: {gathered:?}"); + for (index, item) in gathered.iter().enumerate() { + let lane = format!("scatter#{index}"); + let spawned = &state["nodes"]["spawn"]["lanes"][lane.as_str()]; + let ticket = &spawned["items"][0]["json"]["ticket"]; + assert!(ticket.is_string(), "{when}: lane {index} spawned no ticket"); + assert_eq!(&item["json"]["ticket"], ticket, "{when}: lane {index}"); + } + + // Every boundary: no top-level write, and the gather never released + // while a lane was still re-entering. + let history = checkpointer.get_thread(thread).await.expect("checkpoints"); + assert!(!history.is_empty(), "{when}: no checkpoints were written"); + let mut saw_pending = false; + for checkpoint in &history { + let at = format!("{when}, checkpoint {}", checkpoint.checkpoint_id); + assert_no_top_level_items(&checkpoint.state, &["spawn", "gate"], &at); + let gate = statuses(&checkpoint.state, "gate"); + saw_pending |= gate.iter().any(|status| status == "pending"); + let released = checkpoint.state["nodes"]["gather"]["items"] + .as_array() + .is_some_and(|items| !items.is_empty()); + if released { + assert_eq!(gate, expected_ok, "{at}: released early"); + } + } + assert!(saw_pending, "{when}: no pending boundary"); + } +}