From ddb499c319fba7c8772d9fbad31434386c33aacb Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 24 Mar 2026 17:49:44 -0400 Subject: [PATCH] Remove core-engine feature flag and old execution loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix parity gaps (handler errors → fail outcomes, panic.txt, goal gate message, fail-with-no-edge message, visit limit source, terminal completion normalization) then delete ~1,250 lines of old-path code (LoopState, run_failed_hook, mirror_graph_attributes, execute_with_retry, run_internal) and all cfg gating. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/crates/fabro-core/src/error.rs | 23 +- lib/crates/fabro-core/src/executor.rs | 29 +- lib/crates/fabro-core/src/lib.rs | 2 +- lib/crates/fabro-workflows/Cargo.toml | 1 - .../src/core_adapter/handler.rs | 4 + .../src/core_adapter/lifecycle/mod.rs | 6 +- lib/crates/fabro-workflows/src/engine.rs | 1385 +---------------- lib/crates/fabro-workflows/src/handler/mod.rs | 2 +- 8 files changed, 76 insertions(+), 1376 deletions(-) diff --git a/lib/crates/fabro-core/src/error.rs b/lib/crates/fabro-core/src/error.rs index 07146daa3..9b97c71d4 100644 --- a/lib/crates/fabro-core/src/error.rs +++ b/lib/crates/fabro-core/src/error.rs @@ -2,6 +2,21 @@ use std::fmt; use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeMeta, StageStatus}; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum VisitLimitSource { + Node, + Graph, +} + +impl fmt::Display for VisitLimitSource { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Node => write!(f, "node"), + Self::Graph => write!(f, "graph"), + } + } +} + /// Structured failure data on handler errors. Maps to FabroError's /// is_retryable(), failure_class(), failure_signature_hint(), to_fail_outcome(). #[derive(Debug, Clone)] @@ -28,11 +43,12 @@ pub enum CoreError { Cancelled, #[error("blocked: {message}")] Blocked { message: String }, - #[error("node \"{node_id}\" visited {visits} times (limit {limit})")] + #[error("node \"{node_id}\" visited {visits} times ({limit_source} limit {limit}); run is stuck in a cycle")] VisitLimitExceeded { node_id: String, visits: usize, limit: usize, + limit_source: VisitLimitSource, }, #[error("stall timeout on node \"{node_id}\"")] StallTimeout { node_id: String }, @@ -101,10 +117,11 @@ mod tests { CoreError::VisitLimitExceeded { node_id: "n1".into(), visits: 5, - limit: 3 + limit: 3, + limit_source: VisitLimitSource::Node, } .to_string(), - "node \"n1\" visited 5 times (limit 3)" + "node \"n1\" visited 5 times (node limit 3); run is stuck in a cycle" ); assert_eq!( CoreError::StallTimeout { diff --git a/lib/crates/fabro-core/src/executor.rs b/lib/crates/fabro-core/src/executor.rs index 835ff8609..699b63b54 100644 --- a/lib/crates/fabro-core/src/executor.rs +++ b/lib/crates/fabro-core/src/executor.rs @@ -131,8 +131,7 @@ impl Executor { continue; } let outcome = Outcome::fail(&format!( - "goal gate failed for node \"{}\"", - failed_node_id + "goal gate unsatisfied for node {failed_node_id} and no retry target" )); self.lifecycle.on_run_end(&outcome, &state).await; return Ok((outcome, state)); @@ -148,6 +147,7 @@ impl Executor { node_id: node.id().to_string(), visits, limit: max, + limit_source: crate::error::VisitLimitSource::Node, }); } } @@ -157,6 +157,7 @@ impl Executor { node_id: node.id().to_string(), visits, limit: global_max, + limit_source: crate::error::VisitLimitSource::Graph, }); } } @@ -225,7 +226,13 @@ impl Executor { self.lifecycle.on_run_start(graph, &state).await?; } NextStep::End => { - let outcome = last_outcome.clone(); + let mut outcome = last_outcome.clone(); + if outcome.status == StageStatus::Fail { + outcome = Outcome::fail(&format!( + "stage {} failed with no outgoing fail edge", + node.id() + )); + } self.lifecycle.on_run_end(&outcome, &state).await; return Ok((outcome, state)); } @@ -317,17 +324,19 @@ impl Executor { tokio::time::sleep(delay).await; } Err(e) => { - let fail_result = - NodeResult::from_error(&e, start.elapsed(), attempt, policy.max_attempts); + // Convert handler error to fail outcome so routing continues + let outcome = e.to_fail_outcome(); + let result = + NodeResult::new(outcome, start.elapsed(), attempt, policy.max_attempts); let ctx = AttemptResultContext { node, - result: &fail_result, + result: &result, attempt, will_retry: false, backoff_delay: None, }; self.lifecycle.after_attempt(&ctx, state).await?; - return Err(e); + return Ok(result); } } } @@ -978,7 +987,8 @@ mod tests { handler.clone() as Arc>, ) .await; - assert!(result.is_err()); + // Non-retryable errors become fail outcomes, routing continues through the linear graph + assert!(result.is_ok()); assert_eq!(handler.calls(), 1); } @@ -998,7 +1008,8 @@ mod tests { handler.clone() as Arc>, ) .await; - assert!(result.is_err()); + // Errors become fail outcomes, routing continues through the linear graph + assert!(result.is_ok()); assert_eq!(handler.calls(), 1); } diff --git a/lib/crates/fabro-core/src/lib.rs b/lib/crates/fabro-core/src/lib.rs index 84c53cedc..e8a19c918 100644 --- a/lib/crates/fabro-core/src/lib.rs +++ b/lib/crates/fabro-core/src/lib.rs @@ -13,7 +13,7 @@ pub mod state; pub mod test_fixtures; pub use context::Context; -pub use error::{CoreError, HandlerErrorDetail, Result}; +pub use error::{CoreError, HandlerErrorDetail, Result, VisitLimitSource}; pub use executor::{Executor, ExecutorBuilder, ExecutorSettings}; pub use graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; pub use handler::NodeHandler; diff --git a/lib/crates/fabro-workflows/Cargo.toml b/lib/crates/fabro-workflows/Cargo.toml index 1d050e278..f979b3843 100644 --- a/lib/crates/fabro-workflows/Cargo.toml +++ b/lib/crates/fabro-workflows/Cargo.toml @@ -14,7 +14,6 @@ doctest = false [features] default = [] -core-engine = [] exedev = ["fabro-sandbox/exe", "fabro-config/exedev"] [dependencies] diff --git a/lib/crates/fabro-workflows/src/core_adapter/handler.rs b/lib/crates/fabro-workflows/src/core_adapter/handler.rs index 574b309d2..684be397b 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/handler.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/handler.rs @@ -97,6 +97,10 @@ impl NodeHandler for WorkflowNodeHandler { } Err(panic_payload) => { let msg = format_panic_message(panic_payload); + let visit = context.node_visit_count().max(1); + let panic_dir = crate::engine::node_dir(&self.run_dir, &gv_node.id, visit); + let _ = std::fs::create_dir_all(&panic_dir); + let _ = std::fs::write(panic_dir.join("panic.txt"), &msg); Err(CoreError::handler(HandlerErrorDetail { message: msg, retryable: false, 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 5893ff98f..bd28360eb 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs @@ -63,7 +63,7 @@ pub struct WorkflowLifecycle { /// Shared git checkpoint result (written by git, read by event) checkpoint_git_result: Arc>>, /// True when constructed with a checkpoint; cleared after first on_run_start. - /// Gates mirror_graph_attributes on initial resume. + /// Gates context seeding on initial resume. is_initial_resume: AtomicBool, // Config needed for context seeding graph: Arc, @@ -194,9 +194,9 @@ impl WorkflowLifecycle { impl RunLifecycle for WorkflowLifecycle { async fn on_run_start(&self, graph: &WorkflowGraph, state: &WfRunState) -> CoreResult<()> { // Re-seed context keys (fires on initial start AND after every loop restart). - // mirror_graph_attributes: skip on initial checkpoint resume (context already has them) + // Skip on initial checkpoint resume (context already has them). if self.is_initial_resume.swap(false, Ordering::Relaxed) { - // First on_run_start after checkpoint resume — skip mirror_graph_attributes + // First on_run_start after checkpoint resume — skip context seeding } else { // Mirror graph-level attributes into the core context if !self.graph.goal().is_empty() { diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index b2d39eda7..d6bdc33c3 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -1,31 +1,28 @@ use std::collections::HashMap; -use std::panic::AssertUnwindSafe; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::AtomicBool; use std::sync::Arc; use std::time::{Duration, Instant}; use chrono::Utc; use fabro_agent::Sandbox; +use fabro_core::executor::ExecutorBuilder; +use fabro_core::state::RunState; use fabro_util::backoff::BackoffPolicy; -use futures::FutureExt; use rand::Rng; use tokio_util::sync::CancellationToken; use fabro_git_storage::trailerlink::{self, Trailer}; -use crate::artifact::{offload_large_values, sync_artifacts_to_env, ArtifactStore}; use crate::asset_snapshot; use crate::checkpoint::Checkpoint; use crate::condition::evaluate_condition; use crate::context; use crate::context::Context; -use crate::error::{FabroError, FailureCategory, FailureSignature, Result}; +use crate::error::{FabroError, FailureCategory, Result}; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::handler::{EngineServices, HandlerRegistry}; -use crate::millis_u64; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; -use crate::preamble::build_preamble; use fabro_config::run::PullRequestConfig; use fabro_graphviz::graph::{Edge, Graph, Node}; use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner}; @@ -55,18 +52,6 @@ pub(crate) 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. - /// Never reset on success — prevents impl-succeeds/verify-fails cycles. - loop_failure_signatures: HashMap, - /// Tracks failure signatures across loop_restart edges. - restart_failure_signatures: HashMap, -} - // --- Retry policy types --- /// Retry policy for node execution. @@ -926,281 +911,15 @@ impl WorkflowRunEngine { .await } - /// Fire a non-blocking RunFailed hook. - #[cfg_attr(feature = "core-engine", allow(dead_code))] - async fn run_failed_hook( - &self, - run_id: &str, - workflow_name: &str, - error: &FabroError, - work_dir: Option<&Path>, - ) { - let mut hook_ctx = HookContext::new( - HookEvent::RunFailed, - run_id.to_string(), - workflow_name.to_string(), - ); - hook_ctx.failure_reason = Some(error.to_string()); - let _ = self.run_hooks(&hook_ctx, work_dir).await; - } - - /// 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())); - } - for (key, val) in &graph.attrs { - context.set( - context::keys::graph_attr_key(key), - serde_json::json!(val.to_string_value()), - ); - } - } - - /// 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, - context: &Context, - graph: &Graph, - run_dir: &Path, - policy: &RetryPolicy, - stage_index: usize, - visit: usize, - asset_globs: &[String], - run_id: &str, - hook_work_dir: Option<&Path>, - ) -> Result<(Outcome, u32)> { - let handler = self.services.registry.resolve(node); - - let node_timeout = node.timeout(); - - for attempt in 1..=policy.max_attempts { - // Run StageStart hook (blocking — can skip or block node) - { - let mut hook_ctx = HookContext::new( - HookEvent::StageStart, - run_id.to_string(), - graph.name.clone(), - ); - hook_ctx.cwd = hook_work_dir.map(|p| p.display().to_string()); - set_hook_node(&mut hook_ctx, node); - hook_ctx.attempt = Some(usize::try_from(attempt).unwrap_or(usize::MAX)); - hook_ctx.max_attempts = - Some(usize::try_from(policy.max_attempts).unwrap_or(usize::MAX)); - let decision = self.run_hooks(&hook_ctx, hook_work_dir).await; - match decision { - HookDecision::Skip { reason } => { - let msg = reason.unwrap_or_else(|| "skipped by StageStart hook".into()); - let mut outcome = Outcome::skipped(&msg); - outcome.notes = Some(msg); - return Ok((outcome, attempt)); - } - HookDecision::Block { reason } => { - let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into()); - return Err(FabroError::engine(msg)); - } - _ => {} - } - } - - // Emit StageStarted (fires once per attempt, only after hook passes) - self.services.emitter.emit(&WorkflowRunEvent::StageStarted { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - handler_type: node.handler_type().map(String::from), - script: node_script(node), - attempt: usize::try_from(attempt).unwrap_or(usize::MAX), - max_attempts: usize::try_from(policy.max_attempts).unwrap_or(usize::MAX), - }); - // Floor to integer seconds: macOS stat reports mtime as integer seconds, - // so a fractional epoch would reject files created in the same second. - let command_start_epoch = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs() as f64) - .unwrap_or(0.0); - - // Gap #11: Panic safety -- catch panics from handler execution - let result = { - let future = crate::handler::dispatch_handler( - handler, - node, - context, - graph, - run_dir, - &self.services, - ); - let panic_safe = AssertUnwindSafe(future).catch_unwind(); - // Gap #2: Timeout enforcement -- wrap with tokio::time::timeout - let timed_result = if let Some(duration) = node_timeout { - match tokio::time::timeout(duration, panic_safe).await { - Ok(inner) => inner, - Err(_elapsed) => Ok(Ok(Outcome::fail_classify(format!( - "handler timed out after {}ms", - duration.as_millis() - )))), - } - } else { - panic_safe.await - }; - match timed_result { - Ok(r) => r, - Err(panic_payload) => { - let msg = crate::handler::format_panic_message(panic_payload); - let panic_dir = node_dir(run_dir, &node.id, visit); - let _ = std::fs::create_dir_all(&panic_dir); - let _ = std::fs::write(panic_dir.join("panic.txt"), &msg); - Err(FabroError::handler(msg)) - } - } - }; - - // Collect assets after handler completes (only when globs are configured) - if !asset_globs.is_empty() { - let node_slug = if visit <= 1 { - node.id.clone() - } else { - format!("{}-visit_{visit}", node.id) - }; - let assets_dir = run_dir - .join("artifacts") - .join("assets") - .join(&node_slug) - .join(format!("retry_{attempt}")); - match asset_snapshot::collect_assets( - self.services.sandbox.as_ref(), - &assets_dir, - asset_globs, - command_start_epoch, - ) - .await - { - Ok(summary) if summary.files_copied > 0 => { - self.services - .emitter - .emit(&WorkflowRunEvent::AssetsCaptured { - node_id: node.id.clone(), - files_copied: summary.files_copied, - total_bytes: summary.total_bytes, - files_skipped: summary.files_skipped, - }); - } - Ok(_) => {} - Err(e) => { - tracing::warn!( - node = %node.id, - error = %e, - "Asset collection failed" - ); - } - } - } - - let outcome = match result { - Ok(o) => o, - Err(e) => { - // Gap #7: Check should_retry predicate before retrying - if attempt < policy.max_attempts && handler.should_retry(&e) { - let delay = policy.backoff.delay_for_attempt(attempt); - self.services.emitter.emit(&WorkflowRunEvent::StageFailed { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - failure: crate::outcome::FailureDetail { - message: e.to_string(), - category: e.failure_category(), - signature: e.failure_signature_hint(), - }, - will_retry: true, - }); - self.services - .emitter - .emit(&WorkflowRunEvent::StageRetrying { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - attempt: usize::try_from(attempt).unwrap_or(usize::MAX), - max_attempts: usize::try_from(policy.max_attempts) - .unwrap_or(usize::MAX), - delay_ms: millis_u64(delay), - }); - tokio::time::sleep(delay).await; - continue; - } - return Ok((e.to_fail_outcome(), attempt)); - } - }; - - match outcome.status { - StageStatus::Success - | StageStatus::PartialSuccess - | StageStatus::Fail - | StageStatus::Skipped => { - return Ok((outcome, attempt)); - } - StageStatus::Retry => { - if attempt < policy.max_attempts { - let delay = policy.backoff.delay_for_attempt(attempt); - self.services - .emitter - .emit(&WorkflowRunEvent::StageRetrying { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - attempt: usize::try_from(attempt).unwrap_or(usize::MAX), - max_attempts: usize::try_from(policy.max_attempts) - .unwrap_or(usize::MAX), - delay_ms: millis_u64(delay), - }); - tokio::time::sleep(delay).await; - continue; - } - if node.allow_partial() { - return Ok(( - Outcome { - status: StageStatus::PartialSuccess, - notes: Some("retries exhausted, partial accepted".to_string()), - ..Outcome::success() - }, - attempt, - )); - } - return Ok((Outcome::fail_classify("max retries exceeded"), attempt)); - } - } - } - - Ok(( - Outcome::fail_classify("max retries exceeded"), - policy.max_attempts, - )) - } - - /// Run the workflow. Returns the final outcome. + /// Execute the workflow graph. Returns the final outcome. /// /// # Errors /// /// 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 { - #[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) - } + let (outcome, _context) = self.run_via_core(graph, config, None, None).await?; + Ok(outcome) } /// Run a workflow with full sandbox lifecycle management. @@ -1211,7 +930,7 @@ impl WorkflowRunEngine { /// 4. Sandbox git setup via `sandbox.setup_git_for_run()` /// 5. Run setup commands /// 6. Run devcontainer lifecycle phases - /// 7. Execute the workflow graph via `run_internal` + /// 7. Execute the workflow graph /// /// The sandbox is left alive after return so the caller can run retro, PR creation, etc. /// Call `cleanup_sandbox()` when done. @@ -1405,25 +1124,8 @@ impl WorkflowRunEngine { config: &RunConfig, seed_context: Context, ) -> Result<(Outcome, Context)> { - #[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(), - ) + self.run_via_core(graph, config, None, Some(seed_context)) .await - } } /// Resume from a checkpoint. Restores context, completed nodes, and continues @@ -1438,31 +1140,13 @@ impl WorkflowRunEngine { config: &RunConfig, checkpoint: &Checkpoint, ) -> Result { - #[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) - } + let (outcome, _context) = self + .run_via_core(graph, config, Some(checkpoint), None) + .await?; + Ok(outcome) } /// Run the workflow through the fabro-core executor with full lifecycle management. - #[cfg(feature = "core-engine")] - #[allow(dead_code)] async fn run_via_core( &self, graph: &Graph, @@ -1470,10 +1154,6 @@ impl WorkflowRunEngine { resume_checkpoint: Option<&Checkpoint>, seed_context: Option, ) -> Result<(Outcome, Context)> { - use fabro_core::executor::ExecutorBuilder; - use fabro_core::state::RunState; - use tokio_util::sync::CancellationToken; - let graph_arc = std::sync::Arc::new(graph.clone()); let wf_graph = crate::core_adapter::WorkflowGraph(Arc::clone(&graph_arc)); @@ -1666,9 +1346,15 @@ impl WorkflowRunEngine { // Convert result match result { Ok((core_outcome, final_state)) => { - // Extract the executor's final context so callers see all state let ctx = final_state.context.clone(); - Ok((core_outcome, ctx)) + let result = if core_outcome.status == StageStatus::Fail { + core_outcome + } else { + let mut out = Outcome::success(); + out.notes = Some("Pipeline completed".to_string()); + out + }; + Ok((result, ctx)) } Err(fabro_core::CoreError::StallTimeout { node_id }) => { let stall_timeout = graph.stall_timeout().unwrap_or_default(); @@ -1685,1030 +1371,9 @@ impl WorkflowRunEngine { } Err(fabro_core::CoreError::Cancelled) => Err(FabroError::Cancelled), Err(fabro_core::CoreError::Blocked { message }) => Err(FabroError::engine(message)), - Err(fabro_core::CoreError::VisitLimitExceeded { - node_id, - visits, - limit, - }) => Err(FabroError::engine(format!( - "node \"{node_id}\" visited {visits} times (limit {limit})" - ))), Err(e) => Err(FabroError::engine(e.to_string())), } } - - /// 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, - config: &RunConfig, - resume_checkpoint: Option<&Checkpoint>, - start_at: Option<&str>, - seed_context: Option, - mut loop_state: LoopState, - ) -> Result<(Outcome, Context)> { - let run_start = Instant::now(); - let run_id = config.run_id.clone(); - let artifact_store = ArtifactStore::new(Some(config.run_dir.clone())); - - // 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: 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 - }; - self.services.set_git_state(git_state); - - // Host-side git checkpoint: sandbox has a host-accessible worktree - let local_git_checkpoint = - config.git_checkpoint_enabled && self.services.sandbox.host_git_dir().is_some(); - - self.services - .emitter - .emit(&WorkflowRunEvent::WorkflowRunStarted { - name: graph.name.clone(), - run_id: run_id.clone(), - base_sha: config.base_sha.clone(), - run_branch: config.run_branch.clone(), - worktree_dir: if local_git_checkpoint { - Some(self.services.sandbox.working_directory().to_string()) - } else { - None - }, - goal: (!graph.goal().is_empty()).then(|| graph.goal().to_string()), - }); - - // Resolve work_dir from config for hooks - let hook_work_dir: Option = if local_git_checkpoint { - Some(PathBuf::from(self.services.sandbox.working_directory())) - } else { - None - }; - - // RunStart hook (blocking — can prevent run) - { - let hook_ctx = - HookContext::new(HookEvent::RunStart, run_id.clone(), graph.name.clone()); - let decision = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - if let HookDecision::Block { reason } = decision { - let msg = reason.unwrap_or_else(|| "blocked by RunStart hook".into()); - return Err(FabroError::engine(msg)); - } - } - - // Write manifest.json (spec 5.6) - let manifest = write_manifest(&config.run_dir, graph, config); - crate::run_status::write_run_status( - &config.run_dir, - crate::run_status::RunStatus::Running, - None, - ); - - // Initialize metadata branch for git-native checkpoint storage (best-effort) - if let (Some(_), Some(ref repo_path)) = (&config.meta_branch, &config.host_repo_path) { - let store = crate::git::MetadataStore::new(repo_path, &config.git_author); - let manifest_bytes = serde_json::to_vec_pretty(&manifest).unwrap_or_default(); - let dot_source = std::fs::read(config.run_dir.join("graph.fabro")) - .or_else(|_| std::fs::read(config.run_dir.join("graph.dot"))) - .unwrap_or_default(); - let sandbox_json = std::fs::read(config.run_dir.join("sandbox.json")).ok(); - let mut extra_files: Vec<(&str, &[u8])> = Vec::new(); - if let Some(ref data) = sandbox_json { - extra_files.push(("sandbox.json", data)); - } - if let Err(e) = - store.init_run(&config.run_id, &manifest_bytes, &dot_source, &extra_files) - { - tracing::warn!(run_id = %config.run_id, error = %e, "Metadata branch init failed"); - } - } - - // Compute effective max-node-visits limit: - // graph attr > 0 → use it; else dry_run → 10; else 0 (disabled) - let graph_limit = graph.max_node_visits(); - let graph_max_node_visits: usize = if graph_limit > 0 { - usize::try_from(graph_limit).unwrap_or(usize::MAX) - } else if config.dry_run { - 10 - } else { - 0 - }; - - // Gap #4: Initialize from checkpoint, start_at, or fresh - let context; - let mut completed_nodes: Vec; - let mut node_outcomes: HashMap = HashMap::new(); - let mut node_retries: HashMap = HashMap::new(); - let mut stage_index: usize; - let mut current_node_id: String; - let mut incoming_edge: Option<&Edge> = None; - let mut previous_node_id: Option = None; - // Gap #6: Track whether fidelity should be degraded on the first resumed node - let mut degrade_fidelity_on_resume = false; - let mut last_git_sha: Option = None; - - if let Some(cp) = resume_checkpoint { - // Restore context from checkpoint - context = Context::new(); - for (key, value) in &cp.context_values { - context.set(key.clone(), value.clone()); - } - completed_nodes = cp.completed_nodes.clone(); - // Use persisted node_visits; fall back to reconstruction for old checkpoints - if cp.node_visits.is_empty() { - for id in &completed_nodes { - *loop_state.node_visits.entry(id.clone()).or_insert(0) += 1; - } - } else { - loop_state.node_visits = cp.node_visits.clone(); - } - // Gap #5: Restore retry counters from checkpoint - node_retries = cp.node_retries.clone(); - // P1: Restore node outcomes for goal gate checks - node_outcomes = cp.node_outcomes.clone(); - stage_index = completed_nodes.len(); - // P1: Use stored next_node_id if available, otherwise fall back - if let Some(ref next_id) = cp.next_node_id { - current_node_id = next_id.clone(); - } else { - let edges = graph.outgoing_edges(&cp.current_node); - if let Some(edge) = edges.first() { - current_node_id = edge.to.clone(); - } else { - current_node_id = cp.current_node.clone(); - } - } - // Gap #6: Check if the checkpointed node used full fidelity - if cp.context_values.get(context::keys::INTERNAL_FIDELITY) - == Some(&serde_json::json!(context::keys::Fidelity::Full.to_string())) - { - degrade_fidelity_on_resume = true; - } - } else if let Some(start) = start_at { - context = Context::new(); - Self::mirror_graph_attributes(graph, &context); - completed_nodes = Vec::new(); - stage_index = 0; - current_node_id = start.to_string(); - } else { - context = seed_context.unwrap_or_default(); - Self::mirror_graph_attributes(graph, &context); - completed_nodes = Vec::new(); - stage_index = 0; - - let start_node = graph - .find_start_node() - .ok_or_else(|| FabroError::engine("no start node found".to_string()))?; - current_node_id = start_node.id.clone(); - } - - // Store run_id and work_dir in context for handlers - context.set(context::keys::INTERNAL_RUN_ID, serde_json::json!(run_id)); - if local_git_checkpoint { - context.set( - context::keys::INTERNAL_WORK_DIR, - serde_json::json!(self.services.sandbox.working_directory()), - ); - } - - // Stall watchdog: background task that cancels `stall_token` when no events - // have been emitted for longer than `stall_timeout`. - let stall_token = graph.stall_timeout().map(|timeout| { - let token = CancellationToken::new(); - let shutdown = CancellationToken::new(); - let check_interval = (timeout / 10) - .max(std::time::Duration::from_millis(50)) - .min(std::time::Duration::from_secs(5)); - self.services.emitter.touch(); - let emitter = Arc::clone(&self.services.emitter); - let cancel = token.clone(); - let stop = shutdown.clone(); - tracing::debug!( - stall_timeout_ms = timeout.as_millis() as u64, - check_interval_ms = check_interval.as_millis() as u64, - "Stall watchdog started" - ); - tokio::spawn(async move { - loop { - tokio::select! { - () = stop.cancelled() => break, - () = tokio::time::sleep(check_interval) => { - let last = emitter.last_event_at(); - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64; - if now - last >= timeout.as_millis() as i64 { - cancel.cancel(); - break; - } - } - } - } - }); - (token, shutdown) - }); - - let start_node_id = graph.find_start_node().map(|n| n.id.clone()); - - loop { - // Check for cancellation before processing each node - if let Some(ref token) = config.cancel_token { - if token.load(Ordering::Relaxed) { - return Err(FabroError::Cancelled); - } - } - - let node = graph - .nodes - .get(¤t_node_id) - .ok_or_else(|| FabroError::engine(format!("node not found: {current_node_id}")))?; - - // Always track visit count (used for stage directory naming) - let count = loop_state - .node_visits - .entry(current_node_id.clone()) - .or_insert(0); - *count += 1; - - let node_limit = node - .max_visits() - .and_then(|v| usize::try_from(v).ok()) - .filter(|&v| v > 0); - - if let Some(limit) = node_limit { - if *count >= limit { - tracing::warn!(node = %current_node_id, visits = *count, limit, source = "node", "Node visit limit exceeded"); - return Err(FabroError::engine(format!( - "node \"{}\" visited {count} times (node limit {limit}); run is stuck in a cycle", - current_node_id - ))); - } - } else if graph_max_node_visits > 0 && *count >= graph_max_node_visits { - tracing::warn!(node = %current_node_id, visits = *count, limit = graph_max_node_visits, source = "graph", "Node visit limit exceeded"); - return Err(FabroError::engine(format!( - "node \"{}\" visited {count} times (graph limit {graph_max_node_visits}); run is stuck in a cycle", - current_node_id - ))); - } - - // Step 1: Check for terminal node - if is_terminal(node) { - match check_goal_gates(graph, &node_outcomes) { - Ok(()) => { - self.services.emitter.emit(&WorkflowRunEvent::StageStarted { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - handler_type: node.handler_type().map(String::from), - script: node_script(node), - attempt: 1, - max_attempts: 1, - }); - self.services - .emitter - .emit(&WorkflowRunEvent::StageCompleted { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - duration_ms: 0, - status: StageStatus::Success.to_string(), - preferred_label: None, - suggested_next_ids: vec![], - usage: None, - failure: None, - notes: None, - files_touched: vec![], - attempt: 1, - max_attempts: 1, - }); - break; - } - Err(failed_node_id) => { - if let Some(retry_target) = get_retry_target(&failed_node_id, graph) { - current_node_id = retry_target; - continue; - } - let duration_ms = millis_u64(run_start.elapsed()); - let error = FabroError::engine(format!( - "goal gate unsatisfied for node {failed_node_id} and no retry target" - )); - self.services - .emitter - .emit(&WorkflowRunEvent::WorkflowRunFailed { - error: error.clone(), - duration_ms, - git_commit_sha: last_git_sha.clone(), - }); - - self.run_failed_hook( - &run_id, - &graph.name, - &error, - hook_work_dir.as_deref(), - ) - .await; - - return Ok((error.to_fail_outcome(), context)); - } - } - } - - // Resolve fidelity (spec 5.4) and store in context - let mut fidelity = resolve_fidelity(incoming_edge, node, graph); - // Gap #6: On the first node after resume, degrade full -> summary:high - if degrade_fidelity_on_resume { - let original = fidelity; - fidelity = fidelity.degraded(); - if fidelity != original { - tracing::debug!( - node = %current_node_id, - from = %original, - to = %fidelity, - "Fidelity degraded on checkpoint resume" - ); - } - } - degrade_fidelity_on_resume = false; - context.set( - context::keys::INTERNAL_FIDELITY, - serde_json::json!(fidelity.to_string()), - ); - - // Preamble injection at execution time (spec 5.4 / 8.3): synthesize a - // fidelity-appropriate preamble from runtime data for handlers to read - if fidelity == context::keys::Fidelity::Full { - context.set(context::keys::CURRENT_PREAMBLE, serde_json::json!("")); - } else { - let preamble = - build_preamble(fidelity, &context, graph, &completed_nodes, &node_outcomes); - context.set(context::keys::CURRENT_PREAMBLE, serde_json::json!(preamble)); - } - - // Thread context sharing: resolve thread ID and store in context for handlers - let resolved_thread_id = - resolve_thread_id(incoming_edge, node, graph, previous_node_id.as_deref()); - if let Some(ref tid) = resolved_thread_id { - context.set( - context::keys::thread_current_node_key(tid), - serde_json::json!(&node.id), - ); - context.set(context::keys::INTERNAL_THREAD_ID, serde_json::json!(tid)); - } else { - context.set(context::keys::INTERNAL_THREAD_ID, serde_json::Value::Null); - } - - // Step 2: Execute node handler with retry policy - let visit = *loop_state.node_visits.get(¤t_node_id).unwrap_or(&1); - context.set( - context::keys::INTERNAL_NODE_VISIT_COUNT, - serde_json::json!(visit), - ); - context.set(context::keys::CURRENT_NODE, serde_json::json!(&node.id)); - let retry_policy = build_retry_policy(node, graph); - - let stage_start = Instant::now(); - - let (mut outcome, attempts_used) = if let Some((ref token, _)) = stall_token { - tokio::select! { - result = self.execute_with_retry( - node, &context, graph, &config.run_dir, &retry_policy, stage_index, visit, &config.asset_globs, - &run_id, hook_work_dir.as_deref(), - ) => result?, - () = token.cancelled() => { - let idle_secs = graph.stall_timeout().map_or(0, |d| d.as_secs()); - self.services.emitter.emit(&WorkflowRunEvent::StallWatchdogTimeout { - node: node.id.clone(), - idle_seconds: idle_secs, - }); - return Err(FabroError::engine(format!( - "stall watchdog: node \"{}\" had no activity for {}s", - node.id, idle_secs, - ))); - } - } - } else { - self.execute_with_retry( - node, - &context, - graph, - &config.run_dir, - &retry_policy, - stage_index, - visit, - &config.asset_globs, - &run_id, - hook_work_dir.as_deref(), - ) - .await? - }; - // Gap #5: Track retry count per node - let retry_count = attempts_used.saturating_sub(1); - node_retries.insert(node.id.clone(), retry_count); - context.set( - context::keys::retry_count_key(&node.id), - serde_json::json!(retry_count), - ); - - // Gap #1: Auto status -- when auto_status=true and outcome is non-success, - // override to success with auto-status note - if node.auto_status() - && outcome.status != StageStatus::Success - && outcome.status != StageStatus::Skipped - { - outcome = Outcome { - status: StageStatus::Success, - notes: Some( - "auto-status: handler completed without writing status".to_string(), - ), - ..outcome - }; - } - - let stage_duration_ms = millis_u64(stage_start.elapsed()); - - let outcome_failure_class = classify_outcome(&outcome); - - // Circuit breaker: track deterministic/structural failure signatures - let failure_sig = if let Some(fc) = outcome_failure_class { - let sig_hint = outcome - .failure - .as_ref() - .and_then(|f| f.signature.as_deref()); - let sig = FailureSignature::new(&node.id, fc, sig_hint, outcome.failure_reason()); - if fc.is_signature_tracked() { - let count = loop_state - .loop_failure_signatures - .entry(sig.clone()) - .or_insert(0); - *count += 1; - let limit = graph.loop_restart_signature_limit(); - if *count >= limit { - return Err(FabroError::engine(format!( - "deterministic failure cycle detected: signature {sig} repeated {count} times (limit {limit})" - ))); - } - } - Some(sig) - } else { - None - }; - - // Hook-skipped stages have no StageStarted, so skip completion events/hooks - if outcome.status == StageStatus::Skipped { - // No StageCompleted/StageFailed — proceed to write_node_status, edge selection, etc. - } else if outcome.status == StageStatus::Fail { - self.services.emitter.emit(&WorkflowRunEvent::StageFailed { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - failure: outcome.failure.clone().unwrap_or_else(|| { - crate::outcome::FailureDetail::new( - "unknown", - FailureCategory::Deterministic, - ) - }), - will_retry: false, - }); - - // StageFailed hook (non-blocking) - { - let mut hook_ctx = HookContext::new( - HookEvent::StageFailed, - run_id.clone(), - graph.name.clone(), - ); - set_hook_node(&mut hook_ctx, node); - hook_ctx.status = Some("fail".into()); - hook_ctx.failure_reason = outcome.failure_reason().map(String::from); - let _ = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - } - } else { - self.services - .emitter - .emit(&WorkflowRunEvent::StageCompleted { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - duration_ms: stage_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: outcome.failure.clone(), - notes: outcome.notes.clone(), - files_touched: outcome.files_touched.clone(), - attempt: usize::try_from(attempts_used).unwrap_or(usize::MAX), - max_attempts: usize::try_from(retry_policy.max_attempts) - .unwrap_or(usize::MAX), - }); - - // StageComplete hook (non-blocking) - { - let mut hook_ctx = HookContext::new( - HookEvent::StageComplete, - run_id.clone(), - graph.name.clone(), - ); - set_hook_node(&mut hook_ctx, node); - hook_ctx.status = Some(outcome.status.to_string()); - let _ = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - } - } - - // Write per-node status.json (spec 5.6) - write_node_status(&config.run_dir, &node.id, visit, &outcome); - - // Offload large context values to artifact store before recording - if let Err(e) = offload_large_values(&mut outcome.context_updates, &artifact_store) { - self.services.emitter.emit(&WorkflowRunEvent::RunNotice { - level: crate::event::RunNoticeLevel::Warn, - code: "artifact_offload_failed".to_string(), - message: format!("[node: {}] artifact offload failed: {e}", node.id), - }); - } - - // Sync artifact files to the sandbox (no-op for local envs) - if let Err(e) = - sync_artifacts_to_env(&mut outcome.context_updates, &*self.services.sandbox).await - { - self.services.emitter.emit(&WorkflowRunEvent::RunNotice { - level: crate::event::RunNoticeLevel::Warn, - code: "artifact_sync_failed".to_string(), - message: format!("[node: {}] artifact sync failed: {e}", node.id), - }); - } - - // Step 3: Record completion - outcome.duration_ms = Some(stage_duration_ms); - completed_nodes.push(node.id.clone()); - node_outcomes.insert(node.id.clone(), outcome.clone()); - previous_node_id = Some(node.id.clone()); - stage_index += 1; - - // Step 4: Apply context updates from outcome - context.apply_updates(&outcome.context_updates); - context.set( - context::keys::OUTCOME, - serde_json::json!(outcome.status.to_string()), - ); - context.set( - context::keys::FAILURE_CLASS, - serde_json::json!(outcome_failure_class.map_or(String::new(), |fc| fc.to_string())), - ); - context.set( - context::keys::FAILURE_SIGNATURE, - serde_json::json!(failure_sig - .as_ref() - .map_or(String::new(), |s| s.to_string())), - ); - if let Some(ref pref) = outcome.preferred_label { - context.set(context::keys::PREFERRED_LABEL, serde_json::json!(pref)); - } - - // Step 5: Select next edge (done before checkpoint so we can store next_node_id) - // If the handler specified a direct jump (e.g., parallel -> fan-in), - // bypass edge selection entirely. - let stage_status = outcome.status.to_string(); - let (next_edge, jump_target) = if let Some(ref target) = outcome.jump_to_node { - self.services.emitter.emit(&WorkflowRunEvent::EdgeSelected { - from_node: node.id.clone(), - to_node: target.clone(), - label: None, - condition: None, - reason: "jump".to_string(), - preferred_label: outcome.preferred_label.clone(), - suggested_next_ids: outcome.suggested_next_ids.clone(), - stage_status, - is_jump: true, - }); - (None, Some(target.clone())) - } else { - let selection = select_edge(node, &outcome, &context, graph, node.selection()); - if let Some(sel) = &selection { - self.services.emitter.emit(&WorkflowRunEvent::EdgeSelected { - from_node: node.id.clone(), - to_node: sel.edge.to.clone(), - label: sel.edge.label().map(String::from), - condition: sel.edge.condition().map(String::from), - reason: sel.reason.to_string(), - preferred_label: outcome.preferred_label.clone(), - suggested_next_ids: outcome.suggested_next_ids.clone(), - stage_status, - is_jump: false, - }); - } - (selection.map(|s| s.edge), None) - }; - - // EdgeSelected hook (blocking — can override routing) - let (next_edge, jump_target) = { - let edge_to = jump_target - .as_ref() - .cloned() - .or_else(|| next_edge.as_ref().map(|e| e.to.clone())); - if let Some(ref to) = edge_to { - let mut hook_ctx = HookContext::new( - HookEvent::EdgeSelected, - run_id.clone(), - graph.name.clone(), - ); - hook_ctx.edge_from = Some(node.id.clone()); - hook_ctx.edge_to = Some(to.clone()); - hook_ctx.edge_label = - next_edge.as_ref().and_then(|e| e.label().map(String::from)); - let decision = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - match decision { - HookDecision::Override { - edge_to: new_target, - } => { - // Redirect routing to the hook-specified target - (None, Some(new_target)) - } - HookDecision::Block { reason } => { - let msg = - reason.unwrap_or_else(|| "blocked by EdgeSelected hook".into()); - return Err(FabroError::engine(msg)); - } - _ => (next_edge, jump_target), - } - } else { - (next_edge, jump_target) - } - }; - - let next_node_id_for_checkpoint = jump_target - .as_ref() - .cloned() - .or_else(|| next_edge.map(|e| e.to.clone())); - - // Step 6: Save checkpoint with all state - let mut checkpoint = Checkpoint::from_context( - &context, - &node.id, - completed_nodes.clone(), - node_retries.clone(), - node_outcomes.clone(), - next_node_id_for_checkpoint, - loop_state.loop_failure_signatures.clone(), - loop_state.restart_failure_signatures.clone(), - loop_state.node_visits.clone(), - ); - let checkpoint_path = config.run_dir.join("checkpoint.json"); - if let Err(e) = checkpoint.save(&checkpoint_path) { - self.services.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), - }); - } - - // Step 6b: Write shadow branch first, then run branch commit with trailer - // Skip git checkpoint for the start node — it's a no-op, so the commit is always empty. - if start_node_id.as_deref() != Some(&*node.id) && config.git_checkpoint_enabled { - // Shadow commit (best-effort) - let shadow_sha: Option = if let (Some(_), Some(ref repo_path)) = - (&config.meta_branch, &config.host_repo_path) - { - let store = crate::git::MetadataStore::new(repo_path, &config.git_author); - serde_json::to_vec_pretty(&checkpoint) - .ok() - .and_then(|cp_json| { - let mut extra_entries: Vec<(String, Vec)> = artifact_store - .list() - .iter() - .filter_map(|info| { - info.file_path.as_ref().and_then(|path| { - std::fs::read(path).ok().map(|data| { - (format!("artifacts/{}.json", info.id), data) - }) - }) - }) - .collect(); - extra_entries.extend(crate::git::scan_node_files(&config.run_dir)); - let extra_refs: Vec<(&str, &[u8])> = extra_entries - .iter() - .map(|(k, v)| (k.as_str(), v.as_slice())) - .collect(); - match store.write_checkpoint(&config.run_id, &cp_json, &extra_refs) { - Ok(sha) => Some(sha), - Err(e) => { - self.services.emitter.emit(&WorkflowRunEvent::RunNotice { - level: crate::event::RunNoticeLevel::Warn, - code: "checkpoint_metadata_write_failed".to_string(), - message: format!( - "[node: {}] metadata checkpoint write failed: {e}", - node.id - ), - }); - None - } - } - }) - } else { - None - }; - - // Run branch commit via sandbox - let completed_count = completed_nodes.len(); - let commit_result = git_checkpoint( - &*self.services.sandbox, - &run_id, - &node.id, - &outcome.status.to_string(), - completed_count, - shadow_sha, - &config.checkpoint_exclude_globs, - &config.git_author, - ) - .await; - - match commit_result { - Ok(sha) => { - checkpoint.git_commit_sha = Some(sha.clone()); - if let Err(e) = checkpoint.save(&checkpoint_path) { - self.services.emitter.emit(&WorkflowRunEvent::RunNotice { - level: crate::event::RunNoticeLevel::Warn, - code: "checkpoint_resave_failed".to_string(), - message: format!( - "[node: {}] checkpoint re-save with SHA failed: {e}", - node.id - ), - }); - } - self.services - .emitter - .emit(&WorkflowRunEvent::CheckpointCompleted { - node_id: node.id.clone(), - status: outcome.status.to_string(), - git_commit_sha: Some(sha.clone()), - }); - - self.services.emitter.emit(&WorkflowRunEvent::GitCommit { - node_id: Some(node.id.clone()), - sha: sha.clone(), - }); - - // Push run branch (skip in dry-run mode) - if !config.dry_run { - if let Some(ref branch) = config.run_branch { - let push_ok = if self.services.sandbox.git_push_branch(branch).await - { - true - } else if let Some(ref repo_path) = config.host_repo_path { - let refspec = format!("refs/heads/{branch}"); - git_push_host( - repo_path, - &refspec, - &config.github_app, - "run branch", - ) - .await - } else { - false - }; - self.services.emitter.emit(&WorkflowRunEvent::GitPush { - branch: branch.clone(), - success: push_ok, - }); - } - // Push metadata branch (always from host) - if let (Some(ref meta_branch), Some(ref repo_path)) = - (&config.meta_branch, &config.host_repo_path) - { - let refspec = format!("refs/heads/{meta_branch}"); - let meta_push_ok = git_push_host( - repo_path, - &refspec, - &config.github_app, - "metadata branch", - ) - .await; - self.services.emitter.emit(&WorkflowRunEvent::GitPush { - branch: meta_branch.clone(), - success: meta_push_ok, - }); - } - } - - // Save diff.patch for this stage - let prev = last_git_sha - .as_deref() - .or(config.base_sha.as_deref()) - .unwrap_or(&sha); - let diff_base = prev.to_string(); - let diff_dest = - node_dir(&config.run_dir, &node.id, visit).join("diff.patch"); - - match git_diff(&*self.services.sandbox, &diff_base).await { - Ok(patch) if !patch.is_empty() => { - let _ = std::fs::write(&diff_dest, patch); - } - Ok(_) => {} // empty diff, nothing to write - Err(err) => { - self.services.emitter.emit(&WorkflowRunEvent::RunNotice { - level: crate::event::RunNoticeLevel::Warn, - code: "git_diff_failed".to_string(), - message: format!("[node: {}] git diff failed: {err}", node.id), - }); - } - } - - last_git_sha = Some(sha); - } - Err(e) => { - self.services - .emitter - .emit(&WorkflowRunEvent::CheckpointFailed { - node_id: node.id.clone(), - error: e.clone(), - }); - return Err(FabroError::Engine { - message: format!( - "git checkpoint commit failed for node '{}': {e}", - node.id - ), - failure_class: FailureCategory::Deterministic, - }); - } - } - } else { - // Non-git checkpoint path (start node or git disabled) - self.services - .emitter - .emit(&WorkflowRunEvent::CheckpointCompleted { - node_id: node.id.clone(), - status: outcome.status.to_string(), - git_commit_sha: None, - }); - } - - // CheckpointSaved hook (non-blocking) — fires for both git and non-git paths. - // The Err arm above returns early, so this only runs on success. - { - let mut hook_ctx = HookContext::new( - HookEvent::CheckpointSaved, - run_id.clone(), - graph.name.clone(), - ); - hook_ctx.node_id = Some(node.id.clone()); - let _ = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - } - - // Step 7: Follow selected edge (or direct jump) - if let Some(target) = jump_target { - incoming_edge = None; - current_node_id = target; - continue; - } - match next_edge { - None => { - // Gap #1: Failure routing -- when FAIL and no matching edge, - // check retry_target / fallback_retry_target before terminating - if outcome.status == StageStatus::Fail { - if let Some(retry_target) = get_retry_target(&node.id, graph) { - current_node_id = retry_target; - continue; - } - let duration_ms = millis_u64(run_start.elapsed()); - let error = FabroError::engine(format!( - "stage {} failed with no outgoing fail edge", - node.id - )); - self.services - .emitter - .emit(&WorkflowRunEvent::WorkflowRunFailed { - error: error.clone(), - duration_ms, - git_commit_sha: last_git_sha.clone(), - }); - - self.run_failed_hook( - &run_id, - &graph.name, - &error, - hook_work_dir.as_deref(), - ) - .await; - - return Ok((error.to_fail_outcome(), context)); - } - break; - } - Some(edge) => { - // Track incoming edge for fidelity resolution on the next node - incoming_edge = Some(edge); - // Gap #6: Handle loop_restart by recursively running from the target - if edge.loop_restart() { - // Guard: only transient_infra failures may loop_restart (matches Kilroy) - if let Some(fc) = outcome_failure_class { - if fc != FailureCategory::TransientInfra { - return Err(FabroError::engine(format!( - "loop_restart blocked: failure_class={fc} (requires transient_infra), node={}, failure_reason={}", - node.id, - outcome.failure_reason().unwrap_or("none"), - ))); - } - } - // Circuit breaker: check restart failure signatures - if let Some(ref sig) = failure_sig { - let count = loop_state - .restart_failure_signatures - .entry(sig.clone()) - .or_insert(0); - *count += 1; - let limit = graph.loop_restart_signature_limit(); - if *count >= limit { - return Err(FabroError::engine(format!( - "loop_restart circuit breaker: signature {sig} repeated {count} times (limit {limit})" - ))); - } - } - self.services.emitter.emit(&WorkflowRunEvent::LoopRestart { - from_node: node.id.clone(), - to_node: edge.to.clone(), - }); - return Box::pin(self.run_internal( - graph, - config, - None, - Some(&edge.to), - None, - loop_state, - )) - .await; - } - current_node_id.clone_from(&edge.to); - } - } - } - - // Shut down stall watchdog - if let Some((_, ref shutdown)) = stall_token { - shutdown.cancel(); - } - - let duration_ms = millis_u64(run_start.elapsed()); - let total_cost: Option = { - let sum: f64 = node_outcomes - .values() - .filter_map(|o| o.usage.as_ref()?.cost) - .sum(); - if sum > 0.0 { - Some(sum) - } else { - None - } - }; - - let mut last_outcome = Outcome::success(); - last_outcome.notes = Some("Pipeline completed".to_string()); - - let run_usage: Option = node_outcomes - .values() - .filter_map(|o| o.usage.as_ref().map(fabro_llm::types::Usage::from)) - .reduce(|a, b| a + b); - - self.services - .emitter - .emit(&WorkflowRunEvent::WorkflowRunCompleted { - duration_ms, - artifact_count: artifact_store.list().len(), - status: last_outcome.status.to_string(), - total_cost, - final_git_commit_sha: last_git_sha.clone(), - usage: run_usage, - }); - - // RunComplete hook (non-blocking) - { - let hook_ctx = - HookContext::new(HookEvent::RunComplete, run_id.clone(), graph.name.clone()); - let _ = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - } - - // Write final.patch: comprehensive diff from base_sha to HEAD - if config.git_checkpoint_enabled { - if let Some(ref base) = config.base_sha { - if let Ok(patch) = git_diff(&*self.services.sandbox, base).await { - if !patch.is_empty() { - let _ = std::fs::write(config.run_dir.join("final.patch"), patch); - } - } - } - } - - Ok((last_outcome, context)) - } } #[cfg(test)] @@ -5207,7 +3872,9 @@ mod tests { _run_dir: &Path, _services: &crate::handler::EngineServices, ) -> std::result::Result { - let n = self.counter.fetch_add(1, Ordering::Relaxed); + let n = self + .counter + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); let reason = VARYING_REASONS[n % VARYING_REASONS.len()]; Ok(Outcome::fail_classify(reason)) } @@ -6033,7 +4700,9 @@ mod tests { _run_dir: &Path, _services: &crate::handler::EngineServices, ) -> std::result::Result { - let n = self.call_count.fetch_add(1, Ordering::Relaxed); + let n = self + .call_count + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); if n == 0 { Err(FabroError::handler("transient failure")) } else { diff --git a/lib/crates/fabro-workflows/src/handler/mod.rs b/lib/crates/fabro-workflows/src/handler/mod.rs index 449d7b4b9..7d7809069 100644 --- a/lib/crates/fabro-workflows/src/handler/mod.rs +++ b/lib/crates/fabro-workflows/src/handler/mod.rs @@ -33,7 +33,7 @@ pub struct EngineServices { pub emitter: Arc, pub sandbox: Arc, /// Git state for the current run. Set via `set_git_state` at the start of - /// `run_internal` and read by parallel/fan-in handlers. + /// `run_via_core` and read by parallel/fan-in handlers. pub(crate) git_state: std::sync::RwLock>>, /// Hook runner for user-defined lifecycle hooks. pub hook_runner: Option>,