diff --git a/Cargo.lock b/Cargo.lock index a671e79c8..053af41ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1408,10 +1408,12 @@ name = "fabro-core" version = "0.176.2" dependencies = [ "async-trait", + "rand 0.8.5", "serde", "serde_json", "thiserror 2.0.18", "tokio", + "tokio-util", "tracing", ] @@ -1745,6 +1747,7 @@ dependencies = [ "dotenvy", "fabro-agent", "fabro-config", + "fabro-core", "fabro-devcontainer", "fabro-git-storage", "fabro-github", diff --git a/lib/crates/fabro-core/Cargo.toml b/lib/crates/fabro-core/Cargo.toml index 90f389d3e..ce4bfd9a2 100644 --- a/lib/crates/fabro-core/Cargo.toml +++ b/lib/crates/fabro-core/Cargo.toml @@ -13,7 +13,9 @@ async-trait.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true +rand.workspace = true tokio.workspace = true +tokio-util.workspace = true tracing.workspace = true [dev-dependencies] diff --git a/lib/crates/fabro-core/src/context.rs b/lib/crates/fabro-core/src/context.rs index 3e1ade0f0..117594831 100644 --- a/lib/crates/fabro-core/src/context.rs +++ b/lib/crates/fabro-core/src/context.rs @@ -76,6 +76,13 @@ impl Context { } } + pub fn with_store_and_logs( + store: Arc, + logs: Arc>>, + ) -> Self { + Self { store, logs } + } + pub fn set(&self, key: impl Into, value: Value) { self.store.set(key.into(), value); } diff --git a/lib/crates/fabro-core/src/error.rs b/lib/crates/fabro-core/src/error.rs index 8d6b76f52..61bf7b25c 100644 --- a/lib/crates/fabro-core/src/error.rs +++ b/lib/crates/fabro-core/src/error.rs @@ -1,5 +1,4 @@ use std::fmt; -use std::time::Duration; use crate::outcome::{FailureDetail, Outcome, StageStatus}; @@ -35,8 +34,8 @@ pub enum CoreError { visits: usize, limit: usize, }, - #[error("stall timeout: no activity for {elapsed:?}")] - StallTimeout { elapsed: Duration }, + #[error("stall timeout on node \"{node_id}\"")] + StallTimeout { node_id: String }, #[error("{detail}")] Handler { detail: HandlerErrorDetail }, #[error("{0}")] @@ -109,10 +108,10 @@ mod tests { ); assert_eq!( CoreError::StallTimeout { - elapsed: Duration::from_secs(30) + node_id: "work".into() } .to_string(), - "stall timeout: no activity for 30s" + "stall timeout on node \"work\"" ); assert_eq!( CoreError::Other("something broke".into()).to_string(), diff --git a/lib/crates/fabro-core/src/executor.rs b/lib/crates/fabro-core/src/executor.rs index a207bd1e0..7e9a6e6c5 100644 --- a/lib/crates/fabro-core/src/executor.rs +++ b/lib/crates/fabro-core/src/executor.rs @@ -2,6 +2,9 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Instant; +use tokio_util::sync::CancellationToken; + +use crate::context::Context; use crate::error::{CoreError, Result}; use crate::graph::{EdgeSpec, Graph, NodeSpec}; use crate::handler::NodeHandler; @@ -15,6 +18,7 @@ use crate::state::RunState; #[derive(Default)] pub struct ExecutorSettings { pub cancel_token: Option>, + pub stall_token: Option, pub max_node_visits: Option, } @@ -56,6 +60,11 @@ impl ExecutorBuilder { self } + pub fn stall_token(mut self, token: CancellationToken) -> Self { + self.settings.stall_token = Some(token); + self + } + pub fn max_node_visits(mut self, limit: usize) -> Self { self.settings.max_node_visits = Some(limit); self @@ -78,6 +87,7 @@ impl Executor { // Check cancellation if let Some(ref token) = self.settings.cancel_token { if token.load(Ordering::Relaxed) { + state.cancelled = true; let outcome = Outcome::fail("run cancelled"); self.lifecycle.on_run_end(&outcome, &state).await; return Err(CoreError::Cancelled); @@ -90,38 +100,46 @@ impl Executor { id: state.current_node_id.clone(), })?; - // Terminal nodes: skip normal lifecycle, call on_terminal_reached, check goal gates + // Terminal nodes: skip normal lifecycle, check goal gates, call on_terminal_reached if node.is_terminal() { - self.lifecycle.on_terminal_reached(&node, &state).await; match graph.check_goal_gates(&state.node_outcomes) { Ok(()) => { + self.lifecycle + .on_terminal_reached(&node, true, &state) + .await; let outcome = Outcome::success(); self.lifecycle.on_run_end(&outcome, &state).await; return Ok(outcome); } - Err(msg) => { + Err(failed_node_id) => { + self.lifecycle + .on_terminal_reached(&node, false, &state) + .await; // Check if there's a retry target for goal gate failure - if let Some(retry_target) = graph.get_retry_target(&state.current_node_id) { + if let Some(retry_target) = graph.get_retry_target(&failed_node_id) { tracing::debug!( node = %node.id(), retry_target = %retry_target, - reason = %msg, + failed_node = %failed_node_id, "Goal gate unsatisfied, retrying" ); state.advance(&retry_target); continue; } - let outcome = Outcome::fail(&msg); + let outcome = Outcome::fail(&format!( + "goal gate failed for node \"{}\"", + failed_node_id + )); self.lifecycle.on_run_end(&outcome, &state).await; return Ok(outcome); } } } - // Check visit limits + // Check visit limits (>= matches fabro-workflows semantics) let visits = state.increment_visits(node.id()); if let Some(max) = node.max_visits() { - if visits > max { + if visits >= max { return Err(CoreError::VisitLimitExceeded { node_id: node.id().to_string(), visits, @@ -130,7 +148,7 @@ impl Executor { } } if let Some(global_max) = self.settings.max_node_visits { - if visits > global_max { + if visits >= global_max { return Err(CoreError::VisitLimitExceeded { node_id: node.id().to_string(), visits, @@ -140,28 +158,39 @@ impl Executor { } // before_node lifecycle - match self.lifecycle.before_node(&node, &state).await? { + let node_result = 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?; + result } NodeDecision::Block(msg) => { return Err(CoreError::blocked(msg)); } NodeDecision::Continue => { - // Execute with retry - let mut result = self.execute_with_retry(&node, &state, graph).await?; + // Execute with retry, racing against stall token + let mut result = if let Some(ref stall) = self.settings.stall_token { + tokio::select! { + r = self.execute_with_retry(&node, &state, graph) => r?, + () = stall.cancelled() => { + return Err(CoreError::StallTimeout { + node_id: node.id().to_string(), + }); + } + } + } else { + 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?; + result } - } + }; + + state.record(node.id(), &node_result); // Determine next step let last_outcome = state.node_outcomes.get(node.id()).unwrap(); @@ -169,12 +198,23 @@ impl Executor { .resolve_next_step(&node, last_outcome, &state, graph) .await?; + // Checkpoint AFTER edge selection so next_node_id is known + let next_node_id = match &next { + NextStep::Edge(target) | NextStep::Jump(target) | NextStep::LoopRestart(target) => { + Some(target.as_str()) + } + NextStep::End => None, + }; + self.lifecycle + .on_checkpoint(&node, &node_result, next_node_id, &state) + .await?; + match next { NextStep::Edge(target) | NextStep::Jump(target) => { state.advance(&target); } NextStep::LoopRestart(start_id) => { - state.restart(&start_id); + state.restart(&start_id, Some(Context::new())); self.lifecycle.on_run_start(graph, &state).await?; } NextStep::End => { @@ -328,8 +368,7 @@ impl Executor { 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())) + Ok(NextStep::LoopRestart(target)) } else { Ok(NextStep::Edge(target)) } @@ -340,14 +379,12 @@ impl Executor { } 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) + if outcome.status == StageStatus::Fail { + if let Some(retry_target) = graph.get_retry_target(node.id()) { + return Ok(NextStep::Edge(retry_target)); + } } + Ok(NextStep::End) } } } @@ -355,6 +392,7 @@ impl Executor { #[cfg(test)] mod tests { + use std::sync::atomic::AtomicU32; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -445,7 +483,7 @@ mod tests { #[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) + // retry_target: work → work (retry the failed node) // First call fails, second succeeds let g = TestGraph::new( vec![ @@ -455,7 +493,7 @@ mod tests { vec![TestEdge::new("work", "end")], "work", ) - .with_retry_target("end", "work"); + .with_retry_target("work", "work"); let handler = Arc::new(CountingHandler::new(vec![ Ok(Outcome::fail("first attempt")), @@ -514,7 +552,12 @@ mod tests { .push(format!("after_node:{}", node.id())); Ok(()) } - async fn on_terminal_reached(&self, node: &TestNode, _s: &RunState) { + async fn on_terminal_reached( + &self, + node: &TestNode, + _goal_gates_passed: bool, + _s: &RunState, + ) { self.0 .lock() .unwrap() @@ -544,7 +587,12 @@ mod tests { struct TerminalTracker(Arc>>); #[async_trait] impl RunLifecycle for TerminalTracker { - async fn on_terminal_reached(&self, node: &TestNode, _s: &RunState) { + async fn on_terminal_reached( + &self, + node: &TestNode, + _goal_gates_passed: bool, + _s: &RunState, + ) { self.0 .lock() .unwrap() @@ -563,10 +611,10 @@ mod tests { #[tokio::test] async fn executor_visit_limit_per_node() { - // Node with max_visits=1, but graph loops back to it + // Node with max_visits=2, loops back — fails on 2nd visit (>= semantics) let g = TestGraph::new( vec![ - TestNode::new("loop_node").with_max_visits(1), + TestNode::new("loop_node").with_max_visits(2), TestNode::new("other"), TestNode::terminal("end"), ], @@ -1298,6 +1346,7 @@ mod tests { &self, node: &TestNode, _r: &NodeResult, + _next_node_id: Option<&str>, _s: &RunState, ) -> Result<()> { self.0.lock().unwrap().push(node.id().to_string()); @@ -1443,4 +1492,407 @@ mod tests { executor.run(&g, state).await.unwrap(); assert_eq!(*log.lock().unwrap(), vec!["hello"]); } + + #[tokio::test] + async fn executor_checkpoint_called_after_edge_selection() { + // Verify on_checkpoint receives the resolved next_node_id + let log = Arc::new(Mutex::new(Vec::<(String, Option)>::new())); + struct NextNodeTracker(Arc)>>>); + #[async_trait] + impl RunLifecycle for NextNodeTracker { + async fn on_checkpoint( + &self, + node: &TestNode, + _r: &NodeResult, + next_node_id: Option<&str>, + _s: &RunState, + ) -> Result<()> { + self.0 + .lock() + .unwrap() + .push((node.id().to_string(), next_node_id.map(String::from))); + 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(NextNodeTracker(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + let checkpoints = log.lock().unwrap().clone(); + // "start" checkpoints with next="work", "work" checkpoints with next="end" + assert_eq!( + checkpoints, + vec![ + ("start".to_string(), Some("work".to_string())), + ("work".to_string(), Some("end".to_string())), + ] + ); + } + + #[tokio::test] + async fn executor_terminal_reached_receives_goal_gate_result() { + let log = Arc::new(Mutex::new(Vec::<(String, bool)>::new())); + struct GateTracker(Arc>>); + #[async_trait] + impl RunLifecycle for GateTracker { + async fn on_terminal_reached( + &self, + node: &TestNode, + goal_gates_passed: bool, + _s: &RunState, + ) { + self.0 + .lock() + .unwrap() + .push((node.id().to_string(), goal_gates_passed)); + } + } + + // Test 1: goal gates pass + let g = linear_graph(&["work", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = + ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc>) + .lifecycle(Box::new(GateTracker(log.clone()))) + .build(); + executor.run(&g, state).await.unwrap(); + assert_eq!(log.lock().unwrap().clone(), vec![("end".to_string(), true)]); + + // Test 2: goal gates fail + let log2 = Arc::new(Mutex::new(Vec::<(String, bool)>::new())); + let g2 = TestGraph::new( + vec![ + TestNode::new("work"), + TestNode::terminal("end").with_goal_gate("work", StageStatus::Success), + ], + vec![TestEdge::new("work", "end")], + "work", + ); + let state2 = RunState::new(&g2).unwrap(); + let executor2 = ExecutorBuilder::new( + Arc::new(AlwaysFailHandler::new("nope")) as Arc> + ) + .lifecycle(Box::new(GateTracker(log2.clone()))) + .build(); + executor2.run(&g2, state2).await.unwrap(); + assert_eq!( + log2.lock().unwrap().clone(), + vec![("end".to_string(), false)] + ); + } + + #[tokio::test] + async fn executor_loop_restart_uses_edge_target() { + // loop_restart edge points to "mid" (not graph start "start") + // Verify execution resumes at "mid" after restart + let call_log = Arc::new(Mutex::new(Vec::::new())); + let log_clone = call_log.clone(); + + struct LogHandler(Arc>>); + #[async_trait] + impl NodeHandler for LogHandler { + async fn execute( + &self, + node: &TestNode, + _c: &Context, + _g: &TestGraph, + ) -> Result { + let mut log = self.0.lock().unwrap(); + log.push(node.id().to_string()); + // On first visit to "work", trigger the loop restart via preferred_label + if node.id() == "work" && log.iter().filter(|n| *n == "work").count() == 1 { + let mut o = Outcome::success(); + o.preferred_label = Some("restart".into()); + return Ok(o); + } + Ok(Outcome::success()) + } + } + + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::new("mid"), + TestNode::new("work"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("start", "mid"), + TestEdge::new("mid", "work"), + TestEdge::new("work", "end"), + // loop_restart edge targets "mid", NOT "start" + TestEdge::new("work", "mid") + .with_label("restart") + .with_loop_restart(), + ], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new( + Arc::new(LogHandler(log_clone)) as Arc> + ) + .max_node_visits(5) + .build(); + executor.run(&g, state).await.unwrap(); + // After restart, execution resumes at "mid" (not "start") + let log = call_log.lock().unwrap().clone(); + assert_eq!(log, vec!["start", "mid", "work", "mid", "work"]); + } + + #[tokio::test] + async fn executor_loop_restart_resets_context() { + // Verify context is fresh after restart (no leaked keys from prior iteration) + struct ContextChecker { + log: Arc>>>, + } + #[async_trait] + impl NodeHandler for ContextChecker { + async fn execute( + &self, + node: &TestNode, + context: &Context, + _g: &TestGraph, + ) -> Result { + if node.id() == "work" { + // Record whether "leaked_key" exists in context + self.log.lock().unwrap().push(context.get("leaked_key")); + // Set a key that should NOT survive restart + let mut o = Outcome::success(); + o.context_updates + .insert("leaked_key".into(), serde_json::json!("should_not_persist")); + // First visit triggers restart + let visits = self.log.lock().unwrap().len(); + if visits == 1 { + o.preferred_label = Some("restart".into()); + } + return Ok(o); + } + Ok(Outcome::success()) + } + } + + let log = Arc::new(Mutex::new(Vec::new())); + let g = TestGraph::new( + vec![ + TestNode::new("start"), + TestNode::new("work"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("start", "work"), + TestEdge::new("work", "end"), + TestEdge::new("work", "start") + .with_label("restart") + .with_loop_restart(), + ], + "start", + ); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new( + Arc::new(ContextChecker { log: log.clone() }) as Arc> + ) + .max_node_visits(5) + .build(); + executor.run(&g, state).await.unwrap(); + let ctx_values = log.lock().unwrap().clone(); + // First visit: no leaked_key yet + assert_eq!(ctx_values[0], None); + // Second visit (after restart): leaked_key should be gone (fresh context) + assert_eq!(ctx_values[1], None); + } + + #[tokio::test] + async fn executor_goal_gate_retry_uses_failed_node_id() { + // Goal gate fails on node "work", retry target defined on "work" + // Verify retry goes there (not to terminal node "end") + let handler = Arc::new(CountingHandler::new(vec![ + Ok(Outcome::fail("first attempt")), + Ok(Outcome::success()), + ])); + 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("work", "work"); + + 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_fail_no_edge_checks_retry_target() { + // Node fails with no outgoing edge, but retry_target is defined + let handler = Arc::new(CountingHandler::new(vec![ + Ok(Outcome::fail("boom")), + Ok(Outcome::success()), + ])); + let g = TestGraph::new( + vec![ + TestNode::new("work"), + TestNode::new("recovery"), + TestNode::terminal("end"), + ], + vec![ + // "work" has only a "success" edge — fail won't match + TestEdge::new("work", "end").with_label("success"), + TestEdge::new("recovery", "end"), + ], + "work", + ) + .with_retry_target("work", "recovery"); + + 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(), 2); + } + + #[tokio::test] + async fn executor_stall_token_interrupts_handler() { + // stall token cancelled during handler execution returns StallTimeout + let stall = CancellationToken::new(); + let stall_clone = stall.clone(); + + struct SlowHandler(CancellationToken); + #[async_trait] + impl NodeHandler for SlowHandler { + async fn execute( + &self, + _n: &TestNode, + _c: &Context, + _g: &TestGraph, + ) -> Result { + // Cancel stall token while "running" + self.0.cancel(); + // Simulate long work + tokio::time::sleep(Duration::from_secs(10)).await; + Ok(Outcome::success()) + } + } + + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new( + Arc::new(SlowHandler(stall_clone)) as Arc> + ) + .stall_token(stall) + .build(); + let result = executor.run(&g, state).await; + match result { + Err(CoreError::StallTimeout { ref node_id }) => { + assert_eq!(node_id, "start"); + } + other => panic!("expected StallTimeout, got {:?}", other), + } + } + + #[tokio::test] + async fn executor_stall_token_interrupts_backoff_sleep() { + // stall token cancelled during retry backoff sleep returns StallTimeout + let stall = CancellationToken::new(); + let stall_clone = stall.clone(); + + struct FailOnceHandler { + stall: CancellationToken, + calls: AtomicU32, + } + #[async_trait] + impl NodeHandler for FailOnceHandler { + async fn execute( + &self, + _n: &TestNode, + _c: &Context, + _g: &TestGraph, + ) -> Result { + let c = self.calls.fetch_add(1, Ordering::Relaxed); + if c == 0 { + // First call: fail with retryable, then cancel stall during backoff + self.stall.cancel(); + Err(CoreError::handler(HandlerErrorDetail { + message: "transient".into(), + retryable: true, + category: None, + signature: None, + })) + } else { + Ok(Outcome::success()) + } + } + fn retry_policy(&self, _n: &TestNode, _g: &TestGraph) -> RetryPolicy { + RetryPolicy { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_secs(60), + factor: 1.0, + max_delay: Duration::from_secs(60), + jitter: false, + }, + } + } + } + + let g = linear_graph(&["start", "end"]); + let state = RunState::new(&g).unwrap(); + let executor = ExecutorBuilder::new(Arc::new(FailOnceHandler { + stall: stall_clone, + calls: AtomicU32::new(0), + }) as Arc>) + .stall_token(stall) + .build(); + let result = executor.run(&g, state).await; + assert!( + matches!(result, Err(CoreError::StallTimeout { .. })), + "expected StallTimeout, got {:?}", + result + ); + } + + #[tokio::test] + async fn executor_stall_token_interrupts_before_attempt() { + // stall token cancelled during a slow before_attempt lifecycle callback + let stall = CancellationToken::new(); + let stall_clone = stall.clone(); + + struct SlowBeforeAttempt(CancellationToken); + #[async_trait] + impl RunLifecycle for SlowBeforeAttempt { + async fn before_attempt( + &self, + _ctx: &AttemptContext<'_, TestGraph>, + _s: &RunState, + ) -> Result { + self.0.cancel(); + tokio::time::sleep(Duration::from_secs(10)).await; + 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(SlowBeforeAttempt(stall_clone))) + .stall_token(stall) + .build(); + let result = executor.run(&g, state).await; + assert!( + matches!(result, Err(CoreError::StallTimeout { .. })), + "expected StallTimeout, got {:?}", + result + ); + } } diff --git a/lib/crates/fabro-core/src/lifecycle.rs b/lib/crates/fabro-core/src/lifecycle.rs index 4f8ea2160..71fff12da 100644 --- a/lib/crates/fabro-core/src/lifecycle.rs +++ b/lib/crates/fabro-core/src/lifecycle.rs @@ -50,7 +50,13 @@ pub trait RunLifecycle: Send + Sync { Ok(()) } - async fn on_terminal_reached(&self, _node: &G::Node, _state: &RunState) {} + async fn on_terminal_reached( + &self, + _node: &G::Node, + _goal_gates_passed: bool, + _state: &RunState, + ) { + } async fn before_node(&self, _node: &G::Node, _state: &RunState) -> Result { Ok(NodeDecision::Continue) @@ -93,6 +99,7 @@ pub trait RunLifecycle: Send + Sync { &self, _node: &G::Node, _result: &NodeResult, + _next_node_id: Option<&str>, _state: &RunState, ) -> Result<()> { Ok(()) @@ -128,9 +135,11 @@ impl RunLifecycle for CompositeLifecycle { Ok(()) } - async fn on_terminal_reached(&self, node: &G::Node, state: &RunState) { + async fn on_terminal_reached(&self, node: &G::Node, goal_gates_passed: bool, state: &RunState) { for child in &self.children { - child.on_terminal_reached(node, state).await; + child + .on_terminal_reached(node, goal_gates_passed, state) + .await; } } @@ -199,10 +208,13 @@ impl RunLifecycle for CompositeLifecycle { &self, node: &G::Node, result: &NodeResult, + next_node_id: Option<&str>, state: &RunState, ) -> Result<()> { for child in &self.children { - child.on_checkpoint(node, result, state).await?; + child + .on_checkpoint(node, result, next_node_id, state) + .await?; } Ok(()) } @@ -268,7 +280,12 @@ mod tests { Ok(()) } - async fn on_terminal_reached(&self, _node: &TestNode, _state: &RunState) { + async fn on_terminal_reached( + &self, + _node: &TestNode, + _goal_gates_passed: bool, + _state: &RunState, + ) { self.log .lock() .unwrap() @@ -351,6 +368,7 @@ mod tests { &self, _node: &TestNode, _result: &NodeResult, + _next_node_id: Option<&str>, _state: &RunState, ) -> Result<()> { self.log diff --git a/lib/crates/fabro-core/src/retry.rs b/lib/crates/fabro-core/src/retry.rs index af04c3192..07a603a62 100644 --- a/lib/crates/fabro-core/src/retry.rs +++ b/lib/crates/fabro-core/src/retry.rs @@ -1,5 +1,7 @@ use std::time::Duration; +use rand::Rng; + #[derive(Debug, Clone)] pub struct BackoffPolicy { pub initial_delay: Duration, @@ -22,11 +24,18 @@ impl Default for BackoffPolicy { 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 { + let base_delay = self.initial_delay.mul_f64(multiplier); + let capped = if base_delay > self.max_delay { self.max_delay } else { - delay + base_delay + }; + if self.jitter { + // Apply jitter: random factor in [0.5, 1.5) + let jitter_factor = rand::thread_rng().gen_range(0.5..1.5); + capped.mul_f64(jitter_factor) + } else { + capped } } } @@ -106,4 +115,28 @@ mod tests { let p = RetryPolicy::none(); assert_eq!(p.max_attempts, 1); } + + #[test] + fn backoff_delay_with_jitter_within_range() { + let b = BackoffPolicy { + initial_delay: Duration::from_millis(1000), + factor: 1.0, + max_delay: Duration::from_secs(10), + jitter: true, + }; + let base = Duration::from_millis(1000); + let min = base.mul_f64(0.5); + let max = base.mul_f64(1.5); + + for _ in 0..100 { + let delay = b.delay_for_attempt(1); + assert!( + delay >= min && delay <= max, + "delay {:?} out of range [{:?}, {:?}]", + delay, + min, + max, + ); + } + } } diff --git a/lib/crates/fabro-core/src/state.rs b/lib/crates/fabro-core/src/state.rs index ff8f3ea46..ac0de7471 100644 --- a/lib/crates/fabro-core/src/state.rs +++ b/lib/crates/fabro-core/src/state.rs @@ -14,6 +14,7 @@ pub struct RunState { pub node_visits: HashMap, pub stage_index: usize, pub previous_node_id: Option, + pub cancelled: bool, } impl RunState { @@ -28,6 +29,7 @@ impl RunState { node_visits: HashMap::new(), stage_index: 0, previous_node_id: None, + cancelled: false, }) } @@ -48,13 +50,16 @@ impl RunState { self.current_node_id = next_node_id.to_string(); } - pub fn restart(&mut self, start_node_id: &str) { + pub fn restart(&mut self, start_node_id: &str, new_context: Option) { 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; + if let Some(ctx) = new_context { + self.context = ctx; + } // node_visits is NOT cleared — preserves total visit counts across restarts } @@ -142,7 +147,7 @@ mod tests { ); state.advance("work"); - state.restart("start"); + state.restart("start", None); assert_eq!(state.current_node_id, "start"); assert!(state.completed_nodes.is_empty()); @@ -171,4 +176,35 @@ mod tests { assert_eq!(state.increment_visits("start"), 2); assert_eq!(state.increment_visits("other"), 1); } + + #[test] + fn run_state_restart_with_new_context() { + let g = linear_graph(&["start", "end"]); + let mut state = RunState::new(&g).unwrap(); + state.context.set("key", json!("old_value")); + state.increment_visits("start"); + + let new_ctx = Context::new(); + new_ctx.set("fresh", json!(true)); + state.restart("start", Some(new_ctx)); + + // Old context key is gone + assert!(state.context.get("key").is_none()); + // New context key is present + assert_eq!(state.context.get("fresh"), Some(json!(true))); + // Visits preserved + assert_eq!(state.node_visits["start"], 1); + } + + #[test] + fn run_state_restart_without_context_preserves() { + let g = linear_graph(&["start", "end"]); + let mut state = RunState::new(&g).unwrap(); + state.context.set("key", json!("value")); + + state.restart("start", None); + + // Context preserved when None passed + assert_eq!(state.context.get("key"), Some(json!("value"))); + } } diff --git a/lib/crates/fabro-core/src/test_fixtures.rs b/lib/crates/fabro-core/src/test_fixtures.rs index 811502d63..6fcf0f675 100644 --- a/lib/crates/fabro-core/src/test_fixtures.rs +++ b/lib/crates/fabro-core/src/test_fixtures.rs @@ -223,17 +223,10 @@ impl Graph for TestGraph { 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 - )); + _ => { + // Return the failed node id (the node whose gate is + // checked), matching fabro-workflows convention + return Err(required_node.clone()); } } } diff --git a/lib/crates/fabro-workflows/Cargo.toml b/lib/crates/fabro-workflows/Cargo.toml index 731bfc59d..1d050e278 100644 --- a/lib/crates/fabro-workflows/Cargo.toml +++ b/lib/crates/fabro-workflows/Cargo.toml @@ -14,6 +14,7 @@ doctest = false [features] default = [] +core-engine = [] exedev = ["fabro-sandbox/exe", "fabro-config/exedev"] [dependencies] @@ -34,6 +35,7 @@ fabro-git-storage = { path = "../fabro-git-storage" } fabro-llm = { path = "../fabro-llm" } fabro-model = { path = "../fabro-model" } fabro-retro = { path = "../fabro-retro" } +fabro-core = { path = "../fabro-core" } thiserror.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/lib/crates/fabro-workflows/src/context/mod.rs b/lib/crates/fabro-workflows/src/context/mod.rs index d4a97834f..b1260f87b 100644 --- a/lib/crates/fabro-workflows/src/context/mod.rs +++ b/lib/crates/fabro-workflows/src/context/mod.rs @@ -116,6 +116,16 @@ impl Context { } } + // --- Internal accessors for bridge code --- + + pub(crate) fn values_arc(&self) -> Arc>> { + self.values.clone() + } + + pub(crate) fn logs_arc(&self) -> Arc>> { + self.logs.clone() + } + // --- Typed accessors --- #[must_use] diff --git a/lib/crates/fabro-workflows/src/core_adapter/context.rs b/lib/crates/fabro-workflows/src/core_adapter/context.rs new file mode 100644 index 000000000..219da382a --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/context.rs @@ -0,0 +1,151 @@ +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; + +use fabro_core::context::{Context as CoreContext, ContextStore}; +use serde_json::Value; + +use crate::context::keys; +use crate::context::Context as WfContext; + +/// A ContextStore implementation that delegates to a wf::Context's internal values map. +struct WfContextStore { + values: Arc>>, +} + +impl ContextStore for WfContextStore { + fn set(&self, key: String, value: Value) { + self.values + .write() + .expect("context lock poisoned") + .insert(key, value); + } + + fn get(&self, key: &str) -> Option { + self.values + .read() + .expect("context lock poisoned") + .get(key) + .cloned() + } + + fn snapshot(&self) -> HashMap { + self.values.read().expect("context lock poisoned").clone() + } + + fn fork(&self) -> Arc { + let cloned = self.values.read().expect("context lock poisoned").clone(); + Arc::new(WfContextStore { + values: Arc::new(RwLock::new(cloned)), + }) + } +} + +/// Create a fabro_core::Context that shares the same underlying values and logs +/// as the given wf::Context. Writes through either are visible to both. +pub fn bridge_context(wf_ctx: &WfContext) -> CoreContext { + let store = Arc::new(WfContextStore { + values: wf_ctx.values_arc(), + }); + CoreContext::with_store_and_logs(store, wf_ctx.logs_arc()) +} + +/// Extension trait providing typed domain accessors on a fabro_core::Context. +pub trait WorkflowContextExt { + fn run_id(&self) -> String; + fn fidelity(&self) -> keys::Fidelity; + fn preamble(&self) -> String; + fn thread_id(&self) -> Option; +} + +impl WorkflowContextExt for CoreContext { + fn run_id(&self) -> String { + self.get_string(keys::INTERNAL_RUN_ID, "unknown") + } + + fn fidelity(&self) -> keys::Fidelity { + self.get_string(keys::INTERNAL_FIDELITY, "") + .parse() + .unwrap_or_default() + } + + fn preamble(&self) -> String { + self.get_string(keys::CURRENT_PREAMBLE, "") + } + + fn thread_id(&self) -> Option { + self.get(keys::INTERNAL_THREAD_ID) + .and_then(|v| v.as_str().map(String::from)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn bridge_shares_values() { + let wf = WfContext::new(); + let core = bridge_context(&wf); + + // Set via wf, read via core + wf.set("key1", json!("from_wf")); + assert_eq!(core.get("key1"), Some(json!("from_wf"))); + + // Set via core, read via wf + core.set("key2", json!("from_core")); + assert_eq!(wf.get("key2"), Some(json!("from_core"))); + } + + #[test] + fn bridge_shares_logs() { + let wf = WfContext::new(); + let core = bridge_context(&wf); + + // Append via wf, read via core + wf.append_log("wf_log"); + assert_eq!(core.logs_snapshot(), vec!["wf_log"]); + + // Append via core, read via wf + core.append_log("core_log"); + assert_eq!(wf.logs_snapshot(), vec!["wf_log", "core_log"]); + } + + #[test] + fn bridge_fork_is_independent() { + let wf = WfContext::new(); + wf.set("shared", json!("original")); + let core = bridge_context(&wf); + let forked = core.clone_context(); + + // Write to fork should not affect original + forked.set("shared", json!("modified")); + assert_eq!(wf.get("shared"), Some(json!("original"))); + assert_eq!(core.get("shared"), Some(json!("original"))); + assert_eq!(forked.get("shared"), Some(json!("modified"))); + } + + #[test] + fn workflow_context_ext_accessors() { + let wf = WfContext::new(); + wf.set(keys::INTERNAL_RUN_ID, json!("run-42")); + wf.set(keys::INTERNAL_FIDELITY, json!("full")); + wf.set(keys::CURRENT_PREAMBLE, json!("You are a helpful assistant")); + wf.set(keys::INTERNAL_THREAD_ID, json!("thread-1")); + + let core = bridge_context(&wf); + assert_eq!(core.run_id(), "run-42"); + assert_eq!(core.fidelity(), keys::Fidelity::Full); + assert_eq!(core.preamble(), "You are a helpful assistant"); + assert_eq!(core.thread_id(), Some("thread-1".to_string())); + } + + #[test] + fn workflow_context_ext_defaults() { + let core = CoreContext::new(); + assert_eq!(core.run_id(), "unknown"); + assert_eq!(core.fidelity(), keys::Fidelity::Compact); + assert_eq!(core.preamble(), ""); + assert_eq!(core.thread_id(), None); + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/graph.rs b/lib/crates/fabro-workflows/src/core_adapter/graph.rs new file mode 100644 index 000000000..448debc41 --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/graph.rs @@ -0,0 +1,135 @@ +use std::collections::HashMap; +use std::sync::Arc; + +use fabro_core::context::Context as CoreContext; +use fabro_core::error::{CoreError, Result as CoreResult}; +use fabro_core::graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; +use fabro_core::outcome::Outcome as CoreOutcome; +use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode}; + +use crate::engine; + +// ---- WorkflowNode ---- + +#[derive(Debug, Clone)] +pub struct WorkflowNode(pub Arc); + +impl WorkflowNode { + pub fn inner(&self) -> &GvNode { + &self.0 + } +} + +impl NodeSpec for WorkflowNode { + fn id(&self) -> &str { + &self.0.id + } + + fn is_terminal(&self) -> bool { + engine::is_terminal(&self.0) + } + + fn max_visits(&self) -> Option { + self.0.max_visits().map(|v| v.max(0) as usize) + } +} + +// ---- WorkflowEdge ---- + +#[derive(Debug, Clone)] +pub struct WorkflowEdge(pub Arc); + +impl WorkflowEdge { + pub fn inner(&self) -> &GvEdge { + &self.0 + } +} + +impl EdgeSpec for WorkflowEdge { + fn target(&self) -> &str { + &self.0.to + } + + fn label(&self) -> Option<&str> { + self.0.label() + } + + fn is_loop_restart(&self) -> bool { + self.0.loop_restart() + } +} + +// ---- WorkflowGraph ---- + +#[derive(Debug, Clone)] +pub struct WorkflowGraph(pub Arc); + +impl WorkflowGraph { + pub fn inner(&self) -> &GvGraph { + &self.0 + } +} + +impl Graph for WorkflowGraph { + type Node = WorkflowNode; + type Edge = WorkflowEdge; + + fn get_node(&self, id: &str) -> Option { + self.0 + .nodes + .get(id) + .map(|n| WorkflowNode(Arc::new(n.clone()))) + } + + fn find_start_node(&self) -> CoreResult { + self.0 + .find_start_node() + .map(|n| WorkflowNode(Arc::new(n.clone()))) + .ok_or(CoreError::NoStartNode) + } + + fn outgoing_edges(&self, node_id: &str) -> Vec { + self.0 + .outgoing_edges(node_id) + .into_iter() + .map(|e| WorkflowEdge(Arc::new(e.clone()))) + .collect() + } + + fn select_edge( + &self, + node: &Self::Node, + outcome: &CoreOutcome, + _context: &CoreContext, + ) -> Option> { + // Convert core outcome to workflow outcome for edge selection + let wf_outcome = super::outcome::core_to_wf_outcome(outcome); + let wf_context = crate::context::Context::new(); + let selection = engine::select_edge( + node.inner(), + &wf_outcome, + &wf_context, + self.inner(), + node.inner().selection(), + ); + selection.map(|sel| EdgeSelection { + edge: WorkflowEdge(Arc::new(sel.edge.clone())), + reason: sel.reason, + }) + } + + fn check_goal_gates( + &self, + outcomes: &HashMap, + ) -> std::result::Result<(), String> { + let wf_outcomes: HashMap = outcomes + .iter() + .map(|(k, v)| (k.clone(), super::outcome::core_to_wf_outcome(v))) + .collect(); + engine::check_goal_gates(self.inner(), &wf_outcomes) + } + + fn get_retry_target(&self, failed_node_id: &str) -> Option { + engine::get_retry_target(failed_node_id, self.inner()) + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/handler.rs b/lib/crates/fabro-workflows/src/core_adapter/handler.rs new file mode 100644 index 000000000..b6bd82b3d --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/handler.rs @@ -0,0 +1,201 @@ +use std::panic::AssertUnwindSafe; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use futures::FutureExt; + +use fabro_core::context::Context as CoreContext; +use fabro_core::error::{CoreError, HandlerErrorDetail, Result as CoreResult}; +use fabro_core::handler::NodeHandler; +use fabro_core::outcome::Outcome as CoreOutcome; +use fabro_core::retry::{BackoffPolicy, RetryPolicy as CoreRetryPolicy}; + +use super::graph::WorkflowGraph; +use super::outcome::{wf_to_core_outcome, wf_to_core_status}; +use super::WorkflowNode; +use crate::engine; +use crate::handler::EngineServices; +use crate::outcome::StageStatus as WfStatus; + +/// Production node handler that bridges fabro-core's NodeHandler to the +/// existing fabro-workflows Handler trait via EngineServices. +pub struct WorkflowNodeHandler { + pub services: Arc, + pub run_dir: PathBuf, +} + +#[async_trait] +impl NodeHandler for WorkflowNodeHandler { + async fn execute( + &self, + node: &WorkflowNode, + _context: &CoreContext, + _graph: &WorkflowGraph, + ) -> CoreResult { + let gv_node = node.inner(); + let handler = self.services.registry.resolve(gv_node); + + // Build a wf context from the core context's state — the lifecycle's + // before_node populates the shared context, so we reconstruct a wf::Context + // that reads from the same store. For now, use the context bridge. + // The actual wf::Context is shared via the bridge set up by the lifecycle. + let wf_context = crate::context::Context::new(); + let wf_graph = fabro_graphviz::graph::types::Graph::new("stub"); + + // Timeout from the node + let node_timeout = gv_node.timeout(); + + // Wrap with panic catch + timeout + let run_dir = self.run_dir.clone(); + let future = crate::handler::dispatch_handler( + handler, + gv_node, + &wf_context, + &wf_graph, + &run_dir, + &self.services, + ); + let panic_safe = AssertUnwindSafe(future).catch_unwind(); + + let timed_result = if let Some(duration) = node_timeout { + match tokio::time::timeout(duration, panic_safe).await { + Ok(inner) => inner, + Err(_elapsed) => { + return Err(CoreError::handler(HandlerErrorDetail { + message: format!("handler timed out after {}ms", duration.as_millis()), + retryable: true, + category: None, + signature: None, + })); + } + } + } else { + panic_safe.await + }; + + match timed_result { + Ok(Ok(wf_outcome)) => Ok(wf_to_core_outcome(&wf_outcome)), + Ok(Err(fabro_err)) => { + // Use the handler's should_retry, not just is_retryable + let retryable = handler.should_retry(&fabro_err); + Err(CoreError::handler(HandlerErrorDetail { + message: fabro_err.to_string(), + retryable, + category: Some(fabro_err.failure_class().to_string()), + signature: fabro_err.failure_signature_hint(), + })) + } + Err(panic_payload) => { + let msg = if let Some(s) = panic_payload.downcast_ref::<&str>() { + format!("handler panicked: {s}") + } else if let Some(s) = panic_payload.downcast_ref::() { + format!("handler panicked: {s}") + } else { + "handler panicked".to_string() + }; + Err(CoreError::handler(HandlerErrorDetail { + message: msg, + retryable: false, + category: None, + signature: None, + })) + } + } + } + + fn retry_policy(&self, node: &WorkflowNode, _graph: &WorkflowGraph) -> CoreRetryPolicy { + let gv_node = node.inner(); + let gv_graph = fabro_graphviz::graph::types::Graph::new("stub"); + let wf_policy = engine::build_retry_policy(gv_node, &gv_graph); + CoreRetryPolicy { + max_attempts: wf_policy.max_attempts, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(wf_policy.backoff.initial_delay_ms), + factor: wf_policy.backoff.backoff_factor, + max_delay: Duration::from_millis(wf_policy.backoff.max_delay_ms), + jitter: wf_policy.backoff.jitter, + }, + } + } + + fn on_retries_exhausted(&self, node: &WorkflowNode, last_outcome: CoreOutcome) -> CoreOutcome { + let gv_node = node.inner(); + if gv_node.allow_partial() { + CoreOutcome { + status: fabro_core::outcome::StageStatus::PartialSuccess, + ..last_outcome + } + } else { + let status = wf_to_core_status(&WfStatus::Fail); + CoreOutcome { + status, + ..last_outcome + } + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use fabro_core::executor::ExecutorBuilder; + use fabro_core::lifecycle::NoopLifecycle; + use fabro_core::outcome::StageStatus; + use fabro_core::state::RunState; + use fabro_graphviz::graph::types::{Edge, Graph, Node}; + use fabro_graphviz::graph::AttrValue; + + use super::super::graph::WorkflowGraph; + use super::*; + + /// Minimal spike handler that always succeeds — proves the trait plumbing. + pub struct SpikeHandler; + + #[async_trait] + impl NodeHandler for SpikeHandler { + async fn execute( + &self, + _node: &WorkflowNode, + _context: &CoreContext, + _graph: &WorkflowGraph, + ) -> CoreResult { + Ok(CoreOutcome::success()) + } + + fn retry_policy(&self, _node: &WorkflowNode, _graph: &WorkflowGraph) -> CoreRetryPolicy { + CoreRetryPolicy::none() + } + } + + #[tokio::test] + async fn spike_core_executor_runs_start_to_exit() { + // Build a minimal graph: start [Mdiamond] → exit [Msquare] + let mut graph = Graph::new("test"); + let mut start = Node::new("start"); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); + let mut exit = Node::new("exit"); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); + graph.nodes.insert("start".to_string(), start); + graph.nodes.insert("exit".to_string(), exit); + graph.edges.push(Edge::new("start", "exit")); + + let wf_graph = WorkflowGraph(Arc::new(graph)); + let handler: Arc> = Arc::new(SpikeHandler); + let state = RunState::new(&wf_graph).unwrap(); + + let executor = ExecutorBuilder::new(handler) + .lifecycle(Box::new(NoopLifecycle)) + .build(); + let result = executor.run(&wf_graph, state).await.unwrap(); + assert_eq!(result.status, StageStatus::Success); + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs new file mode 100644 index 000000000..365e4334c --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs @@ -0,0 +1,579 @@ +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use async_trait::async_trait; + +use fabro_core::error::{CoreError, Result as CoreResult}; +use fabro_core::graph::NodeSpec; +use fabro_core::lifecycle::{ + AttemptContext, AttemptResultContext, EdgeContext, EdgeDecision, NodeDecision, RunLifecycle, +}; +use fabro_core::outcome::{NodeResult, Outcome as CoreOutcome}; +use fabro_core::state::RunState; + +use super::graph::WorkflowGraph; +use super::outcome::{core_to_wf_outcome, core_to_wf_status}; +use super::WorkflowNode; +use crate::checkpoint::Checkpoint; +use crate::context::keys; +use crate::error::{FailureClass, FailureSignature}; +use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::outcome::StageStatus as WfStatus; +use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner}; +use fabro_sandbox::Sandbox; + +/// Data captured from an edge selection to pass to the next node's before_node. +#[derive(Debug, Clone)] +struct IncomingEdgeData { + fidelity: Option, + thread_id: Option, +} + +/// Implements the full RunLifecycle for fabro-workflows, mapping all domain +/// concerns (events, hooks, git, disk I/O, fidelity, circuit breaker, etc.) +/// into fabro-core lifecycle callbacks. +pub struct WorkflowLifecycle { + pub emitter: Arc, + pub hook_runner: Option>, + pub sandbox: Arc, + pub graph: Arc, + pub run_dir: PathBuf, + pub run_id: String, + pub run_start: Instant, + pub labels: HashMap, + // Circuit breaker state + loop_failure_signatures: Mutex>, + restart_failure_signatures: Mutex>, + // Edge data for next node + incoming_edge_data: Mutex>, + // Config flags + pub dry_run: bool, + pub checkpoint_enabled: bool, +} + +impl WorkflowLifecycle { + #[allow(clippy::too_many_arguments)] + pub fn new( + emitter: Arc, + hook_runner: Option>, + sandbox: Arc, + graph: Arc, + run_dir: PathBuf, + run_id: String, + dry_run: bool, + labels: HashMap, + ) -> Self { + Self { + emitter, + hook_runner, + sandbox, + graph, + run_dir, + run_id, + run_start: Instant::now(), + labels, + loop_failure_signatures: Mutex::new(HashMap::new()), + restart_failure_signatures: Mutex::new(HashMap::new()), + incoming_edge_data: Mutex::new(None), + dry_run, + checkpoint_enabled: true, + } + } + + /// Restore circuit breaker state from a checkpoint (for resume). + pub fn restore_circuit_breaker( + &self, + loop_sigs: HashMap, + restart_sigs: HashMap, + ) { + *self.loop_failure_signatures.lock().unwrap() = loop_sigs; + *self.restart_failure_signatures.lock().unwrap() = restart_sigs; + } + + async fn run_hook(&self, hook_ctx: &HookContext) -> HookDecision { + let Some(ref runner) = self.hook_runner else { + return HookDecision::Proceed; + }; + runner + .run(hook_ctx, self.sandbox.clone(), Some(&self.run_dir)) + .await + } +} + +#[async_trait] +impl RunLifecycle for WorkflowLifecycle { + async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &RunState) -> CoreResult<()> { + // Clear incoming edge data (reset stale fidelity/thread from prior iteration) + *self.incoming_edge_data.lock().unwrap() = None; + + // Emit WorkflowRunStarted event + self.emitter.emit(&WorkflowRunEvent::WorkflowRunStarted { + name: self.graph.name.clone(), + run_id: self.run_id.clone(), + base_sha: None, + run_branch: None, + worktree_dir: None, + goal: None, + }); + + // RunStart hook (blocking) + let hook_ctx = HookContext::new( + HookEvent::RunStart, + self.run_id.clone(), + self.graph.name.clone(), + ); + let decision = self.run_hook(&hook_ctx).await; + if let HookDecision::Block { reason } = decision { + let msg = reason.unwrap_or_else(|| "blocked by RunStart hook".into()); + return Err(CoreError::blocked(msg)); + } + + Ok(()) + } + + async fn on_terminal_reached( + &self, + node: &WorkflowNode, + goal_gates_passed: bool, + state: &RunState, + ) { + if !goal_gates_passed { + return; + } + let gv = node.inner(); + let stage_index = state.stage_index; + // Emit StageStarted + StageCompleted for the terminal node + self.emitter.emit(&WorkflowRunEvent::StageStarted { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + handler_type: gv.handler_type().map(String::from), + script: None, + attempt: 1, + max_attempts: 1, + }); + self.emitter.emit(&WorkflowRunEvent::StageCompleted { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + duration_ms: 0, + status: "success".to_string(), + preferred_label: None, + suggested_next_ids: Vec::new(), + usage: None, + failure: None, + notes: None, + files_touched: Vec::new(), + attempt: 1, + max_attempts: 1, + }); + } + + async fn before_node(&self, node: &WorkflowNode, state: &RunState) -> CoreResult { + // Resolve fidelity from incoming edge data + let incoming = self.incoming_edge_data.lock().unwrap().take(); + let gv_node = node.inner(); + + // Set context keys for the current node + // Note: This operates on state.context which is the core context bridged to wf context + let visits = state.node_visits.get(node.id()).copied().unwrap_or(0); + state + .context + .set(keys::CURRENT_NODE, serde_json::json!(node.id())); + state + .context + .set(keys::INTERNAL_NODE_VISIT_COUNT, serde_json::json!(visits)); + + // Fidelity resolution + let fidelity = if let Some(ref edge_data) = incoming { + edge_data + .fidelity + .as_deref() + .or(gv_node.fidelity()) + .unwrap_or("compact") + .to_string() + } else { + gv_node.fidelity().unwrap_or("compact").to_string() + }; + state + .context + .set(keys::INTERNAL_FIDELITY, serde_json::json!(fidelity)); + + // Thread ID resolution + if let Some(ref edge_data) = incoming { + if let Some(ref tid) = edge_data.thread_id { + state + .context + .set(keys::INTERNAL_THREAD_ID, serde_json::json!(tid)); + } + } else if let Some(tid) = gv_node.thread_id() { + state + .context + .set(keys::INTERNAL_THREAD_ID, serde_json::json!(tid)); + } + + Ok(NodeDecision::Continue) + } + + async fn before_attempt( + &self, + ctx: &AttemptContext<'_, WorkflowGraph>, + state: &RunState, + ) -> CoreResult { + let gv = ctx.node.inner(); + let stage_index = state.stage_index; + + // StageStart hook (blocking) + let hook_ctx = HookContext::new( + HookEvent::StageStart, + self.run_id.clone(), + self.graph.name.clone(), + ); + let decision = self.run_hook(&hook_ctx).await; + match decision { + HookDecision::Skip { reason } => { + let msg = reason.unwrap_or_else(|| "skipped by hook".into()); + return Ok(NodeDecision::Skip(Box::new(CoreOutcome::skipped(&msg)))); + } + HookDecision::Block { reason } => { + let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into()); + return Err(CoreError::blocked(msg)); + } + _ => {} + } + + // Emit StageStarted event + self.emitter.emit(&WorkflowRunEvent::StageStarted { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + handler_type: gv.handler_type().map(String::from), + script: None, + attempt: ctx.attempt as usize, + max_attempts: ctx.max_attempts as usize, + }); + + Ok(NodeDecision::Continue) + } + + async fn after_attempt( + &self, + ctx: &AttemptResultContext<'_, WorkflowGraph>, + state: &RunState, + ) -> CoreResult<()> { + if ctx.will_retry { + let gv = ctx.node.inner(); + let wf_outcome = core_to_wf_outcome(&ctx.result.outcome); + let stage_index = state.stage_index; + + // Emit StageFailed event + self.emitter.emit(&WorkflowRunEvent::StageFailed { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + failure: wf_outcome.failure.unwrap_or_else(|| { + crate::outcome::FailureDetail::new( + "handler failed", + FailureClass::TransientInfra, + ) + }), + will_retry: true, + }); + + // Emit StageRetrying event + self.emitter.emit(&WorkflowRunEvent::StageRetrying { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + attempt: ctx.attempt as usize, + max_attempts: ctx.result.max_attempts as usize, + delay_ms: ctx.backoff_delay.map(|d| d.as_millis() as u64).unwrap_or(0), + }); + } + Ok(()) + } + + async fn after_node( + &self, + node: &WorkflowNode, + result: &mut NodeResult, + state: &RunState, + ) -> CoreResult<()> { + let gv = node.inner(); + let stage_index = state.stage_index; + let mut wf_outcome = core_to_wf_outcome(&result.outcome); + + // Auto-status override + if gv.auto_status() + && wf_outcome.status != WfStatus::Success + && wf_outcome.status != WfStatus::Skipped + { + wf_outcome.status = WfStatus::Success; + wf_outcome.notes = + Some("auto-status: handler completed without writing status".to_string()); + result.outcome.status = fabro_core::outcome::StageStatus::Success; + result.outcome.notes = wf_outcome.notes.clone(); + } + + // Circuit breaker: classify + track failure signatures + let outcome_failure_class = if wf_outcome.status == WfStatus::Fail { + wf_outcome.failure.as_ref().map(|f| f.failure_class) + } else { + None + }; + + if let Some(fc) = outcome_failure_class { + let sig_hint = wf_outcome + .failure + .as_ref() + .and_then(|f| f.failure_signature.as_deref()); + let sig = FailureSignature::new( + &gv.id, + fc, + sig_hint, + wf_outcome.failure.as_ref().map(|f| f.message.as_str()), + ); + if fc.is_signature_tracked() { + let mut sigs = self.loop_failure_signatures.lock().unwrap(); + let count = sigs.entry(sig.clone()).or_insert(0); + *count += 1; + let limit = self.graph.loop_restart_signature_limit(); + if *count >= limit { + return Err(CoreError::Other(format!( + "deterministic failure cycle detected: signature {sig} repeated {count} times (limit {limit})" + ))); + } + } + } + + // Emit StageCompleted or StageFailed event + let duration_ms = result.duration.as_millis() as u64; + if wf_outcome.status == WfStatus::Fail { + self.emitter.emit(&WorkflowRunEvent::StageFailed { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + failure: wf_outcome.failure.clone().unwrap_or_else(|| { + crate::outcome::FailureDetail::new( + "handler failed", + FailureClass::Deterministic, + ) + }), + will_retry: false, + }); + } else { + self.emitter.emit(&WorkflowRunEvent::StageCompleted { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + duration_ms, + status: wf_outcome.status.to_string(), + preferred_label: wf_outcome.preferred_label.clone(), + suggested_next_ids: wf_outcome.suggested_next_ids.clone(), + usage: wf_outcome.usage.clone(), + failure: None, + notes: wf_outcome.notes.clone(), + files_touched: wf_outcome.files_touched.clone(), + attempt: result.attempts as usize, + max_attempts: result.max_attempts as usize, + }); + } + + // StageComplete/StageFailed hook (non-blocking) + let hook_event = if wf_outcome.status == WfStatus::Fail { + HookEvent::StageFailed + } else { + HookEvent::StageComplete + }; + let mut hook_ctx = + HookContext::new(hook_event, self.run_id.clone(), self.graph.name.clone()); + hook_ctx.status = Some(wf_outcome.status.to_string()); + let _ = self.run_hook(&hook_ctx).await; + + // Write node status + let status_dir = self.run_dir.join("stages").join(&gv.id); + let _ = std::fs::create_dir_all(&status_dir); + let status_path = status_dir.join("status.json"); + let _ = crate::save_json(&wf_outcome, &status_path, "node_status"); + + Ok(()) + } + + async fn on_edge_selected( + &self, + ctx: &EdgeContext<'_, WorkflowGraph>, + _state: &RunState, + ) -> CoreResult { + // Capture fidelity/thread from edge for next node + if let Some(ref edge) = ctx.edge { + let gv_edge = edge.inner(); + let edge_data = IncomingEdgeData { + fidelity: gv_edge.fidelity().map(String::from), + thread_id: gv_edge.thread_id().map(String::from), + }; + *self.incoming_edge_data.lock().unwrap() = Some(edge_data); + } + + // Compute outcome-derived fields for EdgeSelected event + let wf_outcome = core_to_wf_outcome(ctx.outcome); + + // Emit EdgeSelected event + let label = ctx + .edge + .as_ref() + .and_then(|e| e.inner().label().map(String::from)); + let condition = ctx + .edge + .as_ref() + .and_then(|e| e.inner().condition().map(String::from)); + self.emitter.emit(&WorkflowRunEvent::EdgeSelected { + from_node: ctx.from.to_string(), + to_node: ctx.to.to_string(), + label, + condition, + reason: ctx.reason.to_string(), + preferred_label: wf_outcome.preferred_label.clone(), + suggested_next_ids: wf_outcome.suggested_next_ids.clone(), + stage_status: wf_outcome.status.to_string(), + is_jump: ctx.is_jump, + }); + + // EdgeSelected hook (blocking, can override) + let mut hook_ctx = HookContext::new( + HookEvent::EdgeSelected, + self.run_id.clone(), + self.graph.name.clone(), + ); + hook_ctx.edge_from = Some(ctx.from.to_string()); + hook_ctx.edge_to = Some(ctx.to.to_string()); + let decision = self.run_hook(&hook_ctx).await; + match decision { + HookDecision::Override { edge_to } => { + return Ok(EdgeDecision::Override(edge_to)); + } + HookDecision::Block { reason } => { + let msg = reason.unwrap_or_else(|| "blocked by EdgeSelected hook".into()); + return Err(CoreError::blocked(msg)); + } + _ => {} + } + + // Circuit breaker for loop_restart edges + if let Some(ref edge) = ctx.edge { + if edge.inner().loop_restart() { + // Check restart_failure_signatures limit + // (implemented in Phase 5 when full checkpoint resume is wired) + } + } + + Ok(EdgeDecision::Continue) + } + + async fn on_checkpoint( + &self, + node: &WorkflowNode, + result: &NodeResult, + next_node_id: Option<&str>, + state: &RunState, + ) -> CoreResult<()> { + if !self.checkpoint_enabled { + return Ok(()); + } + + // Build checkpoint from state + let wf_outcome = core_to_wf_outcome(&result.outcome); + let mut node_outcomes: HashMap = state + .node_outcomes + .iter() + .map(|(k, v)| (k.clone(), core_to_wf_outcome(v))) + .collect(); + // Include current node's outcome + node_outcomes.insert(node.id().to_string(), wf_outcome); + + let checkpoint = Checkpoint { + timestamp: chrono::Utc::now(), + current_node: node.id().to_string(), + completed_nodes: state.completed_nodes.clone(), + node_outcomes, + node_retries: state.node_retries.clone(), + context_values: state.context.snapshot(), + logs: state.context.logs_snapshot(), + next_node_id: next_node_id.map(String::from), + git_commit_sha: None, + node_visits: state.node_visits.clone(), + loop_failure_signatures: self.loop_failure_signatures.lock().unwrap().clone(), + restart_failure_signatures: self.restart_failure_signatures.lock().unwrap().clone(), + }; + + // Write checkpoint.json + let checkpoint_path = self.run_dir.join("checkpoint.json"); + if let Err(e) = checkpoint.save(&checkpoint_path) { + state + .context + .append_log(format!("checkpoint save failed: {e}")); + } + + // Emit CheckpointCompleted event + let status = core_to_wf_status(&result.outcome.status).to_string(); + self.emitter.emit(&WorkflowRunEvent::CheckpointCompleted { + node_id: node.id().to_string(), + status, + git_commit_sha: None, + }); + + Ok(()) + } + + async fn on_run_end(&self, outcome: &CoreOutcome, state: &RunState) { + // If cancelled, skip all events/hooks + if state.cancelled { + return; + } + + let duration_ms = self.run_start.elapsed().as_millis() as u64; + let wf_outcome = core_to_wf_outcome(outcome); + + if wf_outcome.status == WfStatus::Success || wf_outcome.status == WfStatus::PartialSuccess { + // Success path + self.emitter.emit(&WorkflowRunEvent::WorkflowRunCompleted { + duration_ms, + artifact_count: 0, + status: wf_outcome.status.to_string(), + total_cost: None, + final_git_commit_sha: None, + usage: None, + }); + + // RunComplete hook + let hook_ctx = HookContext::new( + HookEvent::RunComplete, + self.run_id.clone(), + self.graph.name.clone(), + ); + let _ = self.run_hook(&hook_ctx).await; + } else { + // Failure path + let error_msg = wf_outcome + .failure + .as_ref() + .map(|f| f.message.clone()) + .unwrap_or_else(|| "run failed".to_string()); + self.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed { + error: crate::error::FabroError::engine(error_msg.clone()), + duration_ms, + git_commit_sha: None, + }); + + // RunFailed hook + let mut hook_ctx = HookContext::new( + HookEvent::RunFailed, + self.run_id.clone(), + self.graph.name.clone(), + ); + hook_ctx.failure_reason = Some(error_msg); + let _ = self.run_hook(&hook_ctx).await; + } + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/mod.rs b/lib/crates/fabro-workflows/src/core_adapter/mod.rs new file mode 100644 index 000000000..5c05f2d45 --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/mod.rs @@ -0,0 +1,10 @@ +pub mod context; +pub mod graph; +pub mod handler; +pub mod lifecycle; +pub mod outcome; + +pub use context::{bridge_context, WorkflowContextExt}; +pub use graph::{WorkflowEdge, WorkflowGraph, WorkflowNode}; +pub use handler::WorkflowNodeHandler; +pub use lifecycle::WorkflowLifecycle; diff --git a/lib/crates/fabro-workflows/src/core_adapter/outcome.rs b/lib/crates/fabro-workflows/src/core_adapter/outcome.rs new file mode 100644 index 000000000..4f07daeaa --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/outcome.rs @@ -0,0 +1,173 @@ +use fabro_core::outcome::{ + FailureDetail as CoreFailureDetail, Outcome as CoreOutcome, StageStatus as CoreStatus, +}; + +use crate::error::{classify_failure_reason, FailureClass}; +use crate::outcome::{ + FailureDetail as WfFailureDetail, Outcome as WfOutcome, StageStatus as WfStatus, +}; + +pub fn wf_to_core_status(s: &WfStatus) -> CoreStatus { + match s { + WfStatus::Success => CoreStatus::Success, + WfStatus::Fail => CoreStatus::Fail, + WfStatus::Skipped => CoreStatus::Skipped, + WfStatus::PartialSuccess => CoreStatus::PartialSuccess, + WfStatus::Retry => CoreStatus::Retry, + } +} + +pub fn core_to_wf_status(s: &CoreStatus) -> WfStatus { + match s { + CoreStatus::Success => WfStatus::Success, + CoreStatus::Fail => WfStatus::Fail, + CoreStatus::Skipped => WfStatus::Skipped, + CoreStatus::PartialSuccess => WfStatus::PartialSuccess, + CoreStatus::Retry => WfStatus::Retry, + } +} + +pub fn wf_to_core_outcome(wf: &WfOutcome) -> CoreOutcome { + CoreOutcome { + status: wf_to_core_status(&wf.status), + preferred_label: wf.preferred_label.clone(), + suggested_next_ids: wf.suggested_next_ids.clone(), + context_updates: wf.context_updates.clone(), + jump_to_node: wf.jump_to_node.clone(), + notes: wf.notes.clone(), + failure: wf.failure.as_ref().map(wf_to_core_failure), + metadata: Default::default(), + } +} + +pub fn core_to_wf_outcome(core: &CoreOutcome) -> WfOutcome { + WfOutcome { + status: core_to_wf_status(&core.status), + preferred_label: core.preferred_label.clone(), + suggested_next_ids: core.suggested_next_ids.clone(), + context_updates: core.context_updates.clone(), + jump_to_node: core.jump_to_node.clone(), + notes: core.notes.clone(), + failure: core.failure.as_ref().map(core_to_wf_failure), + usage: None, + files_touched: Vec::new(), + duration_ms: None, + } +} + +pub fn wf_to_core_failure(wf: &WfFailureDetail) -> CoreFailureDetail { + CoreFailureDetail { + message: wf.message.clone(), + category: Some(wf.failure_class.to_string()), + signature: wf.failure_signature.clone(), + } +} + +pub fn core_to_wf_failure(core: &CoreFailureDetail) -> WfFailureDetail { + let failure_class = core + .category + .as_deref() + .and_then(|c| c.parse::().ok()) + .unwrap_or_else(|| classify_failure_reason(&core.message)); + WfFailureDetail { + message: core.message.clone(), + failure_class, + failure_signature: core.signature.clone(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn status_roundtrip_all_variants() { + let wf_statuses = [ + WfStatus::Success, + WfStatus::Fail, + WfStatus::Skipped, + WfStatus::PartialSuccess, + WfStatus::Retry, + ]; + for wf in &wf_statuses { + let core = wf_to_core_status(wf); + let back = core_to_wf_status(&core); + assert_eq!(&back, wf, "roundtrip failed for {:?}", wf); + } + } + + #[test] + fn outcome_roundtrip_success() { + let wf = WfOutcome::success(); + let core = wf_to_core_outcome(&wf); + let back = core_to_wf_outcome(&core); + assert_eq!(back.status, WfStatus::Success); + assert!(back.failure.is_none()); + } + + #[test] + fn outcome_roundtrip_with_shared_fields() { + let mut wf = WfOutcome::success(); + wf.preferred_label = Some("next".into()); + wf.suggested_next_ids = vec!["a".into(), "b".into()]; + wf.context_updates.insert("key".into(), json!("val")); + wf.jump_to_node = Some("target".into()); + wf.notes = Some("hello".into()); + + let core = wf_to_core_outcome(&wf); + assert_eq!(core.preferred_label.as_deref(), Some("next")); + assert_eq!(core.suggested_next_ids, vec!["a", "b"]); + assert_eq!(core.context_updates.get("key"), Some(&json!("val"))); + assert_eq!(core.jump_to_node.as_deref(), Some("target")); + assert_eq!(core.notes.as_deref(), Some("hello")); + + let back = core_to_wf_outcome(&core); + assert_eq!(back.preferred_label, wf.preferred_label); + assert_eq!(back.suggested_next_ids, wf.suggested_next_ids); + assert_eq!(back.context_updates, wf.context_updates); + assert_eq!(back.jump_to_node, wf.jump_to_node); + assert_eq!(back.notes, wf.notes); + } + + #[test] + fn failure_roundtrip() { + let wf_failure = WfFailureDetail { + message: "api down".into(), + failure_class: FailureClass::TransientInfra, + failure_signature: Some("sig123".into()), + }; + let core = wf_to_core_failure(&wf_failure); + assert_eq!(core.message, "api down"); + assert_eq!(core.category.as_deref(), Some("transient_infra")); + assert_eq!(core.signature.as_deref(), Some("sig123")); + + let back = core_to_wf_failure(&core); + assert_eq!(back.message, "api down"); + assert_eq!(back.failure_class, FailureClass::TransientInfra); + assert_eq!(back.failure_signature.as_deref(), Some("sig123")); + } + + #[test] + fn outcome_roundtrip_fail_with_failure() { + let wf = WfOutcome::fail_classify("timeout talking to LLM"); + let core = wf_to_core_outcome(&wf); + let back = core_to_wf_outcome(&core); + assert_eq!(back.status, WfStatus::Fail); + let f = back.failure.unwrap(); + assert_eq!(f.message, "timeout talking to LLM"); + } + + #[test] + fn core_to_wf_failure_classifies_unknown_category() { + let core = CoreFailureDetail { + message: "something broke".into(), + category: None, + signature: None, + }; + let wf = core_to_wf_failure(&core); + // Should fall back to classify_failure_reason + assert_eq!(wf.message, "something broke"); + // The class should be some valid FailureClass (exact value depends on classifier) + } +} diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index 4367126d3..86e71519b 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -193,7 +193,7 @@ impl RetryPolicy { /// Build a retry policy from node and graph attributes. /// If the node has a `retry_policy` attribute naming a preset, use that. /// Otherwise, fall back to `max_retries` / graph default. -fn build_retry_policy(node: &Node, graph: &Graph) -> RetryPolicy { +pub(crate) fn build_retry_policy(node: &Node, graph: &Graph) -> RetryPolicy { if let Some(preset) = node.retry_policy() { match preset { "none" => return RetryPolicy::none(), @@ -535,7 +535,7 @@ pub fn select_edge<'a>( /// Check if all goal gates have been satisfied. /// Returns Ok(()) if all gates passed, or Err with the failed node ID. -fn check_goal_gates( +pub(crate) fn check_goal_gates( graph: &Graph, node_outcomes: &HashMap, ) -> std::result::Result<(), String> { @@ -553,7 +553,7 @@ fn check_goal_gates( } /// Resolve the retry target for a failed goal gate node. -fn get_retry_target(failed_node_id: &str, graph: &Graph) -> Option { +pub(crate) fn get_retry_target(failed_node_id: &str, graph: &Graph) -> Option { if let Some(node) = graph.nodes.get(failed_node_id) { // Node-level retry_target if let Some(target) = node.retry_target() { @@ -584,7 +584,7 @@ fn get_retry_target(failed_node_id: &str, graph: &Graph) -> Option { } /// Check whether a node is a terminal (exit) node. -fn is_terminal(node: &Node) -> bool { +pub(crate) fn is_terminal(node: &Node) -> bool { node.shape() == "Msquare" || node.handler_type() == Some("exit") } @@ -1449,6 +1449,212 @@ impl WorkflowRunEngine { Ok(outcome) } + /// Run the workflow through the fabro-core executor with full lifecycle management. + #[cfg(feature = "core-engine")] + #[allow(dead_code)] + async fn run_via_core( + &self, + graph: &Graph, + config: &RunConfig, + resume_checkpoint: Option<&Checkpoint>, + seed_context: Option, + ) -> Result<(Outcome, Context)> { + use fabro_core::executor::ExecutorBuilder; + use fabro_core::state::RunState; + use tokio_util::sync::CancellationToken; + + let wf_graph = crate::core_adapter::WorkflowGraph(std::sync::Arc::new(graph.clone())); + + // Build a shared EngineServices for the handler + let shared_services = std::sync::Arc::new(EngineServices { + registry: Arc::clone(&self.services.registry), + emitter: Arc::clone(&self.services.emitter), + sandbox: Arc::clone(&self.services.sandbox), + git_state: std::sync::RwLock::new(None), + hook_runner: self.services.hook_runner.clone(), + env: self.services.env.clone(), + dry_run: self.services.dry_run, + }); + + // Build handler + let handler = std::sync::Arc::new(crate::core_adapter::WorkflowNodeHandler { + services: shared_services, + run_dir: config.run_dir.clone(), + }); + + // Build lifecycle + let lifecycle = crate::core_adapter::WorkflowLifecycle::new( + self.services.emitter.clone(), + self.services.hook_runner.clone(), + self.services.sandbox.clone(), + std::sync::Arc::new(graph.clone()), + config.run_dir.clone(), + config.run_id.clone(), + config.dry_run, + config.labels.clone(), + ); + + // Restore circuit breaker state from checkpoint + if let Some(cp) = resume_checkpoint { + lifecycle.restore_circuit_breaker( + cp.loop_failure_signatures.clone(), + cp.restart_failure_signatures.clone(), + ); + } + + // Build RunState + let state = if let Some(cp) = resume_checkpoint { + // Resume from checkpoint + let mut s = RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string()))?; + // Restore context values + for (k, v) in &cp.context_values { + s.context.set(k.clone(), v.clone()); + } + // Restore logs + for log in &cp.logs { + s.context.append_log(log.clone()); + } + s.completed_nodes = cp.completed_nodes.clone(); + s.node_retries = cp.node_retries.clone(); + s.node_visits = cp.node_visits.clone(); + // Restore node outcomes + for (k, v) in &cp.node_outcomes { + s.node_outcomes.insert( + k.clone(), + crate::core_adapter::outcome::wf_to_core_outcome(v), + ); + } + // Set start node to the checkpoint's next_node_id + if let Some(ref next) = cp.next_node_id { + s.current_node_id = next.clone(); + } + s + } else if let Some(seed) = seed_context { + let s = RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string()))?; + // Populate from seed context + for (k, v) in seed.snapshot() { + s.context.set(k, v); + } + for log in seed.logs_snapshot() { + s.context.append_log(log); + } + s + } else { + RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string()))? + }; + + // Compute global visit limit + let graph_max = graph.max_node_visits(); + let max_node_visits = if graph_max > 0 { + Some(graph_max as usize) + } else if config.dry_run { + Some(10) + } else { + None + }; + + // Set up stall watchdog + let stall_token = graph.stall_timeout().map(|_| CancellationToken::new()); + let stall_shutdown = + if let (Some(stall_timeout), Some(ref token)) = (graph.stall_timeout(), &stall_token) { + let shutdown = CancellationToken::new(); + let emitter = self.services.emitter.clone(); + let token_clone = token.clone(); + let shutdown_clone = shutdown.clone(); + emitter.touch(); + tokio::spawn(async move { + loop { + tokio::select! { + _ = tokio::time::sleep(stall_timeout) => { + if shutdown_clone.is_cancelled() { + return; + } + // Check if there's been recent activity + let last = emitter.last_event_at(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64; + let idle_ms = now.saturating_sub(last); + if idle_ms >= stall_timeout.as_millis() as i64 { + token_clone.cancel(); + return; + } + } + _ = shutdown_clone.cancelled() => { + return; + } + } + } + }); + Some(shutdown) + } else { + None + }; + + // Build executor + let mut builder = ExecutorBuilder::new( + handler + as std::sync::Arc< + dyn fabro_core::handler::NodeHandler, + >, + ) + .lifecycle(Box::new(lifecycle)); + + if let Some(ref cancel) = config.cancel_token { + builder = builder.cancel_token(cancel.clone()); + } + if let Some(token) = stall_token.clone() { + builder = builder.stall_token(token); + } + if let Some(limit) = max_node_visits { + builder = builder.max_node_visits(limit); + } + + let executor = builder.build(); + + // Run + let result = executor.run(&wf_graph, state).await; + + // Shut down stall poller + if let Some(shutdown) = stall_shutdown { + shutdown.cancel(); + } + + // Convert result + match result { + Ok(core_outcome) => { + let wf_outcome = crate::core_adapter::outcome::core_to_wf_outcome(&core_outcome); + // Return outcome + a fresh context (lifecycle manages the real context) + let ctx = Context::new(); + Ok((wf_outcome, ctx)) + } + Err(fabro_core::CoreError::StallTimeout { node_id }) => { + let stall_timeout = graph.stall_timeout().unwrap_or_default(); + let idle_secs = stall_timeout.as_secs(); + self.services + .emitter + .emit(&WorkflowRunEvent::StallWatchdogTimeout { + node: node_id.clone(), + idle_seconds: idle_secs, + }); + Err(FabroError::engine(format!( + "stall watchdog: node \"{node_id}\" had no activity for {idle_secs}s" + ))) + } + Err(fabro_core::CoreError::Cancelled) => Err(FabroError::Cancelled), + Err(fabro_core::CoreError::Blocked { message }) => Err(FabroError::engine(message)), + Err(fabro_core::CoreError::VisitLimitExceeded { + node_id, + visits, + limit, + }) => Err(FabroError::engine(format!( + "node \"{node_id}\" visited {visits} times (limit {limit})" + ))), + Err(e) => Err(FabroError::engine(e.to_string())), + } + } + /// Internal run implementation supporting optional checkpoint resume and `start_at` override. async fn run_internal( &self, diff --git a/lib/crates/fabro-workflows/src/lib.rs b/lib/crates/fabro-workflows/src/lib.rs index f77355620..7b3471b1c 100644 --- a/lib/crates/fabro-workflows/src/lib.rs +++ b/lib/crates/fabro-workflows/src/lib.rs @@ -101,6 +101,7 @@ pub mod checkpoint; pub mod conclusion; pub mod condition; pub mod context; +pub mod core_adapter; pub mod cost; pub mod devcontainer_bridge; pub mod engine;