From 54a97a6159658c214473f4aa8f5018f6ffca375c Mon Sep 17 00:00:00 2001 From: oxoxDev Date: Wed, 12 Aug 2026 23:22:29 +0530 Subject: [PATCH 1/3] feat(observability): signal a non-clean completion via RunStatus::CompletedWithErrors --- src/observability.rs | 30 +++++++++++++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/src/observability.rs b/src/observability.rs index 9b85ccf..1801d17 100644 --- a/src/observability.rs +++ b/src/observability.rs @@ -46,8 +46,18 @@ use serde_json::Value; pub enum RunStatus { /// The run is still executing (not yet driven to completion). Running, - /// The run reached a terminal node and completed successfully. + /// The run reached a terminal node and every node completed cleanly. Completed, + /// The run reached a terminal node, but at least one node failed under a + /// non-`stop` error policy (`continue` or `route`): its failure was turned + /// into data and the run proceeded, so the run did not [`Failed`], yet it + /// did not complete cleanly either. Without this a `continue`/`route` + /// failure is invisible in the terminal status and a host reads success + /// while a node failed (#661 L1). The failing nodes are exactly the + /// [`Run::steps`] whose [`StepStatus`] is [`StepStatus::Error`]. + /// + /// [`Failed`]: RunStatus::Failed + CompletedWithErrors, /// The run ended because a node failed under a `stop` error policy. Failed, } @@ -98,6 +108,24 @@ pub struct Run { pub steps: Vec, } +impl Run { + /// The ids of the nodes that errored in this run — the [`steps`](Run::steps) + /// whose [`StepStatus`] is [`StepStatus::Error`], in the order they finished. + /// + /// Empty for a clean [`RunStatus::Completed`]; non-empty exactly when the + /// status is [`RunStatus::CompletedWithErrors`] (nodes handled by + /// `continue`/`route`) or [`RunStatus::Failed`] (the `stop`-policy node that + /// ended the run). Lets a host act on *which* nodes failed without scanning + /// every step itself. + pub fn failed_node_ids(&self) -> Vec<&str> { + self.steps + .iter() + .filter(|step| matches!(step.status, StepStatus::Error)) + .map(|step| step.node_id.as_str()) + .collect() + } +} + /// A host-implemented hook that receives run/step records as a run executes. /// /// Every method has a default no-op body, so a host overrides only the callbacks From ef20b5ce823eaa2b459ca49c5c25bb414ff0596d Mon Sep 17 00:00:00 2001 From: oxoxDev Date: Wed, 12 Aug 2026 23:22:29 +0530 Subject: [PATCH 2/3] feat(engine): mark a run completed-with-errors when a node fails under continue/route (#661 L1) --- src/engine.rs | 115 ++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 111 insertions(+), 4 deletions(-) diff --git a/src/engine.rs b/src/engine.rs index 224f47c..de41ca1 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1847,12 +1847,24 @@ async fn build_and_run( ); // Reaching here means the run settled without a `stop`-policy failure - // (those bubble out as `Err` above), so it Completed. Per-step Error status - // is recorded independently on nodes handled by `continue`/`route`. + // (those bubble out as `Err` above). A node handled by `continue`/`route` + // still records a per-step `Error` while the run proceeds, so the terminal + // status is derived from the steps rather than assumed clean: any error step + // makes this `CompletedWithErrors`, so a host that reads only `status` still + // learns a node failed instead of seeing an unqualified success (#661 L1). + let collected_steps = steps.lock().expect("steps mutex poisoned").clone(); + let status = if collected_steps + .iter() + .any(|step| matches!(step.status, StepStatus::Error)) + { + RunStatus::CompletedWithErrors + } else { + RunStatus::Completed + }; let run_record = Run { id: run_id, - status: RunStatus::Completed, - steps: steps.lock().expect("steps mutex poisoned").clone(), + status, + steps: collected_steps, }; observer.on_run_finish(&run_record); @@ -3788,6 +3800,101 @@ mod tests { ); } + /// Captures the terminal [`Run`] so a test can assert its status and which + /// nodes it names as failed (#661 L1). + #[derive(Default)] + struct StatusCapture { + status: Mutex>, + failed: Mutex>, + } + + impl RunObserver for StatusCapture { + fn on_run_finish(&self, run: &Run) { + *self.status.lock().unwrap() = Some(run.status.clone()); + *self.failed.lock().unwrap() = run + .failed_node_ids() + .iter() + .map(|id| id.to_string()) + .collect(); + } + } + + async fn observed_status(graph: &WorkflowGraph) -> (RunStatus, Vec) { + let compiled = compile(graph).expect("compile"); + let caps = mock_capabilities(); + let capture = Arc::new(StatusCapture::default()); + let observer: Arc = capture.clone(); + run_with_observer(&compiled, json!({}), &caps, &observer) + .await + .expect("run"); + let status = capture + .status + .lock() + .unwrap() + .clone() + .expect("run finished"); + let failed = capture.failed.lock().unwrap().clone(); + (status, failed) + } + + #[tokio::test] + async fn a_clean_run_is_completed_and_names_no_failed_node() { + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + node("a", NodeKind::OutputParser), + ], + edges: vec![edge("t", "a")], + ..Default::default() + }; + let (status, failed) = observed_status(&graph).await; + assert_eq!(status, RunStatus::Completed); + assert!( + failed.is_empty(), + "a clean run names no failed node: {failed:?}" + ); + } + + #[tokio::test] + async fn on_error_continue_marks_the_run_completed_with_errors() { + // #661 L1: a node that fails under `continue` used to leave the run + // reporting an unqualified `Completed`, so a host read success while a + // node failed. Now the terminal status says so, and names the node. + let mut tool = node("x", NodeKind::ToolCall); + tool.config = json!({ "on_error": "continue" }); + let graph = WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), tool], + edges: vec![edge("t", "x")], + ..Default::default() + }; + let (status, failed) = observed_status(&graph).await; + assert_eq!(status, RunStatus::CompletedWithErrors); + assert_eq!(failed, vec!["x".to_string()], "the failing node is named"); + } + + #[tokio::test] + async fn on_error_route_marks_the_run_completed_with_errors() { + // A routed failure also failed the node; the recovery branch handling it + // downstream does not erase that the node itself errored. + let mut tool = node("x", NodeKind::ToolCall); + tool.config = json!({ "on_error": "route" }); + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + tool, + node("recover", NodeKind::OutputParser), + ], + edges: vec![edge("t", "x"), port_edge("x", "error", "recover")], + ..Default::default() + }; + let (status, failed) = observed_status(&graph).await; + assert_eq!(status, RunStatus::CompletedWithErrors); + assert!( + failed.contains(&"x".to_string()), + "the routed failure names its node: {failed:?}" + ); + } + #[tokio::test] async fn retry_max_attempts_then_continue_completes() { // `retry.max_attempts` retries the failing node; after they are exhausted, From 734b53e910c0f24f5318f75d8f727d26b941718a Mon Sep 17 00:00:00 2001 From: oxoxDev Date: Thu, 13 Aug 2026 01:28:24 +0530 Subject: [PATCH 3/3] docs(observability): don't promise failed_node_ids is non-empty for every Failed run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Failed run built from a driver-level fault (recursion limit, checkpointer error, tinyagents graph error) records no per-node Error step, so failed_node_ids() returns an empty vector even though the status is Failed. Only a node-caused stop-policy failure records its Error step before ending the run. Clarify the contract: non-empty means 'these nodes failed', never a proxy for the run's outcome — consult Run::status for that. --- src/observability.rs | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/src/observability.rs b/src/observability.rs index 1801d17..9044d4e 100644 --- a/src/observability.rs +++ b/src/observability.rs @@ -112,11 +112,21 @@ impl Run { /// The ids of the nodes that errored in this run — the [`steps`](Run::steps) /// whose [`StepStatus`] is [`StepStatus::Error`], in the order they finished. /// - /// Empty for a clean [`RunStatus::Completed`]; non-empty exactly when the - /// status is [`RunStatus::CompletedWithErrors`] (nodes handled by - /// `continue`/`route`) or [`RunStatus::Failed`] (the `stop`-policy node that - /// ended the run). Lets a host act on *which* nodes failed without scanning - /// every step itself. + /// Always empty for a clean [`RunStatus::Completed`], and always non-empty + /// for [`RunStatus::CompletedWithErrors`] (that status *is* derived from at + /// least one `Error` step: every node a `continue`/`route` policy turned + /// into data records one). + /// + /// For [`RunStatus::Failed`] it names the failing node **only when a node + /// caused the failure** — a `stop`-policy node records its `Error` step on + /// the way out before ending the run. A `Failed` run that came from a + /// driver-level fault with no node behind it (a hit recursion limit, a + /// checkpointer error, a tinyagents graph error) records no `Error` step and + /// so returns an **empty** vector. So read a non-empty result as "these + /// nodes failed" — never read emptiness as "the run succeeded", and never + /// use this as a proxy for [`RunStatus::Failed`]; consult the run's + /// [`status`](Run::status) for the outcome. Lets a host act on *which* nodes + /// failed without scanning every step itself. pub fn failed_node_ids(&self) -> Vec<&str> { self.steps .iter()