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

refactor(operations): make create own the full create lifecycle

Browse source 
operations::create now handles the full pipeline: var expansion, parse,
goal override, transform, validate, config normalization, and persist.
This eliminates duplicated RunRecord construction and pipeline::persist
calls across CLI and API callers.

Key changes:
- Rename operations::create → validate, CreateOptions → ValidateOptions
- New operations::create returns Persisted, with RunCreateSettings
- Add ValidationFailed error variant with diagnostics
- Move normalize_config, default_run_dir into operations
- Delete prepare_workflow, PreparedWorkflow, CliFlags from CLI
- Make pipeline::persist and types module pub(crate)
- API catches both Parse and ValidationFailed as 400
- CLI prints diagnostics directly from error (no re-validation)
- ExecutionOverrides struct replaces 9-param function

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-25 23:09:16 -04:00
parent 5a8eff0d63
commit 89ad849208
14 changed files with 972 additions and 737 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -24,8 +24,8 @@ use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer};
use fabro_workflows::context::Context;
use fabro_workflows::event::{EventEmitter, WorkflowRunEvent};
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::operations::{self, CreateOptions};
use fabro_workflows::pipeline::{self, InitOptions, PersistOptions, Persisted};
use fabro_workflows::operations::{self, RunCreateSettings};
use fabro_workflows::pipeline::{self, InitOptions, Persisted};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::run_settings::LifecycleConfig;
use fabro_workflows::run_settings::RunSettings;
@ -475,58 +475,60 @@ async fn start_run(
State(state): State<Arc<AppState>>,
Json(req): Json<StartRunRequest>,
) -> Response {
// Parse and persist the DOT source.
let validated = match operations::create(&req.dot_source, CreateOptions::default()) {
Ok(validated) => {
if let Err(e) = validated.raise_on_errors() {
return ApiError::bad_request(e.to_string()).into_response();
}
validated
}
Err(e) => {
return ApiError::bad_request(e.to_string()).into_response();
}
};
let run_id = ulid::Ulid::new().to_string();
info!(run_id = %run_id, "Run queued");
let created_at = chrono::Utc::now();
let run_dir = std::env::temp_dir().join(format!("fabro-{}", uuid::Uuid::new_v4()));
let run_record = fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at,
config: fabro_config::config::FabroConfig {
dry_run: Some(state.dry_run),
hooks: state.hooks.clone(),
sandbox: Some(fabro_config::sandbox::SandboxConfig {
provider: Some("local".to_string()),
..Default::default()
}),
let config = fabro_config::config::FabroConfig {
dry_run: Some(state.dry_run),
hooks: state.hooks.clone(),
sandbox: Some(fabro_config::sandbox::SandboxConfig {
provider: Some("local".to_string()),
..Default::default()
},
graph: validated.graph().clone(),
workflow_slug: None,
working_directory: std::env::current_dir()
.unwrap_or_else(|_| std::path::PathBuf::from(".")),
host_repo_path: None,
base_branch: None,
labels: std::collections::HashMap::new(),
}),
..Default::default()
};
if let Err(err) = fabro_workflows::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record,
let persisted = match operations::create(
&req.dot_source,
RunCreateSettings {
config,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: None,
labels: std::collections::HashMap::new(),
base_branch: None,
working_directory: Some(
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
),
host_repo_path: None,
goal_override: None,
base_dir: None,
},
) {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to persist run state: {err}"),
)
.into_response();
}
Ok(persisted) => persisted,
Err(ref err @ fabro_workflows::error::FabroError::ValidationFailed { ref diagnostics }) => {
let message = if diagnostics.is_empty() {
err.to_string()
} else {
diagnostics
.iter()
.map(|diagnostic| diagnostic.message.as_str())
.collect::<Vec<_>>()
.join("; ")
};
return ApiError::bad_request(message).into_response();
}
Err(err @ fabro_workflows::error::FabroError::Parse(_)) => {
return ApiError::bad_request(err.to_string()).into_response();
}
Err(err) => {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to persist run state: {err}"),
)
.into_response();
}
};
let created_at = persisted.run_record().created_at;
{
let mut runs = state.runs.lock().expect("runs lock poisoned");

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

@ -1,70 +1,15 @@
use std::path::PathBuf;
use chrono::Utc;
use fabro_config::config::FabroConfig;
use fabro_sandbox::SandboxProvider;
use fabro_workflows::pipeline::PersistOptions;
use fabro_workflows::records::RunRecord;
use super::run::{
cached_graph_path, default_run_dir, prepare_workflow, write_run_config_snapshot, RunArgs,
apply_execution_overrides, cached_graph_path, default_run_dir, load_workflow_source_input,
parse_labels, print_diagnostics_from_error, print_workflow_report_from_persisted,
resolve_sandbox_provider, write_run_config_snapshot, ExecutionOverrides, RunArgs,
};
use fabro_util::terminal::Styles;
/// CLI flag overrides for config normalization.
#[derive(Debug, Clone, Copy)]
pub(crate) struct CliFlags {
pub dry_run: bool,
pub auto_approve: bool,
pub no_retro: bool,
pub verbose: bool,
pub preserve_sandbox: bool,
}
impl From<&RunArgs> for CliFlags {
fn from(args: &RunArgs) -> Self {
Self {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
}
}
}
/// Build a normalized FabroConfig that captures the full execution intent.
///
/// Folds resolved model/provider/sandbox/goal and CLI flag overrides back into
/// a single FabroConfig so the RunRecord is self-contained.
pub(crate) fn normalize_config(
run_cfg: Option<&FabroConfig>,
run_defaults: &FabroConfig,
model: &str,
provider: Option<&str>,
sandbox_provider: SandboxProvider,
graph: &fabro_graphviz::graph::Graph,
flags: CliFlags,
) -> FabroConfig {
let mut config = run_cfg.cloned().unwrap_or_else(|| run_defaults.clone());
// Ensure resolved values are written back into config
config.llm.get_or_insert_default().model = Some(model.to_string());
config.llm.get_or_insert_default().provider = provider.map(String::from);
config.sandbox.get_or_insert_default().provider = Some(sandbox_provider.to_string());
let goal = graph.goal().to_string();
config.goal = if goal.is_empty() { None } else { Some(goal) };
// CLI flag overrides
config.dry_run = Some(flags.dry_run);
config.auto_approve = Some(flags.auto_approve);
config.no_retro = Some(flags.no_retro);
config.verbose = Some(flags.verbose);
if flags.preserve_sandbox {
config.sandbox.get_or_insert_default().preserve = Some(true);
}
config.pull_request = config.pull_request.take().filter(|p| p.enabled);
config
}
/// 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.
@ -74,10 +19,7 @@ pub async fn create_run(
styles: &Styles,
quiet: bool,
) -> anyhow::Result<(String, PathBuf)> {
let prep = prepare_workflow(args, run_defaults, styles, quiet)?;
let dot_source = prep.source().to_string();
let graph = prep.graph().clone();
let source_input = load_workflow_source_input(args, run_defaults, true)?;
let run_id = args
.run_id
.clone()
@ -86,50 +28,72 @@ pub async fn create_run(
.run_dir
.clone()
.unwrap_or_else(|| default_run_dir(&run_id, args.dry_run));
// Build normalized config and RunRecord
let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let labels: std::collections::HashMap<String, String> = args
.label
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
let config = normalize_config(
prep.run_cfg.as_ref(),
&prep.run_defaults,
&prep.model,
prep.provider.as_deref(),
prep.sandbox_provider,
&graph,
CliFlags::from(args),
);
let base_branch = fabro_sandbox::daytona::detect_repo_info(&working_directory)
.ok()
.and_then(|(_, branch)| branch);
let record = RunRecord {
run_id: run_id.clone(),
created_at: Utc::now(),
config,
graph,
workflow_slug: prep.workflow_slug.clone(),
working_directory: working_directory.clone(),
host_repo_path: Some(working_directory.to_string_lossy().to_string()),
base_branch,
labels,
let sandbox_provider = if args.dry_run {
SandboxProvider::Local
} else {
resolve_sandbox_provider(
args.sandbox.map(Into::into),
Some(&source_input.config),
&source_input.run_defaults,
)?
};
fabro_workflows::pipeline::persist(
prep.validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record: record,
let mut config = source_input.config.clone();
apply_execution_overrides(
&mut config,
&ExecutionOverrides {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
model: args.model.as_deref(),
provider: args.provider.as_deref(),
sandbox_provider,
},
)?;
);
let persisted = match fabro_workflows::operations::create(
&source_input.raw_source,
fabro_workflows::operations::RunCreateSettings {
config,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: source_input.workflow_slug.clone(),
labels: parse_labels(&args.label),
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);
}
anyhow::bail!("Validation failed");
}
Err(err) => return Err(err.into()),
};
if !quiet {
print_workflow_report_from_persisted(&persisted, &source_input.dot_path, styles);
}
// Write CLI-owned debug and status artifacts after the run has been persisted.
tokio::fs::write(cached_graph_path(&run_dir), &dot_source).await?;
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(
@ -137,7 +101,7 @@ pub async fn create_run(
fabro_workflows::run_status::RunStatus::Submitted,
None,
);
write_run_config_snapshot(&run_dir, prep.workflow_toml_path.as_deref()).await?;
write_run_config_snapshot(&run_dir, source_input.workflow_toml_path.as_deref()).await?;
Ok((run_id, run_dir))
}

2
lib/crates/fabro-cli/src/commands/graph.rs
View file

@ -57,7 +57,7 @@ static RANKDIR_RE: LazyLock<regex::Regex> =
pub fn run(args: &GraphArgs, styles: &Styles) -> anyhow::Result<()> {
let (dot_path, _cfg) = fabro_config::project::resolve_workflow(&args.workflow)?;
let validated = fabro_workflows::operations::create_from_file(&dot_path)?;
let validated = fabro_workflows::operations::validate_from_file(&dot_path)?;
let diagnostics = validated.diagnostics();
print_diagnostics(diagnostics, styles);

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

@ -14,11 +14,12 @@ use fabro_util::terminal::Styles;
use fabro_workflows::event::{EventEmitter, RunNoticeLevel};
use fabro_workflows::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter};
use fabro_workflows::operations::{
create_from_graph, start, StartFinalizeConfig, StartOptions, StartRetroConfig,
create_from_graph, start, RunCreateSettings, StartFinalizeConfig, StartOptions,
StartRetroConfig,
};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::pipeline::{
build_conclusion, classify_engine_result, persist_terminal_outcome, PersistOptions, Persisted,
build_conclusion, classify_engine_result, persist_terminal_outcome, Persisted,
};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::records::RunRecord;
@ -26,12 +27,13 @@ use fabro_workflows::run_settings::{GitCheckpointSettings, LifecycleConfig, RunS
use super::detached_support::{DetachedRunBootstrapGuard, DetachedRunCompletionGuard};
use super::run::{
build_event_envelope, cached_graph_path, default_run_dir, emit_run_notice,
local_sandbox_with_callback, mint_github_token, prepare_workflow_with_project_config,
print_assets, print_final_output, print_retro_result, print_run_conclusion,
apply_execution_overrides, build_event_envelope, cached_graph_path, default_run_dir,
emit_run_notice, load_workflow_source_input, local_sandbox_with_callback, mint_github_token,
parse_labels, print_assets, print_diagnostics_from_error, print_final_output,
print_retro_result, print_run_conclusion, print_workflow_report_from_persisted,
resolve_daytona_config, resolve_fallback_chain, resolve_model_provider,
resolve_ssh_clone_params, resolve_ssh_config, write_run_config_snapshot, CliSandboxProvider,
RunArgs,
resolve_sandbox_provider, resolve_ssh_clone_params, resolve_ssh_config,
write_run_config_snapshot, CliSandboxProvider, ExecutionOverrides, RunArgs,
};
use fabro_config::project as project_config;
use fabro_config::run as run_config;
@ -197,76 +199,76 @@ async fn prepare_from_checkpoint(
.ok_or_else(|| anyhow::anyhow!("--workflow is required when using --checkpoint"))?;
let checkpoint = Checkpoint::load(checkpoint_path)?;
let prepared = prepare_workflow_with_project_config(
let source_input = load_workflow_source_input(
&resume_as_run_args(args, workflow_path.clone()),
run_defaults.clone(),
styles,
true,
false,
)?;
let source = prepared.raw_source.clone();
let validated = prepared.validated;
let graph = validated.graph().clone();
let run_cfg = prepared.run_cfg;
let sandbox_provider = prepared.sandbox_provider;
let workflow_slug = prepared.workflow_slug;
let prepared_model = prepared.model;
let prepared_provider = prepared.provider;
let prepared_run_defaults = prepared.run_defaults;
let workflow_toml_path = prepared.workflow_toml_path;
eprintln!(
"{} {} from checkpoint {}",
styles.bold.apply_to("Resuming workflow:"),
graph.name,
styles.dim.apply_to(checkpoint_path.display()),
);
let run_id = ulid::Ulid::new().to_string();
let run_dir = args
.run_dir
.clone()
.unwrap_or_else(|| default_run_dir(&run_id, args.dry_run));
let labels = args
.label
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let cli_flags = super::create::CliFlags {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
let sandbox_provider = if args.dry_run {
SandboxProvider::Local
} else {
resolve_sandbox_provider(
args.sandbox.map(Into::into),
Some(&source_input.config),
run_defaults,
)?
};
let normalized = super::create::normalize_config(
run_cfg.as_ref(),
&prepared_run_defaults,
&prepared_model,
prepared_provider.as_deref(),
sandbox_provider,
&graph,
cli_flags,
);
let persisted = fabro_workflows::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record: fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized,
graph: graph.clone(),
workflow_slug: workflow_slug.clone(),
working_directory: working_directory.clone(),
host_repo_path: Some(working_directory.to_string_lossy().to_string()),
base_branch: None,
labels,
},
let mut config = source_input.config.clone();
apply_execution_overrides(
&mut config,
&ExecutionOverrides {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
model: args.model.as_deref(),
provider: args.provider.as_deref(),
sandbox_provider,
},
)?;
);
let persisted = match fabro_workflows::operations::create(
&source_input.raw_source,
RunCreateSettings {
config,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: source_input.workflow_slug.clone(),
labels: parse_labels(&args.label),
base_branch: None,
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 }) => {
print_diagnostics_from_error(&diagnostics, styles);
bail!("Validation failed");
}
Err(err) => return Err(err.into()),
};
print_workflow_report_from_persisted(&persisted, &source_input.dot_path, styles);
eprintln!(
"{} {} from checkpoint {}",
styles.bold.apply_to("Resuming workflow:"),
persisted.graph().name,
styles.dim.apply_to(checkpoint_path.display()),
);
let run_cfg: Option<FabroConfig> = Some(persisted.run_record().config.clone());
let settings_config = persisted.run_record().config.clone();
@ -274,8 +276,8 @@ async fn prepare_from_checkpoint(
fabro_util::run_log::activate(&run_dir.join("cli.log"))
.context("Failed to activate per-run log")?;
let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?;
tokio::fs::write(cached_graph_path(&run_dir), &source).await?;
write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?;
tokio::fs::write(cached_graph_path(&run_dir), &source_input.raw_source).await?;
write_run_config_snapshot(&run_dir, source_input.workflow_toml_path.as_deref()).await?;
let original_cwd = std::env::current_dir()?;
let emitter = Arc::new(EventEmitter::new());
@ -572,65 +574,68 @@ async fn prepare_from_branch(
};
tokio::fs::create_dir_all(&run_dir).await?;
let run_dir = tokio::fs::canonicalize(&run_dir).await.unwrap_or(run_dir);
let cli_flags = super::create::CliFlags {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
};
let (persisted, _run_cfg, mut sandbox_provider, graph_source) =
if let Some(ref workflow_path) = args.workflow {
let prepared = prepare_workflow_with_project_config(
let source_input = load_workflow_source_input(
&resume_as_run_args(args, workflow_path.clone()),
run_defaults.clone(),
styles,
true,
false,
)?;
let graph = prepared.validated.graph().clone();
let (model_str, provider_str) = resolve_model_provider(
args.model.as_deref(),
args.provider.as_deref(),
prepared.run_cfg.as_ref(),
run_defaults,
&graph,
);
let normalized = super::create::normalize_config(
prepared.run_cfg.as_ref(),
run_defaults,
&model_str,
provider_str.as_deref(),
prepared.sandbox_provider,
&graph,
cli_flags,
);
let persisted = fabro_workflows::pipeline::persist(
prepared.validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record: fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized,
graph,
workflow_slug: prepared.workflow_slug,
working_directory: resume_repo_path.clone(),
host_repo_path: Some(resume_repo_path.to_string_lossy().to_string()),
base_branch: detected_base_branch.clone(),
labels: args
.label
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
},
let sandbox_provider = if args.dry_run {
SandboxProvider::Local
} else {
resolve_sandbox_provider(
args.sandbox.map(Into::into),
Some(&source_input.config),
run_defaults,
)?
};
let mut config = source_input.config.clone();
apply_execution_overrides(
&mut config,
&ExecutionOverrides {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
model: args.model.as_deref(),
provider: args.provider.as_deref(),
sandbox_provider,
},
)?;
let graph_source = persisted.source().to_string();
);
let persisted = match fabro_workflows::operations::create(
&source_input.raw_source,
RunCreateSettings {
config,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: source_input.workflow_slug.clone(),
labels: parse_labels(&args.label),
base_branch: detected_base_branch.clone(),
working_directory: Some(resume_repo_path.clone()),
host_repo_path: Some(resume_repo_path.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 }) => {
print_diagnostics_from_error(&diagnostics, styles);
bail!("Validation failed");
}
Err(err) => return Err(err.into()),
};
print_workflow_report_from_persisted(&persisted, &source_input.dot_path, styles);
let graph_source = source_input.raw_source;
let run_cfg = Some(persisted.run_record().config.clone());
let sandbox_provider = prepared.sandbox_provider;
(persisted, run_cfg, sandbox_provider, graph_source)
} else if let Ok(loaded) = Persisted::load(&run_dir) {
let sandbox_provider = if args.dry_run {
@ -662,45 +667,33 @@ async fn prepare_from_branch(
.unwrap_or_default();
args.sandbox.map(Into::into).unwrap_or(sp)
};
let validated = create_from_graph(rec.graph.clone(), String::new());
let graph = validated.graph().clone();
let run_cfg = Some(rec.config.clone());
let (model_str, provider_str) = resolve_model_provider(
args.model.as_deref(),
args.provider.as_deref(),
run_cfg.as_ref(),
run_defaults,
&graph,
let mut config = rec.config.clone();
apply_execution_overrides(
&mut config,
&ExecutionOverrides {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
model: args.model.as_deref(),
provider: args.provider.as_deref(),
sandbox_provider,
},
);
let normalized = super::create::normalize_config(
run_cfg.as_ref(),
run_defaults,
&model_str,
provider_str.as_deref(),
sandbox_provider,
&graph,
cli_flags,
);
let persisted = fabro_workflows::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record: fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized,
graph,
workflow_slug: rec.workflow_slug.clone(),
working_directory: resume_repo_path.clone(),
host_repo_path: Some(resume_repo_path.to_string_lossy().to_string()),
base_branch: detected_base_branch.clone(),
labels: args
.label
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
},
let persisted = create_from_graph(
rec.graph.clone(),
RunCreateSettings {
config,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: rec.workflow_slug.clone(),
labels: parse_labels(&args.label),
base_branch: detected_base_branch.clone(),
working_directory: Some(resume_repo_path.clone()),
host_repo_path: Some(resume_repo_path.to_string_lossy().to_string()),
goal_override: None,
base_dir: None,
},
)?;
let graph_source = persisted.source().to_string();

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

@ -25,8 +25,7 @@ use fabro_workflows::operations::{start, StartFinalizeConfig, StartOptions, Star
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::outcome::{compute_stage_cost, format_cost};
use fabro_workflows::pipeline::{
build_conclusion, classify_engine_result, persist_terminal_outcome, PersistOptions, Persisted,
Validated,
build_conclusion, classify_engine_result, persist_terminal_outcome, Persisted, Validated,
};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
@ -163,36 +162,7 @@ pub(crate) fn resolve_cli_goal(
}
}
/// Apply goal to the graph from TOML config or CLI flag.
/// Precedence: CLI `--goal` / `--goal-file` > TOML `goal` > DOT `graph [goal="..."]`.
pub(crate) fn apply_goal_override(
graph: &mut fabro_graphviz::graph::Graph,
cli_goal: Option<&str>,
toml_goal: Option<&str>,
) {
let goal = cli_goal.or(toml_goal);
if let Some(goal) = goal {
debug!(goal = %goal, "overriding graph goal");
graph.attrs.insert(
"goal".to_string(),
fabro_graphviz::graph::AttrValue::String(goal.to_string()),
);
}
}
/// Compute the default run directory when `--run-dir` is not provided.
pub(crate) fn default_run_dir(run_id: &str, dry_run: bool) -> PathBuf {
let base = fabro_workflows::run_lookup::default_runs_base();
if dry_run {
base.join(format!(
"{}-dry-run-{}",
Local::now().format("%Y%m%d"),
run_id
))
} else {
base.join(format!("{}-{}", Local::now().format("%Y%m%d"), run_id))
}
}
pub(crate) use fabro_workflows::operations::default_run_dir;
pub(crate) fn workflow_slug_from_path(workflow_path: &Path) -> Option<String> {
let file_name = workflow_path.file_name()?.to_string_lossy();
@ -545,56 +515,114 @@ pub(crate) fn resolve_workflow_source(
}
}
/// Result of workflow preparation (shared between `create` and `run` commands).
pub(crate) struct PreparedWorkflow {
pub validated: fabro_workflows::pipeline::Validated,
pub raw_source: String,
pub run_cfg: Option<FabroConfig>,
pub(crate) struct ExecutionOverrides<'a> {
pub dry_run: bool,
pub auto_approve: bool,
pub no_retro: bool,
pub verbose: bool,
pub preserve_sandbox: bool,
pub model: Option<&'a str>,
pub provider: Option<&'a str>,
pub sandbox_provider: SandboxProvider,
pub model: String,
pub provider: Option<String>,
pub workflow_slug: Option<String>,
pub run_defaults: FabroConfig,
/// Resolved TOML path (Some for TOML-based workflows, None for bare .fabro).
pub workflow_toml_path: Option<PathBuf>,
}
impl PreparedWorkflow {
/// Read-through to validated graph.
pub fn graph(&self) -> &fabro_graphviz::graph::Graph {
self.validated.graph()
pub(crate) fn apply_execution_overrides(config: &mut FabroConfig, overrides: &ExecutionOverrides) {
config.dry_run = Some(overrides.dry_run);
config.auto_approve = Some(overrides.auto_approve);
config.no_retro = Some(overrides.no_retro);
config.verbose = Some(overrides.verbose);
if let Some(model) = overrides.model {
config.llm.get_or_insert_default().model = Some(model.to_string());
}
/// Original DOT source as authored on disk, before runtime var expansion.
pub fn source(&self) -> &str {
&self.raw_source
if let Some(provider) = overrides.provider {
config.llm.get_or_insert_default().provider = Some(provider.to_string());
}
config.sandbox.get_or_insert_default().provider = Some(overrides.sandbox_provider.to_string());
if overrides.preserve_sandbox {
config.sandbox.get_or_insert_default().preserve = 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()
}
fn print_workflow_header(
graph: &fabro_graphviz::graph::Graph,
diagnostics: &[fabro_validate::Diagnostic],
dot_path: &Path,
styles: &Styles,
) {
eprintln!(
"{} {} {}",
styles.bold.apply_to("Workflow:"),
graph.name,
styles.dim.apply_to(format!(
"({} nodes, {} edges)",
graph.nodes.len(),
graph.edges.len()
)),
);
eprintln!(
"{} {}",
styles.dim.apply_to("Graph:"),
styles.dim.apply_to(relative_path(dot_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: &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: &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) struct WorkflowSourceInput {
pub raw_source: String,
pub config: FabroConfig,
pub workflow_slug: Option<String>,
pub run_defaults: FabroConfig,
pub workflow_toml_path: Option<PathBuf>,
pub dot_path: PathBuf,
pub goal_override: Option<String>,
}
enum WorkflowState {
Validated(Validated),
Source(Box<WorkflowSourceInput>),
Persisted(Box<Persisted>),
}
/// Resolve config, parse/validate the workflow graph, and resolve sandbox + model.
///
/// Shared between `create_run` (which only persists the spec) and
/// `run_command` (which goes on to execute the workflow).
pub(crate) fn prepare_workflow(
args: &RunArgs,
run_defaults: FabroConfig,
styles: &Styles,
quiet: bool,
) -> anyhow::Result<PreparedWorkflow> {
prepare_workflow_with_project_config(args, run_defaults, styles, quiet, true)
}
pub(crate) fn prepare_workflow_with_project_config(
pub(crate) fn load_workflow_source_input(
args: &RunArgs,
mut run_defaults: FabroConfig,
styles: &Styles,
quiet: bool,
apply_project_config: bool,
) -> anyhow::Result<PreparedWorkflow> {
) -> anyhow::Result<WorkflowSourceInput> {
let workflow_path = args
.workflow
.as_ref()
@ -610,122 +638,28 @@ pub(crate) fn prepare_workflow_with_project_config(
}
}
// Resolve workflow arg, load run config if TOML, merge with defaults
let (resolved_workflow_path, dot_path, run_cfg) = {
// Resolve workflow arg, load run config if TOML, merge with defaults.
let (resolved_workflow_path, dot_path, config) = {
let (resolved, dot, cfg) = resolve_workflow_source(workflow_path)?;
match cfg {
Some(cfg) => {
// run_defaults is the base; cfg (from workflow.toml) is the overlay that wins
let mut merged = run_defaults.clone();
merged.merge_overlay(cfg);
(resolved, dot, Some(merged))
(resolved, dot, merged)
}
None => (resolved, dot, None),
None => (resolved, dot, run_defaults.clone()),
}
};
let workflow_slug = workflow_slug_from_path(&resolved_workflow_path);
let directory = run_cfg
.as_ref()
.and_then(|c| c.work_dir.as_deref())
.or(run_defaults.work_dir.as_deref());
if let Some(dir) = directory {
if let Some(dir) = config.work_dir.as_deref() {
std::env::set_current_dir(dir)
.map_err(|e| anyhow::anyhow!("Failed to set working directory to {dir}: {e}"))?;
}
// Parse and transform workflow using pipeline functions
let raw_source = read_workflow_file(&dot_path)?;
let vars = run_cfg
.as_ref()
.and_then(|c| c.vars.as_ref())
.or(run_defaults.vars.as_ref());
let source = match vars {
Some(vars) => fabro_workflows::vars::expand_vars(&raw_source, vars)?,
None => raw_source.clone(),
};
let dot_dir = dot_path.parent().unwrap_or(std::path::Path::new("."));
let parsed = fabro_workflows::pipeline::parse(&source)?;
let mut transformed = fabro_workflows::pipeline::transform(
parsed,
&fabro_workflows::pipeline::TransformOptions {
base_dir: Some(dot_dir.to_path_buf()),
custom_transforms: vec![],
},
);
// Apply goal override on the mutable transformed graph
let cli_goal = resolve_cli_goal(&args.goal, &args.goal_file)?;
let toml_goal = run_cfg.as_ref().and_then(|c| c.goal.as_deref());
apply_goal_override(&mut transformed.graph, cli_goal.as_deref(), toml_goal);
// Inline @file references in the (possibly overridden) goal
if let Some(fabro_graphviz::graph::AttrValue::String(goal)) =
transformed.graph.attrs.get("goal")
{
let fallback = dirs::home_dir().map(|h| h.join(".fabro"));
let resolved =
fabro_workflows::transform::resolve_file_ref(goal, dot_dir, fallback.as_deref());
if resolved != *goal {
transformed.graph.attrs.insert(
"goal".to_string(),
fabro_graphviz::graph::AttrValue::String(resolved),
);
}
}
let validated = fabro_workflows::pipeline::validate(transformed, &[]);
if !quiet {
eprintln!(
"{} {} {}",
styles.bold.apply_to("Workflow:"),
validated.graph().name,
styles.dim.apply_to(format!(
"({} nodes, {} edges)",
validated.graph().nodes.len(),
validated.graph().edges.len()
)),
);
eprintln!(
"{} {}",
styles.dim.apply_to("Graph:"),
styles.dim.apply_to(relative_path(&dot_path)),
);
let goal = validated.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(validated.diagnostics(), styles);
}
if validated.has_errors() {
bail!("Validation failed");
}
// Resolve sandbox provider
let sandbox_provider = if args.dry_run {
SandboxProvider::Local
} else {
resolve_sandbox_provider(
args.sandbox.map(Into::into),
run_cfg.as_ref(),
&run_defaults,
)?
};
// Resolve model and provider
let (model, provider) = resolve_model_provider(
args.model.as_deref(),
args.provider.as_deref(),
run_cfg.as_ref(),
&run_defaults,
validated.graph(),
);
let goal_override = cli_goal.or_else(|| config.goal.clone());
let workflow_toml_path = if resolved_workflow_path
.extension()
@ -736,31 +670,21 @@ pub(crate) fn prepare_workflow_with_project_config(
None
};
Ok(PreparedWorkflow {
validated,
Ok(WorkflowSourceInput {
raw_source,
run_cfg,
sandbox_provider,
model,
provider,
config,
workflow_slug,
run_defaults,
workflow_toml_path,
dot_path,
goal_override,
})
}
/// Pre-prepared run state, used to skip workflow preparation in `run_command_impl`.
struct RecordBasedRun {
workflow: WorkflowState,
raw_source: String,
run_cfg: Option<FabroConfig>,
sandbox_provider: SandboxProvider,
model: String,
provider: Option<String>,
workflow_slug: Option<String>,
run_defaults: FabroConfig,
/// Original TOML path for debug snapshot (None for record-based or bare .fabro runs).
workflow_toml_path: Option<PathBuf>,
}
/// Execute a workflow run from a saved RunRecord, bypassing workflow preparation.
@ -775,6 +699,11 @@ pub async fn run_from_record(
git_author: fabro_workflows::git::GitAuthor,
) -> anyhow::Result<()> {
let record = persisted.run_record().clone();
let record_run = RecordBasedRun {
workflow: WorkflowState::Persisted(Box::new(persisted)),
run_defaults,
};
let sandbox_provider = record
.config
.sandbox
@ -796,18 +725,6 @@ pub async fn run_from_record(
.and_then(|l| l.provider.clone())
.filter(|s| !s.is_empty());
let record_run = RecordBasedRun {
workflow: WorkflowState::Persisted(Box::new(persisted)),
raw_source: String::new(), // Raw DOT provenance is best-effort for record-based runs
run_cfg: Some(record.config.clone()),
sandbox_provider,
model: model.clone(),
provider: provider.clone(),
workflow_slug: record.workflow_slug.clone(),
run_defaults,
workflow_toml_path: None, // No TOML to copy — config is in RunRecord
};
let args = RunArgs {
workflow: None,
run_dir: Some(run_dir),
@ -851,28 +768,12 @@ pub async fn run_command(
github_app: Option<fabro_github::GitHubAppCredentials>,
git_author: fabro_workflows::git::GitAuthor,
) -> anyhow::Result<()> {
let PreparedWorkflow {
validated,
raw_source,
run_cfg,
sandbox_provider,
model,
provider,
workflow_slug,
run_defaults,
workflow_toml_path,
} = prepare_workflow(&args, run_defaults, styles, false)?;
let source_input = load_workflow_source_input(&args, run_defaults, true)?;
let resolved_run_defaults = source_input.run_defaults.clone();
let record_run = RecordBasedRun {
workflow: WorkflowState::Validated(validated),
raw_source,
run_cfg,
sandbox_provider,
model,
provider,
workflow_slug,
run_defaults,
workflow_toml_path,
workflow: WorkflowState::Source(Box::new(source_input)),
run_defaults: resolved_run_defaults,
};
run_command_impl(args, styles, github_app, git_author, Some(record_run)).await
@ -885,49 +786,15 @@ async fn run_command_impl(
git_author: fabro_workflows::git::GitAuthor,
record_run: Option<RecordBasedRun>,
) -> anyhow::Result<()> {
let (
workflow,
raw_source,
mut run_cfg,
sandbox_provider,
model,
provider,
prepared_workflow_slug,
run_defaults,
workflow_toml_path,
) = match record_run {
Some(rr) => (
rr.workflow,
rr.raw_source,
rr.run_cfg,
rr.sandbox_provider,
rr.model,
rr.provider,
rr.workflow_slug,
rr.run_defaults,
rr.workflow_toml_path,
),
let (workflow, run_defaults) = match record_run {
Some(rr) => (rr.workflow, rr.run_defaults),
None => unreachable!("run_command_impl always receives a RecordBasedRun"),
};
let graph = match &workflow {
WorkflowState::Validated(validated) => validated.graph().clone(),
WorkflowState::Persisted(persisted) => persisted.graph().clone(),
};
// For record-based runs from run_from_record, the workflow has already been persisted.
let from_record = matches!(&workflow, WorkflowState::Persisted(_)) && args.workflow.is_none();
// Collect setup commands — they'll be run inside the sandbox
let setup_commands: Vec<String> = run_cfg
.as_ref()
.and_then(|c| c.setup.as_ref())
.or(run_defaults.setup.as_ref())
.map(|s| s.commands.clone())
.unwrap_or_default();
// Pre-flight: check git cleanliness before creating any files
let preserve_sandbox =
resolve_preserve_sandbox(args.preserve_sandbox, run_cfg.as_ref(), &run_defaults);
let original_cwd = std::env::current_dir()?;
let (origin_url, detected_base_branch) =
fabro_sandbox::daytona::detect_repo_info(&original_cwd)
@ -936,21 +803,6 @@ async fn run_command_impl(
let git_status =
fabro_workflows::git::sync_status(&original_cwd, "origin", detected_base_branch.as_deref());
if args.preflight {
return run_preflight(
&graph,
&run_cfg,
&args,
&run_defaults,
git_status,
sandbox_provider,
styles,
github_app,
origin_url.as_deref(),
)
.await;
}
// 3. Create logs directory
// Extract values from args before partial move
let dry_run_flag = args.dry_run;
@ -959,63 +811,156 @@ async fn run_command_impl(
let verbose_flag = args.verbose;
let preserve_sandbox_flag = args.preserve_sandbox;
let label_vec = args.label.clone();
let run_id = args.run_id.unwrap_or_else(|| ulid::Ulid::new().to_string());
let run_id = args
.run_id
.clone()
.unwrap_or_else(|| ulid::Ulid::new().to_string());
let run_dir = args
.run_dir
.clone()
.unwrap_or_else(|| default_run_dir(&run_id, dry_run_flag));
let cached_run_restart = if from_record {
false
} else {
let workflow_path = args.workflow.as_ref().unwrap();
is_cached_run_restart(workflow_path, &run_dir)
let cached_run_restart = match &workflow {
WorkflowState::Source(_) if !from_record => {
let workflow_path = args.workflow.as_ref().unwrap();
is_cached_run_restart(workflow_path, &run_dir)
}
_ => false,
};
let persisted = match (cached_run_restart, workflow) {
(true, _) => Persisted::load(&run_dir)?,
(false, WorkflowState::Persisted(persisted)) => *persisted,
(false, WorkflowState::Validated(validated)) => {
let cli_flags = super::create::CliFlags {
dry_run: dry_run_flag,
auto_approve: auto_approve_flag,
no_retro: no_retro_flag,
verbose: verbose_flag,
preserve_sandbox: preserve_sandbox_flag,
let (persisted, raw_source, workflow_toml_path) = match workflow {
WorkflowState::Persisted(persisted) => (*persisted, String::new(), None),
WorkflowState::Source(source_input) if cached_run_restart => (
Persisted::load(&run_dir)?,
source_input.raw_source,
source_input.workflow_toml_path,
),
WorkflowState::Source(source_input) => {
let mut config = source_input.config.clone();
let sandbox_provider = if dry_run_flag {
SandboxProvider::Local
} else {
resolve_sandbox_provider(
args.sandbox.map(Into::into),
Some(&config),
&source_input.run_defaults,
)?
};
let normalized_config = super::create::normalize_config(
run_cfg.as_ref(),
&run_defaults,
&model,
provider.as_deref(),
sandbox_provider,
validated.graph(),
cli_flags,
);
let run_record = fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized_config,
graph: validated.graph().clone(),
workflow_slug: prepared_workflow_slug.clone(),
working_directory: original_cwd.clone(),
host_repo_path: Some(original_cwd.to_string_lossy().to_string()),
base_branch: detected_base_branch.clone(),
labels: label_vec
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
};
fabro_workflows::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record,
apply_execution_overrides(
&mut config,
&ExecutionOverrides {
dry_run: dry_run_flag,
auto_approve: auto_approve_flag,
no_retro: no_retro_flag,
verbose: verbose_flag,
preserve_sandbox: preserve_sandbox_flag,
model: args.model.as_deref(),
provider: args.provider.as_deref(),
sandbox_provider,
},
)?
);
if args.preflight {
let validated = fabro_workflows::operations::validate(
&source_input.raw_source,
fabro_workflows::operations::ValidateOptions {
base_dir: Some(
source_input
.dot_path
.parent()
.unwrap_or(Path::new("."))
.to_path_buf(),
),
config: Some(config.clone()),
goal_override: source_input.goal_override.clone(),
..Default::default()
},
)?;
print_workflow_report(&validated, &source_input.dot_path, styles);
if validated.has_errors() {
bail!("Validation failed");
}
return run_preflight(
validated.graph(),
&Some(config),
&args,
&run_defaults,
git_status,
sandbox_provider,
styles,
github_app,
origin_url.as_deref(),
)
.await;
}
match fabro_workflows::operations::create(
&source_input.raw_source,
fabro_workflows::operations::RunCreateSettings {
config,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: source_input.workflow_slug.clone(),
labels: parse_labels(&label_vec),
base_branch: detected_base_branch.clone(),
working_directory: Some(original_cwd.clone()),
host_repo_path: Some(original_cwd.to_string_lossy().to_string()),
goal_override: source_input.goal_override.clone(),
base_dir: Some(
source_input
.dot_path
.parent()
.unwrap_or(Path::new("."))
.to_path_buf(),
),
},
) {
Ok(persisted) => {
print_workflow_report_from_persisted(
&persisted,
&source_input.dot_path,
styles,
);
(
persisted,
source_input.raw_source,
source_input.workflow_toml_path,
)
}
Err(fabro_workflows::error::FabroError::ValidationFailed { diagnostics }) => {
print_diagnostics_from_error(&diagnostics, styles);
bail!("Validation failed");
}
Err(err) => return Err(err.into()),
}
}
};
run_cfg = Some(persisted.run_record().config.clone());
let mut run_cfg = Some(persisted.run_record().config.clone());
let workflow_slug = persisted.run_record().workflow_slug.clone();
let sandbox_provider = run_cfg
.as_ref()
.and_then(|cfg| cfg.sandbox.as_ref())
.and_then(|sandbox| sandbox.provider.as_deref())
.unwrap_or("local")
.parse()
.unwrap_or(SandboxProvider::Local);
let model = run_cfg
.as_ref()
.and_then(|cfg| cfg.llm.as_ref())
.and_then(|llm| llm.model.clone())
.unwrap_or_default();
let provider = run_cfg
.as_ref()
.and_then(|cfg| cfg.llm.as_ref())
.and_then(|llm| llm.provider.clone())
.filter(|value| !value.is_empty());
let preserve_sandbox =
resolve_preserve_sandbox(args.preserve_sandbox, run_cfg.as_ref(), &run_defaults);
let setup_commands: Vec<String> = run_cfg
.as_ref()
.and_then(|c| c.setup.as_ref())
.or(run_defaults.setup.as_ref())
.map(|s| s.commands.clone())
.unwrap_or_default();
tokio::fs::create_dir_all(&run_dir).await?;
fabro_util::run_log::activate(&run_dir.join("cli.log"))
@ -2522,7 +2467,7 @@ mod tests {
}
#[test]
fn prepare_workflow_with_project_config_resolves_workflow_toml_settings() {
fn load_workflow_source_input_resolves_workflow_toml_settings() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(
dir.path().join("workflow.fabro"),
@ -2579,26 +2524,36 @@ include = ["*.md"]
run_id: None,
};
let styles = Styles::new(false);
let prepared = prepare_workflow_with_project_config(
&args,
FabroConfig::default(),
&styles,
true,
false,
let source_input =
load_workflow_source_input(&args, FabroConfig::default(), false).unwrap();
let validated = fabro_workflows::operations::validate(
&source_input.raw_source,
fabro_workflows::operations::ValidateOptions {
base_dir: Some(dir.path().to_path_buf()),
config: Some(source_input.config.clone()),
goal_override: source_input.goal_override.clone(),
..Default::default()
},
)
.unwrap();
assert_eq!(prepared.graph().name, "smoke");
assert_eq!(prepared.graph().goal(), "toml goal");
assert_eq!(prepared.sandbox_provider, SandboxProvider::Docker);
assert_eq!(prepared.model, "gpt-5.2");
assert_eq!(prepared.provider.as_deref(), Some("openai"));
assert_eq!(validated.graph().name, "smoke");
assert_eq!(validated.graph().goal(), "toml goal");
let (model, provider) = resolve_model_provider(
None,
None,
Some(&source_input.config),
&source_input.run_defaults,
validated.graph(),
);
assert_eq!(model, "gpt-5.2");
assert_eq!(provider.as_deref(), Some("openai"));
let sandbox_provider =
resolve_sandbox_provider(None, Some(&source_input.config), &source_input.run_defaults)
.unwrap();
assert_eq!(sandbox_provider, SandboxProvider::Docker);
let run_cfg = prepared
.run_cfg
.as_ref()
.expect("run config should be loaded");
let run_cfg = &source_input.config;
assert_eq!(
run_cfg
.setup
@ -2624,42 +2579,6 @@ include = ["*.md"]
);
}
#[test]
fn apply_goal_override_cli_wins_over_toml() {
use fabro_graphviz::graph::{AttrValue, Graph};
let mut graph = Graph::new("test");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("original".to_string()),
);
apply_goal_override(&mut graph, Some("CLI goal"), Some("TOML goal"));
assert_eq!(graph.goal(), "CLI goal");
}
#[test]
fn apply_goal_override_toml_wins_over_dot() {
use fabro_graphviz::graph::{AttrValue, Graph};
let mut graph = Graph::new("test");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("original".to_string()),
);
apply_goal_override(&mut graph, None, Some("TOML goal"));
assert_eq!(graph.goal(), "TOML goal");
}
#[test]
fn apply_goal_override_noop_when_none() {
use fabro_graphviz::graph::{AttrValue, Graph};
let mut graph = Graph::new("test");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("original".to_string()),
);
apply_goal_override(&mut graph, None, None);
assert_eq!(graph.goal(), "original");
}
#[test]
fn resolve_cli_goal_from_file() {
let dir = tempfile::tempdir().unwrap();

2
lib/crates/fabro-cli/src/commands/validate.rs
View file

@ -16,7 +16,7 @@ pub struct ValidateArgs {
pub fn run(args: &ValidateArgs, styles: &Styles) -> anyhow::Result<()> {
let (dot_path, _cfg) = fabro_config::project::resolve_workflow(&args.workflow)?;
let validated = fabro_workflows::operations::create_from_file(&dot_path)?;
let validated = fabro_workflows::operations::validate_from_file(&dot_path)?;
let graph = validated.graph();
let diagnostics = validated.diagnostics();

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

@ -356,7 +356,7 @@ async fn run_engine_entrypoint(
}
// Use run_from_record: loads config + graph directly from persisted state,
// skipping prepare_workflow() entirely. No TOML/DOT re-parsing needed.
// skipping workflow source loading and preprocessing entirely.
match commands::run::run_from_record(
persisted,
run_dir.clone(),

42
lib/crates/fabro-workflows/src/error.rs
View file

@ -1,6 +1,7 @@
use std::fmt;
use fabro_llm::error::{ProviderErrorKind, SdkError};
use fabro_validate::Diagnostic;
use serde::{Deserialize, Serialize};
use thiserror::Error;
@ -205,6 +206,9 @@ pub enum FabroError {
#[error("Validation error: {0}")]
Validation(String),
#[error("Validation failed")]
ValidationFailed { diagnostics: Vec<Diagnostic> },
#[error("Engine error: {message}")]
Engine {
message: String,
@ -267,6 +271,7 @@ impl FabroError {
Self::Llm(sdk_err) => sdk_err.retryable(),
Self::Parse(_)
| Self::Validation(_)
| Self::ValidationFailed { .. }
| Self::Stylesheet(_)
| Self::Checkpoint(_)
| Self::Cancelled => false,
@ -280,9 +285,11 @@ impl FabroError {
Self::Cancelled => FailureCategory::Canceled,
Self::Llm(sdk_err) => classify_sdk_error(sdk_err),
Self::Io(_) => FailureCategory::TransientInfra,
Self::Parse(_) | Self::Validation(_) | Self::Stylesheet(_) | Self::Checkpoint(_) => {
FailureCategory::Deterministic
}
Self::Parse(_)
| Self::Validation(_)
| Self::ValidationFailed { .. }
| Self::Stylesheet(_)
| Self::Checkpoint(_) => FailureCategory::Deterministic,
Self::Handler { failure_class, .. } | Self::Engine { failure_class, .. } => {
*failure_class
}
@ -360,6 +367,21 @@ mod tests {
assert_eq!(err.to_string(), "Validation error: missing start node");
}
#[test]
fn validation_failed_display() {
let err = FabroError::ValidationFailed {
diagnostics: vec![Diagnostic {
rule: "test".to_string(),
severity: fabro_validate::Severity::Error,
message: "missing start node".to_string(),
node_id: None,
edge: None,
fix: None,
}],
};
assert_eq!(err.to_string(), "Validation failed");
}
#[test]
fn engine_error_display() {
let err = FabroError::engine("no outgoing edge");
@ -416,6 +438,10 @@ mod tests {
fn is_retryable_terminal_errors() {
assert!(!FabroError::Parse("bad".to_string()).is_retryable());
assert!(!FabroError::Validation("bad".to_string()).is_retryable());
assert!(!FabroError::ValidationFailed {
diagnostics: vec![]
}
.is_retryable());
assert!(!FabroError::Stylesheet("bad".to_string()).is_retryable());
assert!(!FabroError::Checkpoint("bad".to_string()).is_retryable());
}
@ -1519,6 +1545,16 @@ mod tests {
let errors: Vec<FabroError> = vec![
FabroError::Parse("bad".into()),
FabroError::Validation("bad".into()),
FabroError::ValidationFailed {
diagnostics: vec![Diagnostic {
rule: "test".into(),
severity: fabro_validate::Severity::Error,
message: "bad".into(),
node_id: None,
edge: None,
fix: None,
}],
},
FabroError::engine("engine err"),
FabroError::handler("handler err"),
FabroError::Llm(SdkError::Network {

6
lib/crates/fabro-workflows/src/handler/manager_loop.rs
View file

@ -10,7 +10,7 @@ use crate::condition::evaluate_condition;
use crate::context::keys;
use crate::context::{Context, WorkflowContext};
use crate::error::FabroError;
use crate::operations::{create, create_from_file, CreateOptions};
use crate::operations::{validate, validate_from_file, ValidateOptions};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::pipeline;
use crate::pipeline::types::Initialized;
@ -54,7 +54,7 @@ fn parse_child_graph(node: &Node) -> Result<Graph, FabroError> {
.get("stack.child_dot_source")
.and_then(|v| v.as_str())
{
let validated = create(dot, CreateOptions::default())?;
let validated = validate(dot, ValidateOptions::default())?;
validated.raise_on_errors()?;
let (graph, _, _) = validated.into_parts();
return Ok(graph);
@ -65,7 +65,7 @@ fn parse_child_graph(node: &Node) -> Result<Graph, FabroError> {
.or_else(|| node.attrs.get("stack.child_dotfile"))
.and_then(|v| v.as_str())
{
let validated = create_from_file(std::path::Path::new(path))?;
let validated = validate_from_file(std::path::Path::new(path))?;
validated.raise_on_errors()?;
let (graph, _, _) = validated.into_parts();
return Ok(graph);

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

@ -1,54 +1,252 @@
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use fabro_graphviz::graph::Graph;
use chrono::{Local, Utc};
use fabro_config::config::FabroConfig;
use fabro_graphviz::graph::{AttrValue, Graph};
use fabro_model::{Catalog, Provider};
use crate::error::FabroError;
use crate::pipeline::{self, TransformOptions, Validated};
use crate::transforms::Transform;
use crate::pipeline::types::PersistOptions;
use crate::pipeline::{self, Persisted, TransformOptions, Validated};
use crate::records::RunRecord;
use crate::transforms::{expand_vars, Transform};
#[derive(Default)]
pub struct CreateOptions {
pub struct ValidateOptions {
pub base_dir: Option<PathBuf>,
pub custom_transforms: Vec<Box<dyn Transform>>,
pub config: Option<FabroConfig>,
pub goal_override: Option<String>,
}
pub struct RunCreateSettings {
pub config: FabroConfig,
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>,
}
/// Parse, transform, and validate a DOT source string.
///
/// Returns `Validated` even when validation produced errors. Call
/// `validated.raise_on_errors()` if the caller wants to fail fast.
pub fn create(dot_source: &str, options: CreateOptions) -> Result<Validated, FabroError> {
let parsed = pipeline::parse(dot_source)?;
let transformed = pipeline::transform(
parsed,
&TransformOptions {
base_dir: options.base_dir,
custom_transforms: options.custom_transforms,
},
);
Ok(pipeline::validate(transformed, &[]))
pub fn validate(dot_source: &str, options: ValidateOptions) -> Result<Validated, FabroError> {
preprocess_and_validate(
dot_source,
options.base_dir,
options.custom_transforms,
options.config.as_ref(),
options.goal_override.as_deref(),
)
}
/// Read a DOT file, apply file inlining from its parent directory, then create.
pub fn create_from_file(path: &Path) -> Result<Validated, FabroError> {
/// Read a DOT file, apply file inlining from its parent directory, then validate.
pub fn validate_from_file(path: &Path) -> Result<Validated, 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("."));
create(
validate(
&source,
CreateOptions {
ValidateOptions {
base_dir: Some(base_dir.to_path_buf()),
..Default::default()
},
)
}
/// Build a validated workflow from an already-materialized graph.
/// Parse, transform, validate, normalize config, and persist a run.
pub fn create(dot_source: &str, settings: RunCreateSettings) -> Result<Persisted, FabroError> {
let validated = preprocess_and_validate(
dot_source,
settings.base_dir.clone(),
Vec::new(),
Some(&settings.config),
settings.goal_override.as_deref(),
)?;
if validated.has_errors() {
return Err(FabroError::ValidationFailed {
diagnostics: validated.diagnostics().to_vec(),
});
}
persist_validated(validated, settings)
}
/// Read a DOT file, apply file inlining from its parent directory, then create.
pub fn create_from_file(
path: &Path,
mut settings: RunCreateSettings,
) -> 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("."));
settings.base_dir = Some(base_dir.to_path_buf());
create(&source, settings)
}
/// Build a persisted workflow from an already-materialized graph.
///
/// This is used by detached/resume CLI paths that load a graph from `RunRecord`
/// instead of re-parsing DOT source.
#[doc(hidden)]
pub fn create_from_graph(graph: Graph, source: impl Into<String>) -> Validated {
Validated::new(graph, source.into(), vec![])
pub fn create_from_graph(
mut graph: Graph,
settings: RunCreateSettings,
) -> Result<Persisted, FabroError> {
if let Some(goal_override) = settings.goal_override.as_deref() {
apply_goal_override(&mut graph, Some(goal_override));
}
let validated = Validated::new(graph, String::new(), vec![]);
persist_validated(validated, settings)
}
fn preprocess_and_validate(
dot_source: &str,
base_dir: Option<PathBuf>,
custom_transforms: Vec<Box<dyn Transform>>,
config: Option<&FabroConfig>,
goal_override: Option<&str>,
) -> Result<Validated, FabroError> {
let source = match config.and_then(|cfg| cfg.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}")))?
}
None => dot_source.to_string(),
};
let mut parsed = pipeline::parse(&source)?;
apply_goal_override(&mut parsed.graph, goal_override);
let transformed = pipeline::transform(
parsed,
&TransformOptions {
base_dir,
custom_transforms,
},
);
Ok(pipeline::validate(transformed, &[]))
}
fn apply_goal_override(graph: &mut Graph, goal_override: Option<&str>) {
if let Some(goal_override) = goal_override {
graph.attrs.insert(
"goal".to_string(),
AttrValue::String(goal_override.to_string()),
);
}
}
fn persist_validated(
validated: Validated,
settings: RunCreateSettings,
) -> Result<Persisted, FabroError> {
let RunCreateSettings {
mut config,
run_dir,
run_id,
workflow_slug,
labels,
base_branch,
working_directory,
host_repo_path,
goal_override: _,
base_dir: _,
} = settings;
finalize_config(&mut config, validated.graph());
let run_id = run_id.unwrap_or_else(|| ulid::Ulid::new().to_string());
let run_dir = run_dir.unwrap_or_else(|| default_run_dir(&run_id, config.dry_run_enabled()));
let working_directory = working_directory
.unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")));
let run_record = RunRecord {
run_id,
created_at: Utc::now(),
config,
graph: validated.graph().clone(),
workflow_slug,
working_directory,
host_repo_path,
base_branch,
labels,
};
pipeline::persist(
validated,
PersistOptions {
run_dir,
run_record,
},
)
}
fn finalize_config(config: &mut FabroConfig, graph: &Graph) {
let llm_config = config.llm.as_ref();
let configured_model = llm_config.and_then(|l| l.model.as_deref());
let configured_provider = llm_config.and_then(|l| l.provider.as_deref());
let graph_provider = graph.attrs.get("default_provider").and_then(|v| v.as_str());
let graph_model = graph.attrs.get("default_model").and_then(|v| v.as_str());
let provider = configured_provider.or(graph_provider).map(str::to_string);
let model = configured_model
.or(graph_model)
.map(str::to_string)
.unwrap_or_else(|| {
let catalog = Catalog::builtin();
provider
.as_deref()
.and_then(|value| value.parse::<Provider>().ok())
.and_then(|provider| catalog.default_for_provider(provider))
.unwrap_or_else(|| catalog.default_from_env())
.id
.clone()
});
let (resolved_model, resolved_provider) = match Catalog::builtin().get(&model) {
Some(info) => (
info.id.clone(),
provider.or(Some(info.provider.to_string())),
),
None => (model, provider),
};
let llm = config.llm.get_or_insert_default();
llm.model = Some(resolved_model);
llm.provider = resolved_provider;
let goal = graph.goal().to_string();
config.goal = if goal.is_empty() { None } else { Some(goal) };
config.pull_request = config
.pull_request
.take()
.filter(|pull_request| pull_request.enabled);
}
pub fn default_run_dir(run_id: &str, dry_run: bool) -> PathBuf {
let base = crate::run_lookup::default_runs_base();
if dry_run {
base.join(format!(
"{}-dry-run-{}",
Local::now().format("%Y%m%d"),
run_id
))
} else {
base.join(format!("{}-{}", Local::now().format("%Y%m%d"), run_id))
}
}
#[cfg(test)]
@ -64,8 +262,8 @@ mod tests {
}"#;
#[test]
fn create_minimal() {
let validated = create(MINIMAL_DOT, CreateOptions::default()).unwrap();
fn validate_minimal() {
let validated = validate(MINIMAL_DOT, ValidateOptions::default()).unwrap();
validated.raise_on_errors().unwrap();
assert_eq!(validated.graph().name, "Test");
@ -74,7 +272,7 @@ mod tests {
}
#[test]
fn create_applies_variable_expansion() {
fn validate_applies_variable_expansion() {
let dot = r#"digraph Test {
graph [goal="Fix bugs"]
start [shape=Mdiamond]
@ -82,7 +280,7 @@ mod tests {
exit [shape=Msquare]
start -> work -> exit
}"#;
let validated = create(dot, CreateOptions::default()).unwrap();
let validated = validate(dot, ValidateOptions::default()).unwrap();
validated.raise_on_errors().unwrap();
let prompt = validated.graph().nodes["work"]
@ -94,7 +292,7 @@ mod tests {
}
#[test]
fn create_applies_stylesheet() {
fn validate_applies_stylesheet() {
let dot = r#"digraph Test {
graph [goal="Test", model_stylesheet="* { model: sonnet; }"]
start [shape=Mdiamond]
@ -102,7 +300,7 @@ mod tests {
exit [shape=Msquare]
start -> work -> exit
}"#;
let validated = create(dot, CreateOptions::default()).unwrap();
let validated = validate(dot, ValidateOptions::default()).unwrap();
validated.raise_on_errors().unwrap();
assert_eq!(
@ -112,25 +310,57 @@ mod tests {
}
#[test]
fn create_returns_error_on_invalid_dot() {
let result = create("not a graph", CreateOptions::default());
fn validate_applies_config_vars_and_goal_override() {
let dot = r#"digraph Test {
graph [goal="original"]
start [shape=Mdiamond]
work [prompt="$who: $goal"]
exit [shape=Msquare]
start -> work -> exit
}"#;
let validated = validate(
dot,
ValidateOptions {
config: Some(FabroConfig {
vars: Some(HashMap::from([("who".to_string(), "agent".to_string())])),
..Default::default()
}),
goal_override: Some("override".to_string()),
..Default::default()
},
)
.unwrap();
validated.raise_on_errors().unwrap();
assert_eq!(validated.graph().goal(), "override");
let prompt = validated.graph().nodes["work"]
.attrs
.get("prompt")
.and_then(AttrValue::as_str)
.unwrap();
assert_eq!(prompt, "agent: override");
}
#[test]
fn validate_returns_error_on_invalid_dot() {
let result = validate("not a graph", ValidateOptions::default());
assert!(result.is_err());
}
#[test]
fn create_returns_validation_diagnostics() {
fn validate_returns_validation_diagnostics() {
let dot = r#"digraph Test {
graph [goal="Test"]
work [label="Work"]
}"#;
let validated = create(dot, CreateOptions::default()).unwrap();
let validated = validate(dot, ValidateOptions::default()).unwrap();
assert!(validated.has_errors());
assert!(validated.raise_on_errors().is_err());
}
#[test]
fn create_supports_custom_transforms() {
fn validate_supports_custom_transforms() {
struct TagTransform;
impl Transform for TagTransform {
@ -142,9 +372,9 @@ mod tests {
}
}
let validated = create(
let validated = validate(
MINIMAL_DOT,
CreateOptions {
ValidateOptions {
custom_transforms: vec![Box::new(TagTransform)],
..Default::default()
},
@ -159,7 +389,7 @@ mod tests {
}
#[test]
fn create_from_file_uses_parent_directory_for_inlining() {
fn validate_from_file_uses_parent_directory_for_inlining() {
let dir = tempfile::tempdir().unwrap();
let data_path = dir.path().join("goal.txt");
let dot_path = dir.path().join("workflow.fabro");
@ -176,8 +406,102 @@ mod tests {
)
.unwrap();
let validated = create_from_file(&dot_path).unwrap();
let validated = validate_from_file(&dot_path).unwrap();
validated.raise_on_errors().unwrap();
assert_eq!(validated.graph().goal(), "ship it");
}
#[test]
fn create_returns_validation_failed_with_diagnostics() {
let dot = r#"digraph Test {
graph [goal="Test"]
work [label="Work"]
}"#;
let err = create(
dot,
RunCreateSettings {
config: FabroConfig::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,
base_dir: None,
},
)
.unwrap_err();
match err {
FabroError::ValidationFailed { diagnostics } => {
assert!(!diagnostics.is_empty());
}
other => panic!("expected ValidationFailed, got {other:?}"),
}
}
#[test]
fn create_persists_normalized_config() {
let dir = tempfile::tempdir().unwrap();
let persisted = create(
MINIMAL_DOT,
RunCreateSettings {
config: FabroConfig {
llm: Some(fabro_config::run::LlmConfig {
model: Some("sonnet".to_string()),
provider: None,
fallbacks: None,
}),
pull_request: Some(fabro_config::run::PullRequestConfig {
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()),
base_dir: None,
},
)
.unwrap();
assert_eq!(persisted.run_record().run_id, "run-123");
assert_eq!(persisted.run_record().graph.goal(), "override goal");
assert_eq!(
persisted
.run_record()
.config
.llm
.as_ref()
.and_then(|llm| llm.model.as_deref()),
Some("claude-sonnet-4-6")
);
assert_eq!(
persisted
.run_record()
.config
.llm
.as_ref()
.and_then(|llm| llm.provider.as_deref()),
Some("anthropic")
);
assert_eq!(
persisted.run_record().config.goal.as_deref(),
Some("override goal")
);
assert!(persisted.run_record().config.pull_request.is_none());
assert_eq!(
persisted.run_record().workflow_slug.as_deref(),
Some("slug")
);
}
}

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

@ -3,7 +3,10 @@ mod fork;
mod rewind;
mod start;
pub use create::{create, create_from_file, create_from_graph, CreateOptions};
pub use create::{
create, create_from_file, create_from_graph, default_run_dir, validate, validate_from_file,
RunCreateSettings, ValidateOptions,
};
pub use fork::fork;
pub use rewind::{
build_timeline, find_run_id_by_prefix, load_parallel_map, parse_target, resolve_target, rewind,

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

@ -123,7 +123,6 @@ mod tests {
use std::sync::atomic::{AtomicUsize, Ordering};
use async_trait::async_trait;
use chrono::Utc;
use fabro_agent::{DirEntry, ExecResult, GrepOptions, LocalSandbox, Sandbox};
use fabro_config::config::FabroConfig;
use fabro_graphviz::graph::{Graph, Node};
@ -136,8 +135,6 @@ mod tests {
use crate::handler::start::StartHandler;
use crate::handler::{Handler, HandlerRegistry};
use crate::outcome::Outcome;
use crate::pipeline::PersistOptions;
use crate::records::RunRecord;
use crate::run_settings::{LifecycleConfig, RunSettings};
const MINIMAL_DOT: &str = r#"digraph Test {
@ -344,25 +341,19 @@ mod tests {
}
fn persisted_workflow(dot: &str, run_dir: &std::path::Path) -> Persisted {
let validated =
crate::operations::create(dot, crate::operations::CreateOptions::default()).unwrap();
validated.raise_on_errors().unwrap();
let graph = validated.graph().clone();
crate::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.to_path_buf(),
run_record: RunRecord {
run_id: "run-test".to_string(),
created_at: Utc::now(),
config: FabroConfig::default(),
graph,
workflow_slug: Some("test".to_string()),
working_directory: std::env::current_dir().unwrap(),
host_repo_path: Some(std::env::current_dir().unwrap().display().to_string()),
base_branch: Some("main".to_string()),
labels: HashMap::new(),
},
crate::operations::create(
dot,
crate::operations::RunCreateSettings {
config: FabroConfig::default(),
run_dir: Some(run_dir.to_path_buf()),
run_id: Some("run-test".to_string()),
workflow_slug: Some("test".to_string()),
labels: std::collections::HashMap::new(),
base_branch: Some("main".to_string()),
working_directory: Some(std::env::current_dir().unwrap()),
host_repo_path: Some(std::env::current_dir().unwrap().display().to_string()),
goal_override: None,
base_dir: None,
},
)
.unwrap()

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

@ -197,7 +197,7 @@ mod tests {
use super::*;
use crate::handler::default_registry;
use crate::pipeline::PersistOptions;
use crate::pipeline::types::PersistOptions;
use crate::records::RunRecord;
use crate::run_settings::RunSettings;

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

@ -5,7 +5,7 @@ mod parse;
mod persist;
mod retro;
mod transform;
pub mod types;
pub(crate) mod types;
mod validate;
pub use execute::execute;
@ -15,8 +15,11 @@ pub use finalize::{
};
pub use initialize::initialize;
pub use parse::parse;
pub use persist::persist;
pub(crate) use persist::persist;
pub use retro::{retro, run_retro};
pub use transform::transform;
pub use types::*;
pub use types::{
Executed, FinalizeOptions, Finalized, InitOptions, Initialized, Parsed, Persisted,
RetroOptions, Retroed, TransformOptions, Transformed, Validated,
};
pub use validate::validate;