diff --git a/lib/crates/fabro-workflows/src/graph.rs b/lib/crates/fabro-workflows/src/graph.rs index ca02b92b1..c72875be2 100644 --- a/lib/crates/fabro-workflows/src/graph.rs +++ b/lib/crates/fabro-workflows/src/graph.rs @@ -1,12 +1,13 @@ +mod routing; + use std::collections::HashMap; use std::sync::Arc; use fabro_core::error::{CoreError, Result as CoreResult}; -use fabro_core::graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; +use fabro_core::graph::{EdgeSelection as CoreEdgeSelection, EdgeSpec, Graph, NodeSpec}; use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode}; use crate::context::Context; -use crate::graph_ops; use crate::outcome::{Outcome, StageUsage}; // ---- WorkflowNode ---- @@ -26,7 +27,7 @@ impl NodeSpec for WorkflowNode { } fn is_terminal(&self) -> bool { - graph_ops::is_terminal(&self.0) + routing::is_terminal(&self.0) } fn max_visits(&self) -> Option { @@ -102,15 +103,15 @@ impl Graph for WorkflowGraph { node: &Self::Node, outcome: &Outcome, context: &Context, - ) -> Option> { - let selection = graph_ops::select_edge( + ) -> Option> { + let selection = routing::select_edge( node.inner(), outcome, context, self.inner(), node.inner().selection(), ); - selection.map(|sel| EdgeSelection { + selection.map(|sel| CoreEdgeSelection { edge: WorkflowEdge(Arc::new(sel.edge.clone())), reason: sel.reason, }) @@ -120,10 +121,10 @@ impl Graph for WorkflowGraph { &self, outcomes: &HashMap, ) -> std::result::Result<(), String> { - graph_ops::check_goal_gates(self.inner(), outcomes) + routing::check_goal_gates(self.inner(), outcomes) } fn get_retry_target(&self, failed_node_id: &str) -> Option { - graph_ops::get_retry_target(failed_node_id, self.inner()) + routing::get_retry_target(failed_node_id, self.inner()) } } diff --git a/lib/crates/fabro-workflows/src/graph_ops.rs b/lib/crates/fabro-workflows/src/graph/routing.rs similarity index 51% rename from lib/crates/fabro-workflows/src/graph_ops.rs rename to lib/crates/fabro-workflows/src/graph/routing.rs index fb323887a..60dc77fce 100644 --- a/lib/crates/fabro-workflows/src/graph_ops.rs +++ b/lib/crates/fabro-workflows/src/graph/routing.rs @@ -1,203 +1,147 @@ use std::collections::HashMap; -use std::time::Duration; -use fabro_graphviz::graph::{Edge, Graph, Node}; -use fabro_hooks::HookContext; -use fabro_util::backoff::BackoffPolicy; +use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode}; use rand::Rng; use crate::condition::evaluate_condition; -use crate::context::{self, Context}; -use crate::error::FailureCategory; -use crate::outcome::{Outcome, OutcomeExt, StageStatus}; +use crate::context::Context; +use crate::outcome::{Outcome, StageStatus}; -/// Populate node-related fields on a `HookContext` from a graph `Node`. -pub(crate) fn set_hook_node(ctx: &mut HookContext, node: &Node) { - ctx.node_id = Some(node.id.clone()); - ctx.node_label = Some(node.label().to_string()); - ctx.handler_type = node.handler_type().map(String::from); +/// Result of edge selection: the chosen edge and the reason it was selected. +pub(crate) struct SelectedGraphEdge<'a> { + pub(crate) edge: &'a GvEdge, + pub(crate) reason: &'static str, } -/// Classify the failure mode of a completed outcome. -/// -/// Returns `None` for `Success`, `PartialSuccess`, and `Skipped` outcomes. -/// For failures, checks (in priority order): -/// 1. Handler hint in `context_updates["failure_class"]` -/// 2. String heuristics on `failure_reason` -/// 3. Default to `Deterministic` -#[must_use] -pub(crate) fn classify_outcome(outcome: &Outcome) -> Option { - match outcome.status { - StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped => None, - StageStatus::Fail | StageStatus::Retry => outcome - .failure_category() - .or(Some(FailureCategory::Deterministic)), - } +/// Check whether a node is a terminal (exit) node. +pub(crate) fn is_terminal(node: &GvNode) -> bool { + node.shape() == "Msquare" || node.handler_type() == Some("exit") } -/// Retry policy for node execution. -#[derive(Clone, Debug)] -pub struct RetryPolicy { - pub max_attempts: u32, - pub backoff: BackoffPolicy, +/// Select the next edge from a node's outgoing edges (spec Section 3.3). +pub(crate) fn select_edge<'a>( + node: &GvNode, + outcome: &Outcome, + context: &Context, + graph: &'a GvGraph, + selection: &str, +) -> Option> { + let node_id = &node.id; + let edges = graph.outgoing_edges(node_id); + if edges.is_empty() { + return None; + } + + let condition_matched: Vec<&GvEdge> = edges + .iter() + .filter(|e| { + e.condition() + .is_some_and(|c| !c.is_empty() && evaluate_condition(c, outcome, context)) + }) + .copied() + .collect(); + if !condition_matched.is_empty() { + return pick_edge(&condition_matched, selection).map(|edge| SelectedGraphEdge { + edge, + reason: "condition", + }); + } + + if let Some(pref) = &outcome.preferred_label { + let normalized_pref = normalize_label(pref); + for edge in &edges { + if edge.condition().is_none_or(str::is_empty) { + if let Some(label) = edge.label() { + if normalize_label(label) == normalized_pref { + return Some(SelectedGraphEdge { + edge, + reason: "preferred_label", + }); + } + } + } + } + } + + for suggested_id in &outcome.suggested_next_ids { + for edge in &edges { + if edge.condition().is_none_or(str::is_empty) && edge.to == *suggested_id { + return Some(SelectedGraphEdge { + edge, + reason: "suggested_next", + }); + } + } + } + + if blocks_unconditional_failure_fallthrough(node, outcome) { + return None; + } + + let unconditional: Vec<&GvEdge> = edges + .iter() + .filter(|e| e.condition().is_none_or(str::is_empty)) + .copied() + .collect(); + if !unconditional.is_empty() { + return pick_edge(&unconditional, selection).map(|edge| SelectedGraphEdge { + edge, + reason: "unconditional", + }); + } + + None } -impl RetryPolicy { - const DEFAULT_BACKOFF: BackoffPolicy = BackoffPolicy { - initial_delay: Duration::from_millis(5_000), - factor: 2.0, - max_delay: Duration::from_millis(60_000), - jitter: true, - }; - - /// No retries -- fail immediately. - #[must_use] - pub fn none() -> Self { - Self { - max_attempts: 1, - backoff: Self::DEFAULT_BACKOFF, - } - } - - /// Standard retry policy: 5 attempts, 5s initial, 2x factor. - #[must_use] - pub fn standard() -> Self { - Self { - max_attempts: 5, - backoff: Self::DEFAULT_BACKOFF, - } - } - - /// Aggressive retry: 5 attempts, 500ms initial, 2x factor. - #[must_use] - pub fn aggressive() -> Self { - Self { - max_attempts: 5, - backoff: BackoffPolicy { - initial_delay: Duration::from_millis(500), - ..Self::DEFAULT_BACKOFF - }, - } - } - - /// Linear retry: 3 attempts, 500ms fixed delay. - #[must_use] - pub fn linear() -> Self { - Self { - max_attempts: 3, - backoff: BackoffPolicy { - initial_delay: Duration::from_millis(500), - factor: 1.0, - ..Self::DEFAULT_BACKOFF - }, - } - } - - /// Patient retry: 3 attempts, 2000ms initial, 3x factor. - #[must_use] - pub fn patient() -> Self { - Self { - max_attempts: 3, - backoff: BackoffPolicy { - initial_delay: Duration::from_millis(2000), - factor: 3.0, - ..Self::DEFAULT_BACKOFF - }, +/// Check if all goal gates have been satisfied. +/// Returns Ok(()) if all gates passed, or Err with the failed node ID. +pub(crate) fn check_goal_gates( + graph: &GvGraph, + node_outcomes: &HashMap, +) -> std::result::Result<(), String> { + for (node_id, outcome) in node_outcomes { + if let Some(node) = graph.nodes.get(node_id) { + if node.goal_gate() + && outcome.status != StageStatus::Success + && outcome.status != StageStatus::PartialSuccess + { + return Err(node_id.clone()); + } } } + Ok(()) } -/// Build a retry policy from node and graph attributes. -/// If the node has a `retry_policy` attribute naming a preset, use that. -/// Otherwise, fall back to `max_retries` / graph default. -pub(crate) fn build_retry_policy(node: &Node, graph: &Graph) -> RetryPolicy { - if let Some(preset) = node.retry_policy() { - match preset { - "none" => return RetryPolicy::none(), - "standard" => return RetryPolicy::standard(), - "aggressive" => return RetryPolicy::aggressive(), - "linear" => return RetryPolicy::linear(), - "patient" => return RetryPolicy::patient(), - _ => {} +/// Resolve the retry target for a failed goal gate node. +pub(crate) fn get_retry_target(failed_node_id: &str, graph: &GvGraph) -> Option { + if let Some(node) = graph.nodes.get(failed_node_id) { + if let Some(target) = node.retry_target() { + if graph.nodes.contains_key(target) { + return Some(target.to_string()); + } + } + if let Some(target) = node.fallback_retry_target() { + if graph.nodes.contains_key(target) { + return Some(target.to_string()); + } } } - let max_retries = node - .max_retries() - .unwrap_or_else(|| graph.default_max_retries()); - let max_attempts = u32::try_from(max_retries + 1).unwrap_or(1).max(1); - RetryPolicy { - max_attempts, - backoff: RetryPolicy::DEFAULT_BACKOFF, - } -} - -/// Resolve the context fidelity for a node, following the precedence: -/// 1. Incoming edge `fidelity` attribute -/// 2. Target node `fidelity` attribute -/// 3. Graph `default_fidelity` attribute -/// 4. Default: Compact -#[must_use] -pub fn resolve_fidelity( - incoming_edge: Option<&Edge>, - node: &Node, - graph: &Graph, -) -> context::keys::Fidelity { - let (resolved, source) = if let Some(f) = incoming_edge - .and_then(|e| e.fidelity()) - .and_then(|s| s.parse().ok()) - { - (f, "edge") - } else if let Some(f) = node.fidelity().and_then(|s| s.parse().ok()) { - (f, "node") - } else if let Some(f) = graph.default_fidelity().and_then(|s| s.parse().ok()) { - (f, "graph") - } else { - (context::keys::Fidelity::default(), "default") - }; - - tracing::debug!( - node = %node.id, - fidelity = %resolved, - source = source, - "Fidelity resolved" - ); - - resolved -} - -/// Resolve the thread ID for a node, following the precedence: -/// 1. Incoming edge `thread_id` attribute -/// 2. Target node `thread_id` attribute -/// 3. Graph-level default thread -/// 4. Derived class from enclosing subgraph (first class from the node's classes list) -/// 5. Fallback to previous node ID -#[must_use] -pub fn resolve_thread_id( - incoming_edge: Option<&Edge>, - node: &Node, - graph: &Graph, - previous_node_id: Option<&str>, -) -> Option { - if let Some(edge) = incoming_edge { - if let Some(tid) = edge.thread_id() { - return Some(tid.to_string()); + if let Some(target) = graph.retry_target() { + if graph.nodes.contains_key(target) { + return Some(target.to_string()); } } - if let Some(tid) = node.thread_id() { - return Some(tid.to_string()); + if let Some(target) = graph.fallback_retry_target() { + if graph.nodes.contains_key(target) { + return Some(target.to_string()); + } } - if let Some(tid) = graph.default_thread() { - return Some(tid.to_string()); - } - if let Some(first_class) = node.classes.first() { - return Some(first_class.clone()); - } - previous_node_id.map(String::from) + None } /// Normalize a label for comparison: lowercase, trim, strip accelerator prefixes. /// Patterns: "[Y] ", "Y) ", "Y - " -pub(crate) fn normalize_label(label: &str) -> String { +fn normalize_label(label: &str) -> String { let s = label.trim().to_lowercase(); if s.starts_with('[') { if let Some(rest) = s @@ -222,7 +166,7 @@ pub(crate) fn normalize_label(label: &str) -> String { } /// Pick the best edge by highest weight, then lexical target node ID tiebreak. -pub(crate) fn best_by_weight_then_lexical<'a>(edges: &[&'a Edge]) -> Option<&'a Edge> { +fn best_by_weight_then_lexical<'a>(edges: &[&'a GvEdge]) -> Option<&'a GvEdge> { if edges.is_empty() { return None; } @@ -237,7 +181,7 @@ pub(crate) fn best_by_weight_then_lexical<'a>(edges: &[&'a Edge]) -> Option<&'a /// Pick a random edge using weighted-random selection. /// Edges with `weight <= 0` are treated as weight 1 for probability calculation. -pub(crate) fn weighted_random<'a>(edges: &[&'a Edge]) -> Option<&'a Edge> { +fn weighted_random<'a>(edges: &[&'a GvEdge]) -> Option<&'a GvEdge> { if edges.is_empty() { return None; } @@ -268,325 +212,44 @@ pub(crate) fn weighted_random<'a>(edges: &[&'a Edge]) -> Option<&'a Edge> { } /// Dispatch to the appropriate edge-picking strategy. -fn pick_edge<'a>(edges: &[&'a Edge], selection: &str) -> Option<&'a Edge> { +fn pick_edge<'a>(edges: &[&'a GvEdge], selection: &str) -> Option<&'a GvEdge> { match selection { "random" => weighted_random(edges), _ => best_by_weight_then_lexical(edges), } } -/// Result of edge selection: the chosen edge and the reason it was selected. -pub struct EdgeSelection<'a> { - pub edge: &'a Edge, - pub reason: &'static str, -} - -fn blocks_unconditional_failure_fallthrough(node: &Node, outcome: &Outcome) -> bool { +fn blocks_unconditional_failure_fallthrough(node: &GvNode, outcome: &Outcome) -> bool { node.handler_type() == Some("human") && outcome.status == StageStatus::Fail && outcome.preferred_label.is_none() && outcome.suggested_next_ids.is_empty() } -/// Select the next edge from a node's outgoing edges (spec Section 3.3). -#[must_use] -pub fn select_edge<'a>( - node: &Node, - outcome: &Outcome, - context: &Context, - graph: &'a Graph, - selection: &str, -) -> Option> { - let node_id = &node.id; - let edges = graph.outgoing_edges(node_id); - if edges.is_empty() { - return None; - } - - let condition_matched: Vec<&Edge> = edges - .iter() - .filter(|e| { - e.condition() - .is_some_and(|c| !c.is_empty() && evaluate_condition(c, outcome, context)) - }) - .copied() - .collect(); - if !condition_matched.is_empty() { - return pick_edge(&condition_matched, selection).map(|edge| EdgeSelection { - edge, - reason: "condition", - }); - } - - if let Some(pref) = &outcome.preferred_label { - let normalized_pref = normalize_label(pref); - for edge in &edges { - if edge.condition().is_none_or(str::is_empty) { - if let Some(label) = edge.label() { - if normalize_label(label) == normalized_pref { - return Some(EdgeSelection { - edge, - reason: "preferred_label", - }); - } - } - } - } - } - - for suggested_id in &outcome.suggested_next_ids { - for edge in &edges { - if edge.condition().is_none_or(str::is_empty) && edge.to == *suggested_id { - return Some(EdgeSelection { - edge, - reason: "suggested_next", - }); - } - } - } - - if blocks_unconditional_failure_fallthrough(node, outcome) { - return None; - } - - let unconditional: Vec<&Edge> = edges - .iter() - .filter(|e| e.condition().is_none_or(str::is_empty)) - .copied() - .collect(); - if !unconditional.is_empty() { - return pick_edge(&unconditional, selection).map(|edge| EdgeSelection { - edge, - reason: "unconditional", - }); - } - - None -} - -/// Check if all goal gates have been satisfied. -/// Returns Ok(()) if all gates passed, or Err with the failed node ID. -pub(crate) fn check_goal_gates( - graph: &Graph, - node_outcomes: &HashMap, -) -> std::result::Result<(), String> { - for (node_id, outcome) in node_outcomes { - if let Some(node) = graph.nodes.get(node_id) { - if node.goal_gate() - && outcome.status != StageStatus::Success - && outcome.status != StageStatus::PartialSuccess - { - return Err(node_id.clone()); - } - } - } - Ok(()) -} - -/// Resolve the retry target for a failed goal gate node. -pub(crate) fn get_retry_target(failed_node_id: &str, graph: &Graph) -> Option { - if let Some(node) = graph.nodes.get(failed_node_id) { - if let Some(target) = node.retry_target() { - if graph.nodes.contains_key(target) { - return Some(target.to_string()); - } - } - if let Some(target) = node.fallback_retry_target() { - if graph.nodes.contains_key(target) { - return Some(target.to_string()); - } - } - } - if let Some(target) = graph.retry_target() { - if graph.nodes.contains_key(target) { - return Some(target.to_string()); - } - } - if let Some(target) = graph.fallback_retry_target() { - if graph.nodes.contains_key(target) { - return Some(target.to_string()); - } - } - None -} - -/// Check whether a node is a terminal (exit) node. -pub(crate) fn is_terminal(node: &Node) -> bool { - node.shape() == "Msquare" || node.handler_type() == Some("exit") -} - -pub(crate) fn node_script(node: &Node) -> Option { - node.attrs - .get("script") - .or_else(|| node.attrs.get("tool_command")) - .and_then(|v| v.as_str()) - .map(String::from) -} - #[cfg(test)] mod tests { + use std::collections::HashMap; + + use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; + use super::*; use crate::context::Context; - use crate::error::FailureCategory; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; - use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; - use std::collections::HashMap; - use std::time::Duration; - // --- RetryPolicy preset tests --- - - #[test] - fn retry_policy_none() { - let policy = RetryPolicy::none(); - assert_eq!(policy.max_attempts, 1); + fn make_graph_with_edges(edges: Vec) -> Graph { + let mut g = Graph::new("test"); + for edge in &edges { + if !g.nodes.contains_key(&edge.from) { + g.nodes.insert(edge.from.clone(), Node::new(&edge.from)); + } + if !g.nodes.contains_key(&edge.to) { + g.nodes.insert(edge.to.clone(), Node::new(&edge.to)); + } + } + g.edges = edges; + g } - #[test] - fn retry_policy_standard() { - let policy = RetryPolicy::standard(); - assert_eq!(policy.max_attempts, 5); - assert_eq!(policy.backoff.initial_delay, Duration::from_millis(5_000)); - } - - #[test] - fn retry_policy_aggressive() { - let policy = RetryPolicy::aggressive(); - assert_eq!(policy.max_attempts, 5); - assert_eq!(policy.backoff.initial_delay, Duration::from_millis(500)); - } - - #[test] - fn retry_policy_linear() { - let policy = RetryPolicy::linear(); - assert_eq!(policy.max_attempts, 3); - assert_eq!(policy.backoff.factor, 1.0); - } - - #[test] - fn retry_policy_patient() { - let policy = RetryPolicy::patient(); - assert_eq!(policy.max_attempts, 3); - assert_eq!(policy.backoff.initial_delay, Duration::from_millis(2000)); - } - - // --- build_retry_policy tests --- - - #[test] - fn build_retry_policy_from_node() { - let mut node = Node::new("n"); - node.attrs - .insert("max_retries".to_string(), AttrValue::Integer(3)); - let graph = Graph::new("test"); - let policy = build_retry_policy(&node, &graph); - assert_eq!(policy.max_attempts, 4); // 3 retries + 1 initial - } - - #[test] - fn build_retry_policy_from_graph_default() { - let node = Node::new("n"); - let mut graph = Graph::new("test"); - graph - .attrs - .insert("default_max_retries".to_string(), AttrValue::Integer(2)); - let policy = build_retry_policy(&node, &graph); - assert_eq!(policy.max_attempts, 3); // 2 retries + 1 initial - } - - #[test] - fn build_retry_policy_no_attrs_uses_graph_default_0() { - let node = Node::new("n"); - let graph = Graph::new("test"); - let policy = build_retry_policy(&node, &graph); - assert_eq!(policy.max_attempts, 1); // default_max_retries=0 + 1 - } - - #[test] - fn build_retry_policy_from_retry_policy_attr() { - let mut node = Node::new("n"); - node.attrs.insert( - "retry_policy".to_string(), - AttrValue::String("aggressive".to_string()), - ); - let graph = Graph::new("test"); - let policy = build_retry_policy(&node, &graph); - assert_eq!(policy.max_attempts, 5); - assert_eq!(policy.backoff.initial_delay, Duration::from_millis(500)); - } - - #[test] - fn build_retry_policy_fallback_when_no_retry_policy_attr() { - let mut node = Node::new("n"); - node.attrs - .insert("max_retries".to_string(), AttrValue::Integer(3)); - let graph = Graph::new("test"); - let policy = build_retry_policy(&node, &graph); - assert_eq!(policy.max_attempts, 4); // 3 retries + 1 initial - // Should use default backoff, not a preset's backoff - assert_eq!(policy.backoff.initial_delay, Duration::from_millis(5_000)); - } - - #[test] - fn build_retry_policy_all_presets() { - let presets = [ - ("none", 1u32), - ("standard", 5), - ("aggressive", 5), - ("linear", 3), - ("patient", 3), - ]; - let graph = Graph::new("test"); - let (name, expected) = presets[0]; - let mut node = Node::new("n"); - node.attrs.insert( - "retry_policy".to_string(), - AttrValue::String(name.to_string()), - ); - assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); - - let (name, expected) = presets[1]; - node.attrs.insert( - "retry_policy".to_string(), - AttrValue::String(name.to_string()), - ); - assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); - - let (name, expected) = presets[2]; - node.attrs.insert( - "retry_policy".to_string(), - AttrValue::String(name.to_string()), - ); - assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); - - let (name, expected) = presets[3]; - node.attrs.insert( - "retry_policy".to_string(), - AttrValue::String(name.to_string()), - ); - assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); - - let (name, expected) = presets[4]; - node.attrs.insert( - "retry_policy".to_string(), - AttrValue::String(name.to_string()), - ); - assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); - } - - #[test] - fn build_retry_policy_unknown_preset_falls_back() { - let mut node = Node::new("n"); - node.attrs.insert( - "retry_policy".to_string(), - AttrValue::String("unknown_preset".to_string()), - ); - let graph = Graph::new("test"); - let policy = build_retry_policy(&node, &graph); - // Unknown preset should fall back to graph default_max_retries=0 - assert_eq!(policy.max_attempts, 1); - } - - // --- normalize_label tests --- - #[test] fn normalize_label_lowercase_and_trim() { assert_eq!(normalize_label(" Yes "), "yes"); @@ -613,8 +276,6 @@ mod tests { assert_eq!(normalize_label("next"), "next"); } - // --- best_by_weight_then_lexical tests --- - #[test] fn best_by_weight_highest_wins() { let e1 = Edge::new("a", "x"); @@ -638,8 +299,6 @@ mod tests { assert!(result.is_none()); } - // --- weighted_random tests --- - #[test] fn weighted_random_empty_returns_none() { assert!(weighted_random(&[]).is_none()); @@ -697,22 +356,6 @@ mod tests { ); } - // --- select_edge tests --- - - fn make_graph_with_edges(edges: Vec) -> Graph { - let mut g = Graph::new("test"); - for edge in &edges { - if !g.nodes.contains_key(&edge.from) { - g.nodes.insert(edge.from.clone(), Node::new(&edge.from)); - } - if !g.nodes.contains_key(&edge.to) { - g.nodes.insert(edge.to.clone(), Node::new(&edge.to)); - } - } - g.edges = edges; - g - } - #[test] fn select_edge_no_edges() { let g = Graph::new("test"); @@ -946,8 +589,6 @@ mod tests { assert!(select_edge(node, &outcome, &context, &g, "random").is_none()); } - // --- check_goal_gates tests --- - #[test] fn goal_gates_all_satisfied() { let mut g = Graph::new("test"); @@ -1003,8 +644,6 @@ mod tests { assert!(check_goal_gates(&g, &outcomes).is_ok()); } - // --- get_retry_target tests --- - #[test] fn retry_target_from_node() { let mut g = Graph::new("test"); @@ -1062,12 +701,9 @@ mod tests { AttrValue::String("nonexistent".to_string()), ); g.nodes.insert("work".to_string(), n); - // No "nonexistent" node -- should fall through to graph-level assert!(get_retry_target("work", &g).is_none()); } - // --- is_terminal tests --- - #[test] fn terminal_by_shape() { let mut n = Node::new("exit"); @@ -1091,279 +727,4 @@ mod tests { let n = Node::new("work"); assert!(!is_terminal(&n)); } - - // --- resolve_fidelity tests --- - - #[test] - fn fidelity_defaults_to_compact() { - use crate::context::keys::Fidelity; - let node = Node::new("work"); - let graph = Graph::new("test"); - assert_eq!(resolve_fidelity(None, &node, &graph), Fidelity::Compact); - } - - #[test] - fn fidelity_from_graph_default() { - use crate::context::keys::Fidelity; - let node = Node::new("work"); - let mut graph = Graph::new("test"); - graph.attrs.insert( - "default_fidelity".to_string(), - AttrValue::String("truncate".to_string()), - ); - assert_eq!(resolve_fidelity(None, &node, &graph), Fidelity::Truncate); - } - - #[test] - fn fidelity_from_node_overrides_graph() { - use crate::context::keys::Fidelity; - let mut node = Node::new("work"); - node.attrs.insert( - "fidelity".to_string(), - AttrValue::String("full".to_string()), - ); - let mut graph = Graph::new("test"); - graph.attrs.insert( - "default_fidelity".to_string(), - AttrValue::String("truncate".to_string()), - ); - assert_eq!(resolve_fidelity(None, &node, &graph), Fidelity::Full); - } - - #[test] - fn fidelity_from_edge_overrides_node() { - use crate::context::keys::Fidelity; - let mut node = Node::new("work"); - node.attrs.insert( - "fidelity".to_string(), - AttrValue::String("full".to_string()), - ); - let mut edge = Edge::new("a", "work"); - edge.attrs.insert( - "fidelity".to_string(), - AttrValue::String("summary:high".to_string()), - ); - let graph = Graph::new("test"); - assert_eq!( - resolve_fidelity(Some(&edge), &node, &graph), - Fidelity::SummaryHigh - ); - } - - // --- resolve_thread_id tests --- - - #[test] - fn thread_id_from_node_attribute() { - let mut node = Node::new("work"); - node.attrs.insert( - "thread_id".to_string(), - AttrValue::String("main-thread".to_string()), - ); - let graph = Graph::new("test"); - assert_eq!( - resolve_thread_id(None, &node, &graph, Some("prev")), - Some("main-thread".to_string()) - ); - } - - #[test] - fn thread_id_from_edge_attribute() { - let node = Node::new("work"); - let mut edge = Edge::new("prev", "work"); - edge.attrs.insert( - "thread_id".to_string(), - AttrValue::String("edge-thread".to_string()), - ); - let graph = Graph::new("test"); - assert_eq!( - resolve_thread_id(Some(&edge), &node, &graph, Some("prev")), - Some("edge-thread".to_string()) - ); - } - - #[test] - fn thread_id_node_used_when_no_edge_thread() { - // When the edge has no thread_id, the node's thread_id is used. - let mut node = Node::new("work"); - node.attrs.insert( - "thread_id".to_string(), - AttrValue::String("node-thread".to_string()), - ); - let edge = Edge::new("prev", "work"); - let graph = Graph::new("test"); - assert_eq!( - resolve_thread_id(Some(&edge), &node, &graph, Some("prev")), - Some("node-thread".to_string()) - ); - } - - #[test] - fn thread_id_edge_overrides_node() { - // Edge thread_id should take precedence over node thread_id, - // matching the fidelity precedence where edge > node. - let mut node = Node::new("work"); - node.attrs.insert( - "thread_id".to_string(), - AttrValue::String("node-thread".to_string()), - ); - let mut edge = Edge::new("prev", "work"); - edge.attrs.insert( - "thread_id".to_string(), - AttrValue::String("edge-thread".to_string()), - ); - let graph = Graph::new("test"); - assert_eq!( - resolve_thread_id(Some(&edge), &node, &graph, Some("prev")), - Some("edge-thread".to_string()), - "edge thread_id should override node thread_id" - ); - } - - #[test] - fn thread_id_from_graph_default_thread() { - let node = Node::new("work"); - let mut graph = Graph::new("test"); - graph.attrs.insert( - "default_thread".to_string(), - AttrValue::String("shared-thread".to_string()), - ); - assert_eq!( - resolve_thread_id(None, &node, &graph, Some("prev")), - Some("shared-thread".to_string()) - ); - } - - #[test] - fn thread_id_edge_overrides_graph_default() { - let node = Node::new("work"); - let mut edge = Edge::new("prev", "work"); - edge.attrs.insert( - "thread_id".to_string(), - AttrValue::String("edge-thread".to_string()), - ); - let mut graph = Graph::new("test"); - graph.attrs.insert( - "default_thread".to_string(), - AttrValue::String("shared-thread".to_string()), - ); - assert_eq!( - resolve_thread_id(Some(&edge), &node, &graph, Some("prev")), - Some("edge-thread".to_string()) - ); - } - - #[test] - fn thread_id_graph_default_overrides_class() { - let mut node = Node::new("work"); - node.classes = vec!["planning".to_string()]; - let mut graph = Graph::new("test"); - graph.attrs.insert( - "default_thread".to_string(), - AttrValue::String("shared-thread".to_string()), - ); - assert_eq!( - resolve_thread_id(None, &node, &graph, Some("prev")), - Some("shared-thread".to_string()) - ); - } - - #[test] - fn thread_id_from_node_class() { - let mut node = Node::new("work"); - node.classes = vec!["planning".to_string(), "review".to_string()]; - let graph = Graph::new("test"); - assert_eq!( - resolve_thread_id(None, &node, &graph, Some("prev")), - Some("planning".to_string()) - ); - } - - #[test] - fn thread_id_fallback_to_previous_node() { - let node = Node::new("work"); - let graph = Graph::new("test"); - assert_eq!( - resolve_thread_id(None, &node, &graph, Some("prev_node")), - Some("prev_node".to_string()) - ); - } - - #[test] - fn thread_id_none_when_no_sources() { - let node = Node::new("start"); - let graph = Graph::new("test"); - assert_eq!(resolve_thread_id(None, &node, &graph, None), None); - } - - // --- classify_outcome tests --- - - #[test] - fn classify_outcome_returns_none_for_success() { - assert!(classify_outcome(&Outcome::success()).is_none()); - } - - #[test] - fn classify_outcome_returns_none_for_skipped() { - assert!(classify_outcome(&Outcome::skipped("")).is_none()); - } - - #[test] - fn classify_outcome_returns_none_for_partial_success() { - let outcome = Outcome { - status: StageStatus::PartialSuccess, - ..Outcome::success() - }; - assert!(classify_outcome(&outcome).is_none()); - } - - #[test] - fn classify_outcome_reads_failure_detail() { - let mut outcome = Outcome::fail_classify("some error"); - // Override the FailureDetail's class directly - outcome.failure.as_mut().unwrap().category = FailureCategory::BudgetExhausted; - assert_eq!( - classify_outcome(&outcome), - Some(FailureCategory::BudgetExhausted) - ); - } - - #[test] - fn classify_outcome_uses_failure_reason_heuristics() { - let outcome = Outcome::fail_classify("rate limited by provider"); - assert_eq!( - classify_outcome(&outcome), - Some(FailureCategory::TransientInfra) - ); - } - - #[test] - fn classify_outcome_defaults_to_deterministic() { - let outcome = Outcome::fail_classify("something went wrong"); - assert_eq!( - classify_outcome(&outcome), - Some(FailureCategory::Deterministic) - ); - } - - #[test] - fn classify_outcome_fail_no_reason_is_deterministic() { - let outcome = Outcome { - status: StageStatus::Fail, - failure: None, - ..Outcome::success() - }; - assert_eq!( - classify_outcome(&outcome), - Some(FailureCategory::Deterministic) - ); - } - - #[test] - fn classify_outcome_retry_status_uses_heuristics() { - let outcome = Outcome::retry_classify("connection refused"); - assert_eq!( - classify_outcome(&outcome), - Some(FailureCategory::TransientInfra) - ); - } } diff --git a/lib/crates/fabro-workflows/src/handler/parallel.rs b/lib/crates/fabro-workflows/src/handler/parallel.rs index 1e6a87b55..b1581a42a 100644 --- a/lib/crates/fabro-workflows/src/handler/parallel.rs +++ b/lib/crates/fabro-workflows/src/handler/parallel.rs @@ -10,7 +10,7 @@ use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; use crate::event::WorkflowRunEvent; -use crate::graph_ops::set_hook_node; +use crate::hook_context::set_hook_node; use crate::millis_u64; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; use fabro_graphviz::graph::{Graph, Node}; diff --git a/lib/crates/fabro-workflows/src/hook_context.rs b/lib/crates/fabro-workflows/src/hook_context.rs new file mode 100644 index 000000000..2fdc1f6ae --- /dev/null +++ b/lib/crates/fabro-workflows/src/hook_context.rs @@ -0,0 +1,35 @@ +use fabro_graphviz::graph::types::Node as GvNode; +use fabro_hooks::HookContext; + +/// Populate node-related fields on a `HookContext` from a graph node. +pub(crate) fn set_hook_node(ctx: &mut HookContext, node: &GvNode) { + ctx.node_id = Some(node.id.clone()); + ctx.node_label = Some(node.label().to_string()); + ctx.handler_type = node.handler_type().map(String::from); +} + +#[cfg(test)] +mod tests { + use fabro_graphviz::graph::{AttrValue, Node}; + use fabro_hooks::HookEvent; + + use super::*; + + #[test] + fn set_hook_node_populates_hook_context_fields() { + let mut node = Node::new("approve"); + node.attrs.insert( + "label".to_string(), + AttrValue::String("Approve PR".to_string()), + ); + node.attrs + .insert("type".to_string(), AttrValue::String("human".to_string())); + + let mut ctx = HookContext::new(HookEvent::StageStart, "run-1".into(), "graph".into()); + set_hook_node(&mut ctx, &node); + + assert_eq!(ctx.node_id.as_deref(), Some("approve")); + assert_eq!(ctx.node_label.as_deref(), Some("Approve PR")); + assert_eq!(ctx.handler_type.as_deref(), Some("human")); + } +} diff --git a/lib/crates/fabro-workflows/src/lib.rs b/lib/crates/fabro-workflows/src/lib.rs index 40d02fcad..92029118a 100644 --- a/lib/crates/fabro-workflows/src/lib.rs +++ b/lib/crates/fabro-workflows/src/lib.rs @@ -98,8 +98,8 @@ pub mod error; pub mod event; pub mod git; pub mod graph; -pub mod graph_ops; pub mod handler; +mod hook_context; pub mod lifecycle; pub mod node_handler; pub mod operations; @@ -107,6 +107,7 @@ pub mod outcome; pub mod pipeline; pub mod pull_request; pub mod records; +mod retry; pub mod run_dir; pub mod run_lookup; pub mod run_settings; diff --git a/lib/crates/fabro-workflows/src/lifecycle/circuit_breaker.rs b/lib/crates/fabro-workflows/src/lifecycle/circuit_breaker.rs index 45ae484ab..3ecbb68fb 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/circuit_breaker.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/circuit_breaker.rs @@ -11,7 +11,6 @@ use fabro_core::state::RunState; use crate::error::{FailureCategory, FailureSignature}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; -use crate::graph_ops::classify_outcome; use crate::outcome::{OutcomeExt, StageStatus, StageUsage}; type WfRunState = RunState>; @@ -69,7 +68,7 @@ impl RunLifecycle for CircuitBreakerLifecycle { let outcome = &result.outcome; let outcome_failure_category = if outcome.status == StageStatus::Fail { - classify_outcome(outcome) + outcome.classified_failure_category() } else { None }; @@ -117,7 +116,7 @@ impl RunLifecycle for CircuitBreakerLifecycle { let outcome = ctx.outcome; // Guard: only TransientInfra failures may trigger loop_restart - let failure_class = classify_outcome(outcome); + let failure_class = outcome.classified_failure_category(); if let Some(fc) = failure_class { if fc != FailureCategory::TransientInfra { return Ok(EdgeDecision::Block(format!( diff --git a/lib/crates/fabro-workflows/src/lifecycle/event.rs b/lib/crates/fabro-workflows/src/lifecycle/event.rs index 0102fb16b..05e0b3014 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/event.rs @@ -13,12 +13,20 @@ use crate::artifact::ArtifactStore; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; -use crate::graph_ops::node_script; use crate::outcome::{FailureCategory, FailureDetail, Outcome, StageStatus, StageUsage}; +use fabro_graphviz::graph::types::Node as GvNode; type WfRunState = RunState>; type WfNodeResult = NodeResult>; +fn node_script(node: &GvNode) -> Option { + node.attrs + .get("script") + .or_else(|| node.attrs.get("tool_command")) + .and_then(|v| v.as_str()) + .map(String::from) +} + /// Sub-lifecycle responsible for emitting workflow run events. pub struct EventLifecycle { pub emitter: Arc, diff --git a/lib/crates/fabro-workflows/src/lifecycle/fidelity.rs b/lib/crates/fabro-workflows/src/lifecycle/fidelity.rs index 0a359fa34..0d1135560 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/fidelity.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/fidelity.rs @@ -5,11 +5,11 @@ use async_trait::async_trait; use fabro_core::graph::NodeSpec; use fabro_core::lifecycle::{EdgeContext, NodeDecision, RunLifecycle}; use fabro_core::state::RunState; +use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode}; use crate::context::keys; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; -use crate::graph_ops::{resolve_fidelity, resolve_thread_id}; use crate::handler::llm::preamble::build_preamble; use crate::outcome::StageUsage; @@ -20,19 +20,19 @@ type WfNodeDecision = NodeDecision>; /// for fidelity/thread resolution. #[derive(Debug, Clone)] struct IncomingEdgeData { - edge: Arc, + edge: Arc, } /// Sub-lifecycle responsible for fidelity/thread resolution and context key setup. pub struct FidelityLifecycle { - pub graph: Arc, + pub graph: Arc, incoming_edge_data: Mutex>, /// True on the first node after checkpoint resume when prior fidelity was Full. degrade_fidelity_on_resume: Mutex, } impl FidelityLifecycle { - pub fn new(graph: Arc) -> Self { + pub fn new(graph: Arc) -> Self { Self { graph, incoming_edge_data: Mutex::new(None), @@ -154,3 +154,265 @@ impl RunLifecycle for FidelityLifecycle { Ok(fabro_core::lifecycle::EdgeDecision::Continue) } } + +/// Resolve the context fidelity for a node, following the precedence: +/// 1. Incoming edge `fidelity` attribute +/// 2. Target node `fidelity` attribute +/// 3. Graph `default_fidelity` attribute +/// 4. Default: Compact +fn resolve_fidelity( + incoming_edge: Option<&GvEdge>, + node: &GvNode, + graph: &GvGraph, +) -> keys::Fidelity { + let (resolved, source) = if let Some(f) = incoming_edge + .and_then(|e| e.fidelity()) + .and_then(|s| s.parse().ok()) + { + (f, "edge") + } else if let Some(f) = node.fidelity().and_then(|s| s.parse().ok()) { + (f, "node") + } else if let Some(f) = graph.default_fidelity().and_then(|s| s.parse().ok()) { + (f, "graph") + } else { + (keys::Fidelity::default(), "default") + }; + + tracing::debug!( + node = %node.id, + fidelity = %resolved, + source = source, + "Fidelity resolved" + ); + + resolved +} + +/// Resolve the thread ID for a node, following the precedence: +/// 1. Incoming edge `thread_id` attribute +/// 2. Target node `thread_id` attribute +/// 3. Graph-level default thread +/// 4. Derived class from enclosing subgraph (first class from the node's classes list) +/// 5. Fallback to previous node ID +fn resolve_thread_id( + incoming_edge: Option<&GvEdge>, + node: &GvNode, + graph: &GvGraph, + previous_node_id: Option<&str>, +) -> Option { + if let Some(edge) = incoming_edge { + if let Some(tid) = edge.thread_id() { + return Some(tid.to_string()); + } + } + if let Some(tid) = node.thread_id() { + return Some(tid.to_string()); + } + if let Some(tid) = graph.default_thread() { + return Some(tid.to_string()); + } + if let Some(first_class) = node.classes.first() { + return Some(first_class.clone()); + } + previous_node_id.map(String::from) +} + +#[cfg(test)] +mod tests { + use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; + + use super::*; + use crate::context::keys::Fidelity; + + #[test] + fn fidelity_defaults_to_compact() { + let node = Node::new("work"); + let graph = Graph::new("test"); + assert_eq!(resolve_fidelity(None, &node, &graph), Fidelity::Compact); + } + + #[test] + fn fidelity_from_graph_default() { + let node = Node::new("work"); + let mut graph = Graph::new("test"); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("truncate".to_string()), + ); + assert_eq!(resolve_fidelity(None, &node, &graph), Fidelity::Truncate); + } + + #[test] + fn fidelity_from_node_overrides_graph() { + let mut node = Node::new("work"); + node.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + let mut graph = Graph::new("test"); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("truncate".to_string()), + ); + assert_eq!(resolve_fidelity(None, &node, &graph), Fidelity::Full); + } + + #[test] + fn fidelity_from_edge_overrides_node() { + let mut node = Node::new("work"); + node.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + let mut edge = Edge::new("a", "work"); + edge.attrs.insert( + "fidelity".to_string(), + AttrValue::String("summary:high".to_string()), + ); + let graph = Graph::new("test"); + assert_eq!( + resolve_fidelity(Some(&edge), &node, &graph), + Fidelity::SummaryHigh + ); + } + + #[test] + fn thread_id_from_node_attribute() { + let mut node = Node::new("work"); + node.attrs.insert( + "thread_id".to_string(), + AttrValue::String("main-thread".to_string()), + ); + let graph = Graph::new("test"); + assert_eq!( + resolve_thread_id(None, &node, &graph, Some("prev")), + Some("main-thread".to_string()) + ); + } + + #[test] + fn thread_id_from_edge_attribute() { + let node = Node::new("work"); + let mut edge = Edge::new("prev", "work"); + edge.attrs.insert( + "thread_id".to_string(), + AttrValue::String("edge-thread".to_string()), + ); + let graph = Graph::new("test"); + assert_eq!( + resolve_thread_id(Some(&edge), &node, &graph, Some("prev")), + Some("edge-thread".to_string()) + ); + } + + #[test] + fn thread_id_node_used_when_no_edge_thread() { + let mut node = Node::new("work"); + node.attrs.insert( + "thread_id".to_string(), + AttrValue::String("node-thread".to_string()), + ); + let edge = Edge::new("prev", "work"); + let graph = Graph::new("test"); + assert_eq!( + resolve_thread_id(Some(&edge), &node, &graph, Some("prev")), + Some("node-thread".to_string()) + ); + } + + #[test] + fn thread_id_edge_overrides_node() { + let mut node = Node::new("work"); + node.attrs.insert( + "thread_id".to_string(), + AttrValue::String("node-thread".to_string()), + ); + let mut edge = Edge::new("prev", "work"); + edge.attrs.insert( + "thread_id".to_string(), + AttrValue::String("edge-thread".to_string()), + ); + let graph = Graph::new("test"); + assert_eq!( + resolve_thread_id(Some(&edge), &node, &graph, Some("prev")), + Some("edge-thread".to_string()), + "edge thread_id should override node thread_id" + ); + } + + #[test] + fn thread_id_from_graph_default_thread() { + let node = Node::new("work"); + let mut graph = Graph::new("test"); + graph.attrs.insert( + "default_thread".to_string(), + AttrValue::String("shared-thread".to_string()), + ); + assert_eq!( + resolve_thread_id(None, &node, &graph, Some("prev")), + Some("shared-thread".to_string()) + ); + } + + #[test] + fn thread_id_edge_overrides_graph_default() { + let node = Node::new("work"); + let mut edge = Edge::new("prev", "work"); + edge.attrs.insert( + "thread_id".to_string(), + AttrValue::String("edge-thread".to_string()), + ); + let mut graph = Graph::new("test"); + graph.attrs.insert( + "default_thread".to_string(), + AttrValue::String("shared-thread".to_string()), + ); + assert_eq!( + resolve_thread_id(Some(&edge), &node, &graph, Some("prev")), + Some("edge-thread".to_string()) + ); + } + + #[test] + fn thread_id_graph_default_overrides_class() { + let mut node = Node::new("work"); + node.classes = vec!["planning".to_string()]; + let mut graph = Graph::new("test"); + graph.attrs.insert( + "default_thread".to_string(), + AttrValue::String("shared-thread".to_string()), + ); + assert_eq!( + resolve_thread_id(None, &node, &graph, Some("prev")), + Some("shared-thread".to_string()) + ); + } + + #[test] + fn thread_id_from_node_class() { + let mut node = Node::new("work"); + node.classes = vec!["planning".to_string(), "review".to_string()]; + let graph = Graph::new("test"); + assert_eq!( + resolve_thread_id(None, &node, &graph, Some("prev")), + Some("planning".to_string()) + ); + } + + #[test] + fn thread_id_fallback_to_previous_node() { + let node = Node::new("work"); + let graph = Graph::new("test"); + assert_eq!( + resolve_thread_id(None, &node, &graph, Some("prev_node")), + Some("prev_node".to_string()) + ); + } + + #[test] + fn thread_id_none_when_no_sources() { + let node = Node::new("start"); + let graph = Graph::new("test"); + assert_eq!(resolve_thread_id(None, &node, &graph, None), None); + } +} diff --git a/lib/crates/fabro-workflows/src/lifecycle/hook.rs b/lib/crates/fabro-workflows/src/lifecycle/hook.rs index 3f52b63fb..2cadc1931 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/hook.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/hook.rs @@ -12,7 +12,7 @@ use fabro_core::state::RunState; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; -use crate::graph_ops::set_hook_node; +use crate::hook_context::set_hook_node; use crate::outcome::{Outcome, OutcomeExt, StageStatus, StageUsage}; use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner}; use fabro_sandbox::Sandbox; diff --git a/lib/crates/fabro-workflows/src/lifecycle/mod.rs b/lib/crates/fabro-workflows/src/lifecycle/mod.rs index e5dd639ef..5deb9204a 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/mod.rs @@ -307,7 +307,7 @@ impl RunLifecycle for WorkflowLifecycle { ) -> CoreResult<()> { let outcome = &result.outcome; let retry_count = state.node_retries.get(node.id()).copied().unwrap_or(0); - let failure_class = crate::graph_ops::classify_outcome(outcome); + let failure_class = outcome.classified_failure_category(); let failure_signature = failure_class .map(|category| { let signature_hint = outcome diff --git a/lib/crates/fabro-workflows/src/node_handler.rs b/lib/crates/fabro-workflows/src/node_handler.rs index 4af683def..07c5cc659 100644 --- a/lib/crates/fabro-workflows/src/node_handler.rs +++ b/lib/crates/fabro-workflows/src/node_handler.rs @@ -16,7 +16,8 @@ use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; use crate::handler::{format_panic_message, EngineServices}; use crate::outcome::{Outcome, StageStatus}; -use crate::{graph_ops, run_dir}; +use crate::retry::build_retry_policy; +use crate::run_dir; /// Production node handler that bridges fabro-core's NodeHandler to the /// existing fabro-workflows Handler trait via EngineServices. @@ -113,11 +114,7 @@ impl NodeHandler for WorkflowNodeHandler { fn retry_policy(&self, node: &WorkflowNode, _graph: &WorkflowGraph) -> CoreRetryPolicy { let gv_node = node.inner(); - let wf_policy = graph_ops::build_retry_policy(gv_node, &self.graph); - CoreRetryPolicy { - max_attempts: wf_policy.max_attempts, - backoff: wf_policy.backoff, - } + build_retry_policy(gv_node, &self.graph) } fn on_retries_exhausted(&self, node: &WorkflowNode, last_outcome: Outcome) -> Outcome { diff --git a/lib/crates/fabro-workflows/src/outcome.rs b/lib/crates/fabro-workflows/src/outcome.rs index 07434a9a8..073b48b21 100644 --- a/lib/crates/fabro-workflows/src/outcome.rs +++ b/lib/crates/fabro-workflows/src/outcome.rs @@ -62,6 +62,13 @@ pub trait OutcomeExt: Sized { /// Get the failure category, if this is a failed outcome. fn failure_category(&self) -> Option; + + /// Resolve the effective failure category for this outcome. + /// + /// Returns `None` for success, partial success, and skipped outcomes. + /// Failed and retry outcomes default to `Deterministic` when no + /// structured failure category is present. + fn classified_failure_category(&self) -> Option; } impl OutcomeExt for Outcome { @@ -114,6 +121,15 @@ impl OutcomeExt for Outcome { fn failure_category(&self) -> Option { self.failure.as_ref().map(|f| f.category) } + + fn classified_failure_category(&self) -> Option { + match self.status { + StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped => None, + StageStatus::Fail | StageStatus::Retry => self + .failure_category() + .or(Some(FailureCategory::Deterministic)), + } + } } /// Compute the dollar cost for a stage's token usage, if pricing is available. @@ -274,6 +290,75 @@ mod tests { ); } + #[test] + fn classified_failure_category_returns_none_for_success() { + assert!(Outcome::success().classified_failure_category().is_none()); + } + + #[test] + fn classified_failure_category_returns_none_for_skipped() { + assert!(Outcome::skipped("").classified_failure_category().is_none()); + } + + #[test] + fn classified_failure_category_returns_none_for_partial_success() { + let outcome = Outcome { + status: StageStatus::PartialSuccess, + ..Outcome::success() + }; + assert!(outcome.classified_failure_category().is_none()); + } + + #[test] + fn classified_failure_category_reads_failure_detail() { + let mut outcome = Outcome::fail_classify("some error"); + outcome.failure.as_mut().unwrap().category = FailureCategory::BudgetExhausted; + assert_eq!( + outcome.classified_failure_category(), + Some(FailureCategory::BudgetExhausted) + ); + } + + #[test] + fn classified_failure_category_uses_failure_reason_heuristics() { + let outcome = Outcome::fail_classify("rate limited by provider"); + assert_eq!( + outcome.classified_failure_category(), + Some(FailureCategory::TransientInfra) + ); + } + + #[test] + fn classified_failure_category_defaults_to_deterministic() { + let outcome = Outcome::fail_classify("something went wrong"); + assert_eq!( + outcome.classified_failure_category(), + Some(FailureCategory::Deterministic) + ); + } + + #[test] + fn classified_failure_category_fail_no_reason_is_deterministic() { + let outcome = Outcome { + status: StageStatus::Fail, + failure: None, + ..Outcome::success() + }; + assert_eq!( + outcome.classified_failure_category(), + Some(FailureCategory::Deterministic) + ); + } + + #[test] + fn classified_failure_category_retry_status_uses_heuristics() { + let outcome = Outcome::retry_classify("connection refused"); + assert_eq!( + outcome.classified_failure_category(), + Some(FailureCategory::TransientInfra) + ); + } + #[test] fn stage_usage_serialization_with_cache_and_reasoning() { let usage = StageUsage { diff --git a/lib/crates/fabro-workflows/src/retry.rs b/lib/crates/fabro-workflows/src/retry.rs new file mode 100644 index 000000000..72b122418 --- /dev/null +++ b/lib/crates/fabro-workflows/src/retry.rs @@ -0,0 +1,190 @@ +use std::time::Duration; + +use fabro_core::retry::{BackoffPolicy, RetryPolicy}; +use fabro_graphviz::graph::types::{Graph as GvGraph, Node as GvNode}; + +const DEFAULT_BACKOFF: BackoffPolicy = BackoffPolicy { + initial_delay: Duration::from_millis(5_000), + factor: 2.0, + max_delay: Duration::from_millis(60_000), + jitter: true, +}; + +/// Build a retry policy from node and graph attributes. +/// If the node has a `retry_policy` attribute naming a preset, use that. +/// Otherwise, fall back to `max_retries` / graph default. +pub(crate) fn build_retry_policy(node: &GvNode, graph: &GvGraph) -> RetryPolicy { + if let Some(preset) = node.retry_policy() { + if let Some(policy) = preset_retry_policy(preset) { + return policy; + } + } + + let max_retries = node + .max_retries() + .unwrap_or_else(|| graph.default_max_retries()); + let max_attempts = u32::try_from(max_retries + 1).unwrap_or(1).max(1); + + RetryPolicy { + max_attempts, + backoff: DEFAULT_BACKOFF, + } +} + +fn preset_retry_policy(preset: &str) -> Option { + match preset { + "none" => Some(RetryPolicy { + max_attempts: 1, + backoff: DEFAULT_BACKOFF, + }), + "standard" => Some(RetryPolicy { + max_attempts: 5, + backoff: DEFAULT_BACKOFF, + }), + "aggressive" => Some(RetryPolicy { + max_attempts: 5, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(500), + ..DEFAULT_BACKOFF + }, + }), + "linear" => Some(RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(500), + factor: 1.0, + ..DEFAULT_BACKOFF + }, + }), + "patient" => Some(RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(2_000), + factor: 3.0, + ..DEFAULT_BACKOFF + }, + }), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use fabro_graphviz::graph::{AttrValue, Graph, Node}; + + use super::*; + + #[test] + fn build_retry_policy_from_node() { + let mut node = Node::new("n"); + node.attrs + .insert("max_retries".to_string(), AttrValue::Integer(3)); + let graph = Graph::new("test"); + let policy = build_retry_policy(&node, &graph); + assert_eq!(policy.max_attempts, 4); + } + + #[test] + fn build_retry_policy_from_graph_default() { + let node = Node::new("n"); + let mut graph = Graph::new("test"); + graph + .attrs + .insert("default_max_retries".to_string(), AttrValue::Integer(2)); + let policy = build_retry_policy(&node, &graph); + assert_eq!(policy.max_attempts, 3); + } + + #[test] + fn build_retry_policy_no_attrs_uses_graph_default_0() { + let node = Node::new("n"); + let graph = Graph::new("test"); + let policy = build_retry_policy(&node, &graph); + assert_eq!(policy.max_attempts, 1); + } + + #[test] + fn build_retry_policy_from_retry_policy_attr() { + let mut node = Node::new("n"); + node.attrs.insert( + "retry_policy".to_string(), + AttrValue::String("aggressive".to_string()), + ); + let graph = Graph::new("test"); + let policy = build_retry_policy(&node, &graph); + assert_eq!(policy.max_attempts, 5); + assert_eq!(policy.backoff.initial_delay, Duration::from_millis(500)); + } + + #[test] + fn build_retry_policy_fallback_when_no_retry_policy_attr() { + let mut node = Node::new("n"); + node.attrs + .insert("max_retries".to_string(), AttrValue::Integer(3)); + let graph = Graph::new("test"); + let policy = build_retry_policy(&node, &graph); + assert_eq!(policy.max_attempts, 4); + assert_eq!(policy.backoff.initial_delay, Duration::from_millis(5_000)); + } + + #[test] + fn build_retry_policy_all_presets() { + let presets = [ + ("none", 1u32), + ("standard", 5), + ("aggressive", 5), + ("linear", 3), + ("patient", 3), + ]; + let graph = Graph::new("test"); + let (name, expected) = presets[0]; + let mut node = Node::new("n"); + node.attrs.insert( + "retry_policy".to_string(), + AttrValue::String(name.to_string()), + ); + assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); + + let (name, expected) = presets[1]; + node.attrs.insert( + "retry_policy".to_string(), + AttrValue::String(name.to_string()), + ); + assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); + + let (name, expected) = presets[2]; + node.attrs.insert( + "retry_policy".to_string(), + AttrValue::String(name.to_string()), + ); + assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); + + let (name, expected) = presets[3]; + node.attrs.insert( + "retry_policy".to_string(), + AttrValue::String(name.to_string()), + ); + assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); + + let (name, expected) = presets[4]; + node.attrs.insert( + "retry_policy".to_string(), + AttrValue::String(name.to_string()), + ); + assert_eq!(build_retry_policy(&node, &graph).max_attempts, expected); + } + + #[test] + fn build_retry_policy_unknown_preset_falls_back() { + let mut node = Node::new("n"); + node.attrs.insert( + "retry_policy".to_string(), + AttrValue::String("unknown_preset".to_string()), + ); + let graph = Graph::new("test"); + let policy = build_retry_policy(&node, &graph); + assert_eq!(policy.max_attempts, 1); + } +}