scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Thin CLI run commands and move execution into workflows operations

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-27 16:27:55 -04:00
parent defcf9c746
commit db3231e987
26 changed files with 2200 additions and 3072 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
lib/crates/fabro-api/src/server.rs
View file

@ -21,7 +21,7 @@ use crate::jwt_auth::{AuthMode, AuthenticatedService, AuthenticatedUser};
use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer};
use fabro_workflows::context::Context;
use fabro_workflows::event::{EventEmitter, WorkflowRunEvent};
use fabro_workflows::operations::{self, RunCreateOptions};
use fabro_workflows::operations::{self, CreateRequest, WorkflowInput};
use fabro_workflows::pipeline::{
self, InitOptions, LlmSpec, Persisted, SandboxEnvSpec, SandboxSpec,
};
@ -526,25 +526,19 @@ async fn start_run(
}),
..Default::default()
};
let run_labels = settings.labels.clone();
let persisted = match operations::create(
&req.dot_source,
RunCreateOptions {
settings,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: None,
labels: run_labels,
base_branch: None,
working_directory: Some(
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
),
host_repo_path: None,
goal_override: None,
let created = match operations::create(CreateRequest {
workflow: WorkflowInput::DotSource {
source: req.dot_source.clone(),
base_dir: None,
workflow_slug: None,
},
) {
Ok(persisted) => persisted,
settings,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
host_repo_path: None,
base_branch: None,
}) {
Ok(created) => created,
Err(ref err @ fabro_workflows::error::FabroError::ValidationFailed { ref diagnostics }) => {
let message = if diagnostics.is_empty() {
err.to_string()
@ -568,6 +562,7 @@ async fn start_run(
.into_response();
}
};
let persisted = created.persisted;
let created_at = persisted.run_record().created_at;
{

8
lib/crates/fabro-cli/src/args.rs
View file

@ -646,12 +646,12 @@ pub(crate) enum RunCommands {
/// Internal: run the engine process (reads run.json from run dir)
#[command(name = "__detached", hide = true)]
Detached {
/// Base storage directory
/// Run directory
#[arg(long)]
storage_dir: PathBuf,
/// Run ID
run_dir: PathBuf,
/// Launcher metadata path
#[arg(long)]
run_id: String,
launcher_path: PathBuf,
/// Resume from checkpoint instead of fresh start
#[arg(long)]
resume: bool,

26
lib/crates/fabro-cli/src/commands/config/mod.rs
View file

@ -2,7 +2,6 @@ use std::io::Write;
use std::path::Path;
use crate::args::{ConfigCommand, ConfigNamespace, ConfigShowArgs};
use anyhow::bail;
use fabro_config::{FabroConfig, FabroSettings};
pub fn dispatch(ns: ConfigNamespace) -> anyhow::Result<()> {
@ -13,25 +12,14 @@ pub fn dispatch(ns: ConfigNamespace) -> anyhow::Result<()> {
fn merged_config(workflow: Option<&Path>) -> anyhow::Result<FabroSettings> {
if let Some(workflow) = workflow {
let (resolved_path, _dot_path, run_config) =
crate::commands::run::execute::resolve_workflow_source(workflow)?;
let missing_workflow = run_config.is_none() && !resolved_path.is_file();
let project_config = fabro_config::project::discover_project_config(
resolved_path.parent().unwrap_or_else(|| Path::new(".")),
)?
.map(|(_, config)| config)
.unwrap_or_default();
let cli_config = fabro_config::cli::load_cli_config(None)?;
let config = run_config
.unwrap_or_default()
.combine(project_config)
.combine(cli_config);
if missing_workflow {
bail!("Workflow not found: {}", resolved_path.display());
}
return config.try_into();
return fabro_workflows::operations::resolve_settings_for_path(
workflow,
cli_config,
FabroConfig::default(),
true,
)
.map_err(Into::into);
}
let cwd = std::env::current_dir()?;

502
lib/crates/fabro-cli/src/commands/preflight.rs
View file

@ -1,12 +1,14 @@
use std::path::Path;
use std::sync::Arc;
use anyhow::bail;
use fabro_agent::{DockerSandbox, DockerSandboxConfig, LocalSandbox, Sandbox};
use fabro_config::{FabroConfig, FabroSettings};
use fabro_model::{Catalog, Provider};
use fabro_sandbox::SandboxProvider;
use fabro_util::terminal::Styles;
use fabro_workflows::git::GitSyncStatus;
use super::run::execute::{
load_workflow_source_input, print_workflow_report, resolve_sandbox_provider, run_preflight,
};
use crate::args::PreflightArgs;
pub async fn execute(mut args: PreflightArgs) -> anyhow::Result<()> {
@ -17,51 +19,503 @@ pub async fn execute(mut args: PreflightArgs) -> anyhow::Result<()> {
let github_app = crate::shared::github::build_github_app_credentials(cli_config.app_id());
let cli_args_config = FabroConfig::try_from(&args)?;
let settings = fabro_workflows::operations::resolve_settings_for_path(
&args.workflow,
cli_defaults,
cli_args_config,
true,
)?;
let resolved = fabro_workflows::operations::resolve_workflow(
fabro_workflows::operations::ResolveWorkflowRequest {
workflow: fabro_workflows::operations::WorkflowInput::Path(args.workflow.clone()),
settings: settings.clone(),
},
)?;
let source_input =
load_workflow_source_input(&args.workflow, cli_args_config, cli_defaults, true)?;
let original_cwd = std::env::current_dir()?;
let (origin_url, detected_base_branch) =
fabro_sandbox::daytona::detect_repo_info(&original_cwd)
fabro_sandbox::daytona::detect_repo_info(&resolved.working_directory)
.map(|(url, branch)| (Some(url), branch))
.unwrap_or((None, None));
let git_status =
fabro_workflows::git::sync_status(&original_cwd, "origin", detected_base_branch.as_deref());
let git_status = fabro_workflows::git::sync_status(
&resolved.working_directory,
"origin",
detected_base_branch.as_deref(),
);
let sandbox_provider =
resolve_sandbox_provider(args.sandbox.map(Into::into), &source_input.settings)?;
let sandbox_provider = resolve_sandbox_provider(args.sandbox.map(Into::into), &settings)?;
let validated = fabro_workflows::operations::validate(
&source_input.raw_source,
&resolved.raw_source,
fabro_workflows::operations::ValidateOptions {
base_dir: Some(
source_input
.dot_path
.parent()
.unwrap_or(Path::new("."))
.to_path_buf(),
),
settings: Some(source_input.settings.clone()),
goal_override: source_input.goal_override.clone(),
base_dir: resolved.base_dir.clone(),
settings: Some(resolved.settings.clone()),
goal_override: resolved.goal_override.clone(),
..Default::default()
},
)?;
print_workflow_report(&validated, &source_input.dot_path, styles);
super::run::output::print_workflow_report(&validated, resolved.dot_path.as_deref(), styles);
if validated.has_errors() {
bail!("Validation failed");
}
run_preflight(
validated.graph(),
&source_input.settings,
&resolved.settings,
args.model.as_deref(),
args.provider.as_deref(),
git_status,
sandbox_provider,
&resolved.working_directory,
styles,
github_app,
origin_url.as_deref(),
)
.await
}
fn resolve_model_provider(
cli_model: Option<&str>,
cli_provider: Option<&str>,
settings: &FabroSettings,
graph: &fabro_graphviz::graph::Graph,
) -> (String, Option<String>) {
let configured_model = settings.llm.as_ref().and_then(|llm| llm.model.as_deref());
let configured_provider = settings
.llm
.as_ref()
.and_then(|llm| llm.provider.as_deref());
let provider = cli_provider
.or(configured_provider)
.or_else(|| graph.attrs.get("default_provider").and_then(|v| v.as_str()))
.map(String::from);
let model = cli_model
.or(configured_model)
.or_else(|| graph.attrs.get("default_model").and_then(|v| v.as_str()))
.map(String::from)
.unwrap_or_else(|| {
let catalog = Catalog::builtin();
let info = provider
.as_deref()
.and_then(|s| s.parse::<Provider>().ok())
.and_then(|p| catalog.default_for_provider(p))
.unwrap_or_else(|| catalog.default_from_env());
info.id.clone()
});
match Catalog::builtin().get(&model) {
Some(info) => (
info.id.clone(),
provider.or(Some(info.provider.to_string())),
),
None => (model, provider),
}
}
fn parse_sandbox_provider(settings: &FabroSettings) -> anyhow::Result<Option<SandboxProvider>> {
settings
.sandbox_settings()
.and_then(|s| s.provider.as_deref())
.map(|s| s.parse::<SandboxProvider>())
.transpose()
.map_err(|e| anyhow::anyhow!("Invalid sandbox provider: {e}"))
}
fn resolve_sandbox_provider(
cli: Option<SandboxProvider>,
settings: &FabroSettings,
) -> anyhow::Result<SandboxProvider> {
Ok(cli
.or(parse_sandbox_provider(settings)?)
.unwrap_or_default())
}
fn resolve_daytona_config(
settings: &FabroSettings,
) -> Option<fabro_sandbox::daytona::DaytonaConfig> {
settings
.sandbox_settings()
.and_then(|sandbox| sandbox.daytona.clone())
}
#[cfg(feature = "exedev")]
fn resolve_exe_config(settings: &FabroSettings) -> Option<fabro_sandbox::exe::ExeConfig> {
settings
.sandbox_settings()
.and_then(|sandbox| sandbox.exe.clone())
}
#[cfg(feature = "exedev")]
fn resolve_exe_clone_params(cwd: &Path) -> Option<fabro_sandbox::exe::GitCloneParams> {
let (detected_url, branch) = match fabro_sandbox::daytona::detect_repo_info(cwd) {
Ok(info) => info,
Err(err) => {
tracing::warn!("No git repo detected for exe.dev clone: {err}");
return None;
}
};
let url = fabro_github::ssh_url_to_https(&detected_url);
Some(fabro_sandbox::exe::GitCloneParams { url, branch })
}
fn resolve_ssh_config(settings: &FabroSettings) -> Option<fabro_sandbox::ssh::SshConfig> {
settings
.sandbox_settings()
.and_then(|sandbox| sandbox.ssh.clone())
}
fn resolve_ssh_clone_params(cwd: &Path) -> Option<fabro_sandbox::ssh::GitCloneParams> {
let (detected_url, branch) = match fabro_sandbox::daytona::detect_repo_info(cwd) {
Ok(info) => info,
Err(err) => {
tracing::warn!("No git repo detected for SSH clone: {err}");
return None;
}
};
let url = fabro_github::ssh_url_to_https(&detected_url);
Some(fabro_sandbox::ssh::GitCloneParams { url, branch })
}
async fn mint_github_token(
creds: &fabro_github::GitHubAppCredentials,
origin_url: &str,
permissions: &std::collections::HashMap<String, String>,
) -> anyhow::Result<String> {
let https_url = fabro_github::ssh_url_to_https(origin_url);
let (owner, repo) =
fabro_github::parse_github_owner_repo(&https_url).map_err(|e| anyhow::anyhow!("{e}"))?;
let jwt = fabro_github::sign_app_jwt(&creds.app_id, &creds.private_key_pem)
.map_err(|e| anyhow::anyhow!("{e}"))?;
let client = reqwest::Client::new();
let perms_json = serde_json::to_value(permissions)?;
let token = fabro_github::create_installation_access_token_with_permissions(
&client,
&jwt,
&owner,
&repo,
fabro_github::GITHUB_API_BASE_URL,
perms_json,
)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
Ok(token)
}
#[allow(clippy::too_many_arguments)]
async fn run_preflight(
graph: &fabro_graphviz::graph::Graph,
settings: &FabroSettings,
cli_model: Option<&str>,
cli_provider: Option<&str>,
git_status: GitSyncStatus,
sandbox_provider: SandboxProvider,
working_directory: &Path,
styles: &'static Styles,
github_app: Option<fabro_github::GitHubAppCredentials>,
origin_url: Option<&str>,
) -> anyhow::Result<()> {
use fabro_util::check_report::{
CheckDetail, CheckReport, CheckResult, CheckSection, CheckStatus,
};
let spinner = indicatif::ProgressBar::new_spinner();
spinner.set_style(
indicatif::ProgressStyle::with_template("{spinner:.cyan} {msg}")
.expect("valid template")
.tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏", ""]),
);
spinner.set_message("Running preflight checks...");
spinner.enable_steady_tick(std::time::Duration::from_millis(80));
let mut checks: Vec<CheckResult> = Vec::new();
let setup_command_count = settings.setup_commands().len();
let repo_summary = origin_url
.map(|url| {
let https = fabro_github::ssh_url_to_https(url);
fabro_github::parse_github_owner_repo(&https)
.map(|(owner, repo)| format!("{owner}/{repo}"))
.unwrap_or_else(|_| url.to_string())
})
.unwrap_or_else(|| "unknown".into());
checks.push(CheckResult {
name: "Repository".into(),
status: CheckStatus::Pass,
summary: repo_summary,
details: vec![
CheckDetail::new(format!("Setup commands: {setup_command_count}")),
CheckDetail {
text: format!("Git: {git_status}"),
warn: git_status != GitSyncStatus::Synced,
},
],
remediation: None,
});
let (model, provider) = resolve_model_provider(cli_model, cli_provider, settings, graph);
checks.push(CheckResult {
name: "Workflow".into(),
status: CheckStatus::Pass,
summary: graph.name.clone(),
details: vec![
CheckDetail::new(format!("Nodes: {}", graph.nodes.len())),
CheckDetail::new(format!("Edges: {}", graph.edges.len())),
CheckDetail::new(format!("Goal: {}", graph.goal())),
],
remediation: None,
});
let daytona_config = resolve_daytona_config(settings);
#[cfg(feature = "exedev")]
let exe_config = resolve_exe_config(settings);
let ssh_config = resolve_ssh_config(settings);
let sandbox_result: Result<Arc<dyn Sandbox>, String> = match sandbox_provider {
SandboxProvider::Docker => {
let config = DockerSandboxConfig {
host_working_directory: working_directory.to_string_lossy().to_string(),
..DockerSandboxConfig::default()
};
DockerSandbox::new(config)
.map(|env| Arc::new(env) as Arc<dyn Sandbox>)
.map_err(|e| format!("Docker sandbox creation failed: {e}"))
}
SandboxProvider::Daytona => {
let config = daytona_config.unwrap_or_default();
match fabro_sandbox::daytona::DaytonaSandbox::new(
config,
github_app.clone(),
None,
None,
)
.await
{
Ok(env) => Ok(Arc::new(env) as Arc<dyn Sandbox>),
Err(e) => Err(format!("Daytona sandbox creation failed: {e}")),
}
}
#[cfg(feature = "exedev")]
SandboxProvider::Exe => {
match fabro_sandbox::exe::OpensshRunner::connect_raw("exe.dev").await {
Ok(mgmt_ssh) => {
let config = exe_config.unwrap_or_default();
let clone_params = resolve_exe_clone_params(working_directory);
let env = fabro_sandbox::exe::ExeSandbox::new(
Box::new(mgmt_ssh),
config,
clone_params,
None,
None,
);
Ok(Arc::new(env) as Arc<dyn Sandbox>)
}
Err(e) => Err(format!("exe.dev SSH connection failed: {e}")),
}
}
#[cfg(not(feature = "exedev"))]
SandboxProvider::Exe => Err("exe sandbox requires the exedev feature".to_string()),
SandboxProvider::Ssh => match ssh_config {
Some(config) => {
let clone_params = resolve_ssh_clone_params(working_directory);
let env = fabro_sandbox::ssh::SshSandbox::new(config, clone_params, None, None);
Ok(Arc::new(env) as Arc<dyn Sandbox>)
}
None => Err("SSH sandbox requires [sandbox.ssh] config".to_string()),
},
SandboxProvider::Local => {
Ok(Arc::new(LocalSandbox::new(working_directory.to_path_buf())) as Arc<dyn Sandbox>)
}
};
let sandbox_ok = match sandbox_result {
Ok(sandbox) => match sandbox.initialize().await {
Ok(()) => {
let _ = sandbox.cleanup().await;
true
}
Err(e) => {
let _ = sandbox.cleanup().await;
checks.push(CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Error,
summary: "failed".into(),
details: vec![CheckDetail::new(format!("Provider: {sandbox_provider}"))],
remediation: Some(format!("Sandbox init failed: {e}")),
});
false
}
},
Err(e) => {
checks.push(CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Error,
summary: "failed".into(),
details: vec![CheckDetail::new(format!("Provider: {sandbox_provider}"))],
remediation: Some(e),
});
false
}
};
if sandbox_ok {
checks.push(CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Pass,
summary: sandbox_provider.to_string(),
details: vec![CheckDetail::new(format!("Provider: {sandbox_provider}"))],
remediation: None,
});
}
let default_provider = provider.as_deref().unwrap_or("anthropic");
let llm_ok = match fabro_llm::client::Client::from_env().await {
Ok(c) => {
let configured: Vec<String> =
c.provider_names().iter().map(|s| s.to_string()).collect();
let mut model_providers = std::collections::BTreeSet::new();
for node in graph.nodes.values() {
if !fabro_graphviz::graph::is_llm_handler_type(node.handler_type()) {
continue;
}
let node_model = node.model().unwrap_or(&model);
let node_provider = node.provider().unwrap_or(default_provider);
let (resolved_model, resolved_provider) =
if let Some(info) = Catalog::builtin().get(node_model) {
(info.id.clone(), info.provider.to_string())
} else {
(node_model.to_string(), node_provider.to_string())
};
let final_provider = if node.provider().is_some() {
node_provider.to_string()
} else {
resolved_provider
};
model_providers.insert((resolved_model, final_provider));
}
if model_providers.is_empty() {
let (resolved_model, resolved_provider) =
if let Some(info) = Catalog::builtin().get(&model) {
(info.id.clone(), info.provider.to_string())
} else {
(model.clone(), default_provider.to_string())
};
model_providers.insert((resolved_model, resolved_provider));
}
let mut all_ok = true;
for (model_id, provider_name) in &model_providers {
match provider_name.parse::<Provider>() {
Ok(_) => {
let mut status = CheckStatus::Pass;
if !configured.iter().any(|n| n == provider_name) {
status = CheckStatus::Warning;
all_ok = false;
}
checks.push(CheckResult {
name: "LLM".into(),
status,
summary: model_id.clone(),
details: vec![CheckDetail::new(format!("Provider: {provider_name}"))],
remediation: if status == CheckStatus::Warning {
Some(format!("Provider \"{provider_name}\" is not configured"))
} else {
None
},
});
}
Err(e) => {
checks.push(CheckResult {
name: "LLM".into(),
status: CheckStatus::Error,
summary: model_id.clone(),
details: vec![CheckDetail::new(format!("Provider: {provider_name}"))],
remediation: Some(format!("Invalid provider \"{provider_name}\": {e}")),
});
all_ok = false;
}
}
}
all_ok
}
Err(e) => {
checks.push(CheckResult {
name: "LLM".into(),
status: CheckStatus::Error,
summary: "initialization failed".into(),
details: vec![],
remediation: Some(format!("LLM client init failed: {e}")),
});
false
}
};
if let Some(github_permissions) = settings.github_permissions() {
if !github_permissions.is_empty() {
let perm_details: Vec<CheckDetail> = github_permissions
.iter()
.map(|(k, v)| CheckDetail::new(format!("{k}: {v}")))
.collect();
match (&github_app, origin_url) {
(Some(creds), Some(url)) => {
match mint_github_token(creds, url, github_permissions).await {
Ok(_) => {
checks.push(CheckResult {
name: "GitHub Token".into(),
status: CheckStatus::Pass,
summary: "minted".into(),
details: perm_details,
remediation: None,
});
}
Err(e) => {
checks.push(CheckResult {
name: "GitHub Token".into(),
status: CheckStatus::Error,
summary: "failed".into(),
details: perm_details,
remediation: Some(format!("Failed to mint GitHub token: {e}")),
});
}
}
}
_ => {
checks.push(CheckResult {
name: "GitHub Token".into(),
status: CheckStatus::Warning,
summary: "skipped".into(),
details: vec![],
remediation: Some(
"No GitHub App credentials or origin URL available".to_string(),
),
});
}
}
}
}
spinner.finish_and_clear();
let report = CheckReport {
title: "Run Preflight".into(),
sections: vec![CheckSection {
title: String::new(),
checks,
}],
};
let term_width = console::Term::stderr().size().1;
print!("{}", report.render(styles, true, None, Some(term_width)));
if sandbox_ok && llm_ok {
Ok(())
} else {
std::process::exit(1);
}
}

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

@ -9,10 +9,8 @@ use anyhow::{bail, Result};
use fabro_interview::{AnswerValue, ConsoleInterviewer};
use fabro_util::terminal::Styles;
use fabro_workflows::event::RunNoticeLevel;
use fabro_workflows::run_status::{RunStatus, RunStatusRecord};
use super::detached::append_run_notice;
use super::run_progress;
#[cfg(test)]
@ -36,7 +34,6 @@ pub async fn attach_run(
let status_path = run_dir.join("status.json");
let interview_request_path = run_dir.join("interview_request.json");
let interview_response_path = run_dir.join("interview_response.json");
let pid_path = run_dir.join("run.pid");
let mut engine_guard = engine_child.map(EngineChildGuard::new);
@ -86,6 +83,16 @@ pub async fn attach_run(
}
}
if let Some(pid) = read_launcher_pid(run_dir) {
if !process_alive(pid) && wait_count > 5 {
progress_ui.finish();
return Ok(determine_exit_code(
&conclusion_path,
read_status_record(&status_path),
));
}
}
if wait_count > 100 {
// Guard's Drop kills+waits on the engine child
drop(engine_guard.take());
@ -114,8 +121,13 @@ pub async fn attach_run(
loop {
if cancelled.load(Ordering::Relaxed) {
if kill_on_detach {
// Kill the engine process
kill_engine(&pid_path);
if let Some(guard) = engine_guard.as_mut() {
if let Some(child) = guard.inner() {
let _ = child.kill();
}
} else {
kill_engine(run_dir);
}
// Wait briefly for a terminal status or conclusion
for _ in 0..20 {
if conclusion_path.exists()
@ -168,12 +180,6 @@ pub async fn attach_run(
progress_ui.show_bars();
if answer_requires_reattach(&answer) {
let _ = append_run_notice(
run_dir,
RunNoticeLevel::Warn,
"interview_unanswered",
INTERVIEW_UNANSWERED_MESSAGE,
);
if let Some(guard) = engine_guard.as_mut() {
guard.defuse();
}
@ -213,7 +219,7 @@ pub async fn attach_run(
let engine_alive = match cached_pid {
Some(pid) => process_alive(pid),
None => {
if let Some(pid) = read_pid(&pid_path) {
if let Some(pid) = read_launcher_pid(run_dir) {
cached_pid = Some(pid);
process_alive(pid)
} else {
@ -264,10 +270,14 @@ fn read_status_record(path: &Path) -> Option<RunStatusRecord> {
RunStatusRecord::load(path).ok()
}
fn read_pid(pid_path: &Path) -> Option<u32> {
std::fs::read_to_string(pid_path)
.ok()
.and_then(|pid| pid.trim().parse::<u32>().ok())
fn read_launcher_pid(run_dir: &Path) -> Option<u32> {
super::launcher::launcher_record_for_run(run_dir)
.map(|record| record.pid)
.or_else(|| {
std::fs::read_to_string(run_dir.join("run.pid"))
.ok()
.and_then(|pid| pid.trim().parse::<u32>().ok())
})
}
fn progress_file_is_empty(path: &Path) -> bool {
@ -390,15 +400,13 @@ fn determine_exit_code(conclusion_path: &Path, status_record: Option<RunStatusRe
}
}
fn kill_engine(pid_path: &Path) {
if let Ok(pid_str) = std::fs::read_to_string(pid_path) {
if let Ok(pid) = pid_str.trim().parse::<i32>() {
#[cfg(unix)]
unsafe {
libc::kill(pid, libc::SIGTERM);
}
let _ = pid;
fn kill_engine(run_dir: &Path) {
if let Some(pid) = read_launcher_pid(run_dir).map(|pid| pid as i32) {
#[cfg(unix)]
unsafe {
libc::kill(pid, libc::SIGTERM);
}
let _ = pid;
}
}

35
lib/crates/fabro-cli/src/commands/run/command.rs Normal file
View file

@ -0,0 +1,35 @@
use anyhow::Result;
use crate::args::{GlobalArgs, RunArgs};
pub async fn execute(mut args: RunArgs, _globals: &GlobalArgs) -> Result<()> {
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
let cli_defaults = fabro_config::cli::load_cli_config(None)?;
let cli_config: fabro_config::FabroSettings = cli_defaults.clone().try_into()?;
args.verbose = args.verbose || cli_config.verbose_enabled();
let quiet = args.detach;
let prevent_idle_sleep = cli_config.prevent_idle_sleep_enabled();
let (run_id, run_dir) = super::create::create_run(&args, cli_defaults, styles, quiet).await?;
#[cfg(feature = "sleep_inhibitor")]
let _sleep_guard = crate::sleep_inhibitor::guard(prevent_idle_sleep);
#[cfg(not(feature = "sleep_inhibitor"))]
let _ = prevent_idle_sleep;
let child = super::start::start_run(&run_dir, false)?;
if args.detach {
println!("{run_id}");
} else {
let exit_code = super::attach::attach_run(&run_dir, true, styles, Some(child)).await?;
super::output::print_run_summary(&run_dir, &run_id, styles);
if exit_code != std::process::ExitCode::SUCCESS {
std::process::exit(1);
}
}
Ok(())
}

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

@ -1,15 +1,12 @@
use std::path::PathBuf;
use crate::args::RunArgs;
use fabro_config::FabroConfig;
use fabro_config::{FabroConfig, FabroSettings};
use super::execute::{
cached_graph_path, default_run_dir, load_workflow_source_input, make_run_dir, parse_labels,
print_diagnostics_from_error, print_workflow_report_from_persisted, resolve_sandbox_provider,
write_run_config_snapshot,
};
use fabro_util::terminal::Styles;
use super::output::{print_diagnostics_from_error, print_workflow_report_from_persisted};
/// Create a workflow run: allocate run directory, persist RunRecord, return (run_id, run_dir).
///
/// This does NOT execute the workflow — it only prepares the run directory.
@ -24,79 +21,43 @@ pub async fn create_run(
.as_ref()
.ok_or_else(|| anyhow::anyhow!("--workflow is required"))?;
let cli_args_config = FabroConfig::try_from(args)?;
let source_input =
load_workflow_source_input(workflow_path, cli_args_config, cli_defaults, true)?;
let run_id = args
.run_id
.clone()
.unwrap_or_else(|| ulid::Ulid::new().to_string());
let run_dir = match args
.storage_dir
.clone()
.or_else(|| source_input.settings.storage_dir.clone())
{
Some(sd) => make_run_dir(&sd.join("runs"), &run_id, args.dry_run),
None => default_run_dir(&run_id, args.dry_run),
};
let settings: FabroSettings = fabro_workflows::operations::resolve_settings_for_path(
workflow_path,
cli_defaults,
cli_args_config,
true,
)?;
let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let base_branch = fabro_sandbox::daytona::detect_repo_info(&working_directory)
.ok()
.and_then(|(_, branch)| branch);
if !args.dry_run {
let _ = resolve_sandbox_provider(args.sandbox.map(Into::into), &source_input.settings)?;
}
let settings = source_input.settings.clone();
let persisted = match fabro_workflows::operations::create(
&source_input.raw_source,
fabro_workflows::operations::RunCreateOptions {
let created =
match fabro_workflows::operations::create(fabro_workflows::operations::CreateRequest {
workflow: fabro_workflows::operations::WorkflowInput::Path(workflow_path.clone()),
settings,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: source_input.workflow_slug.clone(),
labels: {
let mut labels = source_input.settings.labels.clone();
labels.extend(parse_labels(&args.label));
labels
},
run_dir: None,
run_id: args.run_id.clone(),
base_branch,
working_directory: Some(working_directory.clone()),
host_repo_path: Some(working_directory.to_string_lossy().to_string()),
goal_override: source_input.goal_override.clone(),
base_dir: Some(
source_input
.dot_path
.parent()
.unwrap_or(std::path::Path::new("."))
.to_path_buf(),
),
},
) {
Ok(persisted) => persisted,
Err(fabro_workflows::error::FabroError::ValidationFailed { diagnostics }) => {
if !quiet {
print_diagnostics_from_error(&diagnostics, styles);
}) {
Ok(created) => created,
Err(fabro_workflows::error::FabroError::ValidationFailed { diagnostics }) => {
if !quiet {
print_diagnostics_from_error(&diagnostics, styles);
}
anyhow::bail!("Validation failed");
}
anyhow::bail!("Validation failed");
}
Err(err) => return Err(err.into()),
};
Err(err) => return Err(err.into()),
};
if !quiet {
print_workflow_report_from_persisted(&persisted, &source_input.dot_path, styles);
print_workflow_report_from_persisted(
&created.persisted,
created.dot_path.as_deref(),
styles,
);
}
// Write CLI-owned debug and status artifacts after the run has been persisted.
tokio::fs::write(cached_graph_path(&run_dir), &source_input.raw_source).await?;
tokio::fs::write(run_dir.join("id.txt"), &run_id).await?;
std::fs::File::create(run_dir.join("progress.jsonl"))?;
fabro_workflows::run_status::write_run_status(
&run_dir,
fabro_workflows::run_status::RunStatus::Submitted,
None,
);
write_run_config_snapshot(&run_dir, source_input.workflow_toml_path.as_deref()).await?;
Ok((run_id, run_dir))
Ok((created.run_id, created.run_dir))
}

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

@ -1,21 +1,14 @@
use std::path::{Path, PathBuf};
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::{Context, Result};
use chrono::Utc;
use fabro_workflows::event::{RunNoticeLevel, WorkflowRunEvent};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::records::Conclusion;
use fabro_workflows::run_status::{self, RunStatus, StatusReason};
use serde::Serialize;
use anyhow::Result;
use fabro_interview::FileInterviewer;
use fabro_workflows::event::EventEmitter;
use crate::cli_config;
use crate::shared;
pub async fn execute(storage_dir: PathBuf, run_id: String, resume: bool) -> Result<()> {
let runs_base = fabro_workflows::run_lookup::runs_base(&storage_dir);
let run_dir = fabro_workflows::run_lookup::find_run_by_prefix(&runs_base, &run_id)?;
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
pub async fn execute(run_dir: PathBuf, launcher_path: PathBuf, resume: bool) -> Result<()> {
let cli_config = cli_config::load_cli_settings(None)?;
let github_app = shared::github::build_github_app_credentials(cli_config.app_id());
let git_author = fabro_workflows::git::GitAuthor::from_options(
@ -23,370 +16,24 @@ pub async fn execute(storage_dir: PathBuf, run_id: String, resume: bool) -> Resu
cli_config.git_author().and_then(|a| a.email.clone()),
);
let persisted = match fabro_workflows::pipeline::Persisted::load(&run_dir) {
Ok(persisted) => persisted,
Err(err) => {
let anyhow_err: anyhow::Error = anyhow::anyhow!("Failed to load persisted run: {err}");
let _ = persist_detached_failure(
&run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
&anyhow_err,
);
return Err(anyhow_err);
}
let _launcher_guard = scopeguard::guard(launcher_path.clone(), |path| {
super::launcher::remove_launcher_record(&path);
});
let services = fabro_workflows::operations::StartServices {
cancel_token: None,
emitter: Arc::new(EventEmitter::new()),
interviewer: Arc::new(FileInterviewer::new(run_dir.clone())),
git_author,
github_app,
registry_override: None,
};
if let Err(err) =
std::env::set_current_dir(&persisted.run_record().working_directory).map_err(|e| {
anyhow::anyhow!(
"Failed to set working directory to {}: {e}",
persisted.run_record().working_directory.display()
)
})
{
let _ =
persist_detached_failure(&run_dir, "bootstrap", StatusReason::BootstrapFailed, &err);
return Err(err);
}
let result = if resume {
super::execute::resume_from_record(
persisted,
run_dir.clone(),
styles,
github_app,
git_author,
)
.await
if resume {
let _ = fabro_workflows::operations::resume(&run_dir, services).await?;
} else {
super::execute::run_from_record(persisted, run_dir.clone(), styles, github_app, git_author)
.await
};
match result {
Ok(()) => Ok(()),
Err(err) => {
let _ = persist_detached_failure(
&run_dir,
"bootstrap",
StatusReason::SandboxInitFailed,
&err,
);
Err(err)
}
}
}
const POSTRUN_ABORTED_MESSAGE: &str = "Run aborted before post-run finalization completed.";
pub(crate) struct DetachedRunBootstrapGuard {
run_dir: PathBuf,
active: bool,
}
impl DetachedRunBootstrapGuard {
pub(crate) fn arm(run_dir: &Path) -> Result<Self> {
std::fs::write(run_dir.join("run.pid"), std::process::id().to_string())
.with_context(|| format!("Failed to write {}", run_dir.join("run.pid").display()))?;
run_status::write_run_status(
run_dir,
RunStatus::Starting,
Some(StatusReason::SandboxInitializing),
);
Ok(Self {
run_dir: run_dir.to_path_buf(),
active: true,
})
}
pub(crate) fn defuse(&mut self) {
self.active = false;
}
}
impl Drop for DetachedRunBootstrapGuard {
fn drop(&mut self) {
if self.active {
run_status::write_run_status(
&self.run_dir,
RunStatus::Failed,
Some(StatusReason::SandboxInitFailed),
);
}
}
}
pub(crate) struct DetachedRunCompletionGuard {
run_dir: PathBuf,
active: bool,
}
impl DetachedRunCompletionGuard {
pub(crate) fn arm(run_dir: &Path) -> Self {
Self {
run_dir: run_dir.to_path_buf(),
active: true,
}
}
pub(crate) fn defuse(&mut self) {
self.active = false;
}
}
impl Drop for DetachedRunCompletionGuard {
fn drop(&mut self) {
if !self.active {
return;
}
run_status::write_run_status(
&self.run_dir,
RunStatus::Failed,
Some(StatusReason::WorkflowError),
);
if !self.run_dir.join("conclusion.json").exists() {
let _ = write_failure_conclusion(
&self.run_dir,
POSTRUN_ABORTED_MESSAGE,
Some(StatusReason::WorkflowError),
);
}
if let Some(run_id) = load_run_id(&self.run_dir) {
let _ = append_progress_event(
&self.run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Error,
code: "postrun_aborted".to_string(),
message: POSTRUN_ABORTED_MESSAGE.to_string(),
},
);
}
}
}
pub(crate) fn load_run_id(run_dir: &Path) -> Option<String> {
fabro_workflows::records::RunRecord::load(run_dir)
.ok()
.map(|record| record.run_id)
.filter(|run_id| !run_id.trim().is_empty())
.or_else(|| {
std::fs::read_to_string(run_dir.join("id.txt"))
.ok()
.map(|run_id| run_id.trim().to_string())
.filter(|run_id| !run_id.is_empty())
})
}
pub(crate) fn append_progress_event(
run_dir: &Path,
run_id: &str,
event: &WorkflowRunEvent,
) -> Result<()> {
fabro_workflows::event::append_progress_event(run_dir, run_id, event)
}
pub(crate) fn append_run_notice(
run_dir: &Path,
level: RunNoticeLevel,
code: &'static str,
message: impl Into<String>,
) -> Result<()> {
let Some(run_id) = load_run_id(run_dir) else {
return Ok(());
};
append_progress_event(
run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level,
code: code.to_string(),
message: message.into(),
},
)
}
pub(crate) fn persist_detached_failure(
run_dir: &Path,
phase: &'static str,
reason: StatusReason,
error: &anyhow::Error,
) -> Result<()> {
#[derive(Serialize)]
struct DetachedFailureRecord<'a> {
timestamp: chrono::DateTime<Utc>,
phase: &'a str,
reason: StatusReason,
error: String,
}
let message = error.to_string();
let record = DetachedFailureRecord {
timestamp: Utc::now(),
phase,
reason,
error: message.clone(),
};
std::fs::write(
run_dir.join("detached_failure.json"),
serde_json::to_string_pretty(&record)?,
)
.with_context(|| {
format!(
"Failed to write {}",
run_dir.join("detached_failure.json").display()
)
})?;
write_failure_conclusion(run_dir, &message, Some(reason))?;
run_status::write_run_status(run_dir, RunStatus::Failed, Some(reason));
if let Some(run_id) = load_run_id(run_dir) {
append_progress_event(
run_dir,
&run_id,
&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Error,
code: format!("{phase}_failed"),
message,
},
)?;
let _ = fabro_workflows::operations::start(&run_dir, services).await?;
}
Ok(())
}
pub(crate) fn write_failure_conclusion(
run_dir: &Path,
message: &str,
_reason: Option<StatusReason>,
) -> Result<()> {
if run_dir.join("conclusion.json").exists() {
return Ok(());
}
let conclusion = Conclusion {
timestamp: Utc::now(),
status: StageStatus::Fail,
duration_ms: 0,
failure_reason: Some(message.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,
};
conclusion.save(&run_dir.join("conclusion.json"))?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use fabro_workflows::run_status::{RunStatusRecord, StatusReason};
#[test]
fn bootstrap_guard_marks_failed_on_drop() {
let dir = tempfile::tempdir().unwrap();
{
let _guard = DetachedRunBootstrapGuard::arm(dir.path()).unwrap();
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Starting);
assert_eq!(record.reason, Some(StatusReason::SandboxInitializing));
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::SandboxInitFailed));
}
#[test]
fn bootstrap_guard_defuse_leaves_starting_intact() {
let dir = tempfile::tempdir().unwrap();
{
let mut guard = DetachedRunBootstrapGuard::arm(dir.path()).unwrap();
guard.defuse();
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Starting);
assert_eq!(record.reason, Some(StatusReason::SandboxInitializing));
}
#[test]
fn completion_guard_marks_failed_on_drop() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-123").unwrap();
{
let _guard = DetachedRunCompletionGuard::arm(dir.path());
}
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::WorkflowError));
assert!(dir.path().join("conclusion.json").exists());
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("postrun_aborted"));
}
#[test]
fn load_run_id_falls_back_to_id_txt() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-xyz").unwrap();
assert_eq!(load_run_id(dir.path()).as_deref(), Some("run-xyz"));
}
#[test]
fn persist_detached_failure_writes_status_conclusion_and_progress() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-err").unwrap();
let err = anyhow::anyhow!("bootstrap exploded");
persist_detached_failure(dir.path(), "bootstrap", StatusReason::BootstrapFailed, &err)
.unwrap();
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(record.status, RunStatus::Failed);
assert_eq!(record.reason, Some(StatusReason::BootstrapFailed));
let conclusion =
fabro_workflows::records::Conclusion::load(&dir.path().join("conclusion.json"))
.unwrap();
assert_eq!(conclusion.status, StageStatus::Fail);
assert_eq!(
conclusion.failure_reason.as_deref(),
Some("bootstrap exploded")
);
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("bootstrap_failed"));
assert!(dir.path().join("detached_failure.json").exists());
}
#[test]
fn append_run_notice_writes_progress() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("id.txt"), "run-notice").unwrap();
append_run_notice(
dir.path(),
RunNoticeLevel::Warn,
"interview_unanswered",
"The run is still waiting for input.",
)
.unwrap();
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("\"event\":\"RunNotice\""));
assert!(progress.contains("\"code\":\"interview_unanswered\""));
}
}

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

File diff suppressed because it is too large Load diff

50
lib/crates/fabro-cli/src/commands/run/launcher.rs Normal file
View file

@ -0,0 +1,50 @@
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct LauncherRecord {
pub run_id: String,
pub run_dir: PathBuf,
pub pid: u32,
pub resume: bool,
pub log_path: PathBuf,
pub started_at: DateTime<Utc>,
}
pub(crate) fn launcher_dir(storage_dir: &Path) -> PathBuf {
storage_dir.join("launchers")
}
pub(crate) fn launcher_record_path(storage_dir: &Path, run_id: &str) -> PathBuf {
launcher_dir(storage_dir).join(format!("{run_id}.json"))
}
pub(crate) fn launcher_log_path(storage_dir: &Path, run_id: &str) -> PathBuf {
launcher_dir(storage_dir).join(format!("{run_id}.log"))
}
pub(crate) fn write_launcher_record(path: &Path, record: &LauncherRecord) -> Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(path, serde_json::to_string_pretty(record)?)
.with_context(|| format!("Failed to write launcher metadata to {}", path.display()))
}
pub(crate) fn read_launcher_record(path: &Path) -> Option<LauncherRecord> {
let content = std::fs::read_to_string(path).ok()?;
serde_json::from_str(&content).ok()
}
pub(crate) fn remove_launcher_record(path: &Path) {
let _ = std::fs::remove_file(path);
}
pub(crate) fn launcher_record_for_run(run_dir: &Path) -> Option<LauncherRecord> {
let record = fabro_workflows::records::RunRecord::load(run_dir).ok()?;
let path = launcher_record_path(&record.settings.storage_dir(), &record.run_id);
read_launcher_record(&path)
}

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

@ -3,13 +3,16 @@ use anyhow::Result;
use crate::args::{GlobalArgs, RunCommands};
pub(crate) mod attach;
pub(crate) mod command;
pub(crate) mod cp;
pub(crate) mod create;
pub(crate) mod detached;
pub(crate) mod diff;
pub(crate) mod execute;
pub(crate) mod fork;
pub(crate) mod launcher;
pub(crate) mod logs;
pub(crate) mod output;
pub(crate) mod overrides;
pub(crate) mod preview;
pub(crate) mod resume;
pub(crate) mod rewind;
@ -20,7 +23,7 @@ pub(crate) mod wait;
pub async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<()> {
match cmd {
RunCommands::Run(args) => execute::execute(args, globals).await,
RunCommands::Run(args) => command::execute(args, globals).await,
RunCommands::Create(args) => {
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
@ -50,10 +53,10 @@ pub async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<()> {
Ok(())
}
RunCommands::Detached {
storage_dir,
run_id,
run_dir,
launcher_path,
resume,
} => detached::execute(storage_dir, run_id, resume).await,
} => detached::execute(run_dir, launcher_path, resume).await,
RunCommands::Cp(args) => cp::cp_command(args).await,
RunCommands::Preview(args) => preview::run(args).await,
RunCommands::Ssh(args) => ssh::run(args).await,

222
lib/crates/fabro-cli/src/commands/run/output.rs Normal file
View file

@ -0,0 +1,222 @@
use std::path::Path;
use std::time::Duration;
use fabro_util::terminal::Styles;
use fabro_workflows::outcome::{format_cost, StageStatus};
use fabro_workflows::pipeline::{Persisted, Validated};
use fabro_workflows::records::Checkpoint;
use indicatif::HumanDuration;
use crate::shared::{format_tokens_human, print_diagnostics, relative_path, tilde_path};
fn print_workflow_header(
graph: &fabro_graphviz::graph::Graph,
diagnostics: &[fabro_validate::Diagnostic],
dot_path: Option<&Path>,
styles: &Styles,
) {
eprintln!(
"{} {} {}",
styles.bold.apply_to("Workflow:"),
graph.name,
styles.dim.apply_to(format!(
"({} nodes, {} edges)",
graph.nodes.len(),
graph.edges.len()
)),
);
let graph_path = dot_path
.map(relative_path)
.unwrap_or_else(|| "<inline>".to_string());
eprintln!(
"{} {}",
styles.dim.apply_to("Graph:"),
styles.dim.apply_to(graph_path),
);
let goal = graph.goal();
if !goal.is_empty() {
let stripped = fabro_util::text::strip_goal_decoration(goal);
eprintln!("{} {stripped}\n", styles.bold.apply_to("Goal:"));
}
print_diagnostics(diagnostics, styles);
}
pub(crate) fn print_workflow_report(
validated: &Validated,
dot_path: Option<&Path>,
styles: &Styles,
) {
print_workflow_header(validated.graph(), validated.diagnostics(), dot_path, styles);
}
pub(crate) fn print_workflow_report_from_persisted(
persisted: &Persisted,
dot_path: Option<&Path>,
styles: &Styles,
) {
print_workflow_header(persisted.graph(), persisted.diagnostics(), dot_path, styles);
}
pub(crate) fn print_diagnostics_from_error(
diagnostics: &[fabro_validate::Diagnostic],
styles: &Styles,
) {
print_diagnostics(diagnostics, styles);
}
pub(crate) fn print_run_summary(run_dir: &Path, run_id: &str, styles: &Styles) {
let conclusion_path = run_dir.join("conclusion.json");
let Ok(conclusion) = fabro_workflows::records::Conclusion::load(&conclusion_path) else {
return;
};
let pr_url = std::fs::read_to_string(run_dir.join("pull_request.json"))
.ok()
.and_then(|content| {
serde_json::from_str::<fabro_workflows::pull_request::PullRequestRecord>(&content)
.ok()
.map(|record| record.html_url)
});
print_run_conclusion(
&conclusion,
run_id,
run_dir,
None,
pr_url.as_deref(),
styles,
);
print_final_output(run_dir, styles);
print_assets(run_dir, styles);
}
pub(crate) fn print_run_conclusion(
conclusion: &fabro_workflows::records::Conclusion,
run_id: &str,
run_dir: &Path,
pushed_branch: Option<&str>,
pr_url: Option<&str>,
styles: &Styles,
) {
eprintln!("\n{}", styles.bold.apply_to("=== Run Result ==="));
eprintln!("{}", styles.dim.apply_to(format!("Run: {run_id}")));
let status_str = conclusion.status.to_string().to_uppercase();
let status_color = match conclusion.status {
StageStatus::Success | StageStatus::PartialSuccess => &styles.bold_green,
_ => &styles.bold_red,
};
eprintln!("Status: {}", status_color.apply_to(&status_str));
eprintln!(
"Duration: {}",
HumanDuration(Duration::from_millis(conclusion.duration_ms))
);
let total_tokens = conclusion.total_input_tokens + conclusion.total_output_tokens;
if total_tokens > 0 {
if conclusion.has_pricing {
if let Some(cost) = conclusion.total_cost {
if cost > 0.0 {
eprintln!(
"{}",
styles.dim.apply_to(format!(
"Cost: {} ({} toks)",
format_cost(cost),
format_tokens_human(total_tokens)
))
);
}
}
} else {
eprintln!(
"{}",
styles
.dim
.apply_to(format!("Toks: {}", format_tokens_human(total_tokens)))
);
}
if conclusion.total_cache_read_tokens > 0 {
eprintln!(
"{}",
styles.dim.apply_to(format!(
"Cache: {} read, {} write",
format_tokens_human(conclusion.total_cache_read_tokens),
format_tokens_human(conclusion.total_cache_write_tokens),
)),
);
}
if conclusion.total_reasoning_tokens > 0 {
eprintln!(
"{}",
styles.dim.apply_to(format!(
"Reasoning: {} tokens",
format_tokens_human(conclusion.total_reasoning_tokens),
)),
);
}
}
eprintln!(
"{}",
styles
.dim
.apply_to(format!("Run: {}", tilde_path(run_dir)))
);
if let Some(ref failure) = conclusion.failure_reason {
eprintln!("Failure: {}", styles.red.apply_to(failure));
}
if pushed_branch.is_some() || pr_url.is_some() {
eprintln!();
if let Some(branch) = pushed_branch {
eprintln!("{} {branch}", styles.bold.apply_to("Pushed branch:"));
}
if let Some(url) = pr_url {
eprintln!("{} {url}", styles.bold.apply_to("Pull request:"));
}
}
}
pub(crate) fn print_final_output(run_dir: &Path, styles: &Styles) {
let Ok(checkpoint) = Checkpoint::load(&run_dir.join("checkpoint.json")) else {
return;
};
for node_id in checkpoint.completed_nodes.iter().rev() {
let key = format!("response.{node_id}");
if let Some(serde_json::Value::String(response)) = checkpoint.context_values.get(&key) {
let text = response.trim();
if !text.is_empty() {
eprintln!("\n{}", styles.bold.apply_to("=== Output ==="));
eprintln!("{}", styles.render_markdown(text));
}
return;
}
}
}
pub(crate) fn print_assets(run_dir: &Path, styles: &Styles) {
let paths = fabro_workflows::asset_snapshot::collect_asset_paths(run_dir);
if paths.is_empty() {
return;
}
let home = dirs::home_dir();
eprintln!("\n{}", styles.bold.apply_to("=== Assets ==="));
for path in &paths {
let display = match &home {
Some(home_dir) => {
let home_str = home_dir.to_string_lossy();
if let Some(rest) = path.strip_prefix(home_str.as_ref()) {
format!("~{rest}")
} else {
path.clone()
}
}
None => path.clone(),
};
eprintln!("{display}");
}
}

90
lib/crates/fabro-cli/src/commands/run/overrides.rs Normal file
View file

@ -0,0 +1,90 @@
use std::collections::HashMap;
use anyhow::Result;
use fabro_config::{sandbox as sandbox_config, FabroConfig};
use fabro_sandbox::SandboxProvider;
use crate::args::{PreflightArgs, RunArgs};
fn sparse_flag(value: bool) -> Option<bool> {
value.then_some(true)
}
pub(crate) fn parse_labels(labels: &[String]) -> HashMap<String, String> {
labels
.iter()
.filter_map(|label| label.split_once('='))
.map(|(key, value)| (key.to_string(), value.to_string()))
.collect()
}
impl TryFrom<&RunArgs> for FabroConfig {
type Error = anyhow::Error;
fn try_from(args: &RunArgs) -> Result<Self, Self::Error> {
let llm = if args.model.is_some() || args.provider.is_some() {
Some(fabro_config::run::LlmConfig {
model: args.model.clone(),
provider: args.provider.clone(),
fallbacks: None,
})
} else {
None
};
let sandbox = if args.sandbox.is_some() || args.preserve_sandbox {
Some(sandbox_config::SandboxConfig {
provider: args
.sandbox
.map(Into::into)
.map(|provider: SandboxProvider| provider.to_string()),
preserve: sparse_flag(args.preserve_sandbox),
..Default::default()
})
} else {
None
};
Ok(Self {
goal: args.goal.clone(),
goal_file: args.goal_file.clone(),
llm,
sandbox,
verbose: sparse_flag(args.verbose),
dry_run: sparse_flag(args.dry_run),
auto_approve: sparse_flag(args.auto_approve),
no_retro: sparse_flag(args.no_retro),
storage_dir: args.storage_dir.clone(),
labels: parse_labels(&args.label),
..Default::default()
})
}
}
impl TryFrom<&PreflightArgs> for FabroConfig {
type Error = anyhow::Error;
fn try_from(args: &PreflightArgs) -> Result<Self, Self::Error> {
let llm = if args.model.is_some() || args.provider.is_some() {
Some(fabro_config::run::LlmConfig {
model: args.model.clone(),
provider: args.provider.clone(),
fallbacks: None,
})
} else {
None
};
let sandbox = args.sandbox.map(|sandbox| sandbox_config::SandboxConfig {
provider: Some(SandboxProvider::from(sandbox).to_string()),
..Default::default()
});
Ok(Self {
goal: args.goal.clone(),
goal_file: args.goal_file.clone(),
llm,
sandbox,
verbose: sparse_flag(args.verbose),
..Default::default()
})
}
}

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

@ -1,7 +1,6 @@
use anyhow::bail;
use fabro_util::terminal::Styles;
use fabro_workflows::records::{Checkpoint, RunRecord};
use fabro_workflows::run_status::{RunStatus, RunStatusRecord};
use fabro_workflows::records::RunRecord;
use crate::args::ResumeArgs;
@ -21,60 +20,17 @@ pub async fn resume_command(args: ResumeArgs, styles: &'static Styles) -> anyhow
}
let run_id = RunRecord::load(&run_dir)?.run_id;
// Guard against resuming a live run — must happen before checkpoint
// validation because the engine writes checkpoint.json with a plain
// fs::write, so a mid-write read would see a truncated file and
// report "corrupt" for a run that is simply still alive.
if is_pid_alive(&run_dir.join("run.pid")) {
if launcher_pid_alive(&run_dir) {
bail!("an engine process is still running for this run — cannot resume");
}
// Reject runs that completed successfully — only failed/interrupted
// runs should be resumed.
if let Ok(record) = RunStatusRecord::load(&run_dir.join("status.json")) {
if record.status == RunStatus::Succeeded {
bail!("run already succeeded — nothing to resume");
}
}
// Validate checkpoint is parseable before touching any state.
// A crash during the original run can leave a truncated file;
// we must not destroy the old conclusion/failure evidence and
// only then discover the checkpoint is corrupt.
let cp_path = run_dir.join("checkpoint.json");
Checkpoint::load(&cp_path).map_err(|e| {
if cp_path.exists() {
anyhow::anyhow!("checkpoint.json is corrupt — cannot resume: {e}")
} else {
anyhow::anyhow!("no checkpoint found — nothing to resume")
}
})?;
// Clean stale artifacts from previous execution
for name in &[
"conclusion.json",
"pull_request.json",
"detached_failure.json",
"interview_request.json",
"interview_response.json",
"interview_request.claim",
"detach.log",
"run.pid",
"progress.jsonl",
] {
let _ = std::fs::remove_file(run_dir.join(name));
}
// Reset status for re-execution
fabro_workflows::run_status::write_run_status(&run_dir, RunStatus::Submitted, None);
let child = super::start::start_run(&run_dir, true)?;
if args.detach {
println!("{run_id}");
} else {
let exit_code = super::attach::attach_run(&run_dir, true, styles, Some(child)).await?;
super::execute::print_run_summary(&run_dir, &run_id, styles);
super::output::print_run_summary(&run_dir, &run_id, styles);
if exit_code != std::process::ExitCode::SUCCESS {
std::process::exit(1);
}
@ -82,16 +38,16 @@ pub async fn resume_command(args: ResumeArgs, styles: &'static Styles) -> anyhow
Ok(())
}
/// Check whether a PID file contains a live process.
fn is_pid_alive(pid_path: &std::path::Path) -> bool {
let Ok(content) = std::fs::read_to_string(pid_path) else {
return false;
};
let Ok(pid) = content.trim().parse::<i32>() else {
return false;
};
// kill(pid, 0) checks liveness without sending a signal
unsafe { libc::kill(pid, 0) == 0 }
fn launcher_pid_alive(run_dir: &std::path::Path) -> bool {
super::launcher::launcher_record_for_run(run_dir)
.map(|record| process_alive(record.pid))
.or_else(|| {
std::fs::read_to_string(run_dir.join("run.pid"))
.ok()
.and_then(|pid| pid.trim().parse::<u32>().ok())
.map(process_alive)
})
.unwrap_or(false)
}
#[cfg(test)]
@ -99,23 +55,20 @@ mod tests {
use super::*;
#[test]
fn is_pid_alive_returns_false_for_missing_file() {
fn launcher_pid_alive_returns_false_for_missing_record() {
let dir = tempfile::tempdir().unwrap();
assert!(!is_pid_alive(&dir.path().join("run.pid")));
}
#[test]
fn is_pid_alive_returns_false_for_invalid_pid() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("run.pid"), "not-a-pid").unwrap();
assert!(!is_pid_alive(&dir.path().join("run.pid")));
}
#[test]
fn is_pid_alive_returns_true_for_current_process() {
let dir = tempfile::tempdir().unwrap();
let pid = std::process::id();
std::fs::write(dir.path().join("run.pid"), pid.to_string()).unwrap();
assert!(is_pid_alive(&dir.path().join("run.pid")));
assert!(!launcher_pid_alive(dir.path()));
}
}
fn process_alive(pid: u32) -> bool {
#[cfg(unix)]
{
unsafe { libc::kill(pid as i32, 0) == 0 }
}
#[cfg(not(unix))]
{
let _ = pid;
true
}
}

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

@ -196,6 +196,7 @@ pub struct ProgressUI {
working_directory: Option<String>,
}
#[allow(dead_code)]
impl ProgressUI {
pub fn new(is_tty: bool, verbose: bool) -> Self {
let renderer = if is_tty {
@ -1722,11 +1723,13 @@ impl ProgressUI {
/// Wraps a `ConsoleInterviewer` so that progress bars are hidden during
/// interactive prompts (avoids garbled output from concurrent writes).
#[allow(dead_code)]
pub struct ProgressAwareInterviewer {
inner: ConsoleInterviewer,
progress: Arc<Mutex<ProgressUI>>,
}
#[allow(dead_code)]
impl ProgressAwareInterviewer {
pub fn new(inner: ConsoleInterviewer, progress: Arc<Mutex<ProgressUI>>) -> Self {
Self { inner, progress }

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

@ -1,76 +1,37 @@
use std::path::Path;
use anyhow::{bail, Result};
use fabro_workflows::run_status::{RunStatus, StatusReason};
use anyhow::{anyhow, Result};
use chrono::Utc;
use super::detached::persist_detached_failure;
use super::launcher::{
launcher_log_path, launcher_record_path, write_launcher_record, LauncherRecord,
};
/// Spawn a detached engine process for the given run directory.
///
/// The engine process reads `run.json` from the run directory and executes the
/// workflow. Returns the child process handle (use `.id()` for the PID).
///
/// `storage_dir` is the base storage directory (e.g. `~/.fabro`). If `None`,
/// it is derived from the `run_dir` by stripping the `runs/<dir_name>` suffix.
pub fn start_run(run_dir: &Path, resume: bool) -> Result<std::process::Child> {
// Validate status is Submitted
let status_path = run_dir.join("status.json");
match fabro_workflows::run_status::RunStatusRecord::load(&status_path) {
Ok(record) if record.status != RunStatus::Submitted => {
bail!(
"Cannot start run: status is {:?}, expected Submitted",
record.status
);
}
_ => {} // No status file or Submitted — proceed
let record = fabro_workflows::records::RunRecord::load(run_dir)
.map_err(|e| anyhow!("Cannot start run: failed to load run.json: {e}"))?;
let storage_dir = record.settings.storage_dir();
let launcher_path = launcher_record_path(&storage_dir, &record.run_id);
let log_path = launcher_log_path(&storage_dir, &record.run_id);
if let Some(parent) = log_path.parent() {
std::fs::create_dir_all(parent)?;
}
// Validate run.json is loadable
fabro_workflows::records::RunRecord::load(run_dir)
.map_err(|e| anyhow::anyhow!("Cannot start run: failed to load run.json: {e}"))?;
// Write Starting status before spawning to prevent duplicate engines
fabro_workflows::run_status::write_run_status(run_dir, RunStatus::Starting, None);
let log_file = match std::fs::File::create(run_dir.join("detach.log")) {
Ok(file) => file,
Err(err) => {
let err = err.into();
let _ = persist_detached_failure(run_dir, "launch", StatusReason::LaunchFailed, &err);
return Err(err);
}
};
let exe = match std::env::current_exe() {
Ok(exe) => exe,
Err(err) => {
let err = err.into();
let _ = persist_detached_failure(run_dir, "launch", StatusReason::LaunchFailed, &err);
return Err(err);
}
};
let log_file = std::fs::File::create(&log_path)?;
let stdout_log = log_file.try_clone()?;
let exe = std::env::current_exe()?;
let mut cmd = std::process::Command::new(&exe);
let stdout_log = match log_file.try_clone() {
Ok(file) => file,
Err(err) => {
let err = err.into();
let _ = persist_detached_failure(run_dir, "launch", StatusReason::LaunchFailed, &err);
return Err(err);
}
};
// Derive storage_dir (grandparent of run_dir, e.g. ~/.fabro) and run_id
let storage_dir = run_dir
.parent() // runs/
.and_then(|p| p.parent()) // ~/.fabro/
.unwrap_or(run_dir);
let run_id = fabro_workflows::records::RunRecord::load(run_dir)
.map(|r| r.run_id)
.or_else(|_| std::fs::read_to_string(run_dir.join("id.txt")).map(|s| s.trim().to_string()))
.unwrap_or_default();
cmd.args(["__detached", "--storage-dir"])
.arg(storage_dir)
.args(["--run-id", &run_id]);
cmd.args(["__detached", "--run-dir"])
.arg(run_dir)
.args(["--launcher-path"])
.arg(&launcher_path);
if resume {
cmd.arg("--resume");
}
@ -78,7 +39,6 @@ pub fn start_run(run_dir: &Path, resume: bool) -> Result<std::process::Child> {
.stderr(log_file)
.stdin(std::process::Stdio::null());
// Detach from the controlling terminal on unix
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
@ -90,20 +50,20 @@ pub fn start_run(run_dir: &Path, resume: bool) -> Result<std::process::Child> {
}
}
let mut child = match cmd.spawn() {
Ok(child) => child,
Err(err) => {
let err = err.into();
let _ = persist_detached_failure(run_dir, "launch", StatusReason::LaunchFailed, &err);
return Err(err);
}
};
let mut child = cmd.spawn()?;
// Write PID file
if let Err(err) = std::fs::write(run_dir.join("run.pid"), child.id().to_string()) {
if let Err(err) = write_launcher_record(
&launcher_path,
&LauncherRecord {
run_id: record.run_id,
run_dir: run_dir.to_path_buf(),
pid: child.id(),
resume,
log_path,
started_at: Utc::now(),
},
) {
kill_child_best_effort(&mut child);
let err = err.into();
let _ = persist_detached_failure(run_dir, "launch", StatusReason::LaunchFailed, &err);
return Err(err);
}
@ -114,53 +74,3 @@ fn kill_child_best_effort(child: &mut std::process::Child) {
let _ = child.kill();
let _ = child.wait();
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::Utc;
use fabro_config::FabroSettings;
use fabro_graphviz::graph::Graph;
use fabro_workflows::records::RunRecord;
use fabro_workflows::run_status::{write_run_status, RunStatus, RunStatusRecord, StatusReason};
use std::collections::HashMap;
use std::path::PathBuf;
fn sample_record() -> RunRecord {
RunRecord {
run_id: "run-test123".to_string(),
created_at: Utc::now(),
settings: FabroSettings::default(),
graph: Graph {
name: "test".to_string(),
..Default::default()
},
workflow_slug: None,
working_directory: PathBuf::from("/tmp"),
host_repo_path: None,
base_branch: None,
labels: HashMap::new(),
}
}
#[test]
fn start_run_marks_failed_when_spawn_cannot_start_engine() {
let dir = tempfile::tempdir().unwrap();
write_run_status(dir.path(), RunStatus::Submitted, None);
sample_record().save(dir.path()).unwrap();
std::fs::create_dir(dir.path().join("detach.log")).unwrap();
let _ = start_run(dir.path(), false);
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(
record.status,
RunStatus::Failed,
"start_run should persist a terminal failure on launch errors"
);
assert_eq!(record.reason, Some(StatusReason::LaunchFailed));
assert!(dir.path().join("conclusion.json").exists());
let progress = std::fs::read_to_string(dir.path().join("progress.jsonl")).unwrap();
assert!(progress.contains("launch_failed"));
}
}

38
lib/crates/fabro-cli/src/main.rs
View file

@ -338,20 +338,23 @@ mod tests {
let cli = Cli::try_parse_from([
"fabro",
"__detached",
"--storage-dir",
"/tmp/fabro",
"--run-id",
"01ABC",
"--run-dir",
"/tmp/fabro/runs/01ABC",
"--launcher-path",
"/tmp/fabro/launchers/01ABC.json",
])
.expect("should parse");
match *cli.command {
Commands::RunCmd(RunCommands::Detached {
storage_dir,
run_id,
run_dir,
launcher_path,
resume,
}) => {
assert_eq!(storage_dir, std::path::PathBuf::from("/tmp/fabro"));
assert_eq!(run_id, "01ABC");
assert_eq!(run_dir, std::path::PathBuf::from("/tmp/fabro/runs/01ABC"));
assert_eq!(
launcher_path,
std::path::PathBuf::from("/tmp/fabro/launchers/01ABC.json")
);
assert!(!resume);
}
_ => panic!("unexpected command variant"),
@ -363,21 +366,24 @@ mod tests {
let cli = Cli::try_parse_from([
"fabro",
"__detached",
"--storage-dir",
"/tmp/fabro",
"--run-id",
"01ABC",
"--run-dir",
"/tmp/fabro/runs/01ABC",
"--launcher-path",
"/tmp/fabro/launchers/01ABC.json",
"--resume",
])
.expect("should parse");
match *cli.command {
Commands::RunCmd(RunCommands::Detached {
storage_dir,
run_id,
run_dir,
launcher_path,
resume,
}) => {
assert_eq!(storage_dir, std::path::PathBuf::from("/tmp/fabro"));
assert_eq!(run_id, "01ABC");
assert_eq!(run_dir, std::path::PathBuf::from("/tmp/fabro/runs/01ABC"));
assert_eq!(
launcher_path,
std::path::PathBuf::from("/tmp/fabro/launchers/01ABC.json")
);
assert!(resume);
}
_ => panic!("unexpected command variant"),

46
lib/crates/fabro-cli/tests/cli.rs
View file

@ -713,7 +713,8 @@ fn detach_creates_run_dir_with_detach_log() {
let ulid = ulid.trim();
assert!(!ulid.is_empty(), "should print a ULID");
// Run dir should have been created under storage_dir/runs/ with detach.log
// Run dir should have been created under storage_dir/runs/ and the launcher
// log should live under storage_dir/launchers/.
let runs_base = storage_dir.join("runs");
assert!(runs_base.exists(), "runs/ directory should exist");
let entries: Vec<_> = std::fs::read_dir(&runs_base)
@ -723,9 +724,13 @@ fn detach_creates_run_dir_with_detach_log() {
assert_eq!(entries.len(), 1, "should have exactly one run directory");
let run_dir = entries[0].path();
assert!(
run_dir.join("detach.log").exists(),
"detach.log should exist in run dir"
storage_dir
.join("launchers")
.join(format!("{ulid}.log"))
.exists(),
"launcher log should exist under storage_dir/launchers"
);
assert!(!run_dir.join("detach.log").exists());
}
// == Resume ===================================================================
@ -790,7 +795,7 @@ fn setup_run_dir(
let run_record = serde_json::json!({
"run_id": run_id,
"created_at": "2026-01-01T00:00:00Z",
"config": {
"settings": {
"goal": overrides.get("goal").and_then(|v| v.as_str()),
"llm": {
"model": get_str("model", "test-model"),
@ -1146,10 +1151,14 @@ digraph G {
let output = arc()
.args([
"__detached",
"--storage-dir",
storage_dir.to_str().unwrap(),
"--run-id",
"test-bug2",
"--run-dir",
run_dir.to_str().unwrap(),
"--launcher-path",
storage_dir
.join("launchers")
.join("test-bug2.json")
.to_str()
.unwrap(),
])
.env("NO_COLOR", "1")
.timeout(std::time::Duration::from_secs(15))
@ -1213,7 +1222,7 @@ digraph Test {
.success();
let before: serde_json::Value =
serde_json::from_slice(&inspect_before.get_output().stdout).unwrap();
let _run_dir = before[0]["run_dir"].as_str().unwrap().to_string();
let run_dir = before[0]["run_dir"].as_str().unwrap().to_string();
let start_time_before = before[0]["start_record"]["start_time"]
.as_str()
.unwrap()
@ -1228,10 +1237,14 @@ digraph Test {
.env("HOME", home.path())
.args([
"__detached",
"--storage-dir",
storage_dir.to_str().unwrap(),
"--run-id",
&run_id,
"--run-dir",
&run_dir,
"--launcher-path",
storage_dir
.join("launchers")
.join(format!("{run_id}.json"))
.to_str()
.unwrap(),
"--resume",
])
.timeout(std::time::Duration::from_secs(10))
@ -1384,13 +1397,6 @@ fn attach_closed_stdin_keeps_interview_pending() {
!run_dir.join("interview_request.claim").exists(),
"attach with closed stdin must release the claim so a later attach can answer"
);
let progress = std::fs::read_to_string(run_dir.join("progress.jsonl")).unwrap();
assert!(
progress.contains("\"event\":\"RunNotice\"")
&& progress.contains("\"code\":\"interview_unanswered\""),
"attach should emit a structured warning when the interview ends without an answer.\nprogress: {progress}"
);
}
// Bug 4: attach should respect the verbose flag from run.json.

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

@ -614,7 +614,7 @@ mod tests {
init_repo(dir.path());
let store = MetadataStore::new(dir.path(), &GitAuthor::default());
let run_record = br#"{"run_id":"RUN1","created_at":"2025-01-01T00:00:00Z","config":{},"graph":{"name":"test","nodes":{},"edges":[],"attrs":{}},"working_directory":"/tmp"}"#;
let run_record = br#"{"run_id":"RUN1","created_at":"2025-01-01T00:00:00Z","settings":{},"graph":{"name":"test","nodes":{},"edges":[],"attrs":{}},"working_directory":"/tmp"}"#;
store.init_run("RUN1", &[("run.json", run_record)]).unwrap();
let read_record = MetadataStore::read_run_record(dir.path(), "RUN1")
@ -794,7 +794,7 @@ mod tests {
init_repo(dir.path());
let store = MetadataStore::new(dir.path(), &GitAuthor::default());
let run_record = br#"{"run_id":"RUN5","created_at":"2025-01-01T00:00:00Z","config":{},"graph":{"name":"test","nodes":{},"edges":[],"attrs":{}},"working_directory":"/tmp"}"#;
let run_record = br#"{"run_id":"RUN5","created_at":"2025-01-01T00:00:00Z","settings":{},"graph":{"name":"test","nodes":{},"edges":[],"attrs":{}},"working_directory":"/tmp"}"#;
store.init_run("RUN5", &[("run.json", run_record)]).unwrap();
store

300
lib/crates/fabro-workflows/src/operations/create.rs
View file

@ -5,6 +5,7 @@ use chrono::{Local, Utc};
use fabro_config::FabroSettings;
use fabro_graphviz::graph::{AttrValue, Graph};
use fabro_model::{Catalog, Provider};
use fabro_sandbox::SandboxProvider;
use crate::error::FabroError;
use crate::pipeline::types::PersistOptions;
@ -12,6 +13,10 @@ use crate::pipeline::{self, Persisted, TransformOptions, Validated};
use crate::records::RunRecord;
use crate::transforms::{expand_vars, Transform};
use super::source::{resolve_workflow, ResolveWorkflowRequest, WorkflowInput};
const RUN_CONFIG_FILE: &str = "workflow.toml";
#[derive(Default)]
pub struct ValidateOptions {
pub base_dir: Option<PathBuf>,
@ -20,17 +25,52 @@ pub struct ValidateOptions {
pub goal_override: Option<String>,
}
pub struct RunCreateOptions {
#[derive(Clone, Debug)]
pub struct CreateRequest {
pub workflow: WorkflowInput,
pub settings: FabroSettings,
pub run_dir: Option<PathBuf>,
pub run_id: Option<String>,
pub workflow_slug: Option<String>,
pub labels: HashMap<String, String>,
pub base_branch: Option<String>,
pub working_directory: Option<PathBuf>,
pub host_repo_path: Option<String>,
pub goal_override: Option<String>,
pub base_dir: Option<PathBuf>,
pub base_branch: Option<String>,
}
impl Default for CreateRequest {
fn default() -> Self {
Self {
workflow: WorkflowInput::DotSource {
source: String::new(),
base_dir: None,
workflow_slug: None,
},
settings: FabroSettings::default(),
run_dir: None,
run_id: None,
host_repo_path: None,
base_branch: None,
}
}
}
#[derive(Debug)]
pub struct CreatedRun {
pub persisted: Persisted,
pub run_id: String,
pub run_dir: PathBuf,
pub dot_path: Option<PathBuf>,
}
struct PersistCreateOptions {
settings: FabroSettings,
run_dir: Option<PathBuf>,
run_id: Option<String>,
workflow_slug: Option<String>,
labels: HashMap<String, String>,
base_branch: Option<String>,
working_directory: Option<PathBuf>,
host_repo_path: Option<String>,
goal_override: Option<String>,
base_dir: Option<PathBuf>,
}
/// Parse, transform, and validate a DOT source string.
@ -61,8 +101,104 @@ pub fn validate_from_file(path: &Path) -> Result<Validated, FabroError> {
)
}
/// Parse, transform, validate, resolve settings, and persist a run.
pub fn create(dot_source: &str, options: RunCreateOptions) -> Result<Persisted, FabroError> {
/// Resolve workflow inputs, normalize settings, and persist a run directory.
pub fn create(request: CreateRequest) -> Result<CreatedRun, FabroError> {
let resolved = resolve_workflow(ResolveWorkflowRequest {
workflow: request.workflow,
settings: request.settings,
})
.map_err(|err| FabroError::Parse(err.to_string()))?;
if !resolved.settings.dry_run_enabled() {
validate_sandbox_provider(&resolved.settings)?;
}
let CreateRequest {
workflow: _,
settings: _,
run_dir,
run_id,
host_repo_path,
base_branch,
} = request;
let settings = resolved.settings.clone();
let run_id = run_id.unwrap_or_else(|| ulid::Ulid::new().to_string());
let storage_dir = settings.storage_dir();
let run_dir = run_dir.unwrap_or_else(|| {
make_run_dir(
&storage_dir.join("runs"),
&run_id,
settings.dry_run_enabled(),
)
});
let working_directory = resolved.working_directory.clone();
let host_repo_path =
host_repo_path.or_else(|| Some(working_directory.to_string_lossy().to_string()));
let base_branch = base_branch.or_else(|| {
fabro_sandbox::daytona::detect_repo_info(&working_directory)
.ok()
.and_then(|(_, branch)| branch)
});
let persisted = create_from_source(
&resolved.raw_source,
PersistCreateOptions {
settings,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: resolved.workflow_slug.clone(),
labels: resolved.settings.labels.clone(),
base_branch,
working_directory: Some(working_directory),
host_repo_path,
goal_override: resolved.goal_override.clone(),
base_dir: resolved.base_dir.clone(),
},
)?;
write_run_config_snapshot(&run_dir, resolved.workflow_toml_path.as_deref())?;
crate::run_status::write_run_status(&run_dir, crate::run_status::RunStatus::Submitted, None);
Ok(CreatedRun {
persisted,
run_id,
run_dir,
dot_path: resolved.dot_path,
})
}
fn validate_sandbox_provider(settings: &FabroSettings) -> Result<(), FabroError> {
if let Some(provider) = settings
.sandbox_settings()
.and_then(|sandbox| sandbox.provider.as_deref())
{
provider
.parse::<SandboxProvider>()
.map_err(|err| FabroError::Precondition(format!("Invalid sandbox provider: {err}")))?;
}
Ok(())
}
fn write_run_config_snapshot(
run_dir: &Path,
workflow_toml_path: Option<&Path>,
) -> Result<(), FabroError> {
if let Some(toml_path) = workflow_toml_path {
if toml_path.is_file() {
std::fs::copy(toml_path, run_dir.join(RUN_CONFIG_FILE))
.map_err(|err| FabroError::Io(err.to_string()))?;
}
}
Ok(())
}
fn create_from_source(
dot_source: &str,
options: PersistCreateOptions,
) -> Result<Persisted, FabroError> {
let validated = preprocess_and_validate(
dot_source,
options.base_dir.clone(),
@ -80,18 +216,6 @@ pub fn create(dot_source: &str, options: RunCreateOptions) -> Result<Persisted,
persist_validated(validated, options)
}
/// Read a DOT file, apply file inlining from its parent directory, then create.
pub fn create_from_file(
path: &Path,
mut options: RunCreateOptions,
) -> Result<Persisted, FabroError> {
let source = std::fs::read_to_string(path)
.map_err(|e| FabroError::Parse(format!("Failed to read {}: {e}", path.display())))?;
let base_dir = path.parent().unwrap_or(Path::new("."));
options.base_dir = Some(base_dir.to_path_buf());
create(&source, options)
}
fn preprocess_and_validate(
dot_source: &str,
base_dir: Option<PathBuf>,
@ -102,7 +226,6 @@ fn preprocess_and_validate(
let source = match settings.and_then(|resolved| resolved.vars.as_ref()) {
Some(vars) => {
let mut vars = vars.clone();
// `$goal` is resolved later from the graph goal after any goal override.
vars.insert("goal".to_string(), "$goal".to_string());
expand_vars(dot_source, &vars)
.map_err(|e| FabroError::Parse(format!("var expansion failed: {e}")))?
@ -134,9 +257,9 @@ fn apply_goal_override(graph: &mut Graph, goal_override: Option<&str>) {
fn persist_validated(
validated: Validated,
options: RunCreateOptions,
options: PersistCreateOptions,
) -> Result<Persisted, FabroError> {
let RunCreateOptions {
let PersistCreateOptions {
settings,
run_dir,
run_id,
@ -406,21 +529,15 @@ mod tests {
graph [goal="Test"]
work [label="Work"]
}"#;
let err = create(
dot,
RunCreateOptions {
settings: FabroSettings::default(),
run_dir: None,
run_id: None,
workflow_slug: None,
labels: HashMap::new(),
base_branch: None,
working_directory: None,
host_repo_path: None,
goal_override: None,
let err = create(CreateRequest {
workflow: WorkflowInput::DotSource {
source: dot.to_string(),
base_dir: None,
workflow_slug: None,
},
)
run_dir: Some(tempfile::tempdir().unwrap().path().join("run")),
..Default::default()
})
.unwrap_err();
match err {
@ -432,41 +549,42 @@ mod tests {
}
#[test]
fn create_persists_normalized_config() {
fn create_persists_normalized_config_and_initial_state() {
let dir = tempfile::tempdir().unwrap();
let persisted = create(
MINIMAL_DOT,
RunCreateOptions {
settings: FabroSettings {
llm: Some(fabro_config::run::LlmSettings {
model: Some("sonnet".to_string()),
provider: None,
fallbacks: None,
}),
pull_request: Some(fabro_config::run::PullRequestSettings {
enabled: false,
..Default::default()
}),
dry_run: Some(true),
..Default::default()
},
run_dir: Some(dir.path().join("run")),
run_id: Some("run-123".to_string()),
workflow_slug: Some("slug".to_string()),
labels: HashMap::from([("env".to_string(), "test".to_string())]),
base_branch: Some("main".to_string()),
working_directory: Some(dir.path().to_path_buf()),
host_repo_path: Some(dir.path().display().to_string()),
goal_override: Some("override goal".to_string()),
let created = create(CreateRequest {
workflow: WorkflowInput::DotSource {
source: MINIMAL_DOT.to_string(),
base_dir: None,
workflow_slug: Some("slug".to_string()),
},
)
settings: FabroSettings {
llm: Some(fabro_config::run::LlmSettings {
model: Some("sonnet".to_string()),
provider: None,
fallbacks: None,
}),
pull_request: Some(fabro_config::run::PullRequestSettings {
enabled: false,
..Default::default()
}),
goal: Some("override goal".to_string()),
dry_run: Some(true),
labels: HashMap::from([("env".to_string(), "test".to_string())]),
..Default::default()
},
run_dir: Some(dir.path().join("run")),
run_id: Some("run-123".to_string()),
host_repo_path: Some(dir.path().display().to_string()),
base_branch: Some("main".to_string()),
..Default::default()
})
.unwrap();
assert_eq!(persisted.run_record().run_id, "run-123");
assert_eq!(persisted.run_record().graph.goal(), "override goal");
assert_eq!(created.run_id, "run-123");
assert_eq!(created.persisted.run_record().graph.goal(), "override goal");
assert_eq!(
persisted
created
.persisted
.run_record()
.settings
.llm
@ -475,7 +593,8 @@ mod tests {
Some("claude-sonnet-4-6")
);
assert_eq!(
persisted
created
.persisted
.run_record()
.settings
.llm
@ -484,13 +603,54 @@ mod tests {
Some("anthropic")
);
assert_eq!(
persisted.run_record().settings.goal.as_deref(),
created.persisted.run_record().settings.goal.as_deref(),
Some("override goal")
);
assert!(persisted.run_record().settings.pull_request.is_none());
assert!(created
.persisted
.run_record()
.settings
.pull_request
.is_none());
assert_eq!(
persisted.run_record().workflow_slug.as_deref(),
created.persisted.run_record().workflow_slug.as_deref(),
Some("slug")
);
assert_eq!(
crate::run_status::RunStatusRecord::load(&created.run_dir.join("status.json"))
.unwrap()
.status,
crate::run_status::RunStatus::Submitted
);
assert!(!created.run_dir.join("id.txt").exists());
}
#[test]
fn create_copies_workflow_toml_snapshot() {
let dir = tempfile::tempdir().unwrap();
let workflow_dir = dir.path().join("workflow");
std::fs::create_dir_all(&workflow_dir).unwrap();
std::fs::write(workflow_dir.join("workflow.fabro"), MINIMAL_DOT).unwrap();
std::fs::write(
workflow_dir.join("workflow.toml"),
"version = 1\ngraph = \"workflow.fabro\"\n",
)
.unwrap();
let created = create(CreateRequest {
workflow: WorkflowInput::Path(workflow_dir.join("workflow.toml")),
settings: FabroSettings {
storage_dir: Some(dir.path().join("storage")),
dry_run: Some(true),
..Default::default()
},
..Default::default()
})
.unwrap();
assert_eq!(
std::fs::read_to_string(created.run_dir.join("workflow.toml")).unwrap(),
"version = 1\ngraph = \"workflow.fabro\"\n"
);
}
}

12
lib/crates/fabro-workflows/src/operations/mod.rs
View file

@ -1,19 +1,21 @@
mod create;
mod fork;
mod rewind;
mod source;
mod start;
pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec, SandboxSpec};
pub use create::{
create, create_from_file, default_run_dir, make_run_dir, validate, validate_from_file,
RunCreateOptions, ValidateOptions,
create, default_run_dir, make_run_dir, validate, validate_from_file, CreateRequest, CreatedRun,
ValidateOptions,
};
pub use fork::fork;
pub use rewind::{
build_timeline, find_run_id_by_prefix, load_parallel_map, parse_target, resolve_target, rewind,
TimelineEntry,
};
pub use start::{
resume, start, StartFinalizeOptions, StartOptions, StartPullRequestConfig, StartRetroOptions,
Started,
pub use source::{
resolve_settings_for_path, resolve_workflow, resolve_workflow_path, ResolveWorkflowRequest,
ResolvedWorkflow, WorkflowInput, WorkflowPathResolution,
};
pub use start::{resume, start, StartServices, Started};

7
lib/crates/fabro-workflows/src/operations/rewind.rs
View file

@ -327,9 +327,12 @@ pub fn load_parallel_map(store: &Store, run_id: &str) -> HashMap<String, String>
}
}
let graph_bytes = match bs.read_entry("graph.fabro") {
let graph_bytes = match bs.read_entry("workflow.fabro") {
Ok(Some(bytes)) => bytes,
_ => return HashMap::new(),
_ => match bs.read_entry("graph.fabro") {
Ok(Some(bytes)) => bytes,
_ => return HashMap::new(),
},
};
let dot_source = String::from_utf8_lossy(&graph_bytes);
let graph = match fabro_graphviz::parser::parse(&dot_source) {

235
lib/crates/fabro-workflows/src/operations/source.rs Normal file
View file

@ -0,0 +1,235 @@
use std::path::{Path, PathBuf};
use anyhow::Context;
use fabro_config::{project as project_config, run as run_config, FabroConfig, FabroSettings};
const RUN_GRAPH_FILE: &str = "workflow.fabro";
const LEGACY_RUN_GRAPH_FILE: &str = "graph.fabro";
#[derive(Clone, Debug)]
pub enum WorkflowInput {
Path(PathBuf),
DotSource {
source: String,
base_dir: Option<PathBuf>,
workflow_slug: Option<String>,
},
}
#[derive(Clone, Debug)]
pub struct WorkflowPathResolution {
pub resolved_workflow_path: PathBuf,
pub dot_path: PathBuf,
pub workflow_config: Option<FabroConfig>,
pub workflow_toml_path: Option<PathBuf>,
pub workflow_slug: Option<String>,
}
#[derive(Clone, Debug)]
pub struct ResolveWorkflowRequest {
pub workflow: WorkflowInput,
pub settings: FabroSettings,
}
#[derive(Clone, Debug)]
pub struct ResolvedWorkflow {
pub raw_source: String,
pub settings: FabroSettings,
pub workflow_slug: Option<String>,
pub workflow_toml_path: Option<PathBuf>,
pub dot_path: Option<PathBuf>,
pub resolved_workflow_path: Option<PathBuf>,
pub base_dir: Option<PathBuf>,
pub goal_override: Option<String>,
pub working_directory: PathBuf,
}
fn resolve_goal_file(
goal_file: Option<&Path>,
working_directory: &Path,
) -> anyhow::Result<Option<String>> {
let Some(goal_file) = goal_file else {
return Ok(None);
};
let expanded = fabro_util::path::expand_tilde(goal_file);
let goal_path = if expanded.is_absolute() {
expanded
} else {
working_directory.join(expanded)
};
let content = std::fs::read_to_string(&goal_path)
.with_context(|| format!("failed to read goal file: {}", goal_path.display()))?;
tracing::debug!(path = %goal_path.display(), "Goal loaded from file");
Ok(Some(content))
}
fn resolve_working_directory(settings: &FabroSettings, caller_cwd: &Path) -> PathBuf {
let Some(work_dir) = settings.work_dir.as_deref() else {
return caller_cwd.to_path_buf();
};
let path = PathBuf::from(work_dir);
if path.is_absolute() {
path
} else {
caller_cwd.join(path)
}
}
pub fn workflow_slug_from_path(workflow_path: &Path) -> Option<String> {
let file_name = workflow_path.file_name()?.to_string_lossy();
if workflow_path.extension().is_none() {
return Some(file_name.into_owned());
}
let file_stem = workflow_path.file_stem()?.to_string_lossy();
if file_stem == "workflow" {
return workflow_path
.parent()
.and_then(|p| p.file_name())
.map(|n| n.to_string_lossy().into_owned())
.or_else(|| Some(file_stem.into_owned()));
}
Some(file_stem.into_owned())
}
pub fn resolve_workflow_path(workflow_path: &Path) -> anyhow::Result<WorkflowPathResolution> {
let path = project_config::resolve_workflow_arg(workflow_path)?;
let workflow_slug = workflow_slug_from_path(&path);
if path.extension().is_some_and(|ext| ext == "toml") {
match run_config::load_run_config(&path) {
Ok(cfg) => {
let dot_path = run_config::resolve_graph_path(
&path,
cfg.graph.as_deref().unwrap_or(RUN_GRAPH_FILE),
);
Ok(WorkflowPathResolution {
resolved_workflow_path: path.clone(),
dot_path,
workflow_config: Some(cfg),
workflow_toml_path: Some(path),
workflow_slug,
})
}
Err(_)
if !path.exists() && path.starts_with(crate::run_lookup::default_runs_base()) =>
{
let canonical = path.with_file_name(RUN_GRAPH_FILE);
let legacy = path.with_file_name(LEGACY_RUN_GRAPH_FILE);
let dot_path = if canonical.exists() || !legacy.exists() {
canonical
} else {
legacy
};
Ok(WorkflowPathResolution {
resolved_workflow_path: path,
dot_path,
workflow_config: None,
workflow_toml_path: None,
workflow_slug,
})
}
Err(err) => Err(err),
}
} else {
Ok(WorkflowPathResolution {
resolved_workflow_path: path.clone(),
dot_path: path,
workflow_config: None,
workflow_toml_path: None,
workflow_slug,
})
}
}
pub fn resolve_settings_for_path(
workflow_path: &Path,
defaults: FabroConfig,
overrides: FabroConfig,
apply_project_config: bool,
) -> anyhow::Result<FabroSettings> {
let resolution = resolve_workflow_path(workflow_path)?;
if resolution.workflow_config.is_none() && !resolution.resolved_workflow_path.is_file() {
anyhow::bail!(
"Workflow not found: {}",
resolution.resolved_workflow_path.display()
);
}
let project_config = if apply_project_config {
project_config::discover_project_config(
resolution
.resolved_workflow_path
.parent()
.unwrap_or_else(|| Path::new(".")),
)?
.map(|(_, config)| config)
.unwrap_or_default()
} else {
FabroConfig::default()
};
overrides
.combine(resolution.workflow_config.unwrap_or_default())
.combine(project_config)
.combine(defaults)
.try_into()
}
pub fn resolve_workflow(request: ResolveWorkflowRequest) -> anyhow::Result<ResolvedWorkflow> {
let caller_cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
match request.workflow {
WorkflowInput::Path(workflow_path) => {
let resolution = resolve_workflow_path(&workflow_path)?;
let settings = request.settings;
let working_directory = resolve_working_directory(&settings, &caller_cwd);
let raw_source = std::fs::read_to_string(&resolution.dot_path)
.with_context(|| format!("Failed to read {}", resolution.dot_path.display()))?;
let goal_override = settings.goal.clone().or(resolve_goal_file(
settings.goal_file.as_deref(),
&working_directory,
)?);
Ok(ResolvedWorkflow {
raw_source,
settings,
workflow_slug: resolution.workflow_slug,
workflow_toml_path: resolution.workflow_toml_path,
dot_path: Some(resolution.dot_path.clone()),
resolved_workflow_path: Some(resolution.resolved_workflow_path),
base_dir: Some(
resolution
.dot_path
.parent()
.unwrap_or_else(|| Path::new("."))
.to_path_buf(),
),
goal_override,
working_directory,
})
}
WorkflowInput::DotSource {
source,
base_dir,
workflow_slug,
} => {
let settings = request.settings;
let working_directory = resolve_working_directory(&settings, &caller_cwd);
let goal_override = settings.goal.clone().or(resolve_goal_file(
settings.goal_file.as_deref(),
&working_directory,
)?);
Ok(ResolvedWorkflow {
raw_source: source,
settings,
workflow_slug,
workflow_toml_path: None,
dot_path: None,
resolved_workflow_path: None,
base_dir,
goal_override,
working_directory,
})
}
}
}

842
lib/crates/fabro-workflows/src/operations/start.rs
View file

File diff suppressed because it is too large Load diff

15
lib/crates/fabro-workflows/src/pipeline/persist.rs
View file

@ -4,9 +4,10 @@ use crate::error::FabroError;
use super::types::{PersistOptions, Persisted, Validated};
const GRAPH_FILE_NAME: &str = "graph.fabro";
const GRAPH_FILE_NAME: &str = "workflow.fabro";
const LEGACY_GRAPH_FILE_NAME: &str = "graph.fabro";
/// PERSIST phase: create run directory, write graph.fabro and run.json to disk.
/// PERSIST phase: create run directory, write workflow.fabro and run.json to disk.
///
/// Overwrites `run_record.graph` with the validated graph before saving.
pub fn persist(validated: Validated, mut options: PersistOptions) -> Result<Persisted, FabroError> {
@ -30,14 +31,20 @@ pub fn persist(validated: Validated, mut options: PersistOptions) -> Result<Pers
/// Load a previously persisted run from disk.
///
/// `run.json` is authoritative for graph + config; `graph.fabro` provides the
/// `run.json` is authoritative for graph + config; `workflow.fabro` provides the
/// original DOT source string when present.
pub(crate) fn load(run_dir: &Path) -> Result<Persisted, FabroError> {
let run_record = crate::records::RunRecord::load(run_dir)?;
let graph = run_record.graph.clone();
let source = match std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)) {
Ok(source) => source,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
match std::fs::read_to_string(run_dir.join(LEGACY_GRAPH_FILE_NAME)) {
Ok(source) => source,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
Err(err) => return Err(err.into()),
}
}
Err(err) => return Err(err.into()),
};

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

@ -108,7 +108,7 @@ pub struct PersistOptions {
pub run_record: RunRecord,
}
/// Output of the PERSIST phase. Run directory created, run.json and graph.fabro written.
/// Output of the PERSIST phase. Run directory created, run.json and workflow.fabro written.
#[derive(Debug)]
#[non_exhaustive]
pub struct Persisted {