fix(engine): a lane activation never writes the node's top-level slot

CHANGELOG.md

@@ -7,6 +7,34 @@ 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<usize>`); 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.
+
+### 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.<id>` 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

crates/tinyflows/src/engine.rs

@@ -267,9 +267,10 @@ 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,
+    MergeReducer, cancelled_result, collect_input, collect_input_since, lane_context,
+    lane_envelope, lane_input, lane_items_update, merge, stamp_activation_step,
 };
 
 mod routing;

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

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.<id>`. Only a non-lane activation is step-stamped.
+        return cancelled_result(&node.id, lane.as_ref(), Some(ctx.step));
     }
 
     if is_trigger {

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))
                 }
             })

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.
@@ -118,8 +147,14 @@ where
                     let mut routed: Vec<RouteTarget> = 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,

crates/tinyflows/src/engine/state.rs

@@ -96,19 +96,60 @@ 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 } } } }))
 }
 
+/// 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<usize>,
+) -> crate::graph::Result<NodeResult<Value>> {
+    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
+/// lane; every node along the lane passes it on unchanged.
 pub(super) fn lane_envelope(
     origin: &str,
     index: usize,
     count: usize,
+    scatter_step: Option<usize>,
     items: &[Item],
 ) -> crate::graph::Result<Value> {
     Ok(json!({
@@ -117,6 +158,7 @@ pub(super) fn lane_envelope(
             "origin": origin,
             "index": index,
             "count": count,
+            "scatter_step": scatter_step,
         },
         "items": serde_json::to_value(items)?,
     }))
@@ -147,6 +189,15 @@ pub(super) fn lane_input(send_arg: Option<&Value>) -> Vec<Item> {
 /// 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 (`<scatter>#<index>`)
+/// 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 +211,12 @@ pub(super) fn lane_context(send_arg: Option<&Value>) -> Option<crate::nodes::Lan
         origin: lane.get("origin")?.as_str()?.to_string(),
         index: usize::try_from(lane.get("index")?.as_u64()?).ok()?,
         count: usize::try_from(lane.get("count")?.as_u64()?).ok()?,
+        // Optional: an envelope from an older binary has none, and must still
+        // decode as a lane.
+        scatter_step: lane
+            .get("scatter_step")
+            .and_then(Value::as_u64)
+            .and_then(|n| usize::try_from(n).ok()),
     })
 }
 

crates/tinyflows/src/lane_context_tests.rs

@@ -14,6 +14,26 @@ fn a_well_formed_envelope_decodes() {
     assert_eq!(lane.origin, "fan");
     assert_eq!(lane.index, 2);
     assert_eq!(lane.count, 5);
+    assert_eq!(
+        lane.scatter_step, None,
+        "an envelope from before the stamp still decodes, unstamped"
+    );
+}
+
+/// The scatter's activation step survives the envelope and lands on the lane
+/// slot, where a node's own meta cannot overwrite it.
+#[test]
+fn the_scatter_step_travels_from_envelope_to_lane_slot() {
+    let packet = lane_envelope("fan", 1, 2, Some(7), &[]).expect("envelope");
+    let lane = lane_context(Some(&packet)).expect("decodes");
+    assert_eq!(lane.scatter_step, Some(7));
+
+    let forged = json!({ "_scatter_step": 1, "_scatter": "other" });
+    let meta = Some(&forged);
+    let update = lane_items_update("work", &lane, &[], None, "ok", meta).expect("update");
+    let slot = &update["nodes"]["work"]["lanes"]["fan#1"];
+    assert_eq!(slot["_scatter_step"], 7);
+    assert_eq!(slot["_scatter"], "fan");
 }
 
 /// The ordinary case, and the only one that occurs until a fan-out node