scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-05 08:10:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Wire fabro-workflows to use fabro-core executor engine

Browse source 
Fix fabro-core semantics to match fabro-workflows (checkpoint after edge
selection, terminal callback with goal-gate result, loop restart uses edge
target with fresh context, retry-target routing for failed nodes, visit
limit >= semantics, stall token with CancellationToken, backoff jitter).

Add core_adapter module bridging fabro-workflows types to fabro-core traits:
WorkflowGraph/Node/Edge newtypes, bidirectional outcome conversion, context
bridge sharing values/logs via ContextStore, WorkflowNodeHandler with
panic/timeout protection, and full WorkflowLifecycle implementing all 8
RunLifecycle callbacks (events, hooks, fidelity, circuit breaker, checkpoints).

Add run_via_core method behind core-engine feature flag that builds and runs
the fabro-core Executor with the full adapter suite. The existing run_internal
path remains the default.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-23 23:13:00 -04:00
parent 48ea05cfde
commit ef37c656f0
19 changed files with 2074 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Cargo.lock generated
View file

@ -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",

2
lib/crates/fabro-core/Cargo.toml
View file

@ -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]

7
lib/crates/fabro-core/src/context.rs
View file

@ -76,6 +76,13 @@ impl Context {
}
}
pub fn with_store_and_logs(
store: Arc<dyn ContextStore>,
logs: Arc<RwLock<Vec<String>>>,
) -> Self {
Self { store, logs }
}
pub fn set(&self, key: impl Into<String>, value: Value) {
self.store.set(key.into(), value);
}

9
lib/crates/fabro-core/src/error.rs
View file

@ -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(),

518
lib/crates/fabro-core/src/executor.rs
View file

@ -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<Arc<AtomicBool>>,
pub stall_token: Option<CancellationToken>,
pub max_node_visits: Option<usize>,
}
@ -56,6 +60,11 @@ impl<G: Graph + 'static> ExecutorBuilder<G> {
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<G: Graph + 'static> Executor<G> {
// 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<G: Graph + 'static> Executor<G> {
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<G: Graph + 'static> Executor<G> {
}
}
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<G: Graph + 'static> Executor<G> {
}
// 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<G: Graph + 'static> Executor<G> {
.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<G: Graph + 'static> Executor<G> {
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<G: Graph + 'static> Executor<G> {
}
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<G: Graph + 'static> Executor<G> {
#[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<Mutex<Vec<String>>>);
#[async_trait]
impl RunLifecycle<TestGraph> 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<String>)>::new()));
struct NextNodeTracker(Arc<Mutex<Vec<(String, Option<String>)>>>);
#[async_trait]
impl RunLifecycle<TestGraph> 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<dyn NodeHandler<TestGraph>>)
.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<Mutex<Vec<(String, bool)>>>);
#[async_trait]
impl RunLifecycle<TestGraph> 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<dyn NodeHandler<TestGraph>>)
.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<dyn NodeHandler<TestGraph>>
)
.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::<String>::new()));
let log_clone = call_log.clone();
struct LogHandler(Arc<Mutex<Vec<String>>>);
#[async_trait]
impl NodeHandler<TestGraph> for LogHandler {
async fn execute(
&self,
node: &TestNode,
_c: &Context,
_g: &TestGraph,
) -> Result<Outcome> {
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<dyn NodeHandler<TestGraph>>
)
.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<Mutex<Vec<Option<serde_json::Value>>>>,
}
#[async_trait]
impl NodeHandler<TestGraph> for ContextChecker {
async fn execute(
&self,
node: &TestNode,
context: &Context,
_g: &TestGraph,
) -> Result<Outcome> {
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<dyn NodeHandler<TestGraph>>
)
.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<dyn NodeHandler<TestGraph>>).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<dyn NodeHandler<TestGraph>>)
.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<TestGraph> for SlowHandler {
async fn execute(
&self,
_n: &TestNode,
_c: &Context,
_g: &TestGraph,
) -> Result<Outcome> {
// 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<dyn NodeHandler<TestGraph>>
)
.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<TestGraph> for FailOnceHandler {
async fn execute(
&self,
_n: &TestNode,
_c: &Context,
_g: &TestGraph,
) -> Result<Outcome> {
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<dyn NodeHandler<TestGraph>>)
.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<TestGraph> for SlowBeforeAttempt {
async fn before_attempt(
&self,
_ctx: &AttemptContext<'_, TestGraph>,
_s: &RunState,
) -> Result<NodeDecision> {
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<dyn NodeHandler<TestGraph>>)
.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
);
}
}

28
lib/crates/fabro-core/src/lifecycle.rs
View file

@ -50,7 +50,13 @@ pub trait RunLifecycle<G: Graph>: 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<NodeDecision> {
Ok(NodeDecision::Continue)
@ -93,6 +99,7 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
&self,
_node: &G::Node,
_result: &NodeResult,
_next_node_id: Option<&str>,
_state: &RunState,
) -> Result<()> {
Ok(())
@ -128,9 +135,11 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
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<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
&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

39
lib/crates/fabro-core/src/retry.rs
View file

@ -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,
);
}
}
}

40
lib/crates/fabro-core/src/state.rs
View file

@ -14,6 +14,7 @@ pub struct RunState {
pub node_visits: HashMap<String, usize>,
pub stage_index: usize,
pub previous_node_id: Option<String>,
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<Context>) {
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")));
}
}

15
lib/crates/fabro-core/src/test_fixtures.rs
View file

@ -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());
}
}
}

2
lib/crates/fabro-workflows/Cargo.toml
View file

@ -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

10
lib/crates/fabro-workflows/src/context/mod.rs
View file

@ -116,6 +116,16 @@ impl Context {
}
}
// --- Internal accessors for bridge code ---
pub(crate) fn values_arc(&self) -> Arc<RwLock<HashMap<String, Value>>> {
self.values.clone()
}
pub(crate) fn logs_arc(&self) -> Arc<RwLock<Vec<String>>> {
self.logs.clone()
}
// --- Typed accessors ---
#[must_use]

151
lib/crates/fabro-workflows/src/core_adapter/context.rs Normal file
View file

@ -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<RwLock<HashMap<String, Value>>>,
}
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<Value> {
self.values
.read()
.expect("context lock poisoned")
.get(key)
.cloned()
}
fn snapshot(&self) -> HashMap<String, Value> {
self.values.read().expect("context lock poisoned").clone()
}
fn fork(&self) -> Arc<dyn ContextStore> {
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<String>;
}
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<String> {
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);
}
}

135
lib/crates/fabro-workflows/src/core_adapter/graph.rs Normal file
View file

@ -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<GvNode>);
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<usize> {
self.0.max_visits().map(|v| v.max(0) as usize)
}
}
// ---- WorkflowEdge ----
#[derive(Debug, Clone)]
pub struct WorkflowEdge(pub Arc<GvEdge>);
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<GvGraph>);
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::Node> {
self.0
.nodes
.get(id)
.map(|n| WorkflowNode(Arc::new(n.clone())))
}
fn find_start_node(&self) -> CoreResult<Self::Node> {
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::Edge> {
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<EdgeSelection<Self>> {
// 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<String, CoreOutcome>,
) -> std::result::Result<(), String> {
let wf_outcomes: HashMap<String, crate::outcome::Outcome> = 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<String> {
engine::get_retry_target(failed_node_id, self.inner())
}
}

201
lib/crates/fabro-workflows/src/core_adapter/handler.rs Normal file
View file

@ -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<EngineServices>,
pub run_dir: PathBuf,
}
#[async_trait]
impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
async fn execute(
&self,
node: &WorkflowNode,
_context: &CoreContext,
_graph: &WorkflowGraph,
) -> CoreResult<CoreOutcome> {
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::<String>() {
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<WorkflowGraph> for SpikeHandler {
async fn execute(
&self,
_node: &WorkflowNode,
_context: &CoreContext,
_graph: &WorkflowGraph,
) -> CoreResult<CoreOutcome> {
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<dyn NodeHandler<WorkflowGraph>> = 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);
}
}

579
lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs Normal file
View file

@ -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<String>,
thread_id: Option<String>,
}
/// 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<EventEmitter>,
pub hook_runner: Option<Arc<HookRunner>>,
pub sandbox: Arc<dyn Sandbox>,
pub graph: Arc<fabro_graphviz::graph::types::Graph>,
pub run_dir: PathBuf,
pub run_id: String,
pub run_start: Instant,
pub labels: HashMap<String, String>,
// Circuit breaker state
loop_failure_signatures: Mutex<HashMap<FailureSignature, usize>>,
restart_failure_signatures: Mutex<HashMap<FailureSignature, usize>>,
// Edge data for next node
incoming_edge_data: Mutex<Option<IncomingEdgeData>>,
// Config flags
pub dry_run: bool,
pub checkpoint_enabled: bool,
}
impl WorkflowLifecycle {
#[allow(clippy::too_many_arguments)]
pub fn new(
emitter: Arc<EventEmitter>,
hook_runner: Option<Arc<HookRunner>>,
sandbox: Arc<dyn Sandbox>,
graph: Arc<fabro_graphviz::graph::types::Graph>,
run_dir: PathBuf,
run_id: String,
dry_run: bool,
labels: HashMap<String, String>,
) -> 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<FailureSignature, usize>,
restart_sigs: HashMap<FailureSignature, usize>,
) {
*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<WorkflowGraph> 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<NodeDecision> {
// 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<NodeDecision> {
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<EdgeDecision> {
// 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<String, crate::outcome::Outcome> = 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;
}
}
}

10
lib/crates/fabro-workflows/src/core_adapter/mod.rs Normal file
View file

@ -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;

173
lib/crates/fabro-workflows/src/core_adapter/outcome.rs Normal file
View file

@ -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::<FailureClass>().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)
}
}

214
lib/crates/fabro-workflows/src/engine.rs
View file

@ -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<String, Outcome>,
) -> 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<String> {
pub(crate) fn get_retry_target(failed_node_id: &str, graph: &Graph) -> Option<String> {
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<String> {
}
/// 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<Context>,
) -> 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<crate::core_adapter::WorkflowGraph>,
>,
)
.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,

1
lib/crates/fabro-workflows/src/lib.rs
View file

@ -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;