diff --git a/lib/crates/fabro-api/src/server.rs b/lib/crates/fabro-api/src/server.rs index 563f4b305..34c6f4ab6 100644 --- a/lib/crates/fabro-api/src/server.rs +++ b/lib/crates/fabro-api/src/server.rs @@ -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, diff --git a/lib/crates/fabro-cli/src/commands/config/mod.rs b/lib/crates/fabro-cli/src/commands/config/mod.rs index c6203c371..d650d94e4 100644 --- a/lib/crates/fabro-cli/src/commands/config/mod.rs +++ b/lib/crates/fabro-cli/src/commands/config/mod.rs @@ -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 { 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()?; diff --git a/lib/crates/fabro-cli/src/commands/graph.rs b/lib/crates/fabro-cli/src/commands/graph.rs index 5e9811141..de700fcf9 100644 --- a/lib/crates/fabro-cli/src/commands/graph.rs +++ b/lib/crates/fabro-cli/src/commands/graph.rs @@ -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 = 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" ); diff --git a/lib/crates/fabro-cli/src/commands/preflight.rs b/lib/crates/fabro-cli/src/commands/preflight.rs index b2d37312d..4deaaa5e9 100644 --- a/lib/crates/fabro-cli/src/commands/preflight.rs +++ b/lib/crates/fabro-cli/src/commands/preflight.rs @@ -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(), diff --git a/lib/crates/fabro-cli/src/commands/run/create.rs b/lib/crates/fabro-cli/src/commands/run/create.rs index c2d0ca882..a8449b7e4 100644 --- a/lib/crates/fabro-cli/src/commands/run/create.rs +++ b/lib/crates/fabro-cli/src/commands/run/create.rs @@ -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, diff --git a/lib/crates/fabro-cli/src/commands/run/fork.rs b/lib/crates/fabro-cli/src/commands/run/fork.rs index 3eca30139..7a4a69f25 100644 --- a/lib/crates/fabro-cli/src/commands/run/fork.rs +++ b/lib/crates/fabro-cli/src/commands/run/fork.rs @@ -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, ¶llel_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, ¶llel_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::) + .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 {} -> {}", diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index 0bef4f084..680cce59d 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -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, ¶llel_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, ¶llel_map)?; + let target = args + .target + .as_deref() + .unwrap() + .parse::()?; - 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, - 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> = 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() { diff --git a/lib/crates/fabro-cli/src/commands/validate.rs b/lib/crates/fabro-cli/src/commands/validate.rs index a2b205b07..5a6b1b666 100644 --- a/lib/crates/fabro-cli/src/commands/validate.rs +++ b/lib/crates/fabro-cli/src/commands/validate.rs @@ -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); diff --git a/lib/crates/fabro-cli/tests/cli.rs b/lib/crates/fabro-cli/tests/cli.rs index 0a0d50b67..66f41e908 100644 --- a/lib/crates/fabro-cli/tests/cli.rs +++ b/lib/crates/fabro-cli/tests/cli.rs @@ -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")); } diff --git a/lib/crates/fabro-config/src/project.rs b/lib/crates/fabro-config/src/project.rs index 86a313bd3..131d91085 100644 --- a/lib/crates/fabro-config/src/project.rs +++ b/lib/crates/fabro-config/src/project.rs @@ -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, } +#[derive(Clone, Debug)] +pub struct WorkflowPathResolution { + pub resolved_workflow_path: PathBuf, + pub dot_path: PathBuf, + pub workflow_config: Option, + pub workflow_toml_path: Option, + pub workflow_slug: Option, +} + +#[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 { + 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 { + 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 { resolve_workflow_arg_from(arg, &start) } +pub fn resolve_workflow_path( + workflow_path: &Path, + cwd: &Path, +) -> anyhow::Result { + 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 { + 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 { 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 { /// loads the run config and resolves the graph path within it. pub fn resolve_workflow(arg: &Path) -> anyhow::Result<(PathBuf, Option)> { 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)> { - 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. diff --git a/lib/crates/fabro-workflows/src/handler/manager_loop.rs b/lib/crates/fabro-workflows/src/handler/manager_loop.rs index e97d17252..51fcba086 100644 --- a/lib/crates/fabro-workflows/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflows/src/handler/manager_loop.rs @@ -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 { + 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 { .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); diff --git a/lib/crates/fabro-workflows/src/operations/create.rs b/lib/crates/fabro-workflows/src/operations/create.rs index bef8886b5..09f746458 100644 --- a/lib/crates/fabro-workflows/src/operations/create.rs +++ b/lib/crates/fabro-workflows/src/operations/create.rs @@ -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, - pub custom_transforms: Vec>, - pub settings: Option, - pub goal_override: Option, -} - -#[derive(Clone, Debug)] -pub struct CreateRequest { +pub struct ValidateInput { pub workflow: WorkflowInput, pub settings: FabroSettings, pub cwd: PathBuf, + pub custom_transforms: Vec>, +} + +#[derive(Clone, Debug)] +pub struct CreateRunInput { + pub workflow: WorkflowInput, + pub settings: FabroSettings, + pub cwd: PathBuf, + pub workflow_slug: Option, pub run_dir: Option, pub run_id: Option, pub host_repo_path: Option, @@ -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 { - 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 { + 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 { - 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 { - let resolved = resolve_workflow(ResolveWorkflowRequest { +pub fn create(request: CreateRunInput) -> Result { + let resolved = resolve_workflow(ResolveWorkflowInput { workflow: request.workflow, settings: request.settings, cwd: request.cwd, @@ -98,10 +91,11 @@ pub fn create(request: CreateRequest) -> Result { 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 { 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, diff --git a/lib/crates/fabro-workflows/src/operations/fork.rs b/lib/crates/fabro-workflows/src/operations/fork.rs index 70128832e..4db57ab05 100644 --- a/lib/crates/fabro-workflows/src/operations/fork.rs +++ b/lib/crates/fabro-workflows/src/operations/fork.rs @@ -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, + 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 { + 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")); } diff --git a/lib/crates/fabro-workflows/src/operations/mod.rs b/lib/crates/fabro-workflows/src/operations/mod.rs index f01c4dae8..ee63a6dfd 100644 --- a/lib/crates/fabro-workflows/src/operations/mod.rs +++ b/lib/crates/fabro-workflows/src/operations/mod.rs @@ -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}; diff --git a/lib/crates/fabro-workflows/src/operations/rewind.rs b/lib/crates/fabro-workflows/src/operations/rewind.rs index 08aaa8019..a96235eb4 100644 --- a/lib/crates/fabro-workflows/src/operations/rewind.rs +++ b/lib/crates/fabro-workflows/src/operations/rewind.rs @@ -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 { + 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::() { + 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, } -pub fn parse_target(s: &str) -> Result { - 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::() { - 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, + pub parallel_map: HashMap, } -pub fn build_timeline(store: &Store, run_id: &str) -> Result> { +#[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 { 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> } 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 { +fn detect_parallel_interior(graph: &Graph) -> HashMap { let mut interior_map = HashMap::new(); for node in graph.nodes.values() { @@ -172,7 +193,7 @@ pub fn detect_parallel_interior(graph: &Graph) -> HashMap { interior_map } -pub fn resolve_target<'a>( +fn resolve_target<'a>( timeline: &'a [TimelineEntry], target: &RewindTarget, parallel_map: &HashMap, @@ -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 } } -pub fn load_parallel_map(store: &Store, run_id: &str) -> HashMap { +fn load_parallel_map(store: &Store, run_id: &str) -> HashMap { 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::().unwrap(), + RewindTarget::Ordinal(4) + ); } #[test] fn parse_target_latest_visit() { assert_eq!( - parse_target("step2").unwrap(), + "step2".parse::().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); diff --git a/lib/crates/fabro-workflows/src/operations/source.rs b/lib/crates/fabro-workflows/src/operations/source.rs index 19d3d455c..7a6dd5ac8 100644 --- a/lib/crates/fabro-workflows/src/operations/source.rs +++ b/lib/crates/fabro-workflows/src/operations/source.rs @@ -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, - workflow_slug: Option, }, } #[derive(Clone, Debug)] -pub struct WorkflowPathResolution { - pub resolved_workflow_path: PathBuf, - pub dot_path: PathBuf, - pub workflow_config: Option, - pub workflow_toml_path: Option, - pub workflow_slug: Option, -} - -#[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, pub workflow_toml_path: Option, pub dot_path: Option, - pub resolved_workflow_path: Option, pub base_dir: Option, pub goal_override: Option, 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 { - 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 { - 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 { - 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 { - 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 { +pub(crate) fn resolve_workflow(request: ResolveWorkflowInput) -> anyhow::Result { 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 anyhow::Result { + 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 exit }".to_string(), base_dir: None, - workflow_slug: None, }, settings: FabroSettings { work_dir: Some("workspace".to_string()), diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs index 5b70df10a..91cc7e4a8 100644 --- a/lib/crates/fabro-workflows/src/operations/start.rs +++ b/lib/crates/fabro-workflows/src/operations/start.rs @@ -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, diff --git a/lib/crates/fabro-workflows/tests/integration.rs b/lib/crates/fabro-workflows/tests/integration.rs index e31c30dfc..aa98f6185 100644 --- a/lib/crates/fabro-workflows/tests/integration.rs +++ b/lib/crates/fabro-workflows/tests/integration.rs @@ -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(), });