diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs index 0c6a7abe7..1e88430c8 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs @@ -11,7 +11,6 @@ use fabro_core::state::RunState; use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; use crate::artifact::{offload_large_values, sync_artifacts_to_env, ArtifactStore}; -use crate::engine; use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use crate::outcome::StageUsage; use fabro_core::lifecycle::NodeDecision; @@ -64,6 +63,7 @@ impl RunLifecycle for ArtifactLifecycle { // Swap in a fresh artifact store on restart (don't call clear() — preserves files on disk) let mut store = self.artifact_store.lock().unwrap(); *store = ArtifactStore::new(self.artifact_base_dir.clone()); + *self.attempt_start_epoch.lock().unwrap() = None; Ok(()) } @@ -92,7 +92,17 @@ impl RunLifecycle for ArtifactLifecycle { let epoch = self.attempt_start_epoch.lock().unwrap().unwrap_or(0.0); let node_id = ctx.node.id(); let visit = state.node_visits.get(node_id).copied().unwrap_or(1); - let stage_dir = engine::node_dir(&self.run_dir, node_id, visit); + let node_slug = if visit <= 1 { + node_id.to_string() + } else { + format!("{node_id}-visit_{visit}") + }; + let stage_dir = self + .run_dir + .join("artifacts") + .join("assets") + .join(node_slug) + .join(format!("retry_{}", ctx.attempt)); let _ = std::fs::create_dir_all(&stage_dir); match crate::asset_snapshot::collect_assets( 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 index f2ee38238..db5268924 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs @@ -10,6 +10,7 @@ use fabro_core::state::RunState; use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; +use crate::engine; use crate::error::{FailureCategory, FailureSignature}; use crate::outcome::{OutcomeExt, StageStatus, StageUsage}; @@ -68,7 +69,7 @@ impl RunLifecycle for CircuitBreakerLifecycle { let outcome = &result.outcome; let outcome_failure_category = if outcome.status == StageStatus::Fail { - outcome.failure.as_ref().map(|f| f.category) + engine::classify_outcome(outcome) } else { None }; @@ -116,10 +117,10 @@ impl RunLifecycle for CircuitBreakerLifecycle { let outcome = ctx.outcome; // Guard: only TransientInfra failures may trigger loop_restart - let failure_class = outcome.failure_category(); + let failure_class = engine::classify_outcome(outcome); if let Some(fc) = failure_class { if fc != FailureCategory::TransientInfra { - return Err(CoreError::blocked(format!( + return Ok(EdgeDecision::Block(format!( "loop_restart blocked: failure_class={fc} (requires transient_infra), failure_reason={}", outcome.failure_reason().unwrap_or("none"), ))); @@ -140,7 +141,7 @@ impl RunLifecycle for CircuitBreakerLifecycle { *count += 1; let limit = self.loop_restart_signature_limit; if *count >= limit { - return Err(CoreError::blocked(format!( + return Ok(EdgeDecision::Block(format!( "loop_restart circuit breaker: signature {sig} repeated {count} times (limit {limit})" ))); } diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs index 1f5e0067e..5b7f02fe0 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs @@ -12,6 +12,7 @@ use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; use super::git::GitCheckpointResult; use crate::artifact::ArtifactStore; +use crate::engine; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::outcome::{FailureCategory, FailureDetail, Outcome, StageStatus, StageUsage}; @@ -86,7 +87,7 @@ impl RunLifecycle for EventLifecycle { name: gv.label().to_string(), index: stage_index, handler_type: gv.handler_type().map(String::from), - script: None, + script: engine::node_script(gv), attempt: 1, max_attempts: 1, }); @@ -118,7 +119,7 @@ impl RunLifecycle for EventLifecycle { name: gv.label().to_string(), index: state.stage_index, handler_type: gv.handler_type().map(String::from), - script: None, + script: engine::node_script(gv), attempt: ctx.attempt as usize, max_attempts: ctx.max_attempts as usize, }); @@ -192,7 +193,7 @@ impl RunLifecycle for EventLifecycle { preferred_label: outcome.preferred_label.clone(), suggested_next_ids: outcome.suggested_next_ids.clone(), usage: outcome.usage.clone(), - failure: None, + failure: outcome.failure.clone(), notes: outcome.notes.clone(), files_touched: outcome.files_touched.clone(), attempt: result.attempts as usize, @@ -240,7 +241,7 @@ impl RunLifecycle for EventLifecycle { let status = result.outcome.status.to_string(); // Read git checkpoint result (set by GitLifecycle) - let git_result = self.checkpoint_git_result.lock().unwrap().take(); + let git_result = self.checkpoint_git_result.lock().unwrap().clone(); let git_sha = git_result.as_ref().and_then(|r| r.commit_sha.clone()); @@ -276,15 +277,32 @@ impl RunLifecycle for EventLifecycle { let duration_ms = self.run_start.lock().unwrap().elapsed().as_millis() as u64; let artifact_count = self.artifact_store.lock().unwrap().list().len(); let last_sha = self.last_git_sha.lock().unwrap().clone(); + let total_cost = { + let sum: f64 = state + .node_outcomes + .values() + .filter_map(|o| o.usage.as_ref()?.cost) + .sum(); + if sum > 0.0 { + Some(sum) + } else { + None + } + }; + let run_usage = state + .node_outcomes + .values() + .filter_map(|o| o.usage.as_ref().map(fabro_llm::types::Usage::from)) + .reduce(|a, b| a + b); if outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess { self.emitter.emit(&WorkflowRunEvent::WorkflowRunCompleted { duration_ms, artifact_count, status: outcome.status.to_string(), - total_cost: None, + total_cost, final_git_commit_sha: last_sha, - usage: None, + usage: run_usage, }); } else { let error_msg = outcome diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs index 8f2c2d7a6..6df3df1e5 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs @@ -111,7 +111,7 @@ impl RunLifecycle for FidelityLifecycle { // 6. Set thread.{tid}.current_node if let Some(ref tid) = thread_id { - let key = format!("thread.{tid}.current_node"); + let key = keys::thread_current_node_key(tid); state.context.set(key, serde_json::json!(node.id())); } @@ -130,7 +130,7 @@ impl RunLifecycle for FidelityLifecycle { } // 8. Set INTERNAL_NODE_VISIT_COUNT and CURRENT_NODE - let visits = state.node_visits.get(node.id()).copied().unwrap_or(0); + let visits = state.node_visits.get(node.id()).copied().unwrap_or(1); state .context .set(keys::CURRENT_NODE, serde_json::json!(node.id())); diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs index ccdc99a46..c19402a8f 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs @@ -49,6 +49,7 @@ impl RunLifecycle for GitLifecycle { ) -> fabro_core::error::Result<()> { // Reset last_git_sha (diff base parity) *self.last_git_sha.lock().unwrap() = None; + *self.checkpoint_git_result.lock().unwrap() = None; // Init metadata branch (best-effort) if let (Some(_), Some(ref repo_path)) = @@ -91,6 +92,7 @@ impl RunLifecycle for GitLifecycle { // Skip git checkpoint for the start node (always empty) or if git disabled if self.start_node_id.as_deref() == Some(node_id) || !self.config.git_checkpoint_enabled { + *self.checkpoint_git_result.lock().unwrap() = None; return Ok(()); } diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs index 98f7f66ab..7f699131c 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs @@ -13,7 +13,7 @@ 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 crate::outcome::{Outcome, OutcomeExt, StageStatus, StageUsage}; use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner}; use fabro_sandbox::Sandbox; @@ -25,7 +25,7 @@ type WfNodeDecision = NodeDecision>; pub struct HookLifecycle { pub hook_runner: Option>, pub sandbox: Arc, - pub run_dir: PathBuf, + pub hook_work_dir: Option, pub run_id: String, pub graph_name: String, } @@ -36,7 +36,11 @@ impl HookLifecycle { return HookDecision::Proceed; }; runner - .run(hook_ctx, self.sandbox.clone(), Some(&self.run_dir)) + .run( + hook_ctx, + self.sandbox.clone(), + self.hook_work_dir.as_deref(), + ) .await } } @@ -68,13 +72,17 @@ impl RunLifecycle for HookLifecycle { self.run_id.clone(), self.graph_name.clone(), ); + hook_ctx.cwd = self + .hook_work_dir + .as_ref() + .map(|path| path.display().to_string()); 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()); + let msg = reason.unwrap_or_else(|| "skipped by StageStart hook".into()); Ok(NodeDecision::Skip(Box::new(Outcome::skipped(&msg)))) } HookDecision::Block { reason } => { @@ -87,7 +95,7 @@ impl RunLifecycle for HookLifecycle { async fn after_node( &self, - _node: &WorkflowNode, + node: &WorkflowNode, result: &mut WfNodeResult, _state: &WfRunState, ) -> CoreResult<()> { @@ -103,7 +111,9 @@ impl RunLifecycle for HookLifecycle { }; let mut hook_ctx = HookContext::new(hook_event, self.run_id.clone(), self.graph_name.clone()); + set_hook_node(&mut hook_ctx, node.inner()); hook_ctx.status = Some(outcome.status.to_string()); + hook_ctx.failure_reason = outcome.failure_reason().map(String::from); let _ = self.run_hook(&hook_ctx).await; Ok(()) } @@ -120,6 +130,10 @@ impl RunLifecycle for HookLifecycle { ); hook_ctx.edge_from = Some(ctx.from.to_string()); hook_ctx.edge_to = Some(ctx.to.to_string()); + hook_ctx.edge_label = ctx + .edge + .as_ref() + .and_then(|edge| edge.inner().label().map(String::from)); let decision = self.run_hook(&hook_ctx).await; match decision { HookDecision::Override { edge_to } => Ok(EdgeDecision::Override(edge_to)), diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs index 1fda1c1c6..56c8b2ad4 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs @@ -115,7 +115,7 @@ impl WorkflowLifecycle { let hook = HookLifecycle { hook_runner, sandbox: Arc::clone(&sandbox), - run_dir: run_dir.clone(), + hook_work_dir: working_directory.clone().map(PathBuf::from), run_id: config.run_id.clone(), graph_name: graph.name.clone(), }; diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index a8273f90b..bbccb336b 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -46,7 +46,7 @@ pub(crate) fn set_hook_node(ctx: &mut HookContext, node: &Node) { /// 2. String heuristics on `failure_reason` /// 3. Default to `Deterministic` #[must_use] -fn classify_outcome(outcome: &Outcome) -> Option { +pub(crate) fn classify_outcome(outcome: &Outcome) -> Option { match outcome.status { StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped => None, StageStatus::Fail | StageStatus::Retry => outcome @@ -57,6 +57,7 @@ fn classify_outcome(outcome: &Outcome) -> Option { /// Mutable state carried across loop restarts and recursive `run_internal` calls. #[derive(Default)] +#[cfg_attr(feature = "core-engine", allow(dead_code))] struct LoopState { node_visits: HashMap, /// Tracks deterministic/structural failure signatures across main-loop stages. @@ -542,7 +543,7 @@ pub(crate) fn is_terminal(node: &Node) -> bool { node.shape() == "Msquare" || node.handler_type() == Some("exit") } -fn node_script(node: &Node) -> Option { +pub(crate) fn node_script(node: &Node) -> Option { node.attrs .get("script") .or_else(|| node.attrs.get("tool_command")) @@ -923,6 +924,7 @@ impl WorkflowRunEngine { } /// Fire a non-blocking RunFailed hook. + #[cfg_attr(feature = "core-engine", allow(dead_code))] async fn run_failed_hook( &self, run_id: &str, @@ -940,6 +942,7 @@ impl WorkflowRunEngine { } /// Mirror graph-level attributes into the context. + #[cfg_attr(feature = "core-engine", allow(dead_code))] pub(crate) fn mirror_graph_attributes(graph: &Graph, context: &Context) { if !graph.goal().is_empty() { context.set(context::keys::GRAPH_GOAL, serde_json::json!(graph.goal())); @@ -955,6 +958,7 @@ impl WorkflowRunEngine { /// Execute a node handler with retry policy. /// Returns `(outcome, attempts_used)` where `attempts_used` is the 1-indexed count. #[allow(clippy::too_many_arguments)] + #[cfg_attr(feature = "core-engine", allow(dead_code))] async fn execute_with_retry( &self, node: &Node, @@ -1181,10 +1185,19 @@ impl WorkflowRunEngine { /// Returns an error if no start node is found, a node is missing, or a goal gate fails /// without a retry target. pub async fn run(&self, graph: &Graph, config: &RunConfig) -> Result { - let (outcome, _context) = self - .run_internal(graph, config, None, None, None, LoopState::default()) - .await?; - Ok(outcome) + #[cfg(feature = "core-engine")] + { + let (outcome, _context) = self.run_via_core(graph, config, None, None).await?; + return Ok(outcome); + } + + #[cfg(not(feature = "core-engine"))] + { + let (outcome, _context) = self + .run_internal(graph, config, None, None, None, LoopState::default()) + .await?; + Ok(outcome) + } } /// Run a workflow with full sandbox lifecycle management. @@ -1389,15 +1402,25 @@ impl WorkflowRunEngine { config: &RunConfig, seed_context: Context, ) -> Result<(Outcome, Context)> { - self.run_internal( - graph, - config, - None, - None, - Some(seed_context), - LoopState::default(), - ) - .await + #[cfg(feature = "core-engine")] + { + return self + .run_via_core(graph, config, None, Some(seed_context)) + .await; + } + + #[cfg(not(feature = "core-engine"))] + { + self.run_internal( + graph, + config, + None, + None, + Some(seed_context), + LoopState::default(), + ) + .await + } } /// Resume from a checkpoint. Restores context, completed nodes, and continues @@ -1412,15 +1435,26 @@ impl WorkflowRunEngine { config: &RunConfig, checkpoint: &Checkpoint, ) -> Result { - let loop_state = LoopState { - node_visits: HashMap::new(), - loop_failure_signatures: checkpoint.loop_failure_signatures.clone(), - restart_failure_signatures: checkpoint.restart_failure_signatures.clone(), - }; - let (outcome, _context) = self - .run_internal(graph, config, Some(checkpoint), None, None, loop_state) - .await?; - Ok(outcome) + #[cfg(feature = "core-engine")] + { + let (outcome, _context) = self + .run_via_core(graph, config, Some(checkpoint), None) + .await?; + return Ok(outcome); + } + + #[cfg(not(feature = "core-engine"))] + { + let loop_state = LoopState { + node_visits: HashMap::new(), + loop_failure_signatures: checkpoint.loop_failure_signatures.clone(), + restart_failure_signatures: checkpoint.restart_failure_signatures.clone(), + }; + let (outcome, _context) = self + .run_internal(graph, config, Some(checkpoint), None, None, loop_state) + .await?; + Ok(outcome) + } } /// Run the workflow through the fabro-core executor with full lifecycle management. @@ -1660,6 +1694,7 @@ impl WorkflowRunEngine { } /// Internal run implementation supporting optional checkpoint resume and `start_at` override. + #[cfg_attr(feature = "core-engine", allow(dead_code))] async fn run_internal( &self, graph: &Graph,