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

Extract engine helpers and implement pipeline phases

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-24 22:31:08 -04:00
parent 606af8b1ea
commit e55c9a3189
13 changed files with 2223 additions and 1212 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

409
lib/crates/fabro-cli/src/commands/run.rs
View file

@ -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<Outcome, fabro_workflows::error::FabroError>,
) -> (StageStatus, Option<String>, RunStatus, Option<StatusReason>) {
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<String>,
run_duration_ms: u64,
final_git_commit_sha: Option<String>,
) -> 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<f64> = 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<StatusReason>,
) {
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<Arc<EventEmitter>>,
) {
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<Arc<dyn Fn(fabro_agent::SessionEvent) + Send + Sync>> =
emitter_clone.map(
|em| -> Arc<dyn Fn(fabro_agent::SessionEvent) + Send + Sync> {
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"));
}
}

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

@ -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<FailureCategory> {
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<String> {
// 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<f64> = 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<EdgeSelection<'a>> {
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<String, Outcome>,
) -> 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<String> {
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<String> {
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<String>,
pub meta_branch: Option<String>,
pub checkpoint_exclude_globs: Vec<String>,
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<String>,
exclude_globs: &[String],
author: &crate::git::GitAuthor,
) -> std::result::Result<String, String> {
// Stage everything, always excluding EXCLUDE_DIRS plus any user-configured globs
let mut all_excludes: Vec<String> = asset_snapshot::EXCLUDE_DIRS
.iter()
.map(|d| format!("**/{d}/**"))
.collect();
all_excludes.extend(exclude_globs.iter().cloned());
let pathspecs: Vec<String> = 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<fabro_github::GitHubAppCredentials>,
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<String, String> {
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<String>,
pub run_branch: Option<String>,
pub meta_branch: Option<String>,
}
/// Configuration for a workflow run.
#[derive(Clone)]
pub struct RunSettings {
pub config: FabroConfig,
pub run_dir: PathBuf,
pub cancel_token: Option<Arc<AtomicBool>>,
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<String, String>,
/// 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<String>,
/// GitHub App credentials for pushing metadata branches to origin.
pub github_app: Option<fabro_github::GitHubAppCredentials>,
/// Host repo path for MetadataStore (shadow commits) and host-side pushes.
pub host_repo_path: Option<PathBuf>,
/// Name of the branch the run was started from (for PR base).
pub base_branch: Option<String>,
/// Git checkpoint settings; `None` means checkpointing disabled.
pub git: Option<GitCheckpointSettings>,
}
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<String>,
/// 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<fabro_devcontainer::Command>)>,
}
#[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 {

424
lib/crates/fabro-workflows/src/graph_ops.rs Normal file
View file

@ -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<FailureCategory> {
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<String> {
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<f64> = 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<EdgeSelection<'a>> {
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<String, Outcome>,
) -> 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<String> {
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<String> {
node.attrs
.get("script")
.or_else(|| node.attrs.get("tool_command"))
.and_then(|v| v.as_str())
.map(String::from)
}

4
lib/crates/fabro-workflows/src/lib.rs
View file

@ -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;

371
lib/crates/fabro-workflows/src/pipeline/execute.rs
View file

@ -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<dyn fabro_core::handler::NodeHandler<WorkflowGraph>>)
.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"))
);
}
}

435
lib/crates/fabro-workflows/src/pipeline/finalize.rs
View file

@ -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<String>,
message: impl Into<String>,
) {
emitter.emit(&WorkflowRunEvent::RunNotice {
level,
code: code.into(),
message: message.into(),
});
}
pub fn classify_engine_result(
engine_result: &Result<Outcome, FabroError>,
) -> (StageStatus, Option<String>, RunStatus, Option<StatusReason>) {
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<String>,
run_duration_ms: u64,
final_git_commit_sha: Option<String>,
) -> 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<f64> = 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<StatusReason>,
) {
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<dyn fabro_agent::Sandbox>,
) {
let Some(runner) = hook_runner else {
return;
};
let _ = runner.run(hook_context, sandbox, None).await;
}
async fn cleanup_sandbox(
hook_runner: Option<Arc<HookRunner>>,
sandbox: Arc<dyn fabro_agent::Sandbox>,
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<Finalized, FabroError> {
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);
}
}

286
lib/crates/fabro-workflows/src/pipeline/initialize.rs
View file

@ -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<dyn fabro_agent::Sandbox>,
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")
);
}
}

7
lib/crates/fabro-workflows/src/pipeline/mod.rs
View file

@ -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;

266
lib/crates/fabro-workflows/src/pipeline/retro.rs
View file

@ -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<Retro> {
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<Arc<dyn Fn(SessionEvent) + Send + Sync>> =
emitter_clone.map(|emitter| -> Arc<dyn Fn(SessionEvent) + Send + Sync> {
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<dyn fabro_agent::Sandbox> = 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 { .. })));
}
}

33
lib/crates/fabro-workflows/src/pipeline/types.rs
View file

@ -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<EventEmitter>,
pub interviewer: Arc<dyn Interviewer>,
pub sandbox: Arc<dyn Sandbox>,
pub registry: HandlerRegistry,
pub registry: Arc<HandlerRegistry>,
pub lifecycle: LifecycleConfig,
pub run_settings: RunSettings,
pub hooks: fabro_hooks::HookConfig,
pub sandbox_env: HashMap<String, String>,
pub checkpoint: Option<Checkpoint>,
pub seed_context: Option<Context>,
}
/// 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<Checkpoint>,
pub(crate) seed_context: Option<Context>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
pub registry: Arc<HandlerRegistry>,
pub hook_runner: Option<Arc<HookRunner>>,
pub env: HashMap<String, String>,
pub dry_run: bool,
}
/// Output of the EXECUTE phase.
@ -126,10 +132,11 @@ pub struct Executed {
pub graph: Graph,
pub outcome: Result<Outcome, FabroError>,
pub settings: RunSettings,
pub engine: WorkflowRunEngine,
pub hook_runner: Option<Arc<HookRunner>>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
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<Outcome, FabroError>,
pub settings: RunSettings,
pub engine: WorkflowRunEngine,
pub hook_runner: Option<Arc<HookRunner>>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
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<dyn Sandbox>,
pub emitter: Option<Arc<EventEmitter>>,
pub failed: bool,
pub run_duration_ms: u64,
pub enabled: bool,
pub dry_run: bool,
pub llm_client: Option<fabro_llm::client::Client>,
@ -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<Arc<HookRunner>>,
pub preserve_sandbox: bool,
pub pr_config: Option<fabro_config::run::PullRequestConfig>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,

61
lib/crates/fabro-workflows/src/run_dir.rs Normal file
View file

@ -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);
}
}

71
lib/crates/fabro-workflows/src/run_settings.rs Normal file
View file

@ -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<String>,
pub run_branch: Option<String>,
pub meta_branch: Option<String>,
}
/// Configuration for a workflow run.
#[derive(Clone)]
pub struct RunSettings {
pub config: FabroConfig,
pub run_dir: PathBuf,
pub cancel_token: Option<Arc<AtomicBool>>,
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<String, String>,
/// Git author identity for checkpoint commits.
pub git_author: GitAuthor,
/// Workflow directory slug (e.g. "smoke" from `fabro/workflows/smoke/`).
pub workflow_slug: Option<String>,
/// GitHub App credentials for pushing metadata branches to origin.
pub github_app: Option<fabro_github::GitHubAppCredentials>,
/// Host repo path for MetadataStore (shadow commits) and host-side pushes.
pub host_repo_path: Option<PathBuf>,
/// Name of the branch the run was started from (for PR base).
pub base_branch: Option<String>,
/// Git checkpoint settings; `None` means checkpointing disabled.
pub git: Option<GitCheckpointSettings>,
}
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<String>,
/// 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<fabro_devcontainer::Command>)>,
}

230
lib/crates/fabro-workflows/src/sandbox_git.rs Normal file
View file

@ -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<String>,
pub meta_branch: Option<String>,
pub checkpoint_exclude_globs: Vec<String>,
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<String>,
exclude_globs: &[String],
author: &crate::git::GitAuthor,
) -> std::result::Result<String, String> {
let mut all_excludes: Vec<String> = asset_snapshot::EXCLUDE_DIRS
.iter()
.map(|d| format!("**/{d}/**"))
.collect();
all_excludes.extend(exclude_globs.iter().cloned());
let pathspecs: Vec<String> = 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<fabro_github::GitHubAppCredentials>,
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<String, String> {
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
}