From 5e6eebaff805f22a09c619d91523fddb6fb1fd5e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 24 Mar 2026 14:46:10 -0400 Subject: [PATCH] Decompose monolithic WorkflowLifecycle into 8 focused sub-lifecycles 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) --- lib/crates/fabro-core/src/executor.rs | 47 +- lib/crates/fabro-core/src/state.rs | 11 + .../src/core_adapter/handler.rs | 32 +- .../src/core_adapter/lifecycle.rs | 563 ------------------ .../src/core_adapter/lifecycle/artifact.rs | 11 + .../src/core_adapter/lifecycle/auto_status.rs | 37 ++ .../core_adapter/lifecycle/circuit_breaker.rs | 102 ++++ .../src/core_adapter/lifecycle/disk.rs | 89 +++ .../src/core_adapter/lifecycle/event.rs | 270 +++++++++ .../src/core_adapter/lifecycle/fidelity.rs | 135 +++++ .../src/core_adapter/lifecycle/git.rs | 11 + .../src/core_adapter/lifecycle/hook.rs | 161 +++++ .../src/core_adapter/lifecycle/mod.rs | 252 ++++++++ lib/crates/fabro-workflows/src/engine.rs | 45 +- 14 files changed, 1167 insertions(+), 599 deletions(-) delete mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/auto_status.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/disk.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs create mode 100644 lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs diff --git a/lib/crates/fabro-core/src/executor.rs b/lib/crates/fabro-core/src/executor.rs index ea2774c84..a63bc03fa 100644 --- a/lib/crates/fabro-core/src/executor.rs +++ b/lib/crates/fabro-core/src/executor.rs @@ -80,7 +80,11 @@ impl ExecutorBuilder { } impl Executor { - pub async fn run(&self, graph: &G, mut state: RunState) -> Result> { + pub async fn run( + &self, + graph: &G, + mut state: RunState, + ) -> Result<(Outcome, RunState)> { self.lifecycle.on_run_start(graph, &state).await?; loop { @@ -109,7 +113,7 @@ impl Executor { .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 Executor { failed_node_id )); self.lifecycle.on_run_end(&outcome, &state).await; - return Ok(outcome); + return Ok((outcome, state)); } } } @@ -220,7 +224,7 @@ impl Executor { 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>) .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>).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> ) .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> ) .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>) .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>).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>) .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> ) .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>) .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>) .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>) .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>) .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>) .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>).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>) .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); } diff --git a/lib/crates/fabro-core/src/state.rs b/lib/crates/fabro-core/src/state.rs index cd2f1e8d3..8ba97f9ac 100644 --- a/lib/crates/fabro-core/src/state.rs +++ b/lib/crates/fabro-core/src/state.rs @@ -5,6 +5,17 @@ use crate::error::Result; use crate::graph::{Graph, NodeSpec}; use crate::outcome::{NodeResult, Outcome, OutcomeMeta}; +impl std::fmt::Debug for RunState { + 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 { pub context: Context, pub current_node_id: String, diff --git a/lib/crates/fabro-workflows/src/core_adapter/handler.rs b/lib/crates/fabro-workflows/src/core_adapter/handler.rs index e0fd7940a..37d0c7f9a 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/handler.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/handler.rs @@ -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 = - 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, pub run_dir: PathBuf, + pub graph: Arc, } #[async_trait] @@ -33,13 +33,16 @@ impl NodeHandler for WorkflowNodeHandler { async fn execute( &self, node: &WorkflowNode, - _context: &CoreContext, + context: &CoreContext, _graph: &WorkflowGraph, ) -> CoreResult { 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 for WorkflowNodeHandler { handler, gv_node, &wf_context, - &STUB_GRAPH, + &self.graph, &run_dir, &self.services, ); @@ -72,6 +75,15 @@ impl NodeHandler 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 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); } } diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs deleted file mode 100644 index 727312434..000000000 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs +++ /dev/null @@ -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, - thread_id: Option, -} - -/// Implements the full RunLifecycle for fabro-workflows, mapping all domain -/// concerns (events, hooks, git, disk I/O, fidelity, circuit breaker, etc.) -/// into fabro-core lifecycle callbacks. -pub struct WorkflowLifecycle { - pub emitter: Arc, - pub hook_runner: Option>, - pub sandbox: Arc, - pub graph: Arc, - pub run_dir: PathBuf, - pub run_id: String, - pub run_start: Instant, - pub labels: HashMap, - // Circuit breaker state - loop_failure_signatures: Mutex>, - restart_failure_signatures: Mutex>, - // Edge data for next node - incoming_edge_data: Mutex>, - // Config flags - pub dry_run: bool, - pub checkpoint_enabled: bool, -} - -impl WorkflowLifecycle { - #[allow(clippy::too_many_arguments)] - pub fn new( - emitter: Arc, - hook_runner: Option>, - sandbox: Arc, - graph: Arc, - run_dir: PathBuf, - run_id: String, - dry_run: bool, - labels: HashMap, - ) -> Self { - Self { - emitter, - hook_runner, - sandbox, - graph, - run_dir, - run_id, - run_start: Instant::now(), - labels, - loop_failure_signatures: Mutex::new(HashMap::new()), - restart_failure_signatures: Mutex::new(HashMap::new()), - incoming_edge_data: Mutex::new(None), - dry_run, - checkpoint_enabled: true, - } - } - - /// Restore circuit breaker state from a checkpoint (for resume). - pub fn restore_circuit_breaker( - &self, - loop_sigs: HashMap, - restart_sigs: HashMap, - ) { - *self.loop_failure_signatures.lock().unwrap() = loop_sigs; - *self.restart_failure_signatures.lock().unwrap() = restart_sigs; - } - - async fn run_hook(&self, hook_ctx: &HookContext) -> HookDecision { - let Some(ref runner) = self.hook_runner else { - return HookDecision::Proceed; - }; - runner - .run(hook_ctx, self.sandbox.clone(), Some(&self.run_dir)) - .await - } -} - -type WfRunState = RunState>; -type WfNodeResult = NodeResult>; -type WfNodeDecision = NodeDecision>; - -#[async_trait] -impl RunLifecycle 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 { - // 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 { - 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 { - // 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 = 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; - } - } -} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs new file mode 100644 index 000000000..8471ddde6 --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs @@ -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 for ArtifactLifecycle {} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/auto_status.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/auto_status.rs new file mode 100644 index 000000000..a58df4f0a --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/auto_status.rs @@ -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>; +type WfNodeResult = NodeResult>; + +/// Sub-lifecycle responsible for auto-status override on nodes with `auto_status=true`. +pub struct AutoStatusLifecycle; + +#[async_trait] +impl RunLifecycle 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(()) + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs new file mode 100644 index 000000000..5050f7fbd --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs @@ -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>; +type WfNodeResult = NodeResult>; + +/// 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>, + restart_failure_signatures: Mutex>, + 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, + restart_sigs: HashMap, + ) { + *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, + HashMap, + ) { + 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 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(()) + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/disk.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/disk.rs new file mode 100644 index 000000000..07e5b092b --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/disk.rs @@ -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>; +type WfNodeResult = NodeResult>; + +/// 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, + pub circuit_breaker: Arc, + pub checkpoint_enabled: bool, +} + +#[async_trait] +impl RunLifecycle 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(()) + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs new file mode 100644 index 000000000..a56ddaa8d --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs @@ -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>; +type WfNodeResult = NodeResult>; + +/// Sub-lifecycle responsible for emitting workflow run events. +pub struct EventLifecycle { + pub emitter: Arc, + pub graph_name: String, + pub run_id: String, + pub run_start: Mutex, + /// Set in on_edge_selected when loop_restart approved; emitted+cleared in on_run_start. + pub restarted_from: Arc>>, + // Config for WorkflowRunStarted payload + pub base_sha: Option, + pub run_branch: Option, + pub worktree_dir: Option, + pub goal: Option, +} + +#[async_trait] +impl RunLifecycle 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>> { + 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 { + 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, + }); + } + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs new file mode 100644 index 000000000..f801470bb --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs @@ -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>; +type WfNodeDecision = NodeDecision>; + +/// Data captured from an edge selection to pass to the next node's before_node. +#[derive(Debug, Clone)] +struct IncomingEdgeData { + fidelity: Option, + thread_id: Option, +} + +/// Sub-lifecycle responsible for fidelity/thread resolution and context key setup. +pub struct FidelityLifecycle { + pub graph: Arc, + incoming_edge_data: Mutex>, + /// True on the first node after checkpoint resume when prior fidelity was Full. + degrade_fidelity_on_resume: Mutex, +} + +impl FidelityLifecycle { + pub fn new(graph: Arc) -> 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 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 { + 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 { + // 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) + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs new file mode 100644 index 000000000..a49fdda78 --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs @@ -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 for GitLifecycle {} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs new file mode 100644 index 000000000..be1cd842f --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs @@ -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>; +type WfNodeResult = NodeResult>; +type WfNodeDecision = NodeDecision>; + +/// Sub-lifecycle responsible for running workflow hooks. +pub struct HookLifecycle { + pub hook_runner: Option>, + pub sandbox: Arc, + 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 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 { + 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 { + 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; + } + } +} diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs new file mode 100644 index 000000000..a8a73f0d8 --- /dev/null +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs @@ -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>; +type WfNodeResult = NodeResult>; +type WfNodeDecision = NodeDecision>; + +/// Orchestrates all sub-lifecycles with explicit per-callback ordering. +/// Implements `RunLifecycle` by delegating to focused structs. +pub struct WorkflowLifecycle { + event: EventLifecycle, + hook: HookLifecycle, + fidelity: FidelityLifecycle, + auto_status: AutoStatusLifecycle, + circuit_breaker: Arc, + 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>>, +} + +impl WorkflowLifecycle { + #[allow(clippy::too_many_arguments)] + pub fn new( + emitter: Arc, + hook_runner: Option>, + sandbox: Arc, + graph: Arc, + run_dir: PathBuf, + run_id: String, + _dry_run: bool, + _labels: HashMap, + ) -> Self { + let restarted_from: Arc>> = 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, + restart_sigs: HashMap, + ) { + 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 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 { + self.fidelity.before_node(node, state).await + } + + async fn before_attempt( + &self, + ctx: &AttemptContext<'_, WorkflowGraph>, + state: &WfRunState, + ) -> CoreResult { + // 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 { + // 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; + } +} diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index fb00c5ae5..8f03063bc 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -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 }) => {