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

Refactor workflow operations and config resolution

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-27 17:58:05 -04:00
parent f8cc95e97d
commit 75b5aa35b9
18 changed files with 591 additions and 412 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
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, CreateRequest, WorkflowInput};
use fabro_workflows::operations::{self, CreateRunInput, WorkflowInput};
use fabro_workflows::pipeline::{
self, InitOptions, LlmSpec, Persisted, SandboxEnvSpec, SandboxSpec,
};
@ -526,14 +526,14 @@ async fn start_run(
}),
..Default::default()
};
let created = match operations::create(CreateRequest {
let created = match operations::create(CreateRunInput {
workflow: WorkflowInput::DotSource {
source: req.dot_source.clone(),
base_dir: None,
workflow_slug: None,
},
settings,
cwd: std::env::current_dir().unwrap_or_else(|_| std::env::temp_dir()),
workflow_slug: None,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
host_repo_path: None,

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

@ -2,6 +2,7 @@ use std::io::Write;
use std::path::Path;
use crate::args::{ConfigCommand, ConfigNamespace, ConfigShowArgs};
use fabro_config::project::ResolveSettingsInput;
use fabro_config::{FabroConfig, FabroSettings};
pub fn dispatch(ns: ConfigNamespace) -> anyhow::Result<()> {
@ -13,13 +14,14 @@ pub fn dispatch(ns: ConfigNamespace) -> anyhow::Result<()> {
fn merged_config(workflow: Option<&Path>) -> anyhow::Result<FabroSettings> {
if let Some(workflow) = workflow {
let cli_config = fabro_config::cli::load_cli_config(None)?;
return fabro_workflows::operations::resolve_settings_for_path(
workflow,
cli_config,
FabroConfig::default(),
true,
)
.map_err(Into::into);
let cwd = std::env::current_dir()?;
return fabro_config::project::resolve_settings(ResolveSettingsInput {
workflow_path: workflow.to_path_buf(),
cwd,
defaults: cli_config,
overrides: FabroConfig::default(),
apply_project_config: true,
});
}
let cwd = std::env::current_dir()?;

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

@ -3,6 +3,7 @@ use std::io::Write;
use std::sync::LazyLock;
use anyhow::bail;
use fabro_config::project::ResolveSettingsInput;
use fabro_util::terminal::Styles;
use fabro_validate::Severity;
use tracing::debug;
@ -14,9 +15,23 @@ static RANKDIR_RE: LazyLock<regex::Regex> =
LazyLock::new(|| regex::Regex::new(r"rankdir\s*=\s*\w+").unwrap());
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::validate_from_file(&dot_path)?;
let cwd = std::env::current_dir()?;
let cli_defaults = fabro_config::cli::load_cli_config(None)?;
let settings = fabro_config::project::resolve_settings(ResolveSettingsInput {
workflow_path: args.workflow.clone(),
cwd: cwd.clone(),
defaults: cli_defaults,
overrides: fabro_config::FabroConfig::default(),
apply_project_config: true,
})?;
let resolution = fabro_config::project::resolve_workflow_path(&args.workflow, &cwd)?;
let validated =
fabro_workflows::operations::validate(fabro_workflows::operations::ValidateInput {
workflow: fabro_workflows::operations::WorkflowInput::Path(args.workflow.clone()),
settings,
cwd,
custom_transforms: Vec::new(),
})?;
let diagnostics = validated.diagnostics();
print_diagnostics(diagnostics, styles);
@ -25,7 +40,7 @@ pub fn run(args: &GraphArgs, styles: &Styles) -> anyhow::Result<()> {
bail!("Validation failed");
}
let source = read_workflow_file(&dot_path)?;
let source = read_workflow_file(&resolution.dot_path)?;
let source = apply_direction(&source, args.direction);
let rendered = fabro_graphviz::render::render_dot(&source, args.format.into())?;
@ -36,7 +51,7 @@ pub fn run(args: &GraphArgs, styles: &Styles) -> anyhow::Result<()> {
}
debug!(
path = %relative_path(&dot_path),
path = %relative_path(&resolution.dot_path),
format = %args.format,
"Rendered workflow graph"
);

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

@ -3,6 +3,7 @@ use std::sync::Arc;
use anyhow::bail;
use fabro_agent::{DockerSandbox, DockerSandboxConfig, LocalSandbox, Sandbox};
use fabro_config::project::ResolveSettingsInput;
use fabro_config::{FabroConfig, FabroSettings};
use fabro_model::{Catalog, Provider};
use fabro_sandbox::SandboxProvider;
@ -19,54 +20,49 @@ 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(),
cwd: std::env::current_dir()?,
},
)?;
let cwd = std::env::current_dir()?;
let settings = fabro_config::project::resolve_settings(ResolveSettingsInput {
workflow_path: args.workflow.clone(),
cwd: cwd.clone(),
defaults: cli_defaults,
overrides: cli_args_config,
apply_project_config: true,
})?;
let resolution = fabro_config::project::resolve_workflow_path(&args.workflow, &cwd)?;
let working_directory = fabro_config::project::resolve_working_directory(&settings, &cwd);
let (origin_url, detected_base_branch) =
fabro_sandbox::daytona::detect_repo_info(&resolved.working_directory)
fabro_sandbox::daytona::detect_repo_info(&working_directory)
.map(|(url, branch)| (Some(url), branch))
.unwrap_or((None, None));
let git_status = fabro_workflows::git::sync_status(
&resolved.working_directory,
&working_directory,
"origin",
detected_base_branch.as_deref(),
);
let sandbox_provider = resolve_sandbox_provider(args.sandbox.map(Into::into), &settings)?;
let validated = fabro_workflows::operations::validate(
&resolved.raw_source,
fabro_workflows::operations::ValidateOptions {
base_dir: resolved.base_dir.clone(),
settings: Some(resolved.settings.clone()),
goal_override: resolved.goal_override.clone(),
..Default::default()
},
)?;
super::run::output::print_workflow_report(&validated, resolved.dot_path.as_deref(), styles);
let validated =
fabro_workflows::operations::validate(fabro_workflows::operations::ValidateInput {
workflow: fabro_workflows::operations::WorkflowInput::Path(args.workflow.clone()),
settings: settings.clone(),
cwd,
custom_transforms: Vec::new(),
})?;
super::run::output::print_workflow_report(&validated, Some(&resolution.dot_path), styles);
if validated.has_errors() {
bail!("Validation failed");
}
run_preflight(
validated.graph(),
&resolved.settings,
&settings,
args.model.as_deref(),
args.provider.as_deref(),
git_status,
sandbox_provider,
&resolved.working_directory,
&working_directory,
styles,
github_app,
origin_url.as_deref(),

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

@ -1,6 +1,7 @@
use std::path::PathBuf;
use crate::args::RunArgs;
use fabro_config::project::ResolveSettingsInput;
use fabro_config::{FabroConfig, FabroSettings};
use fabro_util::terminal::Styles;
@ -21,19 +22,21 @@ pub async fn create_run(
.as_ref()
.ok_or_else(|| anyhow::anyhow!("--workflow is required"))?;
let cli_args_config = FabroConfig::try_from(args)?;
let settings: FabroSettings = fabro_workflows::operations::resolve_settings_for_path(
workflow_path,
cli_defaults,
cli_args_config,
true,
)?;
let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let settings: FabroSettings = fabro_config::project::resolve_settings(ResolveSettingsInput {
workflow_path: workflow_path.clone(),
cwd: cwd.clone(),
defaults: cli_defaults,
overrides: cli_args_config,
apply_project_config: true,
})?;
let created =
match fabro_workflows::operations::create(fabro_workflows::operations::CreateRequest {
match fabro_workflows::operations::create(fabro_workflows::operations::CreateRunInput {
workflow: fabro_workflows::operations::WorkflowInput::Path(workflow_path.clone()),
settings,
cwd,
workflow_slug: None,
run_dir: None,
run_id: args.run_id.clone(),
base_branch: None,

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

@ -14,22 +14,23 @@ pub fn run(args: &ForkArgs, styles: &Styles) -> Result<()> {
let timeline = fabro_workflows::operations::build_timeline(&store, &run_id)?;
if args.list {
let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id);
super::rewind::print_timeline(&timeline, &parallel_map, styles);
super::rewind::print_timeline(&timeline, styles);
return Ok(());
}
let entry = if let Some(target_str) = &args.target {
let target = fabro_workflows::operations::parse_target(target_str)?;
let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id);
fabro_workflows::operations::resolve_target(&timeline, &target, &parallel_map)?
} else {
timeline
.last()
.ok_or_else(|| anyhow::anyhow!("no checkpoints found for run {run_id}"))?
};
let new_run_id = fabro_workflows::operations::fork(&store, &run_id, entry, !args.no_push)?;
let target = args
.target
.as_deref()
.map(str::parse::<fabro_workflows::operations::RewindTarget>)
.transpose()?;
let new_run_id = fabro_workflows::operations::fork(
&store,
fabro_workflows::operations::ForkRunInput {
source_run_id: run_id.clone(),
target,
push: !args.no_push,
},
)?;
eprintln!(
"\nForked run {} -> {}",

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

@ -17,16 +17,24 @@ pub fn run(args: &RewindArgs, styles: &Styles) -> Result<()> {
let timeline = fabro_workflows::operations::build_timeline(&store, &run_id)?;
if args.list || args.target.is_none() {
let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id);
print_timeline(&timeline, &parallel_map, styles);
print_timeline(&timeline, styles);
return Ok(());
}
let target = fabro_workflows::operations::parse_target(args.target.as_deref().unwrap())?;
let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id);
let entry = fabro_workflows::operations::resolve_target(&timeline, &target, &parallel_map)?;
let target = args
.target
.as_deref()
.unwrap()
.parse::<fabro_workflows::operations::RewindTarget>()?;
fabro_workflows::operations::rewind(&store, &run_id, entry, !args.no_push)?;
fabro_workflows::operations::rewind(
&store,
fabro_workflows::operations::RewindInput {
run_id: run_id.clone(),
target,
push: !args.no_push,
},
)?;
eprintln!(
"\nTo resume: fabro resume {}",
@ -36,12 +44,8 @@ pub fn run(args: &RewindArgs, styles: &Styles) -> Result<()> {
Ok(())
}
pub(crate) fn print_timeline(
timeline: &[fabro_workflows::operations::TimelineEntry],
parallel_map: &std::collections::HashMap<String, String>,
styles: &Styles,
) {
if timeline.is_empty() {
pub(crate) fn print_timeline(timeline: &fabro_workflows::operations::RunTimeline, styles: &Styles) {
if timeline.entries.is_empty() {
eprintln!("No checkpoints found.");
return;
}
@ -54,6 +58,7 @@ pub(crate) fn print_timeline(
];
let rows: Vec<Vec<CellStruct>> = timeline
.entries
.iter()
.map(|entry| {
let ordinal_str = format!("@{}", entry.ordinal);
@ -61,7 +66,7 @@ pub(crate) fn print_timeline(
if entry.visit > 1 {
details.push(format!("visit {}, loop", entry.visit));
}
if parallel_map.contains_key(&entry.node_name) {
if timeline.parallel_map.contains_key(&entry.node_name) {
details.push("parallel interior".to_string());
}
if entry.run_commit_sha.is_none() {

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

@ -1,4 +1,5 @@
use anyhow::bail;
use fabro_config::project::ResolveSettingsInput;
use fabro_util::terminal::Styles;
use fabro_validate::Severity;
@ -6,9 +7,23 @@ use crate::args::ValidateArgs;
use crate::shared::{print_diagnostics, relative_path};
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::validate_from_file(&dot_path)?;
let cwd = std::env::current_dir()?;
let cli_defaults = fabro_config::cli::load_cli_config(None)?;
let settings = fabro_config::project::resolve_settings(ResolveSettingsInput {
workflow_path: args.workflow.clone(),
cwd: cwd.clone(),
defaults: cli_defaults,
overrides: fabro_config::FabroConfig::default(),
apply_project_config: true,
})?;
let resolution = fabro_config::project::resolve_workflow_path(&args.workflow, &cwd)?;
let validated =
fabro_workflows::operations::validate(fabro_workflows::operations::ValidateInput {
workflow: fabro_workflows::operations::WorkflowInput::Path(args.workflow.clone()),
settings,
cwd,
custom_transforms: Vec::new(),
})?;
let graph = validated.graph();
let diagnostics = validated.diagnostics();
@ -21,7 +36,7 @@ pub fn run(args: &ValidateArgs, styles: &Styles) -> anyhow::Result<()> {
eprintln!(
"{} {}",
styles.dim.apply_to("Graph:"),
styles.dim.apply_to(relative_path(&dot_path)),
styles.dim.apply_to(relative_path(&resolution.dot_path)),
);
print_diagnostics(diagnostics, styles);

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

@ -1043,7 +1043,7 @@ fn start_by_workflow_name_prefers_newly_created_submitted_run() {
serde_json::json!({
"run_id": "old-smoke",
"created_at": "2026-01-01T00:00:00Z",
"config": {},
"settings": {},
"graph": {
"name": "Smoke",
"nodes": {},
@ -1119,7 +1119,7 @@ digraph G {
let run_record = serde_json::json!({
"run_id": "test-bug2",
"created_at": "2026-01-01T00:00:00Z",
"config": {
"settings": {
"dry_run": true,
"auto_approve": true,
"no_retro": true,
@ -1698,5 +1698,5 @@ fn config_show_missing_run_config_errors() {
.args(["config", "show", "missing.toml"])
.assert()
.failure()
.stderr(predicate::str::contains("Failed to read"));
.stderr(predicate::str::contains("Workflow not found"));
}

162
lib/crates/fabro-config/src/project.rs
View file

@ -4,15 +4,37 @@ use anyhow::{bail, Context};
use serde::{Deserialize, Serialize};
use crate::config::FabroConfig;
use crate::run;
use crate::FabroSettings;
const CONFIG_FILENAME: &str = "fabro.toml";
const SUPPORTED_VERSION: u32 = 1;
const RUN_GRAPH_FILE: &str = "workflow.fabro";
const LEGACY_RUN_GRAPH_FILE: &str = "graph.fabro";
#[derive(Clone, Debug, Default, Deserialize, PartialEq, Serialize, crate::Combine)]
pub struct ProjectFabroConfig {
pub root: 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 ResolveSettingsInput {
pub workflow_path: PathBuf,
pub cwd: PathBuf,
pub defaults: FabroConfig,
pub overrides: FabroConfig,
pub apply_project_config: bool,
}
fn default_root() -> String {
".".to_string()
}
@ -79,6 +101,42 @@ pub fn discover_project_config(start: &Path) -> anyhow::Result<Option<(PathBuf,
Ok(None)
}
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())
}
fn cached_workflow_graph_path(path: &Path) -> Option<PathBuf> {
if path.file_name().and_then(|name| name.to_str()) != Some("workflow.toml") {
return None;
}
let canonical = path.with_file_name(RUN_GRAPH_FILE);
if canonical.exists() {
return Some(canonical);
}
let legacy = path.with_file_name(LEGACY_RUN_GRAPH_FILE);
if legacy.exists() {
return Some(legacy);
}
None
}
/// Resolve a workflow argument to a path.
///
/// - If the arg has a file extension (`.toml`, `.fabro`, etc.), return it as-is.
@ -90,6 +148,92 @@ pub fn resolve_workflow_arg(arg: &Path) -> anyhow::Result<PathBuf> {
resolve_workflow_arg_from(arg, &start)
}
pub fn resolve_workflow_path(
workflow_path: &Path,
cwd: &Path,
) -> anyhow::Result<WorkflowPathResolution> {
let path = resolve_workflow_arg_from(workflow_path, cwd)?;
let workflow_slug = workflow_slug_from_path(&path);
if path.extension().is_some_and(|ext| ext == "toml") {
match run::load_run_config(&path) {
Ok(cfg) => {
let dot_path =
run::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() => {
let Some(dot_path) = cached_workflow_graph_path(&path) else {
anyhow::bail!("Workflow not found: {}", path.display());
};
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_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 resolve_settings(input: ResolveSettingsInput) -> anyhow::Result<FabroSettings> {
let resolution = resolve_workflow_path(&input.workflow_path, &input.cwd)?;
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 input.apply_project_config {
discover_project_config(
resolution
.resolved_workflow_path
.parent()
.unwrap_or_else(|| Path::new(".")),
)?
.map(|(_, config)| config)
.unwrap_or_default()
} else {
FabroConfig::default()
};
input
.overrides
.combine(resolution.workflow_config.unwrap_or_default())
.combine(project_config)
.combine(input.defaults)
.try_into()
}
fn resolve_workflow_arg_from(arg: &Path, start_dir: &Path) -> anyhow::Result<PathBuf> {
resolve_workflow_arg_impl(arg, start_dir, user_workflows_dir().as_deref())
}
@ -284,22 +428,8 @@ fn find_closest_match(input: &str, candidates: &[String]) -> Option<String> {
/// loads the run config and resolves the graph path within it.
pub fn resolve_workflow(arg: &Path) -> anyhow::Result<(PathBuf, Option<FabroConfig>)> {
let start = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
resolve_workflow_from(arg, &start)
}
fn resolve_workflow_from(
arg: &Path,
start_dir: &Path,
) -> anyhow::Result<(PathBuf, Option<FabroConfig>)> {
let path = resolve_workflow_arg_from(arg, start_dir)?;
if path.extension().is_some_and(|ext| ext == "toml") {
let cfg = crate::run::load_run_config(&path)?;
let dot =
crate::run::resolve_graph_path(&path, cfg.graph.as_deref().unwrap_or("workflow.fabro"));
Ok((dot, Some(cfg)))
} else {
Ok((path, None))
}
let resolution = resolve_workflow_path(arg, &start)?;
Ok((resolution.dot_path, resolution.workflow_config))
}
/// Check whether retros are enabled in the project config.

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

@ -1,16 +1,17 @@
use std::collections::HashMap;
use std::path::Path;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use fabro_config::FabroSettings;
use crate::condition::evaluate_condition;
use crate::context::keys;
use crate::context::{Context, WorkflowContext};
use crate::error::FabroError;
use crate::operations::{validate, validate_from_file, ValidateOptions};
use crate::operations::{validate, ValidateInput, WorkflowInput};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::pipeline;
use crate::pipeline::types::Initialized;
@ -49,12 +50,22 @@ fn parse_duration_str(s: &str) -> Duration {
/// (with file inlining). `stack.child_workflow` is preferred; `stack.child_dotfile`
/// is kept for backward compatibility.
fn parse_child_graph(node: &Node) -> Result<Graph, FabroError> {
let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
if let Some(dot) = node
.attrs
.get("stack.child_dot_source")
.and_then(|v| v.as_str())
{
let validated = validate(dot, ValidateOptions::default())?;
let validated = validate(ValidateInput {
workflow: WorkflowInput::DotSource {
source: dot.to_string(),
base_dir: None,
},
settings: FabroSettings::default(),
cwd: cwd.clone(),
custom_transforms: Vec::new(),
})?;
validated.raise_on_errors()?;
let (graph, _, _) = validated.into_parts();
return Ok(graph);
@ -65,7 +76,12 @@ 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 = validate_from_file(std::path::Path::new(path))?;
let validated = validate(ValidateInput {
workflow: WorkflowInput::Path(PathBuf::from(path)),
settings: FabroSettings::default(),
cwd,
custom_transforms: Vec::new(),
})?;
validated.raise_on_errors()?;
let (graph, _, _) = validated.into_parts();
return Ok(graph);

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

@ -13,23 +13,23 @@ use crate::pipeline::{self, Persisted, TransformOptions, Validated};
use crate::records::RunRecord;
use crate::transforms::{expand_vars, Transform};
use super::source::{resolve_workflow, ResolveWorkflowRequest, WorkflowInput};
use super::source::{resolve_workflow, ResolveWorkflowInput, WorkflowInput};
const RUN_CONFIG_FILE: &str = "workflow.toml";
#[derive(Default)]
pub struct ValidateOptions {
pub base_dir: Option<PathBuf>,
pub custom_transforms: Vec<Box<dyn Transform>>,
pub settings: Option<FabroSettings>,
pub goal_override: Option<String>,
}
#[derive(Clone, Debug)]
pub struct CreateRequest {
pub struct ValidateInput {
pub workflow: WorkflowInput,
pub settings: FabroSettings,
pub cwd: PathBuf,
pub custom_transforms: Vec<Box<dyn Transform>>,
}
#[derive(Clone, Debug)]
pub struct CreateRunInput {
pub workflow: WorkflowInput,
pub settings: FabroSettings,
pub cwd: PathBuf,
pub workflow_slug: Option<String>,
pub run_dir: Option<PathBuf>,
pub run_id: Option<String>,
pub host_repo_path: Option<String>,
@ -61,33 +61,26 @@ struct PersistCreateOptions {
///
/// Returns `Validated` even when validation produced errors. Call
/// `validated.raise_on_errors()` if the caller wants to fail fast.
pub fn validate(dot_source: &str, options: ValidateOptions) -> Result<Validated, FabroError> {
preprocess_and_validate(
dot_source,
options.base_dir,
options.custom_transforms,
options.settings.as_ref(),
options.goal_override.as_deref(),
)
}
pub fn validate(input: ValidateInput) -> Result<Validated, FabroError> {
let resolved = resolve_workflow(ResolveWorkflowInput {
workflow: input.workflow,
settings: input.settings,
cwd: input.cwd,
})
.map_err(|err| FabroError::Parse(err.to_string()))?;
/// 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("."));
validate(
&source,
ValidateOptions {
base_dir: Some(base_dir.to_path_buf()),
..Default::default()
},
preprocess_and_validate(
&resolved.raw_source,
resolved.base_dir,
input.custom_transforms,
Some(&resolved.settings),
resolved.goal_override.as_deref(),
)
}
/// Resolve workflow inputs, normalize settings, and persist a run directory.
pub fn create(request: CreateRequest) -> Result<CreatedRun, FabroError> {
let resolved = resolve_workflow(ResolveWorkflowRequest {
pub fn create(request: CreateRunInput) -> Result<CreatedRun, FabroError> {
let resolved = resolve_workflow(ResolveWorkflowInput {
workflow: request.workflow,
settings: request.settings,
cwd: request.cwd,
@ -98,10 +91,11 @@ pub fn create(request: CreateRequest) -> Result<CreatedRun, FabroError> {
validate_sandbox_provider(&resolved.settings)?;
}
let CreateRequest {
let CreateRunInput {
workflow: _,
settings: _,
cwd: _,
workflow_slug,
run_dir,
run_id,
host_repo_path,
@ -133,7 +127,7 @@ pub fn create(request: CreateRequest) -> Result<CreatedRun, FabroError> {
settings,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id.clone()),
workflow_slug: resolved.workflow_slug.clone(),
workflow_slug: workflow_slug.or(resolved.workflow_slug.clone()),
labels: resolved.settings.labels.clone(),
base_branch,
working_directory,
@ -329,11 +323,11 @@ pub(crate) fn resolve_run_settings(mut settings: FabroSettings, graph: &Graph) -
settings
}
pub fn default_run_dir(run_id: &str, dry_run: bool) -> PathBuf {
pub(crate) fn default_run_dir(run_id: &str, dry_run: bool) -> PathBuf {
make_run_dir(&crate::run_lookup::default_runs_base(), run_id, dry_run)
}
pub fn make_run_dir(runs_base: &Path, run_id: &str, dry_run: bool) -> PathBuf {
pub(crate) fn make_run_dir(runs_base: &Path, run_id: &str, dry_run: bool) -> PathBuf {
if dry_run {
runs_base.join(format!(
"{}-dry-run-{}",
@ -350,6 +344,19 @@ mod tests {
use super::*;
use fabro_graphviz::graph::AttrValue;
fn validate_dot(dot_source: &str, settings: FabroSettings) -> Validated {
validate(ValidateInput {
workflow: WorkflowInput::DotSource {
source: dot_source.to_string(),
base_dir: None,
},
settings,
cwd: PathBuf::from("."),
custom_transforms: Vec::new(),
})
.unwrap()
}
const MINIMAL_DOT: &str = r#"digraph Test {
graph [goal="Build feature"]
start [shape=Mdiamond]
@ -359,7 +366,7 @@ mod tests {
#[test]
fn validate_minimal() {
let validated = validate(MINIMAL_DOT, ValidateOptions::default()).unwrap();
let validated = validate_dot(MINIMAL_DOT, FabroSettings::default());
validated.raise_on_errors().unwrap();
assert_eq!(validated.graph().name, "Test");
@ -376,7 +383,7 @@ mod tests {
exit [shape=Msquare]
start -> work -> exit
}"#;
let validated = validate(dot, ValidateOptions::default()).unwrap();
let validated = validate_dot(dot, FabroSettings::default());
validated.raise_on_errors().unwrap();
let prompt = validated.graph().nodes["work"]
@ -396,7 +403,7 @@ mod tests {
exit [shape=Msquare]
start -> work -> exit
}"#;
let validated = validate(dot, ValidateOptions::default()).unwrap();
let validated = validate_dot(dot, FabroSettings::default());
validated.raise_on_errors().unwrap();
assert_eq!(
@ -414,18 +421,14 @@ mod tests {
exit [shape=Msquare]
start -> work -> exit
}"#;
let validated = validate(
let validated = validate_dot(
dot,
ValidateOptions {
settings: Some(FabroSettings {
vars: Some(HashMap::from([("who".to_string(), "agent".to_string())])),
..Default::default()
}),
goal_override: Some("override".to_string()),
FabroSettings {
vars: Some(HashMap::from([("who".to_string(), "agent".to_string())])),
goal: Some("override".to_string()),
..Default::default()
},
)
.unwrap();
);
validated.raise_on_errors().unwrap();
assert_eq!(validated.graph().goal(), "override");
@ -439,7 +442,15 @@ mod tests {
#[test]
fn validate_returns_error_on_invalid_dot() {
let result = validate("not a graph", ValidateOptions::default());
let result = validate(ValidateInput {
workflow: WorkflowInput::DotSource {
source: "not a graph".to_string(),
base_dir: None,
},
settings: FabroSettings::default(),
cwd: PathBuf::from("."),
custom_transforms: Vec::new(),
});
assert!(result.is_err());
}
@ -449,7 +460,7 @@ mod tests {
graph [goal="Test"]
work [label="Work"]
}"#;
let validated = validate(dot, ValidateOptions::default()).unwrap();
let validated = validate_dot(dot, FabroSettings::default());
assert!(validated.has_errors());
assert!(validated.raise_on_errors().is_err());
@ -468,13 +479,15 @@ mod tests {
}
}
let validated = validate(
MINIMAL_DOT,
ValidateOptions {
custom_transforms: vec![Box::new(TagTransform)],
..Default::default()
let validated = validate(ValidateInput {
workflow: WorkflowInput::DotSource {
source: MINIMAL_DOT.to_string(),
base_dir: None,
},
)
settings: FabroSettings::default(),
cwd: PathBuf::from("."),
custom_transforms: vec![Box::new(TagTransform)],
})
.unwrap();
validated.raise_on_errors().unwrap();
@ -502,7 +515,13 @@ mod tests {
)
.unwrap();
let validated = validate_from_file(&dot_path).unwrap();
let validated = validate(ValidateInput {
workflow: WorkflowInput::Path(dot_path),
settings: FabroSettings::default(),
cwd: dir.path().to_path_buf(),
custom_transforms: Vec::new(),
})
.unwrap();
validated.raise_on_errors().unwrap();
assert_eq!(validated.graph().goal(), "ship it");
}
@ -514,14 +533,14 @@ mod tests {
work [label="Work"]
}"#;
let dir = tempfile::tempdir().unwrap();
let err = create(CreateRequest {
let err = create(CreateRunInput {
workflow: WorkflowInput::DotSource {
source: dot.to_string(),
base_dir: None,
workflow_slug: None,
},
settings: FabroSettings::default(),
cwd: dir.path().to_path_buf(),
workflow_slug: None,
run_dir: Some(dir.path().join("run")),
run_id: None,
host_repo_path: None,
@ -540,11 +559,10 @@ mod tests {
#[test]
fn create_persists_normalized_config_and_initial_state() {
let dir = tempfile::tempdir().unwrap();
let created = create(CreateRequest {
let created = create(CreateRunInput {
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 {
@ -562,6 +580,7 @@ mod tests {
..Default::default()
},
cwd: dir.path().to_path_buf(),
workflow_slug: Some("slug".to_string()),
run_dir: Some(dir.path().join("run")),
run_id: Some("run-123".to_string()),
host_repo_path: Some(dir.path().display().to_string()),
@ -626,7 +645,7 @@ mod tests {
)
.unwrap();
let created = create(CreateRequest {
let created = create(CreateRunInput {
workflow: WorkflowInput::Path(workflow_dir.join("workflow.toml")),
settings: FabroSettings {
storage_dir: Some(dir.path().join("storage")),
@ -634,6 +653,7 @@ mod tests {
..Default::default()
},
cwd: dir.path().to_path_buf(),
workflow_slug: None,
run_dir: None,
run_id: None,
host_repo_path: None,
@ -653,11 +673,10 @@ mod tests {
let workspace = dir.path().join("workspace");
std::fs::create_dir_all(&workspace).unwrap();
let created = create(CreateRequest {
let created = create(CreateRunInput {
workflow: WorkflowInput::DotSource {
source: MINIMAL_DOT.to_string(),
base_dir: None,
workflow_slug: None,
},
settings: FabroSettings {
work_dir: Some("workspace".to_string()),
@ -665,6 +684,7 @@ mod tests {
..Default::default()
},
cwd: dir.path().to_path_buf(),
workflow_slug: None,
run_dir: Some(dir.path().join("run")),
run_id: Some("run-cwd".to_string()),
host_repo_path: None,

98
lib/crates/fabro-workflows/src/operations/fork.rs
View file

@ -7,12 +7,83 @@ use crate::git::MetadataStore;
use crate::records::RunRecord;
use crate::records::StartRecord;
use super::rewind::TimelineEntry;
use super::rewind::{build_timeline, RewindTarget, RunTimeline, TimelineEntry};
#[derive(Debug, Clone)]
pub struct ForkRunInput {
pub source_run_id: String,
pub target: Option<RewindTarget>,
pub push: bool,
}
/// Create a new run that branches from an existing run at a specific checkpoint.
///
/// Returns the new run ID.
pub fn fork(
pub fn fork(store: &Store, input: ForkRunInput) -> Result<String> {
let timeline = build_timeline(store, &input.source_run_id)?;
let entry = match input.target.as_ref() {
Some(target) => resolve_timeline_entry(&timeline, target)?,
None => timeline.entries.last().ok_or_else(|| {
anyhow::anyhow!("no checkpoints found for run {}", input.source_run_id)
})?,
};
fork_from_entry(store, &input.source_run_id, entry, input.push)
}
fn resolve_timeline_entry<'a>(
timeline: &'a RunTimeline,
target: &RewindTarget,
) -> Result<&'a TimelineEntry> {
match target {
RewindTarget::Ordinal(n) => timeline
.entries
.iter()
.find(|e| e.ordinal == *n)
.ok_or_else(|| {
anyhow::anyhow!(
"ordinal @{n} out of range (max @{})",
timeline.entries.len()
)
}),
RewindTarget::LatestVisit(name) => {
let effective_name = timeline.parallel_map.get(name).unwrap_or(name);
timeline
.entries
.iter()
.rev()
.find(|e| e.node_name == *effective_name)
.ok_or_else(|| {
if effective_name != name {
anyhow::anyhow!(
"node '{name}' is inside parallel '{effective_name}'; \
no checkpoint found for '{effective_name}'"
)
} else {
anyhow::anyhow!("no checkpoint found for node '{name}'")
}
})
}
RewindTarget::SpecificVisit(name, visit) => {
let effective_name = timeline.parallel_map.get(name).unwrap_or(name);
timeline
.entries
.iter()
.find(|e| e.node_name == *effective_name && e.visit == *visit)
.ok_or_else(|| {
if effective_name != name {
anyhow::anyhow!(
"node '{name}' is inside parallel '{effective_name}'; \
no visit {visit} found for '{effective_name}'"
)
} else {
anyhow::anyhow!("no visit {visit} found for node '{name}'")
}
})
}
}
}
fn fork_from_entry(
store: &Store,
source_run_id: &str,
entry: &TimelineEntry,
@ -147,11 +218,12 @@ pub fn fork(
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::str::FromStr;
use super::*;
use git2::Repository;
use crate::operations::{build_timeline, find_run_id_by_prefix, parse_target, resolve_target};
use crate::operations::find_run_id_by_prefix;
fn temp_repo() -> (tempfile::TempDir, Store) {
let dir = tempfile::TempDir::new().unwrap();
@ -183,7 +255,7 @@ mod tests {
let record = serde_json::json!({
"run_id": run_id,
"created_at": "2025-01-01T00:00:00Z",
"config": {},
"settings": {},
"graph": {
"name": "test_workflow",
"nodes": {
@ -264,11 +336,15 @@ mod tests {
let source_run_id = "run-source";
let _run_oids = setup_source_run(&store, source_run_id, &["start", "build", "test"]);
let timeline = build_timeline(&store, source_run_id).unwrap();
let entry =
resolve_target(&timeline, &parse_target("@2").unwrap(), &HashMap::new()).unwrap();
let new_run_id = fork(&store, source_run_id, entry, false).unwrap();
let new_run_id = fork(
&store,
ForkRunInput {
source_run_id: source_run_id.to_string(),
target: Some(RewindTarget::from_str("@2").unwrap()),
push: false,
},
)
.unwrap();
let new_run_branch = format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX);
let new_meta_branch = MetadataStore::branch_name(&new_run_id);
@ -306,7 +382,9 @@ mod tests {
run_commit_sha: None,
};
let err = fork(&store, run_id, &entry, false).unwrap_err().to_string();
let err = fork_from_entry(&store, run_id, &entry, false)
.unwrap_err()
.to_string();
assert!(err.contains("cannot fork"));
}

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

@ -5,17 +5,11 @@ mod source;
mod start;
pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec, SandboxSpec};
pub use create::{
create, default_run_dir, make_run_dir, validate, validate_from_file, CreateRequest, CreatedRun,
ValidateOptions,
};
pub use fork::fork;
pub use create::{create, validate, CreateRunInput, CreatedRun, ValidateInput};
pub use fork::{fork, ForkRunInput};
pub use rewind::{
build_timeline, find_run_id_by_prefix, load_parallel_map, parse_target, resolve_target, rewind,
build_timeline, find_run_id_by_prefix, rewind, RewindInput, RewindTarget, RunTimeline,
TimelineEntry,
};
pub use source::{
resolve_settings_for_path, resolve_workflow, resolve_workflow_path, ResolveWorkflowRequest,
ResolvedWorkflow, WorkflowInput, WorkflowPathResolution,
};
pub use source::WorkflowInput;
pub use start::{resume, start, StartServices, Started};

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

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::str::FromStr;
use anyhow::{bail, Context, Result};
use fabro_git_storage::branchstore::{BranchStore, CommitInfo};
@ -16,6 +17,35 @@ pub enum RewindTarget {
SpecificVisit(String, usize),
}
impl FromStr for RewindTarget {
type Err = anyhow::Error;
fn from_str(s: &str) -> Result<Self> {
if let Some(rest) = s.strip_prefix('@') {
let n: usize = rest
.parse()
.with_context(|| format!("invalid ordinal: @{rest}"))?;
if n == 0 {
bail!("ordinal must be >= 1");
}
return Ok(Self::Ordinal(n));
}
if let Some(at_pos) = s.rfind('@') {
let name = &s[..at_pos];
let visit_str = &s[at_pos + 1..];
if !name.is_empty() && !visit_str.is_empty() {
if let Ok(visit) = visit_str.parse::<usize>() {
if visit == 0 {
bail!("visit number must be >= 1");
}
return Ok(Self::SpecificVisit(name.to_string(), visit));
}
}
}
Ok(Self::LatestVisit(s.to_string()))
}
}
#[derive(Debug, Clone)]
pub struct TimelineEntry {
pub ordinal: usize,
@ -25,32 +55,20 @@ pub struct TimelineEntry {
pub run_commit_sha: Option<String>,
}
pub fn parse_target(s: &str) -> Result<RewindTarget> {
if let Some(rest) = s.strip_prefix('@') {
let n: usize = rest
.parse()
.with_context(|| format!("invalid ordinal: @{rest}"))?;
if n == 0 {
bail!("ordinal must be >= 1");
}
return Ok(RewindTarget::Ordinal(n));
}
if let Some(at_pos) = s.rfind('@') {
let name = &s[..at_pos];
let visit_str = &s[at_pos + 1..];
if !name.is_empty() && !visit_str.is_empty() {
if let Ok(visit) = visit_str.parse::<usize>() {
if visit == 0 {
bail!("visit number must be >= 1");
}
return Ok(RewindTarget::SpecificVisit(name.to_string(), visit));
}
}
}
Ok(RewindTarget::LatestVisit(s.to_string()))
#[derive(Debug, Clone)]
pub struct RunTimeline {
pub entries: Vec<TimelineEntry>,
pub parallel_map: HashMap<String, String>,
}
pub fn build_timeline(store: &Store, run_id: &str) -> Result<Vec<TimelineEntry>> {
#[derive(Debug, Clone)]
pub struct RewindInput {
pub run_id: String,
pub target: RewindTarget,
pub push: bool,
}
pub fn build_timeline(store: &Store, run_id: &str) -> Result<RunTimeline> {
let branch = MetadataStore::branch_name(run_id);
let sig = Signature::now("Fabro", "noreply@fabro.sh")?;
let bs = BranchStore::new(store, &branch, &sig);
@ -87,7 +105,10 @@ pub fn build_timeline(store: &Store, run_id: &str) -> Result<Vec<TimelineEntry>>
}
backfill_run_shas(store, run_id, &mut timeline);
Ok(timeline)
Ok(RunTimeline {
entries: timeline,
parallel_map: load_parallel_map(store, run_id),
})
}
fn backfill_run_shas(store: &Store, run_id: &str, timeline: &mut [TimelineEntry]) {
@ -138,7 +159,7 @@ fn backfill_run_shas(store: &Store, run_id: &str, timeline: &mut [TimelineEntry]
}
}
pub fn detect_parallel_interior(graph: &Graph) -> HashMap<String, String> {
fn detect_parallel_interior(graph: &Graph) -> HashMap<String, String> {
let mut interior_map = HashMap::new();
for node in graph.nodes.values() {
@ -172,7 +193,7 @@ pub fn detect_parallel_interior(graph: &Graph) -> HashMap<String, String> {
interior_map
}
pub fn resolve_target<'a>(
fn resolve_target<'a>(
timeline: &'a [TimelineEntry],
target: &RewindTarget,
parallel_map: &HashMap<String, String>,
@ -218,7 +239,13 @@ pub fn resolve_target<'a>(
}
}
pub fn rewind(store: &Store, run_id: &str, entry: &TimelineEntry, push: bool) -> Result<()> {
pub fn rewind(store: &Store, input: RewindInput) -> Result<()> {
let timeline = build_timeline(store, &input.run_id)?;
let entry = resolve_target(&timeline.entries, &input.target, &timeline.parallel_map)?;
rewind_to_entry(store, &input.run_id, entry, input.push)
}
fn rewind_to_entry(store: &Store, run_id: &str, entry: &TimelineEntry, push: bool) -> Result<()> {
let meta_branch = MetadataStore::branch_name(run_id);
store
.update_ref(&meta_branch, entry.metadata_commit_oid)
@ -313,7 +340,7 @@ pub fn find_run_id_by_prefix(repo: &Repository, prefix: &str) -> Result<String>
}
}
pub fn load_parallel_map(store: &Store, run_id: &str) -> HashMap<String, String> {
fn load_parallel_map(store: &Store, run_id: &str) -> HashMap<String, String> {
let branch = MetadataStore::branch_name(run_id);
let sig = match Signature::now("Fabro", "noreply@fabro.sh") {
Ok(s) => s,
@ -374,13 +401,16 @@ mod tests {
#[test]
fn parse_target_ordinal() {
assert_eq!(parse_target("@4").unwrap(), RewindTarget::Ordinal(4));
assert_eq!(
"@4".parse::<RewindTarget>().unwrap(),
RewindTarget::Ordinal(4)
);
}
#[test]
fn parse_target_latest_visit() {
assert_eq!(
parse_target("step2").unwrap(),
"step2".parse::<RewindTarget>().unwrap(),
RewindTarget::LatestVisit("step2".to_string())
);
}
@ -402,41 +432,44 @@ mod tests {
.unwrap();
let timeline = build_timeline(&store, "test-run-1").unwrap();
assert_eq!(timeline.len(), 2);
assert_eq!(timeline[0].node_name, "start");
assert_eq!(timeline[1].node_name, "build");
assert_eq!(timeline.entries.len(), 2);
assert_eq!(timeline.entries[0].node_name, "start");
assert_eq!(timeline.entries[1].node_name, "build");
}
#[test]
fn resolve_latest_visit() {
let timeline = vec![
TimelineEntry {
ordinal: 1,
node_name: "start".to_string(),
visit: 1,
metadata_commit_oid: Oid::zero(),
run_commit_sha: Some("aaa".to_string()),
},
TimelineEntry {
ordinal: 2,
node_name: "build".to_string(),
visit: 1,
metadata_commit_oid: Oid::zero(),
run_commit_sha: Some("bbb".to_string()),
},
TimelineEntry {
ordinal: 3,
node_name: "build".to_string(),
visit: 2,
metadata_commit_oid: Oid::zero(),
run_commit_sha: Some("ccc".to_string()),
},
];
let timeline = RunTimeline {
entries: vec![
TimelineEntry {
ordinal: 1,
node_name: "start".to_string(),
visit: 1,
metadata_commit_oid: Oid::zero(),
run_commit_sha: Some("aaa".to_string()),
},
TimelineEntry {
ordinal: 2,
node_name: "build".to_string(),
visit: 1,
metadata_commit_oid: Oid::zero(),
run_commit_sha: Some("bbb".to_string()),
},
TimelineEntry {
ordinal: 3,
node_name: "build".to_string(),
visit: 2,
metadata_commit_oid: Oid::zero(),
run_commit_sha: Some("ccc".to_string()),
},
],
parallel_map: HashMap::new(),
};
let entry = resolve_target(
&timeline,
&timeline.entries,
&RewindTarget::LatestVisit("build".to_string()),
&HashMap::new(),
&timeline.parallel_map,
)
.unwrap();
assert_eq!(entry.ordinal, 3);
@ -499,8 +532,15 @@ mod tests {
bs.write_entry("checkpoint.json", &cp2, "checkpoint")
.unwrap();
let timeline = build_timeline(&store, "run-1").unwrap();
rewind(&store, "run-1", &timeline[0], false).unwrap();
rewind(
&store,
RewindInput {
run_id: "run-1".to_string(),
target: RewindTarget::Ordinal(1),
push: false,
},
)
.unwrap();
let resolved = store.resolve_ref(&branch).unwrap().unwrap();
assert_eq!(resolved, oid1);

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

@ -1,10 +1,7 @@
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";
use fabro_config::{project as project_config, FabroSettings};
#[derive(Clone, Debug)]
pub enum WorkflowInput {
@ -12,34 +9,23 @@ pub enum WorkflowInput {
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(crate) struct ResolveWorkflowInput {
pub workflow: WorkflowInput,
pub settings: FabroSettings,
pub cwd: PathBuf,
}
#[derive(Clone, Debug)]
pub struct ResolvedWorkflow {
pub(crate) 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,
@ -64,137 +50,13 @@ fn resolve_goal_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())
}
fn cached_workflow_graph_path(path: &Path) -> Option<PathBuf> {
if path.file_name().and_then(|name| name.to_str()) != Some("workflow.toml") {
return None;
}
let canonical = path.with_file_name(RUN_GRAPH_FILE);
if canonical.exists() {
return Some(canonical);
}
let legacy = path.with_file_name(LEGACY_RUN_GRAPH_FILE);
if legacy.exists() {
return Some(legacy);
}
None
}
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() => {
let Some(dot_path) = cached_workflow_graph_path(&path) else {
anyhow::bail!("Workflow not found: {}", path.display());
};
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> {
pub(crate) fn resolve_workflow(request: ResolveWorkflowInput) -> anyhow::Result<ResolvedWorkflow> {
match request.workflow {
WorkflowInput::Path(workflow_path) => {
let resolution = resolve_workflow_path(&workflow_path)?;
let resolution = project_config::resolve_workflow_path(&workflow_path, &request.cwd)?;
let settings = request.settings;
let working_directory = resolve_working_directory(&settings, &request.cwd);
let working_directory =
project_config::resolve_working_directory(&settings, &request.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(
@ -208,7 +70,6 @@ pub fn resolve_workflow(request: ResolveWorkflowRequest) -> anyhow::Result<Resol
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
@ -220,13 +81,10 @@ pub fn resolve_workflow(request: ResolveWorkflowRequest) -> anyhow::Result<Resol
working_directory,
})
}
WorkflowInput::DotSource {
source,
base_dir,
workflow_slug,
} => {
WorkflowInput::DotSource { source, base_dir } => {
let settings = request.settings;
let working_directory = resolve_working_directory(&settings, &request.cwd);
let working_directory =
project_config::resolve_working_directory(&settings, &request.cwd);
let goal_override = settings.goal.clone().or(resolve_goal_file(
settings.goal_file.as_deref(),
&working_directory,
@ -234,10 +92,9 @@ pub fn resolve_workflow(request: ResolveWorkflowRequest) -> anyhow::Result<Resol
Ok(ResolvedWorkflow {
raw_source: source,
settings,
workflow_slug,
workflow_slug: None,
workflow_toml_path: None,
dot_path: None,
resolved_workflow_path: None,
base_dir,
goal_override,
working_directory,
@ -251,7 +108,7 @@ mod tests {
use super::*;
#[test]
fn resolve_workflow_path_uses_cached_graph_sibling_for_missing_workflow_toml() {
fn resolve_workflow_uses_cached_graph_sibling_for_missing_workflow_toml() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir
.path()
@ -265,21 +122,28 @@ mod tests {
)
.unwrap();
let resolution = resolve_workflow_path(&run_dir.join("workflow.toml")).unwrap();
let resolved = resolve_workflow(ResolveWorkflowInput {
workflow: WorkflowInput::Path(run_dir.join("workflow.toml")),
settings: FabroSettings::default(),
cwd: dir.path().to_path_buf(),
})
.unwrap();
assert_eq!(resolution.dot_path, run_dir.join("workflow.fabro"));
assert!(resolution.workflow_config.is_none());
assert!(resolution.workflow_toml_path.is_none());
let expected_dot_path = run_dir.join("workflow.fabro");
assert_eq!(
resolved.dot_path.as_deref(),
Some(expected_dot_path.as_path())
);
assert!(resolved.workflow_toml_path.is_none());
}
#[test]
fn resolve_workflow_uses_explicit_cwd_for_relative_work_dir() {
let dir = tempfile::tempdir().unwrap();
let resolved = resolve_workflow(ResolveWorkflowRequest {
let resolved = resolve_workflow(ResolveWorkflowInput {
workflow: WorkflowInput::DotSource {
source: "digraph Test { start -> exit }".to_string(),
base_dir: None,
workflow_slug: None,
},
settings: FabroSettings {
work_dir: Some("workspace".to_string()),

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

@ -775,11 +775,10 @@ mod tests {
}"#;
fn persisted_workflow(dot: &str, run_dir: &Path) -> Persisted {
crate::operations::create(crate::operations::CreateRequest {
crate::operations::create(crate::operations::CreateRunInput {
workflow: crate::operations::WorkflowInput::DotSource {
source: dot.to_string(),
base_dir: None,
workflow_slug: Some("test".to_string()),
},
settings: FabroSettings {
dry_run: Some(true),
@ -789,6 +788,7 @@ mod tests {
.parent()
.unwrap_or_else(|| Path::new("."))
.to_path_buf(),
workflow_slug: Some("test".to_string()),
run_dir: Some(run_dir.to_path_buf()),
run_id: Some("run-test".to_string()),
host_repo_path: None,

2
lib/crates/fabro-workflows/tests/integration.rs
View file

@ -10609,7 +10609,7 @@ async fn git_checkpoint_host_writes_shadow_branch() {
let run_record_json = serde_json::json!({
"run_id": run_id,
"created_at": "2025-01-01T00:00:00Z",
"config": {},
"settings": {},
"graph": { "name": "ShadowBranchTest", "nodes": {}, "edges": [], "attrs": {} },
"working_directory": worktree_path.to_str().unwrap(),
});