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

Unify Outcome types between fabro-core and fabro-workflows

Browse source 
Make fabro-core's Outcome generic over a usage/metadata type parameter
(OutcomeMeta trait), allowing fabro-workflows to use core's types
directly via a type alias instead of maintaining duplicate Outcome,
StageStatus, and FailureDetail types with bidirectional conversions.

Key changes:
- Add FailureCategory enum to fabro-core (moved from fabro-workflows'
  FailureClass), with Display/FromStr/is_signature_tracked
- Add OutcomeMeta supertrait + blanket impl for the generic parameter
- Make Outcome<M>, NodeResult<M>, RunState<M>, NodeDecision<M> generic
  with default type parameter M=()
- Add Graph::Meta associated type
- Update FailureDetail with serde renames (category→"failure_class",
  signature→"failure_signature") for checkpoint backward compat
- Replace fabro-workflows' Outcome with type alias to
  fabro_core::Outcome<Option<StageUsage>>
- Add OutcomeExt extension trait for wf-specific factory methods
  (fail_classify, fail_deterministic, retry_classify, simulated, etc.)
- Delete core_adapter/outcome.rs (~170 lines of conversion functions)
- Replace FailureClass with FailureCategory throughout fabro-workflows
- Fix timeout handler to use TransientInfra category, panic handler to
  use Deterministic category

Net: -144 lines, zero-cost type unification with no runtime conversions.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-24 09:24:36 -04:00
parent c442ca62e7
commit f7e2391535
31 changed files with 783 additions and 927 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/crates/fabro-cli/src/commands/run.rs
View file

@ -26,7 +26,7 @@ use fabro_workflows::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use fabro_workflows::git::GitSyncStatus;
use fabro_workflows::handler::default_registry;
use fabro_workflows::manifest::Manifest;
use fabro_workflows::outcome::{Outcome, StageStatus};
use fabro_workflows::outcome::{Outcome, OutcomeExt, StageStatus};
use fabro_workflows::run_status::{RunStatus, StatusReason};
use fabro_workflows::sandbox_provider::SandboxProvider;
use fabro_workflows::workflow::WorkflowBuilder;

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

@ -1,6 +1,6 @@
use std::fmt;
use crate::outcome::{FailureDetail, Outcome, StageStatus};
use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeMeta, StageStatus};
/// Structured failure data on handler errors. Maps to FabroError's
/// is_retryable(), failure_class(), failure_signature_hint(), to_fail_outcome().
@ -8,7 +8,7 @@ use crate::outcome::{FailureDetail, Outcome, StageStatus};
pub struct HandlerErrorDetail {
pub message: String,
pub retryable: bool,
pub category: Option<String>,
pub category: Option<FailureCategory>,
pub signature: Option<String>,
}
@ -57,13 +57,13 @@ impl CoreError {
matches!(self, Self::Handler { detail } if detail.retryable)
}
pub fn to_fail_outcome(&self) -> Outcome {
pub fn to_fail_outcome<M: OutcomeMeta>(&self) -> Outcome<M> {
match self {
Self::Handler { detail } => Outcome {
status: StageStatus::Fail,
failure: Some(FailureDetail {
message: detail.message.clone(),
category: detail.category.clone(),
category: detail.category.unwrap_or(FailureCategory::Deterministic),
signature: detail.signature.clone(),
}),
..Outcome::default()
@ -140,17 +140,18 @@ mod tests {
#[test]
fn core_error_handler_to_fail_outcome() {
use crate::outcome::FailureCategory;
let err = CoreError::handler(HandlerErrorDetail {
message: "api down".into(),
retryable: true,
category: Some("transient".into()),
category: Some(FailureCategory::TransientInfra),
signature: Some("sig123".into()),
});
let outcome = err.to_fail_outcome();
let outcome: crate::outcome::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.category, FailureCategory::TransientInfra);
assert_eq!(failure.signature.as_deref(), Some("sig123"));
}

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

@ -80,7 +80,7 @@ impl<G: Graph + 'static> ExecutorBuilder<G> {
}
impl<G: Graph + 'static> Executor<G> {
pub async fn run(&self, graph: &G, mut state: RunState) -> Result<Outcome> {
pub async fn run(&self, graph: &G, mut state: RunState<G::Meta>) -> Result<Outcome<G::Meta>> {
self.lifecycle.on_run_start(graph, &state).await?;
loop {
@ -229,9 +229,9 @@ impl<G: Graph + 'static> Executor<G> {
async fn execute_with_retry(
&self,
node: &G::Node,
state: &RunState,
state: &RunState<G::Meta>,
graph: &G,
) -> Result<NodeResult> {
) -> Result<NodeResult<G::Meta>> {
let policy = self.handler.retry_policy(node, graph);
let start = Instant::now();
@ -330,8 +330,8 @@ impl<G: Graph + 'static> Executor<G> {
async fn resolve_next_step(
&self,
node: &G::Node,
outcome: &Outcome,
state: &RunState,
outcome: &Outcome<G::Meta>,
state: &RunState<G::Meta>,
graph: &G,
) -> Result<NextStep> {
// Jump takes priority

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

@ -2,7 +2,7 @@ use std::collections::HashMap;
use crate::context::Context;
use crate::error::Result;
use crate::outcome::Outcome;
use crate::outcome::{Outcome, OutcomeMeta};
pub trait NodeSpec: Send + Sync + Clone {
fn id(&self) -> &str;
@ -24,6 +24,7 @@ pub struct EdgeSelection<G: Graph + ?Sized> {
pub trait Graph: Send + Sync {
type Node: NodeSpec + Clone;
type Edge: EdgeSpec + Clone;
type Meta: OutcomeMeta;
fn get_node(&self, id: &str) -> Option<Self::Node>;
fn find_start_node(&self) -> Result<Self::Node>;
@ -31,12 +32,12 @@ pub trait Graph: Send + Sync {
fn select_edge(
&self,
node: &Self::Node,
outcome: &Outcome,
outcome: &Outcome<Self::Meta>,
context: &Context,
) -> Option<EdgeSelection<Self>>;
fn check_goal_gates(
&self,
outcomes: &HashMap<String, Outcome>,
outcomes: &HashMap<String, Outcome<Self::Meta>>,
) -> std::result::Result<(), String>;
fn get_retry_target(&self, failed_node_id: &str) -> Option<String>;
}

13
lib/crates/fabro-core/src/handler.rs
View file

@ -8,13 +8,22 @@ 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>;
async fn execute(
&self,
node: &G::Node,
context: &Context,
graph: &G,
) -> Result<Outcome<G::Meta>>;
fn retry_policy(&self, _node: &G::Node, _graph: &G) -> RetryPolicy {
RetryPolicy::none()
}
fn on_retries_exhausted(&self, _node: &G::Node, _last_outcome: Outcome) -> Outcome {
fn on_retries_exhausted(
&self,
_node: &G::Node,
_last_outcome: Outcome<G::Meta>,
) -> Outcome<G::Meta> {
Outcome::fail("max retries exceeded")
}
}

2
lib/crates/fabro-core/src/lib.rs
View file

@ -21,7 +21,7 @@ pub use lifecycle::{
AttemptContext, AttemptResultContext, CompositeLifecycle, EdgeContext, EdgeDecision,
NodeDecision, NoopLifecycle, RunLifecycle,
};
pub use outcome::{FailureDetail, NodeResult, Outcome, StageStatus};
pub use outcome::{FailureCategory, FailureDetail, NodeResult, Outcome, OutcomeMeta, StageStatus};
pub use retry::{BackoffPolicy, RetryPolicy};
pub use stall::{ActivityMonitor, StallGuard, StallWatchdog};
pub use state::RunState;

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

@ -4,13 +4,13 @@ use async_trait::async_trait;
use crate::error::Result;
use crate::graph::Graph;
use crate::outcome::{NodeResult, Outcome};
use crate::outcome::{NodeResult, Outcome, OutcomeMeta};
use crate::state::RunState;
#[derive(Debug, Clone)]
pub enum NodeDecision {
pub enum NodeDecision<M: OutcomeMeta = ()> {
Continue,
Skip(Box<Outcome>),
Skip(Box<Outcome<M>>),
Block(String),
}
@ -29,7 +29,7 @@ pub struct AttemptContext<'a, G: Graph> {
pub struct AttemptResultContext<'a, G: Graph> {
pub node: &'a G::Node,
pub result: &'a NodeResult,
pub result: &'a NodeResult<G::Meta>,
pub attempt: u32,
pub will_retry: bool,
pub backoff_delay: Option<Duration>,
@ -40,13 +40,13 @@ pub struct EdgeContext<'a, G: Graph> {
pub to: &'a str,
pub edge: Option<G::Edge>,
pub is_jump: bool,
pub outcome: &'a Outcome,
pub outcome: &'a Outcome<G::Meta>,
pub reason: &'a str,
}
#[async_trait]
pub trait RunLifecycle<G: Graph>: Send + Sync {
async fn on_run_start(&self, _graph: &G, _state: &RunState) -> Result<()> {
async fn on_run_start(&self, _graph: &G, _state: &RunState<G::Meta>) -> Result<()> {
Ok(())
}
@ -54,26 +54,30 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
&self,
_node: &G::Node,
_goal_gates_passed: bool,
_state: &RunState,
_state: &RunState<G::Meta>,
) {
}
async fn before_node(&self, _node: &G::Node, _state: &RunState) -> Result<NodeDecision> {
async fn before_node(
&self,
_node: &G::Node,
_state: &RunState<G::Meta>,
) -> Result<NodeDecision<G::Meta>> {
Ok(NodeDecision::Continue)
}
async fn before_attempt(
&self,
_ctx: &AttemptContext<'_, G>,
_state: &RunState,
) -> Result<NodeDecision> {
_state: &RunState<G::Meta>,
) -> Result<NodeDecision<G::Meta>> {
Ok(NodeDecision::Continue)
}
async fn after_attempt(
&self,
_ctx: &AttemptResultContext<'_, G>,
_state: &RunState,
_state: &RunState<G::Meta>,
) -> Result<()> {
Ok(())
}
@ -81,8 +85,8 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
async fn after_node(
&self,
_node: &G::Node,
_result: &mut NodeResult,
_state: &RunState,
_result: &mut NodeResult<G::Meta>,
_state: &RunState<G::Meta>,
) -> Result<()> {
Ok(())
}
@ -90,7 +94,7 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
async fn on_edge_selected(
&self,
_ctx: &EdgeContext<'_, G>,
_state: &RunState,
_state: &RunState<G::Meta>,
) -> Result<EdgeDecision> {
Ok(EdgeDecision::Continue)
}
@ -98,14 +102,14 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
async fn on_checkpoint(
&self,
_node: &G::Node,
_result: &NodeResult,
_result: &NodeResult<G::Meta>,
_next_node_id: Option<&str>,
_state: &RunState,
_state: &RunState<G::Meta>,
) -> Result<()> {
Ok(())
}
async fn on_run_end(&self, _outcome: &Outcome, _state: &RunState) {}
async fn on_run_end(&self, _outcome: &Outcome<G::Meta>, _state: &RunState<G::Meta>) {}
}
/// No-op lifecycle that passes through everything.
@ -128,14 +132,19 @@ impl<G: Graph> CompositeLifecycle<G> {
#[async_trait]
impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
async fn on_run_start(&self, graph: &G, state: &RunState) -> Result<()> {
async fn on_run_start(&self, graph: &G, state: &RunState<G::Meta>) -> Result<()> {
for child in &self.children {
child.on_run_start(graph, state).await?;
}
Ok(())
}
async fn on_terminal_reached(&self, node: &G::Node, goal_gates_passed: bool, state: &RunState) {
async fn on_terminal_reached(
&self,
node: &G::Node,
goal_gates_passed: bool,
state: &RunState<G::Meta>,
) {
for child in &self.children {
child
.on_terminal_reached(node, goal_gates_passed, state)
@ -143,7 +152,11 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
}
}
async fn before_node(&self, node: &G::Node, state: &RunState) -> Result<NodeDecision> {
async fn before_node(
&self,
node: &G::Node,
state: &RunState<G::Meta>,
) -> Result<NodeDecision<G::Meta>> {
for child in &self.children {
match child.before_node(node, state).await? {
NodeDecision::Continue => {}
@ -156,8 +169,8 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
async fn before_attempt(
&self,
ctx: &AttemptContext<'_, G>,
state: &RunState,
) -> Result<NodeDecision> {
state: &RunState<G::Meta>,
) -> Result<NodeDecision<G::Meta>> {
for child in &self.children {
match child.before_attempt(ctx, state).await? {
NodeDecision::Continue => {}
@ -170,7 +183,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
async fn after_attempt(
&self,
ctx: &AttemptResultContext<'_, G>,
state: &RunState,
state: &RunState<G::Meta>,
) -> Result<()> {
for child in &self.children {
child.after_attempt(ctx, state).await?;
@ -181,8 +194,8 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
async fn after_node(
&self,
node: &G::Node,
result: &mut NodeResult,
state: &RunState,
result: &mut NodeResult<G::Meta>,
state: &RunState<G::Meta>,
) -> Result<()> {
for child in &self.children {
child.after_node(node, result, state).await?;
@ -193,7 +206,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
async fn on_edge_selected(
&self,
ctx: &EdgeContext<'_, G>,
state: &RunState,
state: &RunState<G::Meta>,
) -> Result<EdgeDecision> {
for child in &self.children {
match child.on_edge_selected(ctx, state).await? {
@ -207,9 +220,9 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
async fn on_checkpoint(
&self,
node: &G::Node,
result: &NodeResult,
result: &NodeResult<G::Meta>,
next_node_id: Option<&str>,
state: &RunState,
state: &RunState<G::Meta>,
) -> Result<()> {
for child in &self.children {
child
@ -219,7 +232,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
Ok(())
}
async fn on_run_end(&self, outcome: &Outcome, state: &RunState) {
async fn on_run_end(&self, outcome: &Outcome<G::Meta>, state: &RunState<G::Meta>) {
for child in &self.children {
child.on_run_end(outcome, state).await;
}

301
lib/crates/fabro-core/src/outcome.rs
View file

@ -3,9 +3,21 @@ use std::fmt;
use std::str::FromStr;
use std::time::Duration;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::Value;
/// Supertrait for the generic usage/metadata type parameter on `Outcome`.
pub trait OutcomeMeta:
Default + Clone + Send + Sync + fmt::Debug + Serialize + DeserializeOwned + 'static
{
}
impl<T> OutcomeMeta for T where
T: Default + Clone + Send + Sync + fmt::Debug + Serialize + DeserializeOwned + 'static
{
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StageStatus {
@ -43,26 +55,149 @@ impl FromStr for StageStatus {
}
}
#[derive(Debug, Clone)]
/// Classification of failure modes.
///
/// Pipeline authors can write edge conditions like `context.failure_class=budget_exhausted`
/// to route execution based on the nature of the failure.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FailureCategory {
/// Temporary infrastructure failure (rate limit, timeout, network, 5xx).
TransientInfra,
/// Permanent failure (auth, bad config, code bug).
Deterministic,
/// Context length, token/turn limit, quota exceeded.
BudgetExhausted,
/// Reserved for future loop detection.
CompilationLoop,
/// User/system cancellation.
Canceled,
/// Reserved for future scope enforcement.
Structural,
}
impl fmt::Display for FailureCategory {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let s = match self {
Self::TransientInfra => "transient_infra",
Self::Deterministic => "deterministic",
Self::BudgetExhausted => "budget_exhausted",
Self::CompilationLoop => "compilation_loop",
Self::Canceled => "canceled",
Self::Structural => "structural",
};
write!(f, "{s}")
}
}
impl FromStr for FailureCategory {
type Err = std::convert::Infallible;
fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
let normalized = s.trim().to_lowercase();
Ok(match normalized.as_str() {
// Canonical names
"transient_infra" => Self::TransientInfra,
"deterministic" => Self::Deterministic,
"budget_exhausted" => Self::BudgetExhausted,
"compilation_loop" => Self::CompilationLoop,
"canceled" => Self::Canceled,
"structural" => Self::Structural,
// Aliases: transient_infra
"transient"
| "transient-infra"
| "infra_transient"
| "transient infra"
| "infrastructure_transient"
| "retryable"
| "toolchain_workspace_io"
| "toolchain-workspace-io"
| "toolchain_or_dependency_registry_unavailable"
| "toolchain-dependency-registry-unavailable" => Self::TransientInfra,
// Aliases: deterministic
"non_transient" | "non-transient" | "permanent" | "logic" | "product" => {
Self::Deterministic
}
// Aliases: canceled
"cancelled" => Self::Canceled,
// Aliases: budget_exhausted
"budget-exhausted" | "budget exhausted" | "budget" => Self::BudgetExhausted,
// Aliases: compilation_loop
"compilation-loop" | "compilation loop" | "compile_loop" | "compile-loop" => {
Self::CompilationLoop
}
// Aliases: structural
"structure" | "scope_violation" | "write_scope_violation" => Self::Structural,
// Unknown → fail-closed to Deterministic
_ => Self::Deterministic,
})
}
}
impl FailureCategory {
/// Whether this failure category should be tracked by the cycle breaker.
pub fn is_signature_tracked(self) -> bool {
matches!(self, Self::Deterministic | Self::Structural)
}
}
/// Structured failure information.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FailureDetail {
pub message: String,
pub category: Option<String>,
#[serde(rename = "failure_class")]
pub category: FailureCategory,
#[serde(
rename = "failure_signature",
default,
skip_serializing_if = "Option::is_none"
)]
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 FailureDetail {
pub fn new(message: impl Into<String>, category: FailureCategory) -> Self {
Self {
message: message.into(),
category,
signature: None,
}
}
}
impl Default for Outcome {
/// The result of executing a node handler.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(bound = "M: OutcomeMeta")]
pub struct Outcome<M: OutcomeMeta = ()> {
pub status: StageStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub preferred_label: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub suggested_next_ids: Vec<String>,
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub context_updates: HashMap<String, Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub jump_to_node: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub notes: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub failure: Option<FailureDetail>,
#[serde(default)]
pub usage: M,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub files_touched: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub duration_ms: Option<u64>,
}
impl<M: OutcomeMeta> Default for Outcome<M> {
fn default() -> Self {
Self {
status: StageStatus::Success,
@ -72,12 +207,14 @@ impl Default for Outcome {
jump_to_node: None,
notes: None,
failure: None,
metadata: HashMap::new(),
usage: M::default(),
files_touched: Vec::new(),
duration_ms: None,
}
}
}
impl Outcome {
impl<M: OutcomeMeta> Outcome<M> {
pub fn success() -> Self {
Self::default()
}
@ -87,7 +224,7 @@ impl Outcome {
status: StageStatus::Fail,
failure: Some(FailureDetail {
message: message.to_string(),
category: None,
category: FailureCategory::Deterministic,
signature: None,
}),
..Self::default()
@ -104,15 +241,15 @@ impl Outcome {
}
#[derive(Debug, Clone)]
pub struct NodeResult {
pub outcome: Outcome,
pub struct NodeResult<M: OutcomeMeta = ()> {
pub outcome: Outcome<M>,
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 {
impl<M: OutcomeMeta> NodeResult<M> {
pub fn new(outcome: Outcome<M>, duration: Duration, attempts: u32, max_attempts: u32) -> Self {
Self {
outcome,
duration,
@ -135,7 +272,7 @@ impl NodeResult {
}
}
pub fn from_skip(outcome: Outcome) -> Self {
pub fn from_skip(outcome: Outcome<M>) -> Self {
Self {
outcome,
duration: Duration::ZERO,
@ -181,9 +318,55 @@ mod tests {
}
}
#[test]
fn failure_category_display_roundtrip() {
let categories = [
FailureCategory::TransientInfra,
FailureCategory::Deterministic,
FailureCategory::BudgetExhausted,
FailureCategory::CompilationLoop,
FailureCategory::Canceled,
FailureCategory::Structural,
];
for cat in &categories {
let s = cat.to_string();
let parsed: FailureCategory = s.parse().unwrap();
assert_eq!(&parsed, cat);
}
}
#[test]
fn failure_category_aliases() {
assert_eq!(
"transient".parse::<FailureCategory>().unwrap(),
FailureCategory::TransientInfra
);
assert_eq!(
"cancelled".parse::<FailureCategory>().unwrap(),
FailureCategory::Canceled
);
assert_eq!(
"permanent".parse::<FailureCategory>().unwrap(),
FailureCategory::Deterministic
);
assert_eq!(
"budget".parse::<FailureCategory>().unwrap(),
FailureCategory::BudgetExhausted
);
}
#[test]
fn failure_category_is_signature_tracked() {
assert!(FailureCategory::Deterministic.is_signature_tracked());
assert!(FailureCategory::Structural.is_signature_tracked());
assert!(!FailureCategory::TransientInfra.is_signature_tracked());
assert!(!FailureCategory::BudgetExhausted.is_signature_tracked());
assert!(!FailureCategory::Canceled.is_signature_tracked());
}
#[test]
fn outcome_success_factory() {
let o = Outcome::success();
let o: Outcome = Outcome::success();
assert_eq!(o.status, StageStatus::Success);
assert!(o.failure.is_none());
assert!(o.notes.is_none());
@ -191,24 +374,24 @@ mod tests {
#[test]
fn outcome_fail_factory() {
let o = Outcome::fail("broken");
let o: Outcome = 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_eq!(f.category, FailureCategory::Deterministic);
assert!(f.signature.is_none());
}
#[test]
fn outcome_skipped_factory() {
let o = Outcome::skipped("not needed");
let o: Outcome = 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();
let mut o: Outcome = Outcome::success();
o.context_updates
.insert("key".into(), serde_json::json!("value"));
assert_eq!(o.context_updates["key"], serde_json::json!("value"));
@ -216,37 +399,71 @@ mod tests {
#[test]
fn outcome_with_jump() {
let mut o = Outcome::success();
let mut o: Outcome = 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);
let mut o: Outcome = Outcome::success();
o.notes = Some("done".to_string());
o.context_updates
.insert("key".into(), serde_json::json!("val"));
let json = serde_json::to_string(&o).unwrap();
let parsed: Outcome = serde_json::from_str(&json).unwrap();
assert_eq!(parsed.status, StageStatus::Success);
assert_eq!(parsed.notes.as_deref(), Some("done"));
assert_eq!(
parsed.context_updates.get("key"),
Some(&serde_json::json!("val"))
);
}
#[test]
fn outcome_deserialize_without_usage_key() {
// Old checkpoints may not have "usage" key — serde(default) handles this
let json = r#"{"status":"success"}"#;
let o: Outcome = serde_json::from_str(json).unwrap();
assert_eq!(o.status, StageStatus::Success);
}
#[test]
fn failure_detail_construction() {
let f = FailureDetail {
message: "timeout".into(),
category: Some("transient".into()),
signature: Some("sig".into()),
};
let f = FailureDetail::new("timeout", FailureCategory::TransientInfra);
assert_eq!(f.message, "timeout");
assert_eq!(f.category.as_deref(), Some("transient"));
assert_eq!(f.signature.as_deref(), Some("sig"));
assert_eq!(f.category, FailureCategory::TransientInfra);
assert!(f.signature.is_none());
}
#[test]
fn failure_detail_serde_uses_renamed_keys() {
let f = FailureDetail::new("timeout", FailureCategory::TransientInfra);
let json = serde_json::to_string(&f).unwrap();
// category serializes as "failure_class"
assert!(json.contains("\"failure_class\""));
assert!(!json.contains("\"category\""));
// signature omitted when None
assert!(!json.contains("failure_signature"));
}
#[test]
fn failure_detail_serde_roundtrip() {
let f = FailureDetail {
message: "api down".into(),
category: FailureCategory::TransientInfra,
signature: Some("sig123".into()),
};
let json = serde_json::to_string(&f).unwrap();
let parsed: FailureDetail = serde_json::from_str(&json).unwrap();
assert_eq!(parsed.message, "api down");
assert_eq!(parsed.category, FailureCategory::TransientInfra);
assert_eq!(parsed.signature.as_deref(), Some("sig123"));
}
#[test]
fn node_result_from_outcome() {
let o = Outcome::success();
let o: Outcome = 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));

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

@ -3,13 +3,13 @@ use std::collections::HashMap;
use crate::context::Context;
use crate::error::Result;
use crate::graph::{Graph, NodeSpec};
use crate::outcome::{NodeResult, Outcome};
use crate::outcome::{NodeResult, Outcome, OutcomeMeta};
pub struct RunState {
pub struct RunState<M: OutcomeMeta = ()> {
pub context: Context,
pub current_node_id: String,
pub completed_nodes: Vec<String>,
pub node_outcomes: HashMap<String, Outcome>,
pub node_outcomes: HashMap<String, Outcome<M>>,
pub node_retries: HashMap<String, u32>,
pub node_visits: HashMap<String, usize>,
pub stage_index: usize,
@ -17,7 +17,7 @@ pub struct RunState {
pub cancelled: bool,
}
impl RunState {
impl<M: OutcomeMeta> RunState<M> {
pub fn new<G: Graph>(graph: &G) -> Result<Self> {
let start = graph.find_start_node()?;
Ok(Self {
@ -33,7 +33,7 @@ impl RunState {
})
}
pub fn record(&mut self, node_id: &str, result: &NodeResult) {
pub fn record(&mut self, node_id: &str, result: &NodeResult<M>) {
self.completed_nodes.push(node_id.to_string());
self.node_outcomes
.insert(node_id.to_string(), result.outcome.clone());
@ -87,7 +87,7 @@ mod tests {
#[test]
fn run_state_new_from_graph() {
let g = linear_graph(&["start", "work", "end"]);
let state = RunState::new(&g).unwrap();
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());
@ -98,7 +98,7 @@ mod tests {
#[test]
fn run_state_record_updates_all_fields() {
let g = linear_graph(&["start", "end"]);
let mut state = RunState::new(&g).unwrap();
let mut state = RunState::<()>::new(&g).unwrap();
let result = NodeResult::new(Outcome::success(), Duration::from_millis(50), 2, 3);
state.record("start", &result);
@ -111,7 +111,7 @@ mod tests {
#[test]
fn run_state_record_applies_context_updates() {
let g = linear_graph(&["start", "end"]);
let mut state = RunState::new(&g).unwrap();
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);
@ -122,7 +122,7 @@ mod tests {
#[test]
fn run_state_advance_updates_current_and_previous() {
let g = linear_graph(&["start", "mid", "end"]);
let mut state = RunState::new(&g).unwrap();
let mut state = RunState::<()>::new(&g).unwrap();
assert_eq!(state.current_node_id, "start");
assert!(state.previous_node_id.is_none());
@ -138,7 +138,7 @@ mod tests {
#[test]
fn run_state_restart_clears_progress_keeps_visits() {
let g = linear_graph(&["start", "work", "end"]);
let mut state = RunState::new(&g).unwrap();
let mut state = RunState::<()>::new(&g).unwrap();
state.increment_visits("start");
state.increment_visits("work");
state.record(
@ -163,7 +163,7 @@ mod tests {
#[test]
fn run_state_current_node_from_graph() {
let g = linear_graph(&["start", "end"]);
let state = RunState::new(&g).unwrap();
let state = RunState::<()>::new(&g).unwrap();
let node = state.current_node(&g).unwrap();
assert_eq!(node.id(), "start");
}
@ -171,7 +171,7 @@ mod tests {
#[test]
fn run_state_increment_visits() {
let g = linear_graph(&["start", "end"]);
let mut state = RunState::new(&g).unwrap();
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);
@ -180,7 +180,7 @@ mod tests {
#[test]
fn run_state_restart_with_new_context() {
let g = linear_graph(&["start", "end"]);
let mut state = RunState::new(&g).unwrap();
let mut state = RunState::<()>::new(&g).unwrap();
state.context.set("key", json!("old_value"));
state.increment_visits("start");
@ -199,7 +199,7 @@ mod tests {
#[test]
fn run_state_restart_without_context_preserves() {
let g = linear_graph(&["start", "end"]);
let mut state = RunState::new(&g).unwrap();
let mut state = RunState::<()>::new(&g).unwrap();
state.context.set("key", json!("value"));
state.restart("start", None);

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

@ -139,6 +139,7 @@ impl TestGraph {
impl Graph for TestGraph {
type Node = TestNode;
type Edge = TestEdge;
type Meta = ();
fn get_node(&self, id: &str) -> Option<Self::Node> {
self.nodes.iter().find(|n| n.id == id).cloned()

13
lib/crates/fabro-workflows/src/checkpoint.rs
View file

@ -207,7 +207,7 @@ mod tests {
#[test]
fn signature_maps_roundtrip() {
use crate::error::{FailureClass, FailureSignature};
use crate::error::{FailureCategory, FailureSignature};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("checkpoint.json");
@ -217,7 +217,7 @@ mod tests {
loop_sigs.insert(
FailureSignature::new(
"verify",
FailureClass::Deterministic,
FailureCategory::Deterministic,
None,
Some("test failed"),
),
@ -225,7 +225,12 @@ mod tests {
);
let mut restart_sigs = HashMap::new();
restart_sigs.insert(
FailureSignature::new("build", FailureClass::Structural, None, Some("scope error")),
FailureSignature::new(
"build",
FailureCategory::Structural,
None,
Some("scope error"),
),
1,
);
@ -247,7 +252,7 @@ mod tests {
assert_eq!(loaded.restart_failure_signatures.len(), 1);
let sig = FailureSignature::new(
"verify",
FailureClass::Deterministic,
FailureCategory::Deterministic,
None,
Some("test failed"),
);

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

@ -4,10 +4,10 @@ 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;
use crate::outcome::{Outcome, StageUsage};
// ---- WorkflowNode ----
@ -73,6 +73,7 @@ impl WorkflowGraph {
impl Graph for WorkflowGraph {
type Node = WorkflowNode;
type Edge = WorkflowEdge;
type Meta = Option<StageUsage>;
fn get_node(&self, id: &str) -> Option<Self::Node> {
self.0
@ -99,15 +100,14 @@ impl Graph for WorkflowGraph {
fn select_edge(
&self,
node: &Self::Node,
outcome: &CoreOutcome,
outcome: &Outcome,
_context: &CoreContext,
) -> Option<EdgeSelection<Self>> {
// Convert core outcome to workflow outcome for edge selection
let wf_outcome = super::outcome::core_to_wf_outcome(outcome);
// Outcome is now the wf type directly — no conversion needed
let wf_context = crate::context::Context::new();
let selection = engine::select_edge(
node.inner(),
&wf_outcome,
outcome,
&wf_context,
self.inner(),
node.inner().selection(),
@ -120,13 +120,9 @@ impl Graph for WorkflowGraph {
fn check_goal_gates(
&self,
outcomes: &HashMap<String, CoreOutcome>,
outcomes: &HashMap<String, Outcome>,
) -> 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)
engine::check_goal_gates(self.inner(), outcomes)
}
fn get_retry_target(&self, failed_node_id: &str) -> Option<String> {

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

@ -8,15 +8,14 @@ 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::outcome::FailureCategory;
use fabro_core::retry::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;
use crate::outcome::{Outcome, StageStatus};
/// Production node handler that bridges fabro-core's NodeHandler to the
/// existing fabro-workflows Handler trait via EngineServices.
@ -32,14 +31,10 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
node: &WorkflowNode,
_context: &CoreContext,
_graph: &WorkflowGraph,
) -> CoreResult<CoreOutcome> {
) -> CoreResult<Outcome> {
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");
@ -65,7 +60,7 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
return Err(CoreError::handler(HandlerErrorDetail {
message: format!("handler timed out after {}ms", duration.as_millis()),
retryable: true,
category: None,
category: Some(FailureCategory::TransientInfra),
signature: None,
}));
}
@ -75,14 +70,13 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
};
match timed_result {
Ok(Ok(wf_outcome)) => Ok(wf_to_core_outcome(&wf_outcome)),
Ok(Ok(wf_outcome)) => Ok(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()),
category: Some(fabro_err.failure_category()),
signature: fabro_err.failure_signature_hint(),
}))
}
@ -97,7 +91,7 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
Err(CoreError::handler(HandlerErrorDetail {
message: msg,
retryable: false,
category: None,
category: Some(FailureCategory::Deterministic),
signature: None,
}))
}
@ -114,17 +108,16 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
}
}
fn on_retries_exhausted(&self, node: &WorkflowNode, last_outcome: CoreOutcome) -> CoreOutcome {
fn on_retries_exhausted(&self, node: &WorkflowNode, last_outcome: Outcome) -> Outcome {
let gv_node = node.inner();
if gv_node.allow_partial() {
CoreOutcome {
status: fabro_core::outcome::StageStatus::PartialSuccess,
Outcome {
status: StageStatus::PartialSuccess,
..last_outcome
}
} else {
let status = wf_to_core_status(&WfStatus::Fail);
CoreOutcome {
status,
Outcome {
status: StageStatus::Fail,
..last_outcome
}
}
@ -155,8 +148,8 @@ mod tests {
_node: &WorkflowNode,
_context: &CoreContext,
_graph: &WorkflowGraph,
) -> CoreResult<CoreOutcome> {
Ok(CoreOutcome::success())
) -> CoreResult<Outcome> {
Ok(Outcome::success())
}
fn retry_policy(&self, _node: &WorkflowNode, _graph: &WorkflowGraph) -> CoreRetryPolicy {

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

@ -10,17 +10,16 @@ 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::outcome::NodeResult;
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::error::{FailureCategory, FailureSignature};
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::outcome::StageStatus as WfStatus;
use crate::outcome::{FailureDetail, Outcome, StageStatus, StageUsage};
use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner};
use fabro_sandbox::Sandbox;
@ -102,9 +101,13 @@ impl WorkflowLifecycle {
}
}
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
type WfNodeDecision = NodeDecision<Option<StageUsage>>;
#[async_trait]
impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &RunState) -> CoreResult<()> {
async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> {
// Clear incoming edge data (reset stale fidelity/thread from prior iteration)
*self.incoming_edge_data.lock().unwrap() = None;
@ -137,7 +140,7 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
&self,
node: &WorkflowNode,
goal_gates_passed: bool,
state: &RunState,
state: &WfRunState,
) {
if !goal_gates_passed {
return;
@ -171,13 +174,16 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
});
}
async fn before_node(&self, node: &WorkflowNode, state: &RunState) -> CoreResult<NodeDecision> {
async fn before_node(
&self,
node: &WorkflowNode,
state: &WfRunState,
) -> CoreResult<WfNodeDecision> {
// 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
@ -220,8 +226,8 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
async fn before_attempt(
&self,
ctx: &AttemptContext<'_, WorkflowGraph>,
state: &RunState,
) -> CoreResult<NodeDecision> {
state: &WfRunState,
) -> CoreResult<WfNodeDecision> {
let gv = ctx.node.inner();
let stage_index = state.stage_index;
@ -235,7 +241,7 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
match decision {
HookDecision::Skip { reason } => {
let msg = reason.unwrap_or_else(|| "skipped by hook".into());
return Ok(NodeDecision::Skip(Box::new(CoreOutcome::skipped(&msg))));
return Ok(NodeDecision::Skip(Box::new(Outcome::skipped(&msg))));
}
HookDecision::Block { reason } => {
let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into());
@ -261,11 +267,11 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
async fn after_attempt(
&self,
ctx: &AttemptResultContext<'_, WorkflowGraph>,
state: &RunState,
state: &WfRunState,
) -> CoreResult<()> {
if ctx.will_retry {
let gv = ctx.node.inner();
let wf_outcome = core_to_wf_outcome(&ctx.result.outcome);
let outcome = &ctx.result.outcome;
let stage_index = state.stage_index;
// Emit StageFailed event
@ -273,11 +279,8 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
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,
)
failure: outcome.failure.clone().unwrap_or_else(|| {
FailureDetail::new("handler failed", FailureCategory::TransientInfra)
}),
will_retry: true,
});
@ -298,42 +301,40 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
async fn after_node(
&self,
node: &WorkflowNode,
result: &mut NodeResult,
state: &RunState,
result: &mut WfNodeResult,
state: &WfRunState,
) -> CoreResult<()> {
let gv = node.inner();
let stage_index = state.stage_index;
let mut wf_outcome = core_to_wf_outcome(&result.outcome);
let outcome = &mut result.outcome;
// Auto-status override
if gv.auto_status()
&& wf_outcome.status != WfStatus::Success
&& wf_outcome.status != WfStatus::Skipped
&& outcome.status != StageStatus::Success
&& outcome.status != StageStatus::Skipped
{
wf_outcome.status = WfStatus::Success;
wf_outcome.notes =
outcome.status = StageStatus::Success;
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)
let outcome_failure_category = if outcome.status == StageStatus::Fail {
outcome.failure.as_ref().map(|f| f.category)
} else {
None
};
if let Some(fc) = outcome_failure_class {
let sig_hint = wf_outcome
if let Some(fc) = outcome_failure_category {
let sig_hint = outcome
.failure
.as_ref()
.and_then(|f| f.failure_signature.as_deref());
.and_then(|f| f.signature.as_deref());
let sig = FailureSignature::new(
&gv.id,
fc,
sig_hint,
wf_outcome.failure.as_ref().map(|f| f.message.as_str()),
outcome.failure.as_ref().map(|f| f.message.as_str()),
);
if fc.is_signature_tracked() {
let mut sigs = self.loop_failure_signatures.lock().unwrap();
@ -350,16 +351,13 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
// Emit StageCompleted or StageFailed event
let duration_ms = result.duration.as_millis() as u64;
if wf_outcome.status == WfStatus::Fail {
if outcome.status == StageStatus::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,
)
failure: outcome.failure.clone().unwrap_or_else(|| {
FailureDetail::new("handler failed", FailureCategory::Deterministic)
}),
will_retry: false,
});
@ -369,34 +367,34 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
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(),
status: outcome.status.to_string(),
preferred_label: outcome.preferred_label.clone(),
suggested_next_ids: outcome.suggested_next_ids.clone(),
usage: outcome.usage.clone(),
failure: None,
notes: wf_outcome.notes.clone(),
files_touched: wf_outcome.files_touched.clone(),
notes: outcome.notes.clone(),
files_touched: 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 {
let hook_event = if outcome.status == StageStatus::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());
hook_ctx.status = Some(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");
let _ = crate::save_json(outcome, &status_path, "node_status");
Ok(())
}
@ -404,7 +402,7 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
async fn on_edge_selected(
&self,
ctx: &EdgeContext<'_, WorkflowGraph>,
_state: &RunState,
_state: &WfRunState,
) -> CoreResult<EdgeDecision> {
// Capture fidelity/thread from edge for next node
if let Some(ref edge) = ctx.edge {
@ -416,8 +414,7 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
*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);
let outcome = ctx.outcome;
// Emit EdgeSelected event
let label = ctx
@ -434,9 +431,9 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
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(),
preferred_label: outcome.preferred_label.clone(),
suggested_next_ids: outcome.suggested_next_ids.clone(),
stage_status: outcome.status.to_string(),
is_jump: ctx.is_jump,
});
@ -466,23 +463,18 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
async fn on_checkpoint(
&self,
node: &WorkflowNode,
result: &NodeResult,
result: &WfNodeResult,
next_node_id: Option<&str>,
state: &RunState,
state: &WfRunState,
) -> 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();
// Build checkpoint from state — outcomes are already the wf type
let mut node_outcomes: HashMap<String, Outcome> = state.node_outcomes.clone();
// Include current node's outcome
node_outcomes.insert(node.id().to_string(), wf_outcome);
node_outcomes.insert(node.id().to_string(), result.outcome.clone());
let checkpoint = Checkpoint {
timestamp: chrono::Utc::now(),
@ -508,7 +500,7 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
}
// Emit CheckpointCompleted event
let status = core_to_wf_status(&result.outcome.status).to_string();
let status = result.outcome.status.to_string();
self.emitter.emit(&WorkflowRunEvent::CheckpointCompleted {
node_id: node.id().to_string(),
status,
@ -518,21 +510,20 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
Ok(())
}
async fn on_run_end(&self, outcome: &CoreOutcome, state: &RunState) {
async fn on_run_end(&self, outcome: &Outcome, state: &WfRunState) {
// 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 {
if outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess {
// Success path
self.emitter.emit(&WorkflowRunEvent::WorkflowRunCompleted {
duration_ms,
artifact_count: 0,
status: wf_outcome.status.to_string(),
status: outcome.status.to_string(),
total_cost: None,
final_git_commit_sha: None,
usage: None,
@ -547,7 +538,7 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
let _ = self.run_hook(&hook_ctx).await;
} else {
// Failure path
let error_msg = wf_outcome
let error_msg = outcome
.failure
.as_ref()
.map(|f| f.message.clone())

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

@ -2,7 +2,6 @@ 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};

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

@ -1,173 +0,0 @@
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)
}
}

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

@ -20,11 +20,11 @@ use crate::checkpoint::Checkpoint;
use crate::condition::evaluate_condition;
use crate::context;
use crate::context::Context;
use crate::error::{FabroError, FailureClass, FailureSignature, Result};
use crate::error::{FabroError, FailureCategory, FailureSignature, Result};
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::handler::{EngineServices, HandlerRegistry};
use crate::millis_u64;
use crate::outcome::{Outcome, StageStatus};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::preamble::build_preamble;
use fabro_config::run::PullRequestConfig;
use fabro_graphviz::graph::{Edge, Graph, Node};
@ -46,12 +46,12 @@ pub(crate) fn set_hook_node(ctx: &mut HookContext, node: &Node) {
/// 2. String heuristics on `failure_reason`
/// 3. Default to `Deterministic`
#[must_use]
fn classify_outcome(outcome: &Outcome) -> Option<FailureClass> {
fn classify_outcome(outcome: &Outcome) -> Option<FailureCategory> {
match outcome.status {
StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped => None,
StageStatus::Fail | StageStatus::Retry => outcome
.failure_class()
.or(Some(FailureClass::Deterministic)),
.failure_category()
.or(Some(FailureCategory::Deterministic)),
}
}
@ -991,9 +991,9 @@ impl WorkflowRunEngine {
let decision = self.run_hooks(&hook_ctx, hook_work_dir).await;
match decision {
HookDecision::Skip { reason } => {
let mut outcome = Outcome::skipped();
outcome.notes =
Some(reason.unwrap_or_else(|| "skipped by StageStart hook".into()));
let msg = reason.unwrap_or_else(|| "skipped by StageStart hook".into());
let mut outcome = Outcome::skipped(&msg);
outcome.notes = Some(msg);
return Ok((outcome, attempt));
}
HookDecision::Block { reason } => {
@ -1115,8 +1115,8 @@ impl WorkflowRunEngine {
index: stage_index,
failure: crate::outcome::FailureDetail {
message: e.to_string(),
failure_class: e.failure_class(),
failure_signature: e.failure_signature_hint(),
category: e.failure_category(),
signature: e.failure_signature_hint(),
},
will_retry: true,
});
@ -1482,10 +1482,7 @@ impl WorkflowRunEngine {
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),
);
s.node_outcomes.insert(k.clone(), v.clone());
}
// Set start node to the checkpoint's next_node_id
if let Some(ref next) = cp.next_node_id {
@ -1587,10 +1584,9 @@ impl WorkflowRunEngine {
// 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)
// Outcome is now the wf type directly — no conversion needed
let ctx = Context::new();
Ok((wf_outcome, ctx))
Ok((core_outcome, ctx))
}
Err(fabro_core::CoreError::StallTimeout { node_id }) => {
let stall_timeout = graph.stall_timeout().unwrap_or_default();
@ -2066,7 +2062,7 @@ impl WorkflowRunEngine {
let sig_hint = outcome
.failure
.as_ref()
.and_then(|f| f.failure_signature.as_deref());
.and_then(|f| f.signature.as_deref());
let sig = FailureSignature::new(&node.id, fc, sig_hint, outcome.failure_reason());
if fc.is_signature_tracked() {
let count = loop_state
@ -2095,7 +2091,10 @@ impl WorkflowRunEngine {
name: node.label().to_string(),
index: stage_index,
failure: outcome.failure.clone().unwrap_or_else(|| {
crate::outcome::FailureDetail::new("unknown", FailureClass::Deterministic)
crate::outcome::FailureDetail::new(
"unknown",
FailureCategory::Deterministic,
)
}),
will_retry: false,
});
@ -2428,7 +2427,7 @@ impl WorkflowRunEngine {
"git checkpoint commit failed for node '{}': {e}",
node.id
),
failure_class: FailureClass::Deterministic,
failure_class: FailureCategory::Deterministic,
});
}
}
@ -2502,7 +2501,7 @@ impl WorkflowRunEngine {
if edge.loop_restart() {
// Guard: only transient_infra failures may loop_restart (matches Kilroy)
if let Some(fc) = outcome_failure_class {
if fc != FailureClass::TransientInfra {
if fc != FailureCategory::TransientInfra {
return Err(FabroError::engine(format!(
"loop_restart blocked: failure_class={fc} (requires transient_infra), node={}, failure_reason={}",
node.id,
@ -4927,7 +4926,7 @@ mod tests {
#[test]
fn classify_outcome_returns_none_for_skipped() {
assert!(classify_outcome(&Outcome::skipped()).is_none());
assert!(classify_outcome(&Outcome::skipped("")).is_none());
}
#[test]
@ -4943,10 +4942,10 @@ mod tests {
fn classify_outcome_reads_failure_detail() {
let mut outcome = Outcome::fail_classify("some error");
// Override the FailureDetail's class directly
outcome.failure.as_mut().unwrap().failure_class = FailureClass::BudgetExhausted;
outcome.failure.as_mut().unwrap().category = FailureCategory::BudgetExhausted;
assert_eq!(
classify_outcome(&outcome),
Some(FailureClass::BudgetExhausted)
Some(FailureCategory::BudgetExhausted)
);
}
@ -4955,7 +4954,7 @@ mod tests {
let outcome = Outcome::fail_classify("rate limited by provider");
assert_eq!(
classify_outcome(&outcome),
Some(FailureClass::TransientInfra)
Some(FailureCategory::TransientInfra)
);
}
@ -4964,7 +4963,7 @@ mod tests {
let outcome = Outcome::fail_classify("something went wrong");
assert_eq!(
classify_outcome(&outcome),
Some(FailureClass::Deterministic)
Some(FailureCategory::Deterministic)
);
}
@ -4977,7 +4976,7 @@ mod tests {
};
assert_eq!(
classify_outcome(&outcome),
Some(FailureClass::Deterministic)
Some(FailureCategory::Deterministic)
);
}
@ -4986,7 +4985,7 @@ mod tests {
let outcome = Outcome::retry_classify("connection refused");
assert_eq!(
classify_outcome(&outcome),
Some(FailureClass::TransientInfra)
Some(FailureCategory::TransientInfra)
);
}

477
lib/crates/fabro-workflows/src/error.rs
View file

File diff suppressed because it is too large Load diff

16
lib/crates/fabro-workflows/src/event.rs
View file

@ -1294,8 +1294,7 @@ mod tests {
#[test]
fn stage_completed_event_serialization_with_new_fields() {
use crate::error::FailureClass;
use crate::outcome::FailureDetail;
use crate::outcome::{FailureCategory, FailureDetail};
let event = WorkflowRunEvent::StageCompleted {
node_id: "plan".to_string(),
@ -1308,7 +1307,7 @@ mod tests {
usage: None,
failure: Some(FailureDetail::new(
"lint errors remain",
FailureClass::Deterministic,
FailureCategory::Deterministic,
)),
notes: Some("fixed 3 of 5 issues".to_string()),
files_touched: vec!["src/main.rs".to_string()],
@ -1343,8 +1342,7 @@ mod tests {
#[test]
fn stage_failed_event_serialization() {
use crate::error::FailureClass;
use crate::outcome::FailureDetail;
use crate::outcome::{FailureCategory, FailureDetail};
let event = WorkflowRunEvent::StageFailed {
node_id: "plan".to_string(),
@ -1352,8 +1350,8 @@ mod tests {
index: 0,
failure: FailureDetail {
message: "LLM request timed out".to_string(),
failure_class: FailureClass::TransientInfra,
failure_signature: None,
category: FailureCategory::TransientInfra,
signature: None,
},
will_retry: true,
};
@ -1364,14 +1362,14 @@ mod tests {
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(
deserialized,
WorkflowRunEvent::StageFailed { failure, .. } if failure.failure_class == FailureClass::TransientInfra
WorkflowRunEvent::StageFailed { failure, .. } if failure.category == FailureCategory::TransientInfra
));
let event_terminal = WorkflowRunEvent::StageFailed {
node_id: "plan".to_string(),
name: "plan".to_string(),
index: 0,
failure: FailureDetail::new("timeout", FailureClass::Deterministic),
failure: FailureDetail::new("timeout", FailureCategory::Deterministic),
will_retry: false,
};
let json_terminal = serde_json::to_string(&event_terminal).unwrap();

4
lib/crates/fabro-workflows/src/handler/agent.rs
View file

@ -9,7 +9,7 @@ use crate::context::keys;
use crate::context::Context;
use crate::error::FabroError;
use crate::event::EventEmitter;
use crate::outcome::{Outcome, StageUsage};
use crate::outcome::{Outcome, OutcomeExt, StageUsage};
use fabro_graphviz::graph::{Graph, Node};
use super::{EngineServices, Handler};
@ -170,7 +170,7 @@ pub(crate) fn extract_status_fields(text: &str, outcome: &mut Outcome) -> bool {
if let Some(reason) = obj.get("failure_reason").and_then(|v| v.as_str()) {
outcome.failure = Some(crate::outcome::FailureDetail::new(
reason,
crate::error::FailureClass::Deterministic,
crate::outcome::FailureCategory::Deterministic,
));
}
}

2
lib/crates/fabro-workflows/src/handler/command.rs
View file

@ -5,7 +5,7 @@ use async_trait::async_trait;
use crate::context::keys;
use crate::context::Context;
use crate::error::FabroError;
use crate::outcome::Outcome;
use crate::outcome::{Outcome, OutcomeExt};
use fabro_graphviz::graph::{Graph, Node};
use super::{EngineServices, Handler};

2
lib/crates/fabro-workflows/src/handler/fan_in.rs
View file

@ -8,7 +8,7 @@ use crate::context::keys;
use crate::context::Context;
use crate::error::FabroError;
use crate::event::EventEmitter;
use crate::outcome::Outcome;
use crate::outcome::{Outcome, OutcomeExt};
use fabro_graphviz::graph::{Graph, Node};
use super::agent::{CodergenBackend, CodergenResult};

2
lib/crates/fabro-workflows/src/handler/human.rs
View file

@ -9,7 +9,7 @@ use crate::context::Context;
use crate::error::FabroError;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::millis_u64;
use crate::outcome::Outcome;
use crate::outcome::{Outcome, OutcomeExt};
use fabro_graphviz::graph::{Graph, Node};
use fabro_interview::{Answer, AnswerValue, Interviewer, Question, QuestionOption, QuestionType};

2
lib/crates/fabro-workflows/src/handler/manager_loop.rs
View file

@ -11,7 +11,7 @@ use crate::context::keys;
use crate::context::Context;
use crate::engine::{RunConfig, WorkflowRunEngine};
use crate::error::FabroError;
use crate::outcome::{Outcome, StageStatus};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::workflow::{prepare_from_file, prepare_from_source};
use fabro_graphviz::graph::{Graph, Node};

2
lib/crates/fabro-workflows/src/handler/mod.rs
View file

@ -21,7 +21,7 @@ use crate::context::Context;
use crate::engine::GitState;
use crate::error::FabroError;
use crate::event::EventEmitter;
use crate::outcome::Outcome;
use crate::outcome::{Outcome, OutcomeExt};
use fabro_graphviz::graph::{shape_to_handler_type, Graph, Node};
use fabro_hooks::{HookContext, HookDecision, HookRunner};
use fabro_interview::Interviewer;

4
lib/crates/fabro-workflows/src/handler/parallel.rs
View file

@ -12,7 +12,7 @@ use crate::engine::set_hook_node;
use crate::error::FabroError;
use crate::event::WorkflowRunEvent;
use crate::millis_u64;
use crate::outcome::{Outcome, StageStatus};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use fabro_graphviz::graph::{Graph, Node};
use fabro_hooks::{HookContext, HookEvent};
@ -516,7 +516,7 @@ impl Handler for ParallelHandler {
failure: if is_fail {
Some(crate::outcome::FailureDetail::new(
format!("Join policy not satisfied: {success_count}/{total} succeeded"),
crate::error::FailureClass::Deterministic,
crate::outcome::FailureCategory::Deterministic,
))
} else {
None

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

@ -31,7 +31,7 @@ pub fn build_completed_stages(
cp: &checkpoint::Checkpoint,
run_failed: bool,
) -> Vec<fabro_retro::retro::CompletedStage> {
use outcome::StageStatus;
use outcome::{OutcomeExt, StageStatus};
let mut stages = Vec::new();
let mut any_stage_failed = false;

284
lib/crates/fabro-workflows/src/outcome.rs
View file

@ -1,49 +1,8 @@
use std::collections::HashMap;
use std::fmt;
use std::str::FromStr;
use serde::{Deserialize, Serialize};
use crate::error::{classify_failure_reason, FailureClass};
pub use fabro_core::outcome::{FailureCategory, FailureDetail, OutcomeMeta, StageStatus};
/// Status of a pipeline stage execution.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StageStatus {
Success,
Fail,
PartialSuccess,
Retry,
Skipped,
}
impl fmt::Display for StageStatus {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let s = match self {
Self::Success => "success",
Self::Fail => "fail",
Self::PartialSuccess => "partial_success",
Self::Retry => "retry",
Self::Skipped => "skipped",
};
write!(f, "{s}")
}
}
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),
"partial_success" => Ok(Self::PartialSuccess),
"retry" => Ok(Self::Retry),
"skipped" => Ok(Self::Skipped),
other => Err(format!("unknown stage status: {other}")),
}
}
}
use crate::error::classify_failure_reason;
/// Token usage from a single pipeline stage.
#[derive(Debug, Clone, Serialize, Deserialize)]
@ -78,163 +37,82 @@ impl From<&StageUsage> for fabro_llm::types::Usage {
}
}
/// Structured failure information carried through the pipeline.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FailureDetail {
pub message: String,
pub failure_class: FailureClass,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub failure_signature: Option<String>,
}
/// The workflow-specific Outcome type, parameterized with optional stage usage.
pub type Outcome = fabro_core::Outcome<Option<StageUsage>>;
impl FailureDetail {
pub fn new(message: impl Into<String>, failure_class: FailureClass) -> Self {
Self {
message: message.into(),
failure_class,
failure_signature: None,
}
}
}
/// Extension trait for workflow-specific Outcome factory methods and accessors.
pub trait OutcomeExt: Sized {
/// Create a failed outcome with a deterministic failure category.
fn fail_deterministic(reason: impl Into<String>) -> Self;
/// The result of executing a node handler.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Outcome {
pub status: StageStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub preferred_label: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub suggested_next_ids: Vec<String>,
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub context_updates: HashMap<String, serde_json::Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub notes: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub failure: Option<FailureDetail>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub usage: Option<StageUsage>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub files_touched: Vec<String>,
/// When set, the engine bypasses edge selection and jumps directly to this node.
/// Used by the parallel handler to skip re-executing branch nodes.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub jump_to_node: Option<String>,
/// Wall-clock duration of the stage execution in milliseconds.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub duration_ms: Option<u64>,
}
/// Create a failed outcome with the failure category inferred from the message via heuristics.
fn fail_classify(reason: impl Into<String>) -> Self;
impl Outcome {
#[must_use]
pub fn success() -> Self {
Self {
status: StageStatus::Success,
preferred_label: None,
suggested_next_ids: Vec::new(),
context_updates: HashMap::new(),
notes: None,
failure: None,
usage: None,
files_touched: Vec::new(),
jump_to_node: None,
duration_ms: None,
}
}
/// Create a failed outcome with a deterministic failure class.
pub fn fail_deterministic(reason: impl Into<String>) -> Self {
Self {
status: StageStatus::Fail,
preferred_label: None,
suggested_next_ids: Vec::new(),
context_updates: HashMap::new(),
notes: None,
failure: Some(FailureDetail::new(reason, FailureClass::Deterministic)),
usage: None,
files_touched: Vec::new(),
jump_to_node: None,
duration_ms: None,
}
}
/// Create a failed outcome with the failure class inferred from the message via heuristics.
pub fn fail_classify(reason: impl Into<String>) -> Self {
let reason = reason.into();
let failure_class = classify_failure_reason(&reason);
Self {
status: StageStatus::Fail,
preferred_label: None,
suggested_next_ids: Vec::new(),
context_updates: HashMap::new(),
notes: None,
failure: Some(FailureDetail::new(reason, failure_class)),
usage: None,
files_touched: Vec::new(),
jump_to_node: None,
duration_ms: None,
}
}
/// Create a retry outcome with the failure class inferred from the message via heuristics.
pub fn retry_classify(reason: impl Into<String>) -> Self {
let reason = reason.into();
let failure_class = classify_failure_reason(&reason);
Self {
status: StageStatus::Retry,
preferred_label: None,
suggested_next_ids: Vec::new(),
context_updates: HashMap::new(),
notes: None,
failure: Some(FailureDetail::new(reason, failure_class)),
usage: None,
files_touched: Vec::new(),
jump_to_node: None,
duration_ms: None,
}
}
/// Set the failure signature on this outcome. Returns self for chaining.
#[must_use]
pub fn with_signature(mut self, sig: Option<impl Into<String>>) -> Self {
if let Some(ref mut f) = self.failure {
f.failure_signature = sig.map(Into::into);
}
self
}
#[must_use]
pub fn skipped() -> Self {
Self {
status: StageStatus::Skipped,
preferred_label: None,
suggested_next_ids: Vec::new(),
context_updates: HashMap::new(),
notes: None,
failure: None,
usage: None,
files_touched: Vec::new(),
jump_to_node: None,
duration_ms: None,
}
}
/// Create a retry outcome with the failure category inferred from the message via heuristics.
fn retry_classify(reason: impl Into<String>) -> Self;
/// Create a simulated success outcome for dry-run mode.
#[must_use]
pub fn simulated(node_id: &str) -> Self {
fn simulated(node_id: &str) -> Self;
/// Set the failure signature on this outcome. Returns self for chaining.
fn with_signature(self, sig: Option<impl Into<String>>) -> Self;
/// Get the failure reason message, if any.
fn failure_reason(&self) -> Option<&str>;
/// Get the failure category, if this is a failed outcome.
fn failure_category(&self) -> Option<FailureCategory>;
}
impl OutcomeExt for Outcome {
fn fail_deterministic(reason: impl Into<String>) -> Self {
Self {
status: StageStatus::Fail,
failure: Some(FailureDetail::new(reason, FailureCategory::Deterministic)),
..Self::default()
}
}
fn fail_classify(reason: impl Into<String>) -> Self {
let reason = reason.into();
let category = classify_failure_reason(&reason);
Self {
status: StageStatus::Fail,
failure: Some(FailureDetail::new(reason, category)),
..Self::default()
}
}
fn retry_classify(reason: impl Into<String>) -> Self {
let reason = reason.into();
let category = classify_failure_reason(&reason);
Self {
status: StageStatus::Retry,
failure: Some(FailureDetail::new(reason, category)),
..Self::default()
}
}
fn simulated(node_id: &str) -> Self {
Self {
notes: Some(format!("[Simulated] {node_id}")),
..Self::success()
}
}
/// Get the failure reason message, if any.
pub fn failure_reason(&self) -> Option<&str> {
fn with_signature(mut self, sig: Option<impl Into<String>>) -> Self {
if let Some(ref mut f) = self.failure {
f.signature = sig.map(Into::into);
}
self
}
fn failure_reason(&self) -> Option<&str> {
self.failure.as_ref().map(|f| f.message.as_str())
}
/// Get the failure class, if this is a failed outcome.
pub fn failure_class(&self) -> Option<FailureClass> {
self.failure.as_ref().map(|f| f.failure_class)
fn failure_category(&self) -> Option<FailureCategory> {
self.failure.as_ref().map(|f| f.category)
}
}
@ -290,7 +168,7 @@ mod tests {
let o = Outcome::fail_deterministic("something broke");
assert_eq!(o.status, StageStatus::Fail);
assert_eq!(o.failure_reason(), Some("something broke"));
assert_eq!(o.failure_class(), Some(FailureClass::Deterministic));
assert_eq!(o.failure_category(), Some(FailureCategory::Deterministic));
}
#[test]
@ -298,7 +176,7 @@ mod tests {
let o = Outcome::fail_classify("connection refused");
assert_eq!(o.status, StageStatus::Fail);
assert_eq!(o.failure_reason(), Some("connection refused"));
assert_eq!(o.failure_class(), Some(FailureClass::TransientInfra));
assert_eq!(o.failure_category(), Some(FailureCategory::TransientInfra));
}
#[test]
@ -310,46 +188,46 @@ mod tests {
#[test]
fn outcome_skipped_factory() {
let o = Outcome::skipped();
let o = Outcome::skipped("");
assert_eq!(o.status, StageStatus::Skipped);
assert!(o.failure.is_none());
}
#[test]
fn failure_detail_construction() {
let fd = FailureDetail::new("timeout", FailureClass::TransientInfra);
let fd = FailureDetail::new("timeout", FailureCategory::TransientInfra);
assert_eq!(fd.message, "timeout");
assert_eq!(fd.failure_class, FailureClass::TransientInfra);
assert!(fd.failure_signature.is_none());
assert_eq!(fd.category, FailureCategory::TransientInfra);
assert!(fd.signature.is_none());
}
#[test]
fn failure_detail_serde_roundtrip() {
let fd = FailureDetail {
message: "timeout".into(),
failure_class: FailureClass::TransientInfra,
failure_signature: Some("sig".into()),
category: FailureCategory::TransientInfra,
signature: Some("sig".into()),
};
let json = serde_json::to_string(&fd).unwrap();
let deserialized: FailureDetail = serde_json::from_str(&json).unwrap();
assert_eq!(deserialized.message, "timeout");
assert_eq!(deserialized.failure_class, FailureClass::TransientInfra);
assert_eq!(deserialized.failure_signature.as_deref(), Some("sig"));
assert_eq!(deserialized.category, FailureCategory::TransientInfra);
assert_eq!(deserialized.signature.as_deref(), Some("sig"));
}
#[test]
fn fail_classify_known_patterns() {
assert_eq!(
Outcome::fail_classify("timeout").failure_class(),
Some(FailureClass::TransientInfra)
Outcome::fail_classify("timeout").failure_category(),
Some(FailureCategory::TransientInfra)
);
assert_eq!(
Outcome::fail_classify("context length exceeded").failure_class(),
Some(FailureClass::BudgetExhausted)
Outcome::fail_classify("context length exceeded").failure_category(),
Some(FailureCategory::BudgetExhausted)
);
assert_eq!(
Outcome::fail_classify("cancel").failure_class(),
Some(FailureClass::Canceled)
Outcome::fail_classify("cancel").failure_category(),
Some(FailureCategory::Canceled)
);
}
@ -367,7 +245,7 @@ mod tests {
fn with_signature_builder() {
let o = Outcome::fail_deterministic("x").with_signature(Some("sig"));
assert_eq!(
o.failure.as_ref().unwrap().failure_signature.as_deref(),
o.failure.as_ref().unwrap().signature.as_deref(),
Some("sig")
);
}

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

@ -4,6 +4,7 @@ use crate::artifact::{artifact_path, format_artifact_reference};
use crate::context::keys;
use crate::context::Context;
use crate::outcome::Outcome;
use crate::outcome::OutcomeExt;
use fabro_graphviz::graph::{is_llm_handler_type, Graph, Node};
const COMPACT_OUTPUT_MAX_LINES: usize = 25;

2
lib/crates/fabro-workflows/tests/daytona_integration.rs
View file

@ -20,7 +20,7 @@ use fabro_workflows::event::EventEmitter;
use fabro_workflows::handler::exit::ExitHandler;
use fabro_workflows::handler::start::StartHandler;
use fabro_workflows::handler::{Handler, HandlerRegistry};
use fabro_workflows::outcome::{Outcome, StageStatus};
use fabro_workflows::outcome::{Outcome, OutcomeExt, StageStatus};
async fn create_env() -> DaytonaSandbox {
let creds = load_github_app_credentials();

34
lib/crates/fabro-workflows/tests/integration.rs
View file

@ -29,7 +29,7 @@ use fabro_workflows::handler::manager_loop::SubWorkflowHandler;
use fabro_workflows::handler::start::StartHandler;
use fabro_workflows::handler::wait::WaitHandler;
use fabro_workflows::handler::{Handler, HandlerRegistry};
use fabro_workflows::outcome::{Outcome, StageStatus};
use fabro_workflows::outcome::{Outcome, OutcomeExt, StageStatus};
use fabro_workflows::stylesheet::{apply_stylesheet, parse_stylesheet};
use fabro_workflows::transform::{
StylesheetApplicationTransform, Transform, VariableExpansionTransform,
@ -11821,11 +11821,11 @@ fn e2e_normalize_failure_reason_strips_variable_data() {
#[test]
fn e2e_failure_signature_composite_key() {
use fabro_workflows::error::{FailureClass, FailureSignature};
use fabro_workflows::error::{FailureCategory, FailureSignature};
let sig = FailureSignature::new(
"verify",
FailureClass::Deterministic,
FailureCategory::Deterministic,
None,
Some("assertion failed at line 42"),
);
@ -11848,11 +11848,11 @@ fn e2e_failure_signature_composite_key() {
#[test]
fn e2e_failure_signature_hint_priority() {
use fabro_workflows::error::{FailureClass, FailureSignature};
use fabro_workflows::error::{FailureCategory, FailureSignature};
let sig = FailureSignature::new(
"build",
FailureClass::Deterministic,
FailureCategory::Deterministic,
Some("custom-key-abc"),
Some("raw error with line 123 and hash deadbeef"),
);
@ -11865,17 +11865,17 @@ fn e2e_failure_signature_hint_priority() {
#[test]
fn e2e_only_deterministic_and_structural_tracked() {
use fabro_workflows::error::FailureClass;
use fabro_workflows::error::FailureCategory;
// These should be tracked
assert!(FailureClass::Deterministic.is_signature_tracked());
assert!(FailureClass::Structural.is_signature_tracked());
assert!(FailureCategory::Deterministic.is_signature_tracked());
assert!(FailureCategory::Structural.is_signature_tracked());
// These should NOT be tracked (transient failures retry naturally)
assert!(!FailureClass::TransientInfra.is_signature_tracked());
assert!(!FailureClass::BudgetExhausted.is_signature_tracked());
assert!(!FailureClass::Canceled.is_signature_tracked());
assert!(!FailureClass::CompilationLoop.is_signature_tracked());
assert!(!FailureCategory::TransientInfra.is_signature_tracked());
assert!(!FailureCategory::BudgetExhausted.is_signature_tracked());
assert!(!FailureCategory::Canceled.is_signature_tracked());
assert!(!FailureCategory::CompilationLoop.is_signature_tracked());
}
// --- E2E Test: loop_restart_signature_limit graph attribute ---
@ -12373,7 +12373,7 @@ fn e2e_checkpoint_backward_compat_no_signatures() {
#[test]
fn e2e_checkpoint_signatures_roundtrip() {
use fabro_workflows::error::{FailureClass, FailureSignature};
use fabro_workflows::error::{FailureCategory, FailureSignature};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("cp.json");
@ -12384,7 +12384,7 @@ fn e2e_checkpoint_signatures_roundtrip() {
let mut loop_sigs = std::collections::HashMap::new();
let sig1 = FailureSignature::new(
"verify",
FailureClass::Deterministic,
FailureCategory::Deterministic,
None,
Some("assertion failed"),
);
@ -12393,7 +12393,7 @@ fn e2e_checkpoint_signatures_roundtrip() {
let mut restart_sigs = std::collections::HashMap::new();
let sig2 = FailureSignature::new(
"build",
FailureClass::Structural,
FailureCategory::Structural,
None,
Some("scope violation"),
);
@ -12692,11 +12692,11 @@ impl Handler for ClassifiedFailHandler {
if n >= self.succeed_on {
return Ok(Outcome::success());
}
let failure_class: fabro_workflows::error::FailureClass =
let failure_class: fabro_workflows::error::FailureCategory =
self.failure_class.parse().unwrap();
let mut outcome = Outcome::fail_classify("classified failure");
if let Some(ref mut f) = outcome.failure {
f.failure_class = failure_class;
f.category = failure_class;
}
Ok(outcome)
}