diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index e528b916d..b3b0f00a5 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -5,7 +5,7 @@ use std::sync::{Arc, Mutex}; use std::time::Instant; use anyhow::{bail, Context}; -use chrono::{Local, Utc}; +use chrono::Local; use clap::{Args, ValueEnum}; use fabro_agent::{ DockerSandbox, DockerSandboxConfig, LocalSandbox, Sandbox, WorktreeConfig, WorktreeSandbox, @@ -17,15 +17,13 @@ use fabro_model::{Catalog, FallbackTarget, Provider}; use fabro_util::terminal::Styles; use fabro_workflows::backend::{AgentApiBackend, AgentCliBackend, BackendRouter}; use fabro_workflows::checkpoint::Checkpoint; -use fabro_workflows::conclusion::Conclusion; use fabro_workflows::cost::{compute_stage_cost, format_cost}; use fabro_workflows::devcontainer_bridge; use fabro_workflows::engine::{GitCheckpointSettings, RunSettings, WorkflowRunEngine}; use fabro_workflows::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use fabro_workflows::git::GitSyncStatus; use fabro_workflows::handler::default_registry; -use fabro_workflows::outcome::{Outcome, OutcomeExt, StageStatus}; -use fabro_workflows::run_status::{RunStatus, StatusReason}; +use fabro_workflows::outcome::StageStatus; use fabro_workflows::sandbox_provider::SandboxProvider; use indicatif::HumanDuration; use std::time::Duration; @@ -37,6 +35,10 @@ use crate::commands::shared::{ format_tokens_human, print_diagnostics, read_workflow_file, relative_path, tilde_path, }; +pub(crate) use fabro_workflows::pipeline::{ + build_conclusion, classify_engine_result, persist_terminal_outcome, write_finalize_commit, +}; + #[derive(Debug, Clone, Copy, ValueEnum)] pub enum CliSandboxProvider { Local, @@ -2066,128 +2068,6 @@ pub(crate) fn emit_run_notice( }); } -pub(crate) fn classify_engine_result( - engine_result: &Result, -) -> (StageStatus, Option, RunStatus, Option) { - match engine_result { - Ok(outcome) => { - let status = outcome.status.clone(); - let failure_reason = outcome.failure_reason().map(String::from); - let (run_status, status_reason) = match status { - StageStatus::Success | StageStatus::Skipped => { - (RunStatus::Succeeded, Some(StatusReason::Completed)) - } - StageStatus::PartialSuccess => { - (RunStatus::Succeeded, Some(StatusReason::PartialSuccess)) - } - StageStatus::Fail | StageStatus::Retry => { - (RunStatus::Failed, Some(StatusReason::WorkflowError)) - } - }; - (status, failure_reason, run_status, status_reason) - } - Err(fabro_workflows::error::FabroError::Cancelled) => ( - StageStatus::Fail, - Some("Cancelled".to_string()), - RunStatus::Failed, - Some(StatusReason::Cancelled), - ), - Err(err) => ( - StageStatus::Fail, - Some(err.to_string()), - RunStatus::Failed, - Some(StatusReason::WorkflowError), - ), - } -} - -pub(crate) fn build_conclusion( - run_dir: &Path, - status: StageStatus, - failure_reason: Option, - run_duration_ms: u64, - final_git_commit_sha: Option, -) -> Conclusion { - let checkpoint = Checkpoint::load(&run_dir.join("checkpoint.json")).ok(); - let stage_durations = fabro_retro::retro::extract_stage_durations(run_dir); - - let mut total_input_tokens: i64 = 0; - let mut total_output_tokens: i64 = 0; - let mut total_cache_read_tokens: i64 = 0; - let mut total_cache_write_tokens: i64 = 0; - let mut total_reasoning_tokens: i64 = 0; - let mut has_pricing = false; - - let (stages, total_cost, total_retries) = if let Some(ref cp) = checkpoint { - let mut stages = Vec::new(); - let mut cost_sum: Option = None; - let mut retries_sum: u32 = 0; - - for node_id in &cp.completed_nodes { - let outcome = cp.node_outcomes.get(node_id); - let retries = cp - .node_retries - .get(node_id) - .copied() - .unwrap_or(1) - .saturating_sub(1); - retries_sum += retries; - - let cost = outcome.and_then(|o| o.usage.as_ref()).and_then(|u| u.cost); - if let Some(c) = cost { - *cost_sum.get_or_insert(0.0) += c; - has_pricing = true; - } - - if let Some(usage) = outcome.and_then(|o| o.usage.as_ref()) { - total_input_tokens += usage.input_tokens; - total_output_tokens += usage.output_tokens; - total_cache_read_tokens += usage.cache_read_tokens.unwrap_or(0); - total_cache_write_tokens += usage.cache_write_tokens.unwrap_or(0); - total_reasoning_tokens += usage.reasoning_tokens.unwrap_or(0); - } - - stages.push(fabro_workflows::conclusion::StageSummary { - stage_id: node_id.clone(), - stage_label: node_id.clone(), - duration_ms: stage_durations.get(node_id).copied().unwrap_or(0), - cost, - retries, - }); - } - (stages, cost_sum, retries_sum) - } else { - (vec![], None, 0) - }; - - Conclusion { - timestamp: Utc::now(), - status, - duration_ms: run_duration_ms, - failure_reason, - final_git_commit_sha, - stages, - total_cost, - total_retries, - total_input_tokens, - total_output_tokens, - total_cache_read_tokens, - total_cache_write_tokens, - total_reasoning_tokens, - has_pricing, - } -} - -pub(crate) fn persist_terminal_outcome( - run_dir: &Path, - conclusion: &Conclusion, - run_status: RunStatus, - status_reason: Option, -) { - let _ = conclusion.save(&run_dir.join("conclusion.json")); - fabro_workflows::run_status::write_run_status(run_dir, run_status, status_reason); -} - /// Print a summary of the completed run from `conclusion.json` and `pull_request.json`. /// /// Used by the unified create+start+attach path in `main.rs` to display @@ -2677,43 +2557,6 @@ async fn run_preflight( } } -/// Write a finalize commit to the shadow branch with retro.json and final node files. -/// -/// This captures the last diff.patch (written after the final checkpoint) and retro.json. -/// Best-effort: errors are logged as warnings. -pub(crate) async fn write_finalize_commit(config: &RunSettings, run_dir: &std::path::Path) { - let (Some(meta_branch), Some(repo_path)) = ( - config.git.as_ref().and_then(|g| g.meta_branch.as_ref()), - config.host_repo_path.as_ref(), - ) else { - return; - }; - - let store = fabro_workflows::git::MetadataStore::new(repo_path, &config.git_author); - let mut entries = fabro_workflows::git::scan_node_files(run_dir); - if let Ok(retro_bytes) = std::fs::read(run_dir.join("retro.json")) { - entries.push(("retro.json".to_string(), retro_bytes)); - } - let refs: Vec<(&str, &[u8])> = entries - .iter() - .map(|(k, v)| (k.as_str(), v.as_slice())) - .collect(); - if let Err(e) = store.write_files(&config.run_id, &refs, "finalize run") { - tracing::warn!(error = %e, "Failed to write finalize commit to metadata branch"); - return; - } - - // Push the finalize commit - let refspec = format!("refs/heads/{meta_branch}"); - fabro_workflows::engine::git_push_host( - repo_path, - &refspec, - &config.github_app, - "finalize metadata", - ) - .await; -} - /// Generate a retro report for a completed workflow run. /// /// Derives a basic retro from the checkpoint, then optionally runs the retro agent @@ -2734,185 +2577,85 @@ pub(crate) async fn generate_retro( styles: &'static Styles, emitter: Option>, ) { - let cp = match Checkpoint::load(&run_dir.join("checkpoint.json")) { - Ok(cp) => cp, - Err(e) => { - eprintln!( - "{} Could not load checkpoint, skipping retro: {e}", - styles.yellow.apply_to("Warning:"), - ); - return; - } - }; - - let completed_stages = fabro_workflows::build_completed_stages(&cp, failed); - let stage_durations = fabro_retro::retro::extract_stage_durations(run_dir); - let mut retro = fabro_retro::retro::derive_retro( - run_id, - workflow_name, - goal, - completed_stages, - run_duration_ms, - &stage_durations, - ); - - match retro.save(run_dir) { - Ok(()) => {} - Err(e) => { - eprintln!( - "{} Failed to save initial retro: {e}", - styles.yellow.apply_to("Warning:"), - ); - } - } - - // Run retro agent session eprintln!("\n{}", styles.bold.apply_to("=== Retro ===")); - - let retro_start = std::time::Instant::now(); - if let Some(ref em) = emitter { - em.emit(&fabro_workflows::event::WorkflowRunEvent::RetroStarted); - } else { + if emitter.is_none() { eprintln!( "{}", styles.dim.apply_to(format!("Running retro ({model})...")) ); } - let narrative_result = if dry_run_mode { - Ok(fabro_retro::retro_agent::dry_run_narrative()) - } else if let Some(client) = llm_client { - let emitter_clone = emitter.clone(); - let event_callback: Option> = - emitter_clone.map( - |em| -> Arc { - Arc::new(move |event: fabro_agent::SessionEvent| { - em.touch(); + let retro_start = std::time::Instant::now(); + let retro = fabro_workflows::pipeline::run_retro(&fabro_workflows::pipeline::RetroOptions { + run_id: run_id.to_string(), + workflow_name: workflow_name.to_string(), + goal: goal.to_string(), + run_dir: run_dir.to_path_buf(), + sandbox: Arc::clone(sandbox), + emitter, + failed, + run_duration_ms, + enabled: true, + dry_run: dry_run_mode, + llm_client: llm_client.cloned(), + provider: provider_enum, + model: model.to_string(), + }) + .await; - if !matches!( - &event.event, - fabro_agent::AgentEvent::SessionStarted - | fabro_agent::AgentEvent::SessionEnded - | fabro_agent::AgentEvent::AssistantTextStart - | fabro_agent::AgentEvent::AssistantOutputReplace { .. } - | fabro_agent::AgentEvent::TextDelta { .. } - | fabro_agent::AgentEvent::ReasoningDelta { .. } - | fabro_agent::AgentEvent::ToolCallOutputDelta { .. } - | fabro_agent::AgentEvent::SkillExpanded { .. } - ) { - em.emit(&fabro_workflows::event::WorkflowRunEvent::Agent { - stage: "retro".to_string(), - event: event.event.clone(), - }); - } - }) - }, - ); - fabro_retro::retro_agent::run_retro_agent( - sandbox, - run_dir, - client, - provider_enum, - model, - event_callback, - ) - .await + let retro_dur = run_progress::format_duration_short(retro_start.elapsed()); + if let Some(retro) = retro { + let smoothness_str = retro + .smoothness + .as_ref() + .map(|s| s.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + let outcome_str = retro.outcome.as_deref().unwrap_or("No outcome recorded"); + let line1_content = format!("Retro: {smoothness_str} \u{2014} {outcome_str}"); + let term_width = console::Term::stderr().size().1 as usize; + let dur_len = retro_dur.len(); + let pad1 = term_width.saturating_sub(line1_content.len() + dur_len); + eprintln!( + "{} {}{:pad1$}{}", + styles.bold.apply_to("Retro:"), + styles + .dim + .apply_to(format!("{smoothness_str} \u{2014} {outcome_str}")), + "", + styles.dim.apply_to(&retro_dur), + ); + + let friction_count = retro.friction_points.as_ref().map(|v| v.len()).unwrap_or(0); + let open_count = retro.open_items.as_ref().map(|v| v.len()).unwrap_or(0); + if friction_count > 0 || open_count > 0 { + let mut parts = Vec::new(); + if friction_count > 0 { + let noun = if friction_count == 1 { + "friction point" + } else { + "friction points" + }; + parts.push(format!("{friction_count} {noun}")); + } + if open_count > 0 { + let noun = if open_count == 1 { + "open item" + } else { + "open items" + }; + parts.push(format!("{open_count} {noun}")); + } + eprintln!(" {}", styles.dim.apply_to(parts.join(" \u{00b7} "))); + } + + let retro_path = format!("{}/retro.json", tilde_path(run_dir)); + eprintln!( + " {} {}", + styles.dim.apply_to("Retro saved to"), + styles.underline.apply_to(&retro_path), + ); } else { - Err(anyhow::anyhow!("No LLM client available")) - }; - let retro_dur_elapsed = retro_start.elapsed(); - - if let Some(ref em) = emitter { - match &narrative_result { - Ok(_) => { - em.emit(&fabro_workflows::event::WorkflowRunEvent::RetroCompleted { - duration_ms: retro_dur_elapsed.as_millis() as u64, - }); - } - Err(e) => { - em.emit(&fabro_workflows::event::WorkflowRunEvent::RetroFailed { - error: e.to_string(), - duration_ms: retro_dur_elapsed.as_millis() as u64, - }); - } - } - } - - let retro_dur = run_progress::format_duration_short(retro_dur_elapsed); - - match narrative_result { - Ok(narrative) => { - retro.apply_narrative(narrative); - match retro.save(run_dir) { - Ok(()) => { - // Line 1: smoothness + outcome with right-aligned duration - let smoothness_str = retro - .smoothness - .as_ref() - .map(|s| s.to_string()) - .unwrap_or_else(|| "unknown".to_string()); - let outcome_str = retro.outcome.as_deref().unwrap_or("No outcome recorded"); - let line1_content = format!("Retro: {smoothness_str} \u{2014} {outcome_str}"); - let term_width = console::Term::stderr().size().1 as usize; - let dur_len = retro_dur.len(); - let pad1 = term_width.saturating_sub(line1_content.len() + dur_len); - eprintln!( - "{} {}{:pad1$}{}", - styles.bold.apply_to("Retro:"), - styles - .dim - .apply_to(format!("{smoothness_str} \u{2014} {outcome_str}")), - "", - styles.dim.apply_to(&retro_dur), - ); - - // Line 2: friction + open items (only if non-zero) - let friction_count = - retro.friction_points.as_ref().map(|v| v.len()).unwrap_or(0); - let open_count = retro.open_items.as_ref().map(|v| v.len()).unwrap_or(0); - if friction_count > 0 || open_count > 0 { - let mut parts = Vec::new(); - if friction_count > 0 { - let noun = if friction_count == 1 { - "friction point" - } else { - "friction points" - }; - parts.push(format!("{friction_count} {noun}")); - } - if open_count > 0 { - let noun = if open_count == 1 { - "open item" - } else { - "open items" - }; - parts.push(format!("{open_count} {noun}")); - } - eprintln!(" {}", styles.dim.apply_to(parts.join(" \u{00b7} "))); - } - - // Line 3: file path - let retro_path = format!("{}/retro.json", tilde_path(run_dir)); - eprintln!( - " {} {}", - styles.dim.apply_to("Retro saved to"), - styles.underline.apply_to(&retro_path), - ); - } - Err(e) => { - eprintln!( - "{} Failed to save retro with narrative: {e}", - styles.yellow.apply_to("Warning:"), - ); - } - } - } - Err(e) => { - eprintln!( - "{}", - styles.dim.apply_to(format!("Retro agent skipped: {e}")), - ); - } + eprintln!("{}", styles.dim.apply_to("Retro unavailable")); } } diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index 4ddfd855d..a61362327 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -1,822 +1,54 @@ use std::collections::HashMap; -use std::path::{Path, PathBuf}; -use std::sync::atomic::AtomicBool; +use std::path::Path; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Instant; + +#[cfg(test)] +use std::path::PathBuf; +#[cfg(test)] +use std::sync::atomic::AtomicBool; -use chrono::Utc; use fabro_agent::Sandbox; use fabro_core::executor::ExecutorBuilder; use fabro_core::state::RunState; -use fabro_util::backoff::BackoffPolicy; -use rand::Rng; use tokio_util::sync::CancellationToken; -use fabro_git_storage::trailerlink::{self, Trailer}; - -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, Result}; +#[cfg(test)] +use crate::error::FailureCategory; +use crate::error::{FabroError, Result}; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::handler::{EngineServices, HandlerRegistry}; -use crate::outcome::{Outcome, OutcomeExt, StageStatus}; -use fabro_config::{config::FabroConfig, run::PullRequestConfig}; -use fabro_graphviz::graph::{Edge, Graph, Node}; +#[cfg(test)] +use crate::outcome::OutcomeExt; +use crate::outcome::{Outcome, StageStatus}; +#[cfg(test)] +use fabro_config::config::FabroConfig; +use fabro_graphviz::graph::Graph; +#[cfg(test)] +use fabro_graphviz::graph::{Edge, Node}; use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner}; use fabro_interview::Interviewer; -/// Populate node-related fields on a `HookContext` from a graph `Node`. -pub(crate) fn set_hook_node(ctx: &mut HookContext, node: &Node) { - ctx.node_id = Some(node.id.clone()); - ctx.node_label = Some(node.label().to_string()); - ctx.handler_type = node.handler_type().map(String::from); -} - -/// Classify the failure mode of a completed outcome. -/// -/// Returns `None` for `Success`, `PartialSuccess`, and `Skipped` outcomes. -/// For failures, checks (in priority order): -/// 1. Handler hint in `context_updates["failure_class"]` -/// 2. String heuristics on `failure_reason` -/// 3. Default to `Deterministic` -#[must_use] -pub(crate) fn classify_outcome(outcome: &Outcome) -> Option { - match outcome.status { - StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped => None, - StageStatus::Fail | StageStatus::Retry => outcome - .failure_category() - .or(Some(FailureCategory::Deterministic)), - } -} - -// --- Retry policy types --- - -/// Retry policy for node execution. -#[derive(Clone, Debug)] -pub struct RetryPolicy { - pub max_attempts: u32, - pub backoff: BackoffPolicy, -} - -impl RetryPolicy { - const DEFAULT_BACKOFF: BackoffPolicy = BackoffPolicy { - initial_delay: Duration::from_millis(5_000), - factor: 2.0, - max_delay: Duration::from_millis(60_000), - jitter: true, - }; - - /// No retries -- fail immediately. - #[must_use] - pub fn none() -> Self { - Self { - max_attempts: 1, - backoff: Self::DEFAULT_BACKOFF, - } - } - - /// Standard retry policy: 5 attempts, 5s initial, 2x factor. - #[must_use] - pub fn standard() -> Self { - Self { - max_attempts: 5, - backoff: Self::DEFAULT_BACKOFF, - } - } - - /// Aggressive retry: 5 attempts, 500ms initial, 2x factor. - #[must_use] - pub fn aggressive() -> Self { - Self { - max_attempts: 5, - backoff: BackoffPolicy { - initial_delay: Duration::from_millis(500), - ..Self::DEFAULT_BACKOFF - }, - } - } - - /// Linear retry: 3 attempts, 500ms fixed delay. - #[must_use] - pub fn linear() -> Self { - Self { - max_attempts: 3, - backoff: BackoffPolicy { - initial_delay: Duration::from_millis(500), - factor: 1.0, - ..Self::DEFAULT_BACKOFF - }, - } - } - - /// Patient retry: 3 attempts, 2000ms initial, 3x factor. - #[must_use] - pub fn patient() -> Self { - Self { - max_attempts: 3, - backoff: BackoffPolicy { - initial_delay: Duration::from_millis(2000), - factor: 3.0, - ..Self::DEFAULT_BACKOFF - }, - } - } -} - -/// Build a retry policy from node and graph attributes. -/// If the node has a `retry_policy` attribute naming a preset, use that. -/// Otherwise, fall back to `max_retries` / graph default. -pub(crate) fn build_retry_policy(node: &Node, graph: &Graph) -> RetryPolicy { - if let Some(preset) = node.retry_policy() { - match preset { - "none" => return RetryPolicy::none(), - "standard" => return RetryPolicy::standard(), - "aggressive" => return RetryPolicy::aggressive(), - "linear" => return RetryPolicy::linear(), - "patient" => return RetryPolicy::patient(), - _ => {} // Unknown preset, fall through to max_retries behavior - } - } - let max_retries = node - .max_retries() - .unwrap_or_else(|| graph.default_max_retries()); - // max_retries=0 means 1 attempt (no retries) - let max_attempts = u32::try_from(max_retries + 1).unwrap_or(1).max(1); - RetryPolicy { - max_attempts, - backoff: RetryPolicy::DEFAULT_BACKOFF, - } -} - -// --- Fidelity resolution (spec 5.4) --- - -/// Resolve the context fidelity for a node, following the precedence: -/// 1. Incoming edge `fidelity` attribute -/// 2. Target node `fidelity` attribute -/// 3. Graph `default_fidelity` attribute -/// 4. Default: Compact -#[must_use] -pub fn resolve_fidelity( - incoming_edge: Option<&Edge>, - node: &Node, - graph: &Graph, -) -> context::keys::Fidelity { - let (resolved, source) = if let Some(f) = incoming_edge - .and_then(|e| e.fidelity()) - .and_then(|s| s.parse().ok()) - { - (f, "edge") - } else if let Some(f) = node.fidelity().and_then(|s| s.parse().ok()) { - (f, "node") - } else if let Some(f) = graph.default_fidelity().and_then(|s| s.parse().ok()) { - (f, "graph") - } else { - (context::keys::Fidelity::default(), "default") - }; - - tracing::debug!( - node = %node.id, - fidelity = %resolved, - source = source, - "Fidelity resolved" - ); - - resolved -} - -// --- Thread ID resolution (spec 5.4) --- - -/// Resolve the thread ID for a node, following the precedence: -/// 1. Incoming edge `thread_id` attribute -/// 2. Target node `thread_id` attribute -/// 3. Graph-level default thread -/// 4. Derived class from enclosing subgraph (first class from the node's classes list) -/// 5. Fallback to previous node ID -#[must_use] -pub fn resolve_thread_id( - incoming_edge: Option<&Edge>, - node: &Node, - graph: &Graph, - previous_node_id: Option<&str>, -) -> Option { - // Step 1: Edge thread_id - if let Some(edge) = incoming_edge { - if let Some(tid) = edge.thread_id() { - return Some(tid.to_string()); - } - } - // Step 2: Node thread_id - if let Some(tid) = node.thread_id() { - return Some(tid.to_string()); - } - // Step 3: Graph-level default thread - if let Some(tid) = graph.default_thread() { - return Some(tid.to_string()); - } - // Step 4: Derived class from enclosing subgraph - if let Some(first_class) = node.classes.first() { - return Some(first_class.clone()); - } - // Step 5: Fallback to previous node ID - previous_node_id.map(String::from) -} - -// --- Run directory helpers (spec 5.6) --- - -/// Write start.json at the start of a workflow run. Returns the StartRecord. -pub(crate) fn write_start_record( - run_dir: &Path, - settings: &RunSettings, -) -> crate::start_record::StartRecord { - let git_state = settings.git.as_ref(); - let record = crate::start_record::StartRecord { - run_id: settings.run_id.clone(), - start_time: Utc::now(), - run_branch: git_state.and_then(|g| g.run_branch.clone()), - base_sha: git_state.and_then(|g| g.base_sha.clone()), - }; - let _ = std::fs::create_dir_all(run_dir); - let _ = record.save(run_dir); - record -} - -/// Return the directory for a node's logs. -/// -/// First visit (`visit <= 1`): `{run_dir}/nodes/{node_id}` -/// Subsequent visits: `{run_dir}/nodes/{node_id}-visit_{visit}` -pub fn node_dir(run_dir: &Path, node_id: &str, visit: usize) -> PathBuf { - if visit <= 1 { - run_dir.join("nodes").join(node_id) - } else { - run_dir - .join("nodes") - .join(format!("{node_id}-visit_{visit}")) - } -} - -/// Read the workflow visit ordinal from context. -/// -/// The raw context value is `0` when unset; workflow execution code treats -/// missing counts as the first visit for stage/log naming. -pub fn visit_from_context(context: &Context) -> usize { - context.node_visit_count().max(1) -} - -/// Write status.json for a completed node into {`run_dir}/nodes/{node_id}/status.json`. -pub(crate) fn write_node_status(run_dir: &Path, node_id: &str, visit: usize, outcome: &Outcome) { - let node_dir = node_dir(run_dir, node_id, visit); - let _ = std::fs::create_dir_all(&node_dir); - let status = serde_json::json!({ - "status": outcome.status.to_string(), - "notes": outcome.notes, - "failure_reason": outcome.failure_reason(), - "timestamp": Utc::now().to_rfc3339(), - }); - if let Ok(json) = serde_json::to_string_pretty(&status) { - let _ = std::fs::write(node_dir.join("status.json"), json); - } -} - -// --- Edge selection --- - -/// Normalize a label for comparison: lowercase, trim, strip accelerator prefixes. -/// Patterns: "[Y] ", "Y) ", "Y - " -fn normalize_label(label: &str) -> String { - let s = label.trim().to_lowercase(); - // Strip "[X] " prefix - if s.starts_with('[') { - if let Some(rest) = s - .strip_prefix('[') - .and_then(|s| s.find(']').map(|i| s[i + 1..].trim_start().to_string())) - { - return rest; - } - } - // Strip "X) " prefix - if s.len() >= 2 { - let bytes = s.as_bytes(); - if bytes.get(1) == Some(&b')') { - return s[2..].trim_start().to_string(); - } - } - // Strip "X - " prefix - if s.len() >= 3 { - if let Some(rest) = s.get(1..).and_then(|r| r.strip_prefix(" - ")) { - return rest.to_string(); - } - } - s -} - -/// Pick the best edge by highest weight, then lexical target node ID tiebreak. -fn best_by_weight_then_lexical<'a>(edges: &[&'a Edge]) -> Option<&'a Edge> { - if edges.is_empty() { - return None; - } - let mut best = edges[0]; - for &edge in &edges[1..] { - if edge.weight() > best.weight() || (edge.weight() == best.weight() && edge.to < best.to) { - best = edge; - } - } - Some(best) -} - -/// Pick a random edge using weighted-random selection. -/// Edges with `weight <= 0` are treated as weight 1 for probability calculation. -fn weighted_random<'a>(edges: &[&'a Edge]) -> Option<&'a Edge> { - if edges.is_empty() { - return None; - } - if edges.len() == 1 { - return Some(edges[0]); - } - let weights: Vec = edges - .iter() - .map(|e| { - let w = e.weight(); - if w <= 0 { - 1.0 - } else { - w as f64 - } - }) - .collect(); - let total: f64 = weights.iter().sum(); - let mut rng = rand::thread_rng(); - let mut roll: f64 = rng.gen_range(0.0..total); - for (i, &w) in weights.iter().enumerate() { - roll -= w; - if roll < 0.0 { - return Some(edges[i]); - } - } - Some(edges[edges.len() - 1]) -} - -/// Dispatch to the appropriate edge-picking strategy. -fn pick_edge<'a>(edges: &[&'a Edge], selection: &str) -> Option<&'a Edge> { - match selection { - "random" => weighted_random(edges), - _ => best_by_weight_then_lexical(edges), - } -} - -/// Select the next edge from a node's outgoing edges (spec Section 3.3). -#[must_use] -/// Result of edge selection: the chosen edge and the reason it was selected. -pub struct EdgeSelection<'a> { - pub edge: &'a Edge, - pub reason: &'static str, -} - -fn blocks_unconditional_failure_fallthrough(node: &Node, outcome: &Outcome) -> bool { - node.handler_type() == Some("human") - && outcome.status == StageStatus::Fail - && outcome.preferred_label.is_none() - && outcome.suggested_next_ids.is_empty() -} - -pub fn select_edge<'a>( - node: &Node, - outcome: &Outcome, - context: &Context, - graph: &'a Graph, - selection: &str, -) -> Option> { - let node_id = &node.id; - let edges = graph.outgoing_edges(node_id); - if edges.is_empty() { - return None; - } - - // Step 1: Condition matching - let condition_matched: Vec<&Edge> = edges - .iter() - .filter(|e| { - e.condition() - .is_some_and(|c| !c.is_empty() && evaluate_condition(c, outcome, context)) - }) - .copied() - .collect(); - if !condition_matched.is_empty() { - return pick_edge(&condition_matched, selection).map(|edge| EdgeSelection { - edge, - reason: "condition", - }); - } - - // Step 2: Preferred label match (unconditional edges only) - if let Some(pref) = &outcome.preferred_label { - let normalized_pref = normalize_label(pref); - for edge in &edges { - if edge.condition().is_none_or(str::is_empty) { - if let Some(label) = edge.label() { - if normalize_label(label) == normalized_pref { - return Some(EdgeSelection { - edge, - reason: "preferred_label", - }); - } - } - } - } - } - - // Step 3: Suggested next IDs (unconditional edges only) - for suggested_id in &outcome.suggested_next_ids { - for edge in &edges { - if edge.condition().is_none_or(str::is_empty) && edge.to == *suggested_id { - return Some(EdgeSelection { - edge, - reason: "suggested_next", - }); - } - } - } - - if blocks_unconditional_failure_fallthrough(node, outcome) { - return None; - } - - // Step 4 & 5: Weight with lexical tiebreak (unconditional edges only) - let unconditional: Vec<&Edge> = edges - .iter() - .filter(|e| e.condition().is_none_or(str::is_empty)) - .copied() - .collect(); - if !unconditional.is_empty() { - return pick_edge(&unconditional, selection).map(|edge| EdgeSelection { - edge, - reason: "unconditional", - }); - } - - None -} - -// --- Goal gate enforcement --- - -/// Check if all goal gates have been satisfied. -/// Returns Ok(()) if all gates passed, or Err with the failed node ID. -pub(crate) fn check_goal_gates( - graph: &Graph, - node_outcomes: &HashMap, -) -> std::result::Result<(), String> { - for (node_id, outcome) in node_outcomes { - if let Some(node) = graph.nodes.get(node_id) { - if node.goal_gate() - && outcome.status != StageStatus::Success - && outcome.status != StageStatus::PartialSuccess - { - return Err(node_id.clone()); - } - } - } - Ok(()) -} - -/// Resolve the retry target for a failed goal gate node. -pub(crate) fn get_retry_target(failed_node_id: &str, graph: &Graph) -> Option { - if let Some(node) = graph.nodes.get(failed_node_id) { - // Node-level retry_target - if let Some(target) = node.retry_target() { - if graph.nodes.contains_key(target) { - return Some(target.to_string()); - } - } - // Node-level fallback_retry_target - if let Some(target) = node.fallback_retry_target() { - if graph.nodes.contains_key(target) { - return Some(target.to_string()); - } - } - } - // Graph-level retry_target - if let Some(target) = graph.retry_target() { - if graph.nodes.contains_key(target) { - return Some(target.to_string()); - } - } - // Graph-level fallback_retry_target - if let Some(target) = graph.fallback_retry_target() { - if graph.nodes.contains_key(target) { - return Some(target.to_string()); - } - } - None -} - -/// Check whether a node is a terminal (exit) node. -pub(crate) fn is_terminal(node: &Node) -> bool { - node.shape() == "Msquare" || node.handler_type() == Some("exit") -} - -pub(crate) fn node_script(node: &Node) -> Option { - node.attrs - .get("script") - .or_else(|| node.attrs.get("tool_command")) - .and_then(|v| v.as_str()) - .map(String::from) -} - -// --- Workflow run engine --- - -/// Captured git state for a workflow run, shared with handlers. -#[derive(Debug, Clone)] -pub struct GitState { - pub run_id: String, - pub base_sha: String, - pub run_branch: Option, - pub meta_branch: Option, - pub checkpoint_exclude_globs: Vec, - pub git_author: crate::git::GitAuthor, -} - -pub const GIT_REMOTE: &str = "git -c maintenance.auto=0 -c gc.auto=0"; - -/// Shell-escape a string using `shlex::try_quote` (POSIX-safe). -fn shell_quote(s: &str) -> String { - shlex::try_quote(s).map_or_else( - |_| format!("'{}'", s.replace('\'', "'\\''")), - |q| q.to_string(), - ) -} - -/// Run a git checkpoint commit via the sandbox. -#[allow(clippy::too_many_arguments)] -pub async fn git_checkpoint( - sandbox: &dyn Sandbox, - run_id: &str, - node_id: &str, - status: &str, - completed_count: usize, - shadow_sha: Option, - exclude_globs: &[String], - author: &crate::git::GitAuthor, -) -> std::result::Result { - // Stage everything, always excluding EXCLUDE_DIRS plus any user-configured globs - let mut all_excludes: Vec = asset_snapshot::EXCLUDE_DIRS - .iter() - .map(|d| format!("**/{d}/**")) - .collect(); - all_excludes.extend(exclude_globs.iter().cloned()); - - let pathspecs: Vec = all_excludes - .iter() - .map(|g| format!("':(glob,exclude){g}'")) - .collect(); - let add_cmd = format!("{GIT_REMOTE} add -A -- . {}", pathspecs.join(" ")); - let add_result = sandbox - .exec_command(&add_cmd, 30_000, None, None, None) - .await; - match &add_result { - Ok(r) if r.exit_code == 0 => {} - Ok(r) => { - return Err(format!( - "git add failed (exit {}): {}{}", - r.exit_code, r.stdout, r.stderr - )); - } - Err(e) => return Err(format!("git add failed: {e}")), - } - - // Build commit message with trailers (same format as checkpoint_commit in git.rs) - let subject = format!("fabro({run_id}): {node_id} ({status})"); - let completed_str = completed_count.to_string(); - let mut trailers = vec![ - Trailer { - key: "Fabro-Run", - value: run_id, - }, - Trailer { - key: "Fabro-Completed", - value: &completed_str, - }, - ]; - let shadow_sha_ref = shadow_sha.as_deref().unwrap_or(""); - if shadow_sha.is_some() { - trailers.push(Trailer { - key: "Fabro-Checkpoint", - value: shadow_sha_ref, - }); - } - let mut message = trailerlink::format_message(&subject, "", &trailers); - author.append_footer(&mut message); - - // Write message to a unique temp file to avoid races between concurrent local runs - let msg_path = format!("/tmp/fabro-commit-msg-{run_id}-{node_id}"); - if let Err(e) = sandbox.write_file(&msg_path, &message).await { - return Err(format!("failed to write commit message file: {e}")); - } - - // Commit with configured identity using the message file - let commit_cmd = format!( - "{GIT_REMOTE} -c user.name={name} -c user.email={email} commit --allow-empty -F {msg_path}", - name = shell_quote(&author.name), - email = shell_quote(&author.email), - ); - let commit_result = sandbox - .exec_command(&commit_cmd, 30_000, None, None, None) - .await; - match &commit_result { - Ok(r) if r.exit_code == 0 => {} - Ok(r) => { - return Err(format!( - "git commit failed (exit {}): {}{}", - r.exit_code, r.stdout, r.stderr - )); - } - Err(e) => return Err(format!("git commit failed: {e}")), - } - - // Get the new HEAD SHA - let sha_cmd = format!("{GIT_REMOTE} rev-parse HEAD"); - let sha_result = sandbox - .exec_command(&sha_cmd, 10_000, None, None, None) - .await; - match sha_result { - Ok(r) if r.exit_code == 0 => Ok(r.stdout.trim().to_string()), - Ok(r) => Err(format!( - "git rev-parse HEAD failed (exit {}): {}{}", - r.exit_code, r.stdout, r.stderr - )), - Err(e) => Err(format!("git rev-parse HEAD failed: {e}")), - } -} - -/// Push a refspec from the host repo to origin (best-effort). -/// -/// Authenticates via a GitHub App installation token so we don't depend -/// on the host's ambient git credentials. -pub async fn git_push_host( - repo_path: &Path, - refspec: &str, - github_app: &Option, - label: &str, -) -> bool { - let (origin_url, _) = match fabro_sandbox::daytona::detect_repo_info(repo_path) { - Ok(info) => info, - Err(e) => { - tracing::warn!(error = %e, label, "Cannot detect origin for push"); - return false; - } - }; - - let https_url = fabro_github::ssh_url_to_https(&origin_url); - let push_url = match github_app { - Some(creds) => match fabro_github::resolve_authenticated_url(creds, &https_url).await { - Ok(url) => url, - Err(e) => { - tracing::warn!(error = %e, label, "Failed to get token for push"); - return false; - } - }, - None => { - tracing::warn!(label, "No GitHub App credentials for push"); - return false; - } - }; - - let rp = repo_path.to_path_buf(); - let refspec_owned = refspec.to_string(); - let result = crate::git::blocking_push_with_timeout(60, move || { - crate::git::push_ref(&rp, &push_url, &refspec_owned) - }) - .await; - match result { - Ok(()) => { - tracing::info!(label, "Pushed to origin"); - true - } - Err(e) => { - tracing::warn!(error = %e, label, "Failed to push"); - false - } - } -} - -/// Run a git diff via the sandbox. -pub(crate) async fn git_diff( - sandbox: &dyn Sandbox, - base: &str, -) -> std::result::Result { - let cmd = format!("{GIT_REMOTE} diff {base} HEAD"); - match sandbox.exec_command(&cmd, 30_000, None, None, None).await { - Ok(r) if r.exit_code == 0 => Ok(r.stdout), - Ok(r) => Err(format!("exit {}: {}", r.exit_code, r.stderr.trim())), - Err(e) => Err(e.to_string()), - } -} - -// --- Sandbox git helpers --- - -/// Create a branch at a specific SHA via the sandbox. -pub async fn git_create_branch_at(sandbox: &dyn Sandbox, name: &str, sha: &str) -> bool { - let cmd = format!("{GIT_REMOTE} branch --force {name} {sha}"); - matches!( - sandbox.exec_command(&cmd, 30_000, None, None, None).await, - Ok(r) if r.exit_code == 0 - ) -} - -/// Add a git worktree via the sandbox. -pub async fn git_add_worktree(sandbox: &dyn Sandbox, path: &str, branch: &str) -> bool { - let cmd = format!("{GIT_REMOTE} worktree add {path} {branch}"); - matches!( - sandbox.exec_command(&cmd, 30_000, None, None, None).await, - Ok(r) if r.exit_code == 0 - ) -} - -/// Remove a git worktree via the sandbox. -pub async fn git_remove_worktree(sandbox: &dyn Sandbox, path: &str) -> bool { - let cmd = format!("{GIT_REMOTE} worktree remove --force {path}"); - matches!( - sandbox.exec_command(&cmd, 30_000, None, None, None).await, - Ok(r) if r.exit_code == 0 - ) -} - -/// Fast-forward merge to a given SHA via the sandbox. -pub async fn git_merge_ff_only(sandbox: &dyn Sandbox, sha: &str) -> bool { - let cmd = format!("{GIT_REMOTE} merge --ff-only {sha}"); - matches!( - sandbox.exec_command(&cmd, 30_000, None, None, None).await, - Ok(r) if r.exit_code == 0 - ) -} - -/// Remove any stale worktree at `path` (best-effort), then add a fresh one. -pub async fn git_replace_worktree(sandbox: &dyn Sandbox, path: &str, branch: &str) -> bool { - let _ = git_remove_worktree(sandbox, path).await; - git_add_worktree(sandbox, path, branch).await -} - -/// Configuration for a workflow run. -#[derive(Clone)] -pub struct GitCheckpointSettings { - pub base_sha: Option, - pub run_branch: Option, - pub meta_branch: Option, -} - -/// Configuration for a workflow run. -#[derive(Clone)] -pub struct RunSettings { - pub config: FabroConfig, - pub run_dir: PathBuf, - pub cancel_token: Option>, - pub dry_run: bool, - /// Unique identifier for this workflow run. - pub run_id: String, - /// User-defined key-value labels for this run. - pub labels: HashMap, - /// Git author identity for checkpoint commits. - pub git_author: crate::git::GitAuthor, - /// Workflow directory slug (e.g. "smoke" from `fabro/workflows/smoke/`). - pub workflow_slug: Option, - /// GitHub App credentials for pushing metadata branches to origin. - pub github_app: Option, - /// Host repo path for MetadataStore (shadow commits) and host-side pushes. - pub host_repo_path: Option, - /// Name of the branch the run was started from (for PR base). - pub base_branch: Option, - /// Git checkpoint settings; `None` means checkpointing disabled. - pub git: Option, -} - -impl RunSettings { - pub fn checkpoint_exclude_globs(&self) -> &[String] { - &self.config.checkpoint.exclude_globs - } - - /// PR config (already normalized — disabled entries stripped at construction). - pub fn pull_request(&self) -> Option<&PullRequestConfig> { - self.config.pull_request.as_ref() - } - - pub fn asset_globs(&self) -> &[String] { - self.config - .assets - .as_ref() - .map(|a| a.include.as_slice()) - .unwrap_or(&[]) - } -} - -/// Configuration for sandbox lifecycle management within the engine. -pub struct LifecycleConfig { - /// Setup commands to run inside the sandbox after initialization. - pub setup_commands: Vec, - /// Timeout in milliseconds for each setup command. - pub setup_command_timeout_ms: u64, - /// Devcontainer lifecycle phases and their commands. - pub devcontainer_phases: Vec<(String, Vec)>, -} +#[cfg(test)] +use crate::graph_ops::{best_by_weight_then_lexical, normalize_label, weighted_random}; +pub(crate) use crate::graph_ops::{ + build_retry_policy, check_goal_gates, classify_outcome, get_retry_target, is_terminal, + node_script, set_hook_node, +}; +pub use crate::graph_ops::{ + resolve_fidelity, resolve_thread_id, select_edge, EdgeSelection, RetryPolicy, +}; +pub use crate::run_dir::{node_dir, visit_from_context}; +pub(crate) use crate::run_dir::{write_node_status, write_start_record}; +pub use crate::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings}; +pub(crate) use crate::sandbox_git::git_diff; +pub use crate::sandbox_git::{ + git_add_worktree, git_checkpoint, git_create_branch_at, git_merge_ff_only, git_push_host, + git_remove_worktree, git_replace_worktree, GitState, GIT_REMOTE, +}; /// The workflow run execution engine. pub struct WorkflowRunEngine { diff --git a/lib/crates/fabro-workflows/src/graph_ops.rs b/lib/crates/fabro-workflows/src/graph_ops.rs new file mode 100644 index 000000000..16c8b81b5 --- /dev/null +++ b/lib/crates/fabro-workflows/src/graph_ops.rs @@ -0,0 +1,424 @@ +use std::collections::HashMap; +use std::time::Duration; + +use fabro_graphviz::graph::{Edge, Graph, Node}; +use fabro_hooks::HookContext; +use fabro_util::backoff::BackoffPolicy; +use rand::Rng; + +use crate::condition::evaluate_condition; +use crate::context::{self, Context}; +use crate::error::FailureCategory; +use crate::outcome::{Outcome, OutcomeExt, StageStatus}; + +/// Populate node-related fields on a `HookContext` from a graph `Node`. +pub(crate) fn set_hook_node(ctx: &mut HookContext, node: &Node) { + ctx.node_id = Some(node.id.clone()); + ctx.node_label = Some(node.label().to_string()); + ctx.handler_type = node.handler_type().map(String::from); +} + +/// Classify the failure mode of a completed outcome. +/// +/// Returns `None` for `Success`, `PartialSuccess`, and `Skipped` outcomes. +/// For failures, checks (in priority order): +/// 1. Handler hint in `context_updates["failure_class"]` +/// 2. String heuristics on `failure_reason` +/// 3. Default to `Deterministic` +#[must_use] +pub(crate) fn classify_outcome(outcome: &Outcome) -> Option { + match outcome.status { + StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped => None, + StageStatus::Fail | StageStatus::Retry => outcome + .failure_category() + .or(Some(FailureCategory::Deterministic)), + } +} + +/// Retry policy for node execution. +#[derive(Clone, Debug)] +pub struct RetryPolicy { + pub max_attempts: u32, + pub backoff: BackoffPolicy, +} + +impl RetryPolicy { + const DEFAULT_BACKOFF: BackoffPolicy = BackoffPolicy { + initial_delay: Duration::from_millis(5_000), + factor: 2.0, + max_delay: Duration::from_millis(60_000), + jitter: true, + }; + + /// No retries -- fail immediately. + #[must_use] + pub fn none() -> Self { + Self { + max_attempts: 1, + backoff: Self::DEFAULT_BACKOFF, + } + } + + /// Standard retry policy: 5 attempts, 5s initial, 2x factor. + #[must_use] + pub fn standard() -> Self { + Self { + max_attempts: 5, + backoff: Self::DEFAULT_BACKOFF, + } + } + + /// Aggressive retry: 5 attempts, 500ms initial, 2x factor. + #[must_use] + pub fn aggressive() -> Self { + Self { + max_attempts: 5, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(500), + ..Self::DEFAULT_BACKOFF + }, + } + } + + /// Linear retry: 3 attempts, 500ms fixed delay. + #[must_use] + pub fn linear() -> Self { + Self { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(500), + factor: 1.0, + ..Self::DEFAULT_BACKOFF + }, + } + } + + /// Patient retry: 3 attempts, 2000ms initial, 3x factor. + #[must_use] + pub fn patient() -> Self { + Self { + max_attempts: 3, + backoff: BackoffPolicy { + initial_delay: Duration::from_millis(2000), + factor: 3.0, + ..Self::DEFAULT_BACKOFF + }, + } + } +} + +/// Build a retry policy from node and graph attributes. +/// If the node has a `retry_policy` attribute naming a preset, use that. +/// Otherwise, fall back to `max_retries` / graph default. +pub(crate) fn build_retry_policy(node: &Node, graph: &Graph) -> RetryPolicy { + if let Some(preset) = node.retry_policy() { + match preset { + "none" => return RetryPolicy::none(), + "standard" => return RetryPolicy::standard(), + "aggressive" => return RetryPolicy::aggressive(), + "linear" => return RetryPolicy::linear(), + "patient" => return RetryPolicy::patient(), + _ => {} + } + } + let max_retries = node + .max_retries() + .unwrap_or_else(|| graph.default_max_retries()); + let max_attempts = u32::try_from(max_retries + 1).unwrap_or(1).max(1); + RetryPolicy { + max_attempts, + backoff: RetryPolicy::DEFAULT_BACKOFF, + } +} + +/// Resolve the context fidelity for a node, following the precedence: +/// 1. Incoming edge `fidelity` attribute +/// 2. Target node `fidelity` attribute +/// 3. Graph `default_fidelity` attribute +/// 4. Default: Compact +#[must_use] +pub fn resolve_fidelity( + incoming_edge: Option<&Edge>, + node: &Node, + graph: &Graph, +) -> context::keys::Fidelity { + let (resolved, source) = if let Some(f) = incoming_edge + .and_then(|e| e.fidelity()) + .and_then(|s| s.parse().ok()) + { + (f, "edge") + } else if let Some(f) = node.fidelity().and_then(|s| s.parse().ok()) { + (f, "node") + } else if let Some(f) = graph.default_fidelity().and_then(|s| s.parse().ok()) { + (f, "graph") + } else { + (context::keys::Fidelity::default(), "default") + }; + + tracing::debug!( + node = %node.id, + fidelity = %resolved, + source = source, + "Fidelity resolved" + ); + + resolved +} + +/// Resolve the thread ID for a node, following the precedence: +/// 1. Incoming edge `thread_id` attribute +/// 2. Target node `thread_id` attribute +/// 3. Graph-level default thread +/// 4. Derived class from enclosing subgraph (first class from the node's classes list) +/// 5. Fallback to previous node ID +#[must_use] +pub fn resolve_thread_id( + incoming_edge: Option<&Edge>, + node: &Node, + graph: &Graph, + previous_node_id: Option<&str>, +) -> Option { + if let Some(edge) = incoming_edge { + if let Some(tid) = edge.thread_id() { + return Some(tid.to_string()); + } + } + if let Some(tid) = node.thread_id() { + return Some(tid.to_string()); + } + if let Some(tid) = graph.default_thread() { + return Some(tid.to_string()); + } + if let Some(first_class) = node.classes.first() { + return Some(first_class.clone()); + } + previous_node_id.map(String::from) +} + +/// Normalize a label for comparison: lowercase, trim, strip accelerator prefixes. +/// Patterns: "[Y] ", "Y) ", "Y - " +pub(crate) fn normalize_label(label: &str) -> String { + let s = label.trim().to_lowercase(); + if s.starts_with('[') { + if let Some(rest) = s + .strip_prefix('[') + .and_then(|s| s.find(']').map(|i| s[i + 1..].trim_start().to_string())) + { + return rest; + } + } + if s.len() >= 2 { + let bytes = s.as_bytes(); + if bytes.get(1) == Some(&b')') { + return s[2..].trim_start().to_string(); + } + } + if s.len() >= 3 { + if let Some(rest) = s.get(1..).and_then(|r| r.strip_prefix(" - ")) { + return rest.to_string(); + } + } + s +} + +/// Pick the best edge by highest weight, then lexical target node ID tiebreak. +pub(crate) fn best_by_weight_then_lexical<'a>(edges: &[&'a Edge]) -> Option<&'a Edge> { + if edges.is_empty() { + return None; + } + let mut best = edges[0]; + for &edge in &edges[1..] { + if edge.weight() > best.weight() || (edge.weight() == best.weight() && edge.to < best.to) { + best = edge; + } + } + Some(best) +} + +/// Pick a random edge using weighted-random selection. +/// Edges with `weight <= 0` are treated as weight 1 for probability calculation. +pub(crate) fn weighted_random<'a>(edges: &[&'a Edge]) -> Option<&'a Edge> { + if edges.is_empty() { + return None; + } + if edges.len() == 1 { + return Some(edges[0]); + } + let weights: Vec = edges + .iter() + .map(|e| { + let w = e.weight(); + if w <= 0 { + 1.0 + } else { + w as f64 + } + }) + .collect(); + let total: f64 = weights.iter().sum(); + let mut rng = rand::thread_rng(); + let mut roll: f64 = rng.gen_range(0.0..total); + for (i, &w) in weights.iter().enumerate() { + roll -= w; + if roll < 0.0 { + return Some(edges[i]); + } + } + Some(edges[edges.len() - 1]) +} + +/// Dispatch to the appropriate edge-picking strategy. +fn pick_edge<'a>(edges: &[&'a Edge], selection: &str) -> Option<&'a Edge> { + match selection { + "random" => weighted_random(edges), + _ => best_by_weight_then_lexical(edges), + } +} + +/// Result of edge selection: the chosen edge and the reason it was selected. +pub struct EdgeSelection<'a> { + pub edge: &'a Edge, + pub reason: &'static str, +} + +fn blocks_unconditional_failure_fallthrough(node: &Node, outcome: &Outcome) -> bool { + node.handler_type() == Some("human") + && outcome.status == StageStatus::Fail + && outcome.preferred_label.is_none() + && outcome.suggested_next_ids.is_empty() +} + +/// Select the next edge from a node's outgoing edges (spec Section 3.3). +#[must_use] +pub fn select_edge<'a>( + node: &Node, + outcome: &Outcome, + context: &Context, + graph: &'a Graph, + selection: &str, +) -> Option> { + let node_id = &node.id; + let edges = graph.outgoing_edges(node_id); + if edges.is_empty() { + return None; + } + + let condition_matched: Vec<&Edge> = edges + .iter() + .filter(|e| { + e.condition() + .is_some_and(|c| !c.is_empty() && evaluate_condition(c, outcome, context)) + }) + .copied() + .collect(); + if !condition_matched.is_empty() { + return pick_edge(&condition_matched, selection).map(|edge| EdgeSelection { + edge, + reason: "condition", + }); + } + + if let Some(pref) = &outcome.preferred_label { + let normalized_pref = normalize_label(pref); + for edge in &edges { + if edge.condition().is_none_or(str::is_empty) { + if let Some(label) = edge.label() { + if normalize_label(label) == normalized_pref { + return Some(EdgeSelection { + edge, + reason: "preferred_label", + }); + } + } + } + } + } + + for suggested_id in &outcome.suggested_next_ids { + for edge in &edges { + if edge.condition().is_none_or(str::is_empty) && edge.to == *suggested_id { + return Some(EdgeSelection { + edge, + reason: "suggested_next", + }); + } + } + } + + if blocks_unconditional_failure_fallthrough(node, outcome) { + return None; + } + + let unconditional: Vec<&Edge> = edges + .iter() + .filter(|e| e.condition().is_none_or(str::is_empty)) + .copied() + .collect(); + if !unconditional.is_empty() { + return pick_edge(&unconditional, selection).map(|edge| EdgeSelection { + edge, + reason: "unconditional", + }); + } + + None +} + +/// Check if all goal gates have been satisfied. +/// Returns Ok(()) if all gates passed, or Err with the failed node ID. +pub(crate) fn check_goal_gates( + graph: &Graph, + node_outcomes: &HashMap, +) -> std::result::Result<(), String> { + for (node_id, outcome) in node_outcomes { + if let Some(node) = graph.nodes.get(node_id) { + if node.goal_gate() + && outcome.status != StageStatus::Success + && outcome.status != StageStatus::PartialSuccess + { + return Err(node_id.clone()); + } + } + } + Ok(()) +} + +/// Resolve the retry target for a failed goal gate node. +pub(crate) fn get_retry_target(failed_node_id: &str, graph: &Graph) -> Option { + if let Some(node) = graph.nodes.get(failed_node_id) { + if let Some(target) = node.retry_target() { + if graph.nodes.contains_key(target) { + return Some(target.to_string()); + } + } + if let Some(target) = node.fallback_retry_target() { + if graph.nodes.contains_key(target) { + return Some(target.to_string()); + } + } + } + if let Some(target) = graph.retry_target() { + if graph.nodes.contains_key(target) { + return Some(target.to_string()); + } + } + if let Some(target) = graph.fallback_retry_target() { + if graph.nodes.contains_key(target) { + return Some(target.to_string()); + } + } + None +} + +/// Check whether a node is a terminal (exit) node. +pub(crate) fn is_terminal(node: &Node) -> bool { + node.shape() == "Msquare" || node.handler_type() == Some("exit") +} + +pub(crate) fn node_script(node: &Node) -> Option { + node.attrs + .get("script") + .or_else(|| node.attrs.get("tool_command")) + .and_then(|v| v.as_str()) + .map(String::from) +} diff --git a/lib/crates/fabro-workflows/src/lib.rs b/lib/crates/fabro-workflows/src/lib.rs index ee18d299f..3df4eb802 100644 --- a/lib/crates/fabro-workflows/src/lib.rs +++ b/lib/crates/fabro-workflows/src/lib.rs @@ -103,17 +103,21 @@ pub mod engine; pub mod error; pub mod event; pub mod git; +pub mod graph_ops; pub mod graph_render; pub mod handler; pub mod outcome; pub mod pipeline; pub mod preamble; pub mod pull_request; +pub mod run_dir; pub mod run_fork; pub mod run_lookup; pub mod run_record; pub mod run_rewind; +pub mod run_settings; pub mod run_status; +pub mod sandbox_git; pub mod sandbox_provider; pub mod sandbox_reconnect; pub mod sandbox_record; diff --git a/lib/crates/fabro-workflows/src/pipeline/execute.rs b/lib/crates/fabro-workflows/src/pipeline/execute.rs index e65b00a85..0499d869e 100644 --- a/lib/crates/fabro-workflows/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflows/src/pipeline/execute.rs @@ -1,7 +1,29 @@ +use std::sync::Arc; use std::time::Instant; +use fabro_core::executor::ExecutorBuilder; +use fabro_core::state::RunState; +use tokio_util::sync::CancellationToken; + +use crate::context::{self, Context}; +use crate::core_adapter::{WorkflowGraph, WorkflowLifecycle, WorkflowNodeHandler}; +use crate::error::FabroError; +use crate::handler::EngineServices; +use crate::outcome::{Outcome, StageStatus}; +use crate::sandbox_git::GitState; + use super::types::{Executed, Initialized}; +fn seed_context_from_checkpoint(checkpoint: Option<&crate::checkpoint::Checkpoint>) -> Context { + let context = Context::new(); + if let Some(cp) = checkpoint { + for (k, v) in &cp.context_values { + context.set(k.clone(), v.clone()); + } + } + context +} + /// EXECUTE phase: run the workflow graph. /// /// Infallible at the function level — engine errors are captured in `outcome`. @@ -9,18 +31,257 @@ pub async fn execute(init: Initialized) -> Executed { let Initialized { graph, source: _, - engine, settings, checkpoint, + seed_context, emitter, sandbox, + registry, + hook_runner, + env, + dry_run, } = init; let start = Instant::now(); + let graph_arc = Arc::new(graph.clone()); + let wf_graph = WorkflowGraph(Arc::clone(&graph_arc)); - let outcome = engine - .execute_graph(&graph, &settings, checkpoint.as_ref()) - .await; + let git_state = settings.git.as_ref().and_then(|git| { + let base_sha = git.base_sha.clone()?; + Some(Arc::new(GitState { + run_id: settings.run_id.clone(), + base_sha, + run_branch: git.run_branch.clone(), + meta_branch: git.meta_branch.clone(), + checkpoint_exclude_globs: settings.checkpoint_exclude_globs().to_vec(), + git_author: settings.git_author.clone(), + })) + }); + + let shared_services = Arc::new(EngineServices { + registry, + emitter: Arc::clone(&emitter), + sandbox: Arc::clone(&sandbox), + git_state: std::sync::RwLock::new(git_state), + hook_runner: hook_runner.clone(), + env, + dry_run, + }); + + let handler = Arc::new(WorkflowNodeHandler { + services: shared_services, + run_dir: settings.run_dir.clone(), + graph: Arc::clone(&graph_arc), + }); + + let settings_arc = Arc::new(settings.clone()); + let lifecycle = WorkflowLifecycle::new( + Arc::clone(&emitter), + hook_runner.clone(), + Arc::clone(&sandbox), + graph_arc, + settings.run_dir.clone(), + settings_arc, + checkpoint.is_some(), + ); + + if let Some(ref cp) = checkpoint { + lifecycle.restore_circuit_breaker( + cp.loop_failure_signatures.clone(), + cp.restart_failure_signatures.clone(), + ); + if cp.context_values.get(context::keys::INTERNAL_FIDELITY) + == Some(&serde_json::json!(context::keys::Fidelity::Full.to_string())) + { + lifecycle.set_degrade_fidelity_on_resume(true); + } + } + + let state = if let Some(ref cp) = checkpoint { + match RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) { + Ok(mut s) => { + for (k, v) in &cp.context_values { + s.context.set(k.clone(), v.clone()); + } + s.completed_nodes = cp.completed_nodes.clone(); + s.node_retries = cp.node_retries.clone(); + 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(); + } + for (k, v) in &cp.node_outcomes { + s.node_outcomes.insert(k.clone(), v.clone()); + } + s.stage_index = cp.completed_nodes.len(); + 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 + } + Err(err) => { + return Executed { + graph, + outcome: Err(err), + settings, + hook_runner, + emitter, + sandbox, + duration_ms: crate::millis_u64(start.elapsed()), + final_context: seed_context_from_checkpoint(checkpoint.as_ref()), + }; + } + } + } else if let Some(seed) = seed_context { + match RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) { + Ok(s) => { + for (k, v) in seed.snapshot() { + s.context.set(k, v); + } + s + } + Err(err) => { + return Executed { + graph, + outcome: Err(err), + settings, + hook_runner, + emitter, + sandbox, + duration_ms: crate::millis_u64(start.elapsed()), + final_context: seed, + }; + } + } + } else { + match RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) { + Ok(s) => s, + Err(err) => { + return Executed { + graph, + outcome: Err(err), + settings, + hook_runner, + emitter, + sandbox, + duration_ms: crate::millis_u64(start.elapsed()), + final_context: Context::new(), + }; + } + } + }; + + let initial_context = state.context.clone(); + + let graph_max = graph.max_node_visits(); + let max_node_visits = if graph_max > 0 { + Some(graph_max as usize) + } else if settings.dry_run { + Some(10) + } else { + None + }; + + let stall_timeout_opt = graph.stall_timeout(); + let stall_token = stall_timeout_opt.map(|_| CancellationToken::new()); + let stall_shutdown = + if let (Some(stall_timeout), Some(ref token)) = (stall_timeout_opt, &stall_token) { + let shutdown = CancellationToken::new(); + let emitter = Arc::clone(&emitter); + let token_clone = token.clone(); + let shutdown_clone = shutdown.clone(); + emitter.touch(); + tokio::spawn(async move { + loop { + tokio::select! { + _ = tokio::time::sleep(stall_timeout) => { + if shutdown_clone.is_cancelled() { + return; + } + 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; + let idle_ms = now.saturating_sub(last); + if idle_ms >= stall_timeout.as_millis() as i64 { + token_clone.cancel(); + return; + } + } + _ = shutdown_clone.cancelled() => { + return; + } + } + } + }); + Some(shutdown) + } else { + None + }; + + let mut builder = + ExecutorBuilder::new(handler as Arc>) + .lifecycle(Box::new(lifecycle)); + + if let Some(ref cancel) = settings.cancel_token { + builder = builder.cancel_token(cancel.clone()); + } + if let Some(token) = stall_token.clone() { + builder = builder.stall_token(token); + } + if let Some(limit) = max_node_visits { + builder = builder.max_node_visits(limit); + } + + let executor = builder.build(); + let result = executor.run(&wf_graph, state).await; + + if let Some(shutdown) = stall_shutdown { + shutdown.cancel(); + } + + let (outcome, final_context) = match result { + Ok((core_outcome, final_state)) => { + let ctx = final_state.context.clone(); + 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(); + let idle_secs = stall_timeout.as_secs(); + emitter.emit(&crate::event::WorkflowRunEvent::StallWatchdogTimeout { + node: node_id.clone(), + idle_seconds: idle_secs, + }); + ( + Err(FabroError::engine(format!( + "stall watchdog: node \"{node_id}\" had no activity for {idle_secs}s" + ))), + initial_context, + ) + } + Err(fabro_core::CoreError::Cancelled) => (Err(FabroError::Cancelled), initial_context), + Err(fabro_core::CoreError::Blocked { message }) => { + (Err(FabroError::engine(message)), initial_context) + } + Err(e) => (Err(FabroError::engine(e.to_string())), initial_context), + }; let duration_ms = crate::millis_u64(start.elapsed()); @@ -28,9 +289,109 @@ pub async fn execute(init: Initialized) -> Executed { graph, outcome, settings, - engine, + hook_runner, emitter, sandbox, duration_ms, + final_context, + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use fabro_config::config::FabroConfig; + use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; + use fabro_interview::AutoApproveInterviewer; + + use super::*; + use crate::handler::default_registry; + use crate::pipeline::initialize; + use crate::pipeline::types::{InitOptions, Validated}; + use crate::run_settings::{LifecycleConfig, RunSettings}; + + fn simple_graph() -> (Graph, String) { + let source = + "digraph test { start [shape=Mdiamond]; exit [shape=Msquare]; start -> exit; }" + .to_string(); + let mut graph = Graph::new("test"); + let mut start = Node::new("start"); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); + let mut exit = Node::new("exit"); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); + graph.nodes.insert("start".to_string(), start); + graph.nodes.insert("exit".to_string(), exit); + graph.edges.push(Edge::new("start", "exit")); + (graph, source) + } + + fn test_settings(run_dir: &std::path::Path) -> RunSettings { + RunSettings { + config: FabroConfig::default(), + run_dir: run_dir.to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "run-test".to_string(), + labels: HashMap::new(), + git_author: crate::git::GitAuthor::default(), + workflow_slug: None, + github_app: None, + host_repo_path: None, + base_branch: None, + git: None, + } + } + + #[tokio::test] + async fn execute_runs_start_to_exit_and_returns_final_context() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + let (graph, source) = simple_graph(); + let initialized = initialize( + Validated::new(graph, source, vec![]), + InitOptions { + run_id: "run-test".to_string(), + run_dir: run_dir.clone(), + dry_run: false, + emitter: Arc::new(crate::event::EventEmitter::new()), + sandbox: Arc::new(fabro_agent::LocalSandbox::new( + std::env::current_dir().unwrap(), + )), + registry: Arc::new(default_registry(Arc::new(AutoApproveInterviewer), || None)), + lifecycle: LifecycleConfig { + setup_commands: vec![], + setup_command_timeout_ms: 1_000, + devcontainer_phases: vec![], + }, + run_settings: test_settings(&run_dir), + hooks: fabro_hooks::HookConfig { hooks: vec![] }, + sandbox_env: HashMap::new(), + checkpoint: None, + seed_context: None, + }, + ) + .await + .unwrap(); + + let executed = execute(initialized).await; + + assert_eq!( + executed.outcome.as_ref().unwrap().status, + crate::outcome::StageStatus::Success + ); + assert_eq!( + executed + .final_context + .get(crate::context::keys::INTERNAL_RUN_ID), + Some(serde_json::json!("run-test")) + ); } } diff --git a/lib/crates/fabro-workflows/src/pipeline/finalize.rs b/lib/crates/fabro-workflows/src/pipeline/finalize.rs index 7d5652f2e..953130488 100644 --- a/lib/crates/fabro-workflows/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflows/src/pipeline/finalize.rs @@ -1,7 +1,213 @@ +use std::path::Path; +use std::sync::Arc; + +use crate::checkpoint::Checkpoint; +use crate::conclusion::Conclusion; use crate::error::FabroError; +use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; +use crate::outcome::{Outcome, OutcomeExt, StageStatus}; +use crate::run_settings::RunSettings; +use crate::run_status::{RunStatus, StatusReason}; +use fabro_hooks::{HookContext, HookEvent, HookRunner}; use super::types::{FinalizeOptions, Finalized, Retroed}; +fn emit_run_notice( + emitter: &EventEmitter, + level: RunNoticeLevel, + code: impl Into, + message: impl Into, +) { + emitter.emit(&WorkflowRunEvent::RunNotice { + level, + code: code.into(), + message: message.into(), + }); +} + +pub fn classify_engine_result( + engine_result: &Result, +) -> (StageStatus, Option, RunStatus, Option) { + match engine_result { + Ok(outcome) => { + let status = outcome.status.clone(); + let failure_reason = outcome.failure_reason().map(String::from); + let (run_status, status_reason) = match status { + StageStatus::Success | StageStatus::Skipped => { + (RunStatus::Succeeded, Some(StatusReason::Completed)) + } + StageStatus::PartialSuccess => { + (RunStatus::Succeeded, Some(StatusReason::PartialSuccess)) + } + StageStatus::Fail | StageStatus::Retry => { + (RunStatus::Failed, Some(StatusReason::WorkflowError)) + } + }; + (status, failure_reason, run_status, status_reason) + } + Err(FabroError::Cancelled) => ( + StageStatus::Fail, + Some("Cancelled".to_string()), + RunStatus::Failed, + Some(StatusReason::Cancelled), + ), + Err(err) => ( + StageStatus::Fail, + Some(err.to_string()), + RunStatus::Failed, + Some(StatusReason::WorkflowError), + ), + } +} + +pub fn build_conclusion( + run_dir: &Path, + status: StageStatus, + failure_reason: Option, + run_duration_ms: u64, + final_git_commit_sha: Option, +) -> Conclusion { + let checkpoint = Checkpoint::load(&run_dir.join("checkpoint.json")).ok(); + let stage_durations = fabro_retro::retro::extract_stage_durations(run_dir); + + let mut total_input_tokens: i64 = 0; + let mut total_output_tokens: i64 = 0; + let mut total_cache_read_tokens: i64 = 0; + let mut total_cache_write_tokens: i64 = 0; + let mut total_reasoning_tokens: i64 = 0; + let mut has_pricing = false; + + let (stages, total_cost, total_retries) = if let Some(ref cp) = checkpoint { + let mut stages = Vec::new(); + let mut cost_sum: Option = None; + let mut retries_sum: u32 = 0; + + for node_id in &cp.completed_nodes { + let outcome = cp.node_outcomes.get(node_id); + let retries = cp + .node_retries + .get(node_id) + .copied() + .unwrap_or(1) + .saturating_sub(1); + retries_sum += retries; + + let cost = outcome.and_then(|o| o.usage.as_ref()).and_then(|u| u.cost); + if let Some(c) = cost { + *cost_sum.get_or_insert(0.0) += c; + has_pricing = true; + } + + if let Some(usage) = outcome.and_then(|o| o.usage.as_ref()) { + total_input_tokens += usage.input_tokens; + total_output_tokens += usage.output_tokens; + total_cache_read_tokens += usage.cache_read_tokens.unwrap_or(0); + total_cache_write_tokens += usage.cache_write_tokens.unwrap_or(0); + total_reasoning_tokens += usage.reasoning_tokens.unwrap_or(0); + } + + stages.push(crate::conclusion::StageSummary { + stage_id: node_id.clone(), + stage_label: node_id.clone(), + duration_ms: stage_durations.get(node_id).copied().unwrap_or(0), + cost, + retries, + }); + } + (stages, cost_sum, retries_sum) + } else { + (vec![], None, 0) + }; + + Conclusion { + timestamp: chrono::Utc::now(), + status, + duration_ms: run_duration_ms, + failure_reason, + final_git_commit_sha, + stages, + total_cost, + total_retries, + total_input_tokens, + total_output_tokens, + total_cache_read_tokens, + total_cache_write_tokens, + total_reasoning_tokens, + has_pricing, + } +} + +pub fn persist_terminal_outcome( + run_dir: &Path, + conclusion: &Conclusion, + run_status: RunStatus, + status_reason: Option, +) { + let _ = conclusion.save(&run_dir.join("conclusion.json")); + crate::run_status::write_run_status(run_dir, run_status, status_reason); +} + +/// Write a finalize commit to the shadow branch with retro.json and final node files. +/// +/// This captures the last diff.patch (written after the final checkpoint) and retro.json. +/// Best-effort: errors are logged as warnings. +pub async fn write_finalize_commit(config: &RunSettings, run_dir: &Path) { + let (Some(meta_branch), Some(repo_path)) = ( + config.git.as_ref().and_then(|g| g.meta_branch.as_ref()), + config.host_repo_path.as_ref(), + ) else { + return; + }; + + let store = crate::git::MetadataStore::new(repo_path, &config.git_author); + let mut entries = crate::git::scan_node_files(run_dir); + if let Ok(retro_bytes) = std::fs::read(run_dir.join("retro.json")) { + entries.push(("retro.json".to_string(), retro_bytes)); + } + let refs: Vec<(&str, &[u8])> = entries + .iter() + .map(|(k, v)| (k.as_str(), v.as_slice())) + .collect(); + if let Err(e) = store.write_files(&config.run_id, &refs, "finalize run") { + tracing::warn!(error = %e, "Failed to write finalize commit to metadata branch"); + return; + } + + let refspec = format!("refs/heads/{meta_branch}"); + crate::sandbox_git::git_push_host(repo_path, &refspec, &config.github_app, "finalize metadata") + .await; +} + +async fn run_hooks( + hook_runner: Option<&HookRunner>, + hook_context: &HookContext, + sandbox: Arc, +) { + let Some(runner) = hook_runner else { + return; + }; + let _ = runner.run(hook_context, sandbox, None).await; +} + +async fn cleanup_sandbox( + hook_runner: Option>, + sandbox: Arc, + run_id: &str, + workflow_name: &str, + preserve: bool, +) -> std::result::Result<(), String> { + let hook_ctx = HookContext::new( + HookEvent::SandboxCleanup, + run_id.to_string(), + workflow_name.to_string(), + ); + run_hooks(hook_runner.as_deref(), &hook_ctx, Arc::clone(&sandbox)).await; + if !preserve { + sandbox.cleanup().await?; + } + Ok(()) +} + /// FINALIZE phase: classify outcome, build conclusion, persist terminal state. /// /// # Errors @@ -9,45 +215,218 @@ use super::types::{FinalizeOptions, Finalized, Retroed}; /// Returns `FabroError` if persisting terminal state fails. pub async fn finalize( retroed: Retroed, - _options: &FinalizeOptions, + options: &FinalizeOptions, ) -> Result { let Retroed { - graph: _, + graph, outcome, settings, - engine: _, - emitter: _, - sandbox: _, - duration_ms: _, + hook_runner, + emitter, + sandbox, + duration_ms, retro: _, } = retroed; - // TODO: Extract finalize logic from CLI run.rs in Step 5. - // For now, return a minimal Finalized. - let conclusion = crate::conclusion::Conclusion { - timestamp: chrono::Utc::now(), - status: match &outcome { - Ok(o) => o.status.clone(), - Err(_) => crate::outcome::StageStatus::Fail, - }, - duration_ms: 0, - failure_reason: outcome.as_ref().err().map(|e| e.to_string()), - final_git_commit_sha: None, - stages: vec![], - total_cost: None, - total_retries: 0, - total_input_tokens: 0, - total_output_tokens: 0, - total_cache_read_tokens: 0, - total_cache_write_tokens: 0, - total_reasoning_tokens: 0, - has_pricing: false, - }; + let (final_status, failure_reason, run_status, status_reason) = + classify_engine_result(&outcome); + let conclusion = build_conclusion( + &options.run_dir, + final_status, + failure_reason, + duration_ms, + options.last_git_sha.clone(), + ); + + write_finalize_commit(&settings, &options.run_dir).await; + + let mut pr_url = None; + if let Some(pr_cfg) = &options.pr_config { + if let Err(ref e) = outcome { + tracing::debug!(error = %e, "Skipping PR creation: engine returned an error"); + } else if let Ok(ref result) = outcome { + if matches!( + result.status, + StageStatus::Success | StageStatus::PartialSuccess + ) { + let diff = tokio::fs::read_to_string(options.run_dir.join("final.patch")) + .await + .unwrap_or_default(); + if let ( + Some(ref base_branch), + Some(ref run_branch), + Some(ref creds), + Some(ref origin), + ) = ( + &settings.base_branch, + settings.git.as_ref().and_then(|g| g.run_branch.as_ref()), + &options.github_app, + &options.origin_url, + ) { + let auto_merge = if pr_cfg.auto_merge { + Some(crate::pull_request::AutoMergeConfig { + merge_strategy: pr_cfg.merge_strategy, + }) + } else { + None + }; + + match crate::pull_request::maybe_open_pull_request( + creds, + origin, + base_branch, + run_branch, + graph.goal(), + &diff, + &options.model, + pr_cfg.draft, + auto_merge, + &options.run_dir, + ) + .await + { + Ok(Some(record)) => { + emitter.emit(&WorkflowRunEvent::PullRequestCreated { + pr_url: record.html_url.clone(), + pr_number: record.number, + draft: pr_cfg.draft, + }); + pr_url = Some(record.html_url.clone()); + if let Err(e) = record.save(&options.run_dir.join("pull_request.json")) + { + tracing::warn!(error = %e, "Failed to save pull_request.json"); + } + } + Ok(None) => {} + Err(e) => { + emitter.emit(&WorkflowRunEvent::PullRequestFailed { + error: e.to_string(), + }); + emit_run_notice( + &emitter, + RunNoticeLevel::Warn, + "pull_request_failed", + format!("PR creation failed: {e}"), + ); + } + } + } + } + } + } + + if options.preserve_sandbox { + let info = sandbox.sandbox_info(); + if !info.is_empty() { + emit_run_notice( + &emitter, + RunNoticeLevel::Info, + "sandbox_preserved", + format!("sandbox preserved: {info}"), + ); + } else { + emit_run_notice( + &emitter, + RunNoticeLevel::Info, + "sandbox_preserved", + "sandbox preserved", + ); + } + } + if let Err(e) = cleanup_sandbox( + options.hook_runner.clone().or(hook_runner), + sandbox, + &options.run_id, + &options.workflow_name, + options.preserve_sandbox, + ) + .await + { + tracing::warn!(error = %e, "Sandbox cleanup failed"); + emit_run_notice( + &emitter, + RunNoticeLevel::Warn, + "sandbox_cleanup_failed", + format!("sandbox cleanup failed: {e}"), + ); + } + + persist_terminal_outcome(&options.run_dir, &conclusion, run_status, status_reason); Ok(Finalized { run_id: settings.run_id, outcome, conclusion, - pr_url: None, + pr_url, }) } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use fabro_config::config::FabroConfig; + use fabro_graphviz::graph::Graph; + + use super::*; + use crate::pipeline::types::Retroed; + use crate::run_settings::RunSettings; + + fn test_settings(run_dir: &std::path::Path) -> RunSettings { + RunSettings { + config: FabroConfig::default(), + run_dir: run_dir.to_path_buf(), + cancel_token: None, + dry_run: true, + run_id: "run-test".to_string(), + labels: HashMap::new(), + git_author: crate::git::GitAuthor::default(), + workflow_slug: None, + github_app: None, + host_repo_path: None, + base_branch: None, + git: None, + } + } + + #[tokio::test] + async fn finalize_writes_conclusion_json() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); + let retroed = Retroed { + graph: Graph::new("test"), + outcome: Ok(Outcome::success()), + settings: test_settings(&run_dir), + hook_runner: None, + emitter: Arc::new(EventEmitter::new()), + sandbox: Arc::new(fabro_agent::LocalSandbox::new( + std::env::current_dir().unwrap(), + )), + duration_ms: 5, + retro: None, + }; + + let finalized = finalize( + retroed, + &FinalizeOptions { + run_dir: run_dir.clone(), + run_id: "run-test".to_string(), + workflow_name: "test".to_string(), + hook_runner: None, + preserve_sandbox: true, + pr_config: None, + github_app: None, + origin_url: None, + model: "test-model".to_string(), + last_git_sha: None, + }, + ) + .await + .unwrap(); + + assert!(run_dir.join("conclusion.json").exists()); + assert_eq!(finalized.conclusion.status, StageStatus::Success); + } +} diff --git a/lib/crates/fabro-workflows/src/pipeline/initialize.rs b/lib/crates/fabro-workflows/src/pipeline/initialize.rs index 061a46406..4d7ae3d62 100644 --- a/lib/crates/fabro-workflows/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflows/src/pipeline/initialize.rs @@ -1,18 +1,28 @@ +use std::path::Path; use std::sync::Arc; +use std::time::Instant; -use fabro_hooks::HookRunner; +use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner}; -use crate::engine::WorkflowRunEngine; use crate::error::FabroError; +use crate::event::WorkflowRunEvent; +use crate::run_settings::GitCheckpointSettings; use super::types::{InitOptions, Initialized, Validated}; -/// INITIALIZE phase: set up the engine and prepare the sandbox for execution. -/// -/// - Creates run directory, writes `graph.fabro` -/// - Builds `WorkflowRunEngine` from components -/// - Wires hooks, env, dry_run onto engine -/// - Calls `engine.prepare_sandbox()` (sandbox init, git setup, setup commands, devcontainer) +async fn run_hooks( + hook_runner: Option<&HookRunner>, + hook_context: &HookContext, + sandbox: Arc, + work_dir: Option<&Path>, +) -> HookDecision { + let Some(runner) = hook_runner else { + return HookDecision::Proceed; + }; + runner.run(hook_context, sandbox, work_dir).await +} + +/// INITIALIZE phase: prepare the sandbox for execution. /// /// # Errors /// @@ -28,37 +38,257 @@ pub async fn initialize( let graph_path = options.run_dir.join("graph.fabro"); std::fs::write(&graph_path, &source)?; - // Build engine - let mut engine = WorkflowRunEngine::with_interviewer( - options.registry, - Arc::clone(&options.emitter), - options.interviewer, - Arc::clone(&options.sandbox), - ); + let hook_runner = if options.hooks.hooks.is_empty() { + None + } else { + Some(Arc::new(HookRunner::new(options.hooks))) + }; - // Wire hooks - if !options.hooks.hooks.is_empty() { - engine.set_hook_runner(Arc::new(HookRunner::new(options.hooks))); + options + .sandbox + .initialize() + .await + .map_err(|e| FabroError::engine(format!("Failed to initialize sandbox: {e}")))?; + + let hook_ctx = HookContext::new( + HookEvent::SandboxReady, + options.run_settings.run_id.clone(), + graph.name.clone(), + ); + let decision = run_hooks( + hook_runner.as_deref(), + &hook_ctx, + Arc::clone(&options.sandbox), + None, + ) + .await; + if let HookDecision::Block { reason } = decision { + let msg = reason.unwrap_or_else(|| "blocked by SandboxReady hook".into()); + return Err(FabroError::engine(msg)); } - // Wire env and dry_run - engine.set_env(options.sandbox_env); - engine.set_dry_run(options.dry_run); + options.emitter.emit(&WorkflowRunEvent::SandboxInitialized { + working_directory: options.sandbox.working_directory().to_string(), + }); - // Prepare sandbox (initialize, git setup, setup commands, devcontainer) - engine - .prepare_sandbox(&graph, &mut options.run_settings, options.lifecycle) - .await?; + let has_run_branch = options + .run_settings + .git + .as_ref() + .and_then(|g| g.run_branch.as_ref()) + .is_some(); + if !has_run_branch { + match options + .sandbox + .setup_git_for_run(&options.run_settings.run_id) + .await + { + Ok(Some(info)) => { + let base_sha = options + .run_settings + .git + .as_ref() + .and_then(|g| g.base_sha.clone()) + .or(Some(info.base_sha)); + options.run_settings.git = Some(GitCheckpointSettings { + base_sha, + run_branch: Some(info.run_branch.clone()), + meta_branch: Some(crate::git::MetadataStore::branch_name( + &options.run_settings.run_id, + )), + }); + if options.run_settings.base_branch.is_none() { + options.run_settings.base_branch = info.base_branch; + } + } + Ok(None) => {} + Err(e) => { + tracing::warn!( + error = %e, + "Sandbox git setup failed, running without git checkpoints" + ); + } + } + } - // At this point run_settings may have been mutated by prepare_sandbox (base_sha, run_branch, etc.) + if !options.lifecycle.setup_commands.is_empty() { + options.emitter.emit(&WorkflowRunEvent::SetupStarted { + command_count: options.lifecycle.setup_commands.len(), + }); + let setup_start = Instant::now(); + for (index, cmd) in options.lifecycle.setup_commands.iter().enumerate() { + options + .emitter + .emit(&WorkflowRunEvent::SetupCommandStarted { + command: cmd.clone(), + index, + }); + let cmd_start = Instant::now(); + let result = options + .sandbox + .exec_command( + cmd, + options.lifecycle.setup_command_timeout_ms, + None, + None, + None, + ) + .await + .map_err(|e| FabroError::engine(format!("Setup command failed: {e}")))?; + let cmd_duration = crate::millis_u64(cmd_start.elapsed()); + if result.exit_code != 0 { + options.emitter.emit(&WorkflowRunEvent::SetupFailed { + command: cmd.clone(), + index, + exit_code: result.exit_code, + stderr: result.stderr.clone(), + }); + return Err(FabroError::engine(format!( + "Setup command failed (exit code {}): {cmd}\n{}", + result.exit_code, result.stderr, + ))); + } + options + .emitter + .emit(&WorkflowRunEvent::SetupCommandCompleted { + command: cmd.clone(), + index, + exit_code: result.exit_code, + duration_ms: cmd_duration, + }); + } + options.emitter.emit(&WorkflowRunEvent::SetupCompleted { + duration_ms: crate::millis_u64(setup_start.elapsed()), + }); + } + + for (phase, commands) in &options.lifecycle.devcontainer_phases { + crate::devcontainer_bridge::run_devcontainer_lifecycle( + options.sandbox.as_ref(), + &options.emitter, + phase, + commands, + options.lifecycle.setup_command_timeout_ms, + ) + .await + .map_err(|e| FabroError::engine(e.to_string()))?; + } Ok(Initialized { graph, source, - engine, settings: options.run_settings, - checkpoint: None, + checkpoint: options.checkpoint, + seed_context: options.seed_context, emitter: options.emitter, sandbox: options.sandbox, + registry: options.registry, + hook_runner, + env: options.sandbox_env, + dry_run: options.dry_run, }) } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use fabro_config::config::FabroConfig; + use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; + use fabro_interview::AutoApproveInterviewer; + + use super::*; + use crate::handler::default_registry; + use crate::pipeline::types::Validated; + use crate::run_settings::RunSettings; + + fn simple_graph() -> (Graph, String) { + let source = r#"digraph test { + start [shape=Mdiamond]; + exit [shape=Msquare]; + start -> exit; +}"# + .to_string(); + let mut graph = Graph::new("test"); + let mut start = Node::new("start"); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); + let mut exit = Node::new("exit"); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); + graph.nodes.insert("start".to_string(), start); + graph.nodes.insert("exit".to_string(), exit); + graph.edges.push(Edge::new("start", "exit")); + (graph, source) + } + + fn test_settings(run_dir: &std::path::Path) -> RunSettings { + RunSettings { + config: FabroConfig::default(), + run_dir: run_dir.to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "run-test".to_string(), + labels: HashMap::new(), + git_author: crate::git::GitAuthor::default(), + workflow_slug: None, + github_app: None, + host_repo_path: None, + base_branch: None, + git: None, + } + } + + #[tokio::test] + async fn initialize_prepares_sandbox_and_writes_graph() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + let (graph, source) = simple_graph(); + let validated = Validated::new(graph, source.clone(), vec![]); + let emitter = Arc::new(crate::event::EventEmitter::new()); + let sandbox = Arc::new(fabro_agent::LocalSandbox::new( + std::env::current_dir().unwrap(), + )); + let registry = Arc::new(default_registry(Arc::new(AutoApproveInterviewer), || None)); + + let initialized = initialize( + validated, + InitOptions { + run_id: "run-test".to_string(), + run_dir: run_dir.clone(), + dry_run: false, + emitter, + sandbox, + registry, + lifecycle: crate::run_settings::LifecycleConfig { + setup_commands: vec![], + setup_command_timeout_ms: 1_000, + devcontainer_phases: vec![], + }, + run_settings: test_settings(&run_dir), + hooks: fabro_hooks::HookConfig { hooks: vec![] }, + sandbox_env: HashMap::from([("TEST_KEY".to_string(), "value".to_string())]), + checkpoint: None, + seed_context: None, + }, + ) + .await + .unwrap(); + + assert!(run_dir.join("graph.fabro").exists()); + assert_eq!( + std::fs::read_to_string(run_dir.join("graph.fabro")).unwrap(), + source + ); + assert!(initialized.hook_runner.is_none()); + assert_eq!( + initialized.env.get("TEST_KEY").map(String::as_str), + Some("value") + ); + } +} diff --git a/lib/crates/fabro-workflows/src/pipeline/mod.rs b/lib/crates/fabro-workflows/src/pipeline/mod.rs index 76ed9e8cf..ec100aa35 100644 --- a/lib/crates/fabro-workflows/src/pipeline/mod.rs +++ b/lib/crates/fabro-workflows/src/pipeline/mod.rs @@ -8,10 +8,13 @@ pub mod types; mod validate; pub use execute::execute; -pub use finalize::finalize; +pub use finalize::{ + build_conclusion, classify_engine_result, finalize, persist_terminal_outcome, + write_finalize_commit, +}; pub use initialize::initialize; pub use parse::parse; -pub use retro::retro; +pub use retro::{retro, run_retro}; pub use transform::transform; pub use types::*; pub use validate::validate; diff --git a/lib/crates/fabro-workflows/src/pipeline/retro.rs b/lib/crates/fabro-workflows/src/pipeline/retro.rs index ad85cd202..4b1b72981 100644 --- a/lib/crates/fabro-workflows/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflows/src/pipeline/retro.rs @@ -1,30 +1,284 @@ +use std::sync::Arc; + +use fabro_agent::SessionEvent; +use fabro_retro::retro::Retro; + +use crate::checkpoint::Checkpoint; +use crate::event::WorkflowRunEvent; + use super::types::{Executed, RetroOptions, Retroed}; +pub async fn run_retro(options: &RetroOptions) -> Option { + let cp = match Checkpoint::load(&options.run_dir.join("checkpoint.json")) { + Ok(cp) => cp, + Err(e) => { + tracing::warn!(error = %e, "Could not load checkpoint, skipping retro"); + if let Some(ref emitter) = options.emitter { + emitter.emit(&WorkflowRunEvent::RetroFailed { + error: e.to_string(), + duration_ms: 0, + }); + } + return None; + } + }; + + let completed_stages = crate::build_completed_stages(&cp, options.failed); + let stage_durations = fabro_retro::retro::extract_stage_durations(&options.run_dir); + let mut retro = fabro_retro::retro::derive_retro( + &options.run_id, + &options.workflow_name, + &options.goal, + completed_stages, + options.run_duration_ms, + &stage_durations, + ); + + if let Err(e) = retro.save(&options.run_dir) { + tracing::warn!(error = %e, "Failed to save initial retro"); + } + + let retro_start = std::time::Instant::now(); + if let Some(ref emitter) = options.emitter { + emitter.emit(&WorkflowRunEvent::RetroStarted); + } + + let narrative_result = if options.dry_run { + Ok(fabro_retro::retro_agent::dry_run_narrative()) + } else if let Some(client) = options.llm_client.as_ref() { + let emitter_clone = options.emitter.clone(); + let event_callback: Option> = + emitter_clone.map(|emitter| -> Arc { + Arc::new(move |event: SessionEvent| { + emitter.touch(); + if !matches!( + &event.event, + fabro_agent::AgentEvent::SessionStarted + | fabro_agent::AgentEvent::SessionEnded + | fabro_agent::AgentEvent::AssistantTextStart + | fabro_agent::AgentEvent::AssistantOutputReplace { .. } + | fabro_agent::AgentEvent::TextDelta { .. } + | fabro_agent::AgentEvent::ReasoningDelta { .. } + | fabro_agent::AgentEvent::ToolCallOutputDelta { .. } + | fabro_agent::AgentEvent::SkillExpanded { .. } + ) { + emitter.emit(&WorkflowRunEvent::Agent { + stage: "retro".to_string(), + event: event.event.clone(), + }); + } + }) + }); + fabro_retro::retro_agent::run_retro_agent( + &options.sandbox, + &options.run_dir, + client, + options.provider, + &options.model, + event_callback, + ) + .await + } else { + Err(anyhow::anyhow!("No LLM client available")) + }; + + let duration_ms = retro_start.elapsed().as_millis() as u64; + if let Some(ref emitter) = options.emitter { + match &narrative_result { + Ok(_) => emitter.emit(&WorkflowRunEvent::RetroCompleted { duration_ms }), + Err(e) => emitter.emit(&WorkflowRunEvent::RetroFailed { + error: e.to_string(), + duration_ms, + }), + } + } + + match narrative_result { + Ok(narrative) => { + retro.apply_narrative(narrative); + if let Err(e) = retro.save(&options.run_dir) { + tracing::warn!(error = %e, "Failed to save retro with narrative"); + } + } + Err(e) => { + tracing::debug!(error = %e, "Retro agent skipped"); + } + } + + Some(retro) +} + /// RETRO phase: generate a retrospective for the workflow run. /// /// Infallible — errors are logged, not propagated. If disabled, passes through /// with `retro: None`. -pub async fn retro(executed: Executed, _options: &RetroOptions) -> Retroed { +pub async fn retro(executed: Executed, options: &RetroOptions) -> Retroed { let Executed { graph, outcome, settings, - engine, + hook_runner, emitter, sandbox, duration_ms, + final_context: _, } = executed; - // TODO: Extract core retro logic from CLI run.rs in Step 5. - // For now, pass through with no retro. + let retro = if options.enabled { + run_retro(options).await + } else { + None + }; + Retroed { graph, outcome, settings, - engine, + hook_runner, emitter, sandbox, duration_ms, - retro: None, + retro, + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + + use fabro_config::config::FabroConfig; + use fabro_graphviz::graph::Graph; + + use super::*; + use crate::checkpoint::Checkpoint; + use crate::context::Context; + use crate::event::{EventEmitter, WorkflowRunEvent}; + use crate::pipeline::types::Executed; + use crate::run_settings::RunSettings; + + fn write_checkpoint(run_dir: &std::path::Path) { + let context = Context::new(); + context.set("response.work", serde_json::json!("done")); + let mut outcomes = HashMap::new(); + outcomes.insert("work".to_string(), crate::outcome::Outcome::success()); + let checkpoint = Checkpoint::from_context( + &context, + "work", + vec!["work".to_string()], + HashMap::new(), + outcomes, + None, + HashMap::new(), + HashMap::new(), + HashMap::new(), + ); + checkpoint.save(&run_dir.join("checkpoint.json")).unwrap(); + } + + fn test_settings(run_dir: &std::path::Path) -> RunSettings { + RunSettings { + config: FabroConfig::default(), + run_dir: run_dir.to_path_buf(), + cancel_token: None, + dry_run: true, + run_id: "run-test".to_string(), + labels: HashMap::new(), + git_author: crate::git::GitAuthor::default(), + workflow_slug: None, + github_app: None, + host_repo_path: None, + base_branch: None, + git: None, + } + } + + #[tokio::test] + async fn retro_phase_writes_retro_json() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); + write_checkpoint(&run_dir); + + let emitter = Arc::new(EventEmitter::new()); + let sandbox: Arc = Arc::new(fabro_agent::LocalSandbox::new( + std::env::current_dir().unwrap(), + )); + let executed = Executed { + graph: Graph::new("test"), + outcome: Ok(crate::outcome::Outcome::success()), + settings: test_settings(&run_dir), + hook_runner: None, + emitter: Arc::clone(&emitter), + sandbox: Arc::clone(&sandbox), + duration_ms: 1, + final_context: Context::new(), + }; + + let retroed = retro( + executed, + &RetroOptions { + run_id: "run-test".to_string(), + workflow_name: "test".to_string(), + goal: "Ship it".to_string(), + run_dir: run_dir.clone(), + sandbox, + emitter: Some(emitter), + failed: false, + run_duration_ms: 1, + enabled: true, + dry_run: true, + llm_client: None, + provider: fabro_llm::Provider::Anthropic, + model: "test-model".to_string(), + }, + ) + .await; + + assert!(run_dir.join("retro.json").exists()); + assert!(retroed.retro.is_some()); + } + + #[tokio::test] + async fn run_retro_emits_retro_events() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); + write_checkpoint(&run_dir); + + let emitter = Arc::new(EventEmitter::new()); + let seen = Arc::new(Mutex::new(Vec::new())); + emitter.on_event({ + let seen = Arc::clone(&seen); + move |event| seen.lock().unwrap().push(event.clone()) + }); + + let retro = run_retro(&RetroOptions { + run_id: "run-test".to_string(), + workflow_name: "test".to_string(), + goal: "Ship it".to_string(), + run_dir: run_dir.clone(), + sandbox: Arc::new(fabro_agent::LocalSandbox::new( + std::env::current_dir().unwrap(), + )), + emitter: Some(Arc::clone(&emitter)), + failed: false, + run_duration_ms: 1, + enabled: true, + dry_run: true, + llm_client: None, + provider: fabro_llm::Provider::Anthropic, + model: "test-model".to_string(), + }) + .await; + + assert!(retro.is_some()); + let seen = seen.lock().unwrap(); + assert!(seen + .iter() + .any(|event| matches!(event, WorkflowRunEvent::RetroStarted))); + assert!(seen + .iter() + .any(|event| matches!(event, WorkflowRunEvent::RetroCompleted { .. }))); } } diff --git a/lib/crates/fabro-workflows/src/pipeline/types.rs b/lib/crates/fabro-workflows/src/pipeline/types.rs index 2165eb61e..1923f89c4 100644 --- a/lib/crates/fabro-workflows/src/pipeline/types.rs +++ b/lib/crates/fabro-workflows/src/pipeline/types.rs @@ -4,16 +4,17 @@ use std::sync::Arc; use fabro_agent::Sandbox; use fabro_graphviz::graph::Graph; +use fabro_hooks::HookRunner; use fabro_validate::Diagnostic; use crate::checkpoint::Checkpoint; use crate::conclusion::Conclusion; -use crate::engine::{LifecycleConfig, RunSettings, WorkflowRunEngine}; +use crate::context::Context; use crate::error::FabroError; use crate::event::EventEmitter; use crate::handler::HandlerRegistry; use crate::outcome::Outcome; -use fabro_interview::Interviewer; +use crate::run_settings::{LifecycleConfig, RunSettings}; use fabro_validate::Severity; /// Output of the PARSE phase. @@ -99,13 +100,14 @@ pub struct InitOptions { pub run_dir: PathBuf, pub dry_run: bool, pub emitter: Arc, - pub interviewer: Arc, pub sandbox: Arc, - pub registry: HandlerRegistry, + pub registry: Arc, pub lifecycle: LifecycleConfig, pub run_settings: RunSettings, pub hooks: fabro_hooks::HookConfig, pub sandbox_env: HashMap, + pub checkpoint: Option, + pub seed_context: Option, } /// Output of the INITIALIZE phase. @@ -113,11 +115,15 @@ pub struct InitOptions { pub struct Initialized { pub graph: Graph, pub source: String, - pub engine: WorkflowRunEngine, pub settings: RunSettings, pub(crate) checkpoint: Option, + pub(crate) seed_context: Option, pub emitter: Arc, pub sandbox: Arc, + pub registry: Arc, + pub hook_runner: Option>, + pub env: HashMap, + pub dry_run: bool, } /// Output of the EXECUTE phase. @@ -126,10 +132,11 @@ pub struct Executed { pub graph: Graph, pub outcome: Result, pub settings: RunSettings, - pub engine: WorkflowRunEngine, + pub hook_runner: Option>, pub emitter: Arc, pub sandbox: Arc, pub duration_ms: u64, + pub final_context: Context, } /// Output of the RETRO phase. @@ -138,7 +145,7 @@ pub struct Retroed { pub graph: Graph, pub outcome: Result, pub settings: RunSettings, - pub engine: WorkflowRunEngine, + pub hook_runner: Option>, pub emitter: Arc, pub sandbox: Arc, pub duration_ms: u64, @@ -162,6 +169,14 @@ pub struct TransformOptions { /// Options for the RETRO phase. pub struct RetroOptions { + pub run_id: String, + pub workflow_name: String, + pub goal: String, + pub run_dir: PathBuf, + pub sandbox: Arc, + pub emitter: Option>, + pub failed: bool, + pub run_duration_ms: u64, pub enabled: bool, pub dry_run: bool, pub llm_client: Option, @@ -171,6 +186,10 @@ pub struct RetroOptions { /// Options for the FINALIZE phase. pub struct FinalizeOptions { + pub run_dir: PathBuf, + pub run_id: String, + pub workflow_name: String, + pub hook_runner: Option>, pub preserve_sandbox: bool, pub pr_config: Option, pub github_app: Option, diff --git a/lib/crates/fabro-workflows/src/run_dir.rs b/lib/crates/fabro-workflows/src/run_dir.rs new file mode 100644 index 000000000..75d712ff6 --- /dev/null +++ b/lib/crates/fabro-workflows/src/run_dir.rs @@ -0,0 +1,61 @@ +use std::path::{Path, PathBuf}; + +use chrono::Utc; + +use crate::context::Context; +use crate::outcome::{Outcome, OutcomeExt}; +use crate::run_settings::RunSettings; + +/// Write start.json at the start of a workflow run. Returns the StartRecord. +pub(crate) fn write_start_record( + run_dir: &Path, + settings: &RunSettings, +) -> crate::start_record::StartRecord { + let git_state = settings.git.as_ref(); + let record = crate::start_record::StartRecord { + run_id: settings.run_id.clone(), + start_time: Utc::now(), + run_branch: git_state.and_then(|g| g.run_branch.clone()), + base_sha: git_state.and_then(|g| g.base_sha.clone()), + }; + let _ = std::fs::create_dir_all(run_dir); + let _ = record.save(run_dir); + record +} + +/// Return the directory for a node's logs. +/// +/// First visit (`visit <= 1`): `{run_dir}/nodes/{node_id}` +/// Subsequent visits: `{run_dir}/nodes/{node_id}-visit_{visit}` +pub fn node_dir(run_dir: &Path, node_id: &str, visit: usize) -> PathBuf { + if visit <= 1 { + run_dir.join("nodes").join(node_id) + } else { + run_dir + .join("nodes") + .join(format!("{node_id}-visit_{visit}")) + } +} + +/// Read the workflow visit ordinal from context. +/// +/// The raw context value is `0` when unset; workflow execution code treats +/// missing counts as the first visit for stage/log naming. +pub fn visit_from_context(context: &Context) -> usize { + context.node_visit_count().max(1) +} + +/// Write status.json for a completed node into {`run_dir}/nodes/{node_id}/status.json`. +pub(crate) fn write_node_status(run_dir: &Path, node_id: &str, visit: usize, outcome: &Outcome) { + let node_dir = node_dir(run_dir, node_id, visit); + let _ = std::fs::create_dir_all(&node_dir); + let status = serde_json::json!({ + "status": outcome.status.to_string(), + "notes": outcome.notes, + "failure_reason": outcome.failure_reason(), + "timestamp": Utc::now().to_rfc3339(), + }); + if let Ok(json) = serde_json::to_string_pretty(&status) { + let _ = std::fs::write(node_dir.join("status.json"), json); + } +} diff --git a/lib/crates/fabro-workflows/src/run_settings.rs b/lib/crates/fabro-workflows/src/run_settings.rs new file mode 100644 index 000000000..e88f35070 --- /dev/null +++ b/lib/crates/fabro-workflows/src/run_settings.rs @@ -0,0 +1,71 @@ +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::AtomicBool; +use std::sync::Arc; + +use fabro_config::config::FabroConfig; +use fabro_config::run::PullRequestConfig; + +use crate::git::GitAuthor; + +/// Git checkpoint settings for a workflow run. +#[derive(Clone)] +pub struct GitCheckpointSettings { + pub base_sha: Option, + pub run_branch: Option, + pub meta_branch: Option, +} + +/// Configuration for a workflow run. +#[derive(Clone)] +pub struct RunSettings { + pub config: FabroConfig, + pub run_dir: PathBuf, + pub cancel_token: Option>, + pub dry_run: bool, + /// Unique identifier for this workflow run. + pub run_id: String, + /// User-defined key-value labels for this run. + pub labels: HashMap, + /// Git author identity for checkpoint commits. + pub git_author: GitAuthor, + /// Workflow directory slug (e.g. "smoke" from `fabro/workflows/smoke/`). + pub workflow_slug: Option, + /// GitHub App credentials for pushing metadata branches to origin. + pub github_app: Option, + /// Host repo path for MetadataStore (shadow commits) and host-side pushes. + pub host_repo_path: Option, + /// Name of the branch the run was started from (for PR base). + pub base_branch: Option, + /// Git checkpoint settings; `None` means checkpointing disabled. + pub git: Option, +} + +impl RunSettings { + pub fn checkpoint_exclude_globs(&self) -> &[String] { + &self.config.checkpoint.exclude_globs + } + + /// PR config (already normalized — disabled entries stripped at construction). + pub fn pull_request(&self) -> Option<&PullRequestConfig> { + self.config.pull_request.as_ref() + } + + pub fn asset_globs(&self) -> &[String] { + self.config + .assets + .as_ref() + .map(|a| a.include.as_slice()) + .unwrap_or(&[]) + } +} + +/// Configuration for sandbox lifecycle management within the engine. +pub struct LifecycleConfig { + /// Setup commands to run inside the sandbox after initialization. + pub setup_commands: Vec, + /// Timeout in milliseconds for each setup command. + pub setup_command_timeout_ms: u64, + /// Devcontainer lifecycle phases and their commands. + pub devcontainer_phases: Vec<(String, Vec)>, +} diff --git a/lib/crates/fabro-workflows/src/sandbox_git.rs b/lib/crates/fabro-workflows/src/sandbox_git.rs new file mode 100644 index 000000000..617016e55 --- /dev/null +++ b/lib/crates/fabro-workflows/src/sandbox_git.rs @@ -0,0 +1,230 @@ +use std::path::Path; + +use fabro_agent::Sandbox; +use fabro_git_storage::trailerlink::{self, Trailer}; + +use crate::asset_snapshot; + +/// Captured git state for a workflow run, shared with handlers. +#[derive(Debug, Clone)] +pub struct GitState { + pub run_id: String, + pub base_sha: String, + pub run_branch: Option, + pub meta_branch: Option, + pub checkpoint_exclude_globs: Vec, + pub git_author: crate::git::GitAuthor, +} + +pub const GIT_REMOTE: &str = "git -c maintenance.auto=0 -c gc.auto=0"; + +/// Shell-escape a string using `shlex::try_quote` (POSIX-safe). +fn shell_quote(s: &str) -> String { + shlex::try_quote(s).map_or_else( + |_| format!("'{}'", s.replace('\'', "'\\''")), + |q| q.to_string(), + ) +} + +/// Run a git checkpoint commit via the sandbox. +#[allow(clippy::too_many_arguments)] +pub async fn git_checkpoint( + sandbox: &dyn Sandbox, + run_id: &str, + node_id: &str, + status: &str, + completed_count: usize, + shadow_sha: Option, + exclude_globs: &[String], + author: &crate::git::GitAuthor, +) -> std::result::Result { + let mut all_excludes: Vec = asset_snapshot::EXCLUDE_DIRS + .iter() + .map(|d| format!("**/{d}/**")) + .collect(); + all_excludes.extend(exclude_globs.iter().cloned()); + + let pathspecs: Vec = all_excludes + .iter() + .map(|g| format!("':(glob,exclude){g}'")) + .collect(); + let add_cmd = format!("{GIT_REMOTE} add -A -- . {}", pathspecs.join(" ")); + let add_result = sandbox + .exec_command(&add_cmd, 30_000, None, None, None) + .await; + match &add_result { + Ok(r) if r.exit_code == 0 => {} + Ok(r) => { + return Err(format!( + "git add failed (exit {}): {}{}", + r.exit_code, r.stdout, r.stderr + )); + } + Err(e) => return Err(format!("git add failed: {e}")), + } + + let subject = format!("fabro({run_id}): {node_id} ({status})"); + let completed_str = completed_count.to_string(); + let mut trailers = vec![ + Trailer { + key: "Fabro-Run", + value: run_id, + }, + Trailer { + key: "Fabro-Completed", + value: &completed_str, + }, + ]; + let shadow_sha_ref = shadow_sha.as_deref().unwrap_or(""); + if shadow_sha.is_some() { + trailers.push(Trailer { + key: "Fabro-Checkpoint", + value: shadow_sha_ref, + }); + } + let mut message = trailerlink::format_message(&subject, "", &trailers); + author.append_footer(&mut message); + + let msg_path = format!("/tmp/fabro-commit-msg-{run_id}-{node_id}"); + if let Err(e) = sandbox.write_file(&msg_path, &message).await { + return Err(format!("failed to write commit message file: {e}")); + } + + let commit_cmd = format!( + "{GIT_REMOTE} -c user.name={name} -c user.email={email} commit --allow-empty -F {msg_path}", + name = shell_quote(&author.name), + email = shell_quote(&author.email), + ); + let commit_result = sandbox + .exec_command(&commit_cmd, 30_000, None, None, None) + .await; + match &commit_result { + Ok(r) if r.exit_code == 0 => {} + Ok(r) => { + return Err(format!( + "git commit failed (exit {}): {}{}", + r.exit_code, r.stdout, r.stderr + )); + } + Err(e) => return Err(format!("git commit failed: {e}")), + } + + let sha_cmd = format!("{GIT_REMOTE} rev-parse HEAD"); + let sha_result = sandbox + .exec_command(&sha_cmd, 10_000, None, None, None) + .await; + match sha_result { + Ok(r) if r.exit_code == 0 => Ok(r.stdout.trim().to_string()), + Ok(r) => Err(format!( + "git rev-parse HEAD failed (exit {}): {}{}", + r.exit_code, r.stdout, r.stderr + )), + Err(e) => Err(format!("git rev-parse HEAD failed: {e}")), + } +} + +/// Push a refspec from the host repo to origin (best-effort). +/// +/// Authenticates via a GitHub App installation token so we don't depend +/// on the host's ambient git credentials. +pub async fn git_push_host( + repo_path: &Path, + refspec: &str, + github_app: &Option, + label: &str, +) -> bool { + let (origin_url, _) = match fabro_sandbox::daytona::detect_repo_info(repo_path) { + Ok(info) => info, + Err(e) => { + tracing::warn!(error = %e, label, "Cannot detect origin for push"); + return false; + } + }; + + let https_url = fabro_github::ssh_url_to_https(&origin_url); + let push_url = match github_app { + Some(creds) => match fabro_github::resolve_authenticated_url(creds, &https_url).await { + Ok(url) => url, + Err(e) => { + tracing::warn!(error = %e, label, "Failed to get token for push"); + return false; + } + }, + None => { + tracing::warn!(label, "No GitHub App credentials for push"); + return false; + } + }; + + let rp = repo_path.to_path_buf(); + let refspec_owned = refspec.to_string(); + let result = crate::git::blocking_push_with_timeout(60, move || { + crate::git::push_ref(&rp, &push_url, &refspec_owned) + }) + .await; + match result { + Ok(()) => { + tracing::info!(label, "Pushed to origin"); + true + } + Err(e) => { + tracing::warn!(error = %e, label, "Failed to push"); + false + } + } +} + +/// Run a git diff via the sandbox. +pub(crate) async fn git_diff( + sandbox: &dyn Sandbox, + base: &str, +) -> std::result::Result { + let cmd = format!("{GIT_REMOTE} diff {base} HEAD"); + match sandbox.exec_command(&cmd, 30_000, None, None, None).await { + Ok(r) if r.exit_code == 0 => Ok(r.stdout), + Ok(r) => Err(format!("exit {}: {}", r.exit_code, r.stderr.trim())), + Err(e) => Err(e.to_string()), + } +} + +/// Create a branch at a specific SHA via the sandbox. +pub async fn git_create_branch_at(sandbox: &dyn Sandbox, name: &str, sha: &str) -> bool { + let cmd = format!("{GIT_REMOTE} branch --force {name} {sha}"); + matches!( + sandbox.exec_command(&cmd, 30_000, None, None, None).await, + Ok(r) if r.exit_code == 0 + ) +} + +/// Add a git worktree via the sandbox. +pub async fn git_add_worktree(sandbox: &dyn Sandbox, path: &str, branch: &str) -> bool { + let cmd = format!("{GIT_REMOTE} worktree add {path} {branch}"); + matches!( + sandbox.exec_command(&cmd, 30_000, None, None, None).await, + Ok(r) if r.exit_code == 0 + ) +} + +/// Remove a git worktree via the sandbox. +pub async fn git_remove_worktree(sandbox: &dyn Sandbox, path: &str) -> bool { + let cmd = format!("{GIT_REMOTE} worktree remove --force {path}"); + matches!( + sandbox.exec_command(&cmd, 30_000, None, None, None).await, + Ok(r) if r.exit_code == 0 + ) +} + +/// Fast-forward merge to a given SHA via the sandbox. +pub async fn git_merge_ff_only(sandbox: &dyn Sandbox, sha: &str) -> bool { + let cmd = format!("{GIT_REMOTE} merge --ff-only {sha}"); + matches!( + sandbox.exec_command(&cmd, 30_000, None, None, None).await, + Ok(r) if r.exit_code == 0 + ) +} + +/// Remove any stale worktree at `path` (best-effort), then add a fresh one. +pub async fn git_replace_worktree(sandbox: &dyn Sandbox, path: &str, branch: &str) -> bool { + let _ = git_remove_worktree(sandbox, path).await; + git_add_worktree(sandbox, path, branch).await +}