Add fabro-core crate: generic workflow execution engine

Standalone crate with no knowledge of git, fidelity, LLMs, hooks,
sandboxes, artifacts, or disk I/O. Provides a ~55-line executor loop
with pluggable NodeHandler, Graph, and RunLifecycle traits.

Key types: CoreError, StageStatus, Outcome, Context (pluggable store),
RetryPolicy, RunState, Executor/ExecutorBuilder, StallWatchdog.

103 tests covering all milestones: foundation types, graph/handler
traits, lifecycle callbacks, executor (linear paths, terminal nodes,
goal gates, visit limits, edge selection, jumps, loop restarts,
cancellation, retry with backoff), and stall detection.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Cargo.lock

@@ -1403,6 +1403,18 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "fabro-core"
+version = "0.176.2"
+dependencies = [
+ "async-trait",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "fabro-db"
 version = "0.176.2"

lib/crates/fabro-core/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "fabro-core"
+edition.workspace = true
+version.workspace = true
+license.workspace = true
+description = "Generic workflow execution engine"
+
+[lib]
+doctest = false
+
+[dependencies]
+async-trait.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["test-util", "macros"] }

lib/crates/fabro-core/src/context.rs

@@ -0,0 +1,278 @@
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+
+use serde_json::Value;
+
+pub trait ContextStore: Send + Sync {
+    fn set(&self, key: String, value: Value);
+    fn get(&self, key: &str) -> Option<Value>;
+    fn snapshot(&self) -> HashMap<String, Value>;
+    fn fork(&self) -> Arc<dyn ContextStore>;
+}
+
+pub struct InMemoryStore {
+    data: RwLock<HashMap<String, Value>>,
+}
+
+impl InMemoryStore {
+    pub fn new() -> Self {
+        Self {
+            data: RwLock::new(HashMap::new()),
+        }
+    }
+}
+
+impl Default for InMemoryStore {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl ContextStore for InMemoryStore {
+    fn set(&self, key: String, value: Value) {
+        self.data.write().unwrap().insert(key, value);
+    }
+
+    fn get(&self, key: &str) -> Option<Value> {
+        self.data.read().unwrap().get(key).cloned()
+    }
+
+    fn snapshot(&self) -> HashMap<String, Value> {
+        self.data.read().unwrap().clone()
+    }
+
+    fn fork(&self) -> Arc<dyn ContextStore> {
+        let cloned = self.data.read().unwrap().clone();
+        Arc::new(InMemoryStore {
+            data: RwLock::new(cloned),
+        })
+    }
+}
+
+#[derive(Clone)]
+pub struct Context {
+    store: Arc<dyn ContextStore>,
+    logs: Arc<RwLock<Vec<String>>>,
+}
+
+impl Default for Context {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Context {
+    pub fn new() -> Self {
+        Self {
+            store: Arc::new(InMemoryStore::new()),
+            logs: Arc::new(RwLock::new(Vec::new())),
+        }
+    }
+
+    pub fn with_store(store: Arc<dyn ContextStore>) -> Self {
+        Self {
+            store,
+            logs: Arc::new(RwLock::new(Vec::new())),
+        }
+    }
+
+    pub fn set(&self, key: impl Into<String>, value: Value) {
+        self.store.set(key.into(), value);
+    }
+
+    pub fn get(&self, key: &str) -> Option<Value> {
+        self.store.get(key)
+    }
+
+    pub fn get_string(&self, key: &str, default: &str) -> String {
+        self.get(key)
+            .and_then(|v| v.as_str().map(String::from))
+            .unwrap_or_else(|| default.to_string())
+    }
+
+    pub fn apply_updates(&self, updates: &HashMap<String, Value>) {
+        for (k, v) in updates {
+            self.store.set(k.clone(), v.clone());
+        }
+    }
+
+    pub fn snapshot(&self) -> HashMap<String, Value> {
+        self.store.snapshot()
+    }
+
+    pub fn append_log(&self, entry: impl Into<String>) {
+        self.logs.write().unwrap().push(entry.into());
+    }
+
+    pub fn logs_snapshot(&self) -> Vec<String> {
+        self.logs.read().unwrap().clone()
+    }
+
+    pub fn clone_context(&self) -> Self {
+        Self {
+            store: self.store.fork(),
+            logs: Arc::new(RwLock::new(self.logs.read().unwrap().clone())),
+        }
+    }
+
+    // Core typed accessors
+    pub fn current_node_id(&self) -> String {
+        self.get_string("current_node", "")
+    }
+
+    pub fn node_visit_count(&self) -> usize {
+        self.get("internal.node_visit_count")
+            .and_then(|v| v.as_u64())
+            .unwrap_or(0) as usize
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    #[test]
+    fn in_memory_store_set_and_get() {
+        let store = InMemoryStore::new();
+        store.set("k".into(), json!("v"));
+        assert_eq!(store.get("k"), Some(json!("v")));
+        assert_eq!(store.get("missing"), None);
+    }
+
+    #[test]
+    fn in_memory_store_snapshot_is_independent() {
+        let store = InMemoryStore::new();
+        store.set("a".into(), json!(1));
+        let snap = store.snapshot();
+        store.set("b".into(), json!(2));
+        assert!(!snap.contains_key("b"));
+        assert_eq!(snap.len(), 1);
+    }
+
+    #[test]
+    fn in_memory_store_fork() {
+        let store = InMemoryStore::new();
+        store.set("x".into(), json!(10));
+        let forked = store.fork();
+        forked.set("y".into(), json!(20));
+        assert!(store.get("y").is_none());
+        assert_eq!(forked.get("x"), Some(json!(10)));
+        assert_eq!(forked.get("y"), Some(json!(20)));
+    }
+
+    #[test]
+    fn context_set_and_get() {
+        let ctx = Context::new();
+        ctx.set("name", json!("test"));
+        assert_eq!(ctx.get("name"), Some(json!("test")));
+    }
+
+    #[test]
+    fn context_get_missing_returns_none() {
+        let ctx = Context::new();
+        assert_eq!(ctx.get("nope"), None);
+    }
+
+    #[test]
+    fn context_get_string_with_default() {
+        let ctx = Context::new();
+        assert_eq!(ctx.get_string("missing", "fallback"), "fallback");
+        ctx.set("present", json!("value"));
+        assert_eq!(ctx.get_string("present", "fallback"), "value");
+    }
+
+    #[test]
+    fn context_apply_updates() {
+        let ctx = Context::new();
+        let mut updates = HashMap::new();
+        updates.insert("a".into(), json!(1));
+        updates.insert("b".into(), json!(2));
+        ctx.apply_updates(&updates);
+        assert_eq!(ctx.get("a"), Some(json!(1)));
+        assert_eq!(ctx.get("b"), Some(json!(2)));
+    }
+
+    #[test]
+    fn context_clone_is_independent() {
+        let ctx = Context::new();
+        ctx.set("x", json!(1));
+        let cloned = ctx.clone_context();
+        cloned.set("x", json!(2));
+        assert_eq!(ctx.get("x"), Some(json!(1)));
+        assert_eq!(cloned.get("x"), Some(json!(2)));
+    }
+
+    #[test]
+    fn context_append_and_snapshot_logs() {
+        let ctx = Context::new();
+        ctx.append_log("step 1");
+        ctx.append_log("step 2");
+        let logs = ctx.logs_snapshot();
+        assert_eq!(logs, vec!["step 1", "step 2"]);
+    }
+
+    #[test]
+    fn context_with_custom_store() {
+        struct CountingStore {
+            inner: InMemoryStore,
+            set_count: AtomicUsize,
+        }
+        impl ContextStore for CountingStore {
+            fn set(&self, key: String, value: Value) {
+                self.set_count.fetch_add(1, Ordering::Relaxed);
+                self.inner.set(key, value);
+            }
+            fn get(&self, key: &str) -> Option<Value> {
+                self.inner.get(key)
+            }
+            fn snapshot(&self) -> HashMap<String, Value> {
+                self.inner.snapshot()
+            }
+            fn fork(&self) -> Arc<dyn ContextStore> {
+                self.inner.fork()
+            }
+        }
+
+        let store = Arc::new(CountingStore {
+            inner: InMemoryStore::new(),
+            set_count: AtomicUsize::new(0),
+        });
+        let ctx = Context::with_store(store.clone());
+        ctx.set("k", json!(1));
+        ctx.set("k2", json!(2));
+        assert_eq!(store.set_count.load(Ordering::Relaxed), 2);
+        assert_eq!(ctx.get("k"), Some(json!(1)));
+    }
+
+    #[test]
+    fn context_fork_is_independent() {
+        let ctx = Context::new();
+        ctx.set("shared", json!("original"));
+        ctx.append_log("log1");
+        let forked = ctx.clone_context();
+        forked.set("shared", json!("modified"));
+        forked.append_log("log2");
+        assert_eq!(ctx.get("shared"), Some(json!("original")));
+        assert_eq!(ctx.logs_snapshot().len(), 1);
+        assert_eq!(forked.get("shared"), Some(json!("modified")));
+        assert_eq!(forked.logs_snapshot().len(), 2);
+    }
+
+    #[test]
+    fn context_current_node_id() {
+        let ctx = Context::new();
+        assert_eq!(ctx.current_node_id(), "");
+        ctx.set("current_node", json!("node_5"));
+        assert_eq!(ctx.current_node_id(), "node_5");
+    }
+
+    #[test]
+    fn context_node_visit_count() {
+        let ctx = Context::new();
+        assert_eq!(ctx.node_visit_count(), 0);
+        ctx.set("internal.node_visit_count", json!(3));
+        assert_eq!(ctx.node_visit_count(), 3);
+    }
+}

lib/crates/fabro-core/src/error.rs

@@ -0,0 +1,169 @@
+use std::fmt;
+use std::time::Duration;
+
+use crate::outcome::{FailureDetail, Outcome, StageStatus};
+
+/// Structured failure data on handler errors. Maps to FabroError's
+/// is_retryable(), failure_class(), failure_signature_hint(), to_fail_outcome().
+#[derive(Debug, Clone)]
+pub struct HandlerErrorDetail {
+    pub message: String,
+    pub retryable: bool,
+    pub category: Option<String>,
+    pub signature: Option<String>,
+}
+
+impl fmt::Display for HandlerErrorDetail {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.message)
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum CoreError {
+    #[error("node not found: {id}")]
+    NodeNotFound { id: String },
+    #[error("no start node found in graph")]
+    NoStartNode,
+    #[error("run cancelled")]
+    Cancelled,
+    #[error("blocked: {message}")]
+    Blocked { message: String },
+    #[error("node \"{node_id}\" visited {visits} times (limit {limit})")]
+    VisitLimitExceeded {
+        node_id: String,
+        visits: usize,
+        limit: usize,
+    },
+    #[error("stall timeout: no activity for {elapsed:?}")]
+    StallTimeout { elapsed: Duration },
+    #[error("{detail}")]
+    Handler { detail: HandlerErrorDetail },
+    #[error("{0}")]
+    Other(String),
+}
+
+impl CoreError {
+    pub fn handler(detail: HandlerErrorDetail) -> Self {
+        Self::Handler { detail }
+    }
+
+    pub fn blocked(message: impl Into<String>) -> Self {
+        Self::Blocked {
+            message: message.into(),
+        }
+    }
+
+    pub fn is_retryable(&self) -> bool {
+        matches!(self, Self::Handler { detail } if detail.retryable)
+    }
+
+    pub fn to_fail_outcome(&self) -> Outcome {
+        match self {
+            Self::Handler { detail } => Outcome {
+                status: StageStatus::Fail,
+                failure: Some(FailureDetail {
+                    message: detail.message.clone(),
+                    category: detail.category.clone(),
+                    signature: detail.signature.clone(),
+                }),
+                ..Outcome::default()
+            },
+            other => Outcome::fail(&other.to_string()),
+        }
+    }
+}
+
+pub type Result<T> = std::result::Result<T, CoreError>;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn core_error_display() {
+        assert_eq!(
+            CoreError::NodeNotFound { id: "n1".into() }.to_string(),
+            "node not found: n1"
+        );
+        assert_eq!(
+            CoreError::NoStartNode.to_string(),
+            "no start node found in graph"
+        );
+        assert_eq!(CoreError::Cancelled.to_string(), "run cancelled");
+        assert_eq!(
+            CoreError::Blocked {
+                message: "hook denied".into()
+            }
+            .to_string(),
+            "blocked: hook denied"
+        );
+        assert_eq!(
+            CoreError::VisitLimitExceeded {
+                node_id: "n1".into(),
+                visits: 5,
+                limit: 3
+            }
+            .to_string(),
+            "node \"n1\" visited 5 times (limit 3)"
+        );
+        assert_eq!(
+            CoreError::StallTimeout {
+                elapsed: Duration::from_secs(30)
+            }
+            .to_string(),
+            "stall timeout: no activity for 30s"
+        );
+        assert_eq!(
+            CoreError::Other("something broke".into()).to_string(),
+            "something broke"
+        );
+    }
+
+    #[test]
+    fn core_error_handler_is_retryable() {
+        let retryable = CoreError::handler(HandlerErrorDetail {
+            message: "timeout".into(),
+            retryable: true,
+            category: None,
+            signature: None,
+        });
+        assert!(retryable.is_retryable());
+
+        let not_retryable = CoreError::handler(HandlerErrorDetail {
+            message: "bad input".into(),
+            retryable: false,
+            category: None,
+            signature: None,
+        });
+        assert!(!not_retryable.is_retryable());
+    }
+
+    #[test]
+    fn core_error_handler_to_fail_outcome() {
+        let err = CoreError::handler(HandlerErrorDetail {
+            message: "api down".into(),
+            retryable: true,
+            category: Some("transient".into()),
+            signature: Some("sig123".into()),
+        });
+        let outcome = err.to_fail_outcome();
+        assert_eq!(outcome.status, StageStatus::Fail);
+        let failure = outcome.failure.unwrap();
+        assert_eq!(failure.message, "api down");
+        assert_eq!(failure.category.as_deref(), Some("transient"));
+        assert_eq!(failure.signature.as_deref(), Some("sig123"));
+    }
+
+    #[test]
+    fn core_error_non_handler_not_retryable() {
+        assert!(!CoreError::NodeNotFound { id: "x".into() }.is_retryable());
+        assert!(!CoreError::Cancelled.is_retryable());
+        assert!(!CoreError::NoStartNode.is_retryable());
+        assert!(!CoreError::Blocked {
+            message: "no".into()
+        }
+        .is_retryable());
+        assert!(!CoreError::Other("err".into()).is_retryable());
+    }
+}

lib/crates/fabro-core/src/graph.rs

@@ -0,0 +1,42 @@
+use std::collections::HashMap;
+
+use crate::context::Context;
+use crate::error::Result;
+use crate::outcome::Outcome;
+
+pub trait NodeSpec: Send + Sync + Clone {
+    fn id(&self) -> &str;
+    fn is_terminal(&self) -> bool;
+    fn max_visits(&self) -> Option<usize>;
+}
+
+pub trait EdgeSpec: Send + Sync + Clone {
+    fn target(&self) -> &str;
+    fn label(&self) -> Option<&str>;
+    fn is_loop_restart(&self) -> bool;
+}
+
+pub struct EdgeSelection<G: Graph + ?Sized> {
+    pub edge: G::Edge,
+    pub reason: &'static str,
+}
+
+pub trait Graph: Send + Sync {
+    type Node: NodeSpec + Clone;
+    type Edge: EdgeSpec + Clone;
+
+    fn get_node(&self, id: &str) -> Option<Self::Node>;
+    fn find_start_node(&self) -> Result<Self::Node>;
+    fn outgoing_edges(&self, node_id: &str) -> Vec<Self::Edge>;
+    fn select_edge(
+        &self,
+        node: &Self::Node,
+        outcome: &Outcome,
+        context: &Context,
+    ) -> Option<EdgeSelection<Self>>;
+    fn check_goal_gates(
+        &self,
+        outcomes: &HashMap<String, Outcome>,
+    ) -> std::result::Result<(), String>;
+    fn get_retry_target(&self, failed_node_id: &str) -> Option<String>;
+}

lib/crates/fabro-core/src/handler.rs

@@ -0,0 +1,20 @@
+use async_trait::async_trait;
+
+use crate::context::Context;
+use crate::error::Result;
+use crate::graph::Graph;
+use crate::outcome::Outcome;
+use crate::retry::RetryPolicy;
+
+#[async_trait]
+pub trait NodeHandler<G: Graph>: Send + Sync {
+    async fn execute(&self, node: &G::Node, context: &Context, graph: &G) -> Result<Outcome>;
+
+    fn retry_policy(&self, _node: &G::Node, _graph: &G) -> RetryPolicy {
+        RetryPolicy::none()
+    }
+
+    fn on_retries_exhausted(&self, _node: &G::Node, _last_outcome: Outcome) -> Outcome {
+        Outcome::fail("max retries exceeded")
+    }
+}

lib/crates/fabro-core/src/lib.rs

@@ -0,0 +1,27 @@
+pub mod context;
+pub mod error;
+pub mod executor;
+pub mod graph;
+pub mod handler;
+pub mod lifecycle;
+pub mod outcome;
+pub mod retry;
+pub mod stall;
+pub mod state;
+
+#[cfg(test)]
+pub mod test_fixtures;
+
+pub use context::{Context, ContextStore, InMemoryStore};
+pub use error::{CoreError, HandlerErrorDetail, Result};
+pub use executor::{Executor, ExecutorBuilder, ExecutorSettings};
+pub use graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec};
+pub use handler::NodeHandler;
+pub use lifecycle::{
+    AttemptContext, AttemptResultContext, CompositeLifecycle, EdgeContext, EdgeDecision,
+    NodeDecision, NoopLifecycle, RunLifecycle,
+};
+pub use outcome::{FailureDetail, NodeResult, Outcome, StageStatus};
+pub use retry::{BackoffPolicy, RetryPolicy};
+pub use stall::{ActivityMonitor, StallGuard, StallWatchdog};
+pub use state::RunState;

lib/crates/fabro-core/src/lifecycle.rs

@@ -0,0 +1,648 @@
+use std::time::Duration;
+
+use async_trait::async_trait;
+
+use crate::error::Result;
+use crate::graph::Graph;
+use crate::outcome::{NodeResult, Outcome};
+use crate::state::RunState;
+
+#[derive(Debug, Clone)]
+pub enum NodeDecision {
+    Continue,
+    Skip(Box<Outcome>),
+    Block(String),
+}
+
+#[derive(Debug, Clone)]
+pub enum EdgeDecision {
+    Continue,
+    Override(String),
+    Block(String),
+}
+
+pub struct AttemptContext<'a, G: Graph> {
+    pub node: &'a G::Node,
+    pub attempt: u32,
+    pub max_attempts: u32,
+}
+
+pub struct AttemptResultContext<'a, G: Graph> {
+    pub node: &'a G::Node,
+    pub result: &'a NodeResult,
+    pub attempt: u32,
+    pub will_retry: bool,
+    pub backoff_delay: Option<Duration>,
+}
+
+pub struct EdgeContext<'a, G: Graph> {
+    pub from: &'a str,
+    pub to: &'a str,
+    pub edge: Option<G::Edge>,
+    pub is_jump: bool,
+    pub outcome: &'a Outcome,
+    pub reason: &'a str,
+}
+
+#[async_trait]
+pub trait RunLifecycle<G: Graph>: Send + Sync {
+    async fn on_run_start(&self, _graph: &G, _state: &RunState) -> Result<()> {
+        Ok(())
+    }
+
+    async fn on_terminal_reached(&self, _node: &G::Node, _state: &RunState) {}
+
+    async fn before_node(&self, _node: &G::Node, _state: &RunState) -> Result<NodeDecision> {
+        Ok(NodeDecision::Continue)
+    }
+
+    async fn before_attempt(
+        &self,
+        _ctx: &AttemptContext<'_, G>,
+        _state: &RunState,
+    ) -> Result<NodeDecision> {
+        Ok(NodeDecision::Continue)
+    }
+
+    async fn after_attempt(
+        &self,
+        _ctx: &AttemptResultContext<'_, G>,
+        _state: &RunState,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    async fn after_node(
+        &self,
+        _node: &G::Node,
+        _result: &mut NodeResult,
+        _state: &RunState,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    async fn on_edge_selected(
+        &self,
+        _ctx: &EdgeContext<'_, G>,
+        _state: &RunState,
+    ) -> Result<EdgeDecision> {
+        Ok(EdgeDecision::Continue)
+    }
+
+    async fn on_checkpoint(
+        &self,
+        _node: &G::Node,
+        _result: &NodeResult,
+        _state: &RunState,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    async fn on_run_end(&self, _outcome: &Outcome, _state: &RunState) {}
+}
+
+/// No-op lifecycle that passes through everything.
+pub struct NoopLifecycle;
+
+#[async_trait]
+impl<G: Graph> RunLifecycle<G> for NoopLifecycle {}
+
+/// Composes multiple lifecycles, calling them in order. Useful for testing
+/// and simple use cases where fixed ordering suffices.
+pub struct CompositeLifecycle<G: Graph> {
+    children: Vec<Box<dyn RunLifecycle<G>>>,
+}
+
+impl<G: Graph> CompositeLifecycle<G> {
+    pub fn new(children: Vec<Box<dyn RunLifecycle<G>>>) -> Self {
+        Self { children }
+    }
+}
+
+#[async_trait]
+impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
+    async fn on_run_start(&self, graph: &G, state: &RunState) -> Result<()> {
+        for child in &self.children {
+            child.on_run_start(graph, state).await?;
+        }
+        Ok(())
+    }
+
+    async fn on_terminal_reached(&self, node: &G::Node, state: &RunState) {
+        for child in &self.children {
+            child.on_terminal_reached(node, state).await;
+        }
+    }
+
+    async fn before_node(&self, node: &G::Node, state: &RunState) -> Result<NodeDecision> {
+        for child in &self.children {
+            match child.before_node(node, state).await? {
+                NodeDecision::Continue => {}
+                decision => return Ok(decision),
+            }
+        }
+        Ok(NodeDecision::Continue)
+    }
+
+    async fn before_attempt(
+        &self,
+        ctx: &AttemptContext<'_, G>,
+        state: &RunState,
+    ) -> Result<NodeDecision> {
+        for child in &self.children {
+            match child.before_attempt(ctx, state).await? {
+                NodeDecision::Continue => {}
+                decision => return Ok(decision),
+            }
+        }
+        Ok(NodeDecision::Continue)
+    }
+
+    async fn after_attempt(
+        &self,
+        ctx: &AttemptResultContext<'_, G>,
+        state: &RunState,
+    ) -> Result<()> {
+        for child in &self.children {
+            child.after_attempt(ctx, state).await?;
+        }
+        Ok(())
+    }
+
+    async fn after_node(
+        &self,
+        node: &G::Node,
+        result: &mut NodeResult,
+        state: &RunState,
+    ) -> Result<()> {
+        for child in &self.children {
+            child.after_node(node, result, state).await?;
+        }
+        Ok(())
+    }
+
+    async fn on_edge_selected(
+        &self,
+        ctx: &EdgeContext<'_, G>,
+        state: &RunState,
+    ) -> Result<EdgeDecision> {
+        for child in &self.children {
+            match child.on_edge_selected(ctx, state).await? {
+                EdgeDecision::Continue => {}
+                decision => return Ok(decision),
+            }
+        }
+        Ok(EdgeDecision::Continue)
+    }
+
+    async fn on_checkpoint(
+        &self,
+        node: &G::Node,
+        result: &NodeResult,
+        state: &RunState,
+    ) -> Result<()> {
+        for child in &self.children {
+            child.on_checkpoint(node, result, state).await?;
+        }
+        Ok(())
+    }
+
+    async fn on_run_end(&self, outcome: &Outcome, state: &RunState) {
+        for child in &self.children {
+            child.on_run_end(outcome, state).await;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::{AtomicU32, Ordering};
+    use std::sync::{Arc, Mutex};
+
+    use super::*;
+    use crate::test_fixtures::{linear_graph, TestGraph, TestNode};
+
+    /// A lifecycle that records which callbacks were called.
+    struct RecordingLifecycle {
+        name: String,
+        log: Arc<Mutex<Vec<String>>>,
+        before_node_decision: Mutex<Option<NodeDecision>>,
+        before_attempt_decision: Mutex<Option<NodeDecision>>,
+        edge_decision: Mutex<Option<EdgeDecision>>,
+    }
+
+    impl RecordingLifecycle {
+        fn new(name: &str, log: Arc<Mutex<Vec<String>>>) -> Self {
+            Self {
+                name: name.to_string(),
+                log,
+                before_node_decision: Mutex::new(None),
+                before_attempt_decision: Mutex::new(None),
+                edge_decision: Mutex::new(None),
+            }
+        }
+
+        fn with_before_node(self, decision: NodeDecision) -> Self {
+            *self.before_node_decision.lock().unwrap() = Some(decision);
+            self
+        }
+
+        fn with_before_attempt(self, decision: NodeDecision) -> Self {
+            *self.before_attempt_decision.lock().unwrap() = Some(decision);
+            self
+        }
+
+        fn with_edge_decision(self, decision: EdgeDecision) -> Self {
+            *self.edge_decision.lock().unwrap() = Some(decision);
+            self
+        }
+    }
+
+    #[async_trait]
+    impl RunLifecycle<TestGraph> for RecordingLifecycle {
+        async fn on_run_start(&self, _graph: &TestGraph, _state: &RunState) -> Result<()> {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:on_run_start", self.name));
+            Ok(())
+        }
+
+        async fn on_terminal_reached(&self, _node: &TestNode, _state: &RunState) {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:on_terminal_reached", self.name));
+        }
+
+        async fn before_node(&self, _node: &TestNode, _state: &RunState) -> Result<NodeDecision> {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:before_node", self.name));
+            Ok(self
+                .before_node_decision
+                .lock()
+                .unwrap()
+                .take()
+                .unwrap_or(NodeDecision::Continue))
+        }
+
+        async fn before_attempt(
+            &self,
+            _ctx: &AttemptContext<'_, TestGraph>,
+            _state: &RunState,
+        ) -> Result<NodeDecision> {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:before_attempt", self.name));
+            Ok(self
+                .before_attempt_decision
+                .lock()
+                .unwrap()
+                .take()
+                .unwrap_or(NodeDecision::Continue))
+        }
+
+        async fn after_attempt(
+            &self,
+            _ctx: &AttemptResultContext<'_, TestGraph>,
+            _state: &RunState,
+        ) -> Result<()> {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:after_attempt", self.name));
+            Ok(())
+        }
+
+        async fn after_node(
+            &self,
+            _node: &TestNode,
+            _result: &mut NodeResult,
+            _state: &RunState,
+        ) -> Result<()> {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:after_node", self.name));
+            Ok(())
+        }
+
+        async fn on_edge_selected(
+            &self,
+            _ctx: &EdgeContext<'_, TestGraph>,
+            _state: &RunState,
+        ) -> Result<EdgeDecision> {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:on_edge_selected", self.name));
+            Ok(self
+                .edge_decision
+                .lock()
+                .unwrap()
+                .take()
+                .unwrap_or(EdgeDecision::Continue))
+        }
+
+        async fn on_checkpoint(
+            &self,
+            _node: &TestNode,
+            _result: &NodeResult,
+            _state: &RunState,
+        ) -> Result<()> {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:on_checkpoint", self.name));
+            Ok(())
+        }
+
+        async fn on_run_end(&self, _outcome: &Outcome, _state: &RunState) {
+            self.log
+                .lock()
+                .unwrap()
+                .push(format!("{}:on_run_end", self.name));
+        }
+    }
+
+    #[tokio::test]
+    async fn default_lifecycle_is_noop() {
+        let lc = NoopLifecycle;
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        assert!(
+            <NoopLifecycle as RunLifecycle<TestGraph>>::on_run_start(&lc, &g, &state)
+                .await
+                .is_ok()
+        );
+        let node = g.get_node("start").unwrap();
+        assert!(matches!(
+            <NoopLifecycle as RunLifecycle<TestGraph>>::before_node(&lc, &node, &state)
+                .await
+                .unwrap(),
+            NodeDecision::Continue
+        ));
+    }
+
+    #[tokio::test]
+    async fn composite_calls_all_children_on_run_start() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(RecordingLifecycle::new("a", log.clone())),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        lc.on_run_start(&g, &state).await.unwrap();
+        let calls = log.lock().unwrap().clone();
+        assert_eq!(calls, vec!["a:on_run_start", "b:on_run_start"]);
+    }
+
+    #[tokio::test]
+    async fn composite_before_node_skip_short_circuits() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(
+                RecordingLifecycle::new("a", log.clone())
+                    .with_before_node(NodeDecision::Skip(Box::new(Outcome::skipped("hook")))),
+            ),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let node = g.get_node("start").unwrap();
+        let decision = lc.before_node(&node, &state).await.unwrap();
+        assert!(matches!(decision, NodeDecision::Skip(_)));
+        let calls = log.lock().unwrap().clone();
+        assert_eq!(calls, vec!["a:before_node"]);
+        // b was NOT called
+    }
+
+    #[tokio::test]
+    async fn composite_before_node_block_short_circuits() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(
+                RecordingLifecycle::new("a", log.clone())
+                    .with_before_node(NodeDecision::Block("denied".into())),
+            ),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let node = g.get_node("start").unwrap();
+        let decision = lc.before_node(&node, &state).await.unwrap();
+        assert!(matches!(decision, NodeDecision::Block(_)));
+        let calls = log.lock().unwrap().clone();
+        assert_eq!(calls, vec!["a:before_node"]);
+    }
+
+    #[tokio::test]
+    async fn composite_before_attempt_skip_short_circuits() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(
+                RecordingLifecycle::new("a", log.clone())
+                    .with_before_attempt(NodeDecision::Skip(Box::new(Outcome::skipped("skip")))),
+            ),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let node = g.get_node("start").unwrap();
+        let ctx = AttemptContext {
+            node: &node,
+            attempt: 1,
+            max_attempts: 1,
+        };
+        let decision = lc.before_attempt(&ctx, &state).await.unwrap();
+        assert!(matches!(decision, NodeDecision::Skip(_)));
+        let calls = log.lock().unwrap().clone();
+        assert_eq!(calls, vec!["a:before_attempt"]);
+    }
+
+    #[tokio::test]
+    async fn composite_before_attempt_block_short_circuits() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(
+                RecordingLifecycle::new("a", log.clone())
+                    .with_before_attempt(NodeDecision::Block("nope".into())),
+            ),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let node = g.get_node("start").unwrap();
+        let ctx = AttemptContext {
+            node: &node,
+            attempt: 1,
+            max_attempts: 1,
+        };
+        let decision = lc.before_attempt(&ctx, &state).await.unwrap();
+        assert!(matches!(decision, NodeDecision::Block(_)));
+    }
+
+    #[tokio::test]
+    async fn composite_after_attempt_calls_all() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(RecordingLifecycle::new("a", log.clone())),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let node = g.get_node("start").unwrap();
+        let result = NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1);
+        let ctx = AttemptResultContext {
+            node: &node,
+            result: &result,
+            attempt: 1,
+            will_retry: false,
+            backoff_delay: None,
+        };
+        lc.after_attempt(&ctx, &state).await.unwrap();
+        let calls = log.lock().unwrap().clone();
+        assert_eq!(calls, vec!["a:after_attempt", "b:after_attempt"]);
+    }
+
+    #[tokio::test]
+    async fn composite_on_edge_selected_override_short_circuits() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(
+                RecordingLifecycle::new("a", log.clone())
+                    .with_edge_decision(EdgeDecision::Override("other".into())),
+            ),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let outcome = Outcome::success();
+        let edge = g.outgoing_edges("start").into_iter().next().unwrap();
+        let ctx = EdgeContext {
+            from: "start",
+            to: "end",
+            edge: Some(edge),
+            is_jump: false,
+            outcome: &outcome,
+            reason: "unconditional",
+        };
+        let decision = lc.on_edge_selected(&ctx, &state).await.unwrap();
+        assert!(matches!(decision, EdgeDecision::Override(ref t) if t == "other"));
+        let calls = log.lock().unwrap().clone();
+        assert_eq!(calls, vec!["a:on_edge_selected"]);
+    }
+
+    #[tokio::test]
+    async fn composite_on_edge_selected_block_short_circuits() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(
+                RecordingLifecycle::new("a", log.clone())
+                    .with_edge_decision(EdgeDecision::Block("blocked".into())),
+            ),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let outcome = Outcome::success();
+        let ctx = EdgeContext {
+            from: "start",
+            to: "end",
+            edge: None,
+            is_jump: false,
+            outcome: &outcome,
+            reason: "unconditional",
+        };
+        let decision = lc.on_edge_selected(&ctx, &state).await.unwrap();
+        assert!(matches!(decision, EdgeDecision::Block(_)));
+    }
+
+    #[tokio::test]
+    async fn composite_on_edge_selected_none_for_jumps() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![Box::new(RecordingLifecycle::new("a", log.clone()))]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let outcome = Outcome::success();
+        let ctx = EdgeContext::<TestGraph> {
+            from: "start",
+            to: "target",
+            edge: None,
+            is_jump: true,
+            outcome: &outcome,
+            reason: "jump",
+        };
+        let decision = lc.on_edge_selected(&ctx, &state).await.unwrap();
+        assert!(matches!(decision, EdgeDecision::Continue));
+        assert!(ctx.edge.is_none());
+        assert!(ctx.is_jump);
+    }
+
+    #[tokio::test]
+    async fn composite_after_node_calls_all() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(RecordingLifecycle::new("a", log.clone())),
+            Box::new(RecordingLifecycle::new("b", log.clone())),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let node = g.get_node("start").unwrap();
+        let mut result = NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1);
+        lc.after_node(&node, &mut result, &state).await.unwrap();
+        let calls = log.lock().unwrap().clone();
+        assert_eq!(calls, vec!["a:after_node", "b:after_node"]);
+    }
+
+    #[tokio::test]
+    async fn composite_ordering_is_preserved() {
+        let log = Arc::new(Mutex::new(Vec::new()));
+        let counter = Arc::new(AtomicU32::new(0));
+
+        struct OrderedLifecycle {
+            name: String,
+            log: Arc<Mutex<Vec<String>>>,
+            counter: Arc<AtomicU32>,
+        }
+
+        #[async_trait]
+        impl RunLifecycle<TestGraph> for OrderedLifecycle {
+            async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> {
+                let order = self.counter.fetch_add(1, Ordering::SeqCst);
+                self.log
+                    .lock()
+                    .unwrap()
+                    .push(format!("{}:{}", self.name, order));
+                Ok(())
+            }
+        }
+
+        let lc = CompositeLifecycle::new(vec![
+            Box::new(OrderedLifecycle {
+                name: "first".into(),
+                log: log.clone(),
+                counter: counter.clone(),
+            }),
+            Box::new(OrderedLifecycle {
+                name: "second".into(),
+                log: log.clone(),
+                counter: counter.clone(),
+            }),
+            Box::new(OrderedLifecycle {
+                name: "third".into(),
+                log: log.clone(),
+                counter: counter.clone(),
+            }),
+        ]);
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        lc.on_run_start(&g, &state).await.unwrap();
+        let calls = log.lock().unwrap().clone();
+        assert_eq!(calls, vec!["first:0", "second:1", "third:2"]);
+    }
+}

lib/crates/fabro-core/src/outcome.rs

@@ -0,0 +1,256 @@
+use std::collections::HashMap;
+use std::fmt;
+use std::str::FromStr;
+use std::time::Duration;
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum StageStatus {
+    Success,
+    Fail,
+    Skipped,
+    PartialSuccess,
+    Retry,
+}
+
+impl fmt::Display for StageStatus {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Success => write!(f, "success"),
+            Self::Fail => write!(f, "fail"),
+            Self::Skipped => write!(f, "skipped"),
+            Self::PartialSuccess => write!(f, "partial_success"),
+            Self::Retry => write!(f, "retry"),
+        }
+    }
+}
+
+impl FromStr for StageStatus {
+    type Err = String;
+
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+        match s {
+            "success" => Ok(Self::Success),
+            "fail" => Ok(Self::Fail),
+            "skipped" => Ok(Self::Skipped),
+            "partial_success" => Ok(Self::PartialSuccess),
+            "retry" => Ok(Self::Retry),
+            other => Err(format!("unknown stage status: {other}")),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct FailureDetail {
+    pub message: String,
+    pub category: Option<String>,
+    pub signature: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct Outcome {
+    pub status: StageStatus,
+    pub preferred_label: Option<String>,
+    pub suggested_next_ids: Vec<String>,
+    pub context_updates: HashMap<String, Value>,
+    pub jump_to_node: Option<String>,
+    pub notes: Option<String>,
+    pub failure: Option<FailureDetail>,
+    pub metadata: HashMap<String, Value>,
+}
+
+impl Default for Outcome {
+    fn default() -> Self {
+        Self {
+            status: StageStatus::Success,
+            preferred_label: None,
+            suggested_next_ids: Vec::new(),
+            context_updates: HashMap::new(),
+            jump_to_node: None,
+            notes: None,
+            failure: None,
+            metadata: HashMap::new(),
+        }
+    }
+}
+
+impl Outcome {
+    pub fn success() -> Self {
+        Self::default()
+    }
+
+    pub fn fail(message: &str) -> Self {
+        Self {
+            status: StageStatus::Fail,
+            failure: Some(FailureDetail {
+                message: message.to_string(),
+                category: None,
+                signature: None,
+            }),
+            ..Self::default()
+        }
+    }
+
+    pub fn skipped(reason: &str) -> Self {
+        Self {
+            status: StageStatus::Skipped,
+            notes: Some(reason.to_string()),
+            ..Self::default()
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct NodeResult {
+    pub outcome: Outcome,
+    pub duration: Duration,
+    pub attempts: u32,
+    pub max_attempts: u32,
+}
+
+impl NodeResult {
+    pub fn new(outcome: Outcome, duration: Duration, attempts: u32, max_attempts: u32) -> Self {
+        Self {
+            outcome,
+            duration,
+            attempts,
+            max_attempts,
+        }
+    }
+
+    pub fn from_error(
+        error: &crate::error::CoreError,
+        duration: Duration,
+        attempts: u32,
+        max_attempts: u32,
+    ) -> Self {
+        Self {
+            outcome: error.to_fail_outcome(),
+            duration,
+            attempts,
+            max_attempts,
+        }
+    }
+
+    pub fn from_skip(outcome: Outcome) -> Self {
+        Self {
+            outcome,
+            duration: Duration::ZERO,
+            attempts: 0,
+            max_attempts: 0,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn stage_status_display_roundtrip() {
+        let statuses = [
+            StageStatus::Success,
+            StageStatus::Fail,
+            StageStatus::Skipped,
+            StageStatus::PartialSuccess,
+            StageStatus::Retry,
+        ];
+        for status in &statuses {
+            let s = status.to_string();
+            let parsed: StageStatus = s.parse().unwrap();
+            assert_eq!(&parsed, status);
+        }
+    }
+
+    #[test]
+    fn stage_status_serde_roundtrip() {
+        let statuses = [
+            StageStatus::Success,
+            StageStatus::Fail,
+            StageStatus::Skipped,
+            StageStatus::PartialSuccess,
+            StageStatus::Retry,
+        ];
+        for status in &statuses {
+            let json = serde_json::to_string(status).unwrap();
+            let parsed: StageStatus = serde_json::from_str(&json).unwrap();
+            assert_eq!(&parsed, status);
+        }
+    }
+
+    #[test]
+    fn outcome_success_factory() {
+        let o = Outcome::success();
+        assert_eq!(o.status, StageStatus::Success);
+        assert!(o.failure.is_none());
+        assert!(o.notes.is_none());
+    }
+
+    #[test]
+    fn outcome_fail_factory() {
+        let o = Outcome::fail("broken");
+        assert_eq!(o.status, StageStatus::Fail);
+        let f = o.failure.unwrap();
+        assert_eq!(f.message, "broken");
+        assert!(f.category.is_none());
+        assert!(f.signature.is_none());
+    }
+
+    #[test]
+    fn outcome_skipped_factory() {
+        let o = Outcome::skipped("not needed");
+        assert_eq!(o.status, StageStatus::Skipped);
+        assert_eq!(o.notes.as_deref(), Some("not needed"));
+    }
+
+    #[test]
+    fn outcome_with_context_updates() {
+        let mut o = Outcome::success();
+        o.context_updates
+            .insert("key".into(), serde_json::json!("value"));
+        assert_eq!(o.context_updates["key"], serde_json::json!("value"));
+    }
+
+    #[test]
+    fn outcome_with_jump() {
+        let mut o = Outcome::success();
+        o.jump_to_node = Some("target".into());
+        assert_eq!(o.jump_to_node.as_deref(), Some("target"));
+    }
+
+    #[test]
+    fn outcome_serde_roundtrip() {
+        // Test that metadata (the serde-friendly field) roundtrips
+        let mut o = Outcome::success();
+        o.metadata
+            .insert("usage".into(), serde_json::json!({"tokens": 100}));
+        let json = serde_json::to_value(&o.metadata).unwrap();
+        let parsed: HashMap<String, Value> = serde_json::from_value(json).unwrap();
+        assert_eq!(parsed["usage"]["tokens"], 100);
+    }
+
+    #[test]
+    fn failure_detail_construction() {
+        let f = FailureDetail {
+            message: "timeout".into(),
+            category: Some("transient".into()),
+            signature: Some("sig".into()),
+        };
+        assert_eq!(f.message, "timeout");
+        assert_eq!(f.category.as_deref(), Some("transient"));
+        assert_eq!(f.signature.as_deref(), Some("sig"));
+    }
+
+    #[test]
+    fn node_result_from_outcome() {
+        let o = Outcome::success();
+        let r = NodeResult::new(o, Duration::from_millis(100), 1, 3);
+        assert_eq!(r.outcome.status, StageStatus::Success);
+        assert_eq!(r.duration, Duration::from_millis(100));
+        assert_eq!(r.attempts, 1);
+        assert_eq!(r.max_attempts, 3);
+    }
+}

lib/crates/fabro-core/src/retry.rs

@@ -0,0 +1,109 @@
+use std::time::Duration;
+
+#[derive(Debug, Clone)]
+pub struct BackoffPolicy {
+    pub initial_delay: Duration,
+    pub factor: f64,
+    pub max_delay: Duration,
+    pub jitter: bool,
+}
+
+impl Default for BackoffPolicy {
+    fn default() -> Self {
+        Self {
+            initial_delay: Duration::from_secs(1),
+            factor: 2.0,
+            max_delay: Duration::from_secs(60),
+            jitter: false,
+        }
+    }
+}
+
+impl BackoffPolicy {
+    pub fn delay_for_attempt(&self, attempt: u32) -> Duration {
+        let multiplier = self.factor.powi(attempt.saturating_sub(1) as i32);
+        let delay = self.initial_delay.mul_f64(multiplier);
+        if delay > self.max_delay {
+            self.max_delay
+        } else {
+            delay
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct RetryPolicy {
+    pub max_attempts: u32,
+    pub backoff: BackoffPolicy,
+}
+
+impl RetryPolicy {
+    pub fn none() -> Self {
+        Self {
+            max_attempts: 1,
+            backoff: BackoffPolicy::default(),
+        }
+    }
+
+    pub fn with_max_attempts(max_attempts: u32) -> Self {
+        Self {
+            max_attempts,
+            backoff: BackoffPolicy::default(),
+        }
+    }
+}
+
+impl Default for RetryPolicy {
+    fn default() -> Self {
+        Self::none()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn backoff_delay_first_attempt() {
+        let b = BackoffPolicy {
+            initial_delay: Duration::from_millis(100),
+            factor: 2.0,
+            max_delay: Duration::from_secs(10),
+            jitter: false,
+        };
+        assert_eq!(b.delay_for_attempt(1), Duration::from_millis(100));
+    }
+
+    #[test]
+    fn backoff_delay_exponential() {
+        let b = BackoffPolicy {
+            initial_delay: Duration::from_millis(100),
+            factor: 2.0,
+            max_delay: Duration::from_secs(10),
+            jitter: false,
+        };
+        assert_eq!(b.delay_for_attempt(2), Duration::from_millis(200));
+        assert_eq!(b.delay_for_attempt(3), Duration::from_millis(400));
+        assert_eq!(b.delay_for_attempt(4), Duration::from_millis(800));
+    }
+
+    #[test]
+    fn backoff_delay_capped_at_max() {
+        let b = BackoffPolicy {
+            initial_delay: Duration::from_millis(100),
+            factor: 2.0,
+            max_delay: Duration::from_millis(300),
+            jitter: false,
+        };
+        assert_eq!(b.delay_for_attempt(1), Duration::from_millis(100));
+        assert_eq!(b.delay_for_attempt(2), Duration::from_millis(200));
+        assert_eq!(b.delay_for_attempt(3), Duration::from_millis(300)); // capped
+        assert_eq!(b.delay_for_attempt(4), Duration::from_millis(300)); // still capped
+    }
+
+    #[test]
+    fn retry_policy_none_is_single_attempt() {
+        let p = RetryPolicy::none();
+        assert_eq!(p.max_attempts, 1);
+    }
+}

lib/crates/fabro-core/src/stall.rs

@@ -0,0 +1,206 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::Notify;
+
+/// Trait for receiving stall timeout notifications.
+pub trait ActivityMonitor: Send + Sync {
+    /// Called when a stall timeout fires. The implementation should signal
+    /// cancellation (e.g., set a cancel token).
+    fn on_stall_timeout(&self, elapsed: Duration);
+}
+
+/// Watches for inactivity and fires a stall timeout if no activity is
+/// reported within the configured duration.
+pub struct StallWatchdog {
+    timeout: Duration,
+    cancel_token: Arc<AtomicBool>,
+    activity: Arc<Notify>,
+    shutdown: Arc<AtomicBool>,
+    monitor: Arc<dyn ActivityMonitor>,
+}
+
+/// Guard that resets the stall timer on activity. Drop to stop watching.
+pub struct StallGuard {
+    activity: Arc<Notify>,
+    shutdown: Arc<AtomicBool>,
+    handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl StallWatchdog {
+    pub fn new(
+        timeout: Duration,
+        cancel_token: Arc<AtomicBool>,
+        monitor: Arc<dyn ActivityMonitor>,
+    ) -> Self {
+        Self {
+            timeout,
+            cancel_token,
+            activity: Arc::new(Notify::new()),
+            shutdown: Arc::new(AtomicBool::new(false)),
+            monitor,
+        }
+    }
+
+    /// Start watching. Returns a StallGuard — call `guard.report_activity()`
+    /// to reset the timer. Drop the guard to stop the watchdog.
+    pub fn start(self) -> StallGuard {
+        let activity = self.activity.clone();
+        let shutdown = self.shutdown.clone();
+        let timeout = self.timeout;
+        let cancel_token = self.cancel_token;
+        let monitor = self.monitor;
+
+        let handle = tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    _ = tokio::time::sleep(timeout) => {
+                        if shutdown.load(Ordering::Relaxed) {
+                            return;
+                        }
+                        tracing::info!(
+                            timeout_secs = timeout.as_secs(),
+                            "Stall timeout: no activity detected"
+                        );
+                        monitor.on_stall_timeout(timeout);
+                        cancel_token.store(true, Ordering::Relaxed);
+                        return;
+                    }
+                    _ = activity.notified() => {
+                        if shutdown.load(Ordering::Relaxed) {
+                            return;
+                        }
+                        // Activity reported, restart the timer
+                        continue;
+                    }
+                }
+            }
+        });
+
+        StallGuard {
+            activity: self.activity,
+            shutdown: self.shutdown,
+            handle: Some(handle),
+        }
+    }
+}
+
+impl StallGuard {
+    /// Report activity to reset the stall timer.
+    pub fn report_activity(&self) {
+        self.activity.notify_one();
+    }
+}
+
+impl Drop for StallGuard {
+    fn drop(&mut self) {
+        self.shutdown.store(true, Ordering::Relaxed);
+        self.activity.notify_one(); // wake the task so it can exit
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::AtomicU32;
+
+    struct TestMonitor {
+        stall_count: AtomicU32,
+    }
+
+    impl TestMonitor {
+        fn new() -> Arc<Self> {
+            Arc::new(Self {
+                stall_count: AtomicU32::new(0),
+            })
+        }
+
+        fn stalls(&self) -> u32 {
+            self.stall_count.load(Ordering::Relaxed)
+        }
+    }
+
+    impl ActivityMonitor for TestMonitor {
+        fn on_stall_timeout(&self, _elapsed: Duration) {
+            self.stall_count.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn stall_watchdog_cancels_on_inactivity() {
+        let cancel = Arc::new(AtomicBool::new(false));
+        let monitor = TestMonitor::new();
+        let watchdog =
+            StallWatchdog::new(Duration::from_millis(50), cancel.clone(), monitor.clone());
+        let _guard = watchdog.start();
+
+        // Wait for timeout to fire
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        assert!(cancel.load(Ordering::Relaxed));
+        assert_eq!(monitor.stalls(), 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn stall_watchdog_resets_on_activity() {
+        let cancel = Arc::new(AtomicBool::new(false));
+        let monitor = TestMonitor::new();
+        let watchdog =
+            StallWatchdog::new(Duration::from_millis(80), cancel.clone(), monitor.clone());
+        let guard = watchdog.start();
+
+        // Report activity before timeout
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        guard.report_activity();
+
+        // After another 50ms (100ms total, but only 50ms since activity), should not have timed out
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        assert!(!cancel.load(Ordering::Relaxed));
+
+        // Wait long enough for timeout after last activity (80ms + margin)
+        tokio::time::sleep(Duration::from_millis(60)).await;
+        assert!(cancel.load(Ordering::Relaxed));
+        assert_eq!(monitor.stalls(), 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn stall_watchdog_clean_shutdown_on_success() {
+        let cancel = Arc::new(AtomicBool::new(false));
+        let monitor = TestMonitor::new();
+        let watchdog =
+            StallWatchdog::new(Duration::from_millis(50), cancel.clone(), monitor.clone());
+        let guard = watchdog.start();
+
+        // Drop the guard before timeout
+        drop(guard);
+
+        // Wait past timeout
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        // Should NOT have triggered
+        assert!(!cancel.load(Ordering::Relaxed));
+        assert_eq!(monitor.stalls(), 0);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn stall_guard_cleanup_on_drop() {
+        let cancel = Arc::new(AtomicBool::new(false));
+        let monitor = TestMonitor::new();
+        let watchdog =
+            StallWatchdog::new(Duration::from_millis(50), cancel.clone(), monitor.clone());
+        let guard = watchdog.start();
+
+        // Drop guard — should abort the background task
+        drop(guard);
+
+        // Wait well past timeout
+        tokio::time::sleep(Duration::from_millis(150)).await;
+
+        // Cancel should not be set
+        assert!(!cancel.load(Ordering::Relaxed));
+    }
+}

lib/crates/fabro-core/src/state.rs

@@ -0,0 +1,174 @@
+use std::collections::HashMap;
+
+use crate::context::Context;
+use crate::error::Result;
+use crate::graph::{Graph, NodeSpec};
+use crate::outcome::{NodeResult, Outcome};
+
+pub struct RunState {
+    pub context: Context,
+    pub current_node_id: String,
+    pub completed_nodes: Vec<String>,
+    pub node_outcomes: HashMap<String, Outcome>,
+    pub node_retries: HashMap<String, u32>,
+    pub node_visits: HashMap<String, usize>,
+    pub stage_index: usize,
+    pub previous_node_id: Option<String>,
+}
+
+impl RunState {
+    pub fn new<G: Graph>(graph: &G) -> Result<Self> {
+        let start = graph.find_start_node()?;
+        Ok(Self {
+            context: Context::new(),
+            current_node_id: start.id().to_string(),
+            completed_nodes: Vec::new(),
+            node_outcomes: HashMap::new(),
+            node_retries: HashMap::new(),
+            node_visits: HashMap::new(),
+            stage_index: 0,
+            previous_node_id: None,
+        })
+    }
+
+    pub fn record(&mut self, node_id: &str, result: &NodeResult) {
+        self.completed_nodes.push(node_id.to_string());
+        self.node_outcomes
+            .insert(node_id.to_string(), result.outcome.clone());
+        if result.attempts > 1 {
+            self.node_retries
+                .insert(node_id.to_string(), result.attempts - 1);
+        }
+        self.stage_index += 1;
+        self.context.apply_updates(&result.outcome.context_updates);
+    }
+
+    pub fn advance(&mut self, next_node_id: &str) {
+        self.previous_node_id = Some(self.current_node_id.clone());
+        self.current_node_id = next_node_id.to_string();
+    }
+
+    pub fn restart(&mut self, start_node_id: &str) {
+        self.current_node_id = start_node_id.to_string();
+        self.completed_nodes.clear();
+        self.node_outcomes.clear();
+        self.node_retries.clear();
+        self.stage_index = 0;
+        self.previous_node_id = None;
+        // node_visits is NOT cleared — preserves total visit counts across restarts
+    }
+
+    pub fn current_node<G: Graph>(&self, graph: &G) -> Option<G::Node> {
+        graph.get_node(&self.current_node_id)
+    }
+
+    pub fn increment_visits(&mut self, node_id: &str) -> usize {
+        let count = self.node_visits.entry(node_id.to_string()).or_insert(0);
+        *count += 1;
+        *count
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use serde_json::json;
+
+    use super::*;
+    use crate::outcome::{Outcome, StageStatus};
+    use crate::test_fixtures::linear_graph;
+
+    #[test]
+    fn run_state_new_from_graph() {
+        let g = linear_graph(&["start", "work", "end"]);
+        let state = RunState::new(&g).unwrap();
+        assert_eq!(state.current_node_id, "start");
+        assert!(state.completed_nodes.is_empty());
+        assert!(state.node_outcomes.is_empty());
+        assert_eq!(state.stage_index, 0);
+        assert!(state.previous_node_id.is_none());
+    }
+
+    #[test]
+    fn run_state_record_updates_all_fields() {
+        let g = linear_graph(&["start", "end"]);
+        let mut state = RunState::new(&g).unwrap();
+        let result = NodeResult::new(Outcome::success(), Duration::from_millis(50), 2, 3);
+        state.record("start", &result);
+
+        assert_eq!(state.completed_nodes, vec!["start"]);
+        assert_eq!(state.node_outcomes["start"].status, StageStatus::Success);
+        assert_eq!(state.node_retries["start"], 1); // 2 attempts - 1
+        assert_eq!(state.stage_index, 1);
+    }
+
+    #[test]
+    fn run_state_record_applies_context_updates() {
+        let g = linear_graph(&["start", "end"]);
+        let mut state = RunState::new(&g).unwrap();
+        let mut outcome = Outcome::success();
+        outcome.context_updates.insert("key".into(), json!("value"));
+        let result = NodeResult::new(outcome, Duration::ZERO, 1, 1);
+        state.record("start", &result);
+        assert_eq!(state.context.get("key"), Some(json!("value")));
+    }
+
+    #[test]
+    fn run_state_advance_updates_current_and_previous() {
+        let g = linear_graph(&["start", "mid", "end"]);
+        let mut state = RunState::new(&g).unwrap();
+        assert_eq!(state.current_node_id, "start");
+        assert!(state.previous_node_id.is_none());
+
+        state.advance("mid");
+        assert_eq!(state.current_node_id, "mid");
+        assert_eq!(state.previous_node_id.as_deref(), Some("start"));
+
+        state.advance("end");
+        assert_eq!(state.current_node_id, "end");
+        assert_eq!(state.previous_node_id.as_deref(), Some("mid"));
+    }
+
+    #[test]
+    fn run_state_restart_clears_progress_keeps_visits() {
+        let g = linear_graph(&["start", "work", "end"]);
+        let mut state = RunState::new(&g).unwrap();
+        state.increment_visits("start");
+        state.increment_visits("work");
+        state.record(
+            "start",
+            &NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1),
+        );
+        state.advance("work");
+
+        state.restart("start");
+
+        assert_eq!(state.current_node_id, "start");
+        assert!(state.completed_nodes.is_empty());
+        assert!(state.node_outcomes.is_empty());
+        assert!(state.node_retries.is_empty());
+        assert_eq!(state.stage_index, 0);
+        assert!(state.previous_node_id.is_none());
+        // visits preserved
+        assert_eq!(state.node_visits["start"], 1);
+        assert_eq!(state.node_visits["work"], 1);
+    }
+
+    #[test]
+    fn run_state_current_node_from_graph() {
+        let g = linear_graph(&["start", "end"]);
+        let state = RunState::new(&g).unwrap();
+        let node = state.current_node(&g).unwrap();
+        assert_eq!(node.id(), "start");
+    }
+
+    #[test]
+    fn run_state_increment_visits() {
+        let g = linear_graph(&["start", "end"]);
+        let mut state = RunState::new(&g).unwrap();
+        assert_eq!(state.increment_visits("start"), 1);
+        assert_eq!(state.increment_visits("start"), 2);
+        assert_eq!(state.increment_visits("other"), 1);
+    }
+}

lib/crates/fabro-core/src/test_fixtures.rs

@@ -0,0 +1,600 @@
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::Arc;
+
+use async_trait::async_trait;
+
+use crate::context::Context;
+use crate::error::{CoreError, HandlerErrorDetail, Result};
+use crate::graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec};
+use crate::handler::NodeHandler;
+use crate::outcome::{Outcome, StageStatus};
+use crate::retry::RetryPolicy;
+
+// ---- Test node ----
+
+#[derive(Debug, Clone)]
+pub struct TestNode {
+    pub id: String,
+    pub terminal: bool,
+    pub max_visits: Option<usize>,
+    pub goal_gate: Option<(String, StageStatus)>,
+}
+
+impl TestNode {
+    pub fn new(id: &str) -> Self {
+        Self {
+            id: id.to_string(),
+            terminal: false,
+            max_visits: None,
+            goal_gate: None,
+        }
+    }
+
+    pub fn terminal(id: &str) -> Self {
+        Self {
+            id: id.to_string(),
+            terminal: true,
+            max_visits: None,
+            goal_gate: None,
+        }
+    }
+
+    pub fn with_max_visits(mut self, max: usize) -> Self {
+        self.max_visits = Some(max);
+        self
+    }
+
+    pub fn with_goal_gate(mut self, node_id: &str, required_status: StageStatus) -> Self {
+        self.goal_gate = Some((node_id.to_string(), required_status));
+        self
+    }
+}
+
+impl NodeSpec for TestNode {
+    fn id(&self) -> &str {
+        &self.id
+    }
+
+    fn is_terminal(&self) -> bool {
+        self.terminal
+    }
+
+    fn max_visits(&self) -> Option<usize> {
+        self.max_visits
+    }
+}
+
+// ---- Test edge ----
+
+#[derive(Debug, Clone)]
+pub struct TestEdge {
+    pub from: String,
+    pub to: String,
+    pub label: Option<String>,
+    pub loop_restart: bool,
+}
+
+impl TestEdge {
+    pub fn new(from: &str, to: &str) -> Self {
+        Self {
+            from: from.to_string(),
+            to: to.to_string(),
+            label: None,
+            loop_restart: false,
+        }
+    }
+
+    pub fn with_label(mut self, label: &str) -> Self {
+        self.label = Some(label.to_string());
+        self
+    }
+
+    pub fn with_loop_restart(mut self) -> Self {
+        self.loop_restart = true;
+        self
+    }
+}
+
+impl EdgeSpec for TestEdge {
+    fn target(&self) -> &str {
+        &self.to
+    }
+
+    fn label(&self) -> Option<&str> {
+        self.label.as_deref()
+    }
+
+    fn is_loop_restart(&self) -> bool {
+        self.loop_restart
+    }
+}
+
+// ---- Test graph ----
+
+#[derive(Debug, Clone)]
+pub struct TestGraph {
+    pub nodes: Vec<TestNode>,
+    pub edges: Vec<TestEdge>,
+    pub start_node_id: String,
+    pub retry_targets: HashMap<String, String>,
+}
+
+impl TestGraph {
+    pub fn new(nodes: Vec<TestNode>, edges: Vec<TestEdge>, start: &str) -> Self {
+        Self {
+            nodes,
+            edges,
+            start_node_id: start.to_string(),
+            retry_targets: HashMap::new(),
+        }
+    }
+
+    pub fn with_retry_target(mut self, from: &str, to: &str) -> Self {
+        self.retry_targets.insert(from.to_string(), to.to_string());
+        self
+    }
+}
+
+impl Graph for TestGraph {
+    type Node = TestNode;
+    type Edge = TestEdge;
+
+    fn get_node(&self, id: &str) -> Option<Self::Node> {
+        self.nodes.iter().find(|n| n.id == id).cloned()
+    }
+
+    fn find_start_node(&self) -> Result<Self::Node> {
+        self.get_node(&self.start_node_id)
+            .ok_or(CoreError::NoStartNode)
+    }
+
+    fn outgoing_edges(&self, node_id: &str) -> Vec<Self::Edge> {
+        self.edges
+            .iter()
+            .filter(|e| e.from == node_id)
+            .cloned()
+            .collect()
+    }
+
+    fn select_edge(
+        &self,
+        node: &Self::Node,
+        outcome: &Outcome,
+        _context: &Context,
+    ) -> Option<EdgeSelection<Self>> {
+        let edges = self.outgoing_edges(node.id());
+        if edges.is_empty() {
+            return None;
+        }
+
+        // First: match by preferred_label
+        if let Some(ref label) = outcome.preferred_label {
+            if let Some(e) = edges
+                .iter()
+                .find(|e| e.label.as_deref() == Some(label.as_str()))
+            {
+                return Some(EdgeSelection {
+                    edge: e.clone(),
+                    reason: "preferred_label",
+                });
+            }
+        }
+
+        // Second: match by status label (e.g. "fail", "success")
+        let status_label = outcome.status.to_string();
+        if let Some(e) = edges
+            .iter()
+            .find(|e| e.label.as_deref() == Some(status_label.as_str()))
+        {
+            return Some(EdgeSelection {
+                edge: e.clone(),
+                reason: "condition",
+            });
+        }
+
+        // Third: match by suggested_next_ids
+        for suggested in &outcome.suggested_next_ids {
+            if let Some(e) = edges.iter().find(|e| e.to == *suggested) {
+                return Some(EdgeSelection {
+                    edge: e.clone(),
+                    reason: "suggested_next",
+                });
+            }
+        }
+
+        // Fourth: unconditional (no label)
+        if let Some(e) = edges.iter().find(|e| e.label.is_none()) {
+            return Some(EdgeSelection {
+                edge: e.clone(),
+                reason: "unconditional",
+            });
+        }
+
+        None
+    }
+
+    fn check_goal_gates(
+        &self,
+        outcomes: &HashMap<String, Outcome>,
+    ) -> std::result::Result<(), String> {
+        for node in &self.nodes {
+            if let Some((ref required_node, ref required_status)) = node.goal_gate {
+                if node.is_terminal() {
+                    match outcomes.get(required_node) {
+                        Some(o) if o.status == *required_status => {}
+                        Some(o) => {
+                            return Err(format!(
+                                "goal gate failed: {} requires {} to be {:?} but was {:?}",
+                                node.id, required_node, required_status, o.status
+                            ));
+                        }
+                        None => {
+                            return Err(format!(
+                                "goal gate failed: {} requires {} but it was not completed",
+                                node.id, required_node
+                            ));
+                        }
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
+    fn get_retry_target(&self, failed_node_id: &str) -> Option<String> {
+        self.retry_targets.get(failed_node_id).cloned()
+    }
+}
+
+// ---- Test handlers ----
+
+pub struct AlwaysSucceedHandler;
+
+#[async_trait]
+impl NodeHandler<TestGraph> for AlwaysSucceedHandler {
+    async fn execute(
+        &self,
+        _node: &TestNode,
+        _context: &Context,
+        _graph: &TestGraph,
+    ) -> Result<Outcome> {
+        Ok(Outcome::success())
+    }
+}
+
+pub struct AlwaysFailHandler {
+    pub message: String,
+}
+
+impl AlwaysFailHandler {
+    pub fn new(message: &str) -> Self {
+        Self {
+            message: message.to_string(),
+        }
+    }
+}
+
+#[async_trait]
+impl NodeHandler<TestGraph> for AlwaysFailHandler {
+    async fn execute(
+        &self,
+        _node: &TestNode,
+        _context: &Context,
+        _graph: &TestGraph,
+    ) -> Result<Outcome> {
+        Ok(Outcome::fail(&self.message))
+    }
+}
+
+pub struct CountingHandler {
+    pub call_count: AtomicU32,
+    pub outcomes: std::sync::Mutex<Vec<std::result::Result<Outcome, CoreError>>>,
+    pub retry_policy: RetryPolicy,
+}
+
+impl CountingHandler {
+    pub fn new(outcomes: Vec<std::result::Result<Outcome, CoreError>>) -> Self {
+        Self {
+            call_count: AtomicU32::new(0),
+            outcomes: std::sync::Mutex::new(outcomes),
+            retry_policy: RetryPolicy::none(),
+        }
+    }
+
+    pub fn with_retry_policy(mut self, policy: RetryPolicy) -> Self {
+        self.retry_policy = policy;
+        self
+    }
+
+    pub fn calls(&self) -> u32 {
+        self.call_count.load(Ordering::Relaxed)
+    }
+}
+
+#[async_trait]
+impl NodeHandler<TestGraph> for CountingHandler {
+    async fn execute(
+        &self,
+        _node: &TestNode,
+        _context: &Context,
+        _graph: &TestGraph,
+    ) -> Result<Outcome> {
+        let count = self.call_count.fetch_add(1, Ordering::Relaxed);
+        let mut outcomes = self.outcomes.lock().unwrap();
+        if (count as usize) < outcomes.len() {
+            outcomes.remove(0)
+        } else {
+            Ok(Outcome::success())
+        }
+    }
+
+    fn retry_policy(&self, _node: &TestNode, _graph: &TestGraph) -> RetryPolicy {
+        self.retry_policy.clone()
+    }
+}
+
+/// A handler that dispatches based on node ID.
+pub struct DispatchHandler {
+    handlers: HashMap<String, Arc<dyn NodeHandler<TestGraph>>>,
+    default: Arc<dyn NodeHandler<TestGraph>>,
+}
+
+impl DispatchHandler {
+    pub fn new(default: Arc<dyn NodeHandler<TestGraph>>) -> Self {
+        Self {
+            handlers: HashMap::new(),
+            default,
+        }
+    }
+
+    pub fn with_handler(mut self, node_id: &str, handler: Arc<dyn NodeHandler<TestGraph>>) -> Self {
+        self.handlers.insert(node_id.to_string(), handler);
+        self
+    }
+}
+
+#[async_trait]
+impl NodeHandler<TestGraph> for DispatchHandler {
+    async fn execute(
+        &self,
+        node: &TestNode,
+        context: &Context,
+        graph: &TestGraph,
+    ) -> Result<Outcome> {
+        let handler = self.handlers.get(node.id()).unwrap_or(&self.default);
+        handler.execute(node, context, graph).await
+    }
+
+    fn retry_policy(&self, node: &TestNode, graph: &TestGraph) -> RetryPolicy {
+        let handler = self.handlers.get(node.id()).unwrap_or(&self.default);
+        handler.retry_policy(node, graph)
+    }
+
+    fn on_retries_exhausted(&self, node: &TestNode, last_outcome: Outcome) -> Outcome {
+        let handler = self.handlers.get(node.id()).unwrap_or(&self.default);
+        handler.on_retries_exhausted(node, last_outcome)
+    }
+}
+
+/// A handler that returns Err(CoreError::Handler) with configurable retryability.
+pub struct ErrorHandler {
+    pub detail: HandlerErrorDetail,
+    pub retry_policy: RetryPolicy,
+}
+
+impl ErrorHandler {
+    pub fn retryable(message: &str, policy: RetryPolicy) -> Self {
+        Self {
+            detail: HandlerErrorDetail {
+                message: message.to_string(),
+                retryable: true,
+                category: None,
+                signature: None,
+            },
+            retry_policy: policy,
+        }
+    }
+
+    pub fn non_retryable(message: &str) -> Self {
+        Self {
+            detail: HandlerErrorDetail {
+                message: message.to_string(),
+                retryable: false,
+                category: None,
+                signature: None,
+            },
+            retry_policy: RetryPolicy::none(),
+        }
+    }
+}
+
+#[async_trait]
+impl NodeHandler<TestGraph> for ErrorHandler {
+    async fn execute(
+        &self,
+        _node: &TestNode,
+        _context: &Context,
+        _graph: &TestGraph,
+    ) -> Result<Outcome> {
+        Err(CoreError::handler(self.detail.clone()))
+    }
+
+    fn retry_policy(&self, _node: &TestNode, _graph: &TestGraph) -> RetryPolicy {
+        self.retry_policy.clone()
+    }
+}
+
+// ---- Helper for building common graphs ----
+
+/// Build a linear graph: start → a → b → ... → end
+pub fn linear_graph(node_ids: &[&str]) -> TestGraph {
+    assert!(node_ids.len() >= 2, "need at least start and end nodes");
+    let mut nodes = Vec::new();
+    let mut edges = Vec::new();
+
+    for (i, id) in node_ids.iter().enumerate() {
+        if i == node_ids.len() - 1 {
+            nodes.push(TestNode::terminal(id));
+        } else {
+            nodes.push(TestNode::new(id));
+            edges.push(TestEdge::new(id, node_ids[i + 1]));
+        }
+    }
+
+    TestGraph::new(nodes, edges, node_ids[0])
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_graph_finds_start_node() {
+        let g = linear_graph(&["start", "end"]);
+        let start = g.find_start_node().unwrap();
+        assert_eq!(start.id(), "start");
+    }
+
+    #[test]
+    fn test_graph_gets_node_by_id() {
+        let g = linear_graph(&["start", "work", "end"]);
+        let node = g.get_node("work").unwrap();
+        assert_eq!(node.id(), "work");
+        assert!(!node.is_terminal());
+    }
+
+    #[test]
+    fn test_graph_returns_none_for_missing() {
+        let g = linear_graph(&["start", "end"]);
+        assert!(g.get_node("nonexistent").is_none());
+    }
+
+    #[test]
+    fn test_graph_outgoing_edges() {
+        let g = linear_graph(&["start", "mid", "end"]);
+        let edges = g.outgoing_edges("start");
+        assert_eq!(edges.len(), 1);
+        assert_eq!(edges[0].target(), "mid");
+    }
+
+    #[test]
+    fn test_graph_terminal_detection() {
+        let g = linear_graph(&["start", "end"]);
+        assert!(!g.get_node("start").unwrap().is_terminal());
+        assert!(g.get_node("end").unwrap().is_terminal());
+    }
+
+    #[test]
+    fn test_graph_edge_selection_by_label() {
+        let g = TestGraph::new(
+            vec![
+                TestNode::new("start"),
+                TestNode::new("a"),
+                TestNode::new("b"),
+                TestNode::terminal("end"),
+            ],
+            vec![
+                TestEdge::new("start", "a").with_label("success"),
+                TestEdge::new("start", "b").with_label("fail"),
+            ],
+            "start",
+        );
+        let node = g.get_node("start").unwrap();
+        let outcome = Outcome::fail("oops");
+        let ctx = Context::new();
+        let sel = g.select_edge(&node, &outcome, &ctx).unwrap();
+        assert_eq!(sel.edge.target(), "b");
+        assert_eq!(sel.reason, "condition");
+    }
+
+    #[test]
+    fn test_graph_edge_selection_unconditional() {
+        let g = linear_graph(&["start", "end"]);
+        let node = g.get_node("start").unwrap();
+        let outcome = Outcome::success();
+        let ctx = Context::new();
+        let sel = g.select_edge(&node, &outcome, &ctx).unwrap();
+        assert_eq!(sel.edge.target(), "end");
+        assert_eq!(sel.reason, "unconditional");
+    }
+
+    #[test]
+    fn test_graph_goal_gates_pass() {
+        let g = TestGraph::new(
+            vec![
+                TestNode::new("work"),
+                TestNode::terminal("end").with_goal_gate("work", StageStatus::Success),
+            ],
+            vec![TestEdge::new("work", "end")],
+            "work",
+        );
+        let mut outcomes = HashMap::new();
+        outcomes.insert("work".to_string(), Outcome::success());
+        assert!(g.check_goal_gates(&outcomes).is_ok());
+    }
+
+    #[test]
+    fn test_graph_goal_gates_fail() {
+        let g = TestGraph::new(
+            vec![
+                TestNode::new("work"),
+                TestNode::terminal("end").with_goal_gate("work", StageStatus::Success),
+            ],
+            vec![TestEdge::new("work", "end")],
+            "work",
+        );
+        let mut outcomes = HashMap::new();
+        outcomes.insert("work".to_string(), Outcome::fail("oops"));
+        assert!(g.check_goal_gates(&outcomes).is_err());
+    }
+
+    #[test]
+    fn test_graph_retry_target() {
+        let g = linear_graph(&["start", "end"]).with_retry_target("start", "start");
+        assert_eq!(g.get_retry_target("start").as_deref(), Some("start"));
+        assert!(g.get_retry_target("end").is_none());
+    }
+
+    #[tokio::test]
+    async fn always_succeed_handler() {
+        let h = AlwaysSucceedHandler;
+        let g = linear_graph(&["start", "end"]);
+        let node = g.get_node("start").unwrap();
+        let ctx = Context::new();
+        let result = h.execute(&node, &ctx, &g).await.unwrap();
+        assert_eq!(result.status, StageStatus::Success);
+    }
+
+    #[tokio::test]
+    async fn always_fail_handler() {
+        let h = AlwaysFailHandler::new("boom");
+        let g = linear_graph(&["start", "end"]);
+        let node = g.get_node("start").unwrap();
+        let ctx = Context::new();
+        let result = h.execute(&node, &ctx, &g).await.unwrap();
+        assert_eq!(result.status, StageStatus::Fail);
+        assert_eq!(result.failure.unwrap().message, "boom");
+    }
+
+    #[tokio::test]
+    async fn counting_handler_tracks_calls() {
+        let h = CountingHandler::new(vec![Ok(Outcome::fail("first")), Ok(Outcome::success())]);
+        let g = linear_graph(&["start", "end"]);
+        let node = g.get_node("start").unwrap();
+        let ctx = Context::new();
+
+        let r1 = h.execute(&node, &ctx, &g).await.unwrap();
+        assert_eq!(r1.status, StageStatus::Fail);
+        assert_eq!(h.calls(), 1);
+
+        let r2 = h.execute(&node, &ctx, &g).await.unwrap();
+        assert_eq!(r2.status, StageStatus::Success);
+        assert_eq!(h.calls(), 2);
+
+        // Past end of outcomes list → default success
+        let r3 = h.execute(&node, &ctx, &g).await.unwrap();
+        assert_eq!(r3.status, StageStatus::Success);
+        assert_eq!(h.calls(), 3);
+    }
+}