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

Decompose monolithic WorkflowLifecycle into 8 focused sub-lifecycles

Browse source 
Split the 564-line core_adapter/lifecycle.rs into a lifecycle/ directory
with dedicated structs for each domain concern (event, hook, fidelity,
auto_status, circuit_breaker, disk, git, artifact), orchestrated by a
WorkflowLifecycle that enforces explicit per-callback ordering.

Also fixes core adapter boundary gaps:
- Handler now uses per-call snapshot/apply context bridge and real graph
  instead of STUB_GRAPH
- Executor::run() returns (Outcome, RunState) so run_via_core can
  extract the final context instead of returning an empty one
- run_via_core populates git_state on EngineServices for handlers
- Checkpoint resume gains stage_index, next_node_id fallback, and
  node_visits reconstruction for old checkpoints

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-24 14:46:10 -04:00
parent 4a7433a031
commit 5e6eebaff8
14 changed files with 1167 additions and 599 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -80,7 +80,11 @@ impl<G: Graph + 'static> ExecutorBuilder<G> {
}
impl<G: Graph + 'static> Executor<G> {
pub async fn run(&self, graph: &G, mut state: RunState<G::Meta>) -> Result<Outcome<G::Meta>> {
pub async fn run(
&self,
graph: &G,
mut state: RunState<G::Meta>,
) -> Result<(Outcome<G::Meta>, RunState<G::Meta>)> {
self.lifecycle.on_run_start(graph, &state).await?;
loop {
@ -109,7 +113,7 @@ impl<G: Graph + 'static> Executor<G> {
.await;
let outcome = Outcome::success();
self.lifecycle.on_run_end(&outcome, &state).await;
return Ok(outcome);
return Ok((outcome, state));
}
Err(failed_node_id) => {
self.lifecycle
@ -131,7 +135,7 @@ impl<G: Graph + 'static> Executor<G> {
failed_node_id
));
self.lifecycle.on_run_end(&outcome, &state).await;
return Ok(outcome);
return Ok((outcome, state));
}
}
}
@ -220,7 +224,7 @@ impl<G: Graph + 'static> Executor<G> {
NextStep::End => {
let outcome = last_outcome.clone();
self.lifecycle.on_run_end(&outcome, &state).await;
return Ok(outcome);
return Ok((outcome, state));
}
}
}
@ -413,7 +417,10 @@ mod tests {
let g = linear_graph(node_ids);
let state = RunState::new(&g)?;
let executor = ExecutorBuilder::new(handler).build();
executor.run(&g, state).await
executor
.run(&g, state)
.await
.map(|(outcome, _state)| outcome)
}
// ---- Step 8: Linear happy path ----
@ -476,7 +483,7 @@ mod tests {
let executor =
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
}
@ -502,7 +509,7 @@ mod tests {
let state = RunState::new(&g).unwrap();
let executor =
ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>).build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
assert_eq!(handler.calls(), 2);
}
@ -523,7 +530,7 @@ mod tests {
Arc::new(AlwaysFailHandler::new("nope")) as Arc<dyn NodeHandler<TestGraph>>
)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Fail);
}
@ -673,7 +680,7 @@ mod tests {
Arc::new(AlwaysFailHandler::new("oops")) as Arc<dyn NodeHandler<TestGraph>>
)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
// Ends at "bad" terminal with success (goal gates pass since no gates defined)
assert_eq!(result.status, StageStatus::Success);
}
@ -696,7 +703,7 @@ mod tests {
let executor =
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
}
@ -729,7 +736,7 @@ mod tests {
let state = RunState::new(&g).unwrap();
let executor =
ExecutorBuilder::new(Arc::new(JumpHandler) as Arc<dyn NodeHandler<TestGraph>>).build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
}
@ -765,7 +772,7 @@ mod tests {
let executor = ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>)
.max_node_visits(5)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
assert_eq!(handler.calls(), 4);
}
@ -829,7 +836,7 @@ mod tests {
Arc::new(AlwaysFailHandler::new("boom")) as Arc<dyn NodeHandler<TestGraph>>
)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Fail);
}
@ -841,7 +848,7 @@ mod tests {
let executor =
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
}
@ -1033,7 +1040,7 @@ mod tests {
let executor =
ExecutorBuilder::new(Arc::new(ExhaustedHandler) as Arc<dyn NodeHandler<TestGraph>>)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::PartialSuccess);
}
@ -1171,7 +1178,7 @@ mod tests {
let executor = ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>)
.lifecycle(Box::new(SkipOnSecondAttempt(call_count_clone)))
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success); // overall run succeeds via terminal
assert_eq!(handler.calls(), 1); // handler only called once
assert_eq!(call_count.load(Ordering::Relaxed), 2); // before_attempt called twice
@ -1234,7 +1241,7 @@ mod tests {
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
.lifecycle(Box::new(SkipFirst(Mutex::new(false))))
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
}
@ -1309,7 +1316,7 @@ mod tests {
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
.lifecycle(Box::new(Redirector))
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
}
@ -1725,7 +1732,7 @@ mod tests {
let state = RunState::new(&g).unwrap();
let executor =
ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>).build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
assert_eq!(handler.calls(), 2);
}
@ -1756,7 +1763,7 @@ mod tests {
let executor = ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>)
.max_node_visits(5)
.build();
let result = executor.run(&g, state).await.unwrap();
let (result, _) = executor.run(&g, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
assert_eq!(handler.calls(), 2);
}

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

@ -5,6 +5,17 @@ use crate::error::Result;
use crate::graph::{Graph, NodeSpec};
use crate::outcome::{NodeResult, Outcome, OutcomeMeta};
impl<M: OutcomeMeta> std::fmt::Debug for RunState<M> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RunState")
.field("current_node_id", &self.current_node_id)
.field("completed_nodes", &self.completed_nodes)
.field("stage_index", &self.stage_index)
.field("cancelled", &self.cancelled)
.finish_non_exhaustive()
}
}
pub struct RunState<M: OutcomeMeta = ()> {
pub context: Context,
pub current_node_id: String,

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

@ -1,6 +1,6 @@
use std::panic::AssertUnwindSafe;
use std::path::PathBuf;
use std::sync::{Arc, LazyLock};
use std::sync::Arc;
use async_trait::async_trait;
use futures::FutureExt;
@ -17,15 +17,15 @@ use crate::engine;
use crate::handler::{format_panic_message, EngineServices};
use crate::outcome::{Outcome, StageStatus};
/// Cached stub graph for handler dispatch (avoids allocating on every call).
static STUB_GRAPH: LazyLock<fabro_graphviz::graph::types::Graph> =
LazyLock::new(|| fabro_graphviz::graph::types::Graph::new("stub"));
/// Production node handler that bridges fabro-core's NodeHandler to the
/// existing fabro-workflows Handler trait via EngineServices.
///
/// On each `execute()` call, snapshots the CoreContext into a WfContext,
/// runs the handler, then diffs and applies changes back.
pub struct WorkflowNodeHandler {
pub services: Arc<EngineServices>,
pub run_dir: PathBuf,
pub graph: Arc<fabro_graphviz::graph::types::Graph>,
}
#[async_trait]
@ -33,13 +33,16 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
async fn execute(
&self,
node: &WorkflowNode,
_context: &CoreContext,
context: &CoreContext,
_graph: &WorkflowGraph,
) -> CoreResult<Outcome> {
let gv_node = node.inner();
let handler = self.services.registry.resolve(gv_node);
let wf_context = crate::context::Context::new();
// Per-call snapshot/apply context bridge:
// 1. Snapshot the CoreContext into a WfContext
let snapshot = context.snapshot();
let wf_context = crate::context::Context::from_values(snapshot.clone());
// Timeout from the node
let node_timeout = gv_node.timeout();
@ -50,7 +53,7 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
handler,
gv_node,
&wf_context,
&STUB_GRAPH,
&self.graph,
&run_dir,
&self.services,
);
@ -72,6 +75,15 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
panic_safe.await
};
// 2. After handler returns, diff the WfContext against the snapshot
// and apply changes back to the CoreContext
let new_values = wf_context.snapshot();
for (k, v) in &new_values {
if snapshot.get(k) != Some(v) {
context.set(k.clone(), v.clone());
}
}
match timed_result {
Ok(Ok(wf_outcome)) => Ok(wf_outcome),
Ok(Err(fabro_err)) => {
@ -97,7 +109,7 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
fn retry_policy(&self, node: &WorkflowNode, _graph: &WorkflowGraph) -> CoreRetryPolicy {
let gv_node = node.inner();
let wf_policy = engine::build_retry_policy(gv_node, &STUB_GRAPH);
let wf_policy = engine::build_retry_policy(gv_node, &self.graph);
CoreRetryPolicy {
max_attempts: wf_policy.max_attempts,
backoff: wf_policy.backoff,
@ -178,7 +190,7 @@ mod tests {
let executor = ExecutorBuilder::new(handler)
.lifecycle(Box::new(NoopLifecycle))
.build();
let result = executor.run(&wf_graph, state).await.unwrap();
let (result, _) = executor.run(&wf_graph, state).await.unwrap();
assert_eq!(result.status, StageStatus::Success);
}
}

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

@ -1,563 +0,0 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Instant;
use async_trait::async_trait;
use fabro_core::error::{CoreError, Result as CoreResult};
use fabro_core::graph::NodeSpec;
use fabro_core::lifecycle::{
AttemptContext, AttemptResultContext, EdgeContext, EdgeDecision, NodeDecision, RunLifecycle,
};
use fabro_core::outcome::NodeResult;
use fabro_core::state::RunState;
use super::graph::WorkflowGraph;
use super::WorkflowNode;
use crate::checkpoint::Checkpoint;
use crate::context::keys;
use crate::error::{FailureCategory, FailureSignature};
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::outcome::{FailureDetail, Outcome, StageStatus, StageUsage};
use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner};
use fabro_sandbox::Sandbox;
/// Data captured from an edge selection to pass to the next node's before_node.
#[derive(Debug, Clone)]
struct IncomingEdgeData {
fidelity: Option<String>,
thread_id: Option<String>,
}
/// Implements the full RunLifecycle for fabro-workflows, mapping all domain
/// concerns (events, hooks, git, disk I/O, fidelity, circuit breaker, etc.)
/// into fabro-core lifecycle callbacks.
pub struct WorkflowLifecycle {
pub emitter: Arc<EventEmitter>,
pub hook_runner: Option<Arc<HookRunner>>,
pub sandbox: Arc<dyn Sandbox>,
pub graph: Arc<fabro_graphviz::graph::types::Graph>,
pub run_dir: PathBuf,
pub run_id: String,
pub run_start: Instant,
pub labels: HashMap<String, String>,
// Circuit breaker state
loop_failure_signatures: Mutex<HashMap<FailureSignature, usize>>,
restart_failure_signatures: Mutex<HashMap<FailureSignature, usize>>,
// Edge data for next node
incoming_edge_data: Mutex<Option<IncomingEdgeData>>,
// Config flags
pub dry_run: bool,
pub checkpoint_enabled: bool,
}
impl WorkflowLifecycle {
#[allow(clippy::too_many_arguments)]
pub fn new(
emitter: Arc<EventEmitter>,
hook_runner: Option<Arc<HookRunner>>,
sandbox: Arc<dyn Sandbox>,
graph: Arc<fabro_graphviz::graph::types::Graph>,
run_dir: PathBuf,
run_id: String,
dry_run: bool,
labels: HashMap<String, String>,
) -> Self {
Self {
emitter,
hook_runner,
sandbox,
graph,
run_dir,
run_id,
run_start: Instant::now(),
labels,
loop_failure_signatures: Mutex::new(HashMap::new()),
restart_failure_signatures: Mutex::new(HashMap::new()),
incoming_edge_data: Mutex::new(None),
dry_run,
checkpoint_enabled: true,
}
}
/// Restore circuit breaker state from a checkpoint (for resume).
pub fn restore_circuit_breaker(
&self,
loop_sigs: HashMap<FailureSignature, usize>,
restart_sigs: HashMap<FailureSignature, usize>,
) {
*self.loop_failure_signatures.lock().unwrap() = loop_sigs;
*self.restart_failure_signatures.lock().unwrap() = restart_sigs;
}
async fn run_hook(&self, hook_ctx: &HookContext) -> HookDecision {
let Some(ref runner) = self.hook_runner else {
return HookDecision::Proceed;
};
runner
.run(hook_ctx, self.sandbox.clone(), Some(&self.run_dir))
.await
}
}
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: &WfRunState) -> CoreResult<()> {
// Clear incoming edge data (reset stale fidelity/thread from prior iteration)
*self.incoming_edge_data.lock().unwrap() = None;
// Emit WorkflowRunStarted event
self.emitter.emit(&WorkflowRunEvent::WorkflowRunStarted {
name: self.graph.name.clone(),
run_id: self.run_id.clone(),
base_sha: None,
run_branch: None,
worktree_dir: None,
goal: None,
});
// RunStart hook (blocking)
let hook_ctx = HookContext::new(
HookEvent::RunStart,
self.run_id.clone(),
self.graph.name.clone(),
);
let decision = self.run_hook(&hook_ctx).await;
if let HookDecision::Block { reason } = decision {
let msg = reason.unwrap_or_else(|| "blocked by RunStart hook".into());
return Err(CoreError::blocked(msg));
}
Ok(())
}
async fn on_terminal_reached(
&self,
node: &WorkflowNode,
goal_gates_passed: bool,
state: &WfRunState,
) {
if !goal_gates_passed {
return;
}
let gv = node.inner();
let stage_index = state.stage_index;
// Emit StageStarted + StageCompleted for the terminal node
self.emitter.emit(&WorkflowRunEvent::StageStarted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
handler_type: gv.handler_type().map(String::from),
script: None,
attempt: 1,
max_attempts: 1,
});
self.emitter.emit(&WorkflowRunEvent::StageCompleted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
duration_ms: 0,
status: StageStatus::Success.to_string(),
preferred_label: None,
suggested_next_ids: Vec::new(),
usage: None,
failure: None,
notes: None,
files_touched: Vec::new(),
attempt: 1,
max_attempts: 1,
});
}
async fn before_node(
&self,
node: &WorkflowNode,
state: &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
let visits = state.node_visits.get(node.id()).copied().unwrap_or(0);
state
.context
.set(keys::CURRENT_NODE, serde_json::json!(node.id()));
state
.context
.set(keys::INTERNAL_NODE_VISIT_COUNT, serde_json::json!(visits));
// Fidelity resolution
let fidelity = if let Some(ref edge_data) = incoming {
edge_data
.fidelity
.as_deref()
.or(gv_node.fidelity())
.unwrap_or("compact")
.to_string()
} else {
gv_node.fidelity().unwrap_or("compact").to_string()
};
state
.context
.set(keys::INTERNAL_FIDELITY, serde_json::json!(fidelity));
// Thread ID resolution
if let Some(ref edge_data) = incoming {
if let Some(ref tid) = edge_data.thread_id {
state
.context
.set(keys::INTERNAL_THREAD_ID, serde_json::json!(tid));
}
} else if let Some(tid) = gv_node.thread_id() {
state
.context
.set(keys::INTERNAL_THREAD_ID, serde_json::json!(tid));
}
Ok(NodeDecision::Continue)
}
async fn before_attempt(
&self,
ctx: &AttemptContext<'_, WorkflowGraph>,
state: &WfRunState,
) -> CoreResult<WfNodeDecision> {
let gv = ctx.node.inner();
let stage_index = state.stage_index;
// StageStart hook (blocking)
let hook_ctx = HookContext::new(
HookEvent::StageStart,
self.run_id.clone(),
self.graph.name.clone(),
);
let decision = self.run_hook(&hook_ctx).await;
match decision {
HookDecision::Skip { reason } => {
let msg = reason.unwrap_or_else(|| "skipped by hook".into());
return Ok(NodeDecision::Skip(Box::new(Outcome::skipped(&msg))));
}
HookDecision::Block { reason } => {
let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into());
return Err(CoreError::blocked(msg));
}
_ => {}
}
// Emit StageStarted event
self.emitter.emit(&WorkflowRunEvent::StageStarted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
handler_type: gv.handler_type().map(String::from),
script: None,
attempt: ctx.attempt as usize,
max_attempts: ctx.max_attempts as usize,
});
Ok(NodeDecision::Continue)
}
async fn after_attempt(
&self,
ctx: &AttemptResultContext<'_, WorkflowGraph>,
state: &WfRunState,
) -> CoreResult<()> {
if ctx.will_retry {
let gv = ctx.node.inner();
let outcome = &ctx.result.outcome;
let stage_index = state.stage_index;
// Emit StageFailed event
self.emitter.emit(&WorkflowRunEvent::StageFailed {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
failure: outcome.failure.clone().unwrap_or_else(|| {
FailureDetail::new("handler failed", FailureCategory::TransientInfra)
}),
will_retry: true,
});
// Emit StageRetrying event
self.emitter.emit(&WorkflowRunEvent::StageRetrying {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
attempt: ctx.attempt as usize,
max_attempts: ctx.result.max_attempts as usize,
delay_ms: ctx.backoff_delay.map(|d| d.as_millis() as u64).unwrap_or(0),
});
}
Ok(())
}
async fn after_node(
&self,
node: &WorkflowNode,
result: &mut WfNodeResult,
state: &WfRunState,
) -> CoreResult<()> {
let gv = node.inner();
let stage_index = state.stage_index;
let outcome = &mut result.outcome;
// Auto-status override
if gv.auto_status()
&& outcome.status != StageStatus::Success
&& outcome.status != StageStatus::Skipped
{
outcome.status = StageStatus::Success;
outcome.notes =
Some("auto-status: handler completed without writing status".to_string());
}
// Circuit breaker: classify + track failure signatures
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_category {
let sig_hint = outcome
.failure
.as_ref()
.and_then(|f| f.signature.as_deref());
let sig = FailureSignature::new(
&gv.id,
fc,
sig_hint,
outcome.failure.as_ref().map(|f| f.message.as_str()),
);
if fc.is_signature_tracked() {
let mut sigs = self.loop_failure_signatures.lock().unwrap();
let count = sigs.entry(sig.clone()).or_insert(0);
*count += 1;
let limit = self.graph.loop_restart_signature_limit();
if *count >= limit {
return Err(CoreError::Other(format!(
"deterministic failure cycle detected: signature {sig} repeated {count} times (limit {limit})"
)));
}
}
}
// Emit StageCompleted or StageFailed event
let duration_ms = result.duration.as_millis() as u64;
if outcome.status == StageStatus::Fail {
self.emitter.emit(&WorkflowRunEvent::StageFailed {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
failure: outcome.failure.clone().unwrap_or_else(|| {
FailureDetail::new("handler failed", FailureCategory::Deterministic)
}),
will_retry: false,
});
} else {
self.emitter.emit(&WorkflowRunEvent::StageCompleted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
duration_ms,
status: 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: 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 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(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(outcome, &status_path, "node_status");
Ok(())
}
async fn on_edge_selected(
&self,
ctx: &EdgeContext<'_, WorkflowGraph>,
_state: &WfRunState,
) -> CoreResult<EdgeDecision> {
// Capture fidelity/thread from edge for next node
if let Some(ref edge) = ctx.edge {
let gv_edge = edge.inner();
let edge_data = IncomingEdgeData {
fidelity: gv_edge.fidelity().map(String::from),
thread_id: gv_edge.thread_id().map(String::from),
};
*self.incoming_edge_data.lock().unwrap() = Some(edge_data);
}
let outcome = ctx.outcome;
// Emit EdgeSelected event
let label = ctx
.edge
.as_ref()
.and_then(|e| e.inner().label().map(String::from));
let condition = ctx
.edge
.as_ref()
.and_then(|e| e.inner().condition().map(String::from));
self.emitter.emit(&WorkflowRunEvent::EdgeSelected {
from_node: ctx.from.to_string(),
to_node: ctx.to.to_string(),
label,
condition,
reason: ctx.reason.to_string(),
preferred_label: outcome.preferred_label.clone(),
suggested_next_ids: outcome.suggested_next_ids.clone(),
stage_status: outcome.status.to_string(),
is_jump: ctx.is_jump,
});
// EdgeSelected hook (blocking, can override)
let mut hook_ctx = HookContext::new(
HookEvent::EdgeSelected,
self.run_id.clone(),
self.graph.name.clone(),
);
hook_ctx.edge_from = Some(ctx.from.to_string());
hook_ctx.edge_to = Some(ctx.to.to_string());
let decision = self.run_hook(&hook_ctx).await;
match decision {
HookDecision::Override { edge_to } => {
return Ok(EdgeDecision::Override(edge_to));
}
HookDecision::Block { reason } => {
let msg = reason.unwrap_or_else(|| "blocked by EdgeSelected hook".into());
return Err(CoreError::blocked(msg));
}
_ => {}
}
Ok(EdgeDecision::Continue)
}
async fn on_checkpoint(
&self,
node: &WorkflowNode,
result: &WfNodeResult,
next_node_id: Option<&str>,
state: &WfRunState,
) -> CoreResult<()> {
if !self.checkpoint_enabled {
return Ok(());
}
// 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(), result.outcome.clone());
let checkpoint = Checkpoint {
timestamp: chrono::Utc::now(),
current_node: node.id().to_string(),
completed_nodes: state.completed_nodes.clone(),
node_outcomes,
node_retries: state.node_retries.clone(),
context_values: state.context.snapshot(),
next_node_id: next_node_id.map(String::from),
git_commit_sha: None,
node_visits: state.node_visits.clone(),
loop_failure_signatures: self.loop_failure_signatures.lock().unwrap().clone(),
restart_failure_signatures: self.restart_failure_signatures.lock().unwrap().clone(),
};
// Write checkpoint.json
let checkpoint_path = self.run_dir.join("checkpoint.json");
if let Err(e) = checkpoint.save(&checkpoint_path) {
self.emitter.emit(&WorkflowRunEvent::RunNotice {
level: crate::event::RunNoticeLevel::Warn,
code: "checkpoint_disk_save_failed".to_string(),
message: format!("[node: {}] checkpoint save failed: {e}", node.id()),
});
}
// Emit CheckpointCompleted event
let status = result.outcome.status.to_string();
self.emitter.emit(&WorkflowRunEvent::CheckpointCompleted {
node_id: node.id().to_string(),
status,
git_commit_sha: None,
});
Ok(())
}
async fn on_run_end(&self, outcome: &Outcome, state: &WfRunState) {
// If cancelled, skip all events/hooks
if state.cancelled {
return;
}
let duration_ms = self.run_start.elapsed().as_millis() as u64;
if outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess {
// Success path
self.emitter.emit(&WorkflowRunEvent::WorkflowRunCompleted {
duration_ms,
artifact_count: 0,
status: outcome.status.to_string(),
total_cost: None,
final_git_commit_sha: None,
usage: None,
});
// RunComplete hook
let hook_ctx = HookContext::new(
HookEvent::RunComplete,
self.run_id.clone(),
self.graph.name.clone(),
);
let _ = self.run_hook(&hook_ctx).await;
} else {
// Failure path
let error_msg = outcome
.failure
.as_ref()
.map(|f| f.message.clone())
.unwrap_or_else(|| "run failed".to_string());
self.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed {
error: crate::error::FabroError::engine(error_msg.clone()),
duration_ms,
git_commit_sha: None,
});
// RunFailed hook
let mut hook_ctx = HookContext::new(
HookEvent::RunFailed,
self.run_id.clone(),
self.graph.name.clone(),
);
hook_ctx.failure_reason = Some(error_msg);
let _ = self.run_hook(&hook_ctx).await;
}
}
}

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

@ -0,0 +1,11 @@
use async_trait::async_trait;
use fabro_core::lifecycle::RunLifecycle;
use super::super::graph::WorkflowGraph;
/// Sub-lifecycle responsible for artifact collection, offloading, and syncing.
/// Currently a stub — artifact operations are not yet wired through the core adapter.
pub struct ArtifactLifecycle;
#[async_trait]
impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {}

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

@ -0,0 +1,37 @@
use async_trait::async_trait;
use fabro_core::lifecycle::RunLifecycle;
use fabro_core::outcome::NodeResult;
use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use crate::outcome::{StageStatus, StageUsage};
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
/// Sub-lifecycle responsible for auto-status override on nodes with `auto_status=true`.
pub struct AutoStatusLifecycle;
#[async_trait]
impl RunLifecycle<WorkflowGraph> for AutoStatusLifecycle {
async fn after_node(
&self,
node: &WorkflowNode,
result: &mut WfNodeResult,
_state: &WfRunState,
) -> fabro_core::error::Result<()> {
let gv = node.inner();
let outcome = &mut result.outcome;
if gv.auto_status()
&& outcome.status != StageStatus::Success
&& outcome.status != StageStatus::Skipped
{
outcome.status = StageStatus::Success;
outcome.notes =
Some("auto-status: handler completed without writing status".to_string());
}
Ok(())
}
}

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

@ -0,0 +1,102 @@
use std::collections::HashMap;
use std::sync::Mutex;
use async_trait::async_trait;
use fabro_core::error::{CoreError, Result as CoreResult};
use fabro_core::lifecycle::RunLifecycle;
use fabro_core::outcome::NodeResult;
use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use crate::error::FailureSignature;
use crate::outcome::{StageStatus, StageUsage};
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
/// Sub-lifecycle responsible for tracking failure signatures and tripping the
/// circuit breaker when deterministic failure cycles are detected.
pub struct CircuitBreakerLifecycle {
loop_failure_signatures: Mutex<HashMap<FailureSignature, usize>>,
restart_failure_signatures: Mutex<HashMap<FailureSignature, usize>>,
loop_restart_signature_limit: usize,
}
impl CircuitBreakerLifecycle {
pub fn new(loop_restart_signature_limit: usize) -> Self {
Self {
loop_failure_signatures: Mutex::new(HashMap::new()),
restart_failure_signatures: Mutex::new(HashMap::new()),
loop_restart_signature_limit,
}
}
/// Restore circuit breaker state from a checkpoint (for resume).
pub fn restore(
&self,
loop_sigs: HashMap<FailureSignature, usize>,
restart_sigs: HashMap<FailureSignature, usize>,
) {
*self.loop_failure_signatures.lock().unwrap() = loop_sigs;
*self.restart_failure_signatures.lock().unwrap() = restart_sigs;
}
/// Snapshot current state for checkpoint building.
pub fn snapshot(
&self,
) -> (
HashMap<FailureSignature, usize>,
HashMap<FailureSignature, usize>,
) {
let loop_sigs = self.loop_failure_signatures.lock().unwrap().clone();
let restart_sigs = self.restart_failure_signatures.lock().unwrap().clone();
(loop_sigs, restart_sigs)
}
}
#[async_trait]
impl RunLifecycle<WorkflowGraph> for CircuitBreakerLifecycle {
async fn after_node(
&self,
node: &WorkflowNode,
result: &mut WfNodeResult,
_state: &WfRunState,
) -> CoreResult<()> {
let gv = node.inner();
let outcome = &result.outcome;
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_category {
let sig_hint = outcome
.failure
.as_ref()
.and_then(|f| f.signature.as_deref());
let sig = FailureSignature::new(
&gv.id,
fc,
sig_hint,
outcome.failure.as_ref().map(|f| f.message.as_str()),
);
if fc.is_signature_tracked() {
let mut sigs = self.loop_failure_signatures.lock().unwrap();
let count = sigs.entry(sig.clone()).or_insert(0);
*count += 1;
let limit = self.loop_restart_signature_limit;
if *count >= limit {
return Err(CoreError::Other(format!(
"deterministic failure cycle detected: signature {sig} repeated {count} times (limit {limit})"
)));
}
}
}
Ok(())
}
}

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

@ -0,0 +1,89 @@
use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use fabro_core::graph::NodeSpec;
use fabro_core::lifecycle::RunLifecycle;
use fabro_core::outcome::NodeResult;
use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use super::circuit_breaker::CircuitBreakerLifecycle;
use crate::checkpoint::Checkpoint;
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::outcome::StageUsage;
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
/// Sub-lifecycle responsible for writing run state to disk (node status, checkpoints).
pub struct DiskLifecycle {
pub run_dir: PathBuf,
pub run_id: String,
pub emitter: Arc<EventEmitter>,
pub circuit_breaker: Arc<CircuitBreakerLifecycle>,
pub checkpoint_enabled: bool,
}
#[async_trait]
impl RunLifecycle<WorkflowGraph> for DiskLifecycle {
async fn after_node(
&self,
node: &WorkflowNode,
result: &mut WfNodeResult,
_state: &WfRunState,
) -> fabro_core::error::Result<()> {
let gv = node.inner();
let outcome = &result.outcome;
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(outcome, &status_path, "node_status");
Ok(())
}
async fn on_checkpoint(
&self,
node: &WorkflowNode,
result: &WfNodeResult,
next_node_id: Option<&str>,
state: &WfRunState,
) -> fabro_core::error::Result<()> {
if !self.checkpoint_enabled {
return Ok(());
}
let (loop_sigs, restart_sigs) = self.circuit_breaker.snapshot();
// Build checkpoint from state
let mut node_outcomes = state.node_outcomes.clone();
node_outcomes.insert(node.id().to_string(), result.outcome.clone());
let checkpoint = Checkpoint {
timestamp: chrono::Utc::now(),
current_node: node.id().to_string(),
completed_nodes: state.completed_nodes.clone(),
node_outcomes,
node_retries: state.node_retries.clone(),
context_values: state.context.snapshot(),
next_node_id: next_node_id.map(String::from),
git_commit_sha: None,
node_visits: state.node_visits.clone(),
loop_failure_signatures: loop_sigs,
restart_failure_signatures: restart_sigs,
};
let checkpoint_path = self.run_dir.join("checkpoint.json");
if let Err(e) = checkpoint.save(&checkpoint_path) {
self.emitter.emit(&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Warn,
code: "checkpoint_disk_save_failed".to_string(),
message: format!("[node: {}] checkpoint save failed: {e}", node.id()),
});
}
Ok(())
}
}

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

@ -0,0 +1,270 @@
use std::sync::{Arc, Mutex};
use std::time::Instant;
use async_trait::async_trait;
use fabro_core::graph::NodeSpec;
use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, EdgeContext, RunLifecycle};
use fabro_core::outcome::NodeResult;
use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::outcome::{FailureCategory, FailureDetail, Outcome, StageStatus, StageUsage};
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
/// Sub-lifecycle responsible for emitting workflow run events.
pub struct EventLifecycle {
pub emitter: Arc<EventEmitter>,
pub graph_name: String,
pub run_id: String,
pub run_start: Mutex<Instant>,
/// Set in on_edge_selected when loop_restart approved; emitted+cleared in on_run_start.
pub restarted_from: Arc<Mutex<Option<(String, String)>>>,
// Config for WorkflowRunStarted payload
pub base_sha: Option<String>,
pub run_branch: Option<String>,
pub worktree_dir: Option<String>,
pub goal: Option<String>,
}
#[async_trait]
impl RunLifecycle<WorkflowGraph> for EventLifecycle {
async fn on_run_start(
&self,
_graph: &WorkflowGraph,
_state: &WfRunState,
) -> fabro_core::error::Result<()> {
// If restarted_from is Some, emit LoopRestart and clear it
{
let mut restarted = self.restarted_from.lock().unwrap();
if let Some((from_node, to_node)) = restarted.take() {
self.emitter
.emit(&WorkflowRunEvent::LoopRestart { from_node, to_node });
}
}
// Reset run_start for duration measurement
*self.run_start.lock().unwrap() = Instant::now();
// Emit WorkflowRunStarted
self.emitter.emit(&WorkflowRunEvent::WorkflowRunStarted {
name: self.graph_name.clone(),
run_id: self.run_id.clone(),
base_sha: self.base_sha.clone(),
run_branch: self.run_branch.clone(),
worktree_dir: self.worktree_dir.clone(),
goal: self.goal.clone(),
});
Ok(())
}
async fn on_terminal_reached(
&self,
node: &WorkflowNode,
goal_gates_passed: bool,
state: &WfRunState,
) {
if !goal_gates_passed {
return;
}
let gv = node.inner();
let stage_index = state.stage_index;
self.emitter.emit(&WorkflowRunEvent::StageStarted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
handler_type: gv.handler_type().map(String::from),
script: None,
attempt: 1,
max_attempts: 1,
});
self.emitter.emit(&WorkflowRunEvent::StageCompleted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
duration_ms: 0,
status: StageStatus::Success.to_string(),
preferred_label: None,
suggested_next_ids: Vec::new(),
usage: None,
failure: None,
notes: None,
files_touched: Vec::new(),
attempt: 1,
max_attempts: 1,
});
}
async fn before_attempt(
&self,
ctx: &AttemptContext<'_, WorkflowGraph>,
state: &WfRunState,
) -> fabro_core::error::Result<fabro_core::lifecycle::NodeDecision<Option<StageUsage>>> {
let gv = ctx.node.inner();
self.emitter.emit(&WorkflowRunEvent::StageStarted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: state.stage_index,
handler_type: gv.handler_type().map(String::from),
script: None,
attempt: ctx.attempt as usize,
max_attempts: ctx.max_attempts as usize,
});
Ok(fabro_core::lifecycle::NodeDecision::Continue)
}
async fn after_attempt(
&self,
ctx: &AttemptResultContext<'_, WorkflowGraph>,
state: &WfRunState,
) -> fabro_core::error::Result<()> {
if ctx.will_retry {
let gv = ctx.node.inner();
let outcome = &ctx.result.outcome;
let stage_index = state.stage_index;
self.emitter.emit(&WorkflowRunEvent::StageFailed {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
failure: outcome.failure.clone().unwrap_or_else(|| {
FailureDetail::new("handler failed", FailureCategory::TransientInfra)
}),
will_retry: true,
});
self.emitter.emit(&WorkflowRunEvent::StageRetrying {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
attempt: ctx.attempt as usize,
max_attempts: ctx.result.max_attempts as usize,
delay_ms: ctx.backoff_delay.map(|d| d.as_millis() as u64).unwrap_or(0),
});
}
Ok(())
}
async fn after_node(
&self,
node: &WorkflowNode,
result: &mut WfNodeResult,
state: &WfRunState,
) -> fabro_core::error::Result<()> {
let outcome = &result.outcome;
// Skip events for Skipped nodes
if outcome.status == StageStatus::Skipped {
return Ok(());
}
let gv = node.inner();
let stage_index = state.stage_index;
let duration_ms = result.duration.as_millis() as u64;
if outcome.status == StageStatus::Fail {
self.emitter.emit(&WorkflowRunEvent::StageFailed {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
failure: outcome.failure.clone().unwrap_or_else(|| {
FailureDetail::new("handler failed", FailureCategory::Deterministic)
}),
will_retry: false,
});
} else {
self.emitter.emit(&WorkflowRunEvent::StageCompleted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
duration_ms,
status: 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: outcome.notes.clone(),
files_touched: outcome.files_touched.clone(),
attempt: result.attempts as usize,
max_attempts: result.max_attempts as usize,
});
}
Ok(())
}
async fn on_edge_selected(
&self,
ctx: &EdgeContext<'_, WorkflowGraph>,
_state: &WfRunState,
) -> fabro_core::error::Result<fabro_core::lifecycle::EdgeDecision> {
let outcome = ctx.outcome;
let label = ctx
.edge
.as_ref()
.and_then(|e| e.inner().label().map(String::from));
let condition = ctx
.edge
.as_ref()
.and_then(|e| e.inner().condition().map(String::from));
self.emitter.emit(&WorkflowRunEvent::EdgeSelected {
from_node: ctx.from.to_string(),
to_node: ctx.to.to_string(),
label,
condition,
reason: ctx.reason.to_string(),
preferred_label: outcome.preferred_label.clone(),
suggested_next_ids: outcome.suggested_next_ids.clone(),
stage_status: outcome.status.to_string(),
is_jump: ctx.is_jump,
});
Ok(fabro_core::lifecycle::EdgeDecision::Continue)
}
async fn on_checkpoint(
&self,
node: &WorkflowNode,
result: &WfNodeResult,
_next_node_id: Option<&str>,
_state: &WfRunState,
) -> fabro_core::error::Result<()> {
let status = result.outcome.status.to_string();
self.emitter.emit(&WorkflowRunEvent::CheckpointCompleted {
node_id: node.id().to_string(),
status,
git_commit_sha: None,
});
Ok(())
}
async fn on_run_end(&self, outcome: &Outcome, state: &WfRunState) {
if state.cancelled {
return;
}
let duration_ms = self.run_start.lock().unwrap().elapsed().as_millis() as u64;
if outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess {
self.emitter.emit(&WorkflowRunEvent::WorkflowRunCompleted {
duration_ms,
artifact_count: 0,
status: outcome.status.to_string(),
total_cost: None,
final_git_commit_sha: None,
usage: None,
});
} else {
let error_msg = outcome
.failure
.as_ref()
.map(|f| f.message.clone())
.unwrap_or_else(|| "run failed".to_string());
self.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed {
error: crate::error::FabroError::engine(error_msg),
duration_ms,
git_commit_sha: None,
});
}
}
}

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

@ -0,0 +1,135 @@
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use fabro_core::graph::NodeSpec;
use fabro_core::lifecycle::{EdgeContext, NodeDecision, RunLifecycle};
use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use crate::context::keys;
use crate::outcome::StageUsage;
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeDecision = NodeDecision<Option<StageUsage>>;
/// Data captured from an edge selection to pass to the next node's before_node.
#[derive(Debug, Clone)]
struct IncomingEdgeData {
fidelity: Option<String>,
thread_id: Option<String>,
}
/// Sub-lifecycle responsible for fidelity/thread resolution and context key setup.
pub struct FidelityLifecycle {
pub graph: Arc<fabro_graphviz::graph::types::Graph>,
incoming_edge_data: Mutex<Option<IncomingEdgeData>>,
/// True on the first node after checkpoint resume when prior fidelity was Full.
degrade_fidelity_on_resume: Mutex<bool>,
}
impl FidelityLifecycle {
pub fn new(graph: Arc<fabro_graphviz::graph::types::Graph>) -> Self {
Self {
graph,
incoming_edge_data: Mutex::new(None),
degrade_fidelity_on_resume: Mutex::new(false),
}
}
pub fn set_degrade_fidelity_on_resume(&self, flag: bool) {
*self.degrade_fidelity_on_resume.lock().unwrap() = flag;
}
}
#[async_trait]
impl RunLifecycle<WorkflowGraph> for FidelityLifecycle {
async fn on_run_start(
&self,
_graph: &WorkflowGraph,
_state: &WfRunState,
) -> fabro_core::error::Result<()> {
// Clear incoming edge data (restart target must not inherit pre-restart edge)
*self.incoming_edge_data.lock().unwrap() = None;
Ok(())
}
async fn before_node(
&self,
node: &WorkflowNode,
state: &WfRunState,
) -> fabro_core::error::Result<WfNodeDecision> {
let incoming = self.incoming_edge_data.lock().unwrap().take();
let gv_node = node.inner();
// Set context keys for the current node
let visits = state.node_visits.get(node.id()).copied().unwrap_or(0);
state
.context
.set(keys::CURRENT_NODE, serde_json::json!(node.id()));
state
.context
.set(keys::INTERNAL_NODE_VISIT_COUNT, serde_json::json!(visits));
// Fidelity resolution: edge → node → graph default → compact
let fidelity = if let Some(ref edge_data) = incoming {
edge_data
.fidelity
.as_deref()
.or(gv_node.fidelity())
.unwrap_or("compact")
.to_string()
} else {
gv_node.fidelity().unwrap_or("compact").to_string()
};
// Fidelity degradation on resume
let fidelity = {
let mut degrade = self.degrade_fidelity_on_resume.lock().unwrap();
if *degrade {
*degrade = false;
let parsed: keys::Fidelity = fidelity.parse().unwrap_or_default();
parsed.degraded().to_string()
} else {
fidelity
}
};
state
.context
.set(keys::INTERNAL_FIDELITY, serde_json::json!(fidelity));
// Thread ID resolution: edge → node → graph default → previous node
if let Some(ref edge_data) = incoming {
if let Some(ref tid) = edge_data.thread_id {
state
.context
.set(keys::INTERNAL_THREAD_ID, serde_json::json!(tid));
}
} else if let Some(tid) = gv_node.thread_id() {
state
.context
.set(keys::INTERNAL_THREAD_ID, serde_json::json!(tid));
}
Ok(NodeDecision::Continue)
}
async fn on_edge_selected(
&self,
ctx: &EdgeContext<'_, WorkflowGraph>,
_state: &WfRunState,
) -> fabro_core::error::Result<fabro_core::lifecycle::EdgeDecision> {
// Capture fidelity/thread from edge for next node
if let Some(ref edge) = ctx.edge {
let gv_edge = edge.inner();
let edge_data = IncomingEdgeData {
fidelity: gv_edge.fidelity().map(String::from),
thread_id: gv_edge.thread_id().map(String::from),
};
*self.incoming_edge_data.lock().unwrap() = Some(edge_data);
}
Ok(fabro_core::lifecycle::EdgeDecision::Continue)
}
}

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

@ -0,0 +1,11 @@
use async_trait::async_trait;
use fabro_core::lifecycle::RunLifecycle;
use super::super::graph::WorkflowGraph;
/// Sub-lifecycle responsible for git operations (checkpoint commits, pushes, diffs).
/// Currently a stub — git operations are not yet wired through the core adapter.
pub struct GitLifecycle;
#[async_trait]
impl RunLifecycle<WorkflowGraph> for GitLifecycle {}

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

@ -0,0 +1,161 @@
use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use fabro_core::error::{CoreError, Result as CoreResult};
use fabro_core::lifecycle::{
AttemptContext, EdgeContext, EdgeDecision, NodeDecision, RunLifecycle,
};
use fabro_core::outcome::NodeResult;
use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use crate::engine::set_hook_node;
use crate::outcome::{Outcome, StageStatus, StageUsage};
use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner};
use fabro_sandbox::Sandbox;
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
type WfNodeDecision = NodeDecision<Option<StageUsage>>;
/// Sub-lifecycle responsible for running workflow hooks.
pub struct HookLifecycle {
pub hook_runner: Option<Arc<HookRunner>>,
pub sandbox: Arc<dyn Sandbox>,
pub run_dir: PathBuf,
pub run_id: String,
pub graph_name: String,
}
impl HookLifecycle {
async fn run_hook(&self, hook_ctx: &HookContext) -> HookDecision {
let Some(ref runner) = self.hook_runner else {
return HookDecision::Proceed;
};
runner
.run(hook_ctx, self.sandbox.clone(), Some(&self.run_dir))
.await
}
}
#[async_trait]
impl RunLifecycle<WorkflowGraph> for HookLifecycle {
async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> {
let hook_ctx = HookContext::new(
HookEvent::RunStart,
self.run_id.clone(),
self.graph_name.clone(),
);
let decision = self.run_hook(&hook_ctx).await;
if let HookDecision::Block { reason } = decision {
let msg = reason.unwrap_or_else(|| "blocked by RunStart hook".into());
return Err(CoreError::blocked(msg));
}
Ok(())
}
async fn before_attempt(
&self,
ctx: &AttemptContext<'_, WorkflowGraph>,
_state: &WfRunState,
) -> CoreResult<WfNodeDecision> {
let gv = ctx.node.inner();
let mut hook_ctx = HookContext::new(
HookEvent::StageStart,
self.run_id.clone(),
self.graph_name.clone(),
);
set_hook_node(&mut hook_ctx, gv);
hook_ctx.attempt = Some(ctx.attempt as usize);
hook_ctx.max_attempts = Some(ctx.max_attempts as usize);
let decision = self.run_hook(&hook_ctx).await;
match decision {
HookDecision::Skip { reason } => {
let msg = reason.unwrap_or_else(|| "skipped by hook".into());
Ok(NodeDecision::Skip(Box::new(Outcome::skipped(&msg))))
}
HookDecision::Block { reason } => {
let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into());
Err(CoreError::blocked(msg))
}
_ => Ok(NodeDecision::Continue),
}
}
async fn after_node(
&self,
_node: &WorkflowNode,
result: &mut WfNodeResult,
_state: &WfRunState,
) -> CoreResult<()> {
let outcome = &result.outcome;
// Skip hooks for Skipped nodes
if outcome.status == StageStatus::Skipped {
return Ok(());
}
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(outcome.status.to_string());
let _ = self.run_hook(&hook_ctx).await;
Ok(())
}
async fn on_edge_selected(
&self,
ctx: &EdgeContext<'_, WorkflowGraph>,
_state: &WfRunState,
) -> CoreResult<EdgeDecision> {
let mut hook_ctx = HookContext::new(
HookEvent::EdgeSelected,
self.run_id.clone(),
self.graph_name.clone(),
);
hook_ctx.edge_from = Some(ctx.from.to_string());
hook_ctx.edge_to = Some(ctx.to.to_string());
let decision = self.run_hook(&hook_ctx).await;
match decision {
HookDecision::Override { edge_to } => Ok(EdgeDecision::Override(edge_to)),
HookDecision::Block { reason } => {
let msg = reason.unwrap_or_else(|| "blocked by EdgeSelected hook".into());
Err(CoreError::blocked(msg))
}
_ => Ok(EdgeDecision::Continue),
}
}
async fn on_run_end(&self, outcome: &Outcome, state: &WfRunState) {
if state.cancelled {
return;
}
if outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess {
let hook_ctx = HookContext::new(
HookEvent::RunComplete,
self.run_id.clone(),
self.graph_name.clone(),
);
let _ = self.run_hook(&hook_ctx).await;
} else {
let error_msg = outcome
.failure
.as_ref()
.map(|f| f.message.clone())
.unwrap_or_else(|| "run failed".to_string());
let mut hook_ctx = HookContext::new(
HookEvent::RunFailed,
self.run_id.clone(),
self.graph_name.clone(),
);
hook_ctx.failure_reason = Some(error_msg);
let _ = self.run_hook(&hook_ctx).await;
}
}
}

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

@ -0,0 +1,252 @@
pub mod artifact;
pub mod auto_status;
pub mod circuit_breaker;
pub mod disk;
pub mod event;
pub mod fidelity;
pub mod git;
pub mod hook;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Instant;
use async_trait::async_trait;
use fabro_core::error::Result as CoreResult;
use fabro_core::lifecycle::{
AttemptContext, AttemptResultContext, EdgeContext, EdgeDecision, NodeDecision, RunLifecycle,
};
use fabro_core::outcome::NodeResult;
use fabro_core::state::RunState;
use super::graph::WorkflowGraph;
use super::WorkflowNode;
use crate::event::EventEmitter;
use crate::outcome::{Outcome, StageUsage};
use fabro_hooks::HookRunner;
use fabro_sandbox::Sandbox;
use self::artifact::ArtifactLifecycle;
use self::auto_status::AutoStatusLifecycle;
use self::circuit_breaker::CircuitBreakerLifecycle;
use self::disk::DiskLifecycle;
use self::event::EventLifecycle;
use self::fidelity::FidelityLifecycle;
use self::git::GitLifecycle;
use self::hook::HookLifecycle;
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
type WfNodeDecision = NodeDecision<Option<StageUsage>>;
/// Orchestrates all sub-lifecycles with explicit per-callback ordering.
/// Implements `RunLifecycle<WorkflowGraph>` by delegating to focused structs.
pub struct WorkflowLifecycle {
event: EventLifecycle,
hook: HookLifecycle,
fidelity: FidelityLifecycle,
auto_status: AutoStatusLifecycle,
circuit_breaker: Arc<CircuitBreakerLifecycle>,
disk: DiskLifecycle,
#[allow(dead_code)] // stub — will be wired when git operations move to core adapter
git: GitLifecycle,
#[allow(dead_code)] // stub — will be wired when artifact operations move to core adapter
artifact: ArtifactLifecycle,
/// Set in on_edge_selected when loop_restart approved; read+cleared by EventLifecycle::on_run_start
restarted_from: Arc<Mutex<Option<(String, String)>>>,
}
impl WorkflowLifecycle {
#[allow(clippy::too_many_arguments)]
pub fn new(
emitter: Arc<EventEmitter>,
hook_runner: Option<Arc<HookRunner>>,
sandbox: Arc<dyn Sandbox>,
graph: Arc<fabro_graphviz::graph::types::Graph>,
run_dir: PathBuf,
run_id: String,
_dry_run: bool,
_labels: HashMap<String, String>,
) -> Self {
let restarted_from: Arc<Mutex<Option<(String, String)>>> = Arc::new(Mutex::new(None));
let loop_restart_signature_limit = graph.loop_restart_signature_limit();
let circuit_breaker = Arc::new(CircuitBreakerLifecycle::new(loop_restart_signature_limit));
let event = EventLifecycle {
emitter: Arc::clone(&emitter),
graph_name: graph.name.clone(),
run_id: run_id.clone(),
run_start: Mutex::new(Instant::now()),
restarted_from: Arc::clone(&restarted_from),
base_sha: None,
run_branch: None,
worktree_dir: None,
goal: None,
};
let hook = HookLifecycle {
hook_runner,
sandbox: Arc::clone(&sandbox),
run_dir: run_dir.clone(),
run_id: run_id.clone(),
graph_name: graph.name.clone(),
};
let fidelity = FidelityLifecycle::new(Arc::clone(&graph));
let disk = DiskLifecycle {
run_dir: run_dir.clone(),
run_id: run_id.clone(),
emitter: Arc::clone(&emitter),
circuit_breaker: Arc::clone(&circuit_breaker),
checkpoint_enabled: true,
};
Self {
event,
hook,
fidelity,
auto_status: AutoStatusLifecycle,
circuit_breaker,
disk,
git: GitLifecycle,
artifact: ArtifactLifecycle,
restarted_from,
}
}
/// Restore circuit breaker state from a checkpoint (for resume).
pub fn restore_circuit_breaker(
&self,
loop_sigs: HashMap<crate::error::FailureSignature, usize>,
restart_sigs: HashMap<crate::error::FailureSignature, usize>,
) {
self.circuit_breaker.restore(loop_sigs, restart_sigs);
}
/// Set the fidelity degradation flag for checkpoint resume.
pub fn set_degrade_fidelity_on_resume(&self, flag: bool) {
self.fidelity.set_degrade_fidelity_on_resume(flag);
}
}
#[async_trait]
impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
async fn on_run_start(&self, graph: &WorkflowGraph, state: &WfRunState) -> CoreResult<()> {
// Reset restart-scoped state
self.fidelity.on_run_start(graph, state).await?;
// Observable callbacks
self.event.on_run_start(graph, state).await?;
self.hook.on_run_start(graph, state).await?;
Ok(())
}
async fn on_terminal_reached(
&self,
node: &WorkflowNode,
goal_gates_passed: bool,
state: &WfRunState,
) {
self.event
.on_terminal_reached(node, goal_gates_passed, state)
.await;
}
async fn before_node(
&self,
node: &WorkflowNode,
state: &WfRunState,
) -> CoreResult<WfNodeDecision> {
self.fidelity.before_node(node, state).await
}
async fn before_attempt(
&self,
ctx: &AttemptContext<'_, WorkflowGraph>,
state: &WfRunState,
) -> CoreResult<WfNodeDecision> {
// Hook first (can skip/block)
match self.hook.before_attempt(ctx, state).await? {
NodeDecision::Continue => {}
decision => return Ok(decision),
}
// Event emission
self.event.before_attempt(ctx, state).await?;
Ok(NodeDecision::Continue)
}
async fn after_attempt(
&self,
ctx: &AttemptResultContext<'_, WorkflowGraph>,
state: &WfRunState,
) -> CoreResult<()> {
self.event.after_attempt(ctx, state).await?;
Ok(())
}
async fn after_node(
&self,
node: &WorkflowNode,
result: &mut WfNodeResult,
state: &WfRunState,
) -> CoreResult<()> {
self.auto_status.after_node(node, result, state).await?;
self.circuit_breaker.after_node(node, result, state).await?;
self.event.after_node(node, result, state).await?;
self.hook.after_node(node, result, state).await?;
self.disk.after_node(node, result, state).await?;
Ok(())
}
async fn on_edge_selected(
&self,
ctx: &EdgeContext<'_, WorkflowGraph>,
state: &WfRunState,
) -> CoreResult<EdgeDecision> {
// Fidelity captures edge data
self.fidelity.on_edge_selected(ctx, state).await?;
// Event always fires first
self.event.on_edge_selected(ctx, state).await?;
// Hook can override/block
match self.hook.on_edge_selected(ctx, state).await? {
EdgeDecision::Continue => {
// If loop_restart edge approved by hook, mark for LoopRestart emission
if let Some(ref edge) = ctx.edge {
if edge.inner().loop_restart() {
*self.restarted_from.lock().unwrap() =
Some((ctx.from.to_string(), ctx.to.to_string()));
}
}
Ok(EdgeDecision::Continue)
}
decision => Ok(decision),
}
}
async fn on_checkpoint(
&self,
node: &WorkflowNode,
result: &WfNodeResult,
next_node_id: Option<&str>,
state: &WfRunState,
) -> CoreResult<()> {
self.disk
.on_checkpoint(node, result, next_node_id, state)
.await?;
self.event
.on_checkpoint(node, result, next_node_id, state)
.await?;
Ok(())
}
async fn on_run_end(&self, outcome: &Outcome, state: &WfRunState) {
if state.cancelled {
return;
}
self.event.on_run_end(outcome, state).await;
self.hook.on_run_end(outcome, state).await;
}
}

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

@ -1432,12 +1432,28 @@ impl WorkflowRunEngine {
let graph_arc = std::sync::Arc::new(graph.clone());
let wf_graph = crate::core_adapter::WorkflowGraph(Arc::clone(&graph_arc));
// Populate git_state for handlers (parallel, fan_in) when checkpointing is active
let git_state = if config.git_checkpoint_enabled {
config.base_sha.as_ref().map(|base_sha| {
Arc::new(GitState {
run_id: config.run_id.clone(),
base_sha: base_sha.clone(),
run_branch: config.run_branch.clone(),
meta_branch: config.meta_branch.clone(),
checkpoint_exclude_globs: config.checkpoint_exclude_globs.clone(),
git_author: config.git_author.clone(),
})
})
} else {
None
};
// Build a shared EngineServices for the handler
let shared_services = std::sync::Arc::new(EngineServices {
registry: Arc::clone(&self.services.registry),
emitter: Arc::clone(&self.services.emitter),
sandbox: Arc::clone(&self.services.sandbox),
git_state: std::sync::RwLock::new(None),
git_state: std::sync::RwLock::new(git_state),
hook_runner: self.services.hook_runner.clone(),
env: self.services.env.clone(),
dry_run: self.services.dry_run,
@ -1447,6 +1463,7 @@ impl WorkflowRunEngine {
let handler = std::sync::Arc::new(crate::core_adapter::WorkflowNodeHandler {
services: shared_services,
run_dir: config.run_dir.clone(),
graph: Arc::clone(&graph_arc),
});
// Build lifecycle
@ -1479,14 +1496,30 @@ impl WorkflowRunEngine {
}
s.completed_nodes = cp.completed_nodes.clone();
s.node_retries = cp.node_retries.clone();
s.node_visits = cp.node_visits.clone();
// Restore node_visits; reconstruct from completed_nodes for old checkpoints
if cp.node_visits.is_empty() {
for id in &cp.completed_nodes {
*s.node_visits.entry(id.clone()).or_insert(0) += 1;
}
} else {
s.node_visits = cp.node_visits.clone();
}
// Restore node outcomes
for (k, v) in &cp.node_outcomes {
s.node_outcomes.insert(k.clone(), v.clone());
}
// Set start node to the checkpoint's next_node_id
// Set stage_index to number of completed nodes
s.stage_index = cp.completed_nodes.len();
// Use stored next_node_id if available, otherwise fall back
if let Some(ref next) = cp.next_node_id {
s.current_node_id = next.clone();
} else {
let edges = graph.outgoing_edges(&cp.current_node);
if let Some(edge) = edges.first() {
s.current_node_id = edge.to.clone();
} else {
s.current_node_id = cp.current_node.clone();
}
}
s
} else if let Some(seed) = seed_context {
@ -1581,9 +1614,9 @@ impl WorkflowRunEngine {
// Convert result
match result {
Ok(core_outcome) => {
// Outcome is now the wf type directly — no conversion needed
let ctx = Context::new();
Ok((core_outcome, final_state)) => {
// Extract the executor's final context so callers see all state
let ctx = Context::from_values(final_state.context.snapshot());
Ok((core_outcome, ctx))
}
Err(fabro_core::CoreError::StallTimeout { node_id }) => {