diff --git a/Cargo.lock b/Cargo.lock index d6a1a340b..a671e79c8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1403,6 +1403,18 @@ dependencies = [ "tracing", ] +[[package]] +name = "fabro-core" +version = "0.176.2" +dependencies = [ + "async-trait", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tracing", +] + [[package]] name = "fabro-db" version = "0.176.2" diff --git a/lib/crates/fabro-core/Cargo.toml b/lib/crates/fabro-core/Cargo.toml new file mode 100644 index 000000000..90f389d3e --- /dev/null +++ b/lib/crates/fabro-core/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "fabro-core" +edition.workspace = true +version.workspace = true +license.workspace = true +description = "Generic workflow execution engine" + +[lib] +doctest = false + +[dependencies] +async-trait.workspace = true +serde.workspace = true +serde_json.workspace = true +thiserror.workspace = true +tokio.workspace = true +tracing.workspace = true + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util", "macros"] } diff --git a/lib/crates/fabro-core/src/context.rs b/lib/crates/fabro-core/src/context.rs new file mode 100644 index 000000000..3e1ade0f0 --- /dev/null +++ b/lib/crates/fabro-core/src/context.rs @@ -0,0 +1,278 @@ +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; + +use serde_json::Value; + +pub trait ContextStore: Send + Sync { + fn set(&self, key: String, value: Value); + fn get(&self, key: &str) -> Option; + fn snapshot(&self) -> HashMap; + fn fork(&self) -> Arc; +} + +pub struct InMemoryStore { + data: RwLock>, +} + +impl InMemoryStore { + pub fn new() -> Self { + Self { + data: RwLock::new(HashMap::new()), + } + } +} + +impl Default for InMemoryStore { + fn default() -> Self { + Self::new() + } +} + +impl ContextStore for InMemoryStore { + fn set(&self, key: String, value: Value) { + self.data.write().unwrap().insert(key, value); + } + + fn get(&self, key: &str) -> Option { + self.data.read().unwrap().get(key).cloned() + } + + fn snapshot(&self) -> HashMap { + self.data.read().unwrap().clone() + } + + fn fork(&self) -> Arc { + let cloned = self.data.read().unwrap().clone(); + Arc::new(InMemoryStore { + data: RwLock::new(cloned), + }) + } +} + +#[derive(Clone)] +pub struct Context { + store: Arc, + logs: Arc>>, +} + +impl Default for Context { + fn default() -> Self { + Self::new() + } +} + +impl Context { + pub fn new() -> Self { + Self { + store: Arc::new(InMemoryStore::new()), + logs: Arc::new(RwLock::new(Vec::new())), + } + } + + pub fn with_store(store: Arc) -> Self { + Self { + store, + logs: Arc::new(RwLock::new(Vec::new())), + } + } + + pub fn set(&self, key: impl Into, value: Value) { + self.store.set(key.into(), value); + } + + pub fn get(&self, key: &str) -> Option { + self.store.get(key) + } + + pub fn get_string(&self, key: &str, default: &str) -> String { + self.get(key) + .and_then(|v| v.as_str().map(String::from)) + .unwrap_or_else(|| default.to_string()) + } + + pub fn apply_updates(&self, updates: &HashMap) { + for (k, v) in updates { + self.store.set(k.clone(), v.clone()); + } + } + + pub fn snapshot(&self) -> HashMap { + self.store.snapshot() + } + + pub fn append_log(&self, entry: impl Into) { + self.logs.write().unwrap().push(entry.into()); + } + + pub fn logs_snapshot(&self) -> Vec { + self.logs.read().unwrap().clone() + } + + pub fn clone_context(&self) -> Self { + Self { + store: self.store.fork(), + logs: Arc::new(RwLock::new(self.logs.read().unwrap().clone())), + } + } + + // Core typed accessors + pub fn current_node_id(&self) -> String { + self.get_string("current_node", "") + } + + pub fn node_visit_count(&self) -> usize { + self.get("internal.node_visit_count") + .and_then(|v| v.as_u64()) + .unwrap_or(0) as usize + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[test] + fn in_memory_store_set_and_get() { + let store = InMemoryStore::new(); + store.set("k".into(), json!("v")); + assert_eq!(store.get("k"), Some(json!("v"))); + assert_eq!(store.get("missing"), None); + } + + #[test] + fn in_memory_store_snapshot_is_independent() { + let store = InMemoryStore::new(); + store.set("a".into(), json!(1)); + let snap = store.snapshot(); + store.set("b".into(), json!(2)); + assert!(!snap.contains_key("b")); + assert_eq!(snap.len(), 1); + } + + #[test] + fn in_memory_store_fork() { + let store = InMemoryStore::new(); + store.set("x".into(), json!(10)); + let forked = store.fork(); + forked.set("y".into(), json!(20)); + assert!(store.get("y").is_none()); + assert_eq!(forked.get("x"), Some(json!(10))); + assert_eq!(forked.get("y"), Some(json!(20))); + } + + #[test] + fn context_set_and_get() { + let ctx = Context::new(); + ctx.set("name", json!("test")); + assert_eq!(ctx.get("name"), Some(json!("test"))); + } + + #[test] + fn context_get_missing_returns_none() { + let ctx = Context::new(); + assert_eq!(ctx.get("nope"), None); + } + + #[test] + fn context_get_string_with_default() { + let ctx = Context::new(); + assert_eq!(ctx.get_string("missing", "fallback"), "fallback"); + ctx.set("present", json!("value")); + assert_eq!(ctx.get_string("present", "fallback"), "value"); + } + + #[test] + fn context_apply_updates() { + let ctx = Context::new(); + let mut updates = HashMap::new(); + updates.insert("a".into(), json!(1)); + updates.insert("b".into(), json!(2)); + ctx.apply_updates(&updates); + assert_eq!(ctx.get("a"), Some(json!(1))); + assert_eq!(ctx.get("b"), Some(json!(2))); + } + + #[test] + fn context_clone_is_independent() { + let ctx = Context::new(); + ctx.set("x", json!(1)); + let cloned = ctx.clone_context(); + cloned.set("x", json!(2)); + assert_eq!(ctx.get("x"), Some(json!(1))); + assert_eq!(cloned.get("x"), Some(json!(2))); + } + + #[test] + fn context_append_and_snapshot_logs() { + let ctx = Context::new(); + ctx.append_log("step 1"); + ctx.append_log("step 2"); + let logs = ctx.logs_snapshot(); + assert_eq!(logs, vec!["step 1", "step 2"]); + } + + #[test] + fn context_with_custom_store() { + struct CountingStore { + inner: InMemoryStore, + set_count: AtomicUsize, + } + impl ContextStore for CountingStore { + fn set(&self, key: String, value: Value) { + self.set_count.fetch_add(1, Ordering::Relaxed); + self.inner.set(key, value); + } + fn get(&self, key: &str) -> Option { + self.inner.get(key) + } + fn snapshot(&self) -> HashMap { + self.inner.snapshot() + } + fn fork(&self) -> Arc { + self.inner.fork() + } + } + + let store = Arc::new(CountingStore { + inner: InMemoryStore::new(), + set_count: AtomicUsize::new(0), + }); + let ctx = Context::with_store(store.clone()); + ctx.set("k", json!(1)); + ctx.set("k2", json!(2)); + assert_eq!(store.set_count.load(Ordering::Relaxed), 2); + assert_eq!(ctx.get("k"), Some(json!(1))); + } + + #[test] + fn context_fork_is_independent() { + let ctx = Context::new(); + ctx.set("shared", json!("original")); + ctx.append_log("log1"); + let forked = ctx.clone_context(); + forked.set("shared", json!("modified")); + forked.append_log("log2"); + assert_eq!(ctx.get("shared"), Some(json!("original"))); + assert_eq!(ctx.logs_snapshot().len(), 1); + assert_eq!(forked.get("shared"), Some(json!("modified"))); + assert_eq!(forked.logs_snapshot().len(), 2); + } + + #[test] + fn context_current_node_id() { + let ctx = Context::new(); + assert_eq!(ctx.current_node_id(), ""); + ctx.set("current_node", json!("node_5")); + assert_eq!(ctx.current_node_id(), "node_5"); + } + + #[test] + fn context_node_visit_count() { + let ctx = Context::new(); + assert_eq!(ctx.node_visit_count(), 0); + ctx.set("internal.node_visit_count", json!(3)); + assert_eq!(ctx.node_visit_count(), 3); + } +} diff --git a/lib/crates/fabro-core/src/error.rs b/lib/crates/fabro-core/src/error.rs new file mode 100644 index 000000000..8d6b76f52 --- /dev/null +++ b/lib/crates/fabro-core/src/error.rs @@ -0,0 +1,169 @@ +use std::fmt; +use std::time::Duration; + +use crate::outcome::{FailureDetail, Outcome, StageStatus}; + +/// Structured failure data on handler errors. Maps to FabroError's +/// is_retryable(), failure_class(), failure_signature_hint(), to_fail_outcome(). +#[derive(Debug, Clone)] +pub struct HandlerErrorDetail { + pub message: String, + pub retryable: bool, + pub category: Option, + pub signature: Option, +} + +impl fmt::Display for HandlerErrorDetail { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.message) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum CoreError { + #[error("node not found: {id}")] + NodeNotFound { id: String }, + #[error("no start node found in graph")] + NoStartNode, + #[error("run cancelled")] + Cancelled, + #[error("blocked: {message}")] + Blocked { message: String }, + #[error("node \"{node_id}\" visited {visits} times (limit {limit})")] + VisitLimitExceeded { + node_id: String, + visits: usize, + limit: usize, + }, + #[error("stall timeout: no activity for {elapsed:?}")] + StallTimeout { elapsed: Duration }, + #[error("{detail}")] + Handler { detail: HandlerErrorDetail }, + #[error("{0}")] + Other(String), +} + +impl CoreError { + pub fn handler(detail: HandlerErrorDetail) -> Self { + Self::Handler { detail } + } + + pub fn blocked(message: impl Into) -> Self { + Self::Blocked { + message: message.into(), + } + } + + pub fn is_retryable(&self) -> bool { + matches!(self, Self::Handler { detail } if detail.retryable) + } + + pub fn to_fail_outcome(&self) -> Outcome { + match self { + Self::Handler { detail } => Outcome { + status: StageStatus::Fail, + failure: Some(FailureDetail { + message: detail.message.clone(), + category: detail.category.clone(), + signature: detail.signature.clone(), + }), + ..Outcome::default() + }, + other => Outcome::fail(&other.to_string()), + } + } +} + +pub type Result = std::result::Result; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn core_error_display() { + assert_eq!( + CoreError::NodeNotFound { id: "n1".into() }.to_string(), + "node not found: n1" + ); + assert_eq!( + CoreError::NoStartNode.to_string(), + "no start node found in graph" + ); + assert_eq!(CoreError::Cancelled.to_string(), "run cancelled"); + assert_eq!( + CoreError::Blocked { + message: "hook denied".into() + } + .to_string(), + "blocked: hook denied" + ); + assert_eq!( + CoreError::VisitLimitExceeded { + node_id: "n1".into(), + visits: 5, + limit: 3 + } + .to_string(), + "node \"n1\" visited 5 times (limit 3)" + ); + assert_eq!( + CoreError::StallTimeout { + elapsed: Duration::from_secs(30) + } + .to_string(), + "stall timeout: no activity for 30s" + ); + assert_eq!( + CoreError::Other("something broke".into()).to_string(), + "something broke" + ); + } + + #[test] + fn core_error_handler_is_retryable() { + let retryable = CoreError::handler(HandlerErrorDetail { + message: "timeout".into(), + retryable: true, + category: None, + signature: None, + }); + assert!(retryable.is_retryable()); + + let not_retryable = CoreError::handler(HandlerErrorDetail { + message: "bad input".into(), + retryable: false, + category: None, + signature: None, + }); + assert!(!not_retryable.is_retryable()); + } + + #[test] + fn core_error_handler_to_fail_outcome() { + let err = CoreError::handler(HandlerErrorDetail { + message: "api down".into(), + retryable: true, + category: Some("transient".into()), + signature: Some("sig123".into()), + }); + let outcome = err.to_fail_outcome(); + assert_eq!(outcome.status, StageStatus::Fail); + let failure = outcome.failure.unwrap(); + assert_eq!(failure.message, "api down"); + assert_eq!(failure.category.as_deref(), Some("transient")); + assert_eq!(failure.signature.as_deref(), Some("sig123")); + } + + #[test] + fn core_error_non_handler_not_retryable() { + assert!(!CoreError::NodeNotFound { id: "x".into() }.is_retryable()); + assert!(!CoreError::Cancelled.is_retryable()); + assert!(!CoreError::NoStartNode.is_retryable()); + assert!(!CoreError::Blocked { + message: "no".into() + } + .is_retryable()); + assert!(!CoreError::Other("err".into()).is_retryable()); + } +} diff --git a/lib/crates/fabro-core/src/executor.rs b/lib/crates/fabro-core/src/executor.rs new file mode 100644 index 000000000..a207bd1e0 --- /dev/null +++ b/lib/crates/fabro-core/src/executor.rs @@ -0,0 +1,1446 @@ +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Instant; + +use crate::error::{CoreError, Result}; +use crate::graph::{EdgeSpec, Graph, NodeSpec}; +use crate::handler::NodeHandler; +use crate::lifecycle::{ + AttemptContext, AttemptResultContext, EdgeContext, EdgeDecision, NodeDecision, NoopLifecycle, + RunLifecycle, +}; +use crate::outcome::{NodeResult, Outcome, StageStatus}; +use crate::state::RunState; + +#[derive(Default)] +pub struct ExecutorSettings { + pub cancel_token: Option>, + pub max_node_visits: Option, +} + +pub struct Executor { + handler: Arc>, + lifecycle: Box>, + settings: ExecutorSettings, +} + +enum NextStep { + Edge(String), + Jump(String), + LoopRestart(String), + End, +} + +pub struct ExecutorBuilder { + handler: Arc>, + lifecycle: Option>>, + settings: ExecutorSettings, +} + +impl ExecutorBuilder { + pub fn new(handler: Arc>) -> Self { + Self { + handler, + lifecycle: None, + settings: ExecutorSettings::default(), + } + } + + pub fn lifecycle(mut self, lifecycle: Box>) -> Self { + self.lifecycle = Some(lifecycle); + self + } + + pub fn cancel_token(mut self, token: Arc) -> Self { + self.settings.cancel_token = Some(token); + self + } + + pub fn max_node_visits(mut self, limit: usize) -> Self { + self.settings.max_node_visits = Some(limit); + self + } + + pub fn build(self) -> Executor { + Executor { + handler: self.handler, + lifecycle: self.lifecycle.unwrap_or_else(|| Box::new(NoopLifecycle)), + settings: self.settings, + } + } +} + +impl Executor { + pub async fn run(&self, graph: &G, mut state: RunState) -> Result { + self.lifecycle.on_run_start(graph, &state).await?; + + loop { + // Check cancellation + if let Some(ref token) = self.settings.cancel_token { + if token.load(Ordering::Relaxed) { + let outcome = Outcome::fail("run cancelled"); + self.lifecycle.on_run_end(&outcome, &state).await; + return Err(CoreError::Cancelled); + } + } + + let node = state + .current_node(graph) + .ok_or_else(|| CoreError::NodeNotFound { + id: state.current_node_id.clone(), + })?; + + // Terminal nodes: skip normal lifecycle, call on_terminal_reached, check goal gates + if node.is_terminal() { + self.lifecycle.on_terminal_reached(&node, &state).await; + match graph.check_goal_gates(&state.node_outcomes) { + Ok(()) => { + let outcome = Outcome::success(); + self.lifecycle.on_run_end(&outcome, &state).await; + return Ok(outcome); + } + Err(msg) => { + // Check if there's a retry target for goal gate failure + if let Some(retry_target) = graph.get_retry_target(&state.current_node_id) { + tracing::debug!( + node = %node.id(), + retry_target = %retry_target, + reason = %msg, + "Goal gate unsatisfied, retrying" + ); + state.advance(&retry_target); + continue; + } + let outcome = Outcome::fail(&msg); + self.lifecycle.on_run_end(&outcome, &state).await; + return Ok(outcome); + } + } + } + + // Check visit limits + let visits = state.increment_visits(node.id()); + if let Some(max) = node.max_visits() { + if visits > max { + return Err(CoreError::VisitLimitExceeded { + node_id: node.id().to_string(), + visits, + limit: max, + }); + } + } + if let Some(global_max) = self.settings.max_node_visits { + if visits > global_max { + return Err(CoreError::VisitLimitExceeded { + node_id: node.id().to_string(), + visits, + limit: global_max, + }); + } + } + + // before_node lifecycle + match self.lifecycle.before_node(&node, &state).await? { + NodeDecision::Skip(outcome) => { + let mut result = NodeResult::from_skip(*outcome); + self.lifecycle + .after_node(&node, &mut result, &state) + .await?; + state.record(node.id(), &result); + self.lifecycle.on_checkpoint(&node, &result, &state).await?; + } + NodeDecision::Block(msg) => { + return Err(CoreError::blocked(msg)); + } + NodeDecision::Continue => { + // Execute with retry + let mut result = self.execute_with_retry(&node, &state, graph).await?; + self.lifecycle + .after_node(&node, &mut result, &state) + .await?; + state.record(node.id(), &result); + self.lifecycle.on_checkpoint(&node, &result, &state).await?; + } + } + + // Determine next step + let last_outcome = state.node_outcomes.get(node.id()).unwrap(); + let next = self + .resolve_next_step(&node, last_outcome, &state, graph) + .await?; + + match next { + NextStep::Edge(target) | NextStep::Jump(target) => { + state.advance(&target); + } + NextStep::LoopRestart(start_id) => { + state.restart(&start_id); + self.lifecycle.on_run_start(graph, &state).await?; + } + NextStep::End => { + let outcome = last_outcome.clone(); + self.lifecycle.on_run_end(&outcome, &state).await; + return Ok(outcome); + } + } + } + } + + async fn execute_with_retry( + &self, + node: &G::Node, + state: &RunState, + graph: &G, + ) -> Result { + let policy = self.handler.retry_policy(node, graph); + let start = Instant::now(); + + for attempt in 1..=policy.max_attempts { + let attempt_ctx = AttemptContext { + node, + attempt, + max_attempts: policy.max_attempts, + }; + match self.lifecycle.before_attempt(&attempt_ctx, state).await? { + NodeDecision::Skip(o) => return Ok(NodeResult::from_skip(*o)), + NodeDecision::Block(msg) => return Err(CoreError::blocked(msg)), + NodeDecision::Continue => {} + } + + let can_retry = attempt < policy.max_attempts; + + match self.handler.execute(node, &state.context, graph).await { + Ok(outcome) if outcome.status == StageStatus::Retry && can_retry => { + let delay = policy.backoff.delay_for_attempt(attempt); + let result = + NodeResult::new(outcome, start.elapsed(), attempt, policy.max_attempts); + let ctx = AttemptResultContext { + node, + result: &result, + attempt, + will_retry: true, + backoff_delay: Some(delay), + }; + self.lifecycle.after_attempt(&ctx, state).await?; + tokio::time::sleep(delay).await; + } + Ok(outcome) if outcome.status == StageStatus::Retry => { + let final_outcome = self.handler.on_retries_exhausted(node, outcome); + let result = NodeResult::new( + final_outcome, + start.elapsed(), + attempt, + policy.max_attempts, + ); + let ctx = AttemptResultContext { + node, + result: &result, + attempt, + will_retry: false, + backoff_delay: None, + }; + self.lifecycle.after_attempt(&ctx, state).await?; + return Ok(result); + } + Ok(outcome) => { + let result = + NodeResult::new(outcome, start.elapsed(), attempt, policy.max_attempts); + let ctx = AttemptResultContext { + node, + result: &result, + attempt, + will_retry: false, + backoff_delay: None, + }; + self.lifecycle.after_attempt(&ctx, state).await?; + return Ok(result); + } + Err(e) if can_retry && e.is_retryable() => { + let delay = policy.backoff.delay_for_attempt(attempt); + let fail_result = + NodeResult::from_error(&e, start.elapsed(), attempt, policy.max_attempts); + let ctx = AttemptResultContext { + node, + result: &fail_result, + attempt, + will_retry: true, + backoff_delay: Some(delay), + }; + self.lifecycle.after_attempt(&ctx, state).await?; + tokio::time::sleep(delay).await; + } + Err(e) => { + let fail_result = + NodeResult::from_error(&e, start.elapsed(), attempt, policy.max_attempts); + let ctx = AttemptResultContext { + node, + result: &fail_result, + attempt, + will_retry: false, + backoff_delay: None, + }; + self.lifecycle.after_attempt(&ctx, state).await?; + return Err(e); + } + } + } + unreachable!("loop always returns or continues") + } + + async fn resolve_next_step( + &self, + node: &G::Node, + outcome: &Outcome, + state: &RunState, + graph: &G, + ) -> Result { + // Jump takes priority + if let Some(ref target) = outcome.jump_to_node { + let ctx = EdgeContext { + from: node.id(), + to: target, + edge: None, + is_jump: true, + outcome, + reason: "jump", + }; + match self.lifecycle.on_edge_selected(&ctx, state).await? { + EdgeDecision::Continue => return Ok(NextStep::Jump(target.clone())), + EdgeDecision::Override(new_target) => return Ok(NextStep::Edge(new_target)), + EdgeDecision::Block(msg) => return Err(CoreError::blocked(msg)), + } + } + + // Normal edge selection + match graph.select_edge(node, outcome, &state.context) { + Some(selection) => { + let target = selection.edge.target().to_string(); + let is_restart = selection.edge.is_loop_restart(); + + let ctx = EdgeContext { + from: node.id(), + to: &target, + edge: Some(selection.edge.clone()), + is_jump: false, + outcome, + reason: selection.reason, + }; + match self.lifecycle.on_edge_selected(&ctx, state).await? { + EdgeDecision::Continue => { + if is_restart { + let start = graph.find_start_node()?; + Ok(NextStep::LoopRestart(start.id().to_string())) + } else { + Ok(NextStep::Edge(target)) + } + } + EdgeDecision::Override(new_target) => Ok(NextStep::Edge(new_target)), + EdgeDecision::Block(msg) => Err(CoreError::blocked(msg)), + } + } + None => { + // No edge found + if outcome.status == StageStatus::Success + || outcome.status == StageStatus::PartialSuccess + { + Ok(NextStep::End) + } else { + // Fail with no outgoing edge → fail the run + Ok(NextStep::End) + } + } + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use async_trait::async_trait; + + use super::*; + use crate::context::Context; + use crate::error::HandlerErrorDetail; + use crate::lifecycle::RunLifecycle; + use crate::retry::{BackoffPolicy, RetryPolicy}; + use crate::test_fixtures::*; + + // Helper to build and run an executor with default settings + async fn run_linear( + node_ids: &[&str], + handler: Arc>, + ) -> Result { + let g = linear_graph(node_ids); + let state = RunState::new(&g)?; + let executor = ExecutorBuilder::new(handler).build(); + executor.run(&g, state).await + } + + // ---- Step 8: Linear happy path ---- + + #[tokio::test] + async fn executor_linear_three_node_success() { + let result = run_linear(&["start", "work", "end"], Arc::new(AlwaysSucceedHandler)) + .await + .unwrap(); + assert_eq!(result.status, StageStatus::Success); + } + + #[tokio::test] + async fn executor_builder_sets_lifecycle() { + let log = Arc::new(Mutex::new(Vec::::new())); + struct LogLifecycle(Arc>>); + #[async_trait] + impl RunLifecycle for LogLifecycle { + async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> { + self.0.lock().unwrap().push("start".into()); + Ok(()) + } + } + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(LogLifecycle(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + assert_eq!(log.lock().unwrap().clone(), vec!["start"]); + } + + #[tokio::test] + async fn executor_builder_sets_cancel_token() { + let token = Arc::new(AtomicBool::new(true)); // already cancelled + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .cancel_token(token) + .build(); + let result = executor.run(&g, state).await; + assert!(matches!(result, Err(CoreError::Cancelled))); + } + + // ---- Step 9: Terminal nodes, goal gates, visit limits ---- + + #[tokio::test] + async fn executor_goal_gate_satisfied() { + let g = TestGraph::new( + vec![ + TestNode::new("work"), + TestNode::terminal("end").with_goal_gate("work", StageStatus::Success), + ], + vec![TestEdge::new("work", "end")], + "work", + ); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + } + + #[tokio::test] + async fn executor_goal_gate_unsatisfied_with_retry() { + // work → end (goal gate: work must be success) + // retry_target: end → work (so we go back to work) + // First call fails, second succeeds + let g = TestGraph::new( + vec![ + TestNode::new("work"), + TestNode::terminal("end").with_goal_gate("work", StageStatus::Success), + ], + vec![TestEdge::new("work", "end")], + "work", + ) + .with_retry_target("end", "work"); + + let handler = Arc::new(CountingHandler::new(vec![ + Ok(Outcome::fail("first attempt")), + Ok(Outcome::success()), + ])); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(handler.clone() as Arc>).build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + assert_eq!(handler.calls(), 2); + } + + #[tokio::test] + async fn executor_goal_gate_unsatisfied_no_retry_fails() { + let g = TestGraph::new( + vec![ + TestNode::new("work"), + TestNode::terminal("end").with_goal_gate("work", StageStatus::Success), + ], + vec![TestEdge::new("work", "end")], + "work", + ); + // No retry target, and handler fails + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new( + Arc::new(AlwaysFailHandler::new("nope")) as Arc> + ) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Fail); + } + + #[tokio::test] + async fn executor_terminal_node_skips_normal_lifecycle() { + let log = Arc::new(Mutex::new(Vec::::new())); + struct TrackingLifecycle(Arc>>); + #[async_trait] + impl RunLifecycle for TrackingLifecycle { + async fn before_node(&self, node: &TestNode, _s: &RunState) -> Result { + self.0 + .lock() + .unwrap() + .push(format!("before_node:{}", node.id())); + Ok(NodeDecision::Continue) + } + async fn after_node( + &self, + node: &TestNode, + _r: &mut NodeResult, + _s: &RunState, + ) -> Result<()> { + self.0 + .lock() + .unwrap() + .push(format!("after_node:{}", node.id())); + Ok(()) + } + async fn on_terminal_reached(&self, node: &TestNode, _s: &RunState) { + self.0 + .lock() + .unwrap() + .push(format!("terminal:{}", node.id())); + } + } + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(TrackingLifecycle(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + let calls = log.lock().unwrap().clone(); + // before_node and after_node called for "start", NOT for "end" + assert!(calls.contains(&"before_node:start".to_string())); + assert!(calls.contains(&"after_node:start".to_string())); + assert!(!calls.contains(&"before_node:end".to_string())); + assert!(!calls.contains(&"after_node:end".to_string())); + // on_terminal_reached IS called for "end" + assert!(calls.contains(&"terminal:end".to_string())); + } + + #[tokio::test] + async fn executor_terminal_node_calls_on_terminal_reached() { + let log = Arc::new(Mutex::new(Vec::::new())); + struct TerminalTracker(Arc>>); + #[async_trait] + impl RunLifecycle for TerminalTracker { + async fn on_terminal_reached(&self, node: &TestNode, _s: &RunState) { + self.0 + .lock() + .unwrap() + .push(format!("terminal:{}", node.id())); + } + } + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(TerminalTracker(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + assert_eq!(log.lock().unwrap().clone(), vec!["terminal:end"]); + } + + #[tokio::test] + async fn executor_visit_limit_per_node() { + // Node with max_visits=1, but graph loops back to it + let g = TestGraph::new( + vec![ + TestNode::new("loop_node").with_max_visits(1), + TestNode::new("other"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("loop_node", "other"), + TestEdge::new("other", "loop_node"), + ], + "loop_node", + ); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .build(); + let result = executor.run(&g, state).await; + assert!(matches!(result, Err(CoreError::VisitLimitExceeded { .. }))); + } + + #[tokio::test] + async fn executor_visit_limit_global() { + let g = TestGraph::new( + vec![ + TestNode::new("a"), + TestNode::new("b"), + TestNode::terminal("end"), + ], + vec![TestEdge::new("a", "b"), TestEdge::new("b", "a")], + "a", + ); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .max_node_visits(3) + .build(); + let result = executor.run(&g, state).await; + assert!(matches!(result, Err(CoreError::VisitLimitExceeded { .. }))); + } + + // ---- Step 10: Edge selection, jumps, loop restarts ---- + + #[tokio::test] + async fn executor_conditional_edge_on_fail() { + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::terminal("ok"), + TestNode::terminal("bad"), + ], + vec![ + TestEdge::new("start", "ok").with_label("success"), + TestEdge::new("start", "bad").with_label("fail"), + ], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new( + Arc::new(AlwaysFailHandler::new("oops")) as Arc> + ) + .build(); + let result = executor.run(&g, state).await.unwrap(); + // Ends at "bad" terminal with success (goal gates pass since no gates defined) + assert_eq!(result.status, StageStatus::Success); + } + + #[tokio::test] + async fn executor_conditional_edge_on_success() { + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::terminal("ok"), + TestNode::terminal("bad"), + ], + vec![ + TestEdge::new("start", "ok").with_label("success"), + TestEdge::new("start", "bad").with_label("fail"), + ], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + } + + #[tokio::test] + async fn executor_jump_bypasses_edge_selection() { + // start → end (normal), but handler says jump to "target" + struct JumpHandler; + #[async_trait] + impl NodeHandler for JumpHandler { + async fn execute( + &self, + _n: &TestNode, + _c: &Context, + _g: &TestGraph, + ) -> Result { + let mut o = Outcome::success(); + o.jump_to_node = Some("target".into()); + Ok(o) + } + } + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::terminal("end"), + TestNode::terminal("target"), + ], + vec![TestEdge::new("start", "end")], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(JumpHandler) as Arc>).build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + } + + #[tokio::test] + async fn executor_loop_restart_resets_state() { + // start → work → (loop_restart edge back) → start → work → end + let handler = Arc::new(CountingHandler::new(vec![ + Ok(Outcome::success()), // start (1st) + Ok({ + let mut o = Outcome::success(); + o.preferred_label = Some("retry".into()); + o + }), // work (1st) → triggers loop restart + Ok(Outcome::success()), // start (2nd) + Ok(Outcome::success()), // work (2nd) → no label match, takes unconditional to end + ])); + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::new("work"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("start", "work"), + TestEdge::new("work", "start") + .with_label("retry") + .with_loop_restart(), + TestEdge::new("work", "end"), + ], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new(handler.clone() as Arc>) + .max_node_visits(5) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + assert_eq!(handler.calls(), 4); + } + + #[tokio::test] + async fn executor_loop_restart_calls_on_run_start() { + let log = Arc::new(Mutex::new(Vec::::new())); + struct StartTracker(Arc>>); + #[async_trait] + impl RunLifecycle for StartTracker { + async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> { + self.0.lock().unwrap().push("on_run_start".into()); + Ok(()) + } + } + let handler = Arc::new(CountingHandler::new(vec![ + Ok(Outcome::success()), + Ok({ + let mut o = Outcome::success(); + o.preferred_label = Some("retry".into()); + o + }), + Ok(Outcome::success()), + Ok(Outcome::success()), + ])); + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::new("work"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("start", "work"), + TestEdge::new("work", "start") + .with_label("retry") + .with_loop_restart(), + TestEdge::new("work", "end"), + ], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new(handler as Arc>) + .lifecycle(Box::new(StartTracker(log.clone()))) + .max_node_visits(5) + .build(); + executor.run(&g, state).await.unwrap(); + // on_run_start should be called twice: initial + after restart + assert_eq!(log.lock().unwrap().len(), 2); + } + + #[tokio::test] + async fn executor_fail_no_edge_returns_fail() { + // Node fails with no "fail" edge → run ends with that outcome + let g = TestGraph::new( + vec![TestNode::new("start"), TestNode::terminal("end")], + vec![TestEdge::new("start", "end").with_label("success")], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new( + Arc::new(AlwaysFailHandler::new("boom")) as Arc> + ) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Fail); + } + + #[tokio::test] + async fn executor_no_edge_after_success_returns_success() { + // Node succeeds with no outgoing edges → run ends with success + let g = TestGraph::new(vec![TestNode::new("only")], vec![], "only"); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + } + + // ---- Step 11: Cancellation ---- + + #[tokio::test] + async fn executor_cancellation_stops_run() { + let token = Arc::new(AtomicBool::new(false)); + let token_clone = token.clone(); + + struct CancellingHandler(Arc); + #[async_trait] + impl NodeHandler for CancellingHandler { + async fn execute( + &self, + _n: &TestNode, + _c: &Context, + _g: &TestGraph, + ) -> Result { + // Cancel after first node + self.0.store(true, Ordering::Relaxed); + Ok(Outcome::success()) + } + } + + let g = linear_graph(&["start", "work", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new( + Arc::new(CancellingHandler(token_clone)) as Arc> + ) + .cancel_token(token) + .build(); + let result = executor.run(&g, state).await; + assert!(matches!(result, Err(CoreError::Cancelled))); + } + + // ---- Step 12: Retry integration ---- + + #[tokio::test] + async fn executor_retry_on_retryable_error() { + let handler = Arc::new( + CountingHandler::new(vec![ + Err(CoreError::handler(HandlerErrorDetail { + message: "fail1".into(), + retryable: true, + category: None, + signature: None, + })), + Err(CoreError::handler(HandlerErrorDetail { + message: "fail2".into(), + retryable: true, + category: None, + signature: None, + })), + Ok(Outcome::success()), + ]) + .with_retry_policy(RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(1), + factor: 1.0, + max_delay: Duration::from_millis(1), + jitter: false, + }, + }), + ); + let result = run_linear( + &["start", "end"], + handler.clone() as Arc>, + ) + .await + .unwrap(); + assert_eq!(result.status, StageStatus::Success); + assert_eq!(handler.calls(), 3); + } + + #[tokio::test] + async fn executor_retry_on_retry_status() { + let handler = Arc::new( + CountingHandler::new(vec![ + Ok(Outcome { + status: StageStatus::Retry, + ..Outcome::default() + }), + Ok(Outcome { + status: StageStatus::Retry, + ..Outcome::default() + }), + Ok(Outcome::success()), + ]) + .with_retry_policy(RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(1), + factor: 1.0, + max_delay: Duration::from_millis(1), + jitter: false, + }, + }), + ); + let result = run_linear( + &["start", "end"], + handler.clone() as Arc>, + ) + .await + .unwrap(); + assert_eq!(result.status, StageStatus::Success); + assert_eq!(handler.calls(), 3); + } + + #[tokio::test] + async fn executor_retry_non_retryable_error_no_retry() { + let handler = Arc::new( + CountingHandler::new(vec![Err(CoreError::handler(HandlerErrorDetail { + message: "fatal".into(), + retryable: false, + category: None, + signature: None, + }))]) + .with_retry_policy(RetryPolicy::with_max_attempts(3)), + ); + let result = run_linear( + &["start", "end"], + handler.clone() as Arc>, + ) + .await; + assert!(result.is_err()); + assert_eq!(handler.calls(), 1); + } + + #[tokio::test] + async fn executor_retry_no_retry_by_default() { + // Default policy is RetryPolicy::none() (max_attempts=1) + let handler = Arc::new(CountingHandler::new(vec![Err(CoreError::handler( + HandlerErrorDetail { + message: "fail".into(), + retryable: true, + category: None, + signature: None, + }, + ))])); + let result = run_linear( + &["start", "end"], + handler.clone() as Arc>, + ) + .await; + assert!(result.is_err()); + assert_eq!(handler.calls(), 1); + } + + #[tokio::test] + async fn executor_retry_exhausted_calls_on_retries_exhausted() { + struct ExhaustedHandler; + #[async_trait] + impl NodeHandler for ExhaustedHandler { + async fn execute( + &self, + _n: &TestNode, + _c: &Context, + _g: &TestGraph, + ) -> Result { + Ok(Outcome { + status: StageStatus::Retry, + ..Outcome::default() + }) + } + fn retry_policy(&self, _n: &TestNode, _g: &TestGraph) -> RetryPolicy { + RetryPolicy { + max_attempts: 2, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(1), + factor: 1.0, + max_delay: Duration::from_millis(1), + jitter: false, + }, + } + } + fn on_retries_exhausted(&self, _n: &TestNode, _last: Outcome) -> Outcome { + Outcome { + status: StageStatus::PartialSuccess, + notes: Some("exhausted".into()), + ..Outcome::default() + } + } + } + // No outgoing edges from "start" so PartialSuccess becomes the run result + let g = TestGraph::new(vec![TestNode::new("start")], vec![], "start"); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(ExhaustedHandler) as Arc>) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::PartialSuccess); + } + + #[tokio::test] + async fn executor_retry_lifecycle_before_attempt_called_per_attempt() { + let attempt_log = Arc::new(Mutex::new(Vec::::new())); + struct AttemptTracker(Arc>>); + #[async_trait] + impl RunLifecycle for AttemptTracker { + async fn before_attempt( + &self, + ctx: &AttemptContext<'_, TestGraph>, + _s: &RunState, + ) -> Result { + self.0.lock().unwrap().push(ctx.attempt); + Ok(NodeDecision::Continue) + } + } + let handler = Arc::new( + CountingHandler::new(vec![ + Err(CoreError::handler(HandlerErrorDetail { + message: "r".into(), + retryable: true, + category: None, + signature: None, + })), + Ok(Outcome::success()), + ]) + .with_retry_policy(RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(1), + factor: 1.0, + max_delay: Duration::from_millis(1), + jitter: false, + }, + }), + ); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new(handler as Arc>) + .lifecycle(Box::new(AttemptTracker(attempt_log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + assert_eq!(*attempt_log.lock().unwrap(), vec![1, 2]); + } + + #[tokio::test] + async fn executor_retry_lifecycle_after_attempt_called_with_will_retry() { + let retry_log = Arc::new(Mutex::new(Vec::<(u32, bool)>::new())); + struct RetryTracker(Arc>>); + #[async_trait] + impl RunLifecycle for RetryTracker { + async fn after_attempt( + &self, + ctx: &AttemptResultContext<'_, TestGraph>, + _s: &RunState, + ) -> Result<()> { + self.0.lock().unwrap().push((ctx.attempt, ctx.will_retry)); + Ok(()) + } + } + let handler = Arc::new( + CountingHandler::new(vec![ + Err(CoreError::handler(HandlerErrorDetail { + message: "r".into(), + retryable: true, + category: None, + signature: None, + })), + Ok(Outcome::success()), + ]) + .with_retry_policy(RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(1), + factor: 1.0, + max_delay: Duration::from_millis(1), + jitter: false, + }, + }), + ); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new(handler as Arc>) + .lifecycle(Box::new(RetryTracker(retry_log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + let log = retry_log.lock().unwrap().clone(); + assert_eq!(log, vec![(1, true), (2, false)]); + } + + #[tokio::test] + async fn executor_retry_lifecycle_before_attempt_skip_stops_retry() { + let call_count = Arc::new(std::sync::atomic::AtomicU32::new(0)); + let call_count_clone = call_count.clone(); + struct SkipOnSecondAttempt(Arc); + #[async_trait] + impl RunLifecycle for SkipOnSecondAttempt { + async fn before_attempt( + &self, + ctx: &AttemptContext<'_, TestGraph>, + _s: &RunState, + ) -> Result { + self.0.fetch_add(1, Ordering::Relaxed); + if ctx.attempt >= 2 { + Ok(NodeDecision::Skip(Box::new(Outcome::skipped("hook skip")))) + } else { + Ok(NodeDecision::Continue) + } + } + } + let handler = Arc::new( + CountingHandler::new(vec![ + Err(CoreError::handler(HandlerErrorDetail { + message: "r".into(), + retryable: true, + category: None, + signature: None, + })), + Ok(Outcome::success()), // should not be reached + ]) + .with_retry_policy(RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(1), + factor: 1.0, + max_delay: Duration::from_millis(1), + jitter: false, + }, + }), + ); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new(handler.clone() as Arc>) + .lifecycle(Box::new(SkipOnSecondAttempt(call_count_clone))) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); // overall run succeeds via terminal + assert_eq!(handler.calls(), 1); // handler only called once + assert_eq!(call_count.load(Ordering::Relaxed), 2); // before_attempt called twice + } + + #[tokio::test] + async fn executor_retry_backoff_delay() { + tokio::time::pause(); + let handler = Arc::new( + CountingHandler::new(vec![ + Ok(Outcome { + status: StageStatus::Retry, + ..Outcome::default() + }), + Ok(Outcome::success()), + ]) + .with_retry_policy(RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_secs(5), + factor: 2.0, + max_delay: Duration::from_secs(60), + jitter: false, + }, + }), + ); + let start = tokio::time::Instant::now(); + let result = run_linear( + &["start", "end"], + handler as Arc>, + ) + .await + .unwrap(); + assert_eq!(result.status, StageStatus::Success); + // Should have slept ~5s for the retry backoff + assert!(start.elapsed() >= Duration::from_secs(4)); + } + + // ---- Step 13: Full lifecycle integration ---- + + #[tokio::test] + async fn executor_lifecycle_before_node_skip() { + struct SkipFirst(Mutex); + #[async_trait] + impl RunLifecycle for SkipFirst { + async fn before_node(&self, node: &TestNode, _s: &RunState) -> Result { + if node.id() == "start" { + let mut skipped = self.0.lock().unwrap(); + if !*skipped { + *skipped = true; + return Ok(NodeDecision::Skip(Box::new(Outcome::skipped("hook")))); + } + } + Ok(NodeDecision::Continue) + } + } + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(SkipFirst(Mutex::new(false)))) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + } + + #[tokio::test] + async fn executor_lifecycle_before_node_block() { + struct Blocker; + #[async_trait] + impl RunLifecycle for Blocker { + async fn before_node(&self, _n: &TestNode, _s: &RunState) -> Result { + Ok(NodeDecision::Block("blocked".into())) + } + } + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(Blocker)) + .build(); + let result = executor.run(&g, state).await; + assert!(matches!(result, Err(CoreError::Blocked { .. }))); + } + + #[tokio::test] + async fn executor_lifecycle_after_node_mutates_result() { + struct Mutator; + #[async_trait] + impl RunLifecycle for Mutator { + async fn after_node( + &self, + _n: &TestNode, + result: &mut NodeResult, + _s: &RunState, + ) -> Result<()> { + result.outcome.notes = Some("mutated".into()); + Ok(()) + } + } + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(Mutator)) + .build(); + executor.run(&g, state).await.unwrap(); + // The mutation happened (verified by no error; could also check state) + } + + #[tokio::test] + async fn executor_lifecycle_on_edge_override() { + struct Redirector; + #[async_trait] + impl RunLifecycle for Redirector { + async fn on_edge_selected( + &self, + _ctx: &EdgeContext<'_, TestGraph>, + _s: &RunState, + ) -> Result { + Ok(EdgeDecision::Override("alt".into())) + } + } + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::terminal("end"), + TestNode::terminal("alt"), + ], + vec![TestEdge::new("start", "end")], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(Redirector)) + .build(); + let result = executor.run(&g, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + } + + #[tokio::test] + async fn executor_lifecycle_on_edge_block() { + struct EdgeBlocker; + #[async_trait] + impl RunLifecycle for EdgeBlocker { + async fn on_edge_selected( + &self, + _ctx: &EdgeContext<'_, TestGraph>, + _s: &RunState, + ) -> Result { + Ok(EdgeDecision::Block("edge blocked".into())) + } + } + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(EdgeBlocker)) + .build(); + let result = executor.run(&g, state).await; + assert!(matches!(result, Err(CoreError::Blocked { .. }))); + } + + #[tokio::test] + async fn executor_lifecycle_on_checkpoint_called() { + let log = Arc::new(Mutex::new(Vec::::new())); + struct CheckpointTracker(Arc>>); + #[async_trait] + impl RunLifecycle for CheckpointTracker { + async fn on_checkpoint( + &self, + node: &TestNode, + _r: &NodeResult, + _s: &RunState, + ) -> Result<()> { + self.0.lock().unwrap().push(node.id().to_string()); + Ok(()) + } + } + let g = linear_graph(&["start", "work", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(CheckpointTracker(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + assert_eq!(*log.lock().unwrap(), vec!["start", "work"]); + } + + #[tokio::test] + async fn executor_lifecycle_on_run_start_and_end_called() { + let log = Arc::new(Mutex::new(Vec::::new())); + struct RunTracker(Arc>>); + #[async_trait] + impl RunLifecycle for RunTracker { + async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> { + self.0.lock().unwrap().push("start".into()); + Ok(()) + } + async fn on_run_end(&self, _o: &Outcome, _s: &RunState) { + self.0.lock().unwrap().push("end".into()); + } + } + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(RunTracker(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + assert_eq!(*log.lock().unwrap(), vec!["start", "end"]); + } + + #[tokio::test] + async fn executor_lifecycle_on_edge_for_jumps() { + let log = Arc::new(Mutex::new(Vec::<(String, bool)>::new())); + struct JumpTracker(Arc>>); + #[async_trait] + impl RunLifecycle for JumpTracker { + async fn on_edge_selected( + &self, + ctx: &EdgeContext<'_, TestGraph>, + _s: &RunState, + ) -> Result { + self.0 + .lock() + .unwrap() + .push((ctx.to.to_string(), ctx.is_jump)); + Ok(EdgeDecision::Continue) + } + } + struct JumpHandler; + #[async_trait] + impl NodeHandler for JumpHandler { + async fn execute( + &self, + _n: &TestNode, + _c: &Context, + _g: &TestGraph, + ) -> Result { + let mut o = Outcome::success(); + o.jump_to_node = Some("target".into()); + Ok(o) + } + } + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::terminal("end"), + TestNode::terminal("target"), + ], + vec![TestEdge::new("start", "end")], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(JumpHandler) as Arc>) + .lifecycle(Box::new(JumpTracker(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + let entries = log.lock().unwrap().clone(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0], ("target".to_string(), true)); + } + + #[tokio::test] + async fn executor_context_updates_visible_to_next_node() { + use serde_json::json; + + struct ContextWriter; + #[async_trait] + impl NodeHandler for ContextWriter { + async fn execute( + &self, + node: &TestNode, + context: &Context, + _g: &TestGraph, + ) -> Result { + if node.id() == "start" { + let mut o = Outcome::success(); + o.context_updates.insert("shared".into(), json!("hello")); + Ok(o) + } else { + let val = context.get_string("shared", "missing"); + let mut o = Outcome::success(); + o.notes = Some(val); + Ok(o) + } + } + } + let log = Arc::new(Mutex::new(Vec::::new())); + struct NoteCapture(Arc>>); + #[async_trait] + impl RunLifecycle for NoteCapture { + async fn after_node( + &self, + node: &TestNode, + result: &mut NodeResult, + _s: &RunState, + ) -> Result<()> { + if node.id() == "work" { + if let Some(ref notes) = result.outcome.notes { + self.0.lock().unwrap().push(notes.clone()); + } + } + Ok(()) + } + } + + let g = linear_graph(&["start", "work", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(ContextWriter) as Arc>) + .lifecycle(Box::new(NoteCapture(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + assert_eq!(*log.lock().unwrap(), vec!["hello"]); + } +} diff --git a/lib/crates/fabro-core/src/graph.rs b/lib/crates/fabro-core/src/graph.rs new file mode 100644 index 000000000..19502f1bb --- /dev/null +++ b/lib/crates/fabro-core/src/graph.rs @@ -0,0 +1,42 @@ +use std::collections::HashMap; + +use crate::context::Context; +use crate::error::Result; +use crate::outcome::Outcome; + +pub trait NodeSpec: Send + Sync + Clone { + fn id(&self) -> &str; + fn is_terminal(&self) -> bool; + fn max_visits(&self) -> Option; +} + +pub trait EdgeSpec: Send + Sync + Clone { + fn target(&self) -> &str; + fn label(&self) -> Option<&str>; + fn is_loop_restart(&self) -> bool; +} + +pub struct EdgeSelection { + pub edge: G::Edge, + pub reason: &'static str, +} + +pub trait Graph: Send + Sync { + type Node: NodeSpec + Clone; + type Edge: EdgeSpec + Clone; + + fn get_node(&self, id: &str) -> Option; + fn find_start_node(&self) -> Result; + fn outgoing_edges(&self, node_id: &str) -> Vec; + fn select_edge( + &self, + node: &Self::Node, + outcome: &Outcome, + context: &Context, + ) -> Option>; + fn check_goal_gates( + &self, + outcomes: &HashMap, + ) -> std::result::Result<(), String>; + fn get_retry_target(&self, failed_node_id: &str) -> Option; +} diff --git a/lib/crates/fabro-core/src/handler.rs b/lib/crates/fabro-core/src/handler.rs new file mode 100644 index 000000000..ecad071eb --- /dev/null +++ b/lib/crates/fabro-core/src/handler.rs @@ -0,0 +1,20 @@ +use async_trait::async_trait; + +use crate::context::Context; +use crate::error::Result; +use crate::graph::Graph; +use crate::outcome::Outcome; +use crate::retry::RetryPolicy; + +#[async_trait] +pub trait NodeHandler: Send + Sync { + async fn execute(&self, node: &G::Node, context: &Context, graph: &G) -> Result; + + fn retry_policy(&self, _node: &G::Node, _graph: &G) -> RetryPolicy { + RetryPolicy::none() + } + + fn on_retries_exhausted(&self, _node: &G::Node, _last_outcome: Outcome) -> Outcome { + Outcome::fail("max retries exceeded") + } +} diff --git a/lib/crates/fabro-core/src/lib.rs b/lib/crates/fabro-core/src/lib.rs new file mode 100644 index 000000000..0c288b87f --- /dev/null +++ b/lib/crates/fabro-core/src/lib.rs @@ -0,0 +1,27 @@ +pub mod context; +pub mod error; +pub mod executor; +pub mod graph; +pub mod handler; +pub mod lifecycle; +pub mod outcome; +pub mod retry; +pub mod stall; +pub mod state; + +#[cfg(test)] +pub mod test_fixtures; + +pub use context::{Context, ContextStore, InMemoryStore}; +pub use error::{CoreError, HandlerErrorDetail, Result}; +pub use executor::{Executor, ExecutorBuilder, ExecutorSettings}; +pub use graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; +pub use handler::NodeHandler; +pub use lifecycle::{ + AttemptContext, AttemptResultContext, CompositeLifecycle, EdgeContext, EdgeDecision, + NodeDecision, NoopLifecycle, RunLifecycle, +}; +pub use outcome::{FailureDetail, NodeResult, Outcome, StageStatus}; +pub use retry::{BackoffPolicy, RetryPolicy}; +pub use stall::{ActivityMonitor, StallGuard, StallWatchdog}; +pub use state::RunState; diff --git a/lib/crates/fabro-core/src/lifecycle.rs b/lib/crates/fabro-core/src/lifecycle.rs new file mode 100644 index 000000000..4f8ea2160 --- /dev/null +++ b/lib/crates/fabro-core/src/lifecycle.rs @@ -0,0 +1,648 @@ +use std::time::Duration; + +use async_trait::async_trait; + +use crate::error::Result; +use crate::graph::Graph; +use crate::outcome::{NodeResult, Outcome}; +use crate::state::RunState; + +#[derive(Debug, Clone)] +pub enum NodeDecision { + Continue, + Skip(Box), + Block(String), +} + +#[derive(Debug, Clone)] +pub enum EdgeDecision { + Continue, + Override(String), + Block(String), +} + +pub struct AttemptContext<'a, G: Graph> { + pub node: &'a G::Node, + pub attempt: u32, + pub max_attempts: u32, +} + +pub struct AttemptResultContext<'a, G: Graph> { + pub node: &'a G::Node, + pub result: &'a NodeResult, + pub attempt: u32, + pub will_retry: bool, + pub backoff_delay: Option, +} + +pub struct EdgeContext<'a, G: Graph> { + pub from: &'a str, + pub to: &'a str, + pub edge: Option, + pub is_jump: bool, + pub outcome: &'a Outcome, + pub reason: &'a str, +} + +#[async_trait] +pub trait RunLifecycle: Send + Sync { + async fn on_run_start(&self, _graph: &G, _state: &RunState) -> Result<()> { + Ok(()) + } + + async fn on_terminal_reached(&self, _node: &G::Node, _state: &RunState) {} + + async fn before_node(&self, _node: &G::Node, _state: &RunState) -> Result { + Ok(NodeDecision::Continue) + } + + async fn before_attempt( + &self, + _ctx: &AttemptContext<'_, G>, + _state: &RunState, + ) -> Result { + Ok(NodeDecision::Continue) + } + + async fn after_attempt( + &self, + _ctx: &AttemptResultContext<'_, G>, + _state: &RunState, + ) -> Result<()> { + Ok(()) + } + + async fn after_node( + &self, + _node: &G::Node, + _result: &mut NodeResult, + _state: &RunState, + ) -> Result<()> { + Ok(()) + } + + async fn on_edge_selected( + &self, + _ctx: &EdgeContext<'_, G>, + _state: &RunState, + ) -> Result { + Ok(EdgeDecision::Continue) + } + + async fn on_checkpoint( + &self, + _node: &G::Node, + _result: &NodeResult, + _state: &RunState, + ) -> Result<()> { + Ok(()) + } + + async fn on_run_end(&self, _outcome: &Outcome, _state: &RunState) {} +} + +/// No-op lifecycle that passes through everything. +pub struct NoopLifecycle; + +#[async_trait] +impl RunLifecycle for NoopLifecycle {} + +/// Composes multiple lifecycles, calling them in order. Useful for testing +/// and simple use cases where fixed ordering suffices. +pub struct CompositeLifecycle { + children: Vec>>, +} + +impl CompositeLifecycle { + pub fn new(children: Vec>>) -> Self { + Self { children } + } +} + +#[async_trait] +impl RunLifecycle for CompositeLifecycle { + async fn on_run_start(&self, graph: &G, state: &RunState) -> Result<()> { + for child in &self.children { + child.on_run_start(graph, state).await?; + } + Ok(()) + } + + async fn on_terminal_reached(&self, node: &G::Node, state: &RunState) { + for child in &self.children { + child.on_terminal_reached(node, state).await; + } + } + + async fn before_node(&self, node: &G::Node, state: &RunState) -> Result { + for child in &self.children { + match child.before_node(node, state).await? { + NodeDecision::Continue => {} + decision => return Ok(decision), + } + } + Ok(NodeDecision::Continue) + } + + async fn before_attempt( + &self, + ctx: &AttemptContext<'_, G>, + state: &RunState, + ) -> Result { + for child in &self.children { + match child.before_attempt(ctx, state).await? { + NodeDecision::Continue => {} + decision => return Ok(decision), + } + } + Ok(NodeDecision::Continue) + } + + async fn after_attempt( + &self, + ctx: &AttemptResultContext<'_, G>, + state: &RunState, + ) -> Result<()> { + for child in &self.children { + child.after_attempt(ctx, state).await?; + } + Ok(()) + } + + async fn after_node( + &self, + node: &G::Node, + result: &mut NodeResult, + state: &RunState, + ) -> Result<()> { + for child in &self.children { + child.after_node(node, result, state).await?; + } + Ok(()) + } + + async fn on_edge_selected( + &self, + ctx: &EdgeContext<'_, G>, + state: &RunState, + ) -> Result { + for child in &self.children { + match child.on_edge_selected(ctx, state).await? { + EdgeDecision::Continue => {} + decision => return Ok(decision), + } + } + Ok(EdgeDecision::Continue) + } + + async fn on_checkpoint( + &self, + node: &G::Node, + result: &NodeResult, + state: &RunState, + ) -> Result<()> { + for child in &self.children { + child.on_checkpoint(node, result, state).await?; + } + Ok(()) + } + + async fn on_run_end(&self, outcome: &Outcome, state: &RunState) { + for child in &self.children { + child.on_run_end(outcome, state).await; + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::{Arc, Mutex}; + + use super::*; + use crate::test_fixtures::{linear_graph, TestGraph, TestNode}; + + /// A lifecycle that records which callbacks were called. + struct RecordingLifecycle { + name: String, + log: Arc>>, + before_node_decision: Mutex>, + before_attempt_decision: Mutex>, + edge_decision: Mutex>, + } + + impl RecordingLifecycle { + fn new(name: &str, log: Arc>>) -> Self { + Self { + name: name.to_string(), + log, + before_node_decision: Mutex::new(None), + before_attempt_decision: Mutex::new(None), + edge_decision: Mutex::new(None), + } + } + + fn with_before_node(self, decision: NodeDecision) -> Self { + *self.before_node_decision.lock().unwrap() = Some(decision); + self + } + + fn with_before_attempt(self, decision: NodeDecision) -> Self { + *self.before_attempt_decision.lock().unwrap() = Some(decision); + self + } + + fn with_edge_decision(self, decision: EdgeDecision) -> Self { + *self.edge_decision.lock().unwrap() = Some(decision); + self + } + } + + #[async_trait] + impl RunLifecycle for RecordingLifecycle { + async fn on_run_start(&self, _graph: &TestGraph, _state: &RunState) -> Result<()> { + self.log + .lock() + .unwrap() + .push(format!("{}:on_run_start", self.name)); + Ok(()) + } + + async fn on_terminal_reached(&self, _node: &TestNode, _state: &RunState) { + self.log + .lock() + .unwrap() + .push(format!("{}:on_terminal_reached", self.name)); + } + + async fn before_node(&self, _node: &TestNode, _state: &RunState) -> Result { + self.log + .lock() + .unwrap() + .push(format!("{}:before_node", self.name)); + Ok(self + .before_node_decision + .lock() + .unwrap() + .take() + .unwrap_or(NodeDecision::Continue)) + } + + async fn before_attempt( + &self, + _ctx: &AttemptContext<'_, TestGraph>, + _state: &RunState, + ) -> Result { + self.log + .lock() + .unwrap() + .push(format!("{}:before_attempt", self.name)); + Ok(self + .before_attempt_decision + .lock() + .unwrap() + .take() + .unwrap_or(NodeDecision::Continue)) + } + + async fn after_attempt( + &self, + _ctx: &AttemptResultContext<'_, TestGraph>, + _state: &RunState, + ) -> Result<()> { + self.log + .lock() + .unwrap() + .push(format!("{}:after_attempt", self.name)); + Ok(()) + } + + async fn after_node( + &self, + _node: &TestNode, + _result: &mut NodeResult, + _state: &RunState, + ) -> Result<()> { + self.log + .lock() + .unwrap() + .push(format!("{}:after_node", self.name)); + Ok(()) + } + + async fn on_edge_selected( + &self, + _ctx: &EdgeContext<'_, TestGraph>, + _state: &RunState, + ) -> Result { + self.log + .lock() + .unwrap() + .push(format!("{}:on_edge_selected", self.name)); + Ok(self + .edge_decision + .lock() + .unwrap() + .take() + .unwrap_or(EdgeDecision::Continue)) + } + + async fn on_checkpoint( + &self, + _node: &TestNode, + _result: &NodeResult, + _state: &RunState, + ) -> Result<()> { + self.log + .lock() + .unwrap() + .push(format!("{}:on_checkpoint", self.name)); + Ok(()) + } + + async fn on_run_end(&self, _outcome: &Outcome, _state: &RunState) { + self.log + .lock() + .unwrap() + .push(format!("{}:on_run_end", self.name)); + } + } + + #[tokio::test] + async fn default_lifecycle_is_noop() { + let lc = NoopLifecycle; + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + assert!( + >::on_run_start(&lc, &g, &state) + .await + .is_ok() + ); + let node = g.get_node("start").unwrap(); + assert!(matches!( + >::before_node(&lc, &node, &state) + .await + .unwrap(), + NodeDecision::Continue + )); + } + + #[tokio::test] + async fn composite_calls_all_children_on_run_start() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new(RecordingLifecycle::new("a", log.clone())), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + lc.on_run_start(&g, &state).await.unwrap(); + let calls = log.lock().unwrap().clone(); + assert_eq!(calls, vec!["a:on_run_start", "b:on_run_start"]); + } + + #[tokio::test] + async fn composite_before_node_skip_short_circuits() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new( + RecordingLifecycle::new("a", log.clone()) + .with_before_node(NodeDecision::Skip(Box::new(Outcome::skipped("hook")))), + ), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let node = g.get_node("start").unwrap(); + let decision = lc.before_node(&node, &state).await.unwrap(); + assert!(matches!(decision, NodeDecision::Skip(_))); + let calls = log.lock().unwrap().clone(); + assert_eq!(calls, vec!["a:before_node"]); + // b was NOT called + } + + #[tokio::test] + async fn composite_before_node_block_short_circuits() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new( + RecordingLifecycle::new("a", log.clone()) + .with_before_node(NodeDecision::Block("denied".into())), + ), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let node = g.get_node("start").unwrap(); + let decision = lc.before_node(&node, &state).await.unwrap(); + assert!(matches!(decision, NodeDecision::Block(_))); + let calls = log.lock().unwrap().clone(); + assert_eq!(calls, vec!["a:before_node"]); + } + + #[tokio::test] + async fn composite_before_attempt_skip_short_circuits() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new( + RecordingLifecycle::new("a", log.clone()) + .with_before_attempt(NodeDecision::Skip(Box::new(Outcome::skipped("skip")))), + ), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let node = g.get_node("start").unwrap(); + let ctx = AttemptContext { + node: &node, + attempt: 1, + max_attempts: 1, + }; + let decision = lc.before_attempt(&ctx, &state).await.unwrap(); + assert!(matches!(decision, NodeDecision::Skip(_))); + let calls = log.lock().unwrap().clone(); + assert_eq!(calls, vec!["a:before_attempt"]); + } + + #[tokio::test] + async fn composite_before_attempt_block_short_circuits() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new( + RecordingLifecycle::new("a", log.clone()) + .with_before_attempt(NodeDecision::Block("nope".into())), + ), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let node = g.get_node("start").unwrap(); + let ctx = AttemptContext { + node: &node, + attempt: 1, + max_attempts: 1, + }; + let decision = lc.before_attempt(&ctx, &state).await.unwrap(); + assert!(matches!(decision, NodeDecision::Block(_))); + } + + #[tokio::test] + async fn composite_after_attempt_calls_all() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new(RecordingLifecycle::new("a", log.clone())), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let node = g.get_node("start").unwrap(); + let result = NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1); + let ctx = AttemptResultContext { + node: &node, + result: &result, + attempt: 1, + will_retry: false, + backoff_delay: None, + }; + lc.after_attempt(&ctx, &state).await.unwrap(); + let calls = log.lock().unwrap().clone(); + assert_eq!(calls, vec!["a:after_attempt", "b:after_attempt"]); + } + + #[tokio::test] + async fn composite_on_edge_selected_override_short_circuits() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new( + RecordingLifecycle::new("a", log.clone()) + .with_edge_decision(EdgeDecision::Override("other".into())), + ), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let outcome = Outcome::success(); + let edge = g.outgoing_edges("start").into_iter().next().unwrap(); + let ctx = EdgeContext { + from: "start", + to: "end", + edge: Some(edge), + is_jump: false, + outcome: &outcome, + reason: "unconditional", + }; + let decision = lc.on_edge_selected(&ctx, &state).await.unwrap(); + assert!(matches!(decision, EdgeDecision::Override(ref t) if t == "other")); + let calls = log.lock().unwrap().clone(); + assert_eq!(calls, vec!["a:on_edge_selected"]); + } + + #[tokio::test] + async fn composite_on_edge_selected_block_short_circuits() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new( + RecordingLifecycle::new("a", log.clone()) + .with_edge_decision(EdgeDecision::Block("blocked".into())), + ), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let outcome = Outcome::success(); + let ctx = EdgeContext { + from: "start", + to: "end", + edge: None, + is_jump: false, + outcome: &outcome, + reason: "unconditional", + }; + let decision = lc.on_edge_selected(&ctx, &state).await.unwrap(); + assert!(matches!(decision, EdgeDecision::Block(_))); + } + + #[tokio::test] + async fn composite_on_edge_selected_none_for_jumps() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![Box::new(RecordingLifecycle::new("a", log.clone()))]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let outcome = Outcome::success(); + let ctx = EdgeContext:: { + from: "start", + to: "target", + edge: None, + is_jump: true, + outcome: &outcome, + reason: "jump", + }; + let decision = lc.on_edge_selected(&ctx, &state).await.unwrap(); + assert!(matches!(decision, EdgeDecision::Continue)); + assert!(ctx.edge.is_none()); + assert!(ctx.is_jump); + } + + #[tokio::test] + async fn composite_after_node_calls_all() { + let log = Arc::new(Mutex::new(Vec::new())); + let lc = CompositeLifecycle::new(vec![ + Box::new(RecordingLifecycle::new("a", log.clone())), + Box::new(RecordingLifecycle::new("b", log.clone())), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let node = g.get_node("start").unwrap(); + let mut result = NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1); + lc.after_node(&node, &mut result, &state).await.unwrap(); + let calls = log.lock().unwrap().clone(); + assert_eq!(calls, vec!["a:after_node", "b:after_node"]); + } + + #[tokio::test] + async fn composite_ordering_is_preserved() { + let log = Arc::new(Mutex::new(Vec::new())); + let counter = Arc::new(AtomicU32::new(0)); + + struct OrderedLifecycle { + name: String, + log: Arc>>, + counter: Arc, + } + + #[async_trait] + impl RunLifecycle for OrderedLifecycle { + async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> { + let order = self.counter.fetch_add(1, Ordering::SeqCst); + self.log + .lock() + .unwrap() + .push(format!("{}:{}", self.name, order)); + Ok(()) + } + } + + let lc = CompositeLifecycle::new(vec![ + Box::new(OrderedLifecycle { + name: "first".into(), + log: log.clone(), + counter: counter.clone(), + }), + Box::new(OrderedLifecycle { + name: "second".into(), + log: log.clone(), + counter: counter.clone(), + }), + Box::new(OrderedLifecycle { + name: "third".into(), + log: log.clone(), + counter: counter.clone(), + }), + ]); + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + lc.on_run_start(&g, &state).await.unwrap(); + let calls = log.lock().unwrap().clone(); + assert_eq!(calls, vec!["first:0", "second:1", "third:2"]); + } +} diff --git a/lib/crates/fabro-core/src/outcome.rs b/lib/crates/fabro-core/src/outcome.rs new file mode 100644 index 000000000..4df16780d --- /dev/null +++ b/lib/crates/fabro-core/src/outcome.rs @@ -0,0 +1,256 @@ +use std::collections::HashMap; +use std::fmt; +use std::str::FromStr; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StageStatus { + Success, + Fail, + Skipped, + PartialSuccess, + Retry, +} + +impl fmt::Display for StageStatus { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Success => write!(f, "success"), + Self::Fail => write!(f, "fail"), + Self::Skipped => write!(f, "skipped"), + Self::PartialSuccess => write!(f, "partial_success"), + Self::Retry => write!(f, "retry"), + } + } +} + +impl FromStr for StageStatus { + type Err = String; + + fn from_str(s: &str) -> std::result::Result { + match s { + "success" => Ok(Self::Success), + "fail" => Ok(Self::Fail), + "skipped" => Ok(Self::Skipped), + "partial_success" => Ok(Self::PartialSuccess), + "retry" => Ok(Self::Retry), + other => Err(format!("unknown stage status: {other}")), + } + } +} + +#[derive(Debug, Clone)] +pub struct FailureDetail { + pub message: String, + pub category: Option, + pub signature: Option, +} + +#[derive(Debug, Clone)] +pub struct Outcome { + pub status: StageStatus, + pub preferred_label: Option, + pub suggested_next_ids: Vec, + pub context_updates: HashMap, + pub jump_to_node: Option, + pub notes: Option, + pub failure: Option, + pub metadata: HashMap, +} + +impl Default for Outcome { + fn default() -> Self { + Self { + status: StageStatus::Success, + preferred_label: None, + suggested_next_ids: Vec::new(), + context_updates: HashMap::new(), + jump_to_node: None, + notes: None, + failure: None, + metadata: HashMap::new(), + } + } +} + +impl Outcome { + pub fn success() -> Self { + Self::default() + } + + pub fn fail(message: &str) -> Self { + Self { + status: StageStatus::Fail, + failure: Some(FailureDetail { + message: message.to_string(), + category: None, + signature: None, + }), + ..Self::default() + } + } + + pub fn skipped(reason: &str) -> Self { + Self { + status: StageStatus::Skipped, + notes: Some(reason.to_string()), + ..Self::default() + } + } +} + +#[derive(Debug, Clone)] +pub struct NodeResult { + pub outcome: Outcome, + pub duration: Duration, + pub attempts: u32, + pub max_attempts: u32, +} + +impl NodeResult { + pub fn new(outcome: Outcome, duration: Duration, attempts: u32, max_attempts: u32) -> Self { + Self { + outcome, + duration, + attempts, + max_attempts, + } + } + + pub fn from_error( + error: &crate::error::CoreError, + duration: Duration, + attempts: u32, + max_attempts: u32, + ) -> Self { + Self { + outcome: error.to_fail_outcome(), + duration, + attempts, + max_attempts, + } + } + + pub fn from_skip(outcome: Outcome) -> Self { + Self { + outcome, + duration: Duration::ZERO, + attempts: 0, + max_attempts: 0, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stage_status_display_roundtrip() { + let statuses = [ + StageStatus::Success, + StageStatus::Fail, + StageStatus::Skipped, + StageStatus::PartialSuccess, + StageStatus::Retry, + ]; + for status in &statuses { + let s = status.to_string(); + let parsed: StageStatus = s.parse().unwrap(); + assert_eq!(&parsed, status); + } + } + + #[test] + fn stage_status_serde_roundtrip() { + let statuses = [ + StageStatus::Success, + StageStatus::Fail, + StageStatus::Skipped, + StageStatus::PartialSuccess, + StageStatus::Retry, + ]; + for status in &statuses { + let json = serde_json::to_string(status).unwrap(); + let parsed: StageStatus = serde_json::from_str(&json).unwrap(); + assert_eq!(&parsed, status); + } + } + + #[test] + fn outcome_success_factory() { + let o = Outcome::success(); + assert_eq!(o.status, StageStatus::Success); + assert!(o.failure.is_none()); + assert!(o.notes.is_none()); + } + + #[test] + fn outcome_fail_factory() { + let o = Outcome::fail("broken"); + assert_eq!(o.status, StageStatus::Fail); + let f = o.failure.unwrap(); + assert_eq!(f.message, "broken"); + assert!(f.category.is_none()); + assert!(f.signature.is_none()); + } + + #[test] + fn outcome_skipped_factory() { + let o = Outcome::skipped("not needed"); + assert_eq!(o.status, StageStatus::Skipped); + assert_eq!(o.notes.as_deref(), Some("not needed")); + } + + #[test] + fn outcome_with_context_updates() { + let mut o = Outcome::success(); + o.context_updates + .insert("key".into(), serde_json::json!("value")); + assert_eq!(o.context_updates["key"], serde_json::json!("value")); + } + + #[test] + fn outcome_with_jump() { + let mut o = Outcome::success(); + o.jump_to_node = Some("target".into()); + assert_eq!(o.jump_to_node.as_deref(), Some("target")); + } + + #[test] + fn outcome_serde_roundtrip() { + // Test that metadata (the serde-friendly field) roundtrips + let mut o = Outcome::success(); + o.metadata + .insert("usage".into(), serde_json::json!({"tokens": 100})); + let json = serde_json::to_value(&o.metadata).unwrap(); + let parsed: HashMap = serde_json::from_value(json).unwrap(); + assert_eq!(parsed["usage"]["tokens"], 100); + } + + #[test] + fn failure_detail_construction() { + let f = FailureDetail { + message: "timeout".into(), + category: Some("transient".into()), + signature: Some("sig".into()), + }; + assert_eq!(f.message, "timeout"); + assert_eq!(f.category.as_deref(), Some("transient")); + assert_eq!(f.signature.as_deref(), Some("sig")); + } + + #[test] + fn node_result_from_outcome() { + let o = Outcome::success(); + let r = NodeResult::new(o, Duration::from_millis(100), 1, 3); + assert_eq!(r.outcome.status, StageStatus::Success); + assert_eq!(r.duration, Duration::from_millis(100)); + assert_eq!(r.attempts, 1); + assert_eq!(r.max_attempts, 3); + } +} diff --git a/lib/crates/fabro-core/src/retry.rs b/lib/crates/fabro-core/src/retry.rs new file mode 100644 index 000000000..af04c3192 --- /dev/null +++ b/lib/crates/fabro-core/src/retry.rs @@ -0,0 +1,109 @@ +use std::time::Duration; + +#[derive(Debug, Clone)] +pub struct BackoffPolicy { + pub initial_delay: Duration, + pub factor: f64, + pub max_delay: Duration, + pub jitter: bool, +} + +impl Default for BackoffPolicy { + fn default() -> Self { + Self { + initial_delay: Duration::from_secs(1), + factor: 2.0, + max_delay: Duration::from_secs(60), + jitter: false, + } + } +} + +impl BackoffPolicy { + pub fn delay_for_attempt(&self, attempt: u32) -> Duration { + let multiplier = self.factor.powi(attempt.saturating_sub(1) as i32); + let delay = self.initial_delay.mul_f64(multiplier); + if delay > self.max_delay { + self.max_delay + } else { + delay + } + } +} + +#[derive(Debug, Clone)] +pub struct RetryPolicy { + pub max_attempts: u32, + pub backoff: BackoffPolicy, +} + +impl RetryPolicy { + pub fn none() -> Self { + Self { + max_attempts: 1, + backoff: BackoffPolicy::default(), + } + } + + pub fn with_max_attempts(max_attempts: u32) -> Self { + Self { + max_attempts, + backoff: BackoffPolicy::default(), + } + } +} + +impl Default for RetryPolicy { + fn default() -> Self { + Self::none() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn backoff_delay_first_attempt() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(100), + factor: 2.0, + max_delay: Duration::from_secs(10), + jitter: false, + }; + assert_eq!(b.delay_for_attempt(1), Duration::from_millis(100)); + } + + #[test] + fn backoff_delay_exponential() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(100), + factor: 2.0, + max_delay: Duration::from_secs(10), + jitter: false, + }; + assert_eq!(b.delay_for_attempt(2), Duration::from_millis(200)); + assert_eq!(b.delay_for_attempt(3), Duration::from_millis(400)); + assert_eq!(b.delay_for_attempt(4), Duration::from_millis(800)); + } + + #[test] + fn backoff_delay_capped_at_max() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(100), + factor: 2.0, + max_delay: Duration::from_millis(300), + jitter: false, + }; + assert_eq!(b.delay_for_attempt(1), Duration::from_millis(100)); + assert_eq!(b.delay_for_attempt(2), Duration::from_millis(200)); + assert_eq!(b.delay_for_attempt(3), Duration::from_millis(300)); // capped + assert_eq!(b.delay_for_attempt(4), Duration::from_millis(300)); // still capped + } + + #[test] + fn retry_policy_none_is_single_attempt() { + let p = RetryPolicy::none(); + assert_eq!(p.max_attempts, 1); + } +} diff --git a/lib/crates/fabro-core/src/stall.rs b/lib/crates/fabro-core/src/stall.rs new file mode 100644 index 000000000..47a01006a --- /dev/null +++ b/lib/crates/fabro-core/src/stall.rs @@ -0,0 +1,206 @@ +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::Notify; + +/// Trait for receiving stall timeout notifications. +pub trait ActivityMonitor: Send + Sync { + /// Called when a stall timeout fires. The implementation should signal + /// cancellation (e.g., set a cancel token). + fn on_stall_timeout(&self, elapsed: Duration); +} + +/// Watches for inactivity and fires a stall timeout if no activity is +/// reported within the configured duration. +pub struct StallWatchdog { + timeout: Duration, + cancel_token: Arc, + activity: Arc, + shutdown: Arc, + monitor: Arc, +} + +/// Guard that resets the stall timer on activity. Drop to stop watching. +pub struct StallGuard { + activity: Arc, + shutdown: Arc, + handle: Option>, +} + +impl StallWatchdog { + pub fn new( + timeout: Duration, + cancel_token: Arc, + monitor: Arc, + ) -> Self { + Self { + timeout, + cancel_token, + activity: Arc::new(Notify::new()), + shutdown: Arc::new(AtomicBool::new(false)), + monitor, + } + } + + /// Start watching. Returns a StallGuard — call `guard.report_activity()` + /// to reset the timer. Drop the guard to stop the watchdog. + pub fn start(self) -> StallGuard { + let activity = self.activity.clone(); + let shutdown = self.shutdown.clone(); + let timeout = self.timeout; + let cancel_token = self.cancel_token; + let monitor = self.monitor; + + let handle = tokio::spawn(async move { + loop { + tokio::select! { + _ = tokio::time::sleep(timeout) => { + if shutdown.load(Ordering::Relaxed) { + return; + } + tracing::info!( + timeout_secs = timeout.as_secs(), + "Stall timeout: no activity detected" + ); + monitor.on_stall_timeout(timeout); + cancel_token.store(true, Ordering::Relaxed); + return; + } + _ = activity.notified() => { + if shutdown.load(Ordering::Relaxed) { + return; + } + // Activity reported, restart the timer + continue; + } + } + } + }); + + StallGuard { + activity: self.activity, + shutdown: self.shutdown, + handle: Some(handle), + } + } +} + +impl StallGuard { + /// Report activity to reset the stall timer. + pub fn report_activity(&self) { + self.activity.notify_one(); + } +} + +impl Drop for StallGuard { + fn drop(&mut self) { + self.shutdown.store(true, Ordering::Relaxed); + self.activity.notify_one(); // wake the task so it can exit + if let Some(handle) = self.handle.take() { + handle.abort(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicU32; + + struct TestMonitor { + stall_count: AtomicU32, + } + + impl TestMonitor { + fn new() -> Arc { + Arc::new(Self { + stall_count: AtomicU32::new(0), + }) + } + + fn stalls(&self) -> u32 { + self.stall_count.load(Ordering::Relaxed) + } + } + + impl ActivityMonitor for TestMonitor { + fn on_stall_timeout(&self, _elapsed: Duration) { + self.stall_count.fetch_add(1, Ordering::Relaxed); + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stall_watchdog_cancels_on_inactivity() { + let cancel = Arc::new(AtomicBool::new(false)); + let monitor = TestMonitor::new(); + let watchdog = + StallWatchdog::new(Duration::from_millis(50), cancel.clone(), monitor.clone()); + let _guard = watchdog.start(); + + // Wait for timeout to fire + tokio::time::sleep(Duration::from_millis(100)).await; + + assert!(cancel.load(Ordering::Relaxed)); + assert_eq!(monitor.stalls(), 1); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stall_watchdog_resets_on_activity() { + let cancel = Arc::new(AtomicBool::new(false)); + let monitor = TestMonitor::new(); + let watchdog = + StallWatchdog::new(Duration::from_millis(80), cancel.clone(), monitor.clone()); + let guard = watchdog.start(); + + // Report activity before timeout + tokio::time::sleep(Duration::from_millis(50)).await; + guard.report_activity(); + + // After another 50ms (100ms total, but only 50ms since activity), should not have timed out + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(!cancel.load(Ordering::Relaxed)); + + // Wait long enough for timeout after last activity (80ms + margin) + tokio::time::sleep(Duration::from_millis(60)).await; + assert!(cancel.load(Ordering::Relaxed)); + assert_eq!(monitor.stalls(), 1); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stall_watchdog_clean_shutdown_on_success() { + let cancel = Arc::new(AtomicBool::new(false)); + let monitor = TestMonitor::new(); + let watchdog = + StallWatchdog::new(Duration::from_millis(50), cancel.clone(), monitor.clone()); + let guard = watchdog.start(); + + // Drop the guard before timeout + drop(guard); + + // Wait past timeout + tokio::time::sleep(Duration::from_millis(100)).await; + + // Should NOT have triggered + assert!(!cancel.load(Ordering::Relaxed)); + assert_eq!(monitor.stalls(), 0); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stall_guard_cleanup_on_drop() { + let cancel = Arc::new(AtomicBool::new(false)); + let monitor = TestMonitor::new(); + let watchdog = + StallWatchdog::new(Duration::from_millis(50), cancel.clone(), monitor.clone()); + let guard = watchdog.start(); + + // Drop guard — should abort the background task + drop(guard); + + // Wait well past timeout + tokio::time::sleep(Duration::from_millis(150)).await; + + // Cancel should not be set + assert!(!cancel.load(Ordering::Relaxed)); + } +} diff --git a/lib/crates/fabro-core/src/state.rs b/lib/crates/fabro-core/src/state.rs new file mode 100644 index 000000000..ff8f3ea46 --- /dev/null +++ b/lib/crates/fabro-core/src/state.rs @@ -0,0 +1,174 @@ +use std::collections::HashMap; + +use crate::context::Context; +use crate::error::Result; +use crate::graph::{Graph, NodeSpec}; +use crate::outcome::{NodeResult, Outcome}; + +pub struct RunState { + pub context: Context, + pub current_node_id: String, + pub completed_nodes: Vec, + pub node_outcomes: HashMap, + pub node_retries: HashMap, + pub node_visits: HashMap, + pub stage_index: usize, + pub previous_node_id: Option, +} + +impl RunState { + pub fn new(graph: &G) -> Result { + let start = graph.find_start_node()?; + Ok(Self { + context: Context::new(), + current_node_id: start.id().to_string(), + completed_nodes: Vec::new(), + node_outcomes: HashMap::new(), + node_retries: HashMap::new(), + node_visits: HashMap::new(), + stage_index: 0, + previous_node_id: None, + }) + } + + pub fn record(&mut self, node_id: &str, result: &NodeResult) { + self.completed_nodes.push(node_id.to_string()); + self.node_outcomes + .insert(node_id.to_string(), result.outcome.clone()); + if result.attempts > 1 { + self.node_retries + .insert(node_id.to_string(), result.attempts - 1); + } + self.stage_index += 1; + self.context.apply_updates(&result.outcome.context_updates); + } + + pub fn advance(&mut self, next_node_id: &str) { + self.previous_node_id = Some(self.current_node_id.clone()); + self.current_node_id = next_node_id.to_string(); + } + + pub fn restart(&mut self, start_node_id: &str) { + self.current_node_id = start_node_id.to_string(); + self.completed_nodes.clear(); + self.node_outcomes.clear(); + self.node_retries.clear(); + self.stage_index = 0; + self.previous_node_id = None; + // node_visits is NOT cleared — preserves total visit counts across restarts + } + + pub fn current_node(&self, graph: &G) -> Option { + graph.get_node(&self.current_node_id) + } + + pub fn increment_visits(&mut self, node_id: &str) -> usize { + let count = self.node_visits.entry(node_id.to_string()).or_insert(0); + *count += 1; + *count + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use serde_json::json; + + use super::*; + use crate::outcome::{Outcome, StageStatus}; + use crate::test_fixtures::linear_graph; + + #[test] + fn run_state_new_from_graph() { + let g = linear_graph(&["start", "work", "end"]); + let state = RunState::new(&g).unwrap(); + assert_eq!(state.current_node_id, "start"); + assert!(state.completed_nodes.is_empty()); + assert!(state.node_outcomes.is_empty()); + assert_eq!(state.stage_index, 0); + assert!(state.previous_node_id.is_none()); + } + + #[test] + fn run_state_record_updates_all_fields() { + let g = linear_graph(&["start", "end"]); + let mut state = RunState::new(&g).unwrap(); + let result = NodeResult::new(Outcome::success(), Duration::from_millis(50), 2, 3); + state.record("start", &result); + + assert_eq!(state.completed_nodes, vec!["start"]); + assert_eq!(state.node_outcomes["start"].status, StageStatus::Success); + assert_eq!(state.node_retries["start"], 1); // 2 attempts - 1 + assert_eq!(state.stage_index, 1); + } + + #[test] + fn run_state_record_applies_context_updates() { + let g = linear_graph(&["start", "end"]); + let mut state = RunState::new(&g).unwrap(); + let mut outcome = Outcome::success(); + outcome.context_updates.insert("key".into(), json!("value")); + let result = NodeResult::new(outcome, Duration::ZERO, 1, 1); + state.record("start", &result); + assert_eq!(state.context.get("key"), Some(json!("value"))); + } + + #[test] + fn run_state_advance_updates_current_and_previous() { + let g = linear_graph(&["start", "mid", "end"]); + let mut state = RunState::new(&g).unwrap(); + assert_eq!(state.current_node_id, "start"); + assert!(state.previous_node_id.is_none()); + + state.advance("mid"); + assert_eq!(state.current_node_id, "mid"); + assert_eq!(state.previous_node_id.as_deref(), Some("start")); + + state.advance("end"); + assert_eq!(state.current_node_id, "end"); + assert_eq!(state.previous_node_id.as_deref(), Some("mid")); + } + + #[test] + fn run_state_restart_clears_progress_keeps_visits() { + let g = linear_graph(&["start", "work", "end"]); + let mut state = RunState::new(&g).unwrap(); + state.increment_visits("start"); + state.increment_visits("work"); + state.record( + "start", + &NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1), + ); + state.advance("work"); + + state.restart("start"); + + assert_eq!(state.current_node_id, "start"); + assert!(state.completed_nodes.is_empty()); + assert!(state.node_outcomes.is_empty()); + assert!(state.node_retries.is_empty()); + assert_eq!(state.stage_index, 0); + assert!(state.previous_node_id.is_none()); + // visits preserved + assert_eq!(state.node_visits["start"], 1); + assert_eq!(state.node_visits["work"], 1); + } + + #[test] + fn run_state_current_node_from_graph() { + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let node = state.current_node(&g).unwrap(); + assert_eq!(node.id(), "start"); + } + + #[test] + fn run_state_increment_visits() { + let g = linear_graph(&["start", "end"]); + let mut state = RunState::new(&g).unwrap(); + assert_eq!(state.increment_visits("start"), 1); + assert_eq!(state.increment_visits("start"), 2); + assert_eq!(state.increment_visits("other"), 1); + } +} diff --git a/lib/crates/fabro-core/src/test_fixtures.rs b/lib/crates/fabro-core/src/test_fixtures.rs new file mode 100644 index 000000000..811502d63 --- /dev/null +++ b/lib/crates/fabro-core/src/test_fixtures.rs @@ -0,0 +1,600 @@ +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; + +use async_trait::async_trait; + +use crate::context::Context; +use crate::error::{CoreError, HandlerErrorDetail, Result}; +use crate::graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; +use crate::handler::NodeHandler; +use crate::outcome::{Outcome, StageStatus}; +use crate::retry::RetryPolicy; + +// ---- Test node ---- + +#[derive(Debug, Clone)] +pub struct TestNode { + pub id: String, + pub terminal: bool, + pub max_visits: Option, + pub goal_gate: Option<(String, StageStatus)>, +} + +impl TestNode { + pub fn new(id: &str) -> Self { + Self { + id: id.to_string(), + terminal: false, + max_visits: None, + goal_gate: None, + } + } + + pub fn terminal(id: &str) -> Self { + Self { + id: id.to_string(), + terminal: true, + max_visits: None, + goal_gate: None, + } + } + + pub fn with_max_visits(mut self, max: usize) -> Self { + self.max_visits = Some(max); + self + } + + pub fn with_goal_gate(mut self, node_id: &str, required_status: StageStatus) -> Self { + self.goal_gate = Some((node_id.to_string(), required_status)); + self + } +} + +impl NodeSpec for TestNode { + fn id(&self) -> &str { + &self.id + } + + fn is_terminal(&self) -> bool { + self.terminal + } + + fn max_visits(&self) -> Option { + self.max_visits + } +} + +// ---- Test edge ---- + +#[derive(Debug, Clone)] +pub struct TestEdge { + pub from: String, + pub to: String, + pub label: Option, + pub loop_restart: bool, +} + +impl TestEdge { + pub fn new(from: &str, to: &str) -> Self { + Self { + from: from.to_string(), + to: to.to_string(), + label: None, + loop_restart: false, + } + } + + pub fn with_label(mut self, label: &str) -> Self { + self.label = Some(label.to_string()); + self + } + + pub fn with_loop_restart(mut self) -> Self { + self.loop_restart = true; + self + } +} + +impl EdgeSpec for TestEdge { + fn target(&self) -> &str { + &self.to + } + + fn label(&self) -> Option<&str> { + self.label.as_deref() + } + + fn is_loop_restart(&self) -> bool { + self.loop_restart + } +} + +// ---- Test graph ---- + +#[derive(Debug, Clone)] +pub struct TestGraph { + pub nodes: Vec, + pub edges: Vec, + pub start_node_id: String, + pub retry_targets: HashMap, +} + +impl TestGraph { + pub fn new(nodes: Vec, edges: Vec, start: &str) -> Self { + Self { + nodes, + edges, + start_node_id: start.to_string(), + retry_targets: HashMap::new(), + } + } + + pub fn with_retry_target(mut self, from: &str, to: &str) -> Self { + self.retry_targets.insert(from.to_string(), to.to_string()); + self + } +} + +impl Graph for TestGraph { + type Node = TestNode; + type Edge = TestEdge; + + fn get_node(&self, id: &str) -> Option { + self.nodes.iter().find(|n| n.id == id).cloned() + } + + fn find_start_node(&self) -> Result { + self.get_node(&self.start_node_id) + .ok_or(CoreError::NoStartNode) + } + + fn outgoing_edges(&self, node_id: &str) -> Vec { + self.edges + .iter() + .filter(|e| e.from == node_id) + .cloned() + .collect() + } + + fn select_edge( + &self, + node: &Self::Node, + outcome: &Outcome, + _context: &Context, + ) -> Option> { + let edges = self.outgoing_edges(node.id()); + if edges.is_empty() { + return None; + } + + // First: match by preferred_label + if let Some(ref label) = outcome.preferred_label { + if let Some(e) = edges + .iter() + .find(|e| e.label.as_deref() == Some(label.as_str())) + { + return Some(EdgeSelection { + edge: e.clone(), + reason: "preferred_label", + }); + } + } + + // Second: match by status label (e.g. "fail", "success") + let status_label = outcome.status.to_string(); + if let Some(e) = edges + .iter() + .find(|e| e.label.as_deref() == Some(status_label.as_str())) + { + return Some(EdgeSelection { + edge: e.clone(), + reason: "condition", + }); + } + + // Third: match by suggested_next_ids + for suggested in &outcome.suggested_next_ids { + if let Some(e) = edges.iter().find(|e| e.to == *suggested) { + return Some(EdgeSelection { + edge: e.clone(), + reason: "suggested_next", + }); + } + } + + // Fourth: unconditional (no label) + if let Some(e) = edges.iter().find(|e| e.label.is_none()) { + return Some(EdgeSelection { + edge: e.clone(), + reason: "unconditional", + }); + } + + None + } + + fn check_goal_gates( + &self, + outcomes: &HashMap, + ) -> std::result::Result<(), String> { + for node in &self.nodes { + if let Some((ref required_node, ref required_status)) = node.goal_gate { + if node.is_terminal() { + match outcomes.get(required_node) { + Some(o) if o.status == *required_status => {} + Some(o) => { + return Err(format!( + "goal gate failed: {} requires {} to be {:?} but was {:?}", + node.id, required_node, required_status, o.status + )); + } + None => { + return Err(format!( + "goal gate failed: {} requires {} but it was not completed", + node.id, required_node + )); + } + } + } + } + } + Ok(()) + } + + fn get_retry_target(&self, failed_node_id: &str) -> Option { + self.retry_targets.get(failed_node_id).cloned() + } +} + +// ---- Test handlers ---- + +pub struct AlwaysSucceedHandler; + +#[async_trait] +impl NodeHandler for AlwaysSucceedHandler { + async fn execute( + &self, + _node: &TestNode, + _context: &Context, + _graph: &TestGraph, + ) -> Result { + Ok(Outcome::success()) + } +} + +pub struct AlwaysFailHandler { + pub message: String, +} + +impl AlwaysFailHandler { + pub fn new(message: &str) -> Self { + Self { + message: message.to_string(), + } + } +} + +#[async_trait] +impl NodeHandler for AlwaysFailHandler { + async fn execute( + &self, + _node: &TestNode, + _context: &Context, + _graph: &TestGraph, + ) -> Result { + Ok(Outcome::fail(&self.message)) + } +} + +pub struct CountingHandler { + pub call_count: AtomicU32, + pub outcomes: std::sync::Mutex>>, + pub retry_policy: RetryPolicy, +} + +impl CountingHandler { + pub fn new(outcomes: Vec>) -> Self { + Self { + call_count: AtomicU32::new(0), + outcomes: std::sync::Mutex::new(outcomes), + retry_policy: RetryPolicy::none(), + } + } + + pub fn with_retry_policy(mut self, policy: RetryPolicy) -> Self { + self.retry_policy = policy; + self + } + + pub fn calls(&self) -> u32 { + self.call_count.load(Ordering::Relaxed) + } +} + +#[async_trait] +impl NodeHandler for CountingHandler { + async fn execute( + &self, + _node: &TestNode, + _context: &Context, + _graph: &TestGraph, + ) -> Result { + let count = self.call_count.fetch_add(1, Ordering::Relaxed); + let mut outcomes = self.outcomes.lock().unwrap(); + if (count as usize) < outcomes.len() { + outcomes.remove(0) + } else { + Ok(Outcome::success()) + } + } + + fn retry_policy(&self, _node: &TestNode, _graph: &TestGraph) -> RetryPolicy { + self.retry_policy.clone() + } +} + +/// A handler that dispatches based on node ID. +pub struct DispatchHandler { + handlers: HashMap>>, + default: Arc>, +} + +impl DispatchHandler { + pub fn new(default: Arc>) -> Self { + Self { + handlers: HashMap::new(), + default, + } + } + + pub fn with_handler(mut self, node_id: &str, handler: Arc>) -> Self { + self.handlers.insert(node_id.to_string(), handler); + self + } +} + +#[async_trait] +impl NodeHandler for DispatchHandler { + async fn execute( + &self, + node: &TestNode, + context: &Context, + graph: &TestGraph, + ) -> Result { + let handler = self.handlers.get(node.id()).unwrap_or(&self.default); + handler.execute(node, context, graph).await + } + + fn retry_policy(&self, node: &TestNode, graph: &TestGraph) -> RetryPolicy { + let handler = self.handlers.get(node.id()).unwrap_or(&self.default); + handler.retry_policy(node, graph) + } + + fn on_retries_exhausted(&self, node: &TestNode, last_outcome: Outcome) -> Outcome { + let handler = self.handlers.get(node.id()).unwrap_or(&self.default); + handler.on_retries_exhausted(node, last_outcome) + } +} + +/// A handler that returns Err(CoreError::Handler) with configurable retryability. +pub struct ErrorHandler { + pub detail: HandlerErrorDetail, + pub retry_policy: RetryPolicy, +} + +impl ErrorHandler { + pub fn retryable(message: &str, policy: RetryPolicy) -> Self { + Self { + detail: HandlerErrorDetail { + message: message.to_string(), + retryable: true, + category: None, + signature: None, + }, + retry_policy: policy, + } + } + + pub fn non_retryable(message: &str) -> Self { + Self { + detail: HandlerErrorDetail { + message: message.to_string(), + retryable: false, + category: None, + signature: None, + }, + retry_policy: RetryPolicy::none(), + } + } +} + +#[async_trait] +impl NodeHandler for ErrorHandler { + async fn execute( + &self, + _node: &TestNode, + _context: &Context, + _graph: &TestGraph, + ) -> Result { + Err(CoreError::handler(self.detail.clone())) + } + + fn retry_policy(&self, _node: &TestNode, _graph: &TestGraph) -> RetryPolicy { + self.retry_policy.clone() + } +} + +// ---- Helper for building common graphs ---- + +/// Build a linear graph: start → a → b → ... → end +pub fn linear_graph(node_ids: &[&str]) -> TestGraph { + assert!(node_ids.len() >= 2, "need at least start and end nodes"); + let mut nodes = Vec::new(); + let mut edges = Vec::new(); + + for (i, id) in node_ids.iter().enumerate() { + if i == node_ids.len() - 1 { + nodes.push(TestNode::terminal(id)); + } else { + nodes.push(TestNode::new(id)); + edges.push(TestEdge::new(id, node_ids[i + 1])); + } + } + + TestGraph::new(nodes, edges, node_ids[0]) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_graph_finds_start_node() { + let g = linear_graph(&["start", "end"]); + let start = g.find_start_node().unwrap(); + assert_eq!(start.id(), "start"); + } + + #[test] + fn test_graph_gets_node_by_id() { + let g = linear_graph(&["start", "work", "end"]); + let node = g.get_node("work").unwrap(); + assert_eq!(node.id(), "work"); + assert!(!node.is_terminal()); + } + + #[test] + fn test_graph_returns_none_for_missing() { + let g = linear_graph(&["start", "end"]); + assert!(g.get_node("nonexistent").is_none()); + } + + #[test] + fn test_graph_outgoing_edges() { + let g = linear_graph(&["start", "mid", "end"]); + let edges = g.outgoing_edges("start"); + assert_eq!(edges.len(), 1); + assert_eq!(edges[0].target(), "mid"); + } + + #[test] + fn test_graph_terminal_detection() { + let g = linear_graph(&["start", "end"]); + assert!(!g.get_node("start").unwrap().is_terminal()); + assert!(g.get_node("end").unwrap().is_terminal()); + } + + #[test] + fn test_graph_edge_selection_by_label() { + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::new("a"), + TestNode::new("b"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("start", "a").with_label("success"), + TestEdge::new("start", "b").with_label("fail"), + ], + "start", + ); + let node = g.get_node("start").unwrap(); + let outcome = Outcome::fail("oops"); + let ctx = Context::new(); + let sel = g.select_edge(&node, &outcome, &ctx).unwrap(); + assert_eq!(sel.edge.target(), "b"); + assert_eq!(sel.reason, "condition"); + } + + #[test] + fn test_graph_edge_selection_unconditional() { + let g = linear_graph(&["start", "end"]); + let node = g.get_node("start").unwrap(); + let outcome = Outcome::success(); + let ctx = Context::new(); + let sel = g.select_edge(&node, &outcome, &ctx).unwrap(); + assert_eq!(sel.edge.target(), "end"); + assert_eq!(sel.reason, "unconditional"); + } + + #[test] + fn test_graph_goal_gates_pass() { + let g = TestGraph::new( + vec![ + TestNode::new("work"), + TestNode::terminal("end").with_goal_gate("work", StageStatus::Success), + ], + vec![TestEdge::new("work", "end")], + "work", + ); + let mut outcomes = HashMap::new(); + outcomes.insert("work".to_string(), Outcome::success()); + assert!(g.check_goal_gates(&outcomes).is_ok()); + } + + #[test] + fn test_graph_goal_gates_fail() { + let g = TestGraph::new( + vec![ + TestNode::new("work"), + TestNode::terminal("end").with_goal_gate("work", StageStatus::Success), + ], + vec![TestEdge::new("work", "end")], + "work", + ); + let mut outcomes = HashMap::new(); + outcomes.insert("work".to_string(), Outcome::fail("oops")); + assert!(g.check_goal_gates(&outcomes).is_err()); + } + + #[test] + fn test_graph_retry_target() { + let g = linear_graph(&["start", "end"]).with_retry_target("start", "start"); + assert_eq!(g.get_retry_target("start").as_deref(), Some("start")); + assert!(g.get_retry_target("end").is_none()); + } + + #[tokio::test] + async fn always_succeed_handler() { + let h = AlwaysSucceedHandler; + let g = linear_graph(&["start", "end"]); + let node = g.get_node("start").unwrap(); + let ctx = Context::new(); + let result = h.execute(&node, &ctx, &g).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + } + + #[tokio::test] + async fn always_fail_handler() { + let h = AlwaysFailHandler::new("boom"); + let g = linear_graph(&["start", "end"]); + let node = g.get_node("start").unwrap(); + let ctx = Context::new(); + let result = h.execute(&node, &ctx, &g).await.unwrap(); + assert_eq!(result.status, StageStatus::Fail); + assert_eq!(result.failure.unwrap().message, "boom"); + } + + #[tokio::test] + async fn counting_handler_tracks_calls() { + let h = CountingHandler::new(vec![Ok(Outcome::fail("first")), Ok(Outcome::success())]); + let g = linear_graph(&["start", "end"]); + let node = g.get_node("start").unwrap(); + let ctx = Context::new(); + + let r1 = h.execute(&node, &ctx, &g).await.unwrap(); + assert_eq!(r1.status, StageStatus::Fail); + assert_eq!(h.calls(), 1); + + let r2 = h.execute(&node, &ctx, &g).await.unwrap(); + assert_eq!(r2.status, StageStatus::Success); + assert_eq!(h.calls(), 2); + + // Past end of outcomes list → default success + let r3 = h.execute(&node, &ctx, &g).await.unwrap(); + assert_eq!(r3.status, StageStatus::Success); + assert_eq!(h.calls(), 3); + } +}