diff --git a/lib/crates/fabro-api/src/server.rs b/lib/crates/fabro-api/src/server.rs index 26eb75cbf..e2b1d9259 100644 --- a/lib/crates/fabro-api/src/server.rs +++ b/lib/crates/fabro-api/src/server.rs @@ -23,9 +23,10 @@ use crate::jwt_auth::{AuthMode, AuthenticatedService, AuthenticatedUser}; use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer}; use fabro_workflows::checkpoint::Checkpoint; use fabro_workflows::context::Context; -use fabro_workflows::engine::{RunSettings, WorkflowRunEngine}; +use fabro_workflows::engine::WorkflowRunEngine; use fabro_workflows::event::{EventEmitter, WorkflowRunEvent}; use fabro_workflows::handler::HandlerRegistry; +use fabro_workflows::run_settings::RunSettings; pub use fabro_types::{ ApiQuestion, ApiQuestionOption, PaginatedRunList, PaginationMeta, diff --git a/lib/crates/fabro-cli/src/commands/diff.rs b/lib/crates/fabro-cli/src/commands/diff.rs index 6c294b695..13457a1c2 100644 --- a/lib/crates/fabro-cli/src/commands/diff.rs +++ b/lib/crates/fabro-cli/src/commands/diff.rs @@ -108,8 +108,8 @@ fn build_live_diff_cmd(base_sha: &str, stat: bool, shortstat: bool) -> String { ); format!( "{} add -N . && {} diff{flags} {quoted_sha}", - fabro_workflows::engine::GIT_REMOTE, - fabro_workflows::engine::GIT_REMOTE + fabro_workflows::sandbox_git::GIT_REMOTE, + fabro_workflows::sandbox_git::GIT_REMOTE ) } diff --git a/lib/crates/fabro-cli/src/commands/fork.rs b/lib/crates/fabro-cli/src/commands/fork.rs index 06436c778..313992a0d 100644 --- a/lib/crates/fabro-cli/src/commands/fork.rs +++ b/lib/crates/fabro-cli/src/commands/fork.rs @@ -24,29 +24,28 @@ pub struct ForkArgs { pub fn run(args: &ForkArgs, styles: &Styles) -> Result<()> { let repo = Repository::discover(".").context("not in a git repository")?; - let run_id = fabro_workflows::run_rewind::find_run_id_by_prefix(&repo, &args.run_id)?; + let run_id = fabro_workflows::operations::find_run_id_by_prefix(&repo, &args.run_id)?; let store = Store::new(repo); - let timeline = fabro_workflows::run_rewind::build_timeline(&store, &run_id)?; + let timeline = fabro_workflows::operations::build_timeline(&store, &run_id)?; if args.list { - let parallel_map = fabro_workflows::run_rewind::load_parallel_map(&store, &run_id); + let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id); super::rewind::print_timeline(&timeline, ¶llel_map, styles); return Ok(()); } let entry = if let Some(target_str) = &args.target { - let target = fabro_workflows::run_rewind::parse_target(target_str)?; - let parallel_map = fabro_workflows::run_rewind::load_parallel_map(&store, &run_id); - fabro_workflows::run_rewind::resolve_target(&timeline, &target, ¶llel_map)? + 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::run_fork::execute_fork(&store, &run_id, entry, !args.no_push)?; + let new_run_id = fabro_workflows::operations::fork(&store, &run_id, entry, !args.no_push)?; eprintln!( "\nForked run {} -> {}", diff --git a/lib/crates/fabro-cli/src/commands/graph.rs b/lib/crates/fabro-cli/src/commands/graph.rs index 9f362aaaf..a681f7d0b 100644 --- a/lib/crates/fabro-cli/src/commands/graph.rs +++ b/lib/crates/fabro-cli/src/commands/graph.rs @@ -57,7 +57,8 @@ static RANKDIR_RE: LazyLock = pub fn run(args: &GraphArgs, styles: &Styles) -> anyhow::Result<()> { let (dot_path, _cfg) = fabro_config::project::resolve_workflow(&args.workflow)?; - let (_graph, diagnostics) = fabro_workflows::workflow::prepare_from_file(&dot_path)?; + let validated = fabro_workflows::operations::create_from_file(&dot_path)?; + let diagnostics = validated.diagnostics(); print_diagnostics(&diagnostics, styles); diff --git a/lib/crates/fabro-cli/src/commands/resume.rs b/lib/crates/fabro-cli/src/commands/resume.rs index 7026bcd06..1fa293278 100644 --- a/lib/crates/fabro-cli/src/commands/resume.rs +++ b/lib/crates/fabro-cli/src/commands/resume.rs @@ -1,35 +1,38 @@ use std::io::IsTerminal; use std::path::PathBuf; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Instant; use anyhow::{bail, Context}; use clap::Args; use fabro_agent::{DockerSandbox, DockerSandboxConfig, Sandbox, WorktreeConfig, WorktreeSandbox}; use fabro_config::config::FabroConfig; -use fabro_graphviz::graph::Graph; use fabro_interview::{AutoApproveInterviewer, ConsoleInterviewer, Interviewer}; use fabro_model::{Catalog, Provider}; use fabro_util::terminal::Styles; use fabro_workflows::backend::{AgentApiBackend, AgentCliBackend, BackendRouter}; use fabro_workflows::checkpoint::Checkpoint; -use fabro_workflows::engine::{GitCheckpointSettings, RunSettings}; use fabro_workflows::event::{EventEmitter, RunNoticeLevel}; +use fabro_workflows::operations::{ + create_from_graph, start, StartFinalizeConfig, StartOptions, StartRetroConfig, +}; use fabro_workflows::outcome::StageStatus; +use fabro_workflows::pipeline::{ + build_conclusion, classify_engine_result, persist_terminal_outcome, +}; use fabro_workflows::run_record::RunRecord; +use fabro_workflows::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings}; use fabro_workflows::sandbox_provider::SandboxProvider; -use indicatif::HumanDuration; use super::detached_support::{DetachedRunBootstrapGuard, DetachedRunCompletionGuard}; use super::run::{ - build_conclusion, build_event_envelope, cached_graph_path, classify_engine_result, - default_run_dir, emit_run_notice, generate_retro, local_sandbox_with_callback, - mint_github_token, persist_terminal_outcome, prepare_workflow_with_project_config, - print_assets, print_final_output, resolve_daytona_config, resolve_fallback_chain, - resolve_model_provider, resolve_ssh_clone_params, resolve_ssh_config, write_finalize_commit, - write_run_config_snapshot, CliSandboxProvider, RunArgs, + build_event_envelope, cached_graph_path, default_run_dir, emit_run_notice, + local_sandbox_with_callback, mint_github_token, prepare_workflow_with_project_config, + print_assets, print_final_output, print_retro_result, print_run_conclusion, + resolve_daytona_config, resolve_fallback_chain, resolve_model_provider, + resolve_ssh_clone_params, resolve_ssh_config, write_run_config_snapshot, CliSandboxProvider, + RunArgs, }; -use crate::commands::shared::tilde_path; use fabro_config::project as project_config; use fabro_config::run as run_config; use fabro_workflows::devcontainer_bridge; @@ -94,7 +97,7 @@ pub struct ResumeArgs { /// Intermediate state produced by the two resolution paths (checkpoint-file vs. git-branch). struct ResumeContext { checkpoint: Checkpoint, - graph: Graph, + validated: fabro_workflows::pipeline::Validated, run_id: String, run_dir: PathBuf, run_cfg: Option, @@ -201,7 +204,9 @@ async fn prepare_from_checkpoint( true, false, )?; - let (graph, source, _diagnostics) = prepared.validated.into_parts(); + let source = prepared.raw_source.clone(); + let validated = prepared.validated; + let graph = validated.graph().clone(); let run_cfg = prepared.run_cfg; let sandbox_provider = prepared.sandbox_provider; let workflow_slug = prepared.workflow_slug; @@ -473,7 +478,7 @@ async fn prepare_from_checkpoint( Ok(ResumeContext { checkpoint, - graph, + validated, run_id, run_dir, run_cfg, @@ -509,7 +514,7 @@ async fn prepare_from_branch( (stripped.to_string(), run_arg.to_string()) } else { let repo = git2::Repository::discover(".").context("not in a git repository")?; - let id = fabro_workflows::run_rewind::find_run_id_by_prefix(&repo, run_arg)?; + let id = fabro_workflows::operations::find_run_id_by_prefix(&repo, run_arg)?; let branch = format!("{}{}", fabro_workflows::git::RUN_BRANCH_PREFIX, id); (id, branch) }; @@ -544,7 +549,7 @@ async fn prepare_from_branch( .or_else(|| repo_info.as_ref().and_then(|(_, branch)| branch.clone())); let base_sha = start_record.as_ref().and_then(|s| s.base_sha.clone()); - let (graph, graph_source, run_cfg, mut sandbox_provider, workflow_slug) = + let (validated, graph_source, run_cfg, mut sandbox_provider, workflow_slug) = if let Some(ref workflow_path) = args.workflow { let prepared = prepare_workflow_with_project_config( &resume_as_run_args(args, workflow_path.clone()), @@ -553,19 +558,16 @@ async fn prepare_from_branch( true, false, )?; - { - let (graph, source, _diagnostics) = prepared.validated.into_parts(); - ( - graph, - source, - prepared.run_cfg, - prepared.sandbox_provider, - prepared.workflow_slug, - ) - } + ( + prepared.validated, + prepared.raw_source, + prepared.run_cfg, + prepared.sandbox_provider, + prepared.workflow_slug, + ) } else if let Some(ref rec) = record { // Use the fully transformed graph from the RunRecord - let graph = rec.graph.clone(); + let validated = create_from_graph(rec.graph.clone(), String::new()); let source = String::new(); // no DOT source needed — graph is from RunRecord let run_cfg = Some(rec.config.clone()); let sandbox_provider = if args.dry_run { @@ -581,7 +583,7 @@ async fn prepare_from_branch( args.sandbox.map(Into::into).unwrap_or(sp) }; ( - graph, + validated, source, run_cfg, sandbox_provider, @@ -590,6 +592,7 @@ async fn prepare_from_branch( } else { bail!("no run.json found on metadata branch for run {run_id}"); }; + let graph = validated.graph().clone(); eprintln!( "{} {} from branch {} ({})", @@ -903,7 +906,7 @@ async fn prepare_from_branch( Ok(ResumeContext { checkpoint, - graph, + validated, run_id, run_dir, run_cfg, @@ -931,7 +934,7 @@ async fn run_resumed( ) -> anyhow::Result<()> { let ResumeContext { checkpoint, - graph, + validated, run_id, run_dir, mut run_cfg, @@ -948,21 +951,7 @@ async fn run_resumed( github_app, mut status_guard, } = ctx; - - // Track the last git commit SHA from CheckpointCompleted events - let last_git_sha: Arc>> = Arc::new(std::sync::Mutex::new(None)); - { - let sha_clone = Arc::clone(&last_git_sha); - emitter.on_event(move |event| { - if let fabro_workflows::event::WorkflowRunEvent::CheckpointCompleted { - git_commit_sha: Some(sha), - .. - } = event - { - *sha_clone.lock().unwrap() = Some(sha.clone()); - } - }); - } + let graph = validated.graph().clone(); // Create progress UI (verbose mode shows detailed turn/tool counts and token usage) let is_tty = std::io::stderr().is_terminal(); @@ -1238,34 +1227,7 @@ async fn run_resumed( } } }); - let mut engine = fabro_workflows::engine::WorkflowRunEngine::with_interviewer( - registry, - Arc::clone(&emitter), - interviewer, - Arc::clone(&sandbox), - ); - if !sandbox_env.is_empty() { - engine.set_env(sandbox_env); - } - if dry_run_mode { - engine.set_dry_run(true); - } - // Wire up hook runner from run defaults (mirrors run_command) - { - let hooks = run_cfg - .as_ref() - .map(|cfg| &cfg.hooks) - .unwrap_or(&run_defaults.hooks); - if !hooks.is_empty() { - let hook_config = fabro_hooks::HookConfig { - hooks: hooks.clone(), - }; - let runner = fabro_hooks::HookRunner::new(hook_config); - engine.set_hook_runner(Arc::new(runner)); - } - } - - let lifecycle = fabro_workflows::engine::LifecycleConfig { + let lifecycle = LifecycleConfig { setup_commands, setup_command_timeout_ms: 300_000, devcontainer_phases, @@ -1274,29 +1236,56 @@ async fn run_resumed( // Defuse the bootstrap guard — engine.run() has taken ownership of lifecycle status. status_guard.defuse(); - // Safety net: if we panic or return early, best-effort cleanup via spawn (mirrors run_command). let preserve = super::run::resolve_preserve_sandbox( args.preserve_sandbox, run_cfg.as_ref(), &run_defaults, ); - let sandbox_for_cleanup = Arc::clone(&sandbox); - let cleanup_guard = scopeguard::guard((), move |()| { - if preserve { - return; - } - let rt = tokio::runtime::Handle::try_current(); - if let Ok(handle) = rt { - handle.spawn(async move { - let _ = sandbox_for_cleanup.cleanup().await; - }); - } - }); - let run_start = Instant::now(); - let engine_result = engine - .run_with_lifecycle(&graph, &mut settings, lifecycle, Some(&checkpoint)) - .await; + let pr_config = settings.pull_request().cloned(); + let started = start( + validated, + StartOptions { + init: fabro_workflows::pipeline::InitOptions { + run_id: run_id.clone(), + run_dir: run_dir.clone(), + dry_run: dry_run_mode, + emitter: Arc::clone(&emitter), + sandbox: Arc::clone(&sandbox), + registry: Arc::new(registry), + lifecycle, + run_settings: settings, + hooks: fabro_hooks::HookConfig { + hooks: run_cfg + .as_ref() + .map(|cfg| cfg.hooks.clone()) + .unwrap_or_else(|| run_defaults.hooks.clone()), + }, + sandbox_env, + checkpoint: Some(checkpoint), + seed_context: None, + }, + retro: StartRetroConfig { + enabled: !args.no_retro && project_config::is_retro_enabled(), + dry_run: dry_run_mode, + llm_client: if dry_run_mode { + None + } else { + fabro_llm::client::Client::from_env().await.ok() + }, + provider: provider_enum, + model: model.clone(), + }, + finalize: StartFinalizeConfig { + preserve_sandbox: preserve, + pr_config, + github_app: github_app.clone(), + origin_url: origin_url.clone(), + model: model.clone(), + }, + }, + ) + .await; let run_duration_ms = run_start.elapsed().as_millis() as u64; let mut completion_guard = DetachedRunCompletionGuard::arm(&run_dir); @@ -1305,263 +1294,45 @@ async fn run_resumed( let _ = std::env::set_current_dir(cwd); } - let (final_status, failure_reason, run_status, status_reason) = - classify_engine_result(&engine_result); - let conclusion = build_conclusion( - &run_dir, - final_status.clone(), - failure_reason, - run_duration_ms, - last_git_sha.lock().unwrap().clone(), - ); - - // Auto-derive retro - if !args.no_retro && project_config::is_retro_enabled() { - let failed = match &engine_result { - Ok(ref o) => o.status == StageStatus::Fail, - Err(_) => true, - }; - - let llm_client = if dry_run_mode { - None - } else { - fabro_llm::client::Client::from_env().await.ok() - }; - - generate_retro( - &settings.run_id, - &graph.name, - graph.goal(), - &run_dir, - failed, - run_duration_ms, - dry_run_mode, - llm_client.as_ref(), - &sandbox, - provider_enum, - &model, - styles, - Some(Arc::clone(&emitter)), - ) - .await; - } - - // Finish progress bars after retro (retro stage uses the same ProgressUI) progress_ui.lock().expect("progress lock poisoned").finish(); - - // Write finalize commit with retro.json + final node files (captures last diff.patch) - write_finalize_commit(&settings, &run_dir).await; - - // Auto-create PR on successful completion (mirrors run_command) - let mut pushed_branch: Option = None; - let mut pr_url: Option = None; - if let Some(pr_cfg) = settings.pull_request() { - if settings.dry_run { - debug!("Skipping PR creation: dry-run mode"); - } else if let Err(ref e) = engine_result { - debug!(error = %e, "Skipping PR creation: engine returned an error"); - } else if let Ok(ref outcome) = engine_result { - if !matches!( - outcome.status, - StageStatus::Success | StageStatus::PartialSuccess - ) { - debug!(status = ?outcome.status, "Skipping PR creation: run status is not success"); - } else { - let diff = tokio::fs::read_to_string(run_dir.join("final.patch")) - .await - .unwrap_or_default(); - if let ( - Some(ref base_branch), - Some(ref run_branch), - Some(ref creds), - Some(ref origin), - ) = ( - &settings.base_branch, - settings.git.as_ref().and_then(|g| g.run_branch.as_ref()), - &github_app, - &origin_url, - ) { - if settings.git.is_some() { - pushed_branch = Some(run_branch.to_string()); - } - - let auto_merge = if pr_cfg.auto_merge { - Some(fabro_workflows::pull_request::AutoMergeConfig { - merge_strategy: pr_cfg.merge_strategy, - }) - } else { - None - }; - - match fabro_workflows::pull_request::maybe_open_pull_request( - creds, - origin, - base_branch, - run_branch, - graph.goal(), - &diff, - &model, - pr_cfg.draft, - auto_merge, - &run_dir, - ) - .await - { - Ok(Some(record)) => { - emitter.emit( - &fabro_workflows::event::WorkflowRunEvent::PullRequestCreated { - pr_url: record.html_url.clone(), - pr_number: record.number, - draft: pr_cfg.draft, - }, - ); - pr_url = Some(record.html_url.clone()); - if let Err(e) = record.save(&run_dir.join("pull_request.json")) { - tracing::warn!(error = %e, "Failed to save pull_request.json"); - } - } - Ok(None) => {} // empty diff, logged at DEBUG - Err(e) => { - emitter.emit( - &fabro_workflows::event::WorkflowRunEvent::PullRequestFailed { - error: e.to_string(), - }, - ); - emit_run_notice( - &emitter, - RunNoticeLevel::Warn, - "pull_request_failed", - format!("PR creation failed: {e}"), - ); - } - } - } + let final_status = match started { + Ok(started) => { + if let Some(ref retro) = started.retro { + print_retro_result(retro, started.retro_duration, &run_dir, styles); } - } - } else { - debug!("Skipping PR creation: pull_request not enabled in config"); - } - - // Defuse the cleanup guard — we are about to do explicit cleanup - scopeguard::ScopeGuard::into_inner(cleanup_guard); - - // Cleanup sandbox via engine (fires SandboxCleanup hook) - // Before cleanup, print preserve banner (mirrors run_command) - if preserve { - let info = sandbox.sandbox_info(); - if !info.is_empty() { - emit_run_notice( - &emitter, - RunNoticeLevel::Info, - "sandbox_preserved", - format!("sandbox preserved: {info}"), - ); - } else { - emit_run_notice( - &emitter, - RunNoticeLevel::Info, - "sandbox_preserved", - "sandbox preserved", + let finalized = started.finalized; + print_run_conclusion( + &finalized.conclusion, + &run_id, + &run_dir, + finalized.pushed_branch.as_deref(), + finalized.pr_url.as_deref(), + styles, ); + print_final_output(&run_dir, styles); + print_assets(&run_dir, styles); + finalized.conclusion.status.clone() + } + Err(err) => { + let engine_result: Result = Err(err.clone()); + let (final_status, failure_reason, run_status, status_reason) = + classify_engine_result(&engine_result); + let conclusion = build_conclusion( + &run_dir, + final_status.clone(), + failure_reason, + run_duration_ms, + None, + ); + persist_terminal_outcome(&run_dir, &conclusion, run_status, status_reason); + print_run_conclusion(&conclusion, &run_id, &run_dir, None, None, styles); + print_final_output(&run_dir, styles); + print_assets(&run_dir, styles); + final_status } - } - if let Err(e) = engine - .cleanup_sandbox(&settings.run_id, &graph.name, preserve) - .await - { - tracing::warn!(error = %e, "Sandbox cleanup failed"); - emit_run_notice( - &emitter, - RunNoticeLevel::Warn, - "sandbox_cleanup_failed", - format!("sandbox cleanup failed: {e}"), - ); - } - - persist_terminal_outcome(&run_dir, &conclusion, run_status, status_reason); - completion_guard.defuse(); - - eprintln!("\n{}", styles.bold.apply_to("=== Run Result ===")); - eprintln!("{}", styles.dim.apply_to(format!("Run: {run_id}"))); - let status_str = final_status.to_string().to_uppercase(); - let status_color = match final_status { - StageStatus::Success | StageStatus::PartialSuccess => &styles.bold_green, - _ => &styles.bold_red, }; - eprintln!("Status: {}", status_color.apply_to(&status_str)); - eprintln!( - "Duration: {}", - HumanDuration(Duration::from_millis(run_duration_ms)) - ); - { - use crate::commands::shared::format_tokens_human; - use fabro_workflows::cost::format_cost; - let acc = accumulator.lock().unwrap(); - let total_tokens = acc.total_input_tokens + acc.total_output_tokens; - if total_tokens > 0 { - if acc.has_pricing { - eprintln!( - "{}", - styles.dim.apply_to(format!( - "Cost: {} ({} toks)", - format_cost(acc.total_cost), - format_tokens_human(total_tokens) - )) - ); - } else { - eprintln!( - "{}", - styles - .dim - .apply_to(format!("Toks: {}", format_tokens_human(total_tokens))) - ); - } - if acc.total_cache_read_tokens > 0 { - eprintln!( - "{}", - styles.dim.apply_to(format!( - "Cache: {} read, {} write", - format_tokens_human(acc.total_cache_read_tokens), - format_tokens_human(acc.total_cache_write_tokens), - )), - ); - } - if acc.total_reasoning_tokens > 0 { - eprintln!( - "{}", - styles.dim.apply_to(format!( - "Reasoning: {} tokens", - format_tokens_human(acc.total_reasoning_tokens), - )), - ); - } - } - } - - eprintln!( - "{}", - styles - .dim - .apply_to(format!("Run: {}", tilde_path(&run_dir))) - ); - - if let Some(failure) = conclusion.failure_reason.as_deref() { - eprintln!("Failure: {}", styles.red.apply_to(failure)); - } - - if pushed_branch.is_some() || pr_url.is_some() { - eprintln!(); - if let Some(ref branch) = pushed_branch { - eprintln!("{} {branch}", styles.bold.apply_to("Pushed branch:")); - } - if let Some(ref url) = pr_url { - eprintln!("{} {url}", styles.bold.apply_to("Pull request:")); - } - } - - print_final_output(&run_dir, styles); - print_assets(&run_dir, styles); + completion_guard.defuse(); fabro_util::run_log::deactivate(); match final_status { diff --git a/lib/crates/fabro-cli/src/commands/rewind.rs b/lib/crates/fabro-cli/src/commands/rewind.rs index dbbcd5e80..2e6d22ab6 100644 --- a/lib/crates/fabro-cli/src/commands/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/rewind.rs @@ -28,22 +28,22 @@ pub struct RewindArgs { pub fn run(args: &RewindArgs, styles: &Styles) -> Result<()> { let repo = Repository::discover(".").context("not in a git repository")?; - let run_id = fabro_workflows::run_rewind::find_run_id_by_prefix(&repo, &args.run_id)?; + let run_id = fabro_workflows::operations::find_run_id_by_prefix(&repo, &args.run_id)?; let store = Store::new(repo); - let timeline = fabro_workflows::run_rewind::build_timeline(&store, &run_id)?; + let timeline = fabro_workflows::operations::build_timeline(&store, &run_id)?; if args.list || args.target.is_none() { - let parallel_map = fabro_workflows::run_rewind::load_parallel_map(&store, &run_id); + let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id); print_timeline(&timeline, ¶llel_map, styles); return Ok(()); } - let target = fabro_workflows::run_rewind::parse_target(args.target.as_deref().unwrap())?; - let parallel_map = fabro_workflows::run_rewind::load_parallel_map(&store, &run_id); - let entry = fabro_workflows::run_rewind::resolve_target(&timeline, &target, ¶llel_map)?; + 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)?; - fabro_workflows::run_rewind::execute_rewind(&store, &run_id, entry, !args.no_push)?; + fabro_workflows::operations::rewind(&store, &run_id, entry, !args.no_push)?; eprintln!( "\nTo resume: fabro resume {}", @@ -54,7 +54,7 @@ pub fn run(args: &RewindArgs, styles: &Styles) -> Result<()> { } pub(crate) fn print_timeline( - timeline: &[fabro_workflows::run_rewind::TimelineEntry], + timeline: &[fabro_workflows::operations::TimelineEntry], parallel_map: &std::collections::HashMap, styles: &Styles, ) { diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index b3b0f00a5..b2baf4c63 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -19,11 +19,17 @@ use fabro_workflows::backend::{AgentApiBackend, AgentCliBackend, BackendRouter}; use fabro_workflows::checkpoint::Checkpoint; use fabro_workflows::cost::{compute_stage_cost, format_cost}; use fabro_workflows::devcontainer_bridge; -use fabro_workflows::engine::{GitCheckpointSettings, RunSettings, WorkflowRunEngine}; use fabro_workflows::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use fabro_workflows::git::GitSyncStatus; use fabro_workflows::handler::default_registry; +use fabro_workflows::operations::{ + create_from_graph, start, StartFinalizeConfig, StartOptions, StartRetroConfig, +}; use fabro_workflows::outcome::StageStatus; +use fabro_workflows::pipeline::{ + build_conclusion, classify_engine_result, persist_terminal_outcome, +}; +use fabro_workflows::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings}; use fabro_workflows::sandbox_provider::SandboxProvider; use indicatif::HumanDuration; use std::time::Duration; @@ -35,10 +41,6 @@ use crate::commands::shared::{ format_tokens_human, print_diagnostics, read_workflow_file, relative_path, tilde_path, }; -pub(crate) use fabro_workflows::pipeline::{ - build_conclusion, classify_engine_result, persist_terminal_outcome, write_finalize_commit, -}; - #[derive(Debug, Clone, Copy, ValueEnum)] pub enum CliSandboxProvider { Local, @@ -743,7 +745,7 @@ pub(crate) fn prepare_workflow_with_project_config( /// Pre-prepared run state, used to skip workflow preparation in `run_command_impl`. struct RecordBasedRun { - graph: fabro_graphviz::graph::Graph, + validated: fabro_workflows::pipeline::Validated, raw_source: String, run_cfg: Option, sandbox_provider: SandboxProvider, @@ -789,7 +791,7 @@ pub async fn run_from_record( let record_run = RecordBasedRun { raw_source: String::new(), // Raw DOT provenance is best-effort for record-based runs - graph: record.graph.clone(), + validated: create_from_graph(record.graph.clone(), String::new()), run_cfg: Some(record.config.clone()), sandbox_provider, model: model.clone(), @@ -853,10 +855,9 @@ pub async fn run_command( run_defaults, workflow_toml_path, } = prepare_workflow(&args, run_defaults, styles, false)?; - let (graph, _source, _diagnostics) = validated.into_parts(); let record_run = RecordBasedRun { - graph, + validated, raw_source, run_cfg, sandbox_provider, @@ -878,7 +879,7 @@ async fn run_command_impl( record_run: Option, ) -> anyhow::Result<()> { let ( - graph, + validated, raw_source, mut run_cfg, sandbox_provider, @@ -889,7 +890,7 @@ async fn run_command_impl( workflow_toml_path, ) = match record_run { Some(rr) => ( - rr.graph, + rr.validated, rr.raw_source, rr.run_cfg, rr.sandbox_provider, @@ -901,6 +902,7 @@ async fn run_command_impl( ), None => unreachable!("run_command_impl always receives a RecordBasedRun"), }; + let graph = validated.graph().clone(); // For record-based runs from run_from_record, workflow is None (preparation was skipped). let from_record = args.workflow.is_none(); @@ -1048,21 +1050,6 @@ async fn run_command_impl( // 3. Build event emitter let emitter = EventEmitter::new(); - // Track the last git commit SHA from CheckpointCompleted events - let last_git_sha: Arc>> = Arc::new(Mutex::new(None)); - { - let sha_clone = Arc::clone(&last_git_sha); - emitter.on_event(move |event| { - if let fabro_workflows::event::WorkflowRunEvent::CheckpointCompleted { - git_commit_sha: Some(sha), - .. - } = event - { - *sha_clone.lock().unwrap() = Some(sha.clone()); - } - }); - } - // Cost accumulator — shared across all verbosity levels let accumulator = Arc::new(Mutex::new(CostAccumulator::default())); let acc_clone = Arc::clone(&accumulator); @@ -1689,32 +1676,6 @@ async fn run_command_impl( } } }); - let mut engine = WorkflowRunEngine::with_interviewer( - registry, - Arc::clone(&emitter), - interviewer, - Arc::clone(&sandbox), - ); - if !sandbox_env.is_empty() { - engine.set_env(sandbox_env); - } - if dry_run_mode { - engine.set_dry_run(true); - } - // Wire up hook runner from run config or run defaults - { - let hooks = run_cfg - .as_ref() - .map(|c| &c.hooks) - .unwrap_or(&run_defaults.hooks); - if !hooks.is_empty() { - let hook_config = fabro_hooks::HookConfig { - hooks: hooks.clone(), - }; - let runner = fabro_hooks::HookRunner::new(hook_config); - engine.set_hook_runner(Arc::new(runner)); - } - } // 7. Execute // Set up metadata branch for git checkpointing (host or remote — engine fills remote) @@ -1728,7 +1689,7 @@ async fn run_command_impl( None }; - let mut config = RunSettings { + let config = RunSettings { config: settings_config, run_dir: run_dir.clone(), cancel_token: None, @@ -1754,7 +1715,7 @@ async fn run_command_impl( }; // Build lifecycle config for sandbox init, setup commands, and devcontainer phases - let lifecycle = fabro_workflows::engine::LifecycleConfig { + let lifecycle = LifecycleConfig { setup_commands, setup_command_timeout_ms: 300_000, devcontainer_phases: if let Some(ref dc) = devcontainer_config { @@ -1771,287 +1732,97 @@ async fn run_command_impl( // Defuse the bootstrap guard — engine.run() has taken ownership of lifecycle status. status_guard.defuse(); - // Safety net: if we panic or return early, best-effort cleanup via spawn. - let sandbox_for_cleanup = Arc::clone(&sandbox); - let cleanup_guard = scopeguard::guard((), move |()| { - if preserve_sandbox { - return; - } - let rt = tokio::runtime::Handle::try_current(); - if let Ok(handle) = rt { - handle.spawn(async move { - let _ = sandbox_for_cleanup.cleanup().await; - }); - } - }); - let run_start = Instant::now(); - let engine_result = engine - .run_with_lifecycle(&graph, &mut config, lifecycle, None) - .await; + let pr_config = config.pull_request().cloned(); + let started = start( + validated, + StartOptions { + init: fabro_workflows::pipeline::InitOptions { + run_id: run_id.clone(), + run_dir: run_dir.clone(), + dry_run: dry_run_mode, + emitter: Arc::clone(&emitter), + sandbox: Arc::clone(&sandbox), + registry: Arc::new(registry), + lifecycle, + run_settings: config, + hooks: fabro_hooks::HookConfig { + hooks: run_cfg + .as_ref() + .map(|c| c.hooks.clone()) + .unwrap_or_else(|| run_defaults.hooks.clone()), + }, + sandbox_env, + checkpoint: None, + seed_context: None, + }, + retro: StartRetroConfig { + enabled: !no_retro_flag && project_config::is_retro_enabled(), + dry_run: dry_run_mode, + llm_client: llm_client.clone(), + provider: provider_enum, + model: model.clone(), + }, + finalize: StartFinalizeConfig { + preserve_sandbox, + pr_config, + github_app: github_app.clone(), + origin_url: origin_url.clone(), + model: model.clone(), + }, + }, + ) + .await; let run_duration_ms = run_start.elapsed().as_millis() as u64; let mut completion_guard = DetachedRunCompletionGuard::arm(&run_dir); // Restore cwd (worktree is kept for `fabro cp` access; pruned separately) let _ = std::env::set_current_dir(&original_cwd); - - let (final_status, failure_reason, run_status, status_reason) = - classify_engine_result(&engine_result); - let conclusion = build_conclusion( - &run_dir, - final_status.clone(), - failure_reason, - run_duration_ms, - last_git_sha.lock().unwrap().clone(), - ); - - // Auto-derive retro (always, cheap) and optionally run retro agent - if !no_retro_flag && project_config::is_retro_enabled() { - let failed = match &engine_result { - Ok(ref o) => o.status == StageStatus::Fail, - Err(_) => true, - }; - generate_retro( - &config.run_id, - &graph.name, - graph.goal(), - &run_dir, - failed, - run_duration_ms, - dry_run_mode, - llm_client.as_ref(), - &sandbox, - provider_enum, - &model, - styles, - Some(Arc::clone(&emitter)), - ) - .await; - } - - // Finish progress bars after retro (retro stage uses the same ProgressUI) progress_ui.lock().expect("progress lock poisoned").finish(); - // Write finalize commit with retro.json + final node files (captures last diff.patch) - write_finalize_commit(&config, &run_dir).await; - - // Auto-create PR on successful completion (skip in dry-run mode) - let mut pushed_branch: Option = None; - let mut pr_url: Option = None; - if let Some(pr_cfg) = config.pull_request() { - if dry_run_mode { - debug!("Skipping PR creation: dry-run mode"); - } else if let Err(ref e) = engine_result { - debug!(error = %e, "Skipping PR creation: engine returned an error"); - } else if let Ok(ref outcome) = engine_result { - if !matches!( - outcome.status, - StageStatus::Success | StageStatus::PartialSuccess - ) { - debug!(status = ?outcome.status, "Skipping PR creation: run status is not success"); - } else { - let diff = tokio::fs::read_to_string(run_dir.join("final.patch")) - .await - .unwrap_or_default(); - if let ( - Some(ref base_branch), - Some(ref run_branch), - Some(ref creds), - Some(ref origin), - ) = ( - &config.base_branch, - config.git.as_ref().and_then(|g| g.run_branch.as_ref()), - &github_app, - &origin_url, - ) { - // Run branch was pushed during checkpoint commits; - // just record it for the PR creation. - if config.git.is_some() { - pushed_branch = Some(run_branch.to_string()); - } - - let auto_merge = if pr_cfg.auto_merge { - Some(fabro_workflows::pull_request::AutoMergeConfig { - merge_strategy: pr_cfg.merge_strategy, - }) - } else { - None - }; - - match fabro_workflows::pull_request::maybe_open_pull_request( - creds, - origin, - base_branch, - run_branch, - graph.goal(), - &diff, - &model, - pr_cfg.draft, - auto_merge, - &run_dir, - ) - .await - { - Ok(Some(record)) => { - emitter.emit( - &fabro_workflows::event::WorkflowRunEvent::PullRequestCreated { - pr_url: record.html_url.clone(), - pr_number: record.number, - draft: pr_cfg.draft, - }, - ); - pr_url = Some(record.html_url.clone()); - if let Err(e) = record.save(&run_dir.join("pull_request.json")) { - tracing::warn!(error = %e, "Failed to save pull_request.json"); - } - } - Ok(None) => {} // empty diff, logged at DEBUG - Err(e) => { - emitter.emit( - &fabro_workflows::event::WorkflowRunEvent::PullRequestFailed { - error: e.to_string(), - }, - ); - emit_run_notice( - &emitter, - RunNoticeLevel::Warn, - "pull_request_failed", - format!("PR creation failed: {e}"), - ); - } - } - } + let final_status = match started { + Ok(started) => { + if let Some(ref retro) = started.retro { + print_retro_result(retro, started.retro_duration, &run_dir, styles); } + let finalized = started.finalized; + print_run_conclusion( + &finalized.conclusion, + &run_id, + &run_dir, + finalized.pushed_branch.as_deref(), + finalized.pr_url.as_deref(), + styles, + ); + print_final_output(&run_dir, styles); + print_assets(&run_dir, styles); + finalized.conclusion.status.clone() + } + Err(err) => { + let engine_result = Err(err.clone()); + let (final_status, failure_reason, run_status, status_reason) = + classify_engine_result(&engine_result); + let conclusion = build_conclusion( + &run_dir, + final_status.clone(), + failure_reason, + run_duration_ms, + None, + ); + persist_terminal_outcome(&run_dir, &conclusion, run_status, status_reason); + print_run_conclusion(&conclusion, &run_id, &run_dir, None, None, styles); + print_final_output(&run_dir, styles); + print_assets(&run_dir, styles); + final_status } - } else { - debug!("Skipping PR creation: pull_request not enabled in config"); - } - - // 8. Print result - eprintln!("\n{}", styles.bold.apply_to("=== Run Result ==="),); - - eprintln!("{}", styles.dim.apply_to(format!("Run: {run_id}"))); - let status_str = final_status.to_string().to_uppercase(); - let status_color = match final_status { - StageStatus::Success | StageStatus::PartialSuccess => &styles.bold_green, - _ => &styles.bold_red, }; - eprintln!("Status: {}", status_color.apply_to(&status_str),); - eprintln!( - "Duration: {}", - HumanDuration(Duration::from_millis(run_duration_ms)) - ); - { - let acc = accumulator.lock().unwrap(); - let total_tokens = acc.total_input_tokens + acc.total_output_tokens; - if total_tokens > 0 { - if acc.has_pricing { - eprintln!( - "{}", - styles.dim.apply_to(format!( - "Cost: {} ({} toks)", - format_cost(acc.total_cost), - format_tokens_human(total_tokens) - )) - ); - } else { - eprintln!( - "{}", - styles - .dim - .apply_to(format!("Toks: {}", format_tokens_human(total_tokens))) - ); - } - if acc.total_cache_read_tokens > 0 { - eprintln!( - "{}", - styles.dim.apply_to(format!( - "Cache: {} read, {} write", - format_tokens_human(acc.total_cache_read_tokens), - format_tokens_human(acc.total_cache_write_tokens), - )), - ); - } - if acc.total_reasoning_tokens > 0 { - eprintln!( - "{}", - styles.dim.apply_to(format!( - "Reasoning: {} tokens", - format_tokens_human(acc.total_reasoning_tokens), - )), - ); - } - } - } - - eprintln!( - "{}", - styles - .dim - .apply_to(format!("Run: {}", tilde_path(&run_dir))) - ); - - if let Some(failure) = conclusion.failure_reason.as_deref() { - eprintln!("Failure: {}", styles.red.apply_to(failure)); - } - - if pushed_branch.is_some() || pr_url.is_some() { - eprintln!(); - if let Some(ref branch) = pushed_branch { - eprintln!("{} {branch}", styles.bold.apply_to("Pushed branch:")); - } - if let Some(ref url) = pr_url { - eprintln!("{} {url}", styles.bold.apply_to("Pull request:")); - } - } - - print_final_output(&run_dir, styles); - print_assets(&run_dir, styles); - - // 9. Cleanup sandbox (defuse the scopeguard so we await properly) - scopeguard::ScopeGuard::into_inner(cleanup_guard); - if preserve_sandbox { - let info = sandbox.sandbox_info(); - if !info.is_empty() { - emit_run_notice( - &emitter, - RunNoticeLevel::Info, - "sandbox_preserved", - format!("sandbox preserved: {info}"), - ); - } else { - emit_run_notice( - &emitter, - RunNoticeLevel::Info, - "sandbox_preserved", - "sandbox preserved", - ); - } - } - if let Err(e) = engine - .cleanup_sandbox(&run_id, &graph.name, preserve_sandbox) - .await - { - tracing::warn!(error = %e, "Sandbox cleanup failed"); - emit_run_notice( - &emitter, - RunNoticeLevel::Warn, - "sandbox_cleanup_failed", - format!("sandbox cleanup failed: {e}"), - ); - } - - persist_terminal_outcome(&run_dir, &conclusion, run_status, status_reason); completion_guard.defuse(); - // 10. Exit code fabro_util::run_log::deactivate(); match final_status { StageStatus::Success | StageStatus::PartialSuccess => Ok(()), - _ => { - std::process::exit(1); - } + _ => std::process::exit(1), } } @@ -2078,6 +1849,36 @@ pub fn print_run_summary(run_dir: &Path, run_id: &str, styles: &Styles) { return; }; + // PR info from pull_request.json (saved by _run_engine) + let pr_url = std::fs::read_to_string(run_dir.join("pull_request.json")) + .ok() + .and_then(|content| { + serde_json::from_str::(&content) + .ok() + .map(|record| record.html_url) + }); + + print_run_conclusion( + &conclusion, + run_id, + run_dir, + None, + pr_url.as_deref(), + styles, + ); + + print_final_output(run_dir, styles); + print_assets(run_dir, styles); +} + +pub(crate) fn print_run_conclusion( + conclusion: &fabro_workflows::conclusion::Conclusion, + run_id: &str, + run_dir: &Path, + pushed_branch: Option<&str>, + pr_url: Option<&str>, + styles: &Styles, +) { eprintln!("\n{}", styles.bold.apply_to("=== Run Result ===")); eprintln!("{}", styles.dim.apply_to(format!("Run: {run_id}"))); @@ -2147,22 +1948,74 @@ pub fn print_run_summary(run_dir: &Path, run_id: &str, styles: &Styles) { eprintln!("Failure: {}", styles.red.apply_to(failure)); } - // PR info from pull_request.json (saved by _run_engine) - if let Ok(content) = std::fs::read_to_string(run_dir.join("pull_request.json")) { - if let Ok(record) = - serde_json::from_str::(&content) - { - eprintln!(); - eprintln!( - "{} {}", - styles.bold.apply_to("Pull request:"), - record.html_url - ); + if pushed_branch.is_some() || pr_url.is_some() { + eprintln!(); + if let Some(branch) = pushed_branch { + eprintln!("{} {branch}", styles.bold.apply_to("Pushed branch:")); + } + if let Some(url) = pr_url { + eprintln!("{} {url}", styles.bold.apply_to("Pull request:")); } } +} - print_final_output(run_dir, styles); - print_assets(run_dir, styles); +pub(crate) fn print_retro_result( + retro: &fabro_retro::retro::Retro, + duration: Duration, + run_dir: &Path, + styles: &Styles, +) { + eprintln!("\n{}", styles.bold.apply_to("=== Retro ===")); + + let retro_dur = run_progress::format_duration_short(duration); + let smoothness_str = retro + .smoothness + .as_ref() + .map(|s| s.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + let outcome_str = retro.outcome.as_deref().unwrap_or("No outcome recorded"); + let line1_content = format!("Retro: {smoothness_str} - {outcome_str}"); + let term_width = console::Term::stderr().size().1 as usize; + let pad1 = term_width.saturating_sub(line1_content.len() + retro_dur.len()); + eprintln!( + "{} {}{:pad1$}{}", + styles.bold.apply_to("Retro:"), + styles + .dim + .apply_to(format!("{smoothness_str} - {outcome_str}")), + "", + styles.dim.apply_to(&retro_dur), + ); + + let friction_count = retro.friction_points.as_ref().map(|v| v.len()).unwrap_or(0); + let open_count = retro.open_items.as_ref().map(|v| v.len()).unwrap_or(0); + if friction_count > 0 || open_count > 0 { + let mut parts = Vec::new(); + if friction_count > 0 { + let noun = if friction_count == 1 { + "friction point" + } else { + "friction points" + }; + parts.push(format!("{friction_count} {noun}")); + } + if open_count > 0 { + let noun = if open_count == 1 { + "open item" + } else { + "open items" + }; + parts.push(format!("{open_count} {noun}")); + } + eprintln!(" {}", styles.dim.apply_to(parts.join(" · "))); + } + + let retro_path = format!("{}/retro.json", tilde_path(run_dir)); + eprintln!( + " {} {}", + styles.dim.apply_to("Retro saved to"), + styles.underline.apply_to(&retro_path), + ); } /// Print the final stage output from the checkpoint, if available. @@ -2557,108 +2410,6 @@ async fn run_preflight( } } -/// Generate a retro report for a completed workflow run. -/// -/// Derives a basic retro from the checkpoint, then optionally runs the retro agent -/// for a richer narrative. Errors are logged as warnings rather than propagated. -#[allow(clippy::too_many_arguments)] -pub(crate) async fn generate_retro( - run_id: &str, - workflow_name: &str, - goal: &str, - run_dir: &std::path::Path, - failed: bool, - run_duration_ms: u64, - dry_run_mode: bool, - llm_client: Option<&fabro_llm::client::Client>, - sandbox: &Arc, - provider_enum: Provider, - model: &str, - styles: &'static Styles, - emitter: Option>, -) { - eprintln!("\n{}", styles.bold.apply_to("=== Retro ===")); - if emitter.is_none() { - eprintln!( - "{}", - styles.dim.apply_to(format!("Running retro ({model})...")) - ); - } - - let retro_start = std::time::Instant::now(); - let retro = fabro_workflows::pipeline::run_retro(&fabro_workflows::pipeline::RetroOptions { - run_id: run_id.to_string(), - workflow_name: workflow_name.to_string(), - goal: goal.to_string(), - run_dir: run_dir.to_path_buf(), - sandbox: Arc::clone(sandbox), - emitter, - failed, - run_duration_ms, - enabled: true, - dry_run: dry_run_mode, - llm_client: llm_client.cloned(), - provider: provider_enum, - model: model.to_string(), - }) - .await; - - let retro_dur = run_progress::format_duration_short(retro_start.elapsed()); - if let Some(retro) = retro { - let smoothness_str = retro - .smoothness - .as_ref() - .map(|s| s.to_string()) - .unwrap_or_else(|| "unknown".to_string()); - let outcome_str = retro.outcome.as_deref().unwrap_or("No outcome recorded"); - let line1_content = format!("Retro: {smoothness_str} \u{2014} {outcome_str}"); - let term_width = console::Term::stderr().size().1 as usize; - let dur_len = retro_dur.len(); - let pad1 = term_width.saturating_sub(line1_content.len() + dur_len); - eprintln!( - "{} {}{:pad1$}{}", - styles.bold.apply_to("Retro:"), - styles - .dim - .apply_to(format!("{smoothness_str} \u{2014} {outcome_str}")), - "", - styles.dim.apply_to(&retro_dur), - ); - - let friction_count = retro.friction_points.as_ref().map(|v| v.len()).unwrap_or(0); - let open_count = retro.open_items.as_ref().map(|v| v.len()).unwrap_or(0); - if friction_count > 0 || open_count > 0 { - let mut parts = Vec::new(); - if friction_count > 0 { - let noun = if friction_count == 1 { - "friction point" - } else { - "friction points" - }; - parts.push(format!("{friction_count} {noun}")); - } - if open_count > 0 { - let noun = if open_count == 1 { - "open item" - } else { - "open items" - }; - parts.push(format!("{open_count} {noun}")); - } - eprintln!(" {}", styles.dim.apply_to(parts.join(" \u{00b7} "))); - } - - let retro_path = format!("{}/retro.json", tilde_path(run_dir)); - eprintln!( - " {} {}", - styles.dim.apply_to("Retro saved to"), - styles.underline.apply_to(&retro_path), - ); - } else { - eprintln!("{}", styles.dim.apply_to("Retro unavailable")); - } -} - pub(crate) fn build_event_envelope( event: &fabro_workflows::event::WorkflowRunEvent, run_id: &str, diff --git a/lib/crates/fabro-cli/src/commands/validate.rs b/lib/crates/fabro-cli/src/commands/validate.rs index e4bea600e..2a3c4de84 100644 --- a/lib/crates/fabro-cli/src/commands/validate.rs +++ b/lib/crates/fabro-cli/src/commands/validate.rs @@ -16,7 +16,9 @@ pub struct ValidateArgs { pub fn run(args: &ValidateArgs, styles: &Styles) -> anyhow::Result<()> { let (dot_path, _cfg) = fabro_config::project::resolve_workflow(&args.workflow)?; - let (graph, diagnostics) = fabro_workflows::workflow::prepare_from_file(&dot_path)?; + let validated = fabro_workflows::operations::create_from_file(&dot_path)?; + let graph = validated.graph(); + let diagnostics = validated.diagnostics(); eprintln!( "{} ({} nodes, {} edges)", diff --git a/lib/crates/fabro-workflows/src/core_adapter/graph.rs b/lib/crates/fabro-workflows/src/core_adapter/graph.rs index 318f65b08..ca02b92b1 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/graph.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/graph.rs @@ -6,7 +6,7 @@ use fabro_core::graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode}; use crate::context::Context; -use crate::engine; +use crate::graph_ops; use crate::outcome::{Outcome, StageUsage}; // ---- WorkflowNode ---- @@ -26,7 +26,7 @@ impl NodeSpec for WorkflowNode { } fn is_terminal(&self) -> bool { - engine::is_terminal(&self.0) + graph_ops::is_terminal(&self.0) } fn max_visits(&self) -> Option { @@ -103,7 +103,7 @@ impl Graph for WorkflowGraph { outcome: &Outcome, context: &Context, ) -> Option> { - let selection = engine::select_edge( + let selection = graph_ops::select_edge( node.inner(), outcome, context, @@ -120,10 +120,10 @@ impl Graph for WorkflowGraph { &self, outcomes: &HashMap, ) -> std::result::Result<(), String> { - engine::check_goal_gates(self.inner(), outcomes) + graph_ops::check_goal_gates(self.inner(), outcomes) } fn get_retry_target(&self, failed_node_id: &str) -> Option { - engine::get_retry_target(failed_node_id, self.inner()) + graph_ops::get_retry_target(failed_node_id, self.inner()) } } diff --git a/lib/crates/fabro-workflows/src/core_adapter/handler.rs b/lib/crates/fabro-workflows/src/core_adapter/handler.rs index 684be397b..922d1bda1 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/handler.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/handler.rs @@ -14,9 +14,9 @@ use crate::context::Context; use super::graph::WorkflowGraph; use super::WorkflowNode; -use crate::engine; use crate::handler::{format_panic_message, EngineServices}; use crate::outcome::{Outcome, StageStatus}; +use crate::{graph_ops, run_dir}; /// Production node handler that bridges fabro-core's NodeHandler to the /// existing fabro-workflows Handler trait via EngineServices. @@ -98,7 +98,7 @@ impl NodeHandler for WorkflowNodeHandler { Err(panic_payload) => { let msg = format_panic_message(panic_payload); let visit = context.node_visit_count().max(1); - let panic_dir = crate::engine::node_dir(&self.run_dir, &gv_node.id, visit); + let panic_dir = run_dir::node_dir(&self.run_dir, &gv_node.id, visit); let _ = std::fs::create_dir_all(&panic_dir); let _ = std::fs::write(panic_dir.join("panic.txt"), &msg); Err(CoreError::handler(HandlerErrorDetail { @@ -113,7 +113,7 @@ impl NodeHandler for WorkflowNodeHandler { fn retry_policy(&self, node: &WorkflowNode, _graph: &WorkflowGraph) -> CoreRetryPolicy { let gv_node = node.inner(); - let wf_policy = engine::build_retry_policy(gv_node, &self.graph); + let wf_policy = graph_ops::build_retry_policy(gv_node, &self.graph); CoreRetryPolicy { max_attempts: wf_policy.max_attempts, backoff: wf_policy.backoff, diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs index db5268924..bc6e36c94 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs @@ -10,8 +10,8 @@ use fabro_core::state::RunState; use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; -use crate::engine; use crate::error::{FailureCategory, FailureSignature}; +use crate::graph_ops::classify_outcome; use crate::outcome::{OutcomeExt, StageStatus, StageUsage}; type WfRunState = RunState>; @@ -69,7 +69,7 @@ impl RunLifecycle for CircuitBreakerLifecycle { let outcome = &result.outcome; let outcome_failure_category = if outcome.status == StageStatus::Fail { - engine::classify_outcome(outcome) + classify_outcome(outcome) } else { None }; @@ -117,7 +117,7 @@ impl RunLifecycle for CircuitBreakerLifecycle { let outcome = ctx.outcome; // Guard: only TransientInfra failures may trigger loop_restart - let failure_class = engine::classify_outcome(outcome); + let failure_class = classify_outcome(outcome); if let Some(fc) = failure_class { if fc != FailureCategory::TransientInfra { return Ok(EdgeDecision::Block(format!( diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/disk.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/disk.rs index 9b6b481ef..1a2b16324 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/disk.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/disk.rs @@ -12,9 +12,10 @@ use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; use super::circuit_breaker::CircuitBreakerLifecycle; use crate::checkpoint::Checkpoint; -use crate::engine::{self, RunSettings}; use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use crate::outcome::StageUsage; +use crate::run_dir::{write_node_status, write_start_record}; +use crate::run_settings::RunSettings; type WfRunState = RunState>; type WfNodeResult = NodeResult>; @@ -38,7 +39,7 @@ impl RunLifecycle for DiskLifecycle { _state: &WfRunState, ) -> fabro_core::error::Result<()> { // Write start.json - engine::write_start_record(&self.run_dir, &self.config); + write_start_record(&self.run_dir, &self.config); // Write run status as Running crate::run_status::write_run_status( &self.run_dir, @@ -56,7 +57,7 @@ impl RunLifecycle for DiskLifecycle { ) -> fabro_core::error::Result<()> { let gv = node.inner(); let visit = state.node_visits.get(gv.id.as_str()).copied().unwrap_or(1); - engine::write_node_status(&self.run_dir, &gv.id, visit, &result.outcome); + write_node_status(&self.run_dir, &gv.id, visit, &result.outcome); Ok(()) } diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs index 5b7f02fe0..66d285921 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs @@ -12,8 +12,8 @@ use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; use super::git::GitCheckpointResult; use crate::artifact::ArtifactStore; -use crate::engine; use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::graph_ops::node_script; use crate::outcome::{FailureCategory, FailureDetail, Outcome, StageStatus, StageUsage}; type WfRunState = RunState>; @@ -87,7 +87,7 @@ impl RunLifecycle for EventLifecycle { name: gv.label().to_string(), index: stage_index, handler_type: gv.handler_type().map(String::from), - script: engine::node_script(gv), + script: node_script(gv), attempt: 1, max_attempts: 1, }); @@ -119,7 +119,7 @@ impl RunLifecycle for EventLifecycle { name: gv.label().to_string(), index: state.stage_index, handler_type: gv.handler_type().map(String::from), - script: engine::node_script(gv), + script: node_script(gv), attempt: ctx.attempt as usize, max_attempts: ctx.max_attempts as usize, }); diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs index 5323183c5..088e852c4 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs @@ -9,7 +9,7 @@ use fabro_core::state::RunState; use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; use crate::context::keys; -use crate::engine; +use crate::graph_ops::{resolve_fidelity, resolve_thread_id}; use crate::outcome::StageUsage; use crate::preamble::build_preamble; @@ -67,7 +67,7 @@ impl RunLifecycle for FidelityLifecycle { // 1. Fidelity resolution via resolve_fidelity: edge → node → graph default → Compact let incoming_edge_ref = incoming.as_ref().map(|d| d.edge.as_ref()); - let fidelity = engine::resolve_fidelity(incoming_edge_ref, gv_node, &self.graph); + let fidelity = resolve_fidelity(incoming_edge_ref, gv_node, &self.graph); // 2. Fidelity degradation on resume (full → summary:high) let fidelity = { @@ -99,7 +99,7 @@ impl RunLifecycle for FidelityLifecycle { .set(keys::CURRENT_PREAMBLE, serde_json::json!(preamble)); // 5. Thread ID resolution via resolve_thread_id: edge → node → graph default → class → previous - let thread_id = engine::resolve_thread_id( + let thread_id = resolve_thread_id( incoming_edge_ref, gv_node, &self.graph, diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs index f7e3706a9..14d1d54e2 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs @@ -12,9 +12,11 @@ use fabro_core::state::RunState; use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; use crate::artifact::ArtifactStore; -use crate::engine::{self, RunSettings}; use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use crate::outcome::{Outcome, StageStatus, StageUsage}; +use crate::run_dir::node_dir; +use crate::run_settings::RunSettings; +use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host}; type WfRunState = RunState>; type WfNodeResult = NodeResult>; @@ -149,7 +151,7 @@ impl RunLifecycle for GitLifecycle { // Run branch commit via sandbox let completed_count = state.completed_nodes.len(); - let commit_result = engine::git_checkpoint( + let commit_result = git_checkpoint( &*self.sandbox, &self.run_id, node_id, @@ -192,7 +194,7 @@ impl RunLifecycle for GitLifecycle { true } else if let Some(repo_path) = self.config.host_repo_path.as_ref() { let refspec = format!("refs/heads/{branch}"); - engine::git_push_host( + git_push_host( repo_path, &refspec, &self.config.github_app, @@ -213,7 +215,7 @@ impl RunLifecycle for GitLifecycle { self.config.host_repo_path.as_ref(), ) { let refspec = format!("refs/heads/{meta_branch}"); - let meta_push_ok = engine::git_push_host( + let meta_push_ok = git_push_host( repo_path, &refspec, &self.config.github_app, @@ -235,9 +237,9 @@ impl RunLifecycle for GitLifecycle { .clone() .or_else(|| self.config.git.as_ref().and_then(|g| g.base_sha.clone())) .unwrap_or_else(|| sha.clone()); - let diff_dest = engine::node_dir(&self.run_dir, node_id, visit).join("diff.patch"); + let diff_dest = node_dir(&self.run_dir, node_id, visit).join("diff.patch"); - match engine::git_diff(&*self.sandbox, &prev).await { + match git_diff(&*self.sandbox, &prev).await { Ok(patch) if !patch.is_empty() => { let _ = std::fs::write(&diff_dest, patch); } @@ -277,7 +279,7 @@ impl RunLifecycle for GitLifecycle { { if let Some(base_sha) = self.config.git.as_ref().and_then(|g| g.base_sha.clone()) { let diff_dest = self.run_dir.join("final.patch"); - match engine::git_diff(&*self.sandbox, &base_sha).await { + match git_diff(&*self.sandbox, &base_sha).await { Ok(patch) if !patch.is_empty() => { let _ = std::fs::write(&diff_dest, patch); } diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs index 7f699131c..1a73cf51c 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs @@ -12,7 +12,7 @@ use fabro_core::state::RunState; use super::super::graph::WorkflowGraph; use super::super::WorkflowNode; -use crate::engine::set_hook_node; +use crate::graph_ops::set_hook_node; use crate::outcome::{Outcome, OutcomeExt, StageStatus, StageUsage}; use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner}; use fabro_sandbox::Sandbox; diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs index 5ff144142..d547393cf 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs @@ -27,9 +27,9 @@ use super::graph::WorkflowGraph; use super::WorkflowNode; use crate::artifact::ArtifactStore; use crate::context; -use crate::engine::RunSettings; use crate::event::EventEmitter; use crate::outcome::{Outcome, StageUsage}; +use crate::run_settings::RunSettings; use fabro_hooks::HookRunner; use fabro_sandbox::Sandbox; @@ -307,7 +307,7 @@ impl RunLifecycle for WorkflowLifecycle { ) -> CoreResult<()> { let outcome = &result.outcome; let retry_count = state.node_retries.get(node.id()).copied().unwrap_or(0); - let failure_class = crate::engine::classify_outcome(outcome); + let failure_class = crate::graph_ops::classify_outcome(outcome); let failure_signature = failure_class .map(|category| { let signature_hint = outcome diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index 0e2f54f1f..b34206633 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -28,17 +28,11 @@ use fabro_graphviz::graph::{Edge, Node}; use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner}; use fabro_interview::Interviewer; -pub(crate) use crate::graph_ops::{ - build_retry_policy, check_goal_gates, classify_outcome, get_retry_target, is_terminal, - node_script, set_hook_node, -}; pub use crate::graph_ops::{ resolve_fidelity, resolve_thread_id, select_edge, EdgeSelection, RetryPolicy, }; pub use crate::run_dir::{node_dir, visit_from_context}; -pub(crate) use crate::run_dir::{write_node_status, write_start_record}; pub use crate::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings}; -pub(crate) use crate::sandbox_git::git_diff; pub use crate::sandbox_git::{ git_add_worktree, git_checkpoint, git_create_branch_at, git_merge_ff_only, git_push_host, git_remove_worktree, git_replace_worktree, GitState, GIT_REMOTE, @@ -613,6 +607,7 @@ mod tests { use super::*; use crate::handler::start::StartHandler; use crate::handler::Handler as HandlerTrait; + use crate::outcome::OutcomeExt; use async_trait::async_trait; use fabro_graphviz::graph::AttrValue; use std::time::Duration; diff --git a/lib/crates/fabro-workflows/src/handler/agent.rs b/lib/crates/fabro-workflows/src/handler/agent.rs index 7354e16f4..98dd15f56 100644 --- a/lib/crates/fabro-workflows/src/handler/agent.rs +++ b/lib/crates/fabro-workflows/src/handler/agent.rs @@ -249,8 +249,8 @@ impl Handler for AgentHandler { }; // 2. Write prompt to logs - let visit = crate::engine::visit_from_context(context); - let stage_dir = crate::engine::node_dir(run_dir, &node.id, visit); + let visit = crate::run_dir::visit_from_context(context); + let stage_dir = crate::run_dir::node_dir(run_dir, &node.id, visit); tokio::fs::create_dir_all(&stage_dir).await?; tokio::fs::write(stage_dir.join("prompt.md"), &prompt).await?; diff --git a/lib/crates/fabro-workflows/src/handler/command.rs b/lib/crates/fabro-workflows/src/handler/command.rs index 5a422d36f..38e0dd14d 100644 --- a/lib/crates/fabro-workflows/src/handler/command.rs +++ b/lib/crates/fabro-workflows/src/handler/command.rs @@ -84,8 +84,8 @@ impl Handler for CommandHandler { ))); } - let visit = crate::engine::visit_from_context(context); - let stage_dir = crate::engine::node_dir(run_dir, &node.id, visit); + let visit = crate::run_dir::visit_from_context(context); + let stage_dir = crate::run_dir::node_dir(run_dir, &node.id, visit); tokio::fs::create_dir_all(&stage_dir).await?; let invocation = serde_json::json!({ diff --git a/lib/crates/fabro-workflows/src/handler/fan_in.rs b/lib/crates/fabro-workflows/src/handler/fan_in.rs index 4db24d306..9ee1521ef 100644 --- a/lib/crates/fabro-workflows/src/handler/fan_in.rs +++ b/lib/crates/fabro-workflows/src/handler/fan_in.rs @@ -116,7 +116,7 @@ impl Handler for FanInHandler { }; if let (Some(ref sha), Some(_)) = (&best_head_sha, services.git_state()) { - crate::engine::git_merge_ff_only(&*services.sandbox, sha).await; + crate::sandbox_git::git_merge_ff_only(&*services.sandbox, sha).await; } let mut outcome = Outcome::success(); @@ -231,8 +231,8 @@ async fn llm_evaluate( ); // Write prompt to logs - let visit = crate::engine::visit_from_context(context); - let stage_dir = crate::engine::node_dir(run_dir, node_id, visit); + let visit = crate::run_dir::visit_from_context(context); + let stage_dir = crate::run_dir::node_dir(run_dir, node_id, visit); tokio::fs::create_dir_all(&stage_dir).await?; tokio::fs::write(stage_dir.join("prompt.md"), &full_prompt).await?; diff --git a/lib/crates/fabro-workflows/src/handler/manager_loop.rs b/lib/crates/fabro-workflows/src/handler/manager_loop.rs index 1cb563e7d..af6be3175 100644 --- a/lib/crates/fabro-workflows/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflows/src/handler/manager_loop.rs @@ -9,10 +9,11 @@ use async_trait::async_trait; use crate::condition::evaluate_condition; use crate::context::keys; use crate::context::{Context, WorkflowContext}; -use crate::engine::{RunSettings, WorkflowRunEngine}; +use crate::engine::WorkflowRunEngine; use crate::error::FabroError; +use crate::operations::{create, create_from_file, CreateOptions}; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; -use crate::workflow::{prepare_from_file, prepare_from_source}; +use crate::run_settings::RunSettings; use fabro_graphviz::graph::{Graph, Node}; use super::{EngineServices, Handler}; @@ -52,7 +53,10 @@ fn parse_child_graph(node: &Node) -> Result { .get("stack.child_dot_source") .and_then(|v| v.as_str()) { - return prepare_from_source(dot); + let validated = create(dot, CreateOptions::default())?; + validated.raise_on_errors()?; + let (graph, _, _) = validated.into_parts(); + return Ok(graph); } if let Some(path) = node .attrs @@ -60,8 +64,9 @@ fn parse_child_graph(node: &Node) -> Result { .or_else(|| node.attrs.get("stack.child_dotfile")) .and_then(|v| v.as_str()) { - let (graph, diagnostics) = prepare_from_file(std::path::Path::new(path))?; - fabro_validate::raise_on_errors(&diagnostics)?; + let validated = create_from_file(std::path::Path::new(path))?; + validated.raise_on_errors()?; + let (graph, _, _) = validated.into_parts(); return Ok(graph); } Err(FabroError::handler("No child workflow source".to_string())) @@ -128,7 +133,7 @@ impl Handler for SubWorkflowHandler { }; // Build child RunSettings - let visit = crate::engine::visit_from_context(context) as u64; + let visit = crate::run_dir::visit_from_context(context) as u64; let child_logs = run_dir.join(format!("nodes/{}_{visit}/child", node.id)); let _ = std::fs::create_dir_all(&child_logs); diff --git a/lib/crates/fabro-workflows/src/handler/mod.rs b/lib/crates/fabro-workflows/src/handler/mod.rs index 7d7809069..0379e3481 100644 --- a/lib/crates/fabro-workflows/src/handler/mod.rs +++ b/lib/crates/fabro-workflows/src/handler/mod.rs @@ -19,10 +19,10 @@ use async_trait::async_trait; use fabro_agent::Sandbox; use crate::context::Context; -use crate::engine::GitState; use crate::error::FabroError; use crate::event::EventEmitter; use crate::outcome::{Outcome, OutcomeExt}; +use crate::sandbox_git::GitState; use fabro_graphviz::graph::{shape_to_handler_type, Graph, Node}; use fabro_hooks::{HookContext, HookDecision, HookRunner}; use fabro_interview::Interviewer; diff --git a/lib/crates/fabro-workflows/src/handler/parallel.rs b/lib/crates/fabro-workflows/src/handler/parallel.rs index 501b1859c..1e6a87b55 100644 --- a/lib/crates/fabro-workflows/src/handler/parallel.rs +++ b/lib/crates/fabro-workflows/src/handler/parallel.rs @@ -8,9 +8,9 @@ use tokio::sync::Semaphore; use crate::context::keys; use crate::context::{Context, WorkflowContext}; -use crate::engine::set_hook_node; use crate::error::FabroError; use crate::event::WorkflowRunEvent; +use crate::graph_ops::set_hook_node; use crate::millis_u64; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; use fabro_graphviz::graph::{Graph, Node}; @@ -162,7 +162,7 @@ impl Handler for ParallelHandler { // --- Git isolation: checkpoint "parallel base" before fan-out --- let base_sha: Option = if let Some(ref gs) = git_state { - let result = crate::engine::git_checkpoint( + let result = crate::sandbox_git::git_checkpoint( &*services.sandbox, &gs.run_id, &node.id, @@ -205,7 +205,7 @@ impl Handler for ParallelHandler { (&git_state, &base_sha) { let branch_key = &target_id; - let visit = crate::engine::visit_from_context(&branch_context); + let visit = crate::run_dir::visit_from_context(&branch_context); let branch_name = format!( "fabro/run/parallel/{}/{}/pass{}/{}", gs.run_id, @@ -327,7 +327,7 @@ impl Handler for ParallelHandler { let nid = &setup.target_id; let status_str = outcome.status.to_string(); // Use exec_command to commit and capture HEAD in the branch worktree - let git_r = crate::engine::GIT_REMOTE; + let git_r = crate::sandbox_git::GIT_REMOTE; let add_cmd = format!("{git_r} add -A"); let add_result = setup .sandbox @@ -414,7 +414,7 @@ impl Handler for ParallelHandler { for result in &results { if let Some(ref wt_path) = result.worktree_path { let wt_str = wt_path.to_string_lossy().into_owned(); - crate::engine::git_remove_worktree(&*services.sandbox, &wt_str).await; + crate::sandbox_git::git_remove_worktree(&*services.sandbox, &wt_str).await; services .emitter .emit(&WorkflowRunEvent::GitWorktreeRemove { path: wt_str }); @@ -431,7 +431,7 @@ impl Handler for ParallelHandler { successful.sort_by(|a, b| a.id.cmp(&b.id)); if let Some(winner) = successful.first() { let sha = winner.head_sha.as_ref().unwrap(); - crate::engine::git_merge_ff_only(&*services.sandbox, sha).await; + crate::sandbox_git::git_merge_ff_only(&*services.sandbox, sha).await; } } @@ -463,8 +463,8 @@ impl Handler for ParallelHandler { context.set(keys::PARALLEL_RESULTS, serde_json::json!(results_json)); context.set(keys::PARALLEL_BRANCH_COUNT, serde_json::json!(total)); - let visit = crate::engine::visit_from_context(context); - let node_dir = crate::engine::node_dir(run_dir, &node.id, visit); + let visit = crate::run_dir::visit_from_context(context); + let node_dir = crate::run_dir::node_dir(run_dir, &node.id, visit); let _ = tokio::fs::create_dir_all(&node_dir).await; if let Ok(json) = serde_json::to_string_pretty(&results_json) { let _ = tokio::fs::write(node_dir.join("parallel_results.json"), json).await; diff --git a/lib/crates/fabro-workflows/src/handler/prompt.rs b/lib/crates/fabro-workflows/src/handler/prompt.rs index cb137abea..7b2082b01 100644 --- a/lib/crates/fabro-workflows/src/handler/prompt.rs +++ b/lib/crates/fabro-workflows/src/handler/prompt.rs @@ -86,8 +86,8 @@ impl Handler for PromptHandler { }; // 2. Write prompt to logs - let visit = crate::engine::visit_from_context(context); - let stage_dir = crate::engine::node_dir(run_dir, &node.id, visit); + let visit = crate::run_dir::visit_from_context(context); + let stage_dir = crate::run_dir::node_dir(run_dir, &node.id, visit); tokio::fs::create_dir_all(&stage_dir).await?; tokio::fs::write(stage_dir.join("prompt.md"), &prompt).await?; diff --git a/lib/crates/fabro-workflows/src/lib.rs b/lib/crates/fabro-workflows/src/lib.rs index 3df4eb802..478809cfb 100644 --- a/lib/crates/fabro-workflows/src/lib.rs +++ b/lib/crates/fabro-workflows/src/lib.rs @@ -106,6 +106,7 @@ pub mod git; pub mod graph_ops; pub mod graph_render; pub mod handler; +pub mod operations; pub mod outcome; pub mod pipeline; pub mod preamble; @@ -123,6 +124,8 @@ pub mod sandbox_reconnect; pub mod sandbox_record; pub mod start_record; pub mod stylesheet; +#[doc(hidden)] +pub mod test_support; pub mod transform; pub mod vars; pub mod workflow; diff --git a/lib/crates/fabro-workflows/src/operations/create.rs b/lib/crates/fabro-workflows/src/operations/create.rs new file mode 100644 index 000000000..de12204f5 --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/create.rs @@ -0,0 +1,191 @@ +use std::path::{Path, PathBuf}; + +use fabro_graphviz::graph::Graph; + +use crate::error::FabroError; +use crate::pipeline::{self, TransformOptions, Validated}; +use crate::transform::Transform; + +pub struct CreateOptions { + pub base_dir: Option, + pub custom_transforms: Vec>, +} + +impl Default for CreateOptions { + fn default() -> Self { + Self { + base_dir: None, + custom_transforms: Vec::new(), + } + } +} + +/// Parse, transform, and validate a DOT source string. +/// +/// Returns `Validated` even when validation produced errors. Call +/// `validated.raise_on_errors()` if the caller wants to fail fast. +pub fn create(dot_source: &str, options: CreateOptions) -> Result { + let parsed = pipeline::parse(dot_source)?; + let transformed = pipeline::transform( + parsed, + &TransformOptions { + base_dir: options.base_dir, + custom_transforms: options.custom_transforms, + }, + ); + Ok(pipeline::validate(transformed, &[])) +} + +/// Read a DOT file, apply file inlining from its parent directory, then create. +pub fn create_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(".")); + create( + &source, + CreateOptions { + base_dir: Some(base_dir.to_path_buf()), + ..Default::default() + }, + ) +} + +/// Build a validated workflow from an already-materialized graph. +/// +/// This is used by detached/resume CLI paths that load a graph from `RunRecord` +/// instead of re-parsing DOT source. +#[doc(hidden)] +pub fn create_from_graph(graph: Graph, source: impl Into) -> Validated { + Validated::new(graph, source.into(), vec![]) +} + +#[cfg(test)] +mod tests { + use super::*; + use fabro_graphviz::graph::AttrValue; + + const MINIMAL_DOT: &str = r#"digraph Test { + graph [goal="Build feature"] + start [shape=Mdiamond] + exit [shape=Msquare] + start -> exit + }"#; + + #[test] + fn create_minimal() { + let validated = create(MINIMAL_DOT, CreateOptions::default()).unwrap(); + validated.raise_on_errors().unwrap(); + + assert_eq!(validated.graph().name, "Test"); + assert!(validated.graph().find_start_node().is_some()); + assert!(validated.graph().find_exit_node().is_some()); + } + + #[test] + fn create_applies_variable_expansion() { + let dot = r#"digraph Test { + graph [goal="Fix bugs"] + start [shape=Mdiamond] + work [prompt="Goal: $goal"] + exit [shape=Msquare] + start -> work -> exit + }"#; + let validated = create(dot, CreateOptions::default()).unwrap(); + validated.raise_on_errors().unwrap(); + + let prompt = validated.graph().nodes["work"] + .attrs + .get("prompt") + .and_then(AttrValue::as_str) + .unwrap(); + assert_eq!(prompt, "Goal: Fix bugs"); + } + + #[test] + fn create_applies_stylesheet() { + let dot = r#"digraph Test { + graph [goal="Test", model_stylesheet="* { model: sonnet; }"] + start [shape=Mdiamond] + work [label="Work"] + exit [shape=Msquare] + start -> work -> exit + }"#; + let validated = create(dot, CreateOptions::default()).unwrap(); + validated.raise_on_errors().unwrap(); + + assert_eq!( + validated.graph().nodes["work"].attrs.get("model"), + Some(&AttrValue::String("claude-sonnet-4-6".into())) + ); + } + + #[test] + fn create_returns_error_on_invalid_dot() { + let result = create("not a graph", CreateOptions::default()); + assert!(result.is_err()); + } + + #[test] + fn create_returns_validation_diagnostics() { + let dot = r#"digraph Test { + graph [goal="Test"] + work [label="Work"] + }"#; + let validated = create(dot, CreateOptions::default()).unwrap(); + + assert!(validated.has_errors()); + assert!(validated.raise_on_errors().is_err()); + } + + #[test] + fn create_supports_custom_transforms() { + struct TagTransform; + + impl Transform for TagTransform { + fn apply(&self, graph: &mut fabro_graphviz::graph::Graph) { + for node in graph.nodes.values_mut() { + node.attrs + .insert("tagged".to_string(), AttrValue::Boolean(true)); + } + } + } + + let validated = create( + MINIMAL_DOT, + CreateOptions { + custom_transforms: vec![Box::new(TagTransform)], + ..Default::default() + }, + ) + .unwrap(); + validated.raise_on_errors().unwrap(); + + assert_eq!( + validated.graph().nodes["start"].attrs.get("tagged"), + Some(&AttrValue::Boolean(true)) + ); + } + + #[test] + fn create_from_file_uses_parent_directory_for_inlining() { + let dir = tempfile::tempdir().unwrap(); + let data_path = dir.path().join("goal.txt"); + let dot_path = dir.path().join("workflow.fabro"); + + std::fs::write(&data_path, "ship it").unwrap(); + std::fs::write( + &dot_path, + r#"digraph Test { + graph [goal="@goal.txt"] + start [shape=Mdiamond] + exit [shape=Msquare] + start -> exit + }"#, + ) + .unwrap(); + + let validated = create_from_file(&dot_path).unwrap(); + validated.raise_on_errors().unwrap(); + assert_eq!(validated.graph().goal(), "ship it"); + } +} diff --git a/lib/crates/fabro-workflows/src/operations/fork.rs b/lib/crates/fabro-workflows/src/operations/fork.rs new file mode 100644 index 000000000..fc90afbac --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/fork.rs @@ -0,0 +1 @@ +pub use crate::run_fork::execute_fork as fork; diff --git a/lib/crates/fabro-workflows/src/operations/mod.rs b/lib/crates/fabro-workflows/src/operations/mod.rs new file mode 100644 index 000000000..59da571f3 --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/mod.rs @@ -0,0 +1,12 @@ +mod create; +mod fork; +mod rewind; +mod start; + +pub use create::{create, create_from_file, create_from_graph, CreateOptions}; +pub use fork::fork; +pub use rewind::{ + build_timeline, find_run_id_by_prefix, load_parallel_map, parse_target, resolve_target, + rewind, TimelineEntry, +}; +pub use start::{start, StartFinalizeConfig, StartOptions, StartRetroConfig, Started}; diff --git a/lib/crates/fabro-workflows/src/operations/rewind.rs b/lib/crates/fabro-workflows/src/operations/rewind.rs new file mode 100644 index 000000000..d578e16b5 --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/rewind.rs @@ -0,0 +1,4 @@ +pub use crate::run_rewind::{ + build_timeline, execute_rewind as rewind, find_run_id_by_prefix, load_parallel_map, + parse_target, resolve_target, TimelineEntry, +}; diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs new file mode 100644 index 000000000..5f4205903 --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/start.rs @@ -0,0 +1,117 @@ +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use crate::error::FabroError; +use crate::event::WorkflowRunEvent; +use crate::outcome::StageStatus; +use crate::pipeline::{self, FinalizeOptions, Finalized, InitOptions, RetroOptions, Validated}; + +pub struct StartRetroConfig { + pub enabled: bool, + pub dry_run: bool, + pub llm_client: Option, + pub provider: fabro_llm::Provider, + pub model: String, +} + +pub struct StartFinalizeConfig { + pub preserve_sandbox: bool, + pub pr_config: Option, + pub github_app: Option, + pub origin_url: Option, + pub model: String, +} + +pub struct StartOptions { + pub init: InitOptions, + pub retro: StartRetroConfig, + pub finalize: StartFinalizeConfig, +} + +pub struct Started { + pub finalized: Finalized, + pub retro: Option, + pub retro_duration: Duration, +} + +/// Run a validated workflow through initialize, execute, retro, and finalize. +pub async fn start(validated: Validated, options: StartOptions) -> Result { + let preserve_sandbox = options.finalize.preserve_sandbox; + let sandbox_for_cleanup = Arc::clone(&options.init.sandbox); + let cleanup_guard = scopeguard::guard((), move |()| { + if preserve_sandbox { + return; + } + if let Ok(handle) = tokio::runtime::Handle::try_current() { + handle.spawn(async move { + let _ = sandbox_for_cleanup.cleanup().await; + }); + } + }); + + let initialized = pipeline::initialize(validated, options.init).await?; + + let last_git_sha: Arc>> = Arc::new(Mutex::new(None)); + { + let sha_clone = Arc::clone(&last_git_sha); + initialized.emitter.on_event(move |event| { + if let WorkflowRunEvent::CheckpointCompleted { + git_commit_sha: Some(sha), + .. + } = event + { + *sha_clone.lock().unwrap() = Some(sha.clone()); + } + }); + } + + let executed = pipeline::execute(initialized).await; + let failed = !matches!( + executed.outcome.as_ref().map(|outcome| &outcome.status), + Ok(StageStatus::Success) | Ok(StageStatus::PartialSuccess) + ); + + let retro_opts = RetroOptions { + run_id: executed.settings.run_id.clone(), + workflow_name: executed.graph.name.clone(), + goal: executed.graph.goal().to_string(), + run_dir: executed.settings.run_dir.clone(), + sandbox: Arc::clone(&executed.sandbox), + emitter: Some(Arc::clone(&executed.emitter)), + failed, + run_duration_ms: executed.duration_ms, + enabled: options.retro.enabled, + dry_run: options.retro.dry_run, + llm_client: options.retro.llm_client, + provider: options.retro.provider, + model: options.retro.model, + }; + + let retro_start = Instant::now(); + let retroed = pipeline::retro(executed, &retro_opts).await; + let retro_duration = retro_start.elapsed(); + + let finalize_opts = FinalizeOptions { + run_dir: retroed.settings.run_dir.clone(), + run_id: retroed.settings.run_id.clone(), + workflow_name: retroed.graph.name.clone(), + hook_runner: retroed.hook_runner.clone(), + preserve_sandbox: options.finalize.preserve_sandbox, + pr_config: options.finalize.pr_config, + github_app: options.finalize.github_app, + origin_url: options.finalize.origin_url, + model: options.finalize.model, + last_git_sha: last_git_sha.lock().unwrap().clone(), + }; + + let retro = retroed.retro.clone(); + let finalized = pipeline::finalize(retroed, &finalize_opts).await?; + + scopeguard::ScopeGuard::into_inner(cleanup_guard); + + Ok(Started { + finalized, + retro, + retro_duration, + }) +} diff --git a/lib/crates/fabro-workflows/src/pipeline/finalize.rs b/lib/crates/fabro-workflows/src/pipeline/finalize.rs index 4211e2047..681326d83 100644 --- a/lib/crates/fabro-workflows/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflows/src/pipeline/finalize.rs @@ -357,6 +357,7 @@ pub async fn finalize( run_id: settings.run_id, outcome, conclusion, + pushed_branch: settings.git.as_ref().and_then(|g| g.run_branch.clone()), pr_url, }) } diff --git a/lib/crates/fabro-workflows/src/pipeline/types.rs b/lib/crates/fabro-workflows/src/pipeline/types.rs index 1923f89c4..768a40c1a 100644 --- a/lib/crates/fabro-workflows/src/pipeline/types.rs +++ b/lib/crates/fabro-workflows/src/pipeline/types.rs @@ -158,6 +158,7 @@ pub struct Finalized { pub run_id: String, pub outcome: Result, pub conclusion: Conclusion, + pub pushed_branch: Option, pub pr_url: Option, } diff --git a/lib/crates/fabro-workflows/src/test_support.rs b/lib/crates/fabro-workflows/src/test_support.rs new file mode 100644 index 000000000..ef847f39f --- /dev/null +++ b/lib/crates/fabro-workflows/src/test_support.rs @@ -0,0 +1,119 @@ +use std::collections::HashMap; +use std::sync::Arc; + +use fabro_agent::Sandbox; + +use crate::checkpoint::Checkpoint; +use crate::engine::WorkflowRunEngine; +use crate::error::Result; +use crate::event::EventEmitter; +use crate::handler::HandlerRegistry; +use crate::outcome::Outcome; +use crate::run_settings::RunSettings; + +pub async fn run_graph( + registry: HandlerRegistry, + emitter: Arc, + sandbox: Arc, + graph: &fabro_graphviz::graph::Graph, + settings: &RunSettings, +) -> Result { + let engine = WorkflowRunEngine::new(registry, emitter, sandbox); + engine.run(graph, settings).await +} + +pub async fn run_graph_with_hooks( + registry: HandlerRegistry, + emitter: Arc, + sandbox: Arc, + graph: &fabro_graphviz::graph::Graph, + settings: &RunSettings, + hook_runner: Arc, + env: Option>, +) -> Result { + let mut engine = WorkflowRunEngine::new(registry, emitter, sandbox); + engine.set_hook_runner(hook_runner); + if let Some(env) = env { + engine.set_env(env); + } + engine.run(graph, settings).await +} + +pub async fn run_graph_from_checkpoint( + registry: HandlerRegistry, + emitter: Arc, + sandbox: Arc, + graph: &fabro_graphviz::graph::Graph, + settings: &RunSettings, + checkpoint: &Checkpoint, +) -> Result { + let engine = WorkflowRunEngine::new(registry, emitter, sandbox); + engine + .run_from_checkpoint(graph, settings, checkpoint) + .await +} + +pub struct WorkflowRunner { + registry: std::sync::Mutex>, + emitter: Arc, + sandbox: Arc, +} + +impl WorkflowRunner { + #[must_use] + pub fn new( + registry: HandlerRegistry, + emitter: Arc, + sandbox: Arc, + ) -> Self { + Self { + registry: std::sync::Mutex::new(Some(registry)), + emitter, + sandbox, + } + } + + pub async fn run( + &self, + graph: &fabro_graphviz::graph::Graph, + settings: &RunSettings, + ) -> Result { + let registry = self + .registry + .lock() + .unwrap() + .take() + .expect("WorkflowRunner may only be used once"); + run_graph( + registry, + Arc::clone(&self.emitter), + Arc::clone(&self.sandbox), + graph, + settings, + ) + .await + } + + pub async fn run_from_checkpoint( + &self, + graph: &fabro_graphviz::graph::Graph, + settings: &RunSettings, + checkpoint: &Checkpoint, + ) -> Result { + let registry = self + .registry + .lock() + .unwrap() + .take() + .expect("WorkflowRunner may only be used once"); + run_graph_from_checkpoint( + registry, + Arc::clone(&self.emitter), + Arc::clone(&self.sandbox), + graph, + settings, + checkpoint, + ) + .await + } +} diff --git a/lib/crates/fabro-workflows/tests/daytona_integration.rs b/lib/crates/fabro-workflows/tests/daytona_integration.rs index b13ea6c1b..95e769a94 100644 --- a/lib/crates/fabro-workflows/tests/daytona_integration.rs +++ b/lib/crates/fabro-workflows/tests/daytona_integration.rs @@ -15,13 +15,14 @@ use fabro_sandbox::daytona::{DaytonaConfig, DaytonaSandbox, DaytonaSnapshotConfi use fabro_workflows::artifact::sync_artifacts_to_env; use fabro_workflows::checkpoint::Checkpoint; use fabro_workflows::context::Context; -use fabro_workflows::engine::{GitCheckpointSettings, RunSettings, WorkflowRunEngine}; use fabro_workflows::error::FabroError; use fabro_workflows::event::EventEmitter; use fabro_workflows::handler::exit::ExitHandler; use fabro_workflows::handler::start::StartHandler; use fabro_workflows::handler::{Handler, HandlerRegistry}; use fabro_workflows::outcome::{Outcome, OutcomeExt, StageStatus}; +use fabro_workflows::run_settings::{GitCheckpointSettings, RunSettings}; +use fabro_workflows::test_support::WorkflowRunner; async fn create_env() -> DaytonaSandbox { let creds = load_github_app_credentials(); @@ -386,7 +387,7 @@ async fn daytona_pipeline_artifact_offload_and_sync() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env.clone()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -577,7 +578,7 @@ async fn daytona_git_checkpoint_remote_emits_events() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), env.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), env.clone()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -762,7 +763,7 @@ async fn daytona_parallel_git_branching_e2e() { registry.register("parallel", Box::new(ParallelHandler)); registry.register("parallel.fan_in", Box::new(FanInHandler::new(None))); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), Arc::clone(&env)); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), Arc::clone(&env)); let config = RunSettings { config: FabroConfig::default(), @@ -1139,7 +1140,7 @@ async fn daytona_git_checkpoint_with_shadow_branch() { registry.register("exit", Box::new(ExitHandler)); let meta_branch = MetadataStore::branch_name(&run_id); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env.clone()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -1246,7 +1247,7 @@ async fn daytona_asset_collection() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env.clone()); let mut graph = Graph::new("DaytonaAssetTest"); graph.attrs.insert( @@ -1535,7 +1536,7 @@ async fn daytona_git_push_run_branch_to_origin() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env.clone()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), diff --git a/lib/crates/fabro-workflows/tests/integration.rs b/lib/crates/fabro-workflows/tests/integration.rs index 58880046f..affa10edf 100644 --- a/lib/crates/fabro-workflows/tests/integration.rs +++ b/lib/crates/fabro-workflows/tests/integration.rs @@ -16,7 +16,6 @@ use fabro_workflows::backend::cli::{parse_cli_response, AgentCliBackend, Backend use fabro_workflows::backend::AgentApiBackend; use fabro_workflows::checkpoint::Checkpoint; use fabro_workflows::context::Context; -use fabro_workflows::engine::{GitCheckpointSettings, RunSettings, WorkflowRunEngine}; use fabro_workflows::error::FabroError; use fabro_workflows::event::{EventEmitter, WorkflowRunEvent}; use fabro_workflows::handler::agent::{AgentHandler, CodergenBackend, CodergenResult}; @@ -30,7 +29,9 @@ use fabro_workflows::handler::start::StartHandler; use fabro_workflows::handler::wait::WaitHandler; use fabro_workflows::handler::{Handler, HandlerRegistry}; use fabro_workflows::outcome::{Outcome, OutcomeExt, StageStatus}; +use fabro_workflows::run_settings::{GitCheckpointSettings, RunSettings}; use fabro_workflows::stylesheet::{apply_stylesheet, parse_stylesheet}; +use fabro_workflows::test_support::{run_graph_with_hooks, WorkflowRunner}; use fabro_workflows::transform::{ StylesheetApplicationTransform, Transform, VariableExpansionTransform, }; @@ -187,7 +188,7 @@ async fn end_to_end_linear_pipeline() { validate_or_raise(&graph, &[]).expect("validation should pass"); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_linear_registry(), Arc::new(EventEmitter::new()), local_env(), @@ -330,7 +331,7 @@ async fn end_to_end_branching_pipeline() { registry.register("agent", Box::new(AgentHandler::new(None))); registry.register("conditional", Box::new(ConditionalHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -450,7 +451,7 @@ async fn end_to_end_human_gate_pipeline() { registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -546,7 +547,7 @@ async fn human_gate_aborted_input_fails_closed_without_fail_route() { registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -657,7 +658,7 @@ async fn human_gate_aborted_input_routes_via_outcome_fail_condition() { registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -770,7 +771,7 @@ async fn goal_gate_routes_to_retry_target_on_failure() { registry.register("exit", Box::new(ExitHandler)); registry.register("always_fail", Box::new(AlwaysFailHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -891,7 +892,7 @@ async fn goal_gate_routes_to_retry_target_when_present() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -1203,7 +1204,7 @@ async fn retry_on_failure_then_succeed() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -1274,7 +1275,7 @@ async fn pipeline_with_many_nodes() { .push(Edge::new(node_names.last().unwrap(), "exit")); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_linear_registry(), Arc::new(EventEmitter::new()), local_env(), @@ -1600,7 +1601,7 @@ async fn smoke_test_with_mock_codergen_backend() { ); registry.register("conditional", Box::new(ConditionalHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -1701,7 +1702,7 @@ async fn end_to_end_parallel_fan_out_fan_in() { Box::new(FanInHandler::new(Some(Box::new(MockCodergenBackend)))), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -1814,7 +1815,7 @@ async fn resume_from_checkpoint_completes_pipeline() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -1913,7 +1914,7 @@ async fn resume_from_checkpoint_preserves_goal_gate_outcomes() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -1952,7 +1953,7 @@ async fn graph_goal_in_context() { }"#; let graph = parse(input).expect("parse"); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_linear_registry(), Arc::new(EventEmitter::new()), local_env(), @@ -1992,7 +1993,7 @@ async fn event_streaming_lifecycle() { let dir = tempfile::tempdir().unwrap(); let emitter = EventEmitter::new(); let events = collect_events(&emitter); - let engine = WorkflowRunEngine::new(make_linear_registry(), Arc::new(emitter), local_env()); + let engine = WorkflowRunner::new(make_linear_registry(), Arc::new(emitter), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -2068,7 +2069,7 @@ async fn context_flow_between_stages() { graph.edges.push(Edge::new("step_b", "exit")); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_linear_registry(), Arc::new(EventEmitter::new()), local_env(), @@ -2121,7 +2122,7 @@ async fn tool_handler_e2e() { let dir = tempfile::tempdir().unwrap(); let interviewer = Arc::new(AutoApproveInterviewer); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_full_registry(interviewer), Arc::new(EventEmitter::new()), local_env(), @@ -2191,7 +2192,7 @@ async fn auto_approve_interviewer_e2e() { let dir = tempfile::tempdir().unwrap(); let interviewer = Arc::new(AutoApproveInterviewer); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_full_registry(interviewer), Arc::new(EventEmitter::new()), local_env(), @@ -2228,7 +2229,7 @@ async fn codergen_without_backend_simulated() { }"#; let graph = parse(input).expect("parse"); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_linear_registry(), Arc::new(EventEmitter::new()), local_env(), @@ -2337,7 +2338,7 @@ async fn branching_loop_back_on_failure() { call_count: std::sync::atomic::AtomicU32::new(0), }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -2420,7 +2421,7 @@ async fn human_gate_loops_back() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -2474,7 +2475,7 @@ async fn scenario_ship_a_feature() { let dir = tempfile::tempdir().unwrap(); let emitter = EventEmitter::new(); let events = collect_events(&emitter); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_full_registry(interviewer), Arc::new(emitter), local_env(), @@ -2564,7 +2565,7 @@ async fn scenario_parallel_expert_review() { ); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -2648,7 +2649,7 @@ async fn scenario_node_retries_on_retry_status() { call_count: std::sync::atomic::AtomicU32::new(0), }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -2710,7 +2711,7 @@ async fn scenario_loop_restart_resets_context() { call_count: Arc::clone(&call_count), }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -2778,7 +2779,7 @@ async fn scenario_bug_triage_router() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); registry.register("conditional", Box::new(ConditionalHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -2837,7 +2838,7 @@ async fn scenario_crash_recovery() { let mut registry = HandlerRegistry::new(Box::new(StartHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -2946,7 +2947,7 @@ async fn manager_loop_stop_condition_satisfied_e2e() { registry.register("exit", Box::new(ExitHandler)); registry.register("done_setter", Box::new(DoneSetterHandler)); registry.register("stack.manager_loop", Box::new(SubWorkflowHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3023,7 +3024,7 @@ async fn manager_loop_max_cycles_exceeded_e2e() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); registry.register("stack.manager_loop", Box::new(SubWorkflowHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3159,7 +3160,7 @@ async fn conditional_branching_success_fail_paths() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); registry.register("always_fail", Box::new(AlwaysFailHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3212,7 +3213,7 @@ async fn edge_selection_condition_match_wins_over_weight() { let mut registry = HandlerRegistry::new(Box::new(StartHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3259,7 +3260,7 @@ async fn edge_selection_weight_breaks_ties() { let mut registry = HandlerRegistry::new(Box::new(StartHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3298,7 +3299,7 @@ async fn edge_selection_lexical_tiebreak() { let mut registry = HandlerRegistry::new(Box::new(StartHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3356,7 +3357,7 @@ async fn context_updates_visible_across_nodes() { registry.register("exit", Box::new(ExitHandler)); registry.register("conditional", Box::new(ConditionalHandler)); registry.register("context_setter", Box::new(ContextSetterHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3396,7 +3397,7 @@ async fn stylesheet_applies_model_override() { assert_eq!(graph.nodes["work"].model(), Some("custom-model")); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_linear_registry(), Arc::new(EventEmitter::new()), local_env(), @@ -3456,7 +3457,7 @@ async fn custom_handler_registration_and_execution() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); registry.register("my_custom", Box::new(CustomHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3523,7 +3524,7 @@ async fn integration_smoke_plan_implement_review_done() { let dir = tempfile::tempdir().unwrap(); let emitter = EventEmitter::new(); let events = collect_events(&emitter); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_full_registry(interviewer), Arc::new(emitter), local_env(), @@ -3631,7 +3632,7 @@ async fn manager_loop_runs_child_engine_e2e() { registry.register("exit", Box::new(ExitHandler)); registry.register("stack.manager_loop", Box::new(SubWorkflowHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3765,7 +3766,7 @@ async fn manager_loop_context_flows_e2e() { registry.register("setter", Box::new(SetterHandler)); registry.register("stack.manager_loop", Box::new(SubWorkflowHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3838,7 +3839,7 @@ async fn manager_loop_child_dotfile_e2e() { registry.register("exit", Box::new(ExitHandler)); registry.register("stack.manager_loop", Box::new(SubWorkflowHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -3947,7 +3948,7 @@ async fn graph_merge_e2e_through_engine() { assert!(main_graph.nodes.contains_key("dep.release")); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_linear_registry(), Arc::new(EventEmitter::new()), local_env(), @@ -4101,7 +4102,7 @@ async fn fidelity_default_is_compact() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4158,7 +4159,7 @@ async fn fidelity_graph_default_applied() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4211,7 +4212,7 @@ async fn fidelity_node_overrides_graph_default() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4270,7 +4271,7 @@ async fn fidelity_edge_overrides_node_and_graph() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4319,7 +4320,7 @@ async fn fidelity_full_produces_empty_preamble() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4378,7 +4379,7 @@ async fn fidelity_truncate_preamble_minimal() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4450,7 +4451,7 @@ async fn fidelity_summary_low_mode() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4517,7 +4518,7 @@ async fn fidelity_summary_medium_mode() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4584,7 +4585,7 @@ async fn fidelity_summary_high_mode() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4644,7 +4645,7 @@ async fn fidelity_full_sets_thread_id_in_context() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4715,7 +4716,7 @@ async fn fidelity_full_nodes_share_thread_id() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4796,7 +4797,7 @@ async fn fidelity_resume_degrades_full_to_summary_high() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4893,7 +4894,7 @@ async fn fidelity_resume_degrade_only_affects_first_hop() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -4977,7 +4978,7 @@ async fn fidelity_resume_no_degrade_when_not_full() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5019,7 +5020,7 @@ async fn fidelity_stored_in_checkpoint_context() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5105,7 +5106,7 @@ async fn fidelity_precedence_multi_node_pipeline() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5173,7 +5174,7 @@ async fn fidelity_compact_preamble_includes_completed_stages_and_context() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5248,8 +5249,7 @@ async fn fidelity_summary_low_excludes_context_values_in_pipeline() { captures: captures_low.clone(), }), ); - let engine_low = - WorkflowRunEngine::new(registry_low, Arc::new(EventEmitter::new()), local_env()); + let engine_low = WorkflowRunner::new(registry_low, Arc::new(EventEmitter::new()), local_env()); let config_low = RunSettings { config: FabroConfig::default(), run_dir: dir_low.path().to_path_buf(), @@ -5316,8 +5316,7 @@ async fn fidelity_summary_low_excludes_context_values_in_pipeline() { captures: captures_med.clone(), }), ); - let engine_med = - WorkflowRunEngine::new(registry_med, Arc::new(EventEmitter::new()), local_env()); + let engine_med = WorkflowRunner::new(registry_med, Arc::new(EventEmitter::new()), local_env()); let config_med = RunSettings { config: FabroConfig::default(), run_dir: dir_med.path().to_path_buf(), @@ -5388,7 +5387,7 @@ async fn fidelity_thread_id_fallback_to_previous_node_in_pipeline() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5442,7 +5441,7 @@ async fn fidelity_thread_id_from_node_class_in_pipeline() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5499,7 +5498,7 @@ async fn fidelity_edge_thread_id_override_in_pipeline() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5557,7 +5556,7 @@ async fn fidelity_full_without_explicit_thread_id_uses_previous_node() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5625,7 +5624,7 @@ async fn fidelity_from_parsed_dot_pipeline() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5673,7 +5672,7 @@ async fn fidelity_checkpoint_roundtrip_preserves_fidelity() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5743,7 +5742,7 @@ async fn fidelity_node_thread_id_overrides_edge_thread_id_in_pipeline() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5830,7 +5829,7 @@ async fn fidelity_resume_preserves_context_values_across_checkpoint() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -5969,13 +5968,14 @@ mod real_llm { use fabro_graphviz::graph::{AttrValue, Edge, Graph}; use fabro_interview::AutoApproveInterviewer; use fabro_workflows::checkpoint::Checkpoint; - use fabro_workflows::engine::{RunSettings, WorkflowRunEngine}; use fabro_workflows::event::EventEmitter; use fabro_workflows::handler::exit::ExitHandler; use fabro_workflows::handler::human::HumanHandler; use fabro_workflows::handler::start::StartHandler; use fabro_workflows::handler::HandlerRegistry; use fabro_workflows::outcome::StageStatus; + use fabro_workflows::run_settings::RunSettings; + use fabro_workflows::test_support::WorkflowRunner; #[tokio::test] #[ignore] @@ -6044,7 +6044,7 @@ mod real_llm { )))), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -6158,7 +6158,7 @@ mod real_llm { Box::new(AgentHandler::new(Some(make_llm_backend(client)))), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -6297,7 +6297,7 @@ mod real_llm { ); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -6404,7 +6404,7 @@ mod real_llm { ))), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -6500,7 +6500,7 @@ async fn human_gate_freeform_only_routes_text() { registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -6630,7 +6630,7 @@ async fn human_gate_freeform_with_fixed_choice_match() { registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -6745,7 +6745,7 @@ async fn human_gate_freeform_fallback_on_unmatched_text() { registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -6873,7 +6873,7 @@ async fn human_gate_freeform_sets_allow_freeform_on_question() { registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -6981,7 +6981,7 @@ async fn human_gate_without_freeform_sets_allow_freeform_false() { registry.register("exit", Box::new(ExitHandler)); registry.register("human", Box::new(HumanHandler::new(interviewer))); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -7207,38 +7207,59 @@ fn subgraph_without_label_no_class_derived() { // Hook System E2E Tests // --------------------------------------------------------------------------- -/// Helper: create a WorkflowRunEngine with hooks configured from HookDefinitions. -fn engine_with_hooks(hooks: Vec) -> WorkflowRunEngine { - let registry = make_linear_registry(); - let emitter = Arc::new(EventEmitter::new()); - let sandbox = local_env(); - let mut engine = WorkflowRunEngine::new(registry, emitter, sandbox); - if !hooks.is_empty() { - let config = fabro_hooks::HookConfig { hooks }; - let runner = fabro_hooks::HookRunner::new(config); - engine.set_hook_runner(Arc::new(runner)); - } - engine +fn hook_runner_from_defs(hooks: Vec) -> Arc { + Arc::new(fabro_hooks::HookRunner::new(fabro_hooks::HookConfig { + hooks, + })) } -/// Helper: create a WorkflowRunEngine with hooks and event capture. -fn engine_with_hooks_and_events( - hooks: Vec, -) -> ( - WorkflowRunEngine, +struct HookTestRunner { + emitter: Arc, + hook_runner: Arc, +} + +impl HookTestRunner { + async fn run(&self, graph: &Graph, config: &RunSettings) -> Result { + run_graph_with_hooks( + make_linear_registry(), + Arc::clone(&self.emitter), + local_env(), + graph, + config, + Arc::clone(&self.hook_runner), + None, + ) + .await + } +} + +fn emitter_with_events() -> ( + Arc, Arc>>, ) { - let registry = make_linear_registry(); let emitter = EventEmitter::new(); let events = collect_events(&emitter); - let sandbox = local_env(); - let mut engine = WorkflowRunEngine::new(registry, Arc::new(emitter), sandbox); - if !hooks.is_empty() { - let config = fabro_hooks::HookConfig { hooks }; - let runner = fabro_hooks::HookRunner::new(config); - engine.set_hook_runner(Arc::new(runner)); + (Arc::new(emitter), events) +} + +fn engine_with_hooks(hooks: Vec) -> HookTestRunner { + HookTestRunner { + emitter: Arc::new(EventEmitter::new()), + hook_runner: hook_runner_from_defs(hooks), } - (engine, events) +} + +fn engine_with_hooks_and_events( + hooks: Vec, +) -> (HookTestRunner, Arc>>) { + let (emitter, events) = emitter_with_events(); + ( + HookTestRunner { + emitter, + hook_runner: hook_runner_from_defs(hooks), + }, + events, + ) } fn make_run_config(dir: &std::path::Path) -> RunSettings { @@ -8345,7 +8366,7 @@ async fn arc_e2e_with_real_llm() { }); let run_dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: run_dir.path().to_path_buf(), @@ -8473,7 +8494,7 @@ async fn run_fidelity_prompt_pipeline(fidelity: &str) -> String { Box::new(AgentHandler::new(Some(Box::new(MockCodergenBackend)))), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -8672,7 +8693,7 @@ async fn large_context_values_are_offloaded_to_artifact_store() { let emitter = EventEmitter::new(); let events = collect_events(&emitter); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -8891,8 +8912,7 @@ async fn artifact_pointers_rewritten_for_remote_sandbox() { registry.register("exit", Box::new(ExitHandler)); let remote_env = Arc::new(RemoteMockEnv::new("/sandbox")); - let engine = - WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), remote_env.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), remote_env.clone()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -9022,7 +9042,7 @@ async fn node_dir_uses_visit_count_on_revisit() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -9994,7 +10014,7 @@ async fn full_pipeline_with_cli_backend_node() { ); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -10125,7 +10145,7 @@ async fn stylesheet_backend_property_routes_to_cli() { ); let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -10403,7 +10423,7 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() { let mut registry = HandlerRegistry::new(Box::new(ContextSetterHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), env); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), env); let config = RunSettings { config: FabroConfig::default(), @@ -10605,7 +10625,7 @@ async fn git_checkpoint_host_writes_shadow_branch() { let mut registry = HandlerRegistry::new(Box::new(ContextSetterHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), env); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), env); let meta_branch = MetadataStore::branch_name(run_id); let config = RunSettings { @@ -10804,7 +10824,7 @@ async fn parallel_git_branching_host_e2e() { Box::new(FanInHandler::new(None)), // heuristic select — picks branch_a (lexical tiebreak) ); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), env); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), env); let config = RunSettings { config: FabroConfig::default(), @@ -11067,7 +11087,7 @@ async fn git_checkpoint_host_skips_empty_diff_patch() { let mut registry = HandlerRegistry::new(Box::new(ContextSetterHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), env); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), env); let config = RunSettings { config: FabroConfig::default(), @@ -11450,7 +11470,7 @@ async fn e2e_circuit_breaker_deterministic_self_loop() { )), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -11497,7 +11517,7 @@ async fn e2e_circuit_breaker_custom_limit() { Box::new(DeterministicFailHandler::new("same error every time")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -11537,7 +11557,7 @@ async fn e2e_circuit_breaker_ignores_transient_failures() { registry.register("exit", Box::new(ExitHandler)); registry.register("test_handler", Box::new(TransientInfraFailHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -11584,7 +11604,7 @@ async fn e2e_circuit_breaker_different_reasons_separate_counters() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -11624,7 +11644,7 @@ async fn e2e_circuit_breaker_loop_restart() { Box::new(DeterministicFailHandler::new("verify step failed")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -11686,7 +11706,7 @@ async fn e2e_failure_signature_persisted_in_context() { Box::new(DeterministicFailHandler::new("test assertion failed")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -11750,7 +11770,7 @@ async fn e2e_failure_signature_hint_overrides_reason_in_context() { registry.register("exit", Box::new(ExitHandler)); registry.register("hint_handler", Box::new(SignatureHintHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -11806,7 +11826,7 @@ async fn e2e_signature_maps_persist_in_checkpoint() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -11933,7 +11953,7 @@ async fn e2e_circuit_breaker_emits_events_before_abort() { Box::new(DeterministicFailHandler::new("assertion failed")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12000,7 +12020,7 @@ async fn e2e_circuit_breaker_does_not_fire_below_limit() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12096,7 +12116,7 @@ async fn e2e_circuit_breaker_multi_stage_impl_verify_cycle() { )), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12193,7 +12213,7 @@ async fn e2e_loop_restart_blocked_for_deterministic_failure() { Box::new(ClassifiedFailHandler::always("deterministic")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12233,7 +12253,7 @@ async fn e2e_loop_restart_blocked_for_structural_failure() { Box::new(ClassifiedFailHandler::always("structural")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12273,7 +12293,7 @@ async fn e2e_loop_restart_blocked_for_budget_exhausted_failure() { Box::new(ClassifiedFailHandler::always("budget_exhausted")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12313,7 +12333,7 @@ async fn e2e_loop_restart_blocked_for_canceled_failure() { Box::new(ClassifiedFailHandler::always("canceled")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12350,7 +12370,7 @@ async fn e2e_loop_restart_blocked_for_compilation_loop_failure() { Box::new(ClassifiedFailHandler::always("compilation_loop")), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12391,7 +12411,7 @@ async fn e2e_loop_restart_allowed_for_transient_infra() { Box::new(ClassifiedFailHandler::succeed_on("transient_infra", 1)), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12495,7 +12515,7 @@ async fn e2e_stall_watchdog_triggers_from_dot_parsed_pipeline() { events_clone.lock().unwrap().push(format!("{event:?}")); }); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12551,7 +12571,7 @@ async fn e2e_stall_watchdog_kept_alive_by_handler_events() { }), ); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12597,7 +12617,7 @@ async fn e2e_stall_watchdog_disabled_with_zero_timeout() { registry.register("exit", Box::new(ExitHandler)); registry.register("slow", Box::new(SlowTestHandler { sleep_ms: 50 })); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12662,7 +12682,7 @@ async fn e2e_stall_watchdog_with_explicit_timeout_override() { registry.register("exit", Box::new(ExitHandler)); registry.register("hanging", Box::new(HangingHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), local_env()); let config = RunSettings { config: FabroConfig::default(), run_dir: dir.path().to_path_buf(), @@ -12760,7 +12780,7 @@ async fn asset_collection_local_sandbox_success() { let emitter = EventEmitter::new(); let events = collect_events(&emitter); - let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), sandbox.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(emitter), sandbox.clone()); let mut graph = Graph::new("AssetCollectionTest"); graph.attrs.insert( @@ -12874,7 +12894,7 @@ async fn asset_collection_local_sandbox_on_failure() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), sandbox.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), sandbox.clone()); let mut graph = Graph::new("AssetCollectionFailTest"); graph.attrs.insert( @@ -12971,7 +12991,7 @@ async fn asset_collection_docker_sandbox() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), sandbox.clone()); + let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), sandbox.clone()); let mut graph = Graph::new("DockerAssetTest"); graph.attrs.insert( @@ -13074,7 +13094,7 @@ async fn wait_timer_e2e() { let dir = tempfile::tempdir().unwrap(); let interviewer = Arc::new(AutoApproveInterviewer); - let engine = WorkflowRunEngine::new( + let engine = WorkflowRunner::new( make_full_registry(interviewer), Arc::new(EventEmitter::new()), local_env(),