fix(engine): report the same run error whatever order parallel failures finish in

README.md

@@ -67,7 +67,10 @@ Rust 2024 · MSRV 1.85 · `#![forbid(unsafe_code)]` · GPL-3.0-or-later.
 
 - Per-node error handling: `on_error` policy (`stop` / `continue` / `route`),
   bounded `retry`, and an `error` output port for routing failures to a recovery
-  sub-graph.
+  sub-graph. When several parallel branches fail under `stop` in one super-step,
+  the run reports the error of the branch earliest in that step's active set
+  (for a fan-out, its edge order), never the first to finish; an earlier
+  super-step's failure always wins.
 - Node postconditions: `config.postcondition = { "require": "non_empty" }` (or
   `field_present` / `non_empty_list` with a dotted `field`) is checked against
   every emitted item, so an empty or misshapen output is a failed attempt that

crates/tinyflows/src/engine.rs

@@ -295,6 +295,9 @@ use build::build_graph;
 mod run_config;
 use run_config::RunConfig;
 
+mod terminal_error;
+use terminal_error::TerminalSlot;
+
 mod run_state;
 pub use run_state::resume;
 use run_state::{build_and_run, default_thread_id, merge_approvals};

crates/tinyflows/src/engine/build.rs

@@ -30,7 +30,7 @@ pub(super) fn build_graph(
     capabilities: &Capabilities,
     observer: &Arc<dyn RunObserver>,
     steps: &Arc<Mutex<Vec<ExecutionStep>>>,
-    terminal_error: &Arc<Mutex<Option<EngineError>>>,
+    terminal_error: &TerminalSlot,
     config: &RunConfig,
 ) -> Result<(CompiledGraph<Value, Value>, String)> {
     let checkpointer = config.checkpointer.clone();

crates/tinyflows/src/engine/build/activation.rs

@@ -10,7 +10,7 @@ pub(super) struct HandlerData {
     pub(super) agents: Arc<Vec<crate::model::AgentDefinition>>,
     pub(super) observer: Arc<dyn RunObserver>,
     pub(super) steps: Arc<Mutex<Vec<ExecutionStep>>>,
-    pub(super) terminal_error: Arc<Mutex<Option<EngineError>>>,
+    pub(super) terminal_error: TerminalSlot,
     pub(super) token: CancellationToken,
     pub(super) routing: HandlerRouting,
     pub(super) plain_targets: Vec<String>,
@@ -61,6 +61,8 @@ impl HandlerData {
         let lane = lane_context(ctx.send_arg.as_ref());
         let lane_send_arg = lane.is_some().then(|| ctx.send_arg.clone()).flatten();
         let activation_step = ctx.step;
+        // A sequential step has no fork; it stops at its first failure.
+        let branch = ctx.fork.as_ref().map_or(0, |fork| fork.branch_index);
         // A checkpointed resume (see `ResumableRun::resume`) delivers a resume
         // value to the interrupted node via `NodeContext::resume`. A resume
         // approves *this* gate only when it is a bare `true` (backward-compat,
@@ -444,6 +446,7 @@ impl HandlerData {
             &lane,
             &resume_value,
             activation_step,
+            branch,
             &steps,
             &terminal_error,
             &plain_targets,

crates/tinyflows/src/engine/build/handlers.rs

@@ -7,7 +7,7 @@ pub(super) fn register_handlers(
     capabilities: &Capabilities,
     observer: &Arc<dyn RunObserver>,
     steps: &Arc<Mutex<Vec<ExecutionStep>>>,
-    terminal_error: &Arc<Mutex<Option<EngineError>>>,
+    terminal_error: &TerminalSlot,
     token: &CancellationToken,
     node_timeout: Option<std::time::Duration>,
     loop_edges: &std::collections::HashSet<(String, String)>,

crates/tinyflows/src/engine/build/outcome.rs

@@ -22,8 +22,11 @@ pub(super) async fn finish_execution<F>(
     lane: &Option<crate::nodes::LaneContext>,
     resume_value: &Option<Value>,
     activation_step: usize,
+    // This activation's index in its superstep's active set; picks the
+    // reported error when several branches fail together.
+    branch: usize,
     steps: &Arc<Mutex<Vec<ExecutionStep>>>,
-    terminal_error: &Arc<Mutex<Option<EngineError>>>,
+    terminal_error: &TerminalSlot,
     plain_targets: &[String],
     emit: F,
 ) -> crate::graph::Result<NodeResult<Value>>
@@ -269,9 +272,9 @@ where
                 // into `EngineError::Capability` and a host could not
                 // tell a loop that ran away from an HTTP call that
                 // 500'd. `build_and_run` prefers whatever lands here.
-                // First writer wins: the failure that ends the run is
-                // the one worth reporting, and a parallel superstep
-                // may produce several.
+                // A parallel superstep may produce several; the rule in
+                // `terminal_error::offer` keeps the lowest-ordered one, not
+                // the first to finish.
                 _ => {
                     let message = err.to_string();
                     // A lane failure belongs to its gather, whose
@@ -286,9 +289,7 @@ where
                     let mut slot = terminal_error
                         .lock()
                         .expect("terminal error mutex poisoned");
-                    if slot.is_none() {
-                        *slot = Some(err);
-                    }
+                    crate::engine::terminal_error::offer(&mut slot, activation_step, branch, err);
                     drop(slot);
                     // The message still travels the string channel
                     // unchanged, so every path that only sees the

crates/tinyflows/src/engine/resumable/continuation.rs

@@ -55,7 +55,7 @@ async fn resume_with_checkpointer_inner(
     // `resume` loads the state persisted under `thread_id`. Node handlers fire
     // `observer.on_step_finish` for every node that runs after the interrupt
     // boundary, so a host observer sees the resumed steps live.
-    let terminal_error: Arc<Mutex<Option<EngineError>>> = Arc::new(Mutex::new(None));
+    let terminal_error: TerminalSlot = Arc::new(Mutex::new(None));
     let mut config = RunConfig::new(workflow)?.with_checkpointer(checkpointer, thread_id);
     if let Some(token) = &token {
         config = config.with_token(token.clone());
@@ -79,7 +79,8 @@ async fn resume_with_checkpointer_inner(
             let structured = terminal_error
                 .lock()
                 .expect("terminal error mutex poisoned")
-                .take();
+                .take()
+                .map(|held| held.error);
             return Err(structured.unwrap_or_else(|| EngineError::Capability(error.to_string())));
         }
     };

crates/tinyflows/src/engine/run_state.rs

@@ -105,7 +105,7 @@ pub(super) async fn build_and_run(
     // Where a run-ending node failure leaves its structured error on the way
     // out (see the `stop`-policy branch in `build_graph`), so the caller gets
     // the real `EngineError` instead of its `Display` string.
-    let terminal_error: Arc<Mutex<Option<EngineError>>> = Arc::new(Mutex::new(None));
+    let terminal_error: TerminalSlot = Arc::new(Mutex::new(None));
 
     // Build and compile the graph under the host-supplied checkpointer;
     // `trigger_id` is the graph's entry node.
@@ -205,7 +205,8 @@ pub(super) async fn build_and_run(
             let structured = terminal_error
                 .lock()
                 .expect("terminal error mutex poisoned")
-                .take();
+                .take()
+                .map(|held| held.error);
             return Err(structured.unwrap_or_else(|| EngineError::Capability(e.to_string())));
         }
     };

crates/tinyflows/src/engine/terminal_error.rs

@@ -0,0 +1,53 @@
+//! The structured error a failed run reports, chosen independently of timing.
+//!
+//! A `stop`-policy node failure ends the run, but the runtime can only carry a
+//! string out of a node handler, so the handler also leaves its real
+//! [`EngineError`] in a shared slot for the entry point to return. A parallel
+//! superstep can fail on several branches at once, and they reach the slot in
+//! completion order — so the slot must not simply keep whichever came first.
+
+use std::sync::{Arc, Mutex};
+
+use crate::error::EngineError;
+
+/// A run-ending node failure, with the position it was raised at.
+pub(crate) struct TerminalError {
+    step: usize,
+    branch: usize,
+    pub(crate) error: EngineError,
+}
+
+/// The slot every node handler of one run shares.
+pub(crate) type TerminalSlot = Arc<Mutex<Option<TerminalError>>>;
+
+/// Offers `error`, raised at superstep `step` by the activation at index
+/// `branch` of that superstep's active set, as the run's reported error.
+///
+/// The rule: the earliest superstep wins, and within one superstep the lowest
+/// active-set index wins. That index is the order the runtime folds parallel
+/// branches in, and the order it already uses to name a failed step, so the
+/// structured error and the runtime's own step error agree. Completion order
+/// never matters: an earlier superstep's error is never replaced by a later
+/// one, whatever its index. A sequential step has no fork and reports index 0;
+/// it stops at its first failure, so it offers at most one error per step.
+pub(crate) fn offer(
+    slot: &mut Option<TerminalError>,
+    step: usize,
+    branch: usize,
+    error: EngineError,
+) {
+    if slot
+        .as_ref()
+        .is_none_or(|held| (step, branch) < (held.step, held.branch))
+    {
+        *slot = Some(TerminalError {
+            step,
+            branch,
+            error,
+        });
+    }
+}
+
+#[cfg(test)]
+#[path = "terminal_error_tests.rs"]
+mod tests;

crates/tinyflows/src/engine/terminal_error_tests.rs

@@ -0,0 +1,44 @@
+//! Unit tests for choosing a run's reported error among several failures.
+
+use super::*;
+
+fn failure(name: &str) -> EngineError {
+    EngineError::Capability(name.to_string())
+}
+
+fn reported(slot: &Option<TerminalError>) -> Option<String> {
+    slot.as_ref().map(|held| held.error.to_string())
+}
+
+fn expected(name: &str) -> Option<String> {
+    Some(failure(name).to_string())
+}
+
+#[test]
+fn the_lowest_branch_wins_whichever_arrives_first() {
+    let mut lowest_first = None;
+    offer(&mut lowest_first, 2, 0, failure("branch 0"));
+    offer(&mut lowest_first, 2, 1, failure("branch 1"));
+
+    let mut lowest_last = None;
+    offer(&mut lowest_last, 2, 1, failure("branch 1"));
+    offer(&mut lowest_last, 2, 0, failure("branch 0"));
+
+    assert_eq!(reported(&lowest_first), expected("branch 0"));
+    assert_eq!(reported(&lowest_last), expected("branch 0"));
+}
+
+#[test]
+fn an_earlier_superstep_is_not_replaced_by_a_lower_branch_later() {
+    let mut slot = None;
+    offer(&mut slot, 1, 3, failure("step 1"));
+    offer(&mut slot, 2, 0, failure("step 2"));
+    assert_eq!(reported(&slot), expected("step 1"));
+}
+
+#[test]
+fn a_single_failure_is_reported_as_is() {
+    let mut slot = None;
+    offer(&mut slot, 4, 2, failure("only"));
+    assert_eq!(reported(&slot), expected("only"));
+}

crates/tinyflows/tests/parallel_failure_order_tests.rs

@@ -0,0 +1,120 @@
+#![cfg(feature = "mock")]
+//! Which error a run reports when several parallel branches fail together.
+//!
+//! The rule: within one superstep the failing branch with the lowest
+//! active-set index names the run's error, whichever finished first. For a
+//! static fan-out that index follows the fan-out node's edge order.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use serde_json::{Value, json};
+use tinyflows::caps::mock::mock_capabilities;
+use tinyflows::compiler::compile;
+use tinyflows::engine::{CancellationToken, run_intercepted};
+use tinyflows::interception::{StepAction, StepFrame, StepInterceptor, StepPhase};
+use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph};
+use tinyflows::observability::{NoopObserver, RunObserver};
+
+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(),
+    }
+}
+
+fn runtime() -> tokio::runtime::Runtime {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("build runtime")
+}
+
+/// Fails `n1` and `n2` outright, after a per-node yield count.
+struct FailBoth(Vec<u64>);
+
+#[async_trait]
+impl StepInterceptor for FailBoth {
+    async fn intercept(&self, frame: StepFrame<'_>) -> StepAction {
+        let index = match frame.node.id.as_str() {
+            "n1" => 0,
+            "n2" => 1,
+            _ => return StepAction::Continue { state_patch: None },
+        };
+        if frame.phase != StepPhase::Before {
+            return StepAction::Continue { state_patch: None };
+        }
+        for _ in 0..self.0[index] {
+            tokio::task::yield_now().await;
+        }
+        StepAction::Fail {
+            message: format!("unhandled failure at {}", frame.node.id),
+        }
+    }
+}
+
+/// Two parallel nodes fail without a handler, under opposite latencies. `n1`
+/// is declared first, so it holds the lower active-set index and its error is
+/// the one reported — also when `n2` finishes first.
+#[test]
+fn parallel_failures_report_the_lowest_ordered_branch_under_any_latency() {
+    let graph = WorkflowGraph {
+        name: "parallel_failures".to_string(),
+        nodes: vec![
+            node("trigger", NodeKind::Trigger, Value::Null),
+            node("fan", NodeKind::OutputParser, Value::Null),
+            node("n1", NodeKind::OutputParser, Value::Null),
+            node("n2", NodeKind::OutputParser, Value::Null),
+            node("join", NodeKind::Merge, Value::Null),
+        ],
+        edges: vec![
+            edge("trigger", "fan"),
+            edge("fan", "n1"),
+            edge("fan", "n2"),
+            edge("n1", "join"),
+            edge("n2", "join"),
+        ],
+        ..Default::default()
+    };
+    let compiled = compile(&graph).expect("compile");
+
+    let error_under = |delays: Vec<u64>| {
+        runtime().block_on(async {
+            let observer = Arc::new(NoopObserver) as Arc<dyn RunObserver>;
+            run_intercepted(
+                &compiled,
+                json!({}),
+                &mock_capabilities(),
+                &observer,
+                CancellationToken::new(),
+                Arc::new(FailBoth(delays)),
+            )
+            .await
+            .map(|_| ())
+            .expect_err("both branches fail")
+            .to_string()
+        })
+    };
+    // `[10, 0]`: `n2` finishes first. `[0, 10]`: `n1` does.
+    for delays in [vec![10, 0], vec![0, 10]] {
+        let error = error_under(delays.clone());
+        assert!(
+            error.contains("unhandled failure at n1"),
+            "delays {delays:?} reported {error:?}"
+        );
+    }
+}