From d45b4516abdbc75079a0b18b270deceba02299e6 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 26 Mar 2026 12:58:47 -0400 Subject: [PATCH] refactor: clean CREATE/START/RESUME separation Resume now follows the same subprocess pattern as run: look up run directory by ID prefix, validate checkpoint exists, clean stale artifacts, reset status to Submitted, spawn _run_engine --resume, and attach. This eliminates ~1600 lines of duplicated env/sandbox setup from resume.rs. Key changes: - operations::start() and operations::resume() take run_dir instead of Persisted, loading state from disk internally - run_engine() builds RunOptions from RunRecord on disk, so callers no longer extract record fields manually - StartOptions flattened (no more nested InitOptions) - FabroError::Precondition variant for start/resume guard checks - _run_engine accepts --resume flag to dispatch to resume path - operations::restore removed (no longer needed) - Resume CLI stripped to just + --detach Co-Authored-By: Claude Opus 4.6 (1M context) --- docs/core-concepts/how-fabro-works.mdx | 8 +- docs/execution/checkpoints.mdx | 31 +- docs/reference/cli.mdx | 22 +- lib/crates/fabro-cli/src/commands/resume.rs | 1688 +---------------- lib/crates/fabro-cli/src/commands/run.rs | 196 +- lib/crates/fabro-cli/src/commands/start.rs | 12 +- lib/crates/fabro-cli/src/main.rs | 87 +- lib/crates/fabro-cli/tests/cli.rs | 85 +- lib/crates/fabro-workflows/src/error.rs | 5 + .../fabro-workflows/src/operations/mod.rs | 5 +- .../fabro-workflows/src/operations/restore.rs | 163 -- .../fabro-workflows/src/operations/start.rs | 232 ++- 12 files changed, 476 insertions(+), 2058 deletions(-) delete mode 100644 lib/crates/fabro-workflows/src/operations/restore.rs diff --git a/docs/core-concepts/how-fabro-works.mdx b/docs/core-concepts/how-fabro-works.mdx index 9251cd8a2..0bd1bc949 100644 --- a/docs/core-concepts/how-fabro-works.mdx +++ b/docs/core-concepts/how-fabro-works.mdx @@ -98,11 +98,5 @@ Because Fabro checkpoints after every stage, interrupted runs can be resumed fro fabro resume ``` -Or resume from a checkpoint file: - -```bash -fabro resume --checkpoint path/to/checkpoint.json --workflow workflow.fabro -``` - -The engine restores the full context, node visit counts, and retry state, then continues execution from the next node. +The engine restores the full context, node visit counts, and retry state from the run directory, then continues execution from the next node. diff --git a/docs/execution/checkpoints.mdx b/docs/execution/checkpoints.mdx index 891fe474c..60f77298a 100644 --- a/docs/execution/checkpoints.mdx +++ b/docs/execution/checkpoints.mdx @@ -90,35 +90,22 @@ For Daytona sandboxes, the worktree is created inside the remote sandbox instead ## Resuming a run -There are two ways to resume an interrupted run: - -### From a checkpoint file - -Resume from a `checkpoint.json` saved in the run directory: - -```bash -fabro resume --checkpoint path/to/logs/checkpoint.json --workflow workflow.fabro -``` - -Fabro loads the checkpoint, restores the context and execution state, and continues from the next node after the checkpoint. - -### From a run branch - -Resume from the Git branches created during a previous run: +Resume an interrupted run from its checkpoint on disk: ```bash fabro resume 01JKXYZ ``` -This reads the checkpoint, run record, and Graphviz graph from the metadata branch (`fabro/meta/01JKXYZ...`), re-attaches a worktree to the existing run branch, and resumes execution. No workflow file argument is needed — everything is recovered from Git. +Fabro looks up the run directory by ID prefix, loads `checkpoint.json` and `run.json` from the run directory, and spawns a new engine process to continue execution. No workflow file or override flags are needed — all configuration is read from the persisted run state. -1. Fabro reads `checkpoint.json` from the metadata branch -2. Reads `run.json` to reconstruct the workflow graph and config -3. Creates a fresh worktree attached to the existing run branch -4. Restores the full context, completed node list, retry counts, and failure signatures -5. If the checkpointed node used `full` fidelity, downgrades the first resumed node to `summary:high` (since the original conversation thread no longer exists in memory) -6. Continues execution from `next_node_id` +1. Fabro looks up the run directory by ID prefix +2. Validates that `checkpoint.json` exists and no engine process is already running +3. Cleans stale artifacts from the previous execution (conclusion, PID file, etc.) +4. Resets status to `Submitted` and spawns a new engine subprocess with `--resume` +5. The engine loads `run.json` and `checkpoint.json`, restores the full context, completed node list, retry counts, and failure signatures +6. If the checkpointed node used `full` fidelity, downgrades the first resumed node to `summary:high` (since the original conversation thread no longer exists in memory) +7. Continues execution from `next_node_id` ## The checkpoint cycle diff --git a/docs/reference/cli.mdx b/docs/reference/cli.mdx index 6a05e6c99..52bc66977 100644 --- a/docs/reference/cli.mdx +++ b/docs/reference/cli.mdx @@ -68,31 +68,17 @@ fabro run run.toml ## `fabro resume` -Resume an interrupted workflow run from its last checkpoint. +Resume an interrupted workflow run from its last checkpoint. The run is looked up by ID prefix and uses the configuration persisted at create time — no runtime overrides are accepted. ```bash fabro resume -fabro resume --workflow updated.fabro -fabro resume --checkpoint path/to/checkpoint.json --workflow workflow.fabro +fabro resume --detach ``` | Argument / Flag | Description | |---|---| -| `` | Run ID, prefix, or branch (`fabro/run/...`). Not required when using `--checkpoint`. | -| `--checkpoint ` | Resume from a checkpoint file (requires `--workflow`) | -| `--workflow ` | Override workflow graph (required with `--checkpoint`) | -| `--run-dir ` | Run output directory | -| `--dry-run` | Execute with a simulated LLM backend | -| `--auto-approve` | Auto-approve all human gates | -| `--model ` | Override default LLM model | -| `--provider ` | Override default LLM provider | -| `-v, --verbose` | Enable verbose output | -| `--sandbox ` | Sandbox for agent tools: `local`, `docker`, `daytona`, `ssh`, or `exe` | -| `--goal ` | Override the workflow goal | -| `--goal-file ` | Read the goal from a file | -| `--no-retro` | Skip retro generation after the run | -| `--preserve-sandbox` | Keep the sandbox alive after the run finishes | -| `--label ` | Attach a label to this run (repeatable) | +| `` | Run ID or unambiguous prefix | +| `-d, --detach` | Run in the background and print the run ID | ## `fabro ps` diff --git a/lib/crates/fabro-cli/src/commands/resume.rs b/lib/crates/fabro-cli/src/commands/resume.rs index 252ef9a95..8183621ec 100644 --- a/lib/crates/fabro-cli/src/commands/resume.rs +++ b/lib/crates/fabro-cli/src/commands/resume.rs @@ -1,1652 +1,108 @@ -use std::io::IsTerminal; -use std::path::PathBuf; -use std::sync::Arc; -use std::time::Instant; - -use anyhow::{bail, Context}; +use anyhow::bail; use clap::Args; -use fabro_agent::{DockerSandbox, DockerSandboxConfig, Sandbox, WorktreeConfig, WorktreeSandbox}; -use fabro_config::config::FabroConfig; -use fabro_interview::{AutoApproveInterviewer, ConsoleInterviewer, Interviewer}; -use fabro_model::{Catalog, Provider}; -use fabro_sandbox::SandboxProvider; use fabro_util::terminal::Styles; -use fabro_workflows::event::{EventEmitter, RunNoticeLevel}; -use fabro_workflows::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter}; -use fabro_workflows::operations::{ - restore, start, RestoreOptions, RunCreateOptions, StartFinalizeOptions, StartOptions, - StartPullRequestConfig, StartRetroOptions, -}; -use fabro_workflows::outcome::StageStatus; -use fabro_workflows::pipeline::{ - build_conclusion, classify_engine_result, persist_terminal_outcome, Persisted, -}; -use fabro_workflows::records::Checkpoint; use fabro_workflows::records::RunRecord; -use fabro_workflows::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; - -use super::detached_support::{DetachedRunBootstrapGuard, DetachedRunCompletionGuard}; -use super::run::{ - apply_execution_overrides, build_event_envelope, cached_graph_path, default_run_dir, - emit_run_notice, load_workflow_source_input, local_sandbox_with_callback, mint_github_token, - parse_labels, print_assets, print_diagnostics_from_error, print_final_output, - print_retro_result, print_run_conclusion, print_workflow_report_from_persisted, - resolve_daytona_config, resolve_fallback_chain, resolve_model_provider, - resolve_sandbox_provider, resolve_ssh_clone_params, resolve_ssh_config, - write_run_config_snapshot, CliSandboxProvider, ExecutionOverrides, RunArgs, -}; -use fabro_config::project as project_config; -use fabro_config::run as run_config; -use fabro_workflows::devcontainer_bridge; -use std::collections::HashMap; -use tracing::debug; +use fabro_workflows::run_status::RunStatus; #[derive(Debug, Args)] pub struct ResumeArgs { - /// Run ID, prefix, or branch (fabro/run/...) - #[arg(required_unless_present = "checkpoint")] - pub run: Option, + /// Run ID or unambiguous prefix + pub run: String, - /// Resume from a checkpoint file (requires --workflow) - #[arg(long, conflicts_with = "run", requires = "workflow")] - pub checkpoint: Option, - - /// Override workflow graph (required with --checkpoint) - #[arg(long)] - pub workflow: Option, - - /// Run output directory - #[arg(long)] - pub run_dir: Option, - - /// Execute with simulated LLM backend - #[arg(long)] - pub dry_run: bool, - - /// Auto-approve all human gates - #[arg(long)] - pub auto_approve: bool, - - /// Override default LLM model - #[arg(long)] - pub model: Option, - - /// Override default LLM provider - #[arg(long)] - pub provider: Option, - - /// Enable verbose output - #[arg(short, long)] - pub verbose: bool, - - /// Sandbox for agent tools - #[arg(long, value_enum)] - pub sandbox: Option, - - /// Skip retro generation after the run - #[arg(long)] - pub no_retro: bool, - - /// Keep the sandbox alive after the run finishes (for debugging) - #[arg(long)] - pub preserve_sandbox: bool, - - /// Attach a label to this run (repeatable, format: KEY=VALUE) - #[arg(long = "label", value_name = "KEY=VALUE")] - pub label: Vec, -} - -/// Intermediate state produced by the two resolution paths (checkpoint-file vs. git-branch). -struct ResumeContext { - checkpoint: Checkpoint, - persisted: Persisted, - run_id: String, - run_dir: PathBuf, - run_cfg: Option, - sandbox: Arc, - /// Kept as Arc so the sandbox event callbacks can emit through it. Listeners - /// that need to be added later (e.g. ProgressUI) are registered separately. - emitter: Arc, - run_options: RunOptions, - setup_commands: Vec, - /// Devcontainer lifecycle phases (on_create, post_create, post_start) resolved from config. - devcontainer_phases: Vec<(String, Vec)>, - /// Devcontainer remoteEnv values to layer under sandbox_env. - devcontainer_env: HashMap, - /// Original cwd to restore after engine run (git-branch path changes cwd to worktree). - original_cwd: Option, - origin_url: Option, - sandbox_provider: SandboxProvider, - ssh_data_host: Option, - github_app: Option, - status_guard: DetachedRunBootstrapGuard, -} - -fn resume_as_run_args(args: &ResumeArgs, workflow: PathBuf) -> RunArgs { - RunArgs { - workflow: Some(workflow), - run_dir: None, - dry_run: args.dry_run, - preflight: false, - auto_approve: args.auto_approve, - goal: None, - goal_file: None, - model: args.model.clone(), - provider: args.provider.clone(), - verbose: args.verbose, - sandbox: args.sandbox, - label: Vec::new(), - no_retro: args.no_retro, - preserve_sandbox: args.preserve_sandbox, - detach: false, - run_id: None, - } -} - -fn preferred_resume_repo_path( - original_cwd: &std::path::Path, - record: Option<&RunRecord>, -) -> PathBuf { - record - .and_then(|r| r.host_repo_path.as_deref()) - .map(PathBuf::from) - .filter(|path| path.exists()) - .unwrap_or_else(|| original_cwd.to_path_buf()) -} - -fn restore_persisted_for_resume( - rec: &RunRecord, - config: FabroConfig, - run_dir: PathBuf, - run_id: &str, - labels: HashMap, - base_branch: Option, - resume_repo_path: &std::path::Path, -) -> Result { - let mut run_record = rec.clone(); - run_record.run_id = run_id.to_string(); - run_record.config = config; - run_record.labels = labels; - run_record.base_branch = base_branch; - run_record.working_directory = resume_repo_path.to_path_buf(); - run_record.host_repo_path = Some(resume_repo_path.to_string_lossy().to_string()); - - restore(RestoreOptions { - run_dir, - run_record, - }) + /// Run in the background and print the run ID + #[arg(short = 'd', long)] + pub detach: bool, } /// Resume an interrupted workflow run. /// -/// # Errors -/// -/// Returns an error if the run cannot be found, the checkpoint cannot be loaded, -/// or the workflow cannot be resumed. -pub async fn resume_command( - args: ResumeArgs, - mut run_defaults: FabroConfig, - styles: &'static Styles, - github_app: Option, - git_author: fabro_workflows::git::GitAuthor, -) -> anyhow::Result<()> { - // Apply project-level config overrides (fabro.toml) on top of CLI defaults (mirrors run_command). - if let Ok(Some((_config_path, project_config))) = - project_config::discover_project_config(&std::env::current_dir().unwrap_or_default()) - { - tracing::debug!("Applying run defaults from fabro.toml"); - run_defaults.merge_overlay(project_config); +/// Looks up the run by ID prefix, validates a checkpoint exists, cleans stale +/// artifacts from the previous execution, then spawns an engine subprocess +/// (identical to `fabro run`'s create→start→attach flow). +pub async fn resume_command(args: ResumeArgs, styles: &'static Styles) -> anyhow::Result<()> { + let base = fabro_workflows::run_lookup::default_runs_base(); + let run_dir = fabro_workflows::run_lookup::find_run_by_prefix(&base, &args.run)?; + + // find_run_by_prefix can match orphan directories (no run.json). + if !run_dir.join("run.json").exists() { + bail!("run directory exists but has no run.json — cannot resume"); + } + let run_id = RunRecord::load(&run_dir)?.run_id; + + if !run_dir.join("checkpoint.json").exists() { + bail!("no checkpoint found — nothing to resume"); } - let ctx = if args.checkpoint.is_some() { - prepare_from_checkpoint(&args, &run_defaults, styles, &github_app, git_author).await? - } else { - prepare_from_branch(&args, styles, &run_defaults, &github_app, git_author).await? - }; + // Guard against resuming a live run + if is_pid_alive(&run_dir.join("run.pid")) { + bail!("an engine process is still running for this run — cannot resume"); + } - run_resumed(ctx, args, run_defaults, styles).await + // Clean stale artifacts from previous execution + for name in &[ + "conclusion.json", + "pull_request.json", + "detached_failure.json", + "interview_request.json", + "interview_response.json", + "interview_request.claim", + "detach.log", + "run.pid", + ] { + let _ = std::fs::remove_file(run_dir.join(name)); + } + + // Reset status for re-execution + fabro_workflows::run_status::write_run_status(&run_dir, RunStatus::Submitted, None); + + let child = super::start::start_run(&run_dir, true)?; + + if args.detach { + println!("{run_id}"); + } else { + let exit_code = super::attach::attach_run(&run_dir, true, styles, Some(child)).await?; + super::run::print_run_summary(&run_dir, &run_id, styles); + if exit_code != std::process::ExitCode::SUCCESS { + std::process::exit(1); + } + } + Ok(()) } -/// Checkpoint-file path: load checkpoint and graph from files, resolve sandbox from flags/config. -async fn prepare_from_checkpoint( - args: &ResumeArgs, - run_defaults: &FabroConfig, - styles: &Styles, - github_app: &Option, - git_author: fabro_workflows::git::GitAuthor, -) -> anyhow::Result { - let checkpoint_path = args.checkpoint.as_ref().unwrap(); - let workflow_path = args - .workflow - .as_ref() - .ok_or_else(|| anyhow::anyhow!("--workflow is required when using --checkpoint"))?; - - let checkpoint = Checkpoint::load(checkpoint_path)?; - let source_input = load_workflow_source_input( - &resume_as_run_args(args, workflow_path.clone()), - run_defaults.clone(), - false, - )?; - - let run_id = ulid::Ulid::new().to_string(); - let run_dir = args - .run_dir - .clone() - .unwrap_or_else(|| default_run_dir(&run_id, args.dry_run)); - let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")); - let sandbox_provider = if args.dry_run { - SandboxProvider::Local - } else { - resolve_sandbox_provider( - args.sandbox.map(Into::into), - Some(&source_input.config), - run_defaults, - )? +/// Check whether a PID file contains a live process. +fn is_pid_alive(pid_path: &std::path::Path) -> bool { + let Ok(content) = std::fs::read_to_string(pid_path) else { + return false; }; - let mut config = source_input.config.clone(); - apply_execution_overrides( - &mut config, - &ExecutionOverrides { - dry_run: args.dry_run, - auto_approve: args.auto_approve, - no_retro: args.no_retro, - verbose: args.verbose, - preserve_sandbox: args.preserve_sandbox, - model: args.model.as_deref(), - provider: args.provider.as_deref(), - sandbox_provider, - }, - ); - let persisted = match fabro_workflows::operations::create( - &source_input.raw_source, - RunCreateOptions { - config, - run_dir: Some(run_dir.clone()), - run_id: Some(run_id.clone()), - workflow_slug: source_input.workflow_slug.clone(), - labels: parse_labels(&args.label), - base_branch: None, - working_directory: Some(working_directory.clone()), - host_repo_path: Some(working_directory.to_string_lossy().to_string()), - goal_override: source_input.goal_override.clone(), - base_dir: Some( - source_input - .dot_path - .parent() - .unwrap_or(std::path::Path::new(".")) - .to_path_buf(), - ), - }, - ) { - Ok(persisted) => persisted, - Err(fabro_workflows::error::FabroError::ValidationFailed { diagnostics }) => { - print_diagnostics_from_error(&diagnostics, styles); - bail!("Validation failed"); - } - Err(err) => return Err(err.into()), + let Ok(pid) = content.trim().parse::() else { + return false; }; - print_workflow_report_from_persisted(&persisted, &source_input.dot_path, styles); - eprintln!( - "{} {} from checkpoint {}", - styles.bold.apply_to("Resuming workflow:"), - persisted.graph().name, - styles.dim.apply_to(checkpoint_path.display()), - ); - let run_cfg: Option = Some(persisted.run_record().config.clone()); - let settings_config = persisted.run_record().config.clone(); - - tokio::fs::create_dir_all(&run_dir).await?; - fabro_util::run_log::activate(&run_dir.join("cli.log")) - .context("Failed to activate per-run log")?; - let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?; - tokio::fs::write(cached_graph_path(&run_dir), &source_input.raw_source).await?; - write_run_config_snapshot(&run_dir, source_input.workflow_toml_path.as_deref()).await?; - - let original_cwd = std::env::current_dir()?; - let emitter = Arc::new(EventEmitter::new()); - - // Resolve devcontainer BEFORE sandbox creation (mirrors run_command) so that - // the Daytona snapshot config can be overridden with the devcontainer Dockerfile. - let run_defaults = &run_defaults; - let mut daytona_config = resolve_daytona_config(run_cfg.as_ref(), run_defaults); - let devcontainer_config = if run_cfg - .as_ref() - .and_then(|c| c.sandbox.as_ref()) - .or(run_defaults.sandbox.as_ref()) - .and_then(|s| s.devcontainer) - .unwrap_or(false) - { - match fabro_devcontainer::DevcontainerResolver::resolve(&original_cwd).await { - Ok(dc) => { - let lifecycle_command_count = dc.on_create_commands.len() - + dc.post_create_commands.len() - + dc.post_start_commands.len(); - emitter.emit( - &fabro_workflows::event::WorkflowRunEvent::DevcontainerResolved { - dockerfile_lines: dc.dockerfile.lines().count(), - environment_count: dc.environment.len(), - lifecycle_command_count, - workspace_folder: dc.workspace_folder.clone(), - }, - ); - - // Override daytona_config with devcontainer dockerfile - let snapshot = devcontainer_bridge::devcontainer_to_snapshot_config(&dc); - let mut cfg = daytona_config.unwrap_or_default(); - cfg.snapshot = Some(snapshot); - daytona_config = Some(cfg); - - // Run initialize_commands on host (mirrors run_command) - let timeout = std::time::Duration::from_millis(300_000); - for cmd in &dc.initialize_commands { - let shell_cmds = match cmd { - fabro_devcontainer::Command::Shell(s) => vec![s.clone()], - fabro_devcontainer::Command::Args(args) => { - vec![args - .iter() - .map(|a| { - shlex::try_quote(a).unwrap_or_else(|_| a.into()).to_string() - }) - .collect::>() - .join(" ")] - } - fabro_devcontainer::Command::Parallel(map) => { - map.values().cloned().collect() - } - }; - for shell_cmd in &shell_cmds { - let fut = tokio::process::Command::new("sh") - .arg("-c") - .arg(shell_cmd) - .current_dir(&original_cwd) - .output(); - let output = tokio::time::timeout(timeout, fut) - .await - .with_context(|| { - format!("Devcontainer initializeCommand timed out: {shell_cmd}") - })? - .with_context(|| { - format!( - "Failed to execute devcontainer initializeCommand: {shell_cmd}" - ) - })?; - if !output.status.success() { - let code = output - .status - .code() - .map_or("unknown".to_string(), |c| c.to_string()); - let stderr = String::from_utf8_lossy(&output.stderr); - bail!( - "Devcontainer initializeCommand failed (exit code {code}): {shell_cmd}\n{stderr}" - ); - } - } - } - - Some(dc) - } - Err(e) => { - bail!("Failed to resolve devcontainer: {e}"); - } - } - } else { - None - }; - - let devcontainer_phases = if let Some(ref dc) = devcontainer_config { - vec![ - ("on_create".to_string(), dc.on_create_commands.clone()), - ("post_create".to_string(), dc.post_create_commands.clone()), - ("post_start".to_string(), dc.post_start_commands.clone()), - ] - } else { - Vec::new() - }; - - let mut ssh_data_host: Option = None; - let sandbox: Arc = match sandbox_provider { - SandboxProvider::Local => { - local_sandbox_with_callback(original_cwd.clone(), Arc::clone(&emitter)) - } - SandboxProvider::Docker => { - let config = DockerSandboxConfig { - host_working_directory: original_cwd.to_string_lossy().to_string(), - ..DockerSandboxConfig::default() - }; - let mut env = DockerSandbox::new(config) - .map_err(|e| anyhow::anyhow!("Failed to create Docker environment: {e}"))?; - let emitter_cb = Arc::clone(&emitter); - env.set_event_callback(Arc::new(move |event| { - emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); - })); - Arc::new(env) - } - #[cfg(feature = "exedev")] - SandboxProvider::Exe => { - let exe_config = super::run::resolve_exe_config(run_cfg.as_ref(), run_defaults); - let clone_params = super::run::resolve_exe_clone_params(&original_cwd); - let mgmt_ssh = fabro_sandbox::exe::OpensshRunner::connect_raw("exe.dev") - .await - .map_err(|e| anyhow::anyhow!("Failed to connect to exe.dev: {e}"))?; - let config = exe_config.unwrap_or_default(); - let mut env = fabro_sandbox::exe::ExeSandbox::new( - Box::new(mgmt_ssh), - config, - clone_params, - Some(run_id.clone()), - github_app.clone(), - ); - let emitter_cb = Arc::clone(&emitter); - env.set_event_callback(Arc::new(move |event| { - emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); - })); - Arc::new(env) - } - #[cfg(not(feature = "exedev"))] - SandboxProvider::Exe => { - anyhow::bail!("exe sandbox requires the exedev feature"); - } - SandboxProvider::Ssh => { - let config = resolve_ssh_config(run_cfg.as_ref(), run_defaults) - .ok_or_else(|| anyhow::anyhow!("--sandbox ssh requires [sandbox.ssh] config"))?; - ssh_data_host = Some(config.destination.clone()); - let clone_params = resolve_ssh_clone_params(&original_cwd); - let mut env = fabro_sandbox::ssh::SshSandbox::new( - config, - clone_params, - Some(run_id.clone()), - github_app.clone(), - ); - let emitter_cb = Arc::clone(&emitter); - env.set_event_callback(Arc::new(move |event| { - emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); - })); - Arc::new(env) - } - SandboxProvider::Daytona => { - let config = daytona_config.unwrap_or_default(); - let mut env = fabro_sandbox::daytona::DaytonaSandbox::new( - config, - github_app.clone(), - Some(run_id.clone()), - None, - ) - .await - .map_err(|e| anyhow::anyhow!("{e}"))?; - let emitter_cb = Arc::clone(&emitter); - env.set_event_callback(Arc::new(move |event| { - emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); - })); - Arc::new(env) - } - }; - let sandbox: Arc = Arc::new(fabro_agent::ReadBeforeWriteSandbox::new(sandbox)); - - let run_options = RunOptions { - config: settings_config, - run_dir: run_dir.clone(), - cancel_token: None, - dry_run: args.dry_run, - run_id: run_id.clone(), - host_repo_path: persisted - .run_record() - .host_repo_path - .as_deref() - .map(PathBuf::from), - git: None, - labels: persisted.run_record().labels.clone(), - github_app: github_app.clone(), - git_author, - base_branch: persisted.run_record().base_branch.clone(), - workflow_slug: persisted.run_record().workflow_slug.clone(), - }; - - let devcontainer_env = devcontainer_config - .as_ref() - .map(|dc| dc.environment.clone()) - .unwrap_or_default(); - let setup_commands = run_cfg - .as_ref() - .and_then(|cfg| cfg.setup.as_ref()) - .or(run_defaults.setup.as_ref()) - .map(|s| s.commands.clone()) - .unwrap_or_default(); - - Ok(ResumeContext { - checkpoint, - persisted, - run_id, - run_dir, - run_cfg, - sandbox, - emitter, - run_options, - setup_commands, - devcontainer_phases, - devcontainer_env, - original_cwd: None, - origin_url: fabro_sandbox::daytona::detect_repo_info(&original_cwd) - .ok() - .map(|(url, _)| url), - sandbox_provider, - ssh_data_host, - github_app: github_app.clone(), - status_guard, - }) -} - -/// Git-branch path: resolve run ID, read checkpoint + graph from metadata, set up worktree. -async fn prepare_from_branch( - args: &ResumeArgs, - styles: &Styles, - run_defaults: &FabroConfig, - github_app: &Option, - git_author: fabro_workflows::git::GitAuthor, -) -> anyhow::Result { - let run_arg = args.run.as_deref().expect("run is required"); - - let (run_id, run_branch) = - if let Some(stripped) = run_arg.strip_prefix(fabro_workflows::git::RUN_BRANCH_PREFIX) { - (stripped.to_string(), run_arg.to_string()) - } else { - let repo = git2::Repository::discover(".").context("not in a git repository")?; - 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) - }; - - let original_cwd = std::env::current_dir()?; - let record_hint = fabro_workflows::git::MetadataStore::read_run_record(&original_cwd, &run_id) - .ok() - .flatten(); - let resume_repo_path = preferred_resume_repo_path(&original_cwd, record_hint.as_ref()); - let record = if resume_repo_path == original_cwd { - record_hint - } else { - fabro_workflows::git::MetadataStore::read_run_record(&resume_repo_path, &run_id) - .ok() - .flatten() - .or(record_hint) - }; - let start_record = - fabro_workflows::git::MetadataStore::read_start_record(&resume_repo_path, &run_id) - .ok() - .flatten(); - let checkpoint = - fabro_workflows::git::MetadataStore::read_checkpoint(&resume_repo_path, &run_id)? - .ok_or_else(|| { - anyhow::anyhow!("no checkpoint found on metadata branch for run {run_id}") - })?; - let repo_info = fabro_sandbox::daytona::detect_repo_info(&resume_repo_path).ok(); - let origin_url = repo_info.as_ref().map(|(url, _)| url.clone()); - let detected_base_branch = record - .as_ref() - .and_then(|r| r.base_branch.clone()) - .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()); - - // Set up logs directory — reuse existing run dir for this run_id to avoid - // "ambiguous prefix" errors when the resume happens on a different day. - // Skip reuse when dry-running to avoid corrupting real run data. - let run_dir = if let Some(ref dir) = args.run_dir { - dir.clone() - } else if args.dry_run { - default_run_dir(&run_id, true) - } else { - find_existing_run_dir(&run_id, false).unwrap_or_else(|| default_run_dir(&run_id, false)) - }; - tokio::fs::create_dir_all(&run_dir).await?; - let run_dir = tokio::fs::canonicalize(&run_dir).await.unwrap_or(run_dir); - - let (persisted, _run_cfg, mut sandbox_provider, graph_source) = - if let Some(ref workflow_path) = args.workflow { - let source_input = load_workflow_source_input( - &resume_as_run_args(args, workflow_path.clone()), - run_defaults.clone(), - false, - )?; - let sandbox_provider = if args.dry_run { - SandboxProvider::Local - } else { - resolve_sandbox_provider( - args.sandbox.map(Into::into), - Some(&source_input.config), - run_defaults, - )? - }; - let mut config = source_input.config.clone(); - apply_execution_overrides( - &mut config, - &ExecutionOverrides { - dry_run: args.dry_run, - auto_approve: args.auto_approve, - no_retro: args.no_retro, - verbose: args.verbose, - preserve_sandbox: args.preserve_sandbox, - model: args.model.as_deref(), - provider: args.provider.as_deref(), - sandbox_provider, - }, - ); - let persisted = match fabro_workflows::operations::create( - &source_input.raw_source, - RunCreateOptions { - config, - run_dir: Some(run_dir.clone()), - run_id: Some(run_id.clone()), - workflow_slug: source_input.workflow_slug.clone(), - labels: parse_labels(&args.label), - base_branch: detected_base_branch.clone(), - working_directory: Some(resume_repo_path.clone()), - host_repo_path: Some(resume_repo_path.to_string_lossy().to_string()), - goal_override: source_input.goal_override.clone(), - base_dir: Some( - source_input - .dot_path - .parent() - .unwrap_or(std::path::Path::new(".")) - .to_path_buf(), - ), - }, - ) { - Ok(persisted) => persisted, - Err(fabro_workflows::error::FabroError::ValidationFailed { diagnostics }) => { - print_diagnostics_from_error(&diagnostics, styles); - bail!("Validation failed"); - } - Err(err) => return Err(err.into()), - }; - print_workflow_report_from_persisted(&persisted, &source_input.dot_path, styles); - let graph_source = source_input.raw_source; - let run_cfg = Some(persisted.run_record().config.clone()); - (persisted, run_cfg, sandbox_provider, graph_source) - } else if let Ok(loaded) = Persisted::load(&run_dir) { - let sandbox_provider = if args.dry_run { - SandboxProvider::Local - } else { - let persisted_provider = loaded - .run_record() - .config - .sandbox - .as_ref() - .and_then(|s| s.provider.as_deref()) - .and_then(|s| s.parse::().ok()) - .unwrap_or_default(); - args.sandbox.map(Into::into).unwrap_or(persisted_provider) - }; - let graph_source = loaded.source().to_string(); - let run_cfg = Some(loaded.run_record().config.clone()); - (loaded, run_cfg, sandbox_provider, graph_source) - } else if let Some(ref rec) = record { - let sandbox_provider = if args.dry_run { - SandboxProvider::Local - } else { - let sp = rec - .config - .sandbox - .as_ref() - .and_then(|s| s.provider.as_deref()) - .and_then(|s| s.parse::().ok()) - .unwrap_or_default(); - args.sandbox.map(Into::into).unwrap_or(sp) - }; - let mut config = rec.config.clone(); - apply_execution_overrides( - &mut config, - &ExecutionOverrides { - dry_run: args.dry_run, - auto_approve: args.auto_approve, - no_retro: args.no_retro, - verbose: args.verbose, - preserve_sandbox: args.preserve_sandbox, - model: args.model.as_deref(), - provider: args.provider.as_deref(), - sandbox_provider, - }, - ); - let persisted = restore_persisted_for_resume( - rec, - config, - run_dir.clone(), - &run_id, - parse_labels(&args.label), - detected_base_branch.clone(), - &resume_repo_path, - )?; - let graph_source = persisted.source().to_string(); - let run_cfg = Some(persisted.run_record().config.clone()); - (persisted, run_cfg, sandbox_provider, graph_source) - } else { - bail!("no run.json found on metadata branch for run {run_id}"); - }; - - let graph = persisted.graph().clone(); - - eprintln!( - "{} {} from branch {} ({})", - styles.bold.apply_to("Resuming workflow:"), - graph.name, - styles.dim.apply_to(&run_branch), - run_id, - ); - - let run_cfg: Option = Some(persisted.run_record().config.clone()); - let settings_config = persisted.run_record().config.clone(); - - fabro_util::run_log::activate(&run_dir.join("cli.log")) - .context("Failed to activate per-run log")?; - let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?; - if !graph_source.is_empty() { - tokio::fs::write(cached_graph_path(&run_dir), &graph_source).await?; - } - // Git-branch resume: no original TOML available, skip debug snapshot. - write_run_config_snapshot(&run_dir, None).await?; - - let emitter = Arc::new(EventEmitter::new()); - - // Resolve devcontainer BEFORE sandbox creation (mirrors run_command) so that - // the Daytona snapshot config can be overridden with the devcontainer Dockerfile. - let mut daytona_config = resolve_daytona_config(run_cfg.as_ref(), run_defaults); - let devcontainer_config = if run_cfg - .as_ref() - .and_then(|cfg| cfg.sandbox.as_ref()) - .or(run_defaults.sandbox.as_ref()) - .and_then(|s| s.devcontainer) - .unwrap_or(false) - { - match fabro_devcontainer::DevcontainerResolver::resolve(&resume_repo_path).await { - Ok(dc) => { - let lifecycle_command_count = dc.on_create_commands.len() - + dc.post_create_commands.len() - + dc.post_start_commands.len(); - emitter.emit( - &fabro_workflows::event::WorkflowRunEvent::DevcontainerResolved { - dockerfile_lines: dc.dockerfile.lines().count(), - environment_count: dc.environment.len(), - lifecycle_command_count, - workspace_folder: dc.workspace_folder.clone(), - }, - ); - - // Override daytona_config with devcontainer dockerfile - let snapshot = devcontainer_bridge::devcontainer_to_snapshot_config(&dc); - let mut cfg = daytona_config.unwrap_or_default(); - cfg.snapshot = Some(snapshot); - daytona_config = Some(cfg); - - // Run initialize_commands on host (mirrors run_command) - let timeout = std::time::Duration::from_millis(300_000); - for cmd in &dc.initialize_commands { - let shell_cmds = match cmd { - fabro_devcontainer::Command::Shell(s) => vec![s.clone()], - fabro_devcontainer::Command::Args(args) => { - vec![args - .iter() - .map(|a| { - shlex::try_quote(a).unwrap_or_else(|_| a.into()).to_string() - }) - .collect::>() - .join(" ")] - } - fabro_devcontainer::Command::Parallel(map) => { - map.values().cloned().collect() - } - }; - for shell_cmd in &shell_cmds { - let fut = tokio::process::Command::new("sh") - .arg("-c") - .arg(shell_cmd) - .current_dir(&resume_repo_path) - .output(); - let output = tokio::time::timeout(timeout, fut) - .await - .with_context(|| { - format!("Devcontainer initializeCommand timed out: {shell_cmd}") - })? - .with_context(|| { - format!( - "Failed to execute devcontainer initializeCommand: {shell_cmd}" - ) - })?; - if !output.status.success() { - let code = output - .status - .code() - .map_or("unknown".to_string(), |c| c.to_string()); - let stderr = String::from_utf8_lossy(&output.stderr); - bail!( - "Devcontainer initializeCommand failed (exit code {code}): {shell_cmd}\n{stderr}" - ); - } - } - } - - Some(dc) - } - Err(e) => { - bail!("Failed to resolve devcontainer: {e}"); - } - } - } else { - None - }; - - let devcontainer_phases = if let Some(ref dc) = devcontainer_config { - vec![ - ("on_create".to_string(), dc.on_create_commands.clone()), - ("post_create".to_string(), dc.post_create_commands.clone()), - ("post_start".to_string(), dc.post_start_commands.clone()), - ] - } else { - Vec::new() - }; - - let setup_worktree_sandbox = |emitter: &Arc| -> (WorktreeSandbox, PathBuf) { - let wt = run_dir.join("worktree"); - let wt_str = wt.to_string_lossy().into_owned(); - - let inner = local_sandbox_with_callback(resume_repo_path.clone(), Arc::clone(emitter)); - let wt_config = WorktreeConfig { - branch_name: run_branch.clone(), - base_sha: base_sha.clone().unwrap_or_default(), - worktree_path: wt_str.clone(), - skip_branch_creation: true, // branch already exists on resume - }; - let mut wt_sandbox = WorktreeSandbox::new(inner, wt_config); - wt_sandbox.set_event_callback(Arc::clone(emitter).worktree_callback()); - (wt_sandbox, wt) - }; - - let mut ssh_data_host: Option = None; - let (sandbox, _worktree_path): (Arc, Option) = match sandbox_provider { - SandboxProvider::Local => { - let (wt_sandbox, wt) = setup_worktree_sandbox(&emitter); - wt_sandbox - .initialize() - .await - .map_err(|e| anyhow::anyhow!("failed to attach worktree to {run_branch}: {e}"))?; - std::env::set_current_dir(&wt)?; - (Arc::new(wt_sandbox) as Arc, Some(wt)) - } - SandboxProvider::Docker => { - tracing::warn!( - "--sandbox docker is not supported for branch resume; falling back to local worktree sandbox" - ); - eprintln!( - "{} --sandbox docker is not supported for branch resume; falling back to local worktree sandbox.", - styles.yellow.apply_to("Warning:"), - ); - sandbox_provider = SandboxProvider::Local; - let (wt_sandbox, wt) = setup_worktree_sandbox(&emitter); - wt_sandbox - .initialize() - .await - .map_err(|e| anyhow::anyhow!("failed to attach worktree to {run_branch}: {e}"))?; - std::env::set_current_dir(&wt)?; - (Arc::new(wt_sandbox) as Arc, Some(wt)) - } - #[cfg(feature = "exedev")] - SandboxProvider::Exe => { - let exe_config = super::run::resolve_exe_config(run_cfg.as_ref(), run_defaults); - let clone_params = super::run::resolve_exe_clone_params(&resume_repo_path); - let mgmt_ssh = fabro_sandbox::exe::OpensshRunner::connect_raw("exe.dev") - .await - .map_err(|e| anyhow::anyhow!("Failed to connect to exe.dev: {e}"))?; - let config = exe_config.unwrap_or_default(); - let mut env = fabro_sandbox::exe::ExeSandbox::new( - Box::new(mgmt_ssh), - config, - clone_params, - Some(run_id.clone()), - github_app.clone(), - ); - let emitter_cb = Arc::clone(&emitter); - env.set_event_callback(Arc::new(move |event| { - emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); - })); - (Arc::new(env), None) - } - #[cfg(not(feature = "exedev"))] - SandboxProvider::Exe => { - anyhow::bail!("exe sandbox requires the exedev feature"); - } - SandboxProvider::Ssh => { - let config = resolve_ssh_config(run_cfg.as_ref(), run_defaults) - .ok_or_else(|| anyhow::anyhow!("--sandbox ssh requires [sandbox.ssh] config"))?; - ssh_data_host = Some(config.destination.clone()); - let clone_params = resolve_ssh_clone_params(&resume_repo_path); - let mut env = fabro_sandbox::ssh::SshSandbox::new( - config, - clone_params, - Some(run_id.clone()), - github_app.clone(), - ); - let emitter_cb = Arc::clone(&emitter); - env.set_event_callback(Arc::new(move |event| { - emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); - })); - (Arc::new(env), None) - } - SandboxProvider::Daytona => { - let config = daytona_config.unwrap_or_default(); - let mut env = fabro_sandbox::daytona::DaytonaSandbox::new( - config, - github_app.clone(), - Some(run_id.clone()), - Some(run_branch.clone()), - ) - .await - .map_err(|e| anyhow::anyhow!("{e}"))?; - let emitter_cb = Arc::clone(&emitter); - env.set_event_callback(Arc::new(move |event| { - emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); - })); - (Arc::new(env), None) - } - }; - - // Wrap with ReadBeforeWriteSandbox to enforce read-before-write guard - let sandbox: Arc = Arc::new(fabro_agent::ReadBeforeWriteSandbox::new(sandbox)); - - // User-configured setup commands first, then sandbox-specific resume commands - let mut setup_commands: Vec = run_cfg - .as_ref() - .and_then(|cfg| cfg.setup.as_ref()) - .or(run_defaults.setup.as_ref()) - .map(|s| s.commands.clone()) - .unwrap_or_default(); - setup_commands.extend(sandbox.resume_setup_commands(&run_branch)); - - let run_options = RunOptions { - config: settings_config, - run_dir: run_dir.clone(), - cancel_token: None, - dry_run: args.dry_run, - run_id: run_id.clone(), - host_repo_path: persisted - .run_record() - .host_repo_path - .as_deref() - .map(PathBuf::from) - .or_else(|| Some(resume_repo_path.clone())), - git: Some(GitCheckpointOptions { - base_sha, - run_branch: Some(run_branch), - meta_branch: Some(fabro_workflows::git::MetadataStore::branch_name(&run_id)), - }), - labels: persisted.run_record().labels.clone(), - github_app: github_app.clone(), - git_author, - base_branch: persisted - .run_record() - .base_branch - .clone() - .or(detected_base_branch), - workflow_slug: persisted.run_record().workflow_slug.clone(), - }; - - let devcontainer_env = devcontainer_config - .as_ref() - .map(|dc| dc.environment.clone()) - .unwrap_or_default(); - - Ok(ResumeContext { - checkpoint, - persisted, - run_id, - run_dir, - run_cfg, - sandbox, - emitter, - run_options, - setup_commands, - devcontainer_phases, - devcontainer_env, - original_cwd: Some(original_cwd), - origin_url, - sandbox_provider, - ssh_data_host, - github_app: github_app.clone(), - status_guard, - }) -} - -/// Shared tail: build engine, run workflow, generate retro, print results. -async fn run_resumed( - ctx: ResumeContext, - args: ResumeArgs, - run_defaults: FabroConfig, - styles: &'static Styles, -) -> anyhow::Result<()> { - let ResumeContext { - checkpoint, - persisted, - run_id, - run_dir, - mut run_cfg, - sandbox, - emitter, - mut run_options, - setup_commands, - devcontainer_phases, - devcontainer_env, - original_cwd, - origin_url, - sandbox_provider, - ssh_data_host, - github_app, - mut status_guard, - } = ctx; - let graph = persisted.graph().clone(); - - // Create progress UI (verbose mode shows detailed turn/tool counts and token usage) - let is_tty = std::io::stderr().is_terminal(); - let progress_ui = Arc::new(std::sync::Mutex::new(super::run_progress::ProgressUI::new( - is_tty, - args.verbose, - ))); - { - let mut ui = progress_ui.lock().expect("progress lock poisoned"); - ui.show_version(); - ui.show_run_id(&run_id); - ui.show_time(&chrono::Local::now().format("%Y-%m-%d %H:%M:%S").to_string()); - ui.show_run_dir(&run_dir); - } - { - let p = Arc::clone(&progress_ui); - emitter.on_event(move |event| { - let mut ui = p.lock().expect("progress lock poisoned"); - ui.handle_event(event); - }); - } - - // Cost accumulator (mirrors run_command) - let accumulator = Arc::new(std::sync::Mutex::new(super::run::CostAccumulator::default())); - { - let acc_clone = Arc::clone(&accumulator); - emitter.on_event(move |event| { - if let fabro_workflows::event::WorkflowRunEvent::StageCompleted { - usage: Some(u), .. - } = event - { - let mut acc = acc_clone.lock().unwrap(); - acc.total_input_tokens += u.input_tokens; - acc.total_output_tokens += u.output_tokens; - acc.total_cache_read_tokens += u.cache_read_tokens.unwrap_or(0); - acc.total_cache_write_tokens += u.cache_write_tokens.unwrap_or(0); - acc.total_reasoning_tokens += u.reasoning_tokens.unwrap_or(0); - if let Some(cost) = fabro_workflows::outcome::compute_stage_cost(u) { - acc.total_cost += cost; - acc.has_pricing = true; - } - } - }); - } - - // Write sandbox.json when sandbox is initialized (mirrors run_command) - { - let run_dir_for_listener = run_dir.clone(); - let progress_for_listener = Arc::clone(&progress_ui); - let cwd_for_listener = match &original_cwd { - Some(p) => p.to_string_lossy().to_string(), - // original_cwd is None only for checkpoint path where cwd hasn't changed - None => std::env::current_dir() - .unwrap_or_else(|_| PathBuf::from(".")) - .to_string_lossy() - .to_string(), - }; - let sandbox_for_listener = Arc::clone(&sandbox); - let provider = sandbox_provider; - let ssh_host = ssh_data_host.clone(); - emitter.on_event(move |event| { - if let fabro_workflows::event::WorkflowRunEvent::SandboxInitialized { - working_directory, - } = event - { - progress_for_listener - .lock() - .expect("progress lock poisoned") - .set_working_directory(working_directory.clone()); - - let sandbox_info_opt = { - let info = sandbox_for_listener.sandbox_info(); - if info.is_empty() { - None - } else { - Some(info) - } - }; - - let is_docker = provider == SandboxProvider::Docker; - let record = fabro_sandbox::SandboxRecord { - provider: provider.to_string(), - working_directory: working_directory.clone(), - identifier: sandbox_info_opt, - host_working_directory: if is_docker { - Some(cwd_for_listener.clone()) - } else { - None - }, - container_mount_point: if is_docker { - Some(working_directory.clone()) - } else { - None - }, - data_host: if provider == SandboxProvider::Ssh { - ssh_host.clone() - } else { - None - }, - }; - if let Err(e) = record.save(&run_dir_for_listener.join("sandbox.json")) { - tracing::warn!(error = %e, "Failed to save sandbox record"); - } - } - }); - } - - // JSONL progress log + live.json snapshot (mirrors run_command) - { - let jsonl_path = run_dir.join("progress.jsonl"); - let live_path = run_dir.join("live.json"); - let run_id_shared = Arc::new(std::sync::Mutex::new(run_id.clone())); - let run_id_clone = Arc::clone(&run_id_shared); - emitter.on_event(move |event| { - if let fabro_workflows::event::WorkflowRunEvent::WorkflowRunStarted { run_id, .. } = - event - { - *run_id_clone.lock().unwrap() = run_id.clone(); - } - let envelope = build_event_envelope(event, &run_id_clone.lock().unwrap()); - // Append to progress.jsonl - if let Ok(line) = serde_json::to_string(&envelope) { - let line = fabro_util::redact::redact_jsonl_line(&line); - use std::io::Write; - if let Ok(mut f) = std::fs::OpenOptions::new() - .create(true) - .append(true) - .open(&jsonl_path) - { - let _ = writeln!(f, "{line}"); - } - } - // Overwrite live.json - if let Ok(pretty) = serde_json::to_string_pretty(&envelope) { - let pretty = fabro_util::redact::redact_jsonl_line(&pretty); - let _ = std::fs::write(&live_path, pretty); - } - }); - } - - let interviewer: Arc = if args.auto_approve { - Arc::new(AutoApproveInterviewer) - } else { - Arc::new(super::run_progress::ProgressAwareInterviewer::new( - ConsoleInterviewer::new(styles), - Arc::clone(&progress_ui), - )) - }; - - let dry_run_mode = if args.dry_run { - true - } else { - match fabro_llm::client::Client::from_env().await { - Ok(c) if c.provider_names().is_empty() => { - emit_run_notice( - &emitter, - RunNoticeLevel::Warn, - "dry_run_no_llm", - "No LLM providers configured. Running in dry-run mode.", - ); - true - } - Ok(_) => false, - Err(e) => { - emit_run_notice( - &emitter, - RunNoticeLevel::Warn, - "dry_run_llm_init_failed", - format!("Failed to initialize LLM client: {e}. Running in dry-run mode."), - ); - true - } - } - }; - run_options.dry_run = dry_run_mode; - - if let Some(ref mut cfg) = run_cfg { - run_config::resolve_sandbox_env(cfg)?; - } - - let (model, provider) = resolve_model_provider( - args.model.as_deref(), - args.provider.as_deref(), - run_cfg.as_ref(), - &run_defaults, - &graph, - ); - let provider_enum: Provider = provider - .as_deref() - .map(|s| s.parse::()) - .transpose() - .map_err(|e| anyhow::anyhow!("{e}"))? - .unwrap_or_else(Provider::default_from_env); - - let fallback_chain = if run_cfg.is_some() { - resolve_fallback_chain(provider_enum, &model, run_cfg.as_ref()) - } else { - match run_defaults.llm.as_ref().and_then(|l| l.fallbacks.as_ref()) { - Some(map) => Catalog::builtin().build_fallback_chain(provider_enum, &model, map), - None => Vec::new(), - } - }; - - // Build sandbox env: devcontainer env layered underneath TOML env (TOML wins on conflict, mirrors run_command) - let sandbox_env: HashMap = { - let mut env = devcontainer_env; - if let Some(mut toml_env) = run_cfg - .as_ref() - .and_then(|cfg| cfg.sandbox.as_ref()) - .and_then(|s| s.env.clone()) - .or_else(|| run_defaults.sandbox.as_ref().and_then(|s| s.env.clone())) - { - run_config::resolve_env_refs(&mut toml_env)?; - env.extend(toml_env); - } - env - }; - - // Mint a GitHub App IAT and inject as GITHUB_TOKEN if [github] permissions are declared - let mut sandbox_env = sandbox_env; - let github_permissions = run_cfg - .as_ref() - .and_then(|cfg| cfg.github.as_ref()) - .or(run_defaults.github.as_ref()); - if let Some(gh_cfg) = github_permissions { - if !gh_cfg.permissions.is_empty() { - if let (Some(ref creds), Some(ref url)) = (&github_app, &origin_url) { - match mint_github_token(creds, url, &gh_cfg.permissions).await { - Ok(token) => { - debug!("Minted GitHub IAT for sandbox GITHUB_TOKEN"); - sandbox_env.insert("GITHUB_TOKEN".to_string(), token); - } - Err(e) => { - emit_run_notice( - &emitter, - RunNoticeLevel::Warn, - "github_token_failed", - format!("Failed to mint GitHub token: {e}"), - ); - } - } - } else { - debug!("Skipping GitHub token: no GitHub App credentials or origin URL"); - } - } - } - - // Resolve MCP servers from run defaults - let mcp_servers: Vec = run_cfg - .as_ref() - .map(|cfg| cfg.mcp_servers.clone()) - .unwrap_or_else(|| run_defaults.mcp_servers.clone()) - .clone() - .into_iter() - .map(|(name, entry): (String, fabro_config::mcp::McpServerEntry)| entry.into_config(name)) - .collect(); - - let registry = fabro_workflows::handler::default_registry(interviewer.clone(), { - let sandbox_env = sandbox_env.clone(); - let model = model.clone(); - let mcp_servers = mcp_servers.clone(); - move || { - if dry_run_mode { - None - } else { - let api = - AgentApiBackend::new(model.clone(), provider_enum, fallback_chain.clone()) - .with_env(sandbox_env.clone()) - .with_mcp_servers(mcp_servers.clone()); - let cli = AgentCliBackend::new(model.clone(), provider_enum) - .with_env(sandbox_env.clone()); - Some(Box::new(BackendRouter::new(Box::new(api), cli))) - } - } - }); - let lifecycle = LifecycleOptions { - setup_commands, - setup_command_timeout_ms: 300_000, - devcontainer_phases, - }; - - // Defuse the bootstrap guard — engine.run() has taken ownership of lifecycle status. - status_guard.defuse(); - - let preserve = super::run::resolve_preserve_sandbox( - args.preserve_sandbox, - run_cfg.as_ref(), - &run_defaults, - ); - let run_start = Instant::now(); - let pr_config = if dry_run_mode { - None - } else { - run_options.pull_request().cloned() - }; - let started = start( - persisted, - StartOptions { - init: fabro_workflows::pipeline::InitOptions { - run_id: run_id.clone(), - dry_run: dry_run_mode, - emitter: Arc::clone(&emitter), - sandbox: Arc::clone(&sandbox), - registry: Arc::new(registry), - lifecycle, - run_options, - 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: StartRetroOptions { - 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: StartFinalizeOptions { - preserve_sandbox: preserve, - }, - pull_request: StartPullRequestConfig { - 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 if we changed it (worktree is kept for `fabro cp` access; pruned separately) - if let Some(ref cwd) = original_cwd { - let _ = std::env::set_current_dir(cwd); - } - - progress_ui.lock().expect("progress lock poisoned").finish(); - 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 if !args.no_retro && project_config::is_retro_enabled() { - eprintln!("\n{}", styles.bold.apply_to("=== Retro ===")); - eprintln!("{}", styles.dim.apply_to("Retro unavailable")); - } - 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 - } - }; - - completion_guard.defuse(); - - fabro_util::run_log::deactivate(); - match final_status { - StageStatus::Success | StageStatus::PartialSuccess => Ok(()), - _ => std::process::exit(1), - } -} - -/// Scan a runs directory for an existing directory matching the requested dry-run mode. -fn find_existing_run_dir_in( - base: &std::path::Path, - run_id: &str, - dry_run: bool, -) -> Option { - let suffix = format!("-{run_id}"); - let entries = std::fs::read_dir(base).ok()?; - for entry in entries.flatten() { - let name = entry.file_name(); - let name = name.to_string_lossy(); - if entry.path().is_dir() && run_dir_name_matches_mode(&name, &suffix, dry_run) { - return Some(entry.path()); - } - } - None -} - -/// Scan `~/.fabro/runs/` for an existing directory matching the requested dry-run mode. -fn find_existing_run_dir(run_id: &str, dry_run: bool) -> Option { - let base = dirs::home_dir()?.join(".fabro").join("runs"); - find_existing_run_dir_in(&base, run_id, dry_run) -} - -fn run_dir_name_matches_mode(name: &str, run_id_suffix: &str, dry_run: bool) -> bool { - name.strip_suffix(run_id_suffix) - .is_some_and(|prefix| prefix.ends_with("-dry-run") == dry_run) + // kill(pid, 0) checks liveness without sending a signal + unsafe { libc::kill(pid, 0) == 0 } } #[cfg(test)] mod tests { use super::*; - use chrono::{TimeZone, Utc}; - use fabro_workflows::run_status::{RunStatus, RunStatusRecord, StatusReason}; - - fn sample_run_record() -> RunRecord { - RunRecord { - run_id: "run-1".to_string(), - created_at: Utc::now(), - config: fabro_config::config::FabroConfig::default(), - graph: fabro_graphviz::graph::Graph { - name: "resume".to_string(), - ..Default::default() - }, - workflow_slug: None, - working_directory: std::path::PathBuf::from("/tmp"), - host_repo_path: None, - base_branch: Some("main".to_string()), - labels: HashMap::new(), - } - } #[test] - fn preferred_resume_repo_path_uses_record_host_repo_path_when_present() { - let cwd = tempfile::tempdir().unwrap(); - let host_repo = tempfile::tempdir().unwrap(); - let mut record = sample_run_record(); - record.host_repo_path = Some(host_repo.path().to_string_lossy().to_string()); - - let selected = preferred_resume_repo_path(cwd.path(), Some(&record)); - assert_eq!(selected, host_repo.path()); - } - - #[test] - fn preferred_resume_repo_path_falls_back_when_record_path_is_missing() { - let cwd = tempfile::tempdir().unwrap(); - let mut record = sample_run_record(); - record.host_repo_path = Some(cwd.path().join("missing-repo").display().to_string()); - - let selected = preferred_resume_repo_path(cwd.path(), Some(&record)); - assert_eq!(selected, cwd.path()); - } - - #[test] - fn restore_persisted_for_resume_materializes_metadata_record_without_existing_run_dir() { + fn is_pid_alive_returns_false_for_missing_file() { let dir = tempfile::tempdir().unwrap(); - let repo_dir = dir.path().join("repo"); - let run_dir = dir.path().join("runs").join("run-1"); - std::fs::create_dir_all(&repo_dir).unwrap(); - - let mut record = sample_run_record(); - let created_at = Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).single().unwrap(); - record.created_at = created_at; - record.config = fabro_config::config::FabroConfig { - llm: Some(fabro_config::run::LlmConfig { - model: Some("sonnet".to_string()), - provider: None, - fallbacks: None, - }), - dry_run: Some(true), - ..Default::default() - }; - record.labels = HashMap::from([("old".to_string(), "value".to_string())]); - record.host_repo_path = Some("/tmp/old-repo".to_string()); - - assert!(!run_dir.exists()); - - let persisted = restore_persisted_for_resume( - &record, - record.config.clone(), - run_dir.clone(), - &record.run_id, - HashMap::from([("env".to_string(), "test".to_string())]), - Some("develop".to_string()), - &repo_dir, - ) - .unwrap(); - let loaded = Persisted::load(&run_dir).unwrap(); - - assert!(run_dir.exists()); - assert_eq!(persisted.run_record().created_at, created_at); - assert_eq!(loaded.run_record().created_at, created_at); - assert_eq!(loaded.run_record().working_directory, repo_dir); - assert_eq!( - loaded.run_record().host_repo_path.as_deref(), - Some(repo_dir.to_string_lossy().as_ref()) - ); - assert_eq!(loaded.run_record().base_branch.as_deref(), Some("develop")); - assert_eq!( - loaded.run_record().labels.get("env").map(String::as_str), - Some("test") - ); - assert_eq!( - loaded - .run_record() - .config - .llm - .as_ref() - .and_then(|llm| llm.model.as_deref()), - Some("claude-sonnet-4-6") - ); + assert!(!is_pid_alive(&dir.path().join("run.pid"))); } #[test] - fn find_existing_run_dir_in_respects_dry_run_mode() { - let runs = tempfile::tempdir().unwrap(); - let non_dry = runs.path().join("20260323-run-1"); - let dry = runs.path().join("20260323-dry-run-run-1"); - std::fs::create_dir_all(&non_dry).unwrap(); - std::fs::create_dir_all(&dry).unwrap(); - - assert_eq!( - find_existing_run_dir_in(runs.path(), "run-1", false), - Some(non_dry) - ); - assert_eq!( - find_existing_run_dir_in(runs.path(), "run-1", true), - Some(dry) - ); - } - - #[test] - fn find_existing_run_dir_in_does_not_misclassify_run_ids_containing_dry_run() { - let runs = tempfile::tempdir().unwrap(); - let run_id = "feature-dry-run-fix"; - let non_dry = runs.path().join(format!("20260323-{run_id}")); - let dry = runs.path().join(format!("20260323-dry-run-{run_id}")); - std::fs::create_dir_all(&non_dry).unwrap(); - std::fs::create_dir_all(&dry).unwrap(); - - assert_eq!( - find_existing_run_dir_in(runs.path(), run_id, false), - Some(non_dry) - ); - assert_eq!( - find_existing_run_dir_in(runs.path(), run_id, true), - Some(dry) - ); - } - - #[test] - fn resume_bootstrap_guard_marks_failed_on_drop() { + fn is_pid_alive_returns_false_for_invalid_pid() { let dir = tempfile::tempdir().unwrap(); - - { - let guard = DetachedRunBootstrapGuard::arm(dir.path()).unwrap(); - let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap(); - assert_eq!(record.status, RunStatus::Starting); - assert_eq!(record.reason, Some(StatusReason::SandboxInitializing)); - drop(guard); - } - - let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap(); - assert_eq!(record.status, RunStatus::Failed); - assert_eq!(record.reason, Some(StatusReason::SandboxInitFailed)); - assert!(dir.path().join("run.pid").exists()); + std::fs::write(dir.path().join("run.pid"), "not-a-pid").unwrap(); + assert!(!is_pid_alive(&dir.path().join("run.pid"))); } #[test] - fn resume_bootstrap_guard_does_not_overwrite_after_defuse() { + fn is_pid_alive_returns_true_for_current_process() { let dir = tempfile::tempdir().unwrap(); - let mut guard = DetachedRunBootstrapGuard::arm(dir.path()).unwrap(); - guard.defuse(); - drop(guard); - - let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap(); - assert_eq!(record.status, RunStatus::Starting); - assert_eq!(record.reason, Some(StatusReason::SandboxInitializing)); - } - - #[test] - fn resume_completion_guard_marks_failed_on_drop() { - let dir = tempfile::tempdir().unwrap(); - std::fs::write(dir.path().join("id.txt"), "run-resume").unwrap(); - - { - let _guard = DetachedRunCompletionGuard::arm(dir.path()); - } - - let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap(); - assert_eq!(record.status, RunStatus::Failed); - assert_eq!(record.reason, Some(StatusReason::WorkflowError)); - assert!(dir.path().join("conclusion.json").exists()); + let pid = std::process::id(); + std::fs::write(dir.path().join("run.pid"), pid.to_string()).unwrap(); + assert!(is_pid_alive(&dir.path().join("run.pid"))); } } diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index 988b2ae85..99d8c5edc 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -22,7 +22,8 @@ use fabro_workflows::git::GitSyncStatus; use fabro_workflows::handler::default_registry; use fabro_workflows::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter}; use fabro_workflows::operations::{ - start, StartFinalizeOptions, StartOptions, StartPullRequestConfig, StartRetroOptions, + resume as operations_resume, start, StartFinalizeOptions, StartOptions, StartPullRequestConfig, + StartRetroOptions, }; use fabro_workflows::outcome::StageStatus; use fabro_workflows::outcome::{compute_stage_cost, format_cost}; @@ -30,7 +31,7 @@ use fabro_workflows::pipeline::{ build_conclusion, classify_engine_result, persist_terminal_outcome, Persisted, Validated, }; use fabro_workflows::records::Checkpoint; -use fabro_workflows::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; +use fabro_workflows::run_options::{GitCheckpointOptions, LifecycleOptions}; use indicatif::HumanDuration; use std::time::Duration; use tracing::debug; @@ -755,7 +756,82 @@ pub async fn run_from_record( run_id: Some(record.run_id.clone()), }; - run_command_impl(args, styles, github_app, git_author, Some(record_run)).await + run_command_impl( + args, + styles, + github_app, + git_author, + Some(record_run), + false, + ) + .await +} + +/// Resume an existing workflow run from its persisted checkpoint. +pub async fn resume_from_record( + persisted: Persisted, + run_dir: PathBuf, + run_defaults: FabroConfig, + styles: &'static Styles, + github_app: Option, + git_author: fabro_workflows::git::GitAuthor, +) -> anyhow::Result<()> { + let record = persisted.run_record().clone(); + let record_run = RecordBasedRun { + workflow: WorkflowState::Persisted(Box::new(persisted)), + run_defaults, + }; + + let sandbox_provider = record + .config + .sandbox + .as_ref() + .and_then(|s| s.provider.as_deref()) + .unwrap_or("local") + .parse() + .unwrap_or(SandboxProvider::Local); + let model = record + .config + .llm + .as_ref() + .and_then(|l| l.model.clone()) + .unwrap_or_default(); + let provider = record + .config + .llm + .as_ref() + .and_then(|l| l.provider.clone()) + .filter(|s| !s.is_empty()); + + let args = RunArgs { + workflow: None, + run_dir: Some(run_dir), + dry_run: record.config.dry_run_enabled(), + preflight: false, + auto_approve: record.config.auto_approve_enabled(), + goal: record.config.goal.clone(), + goal_file: None, + model: Some(model), + provider, + verbose: record.config.verbose_enabled(), + sandbox: Some(CliSandboxProvider::from(sandbox_provider)), + label: record + .labels + .iter() + .map(|(k, v)| format!("{k}={v}")) + .collect(), + no_retro: record.config.no_retro_enabled(), + preserve_sandbox: record + .config + .sandbox + .as_ref() + .and_then(|s| s.preserve) + .unwrap_or(false), + detach: false, + run_id: Some(record.run_id.clone()), + }; + + run_command_impl(args, styles, github_app, git_author, Some(record_run), true).await } /// Execute a full workflow run. @@ -778,7 +854,15 @@ pub async fn run_command( run_defaults: resolved_run_defaults, }; - run_command_impl(args, styles, github_app, git_author, Some(record_run)).await + run_command_impl( + args, + styles, + github_app, + git_author, + Some(record_run), + false, + ) + .await } async fn run_command_impl( @@ -787,6 +871,7 @@ async fn run_command_impl( github_app: Option, git_author: fabro_workflows::git::GitAuthor, record_run: Option, + resume: bool, ) -> anyhow::Result<()> { let (workflow, run_defaults) = match record_run { Some(rr) => (rr.workflow, rr.run_defaults), @@ -937,7 +1022,6 @@ async fn run_command_impl( } }; let mut run_cfg = Some(persisted.run_record().config.clone()); - let workflow_slug = persisted.run_record().workflow_slug.clone(); let sandbox_provider = run_cfg .as_ref() .and_then(|cfg| cfg.sandbox.as_ref()) @@ -978,8 +1062,6 @@ async fn run_command_impl( write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?; } - let settings_config = persisted.run_record().config.clone(); - // Now resolve ${env.VARNAME} references for runtime use. if let Some(ref mut cfg) = run_cfg { run_config::resolve_sandbox_env(cfg)?; @@ -1645,30 +1727,6 @@ async fn run_command_impl( None }; - let run_options = RunOptions { - config: settings_config, - run_dir: run_dir.clone(), - cancel_token: None, - dry_run: dry_run_mode, - run_id: run_id.clone(), - labels: persisted.run_record().labels.clone(), - git_author: git_author.clone(), - workflow_slug: workflow_slug.clone(), - github_app: github_app.clone(), - base_branch: persisted - .run_record() - .base_branch - .clone() - .or(detected_base_branch), - host_repo_path: persisted - .run_record() - .host_repo_path - .as_deref() - .map(PathBuf::from) - .or_else(|| Some(original_cwd.clone())), - git, - }; - // Build lifecycle config for sandbox init, setup commands, and devcontainer phases let lifecycle = LifecycleOptions { setup_commands, @@ -1691,46 +1749,46 @@ async fn run_command_impl( let pr_config = if dry_run_mode { None } else { - run_options.pull_request().cloned() + persisted.run_record().config.pull_request.clone() }; - let started = start( - persisted, - StartOptions { - init: fabro_workflows::pipeline::InitOptions { - run_id: run_id.clone(), - dry_run: dry_run_mode, - emitter: Arc::clone(&emitter), - sandbox: Arc::clone(&sandbox), - registry: Arc::new(registry), - lifecycle, - run_options, - 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: StartRetroOptions { - 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: StartFinalizeOptions { preserve_sandbox }, - pull_request: StartPullRequestConfig { - pr_config, - github_app: github_app.clone(), - origin_url: origin_url.clone(), - model: model.clone(), - }, + let start_options = StartOptions { + cancel_token: None, + emitter: Arc::clone(&emitter), + sandbox: Arc::clone(&sandbox), + registry: Arc::new(registry), + lifecycle, + hooks: fabro_hooks::HookConfig { + hooks: run_cfg + .as_ref() + .map(|c| c.hooks.clone()) + .unwrap_or_else(|| run_defaults.hooks.clone()), }, - ) - .await; + sandbox_env, + seed_context: None, + git_author, + git, + github_app: github_app.clone(), + dry_run: dry_run_mode, + retro: StartRetroOptions { + 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: StartFinalizeOptions { preserve_sandbox }, + pull_request: StartPullRequestConfig { + pr_config, + github_app: github_app.clone(), + origin_url: origin_url.clone(), + model: model.clone(), + }, + }; + let started = if resume { + operations_resume(&run_dir, start_options).await + } else { + start(&run_dir, start_options).await + }; let run_duration_ms = run_start.elapsed().as_millis() as u64; let mut completion_guard = DetachedRunCompletionGuard::arm(&run_dir); diff --git a/lib/crates/fabro-cli/src/commands/start.rs b/lib/crates/fabro-cli/src/commands/start.rs index 7e6a3463b..971aac1bc 100644 --- a/lib/crates/fabro-cli/src/commands/start.rs +++ b/lib/crates/fabro-cli/src/commands/start.rs @@ -9,7 +9,7 @@ use super::detached_support::persist_detached_failure; /// /// The engine process reads `run.json` from the run directory and executes the /// workflow. Returns the child process handle (use `.id()` for the PID). -pub fn start_run(run_dir: &Path) -> Result { +pub fn start_run(run_dir: &Path, resume: bool) -> Result { // Validate status is Submitted let status_path = run_dir.join("status.json"); match fabro_workflows::run_status::RunStatusRecord::load(&status_path) { @@ -56,9 +56,11 @@ pub fn start_run(run_dir: &Path) -> Result { return Err(err); } }; - cmd.args(["_run_engine", "--run-dir"]) - .arg(run_dir) - .stdout(stdout_log) + cmd.args(["_run_engine", "--run-dir"]).arg(run_dir); + if resume { + cmd.arg("--resume"); + } + cmd.stdout(stdout_log) .stderr(log_file) .stdin(std::process::Stdio::null()); @@ -134,7 +136,7 @@ mod tests { sample_record().save(dir.path()).unwrap(); std::fs::create_dir(dir.path().join("detach.log")).unwrap(); - let _ = start_run(dir.path()); + let _ = start_run(dir.path(), false); let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap(); assert_eq!( diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 8caa8c2d1..0ad42a9d7 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -90,6 +90,9 @@ enum Command { /// Path to the run directory #[arg(long)] run_dir: PathBuf, + /// Resume from checkpoint instead of fresh start + #[arg(long)] + resume: bool, }, /// Validate a workflow Validate(commands::validate::ValidateArgs), @@ -315,6 +318,7 @@ pub(crate) fn build_github_app_credentials( async fn run_engine_entrypoint( run_dir: PathBuf, + resume: bool, styles: &'static fabro_util::terminal::Styles, ) -> Result<()> { let cli_config = cli_config::load_cli_config(None)?; @@ -355,18 +359,29 @@ async fn run_engine_entrypoint( return Err(err); } - // Use run_from_record: loads config + graph directly from persisted state, - // skipping workflow source loading and preprocessing entirely. - match commands::run::run_from_record( - persisted, - run_dir.clone(), - cli_config, - styles, - github_app, - git_author, - ) - .await - { + let result = if resume { + commands::run::resume_from_record( + persisted, + run_dir.clone(), + cli_config, + styles, + github_app, + git_author, + ) + .await + } else { + commands::run::run_from_record( + persisted, + run_dir.clone(), + cli_config, + styles, + github_app, + git_author, + ) + .await + }; + + match result { Ok(()) => Ok(()), Err(err) => { let _ = commands::detached_support::persist_detached_failure( @@ -752,7 +767,7 @@ async fn main_inner() -> (String, Result<()>) { #[cfg(feature = "sleep_inhibitor")] let _sleep_guard = fabro_beastie::guard(_prevent_idle_sleep); - let child = commands::start::start_run(&run_dir)?; + let child = commands::start::start_run(&run_dir, false)?; if args.detach { println!("{run_id}"); @@ -778,7 +793,7 @@ async fn main_inner() -> (String, Result<()>) { Command::Start { run } => { let base = fabro_workflows::run_lookup::default_runs_base(); let run_info = fabro_workflows::run_lookup::resolve_run(&base, &run)?; - let child = commands::start::start_run(&run_info.path)?; + let child = commands::start::start_run(&run_info.path, false)?; eprintln!("Started engine process (PID {})", child.id()); } Command::Attach { run } => { @@ -792,10 +807,10 @@ async fn main_inner() -> (String, Result<()>) { std::process::exit(1); } } - Command::RunEngine { run_dir } => { + Command::RunEngine { run_dir, resume } => { let styles: &'static fabro_util::terminal::Styles = Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr())); - run_engine_entrypoint(run_dir, styles).await?; + run_engine_entrypoint(run_dir, resume, styles).await?; } Command::Validate(args) => { let styles = fabro_util::terminal::Styles::detect_stderr(); @@ -946,20 +961,16 @@ async fn main_inner() -> (String, Result<()>) { commands::secret::set_command(&args)?; } }, - Command::Resume(mut args) => { + Command::Resume(args) => { let styles: &'static fabro_util::terminal::Styles = Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr())); - let cli_config = cli_config::load_cli_config(None)?; - args.verbose = args.verbose || cli_config.verbose_enabled(); #[cfg(feature = "sleep_inhibitor")] - let _sleep_guard = fabro_beastie::guard(cli_config.prevent_idle_sleep_enabled()); - let github_app = build_github_app_credentials(cli_config.app_id()); - let git_author = fabro_workflows::git::GitAuthor::from_options( - cli_config.git_author().and_then(|a| a.name.clone()), - cli_config.git_author().and_then(|a| a.email.clone()), - ); - commands::resume::resume_command(args, cli_config, styles, github_app, git_author) - .await?; + { + let cli_config = cli_config::load_cli_config(None)?; + let _sleep_guard = + fabro_beastie::guard(cli_config.prevent_idle_sleep_enabled()); + } + commands::resume::resume_command(args, styles).await?; } Command::Rewind(args) => { let styles = fabro_util::terminal::Styles::detect_stderr(); @@ -1114,8 +1125,28 @@ mod tests { let cli = Cli::try_parse_from(["fabro", "_run_engine", "--run-dir", "/tmp/runs/test"]) .expect("should parse"); match cli.command { - Command::RunEngine { run_dir } => { + Command::RunEngine { run_dir, resume } => { assert_eq!(run_dir, std::path::PathBuf::from("/tmp/runs/test")); + assert!(!resume); + } + _ => panic!("unexpected command variant"), + } + } + + #[test] + fn parse_run_engine_with_resume() { + let cli = Cli::try_parse_from([ + "fabro", + "_run_engine", + "--run-dir", + "/tmp/runs/test", + "--resume", + ]) + .expect("should parse"); + match cli.command { + Command::RunEngine { run_dir, resume } => { + assert_eq!(run_dir, std::path::PathBuf::from("/tmp/runs/test")); + assert!(resume); } _ => panic!("unexpected command variant"), } diff --git a/lib/crates/fabro-cli/tests/cli.rs b/lib/crates/fabro-cli/tests/cli.rs index 085f5cd3f..dac6dc28e 100644 --- a/lib/crates/fabro-cli/tests/cli.rs +++ b/lib/crates/fabro-cli/tests/cli.rs @@ -523,12 +523,13 @@ fn resume_help_shows_expected_args() { .args(["resume", "--help"]) .assert() .success() - .stdout(predicate::str::contains("--checkpoint")) - .stdout(predicate::str::contains("--workflow")); + .stdout(predicate::str::contains("--detach")) + .stdout(predicate::str::contains("--checkpoint").not()) + .stdout(predicate::str::contains("--workflow").not()); } #[test] -fn resume_requires_run_or_checkpoint() { +fn resume_requires_run_arg() { arc().args(["resume"]).assert().failure(); } @@ -745,84 +746,6 @@ digraph FooWorkflow { assert_eq!(run_record["workflow_slug"].as_str(), Some("alpha")); } -#[test] -fn resumed_run_preserves_workflow_slug_for_lookup() { - let home = tempfile::tempdir().unwrap(); - let project = tempfile::tempdir().unwrap(); - let workflow_dir = project.path().join("workflows").join("sluggy"); - std::fs::create_dir_all(&workflow_dir).unwrap(); - let workflow_path = workflow_dir.join("workflow.fabro"); - std::fs::write( - &workflow_path, - "\ -digraph BarBaz { - start [shape=Mdiamond, label=\"Start\"] - exit [shape=Msquare, label=\"Exit\"] - start -> exit -} -", - ) - .unwrap(); - let original_run_dir = project.path().join("original-run"); - - arc() - .env("HOME", home.path()) - .current_dir(project.path()) - .args([ - "run", - "--dry-run", - "--auto-approve", - "--no-retro", - "--run-dir", - original_run_dir.to_str().unwrap(), - workflow_path.to_str().unwrap(), - ]) - .assert() - .success(); - - arc() - .env("HOME", home.path()) - .current_dir(project.path()) - .args([ - "resume", - "--checkpoint", - original_run_dir.join("checkpoint.json").to_str().unwrap(), - "--workflow", - workflow_path.to_str().unwrap(), - "--dry-run", - "--auto-approve", - "--no-retro", - ]) - .assert() - .success(); - - arc() - .env("HOME", home.path()) - .current_dir(project.path()) - .args(["attach", "sluggy"]) - .timeout(std::time::Duration::from_secs(10)) - .assert() - .success(); - - let resumed_runs_dir = home.path().join(".fabro").join("runs"); - let resumed_run_dir = std::fs::read_dir(&resumed_runs_dir) - .unwrap() - .flatten() - .map(|entry| entry.path()) - .find(|path| path.is_dir()) - .unwrap_or_else(|| { - panic!( - "expected a resumed run under {}", - resumed_runs_dir.display() - ) - }); - let run_record: serde_json::Value = - serde_json::from_str(&std::fs::read_to_string(resumed_run_dir.join("run.json")).unwrap()) - .unwrap(); - assert_eq!(run_record["graph"]["name"].as_str(), Some("BarBaz")); - assert_eq!(run_record["workflow_slug"].as_str(), Some("sluggy")); -} - #[test] fn dry_run_create_start_attach_works_with_default_run_lookup() { let home = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-workflows/src/error.rs b/lib/crates/fabro-workflows/src/error.rs index 901faffeb..a5d600284 100644 --- a/lib/crates/fabro-workflows/src/error.rs +++ b/lib/crates/fabro-workflows/src/error.rs @@ -233,6 +233,9 @@ pub enum FabroError { #[error("I/O error: {0}")] Io(String), + #[error("Precondition failed: {0}")] + Precondition(String), + #[error("Pipeline cancelled")] Cancelled, } @@ -274,6 +277,7 @@ impl FabroError { | Self::ValidationFailed { .. } | Self::Stylesheet(_) | Self::Checkpoint(_) + | Self::Precondition(_) | Self::Cancelled => false, } } @@ -290,6 +294,7 @@ impl FabroError { | Self::ValidationFailed { .. } | Self::Stylesheet(_) | Self::Checkpoint(_) => FailureCategory::Deterministic, + Self::Precondition(_) => FailureCategory::Structural, Self::Handler { failure_class, .. } | Self::Engine { failure_class, .. } => { *failure_class } diff --git a/lib/crates/fabro-workflows/src/operations/mod.rs b/lib/crates/fabro-workflows/src/operations/mod.rs index 114030cf7..8210a8a06 100644 --- a/lib/crates/fabro-workflows/src/operations/mod.rs +++ b/lib/crates/fabro-workflows/src/operations/mod.rs @@ -1,6 +1,5 @@ mod create; mod fork; -mod restore; mod rewind; mod start; @@ -9,11 +8,11 @@ pub use create::{ ValidateOptions, }; pub use fork::fork; -pub use restore::{restore, RestoreOptions}; pub use rewind::{ build_timeline, find_run_id_by_prefix, load_parallel_map, parse_target, resolve_target, rewind, TimelineEntry, }; pub use start::{ - start, StartFinalizeOptions, StartOptions, StartPullRequestConfig, StartRetroOptions, Started, + resume, start, StartFinalizeOptions, StartOptions, StartPullRequestConfig, StartRetroOptions, + Started, }; diff --git a/lib/crates/fabro-workflows/src/operations/restore.rs b/lib/crates/fabro-workflows/src/operations/restore.rs deleted file mode 100644 index 3f415ce85..000000000 --- a/lib/crates/fabro-workflows/src/operations/restore.rs +++ /dev/null @@ -1,163 +0,0 @@ -use std::path::PathBuf; - -use crate::error::FabroError; -use crate::pipeline::types::PersistOptions; -use crate::pipeline::{self, Persisted, Validated}; -use crate::records::RunRecord; - -use super::create::finalize_config; - -pub struct RestoreOptions { - pub run_dir: PathBuf, - pub run_record: RunRecord, -} - -/// Materialize an existing run record to local disk. -/// -/// Unlike `create()`, this skips parsing, transforms, and validation because -/// the caller already has the resolved graph from the original run. -pub fn restore(options: RestoreOptions) -> Result { - let mut run_record = options.run_record; - finalize_config(&mut run_record.config, &run_record.graph); - let graph = run_record.graph.clone(); - let validated = Validated::new(graph, String::new(), vec![]); - - pipeline::persist( - validated, - PersistOptions { - run_dir: options.run_dir, - run_record, - }, - ) -} - -#[cfg(test)] -mod tests { - use super::*; - use std::collections::HashMap; - - use chrono::{TimeZone, Utc}; - use fabro_config::config::FabroConfig; - use fabro_graphviz::graph::{AttrValue, Graph}; - - fn sample_graph() -> Graph { - let mut graph = Graph::new("restore-test"); - graph.attrs.insert( - "goal".to_string(), - AttrValue::String("Ship feature".to_string()), - ); - graph - } - - fn sample_record() -> RunRecord { - RunRecord { - run_id: "restore-run-123".to_string(), - created_at: Utc.with_ymd_and_hms(2025, 1, 2, 3, 4, 5).single().unwrap(), - config: FabroConfig { - llm: Some(fabro_config::run::LlmConfig { - model: Some("sonnet".to_string()), - provider: None, - fallbacks: None, - }), - pull_request: Some(fabro_config::run::PullRequestConfig { - enabled: false, - ..Default::default() - }), - dry_run: Some(true), - ..Default::default() - }, - graph: sample_graph(), - workflow_slug: Some("restore-slug".to_string()), - working_directory: PathBuf::from("/tmp/original-project"), - host_repo_path: Some("/tmp/original-project".to_string()), - base_branch: Some("main".to_string()), - labels: HashMap::from([("env".to_string(), "test".to_string())]), - } - } - - #[test] - fn restore_roundtrips_and_normalizes_config() { - let temp = tempfile::tempdir().unwrap(); - let run_dir = temp.path().join("run"); - - let persisted = restore(RestoreOptions { - run_dir: run_dir.clone(), - run_record: sample_record(), - }) - .unwrap(); - let loaded = Persisted::load(&run_dir).unwrap(); - - assert_eq!(persisted.run_record().run_id, "restore-run-123"); - assert_eq!( - persisted - .run_record() - .config - .llm - .as_ref() - .and_then(|llm| llm.model.as_deref()), - Some("claude-sonnet-4-6") - ); - assert_eq!( - persisted - .run_record() - .config - .llm - .as_ref() - .and_then(|llm| llm.provider.as_deref()), - Some("anthropic") - ); - assert_eq!( - persisted.run_record().config.goal.as_deref(), - Some("Ship feature") - ); - assert!(persisted.run_record().config.pull_request.is_none()); - assert_eq!( - serde_json::to_value(loaded.run_record()).unwrap(), - serde_json::to_value(persisted.run_record()).unwrap() - ); - } - - #[test] - fn restore_preserves_run_record_fields() { - let temp = tempfile::tempdir().unwrap(); - let run_dir = temp.path().join("run"); - let record = sample_record(); - - restore(RestoreOptions { - run_dir: run_dir.clone(), - run_record: record.clone(), - }) - .unwrap(); - let loaded = Persisted::load(&run_dir).unwrap(); - - assert_eq!(loaded.run_record().run_id, record.run_id); - assert_eq!(loaded.run_record().workflow_slug, record.workflow_slug); - assert_eq!(loaded.run_record().labels, record.labels); - assert_eq!( - loaded.run_record().working_directory, - record.working_directory - ); - assert_eq!(loaded.run_record().host_repo_path, record.host_repo_path); - assert_eq!(loaded.run_record().base_branch, record.base_branch); - } - - #[test] - fn restore_preserves_created_at_and_run_lookup_uses_it_without_start_record() { - let temp = tempfile::tempdir().unwrap(); - let runs_base = temp.path().join("runs"); - let run_dir = runs_base.join("restore-run-123"); - let record = sample_record(); - - restore(RestoreOptions { - run_dir, - run_record: record.clone(), - }) - .unwrap(); - - let runs = crate::run_lookup::scan_runs(&runs_base).unwrap(); - assert_eq!(runs.len(), 1); - assert_eq!(runs[0].run_id, record.run_id); - assert_eq!(runs[0].start_time, record.created_at.to_rfc3339()); - assert_eq!(runs[0].start_time_dt, Some(record.created_at)); - } -} diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs index 04d8de4ec..345a02f9b 100644 --- a/lib/crates/fabro-workflows/src/operations/start.rs +++ b/lib/crates/fabro-workflows/src/operations/start.rs @@ -1,12 +1,17 @@ +use std::collections::HashMap; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; +use crate::context::Context; use crate::error::FabroError; -use crate::event::WorkflowRunEvent; +use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::handler::HandlerRegistry; use crate::outcome::StageStatus; use crate::pipeline::{ self, FinalizeOptions, Finalized, InitOptions, Persisted, PullRequestOptions, RetroOptions, }; +use crate::records::Checkpoint; +use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; pub struct StartRetroOptions { pub enabled: bool, @@ -27,8 +32,27 @@ pub struct StartPullRequestConfig { pub model: String, } +/// Options for `start()` and `resume()`. +/// +/// Fields that are derivable from `RunRecord` (run_id, labels, base_branch, +/// host_repo_path, config, workflow_slug) are read from disk by `run_engine()`. +/// Callers only provide truly external values. pub struct StartOptions { - pub init: InitOptions, + // Truly external (not derivable from RunRecord) + pub cancel_token: Option>, + pub emitter: Arc, + pub sandbox: Arc, + pub registry: Arc, + pub lifecycle: LifecycleOptions, + pub hooks: fabro_hooks::HookConfig, + pub sandbox_env: HashMap, + pub seed_context: Option, + pub git_author: crate::git::GitAuthor, + pub git: Option, + pub github_app: Option, + + // Still external for now — could be derived from RunRecord.config in follow-up + pub dry_run: bool, pub retro: StartRetroOptions, pub finalize: StartFinalizeOptions, pub pull_request: StartPullRequestConfig, @@ -40,10 +64,40 @@ pub struct Started { pub retro_duration: Duration, } -/// Run a persisted workflow through initialize, execute, retro, finalize, and pull_request. -pub async fn start(persisted: Persisted, options: StartOptions) -> Result { +/// Start a fresh workflow run. Errors if a checkpoint already exists (use `resume()` instead). +pub async fn start( + run_dir: &std::path::Path, + options: StartOptions, +) -> Result { + if run_dir.join("checkpoint.json").exists() { + return Err(FabroError::Precondition( + "checkpoint.json exists in run directory — did you mean to resume?".to_string(), + )); + } + let persisted = Persisted::load(run_dir)?; + run_engine(persisted, None, options).await +} + +/// Resume a workflow run from its checkpoint. Errors if no checkpoint is found. +pub async fn resume( + run_dir: &std::path::Path, + options: StartOptions, +) -> Result { + let cp_path = run_dir.join("checkpoint.json"); + let checkpoint = Checkpoint::load(&cp_path) + .map_err(|e| FabroError::Precondition(format!("no checkpoint to resume from: {e}")))?; + let persisted = Persisted::load(run_dir)?; + run_engine(persisted, Some(checkpoint), options).await +} + +/// Shared engine: initialize, execute, retro, finalize, pull_request. +async fn run_engine( + persisted: Persisted, + checkpoint: Option, + options: StartOptions, +) -> Result { let preserve_sandbox = options.finalize.preserve_sandbox; - let sandbox_for_cleanup = Arc::clone(&options.init.sandbox); + let sandbox_for_cleanup = Arc::clone(&options.sandbox); let cleanup_guard = scopeguard::guard((), move |()| { if preserve_sandbox { return; @@ -55,7 +109,40 @@ pub async fn start(persisted: Persisted, options: StartOptions) -> Result>> = Arc::new(Mutex::new(None)); { @@ -145,7 +232,7 @@ mod tests { use crate::handler::start::StartHandler; use crate::handler::{Handler, HandlerRegistry}; use crate::outcome::Outcome; - use crate::run_options::{LifecycleOptions, RunOptions}; + use crate::run_options::LifecycleOptions; const MINIMAL_DOT: &str = r#"digraph Test { graph [goal="Build feature"] @@ -369,23 +456,6 @@ mod tests { .unwrap() } - fn test_run_options(run_dir: &std::path::Path) -> RunOptions { - RunOptions { - config: FabroConfig::default(), - run_dir: run_dir.to_path_buf(), - cancel_token: None, - dry_run: false, - run_id: "run-test".to_string(), - labels: HashMap::new(), - git_author: crate::git::GitAuthor::default(), - workflow_slug: None, - github_app: None, - host_repo_path: None, - base_branch: None, - git: None, - } - } - fn test_registry() -> HandlerRegistry { let mut registry = HandlerRegistry::new(Box::new(StartHandler)); registry.register("start", Box::new(StartHandler)); @@ -395,7 +465,7 @@ mod tests { } fn test_start_options( - run_dir: &std::path::Path, + _run_dir: &std::path::Path, sandbox: Arc, emitter: Arc, registry: Arc, @@ -403,19 +473,18 @@ mod tests { preserve_sandbox: bool, ) -> StartOptions { StartOptions { - init: InitOptions { - run_id: "run-test".to_string(), - dry_run: false, - emitter, - sandbox, - registry, - lifecycle, - run_options: test_run_options(run_dir), - hooks: fabro_hooks::HookConfig { hooks: vec![] }, - sandbox_env: HashMap::new(), - checkpoint: None, - seed_context: None, - }, + cancel_token: None, + emitter, + sandbox, + registry, + lifecycle, + hooks: fabro_hooks::HookConfig { hooks: vec![] }, + sandbox_env: HashMap::new(), + seed_context: None, + git_author: crate::git::GitAuthor::default(), + git: None, + github_app: None, + dry_run: false, retro: StartRetroOptions { enabled: false, dry_run: false, @@ -453,8 +522,9 @@ mod tests { let registry = Arc::new(test_registry()); let (sandbox, cleanup_count) = counting_sandbox(); + persisted_workflow(MINIMAL_DOT, &run_dir); let result = start( - persisted_workflow(MINIMAL_DOT, &run_dir), + &run_dir, test_start_options( &run_dir, sandbox, @@ -485,8 +555,9 @@ mod tests { let sandbox: Arc = Arc::new(LocalSandbox::new(std::env::current_dir().unwrap())); + persisted_workflow(EMIT_DOT, &run_dir); let started = start( - persisted_workflow(EMIT_DOT, &run_dir), + &run_dir, test_start_options( &run_dir, sandbox, @@ -512,22 +583,20 @@ mod tests { } #[tokio::test] - async fn start_runs_loaded_persisted_workflow() { + async fn start_loads_persisted_from_run_dir() { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); - let wrong_run_dir = temp.path().join("wrong-run-dir"); let emitter = Arc::new(EventEmitter::new()); let registry = Arc::new(test_registry()); let sandbox: Arc = Arc::new(LocalSandbox::new(std::env::current_dir().unwrap())); persisted_workflow(MINIMAL_DOT, &run_dir); - let loaded = Persisted::load(&run_dir).unwrap(); let started = start( - loaded, + &run_dir, test_start_options( - &wrong_run_dir, + &run_dir, sandbox, emitter, registry, @@ -544,6 +613,77 @@ mod tests { assert_eq!(started.finalized.conclusion.status, StageStatus::Success); assert!(run_dir.join("conclusion.json").exists()); - assert!(!wrong_run_dir.join("conclusion.json").exists()); + } + + #[tokio::test] + async fn start_errors_when_checkpoint_exists() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + let emitter = Arc::new(EventEmitter::new()); + let registry = Arc::new(test_registry()); + let sandbox: Arc = + Arc::new(LocalSandbox::new(std::env::current_dir().unwrap())); + + persisted_workflow(MINIMAL_DOT, &run_dir); + // Create a fake checkpoint file + std::fs::write(run_dir.join("checkpoint.json"), "{}").unwrap(); + + let result = start( + &run_dir, + test_start_options( + &run_dir, + sandbox, + emitter, + registry, + LifecycleOptions { + setup_commands: vec![], + setup_command_timeout_ms: 1_000, + devcontainer_phases: vec![], + }, + false, + ), + ) + .await; + + assert!( + matches!(&result, Err(crate::error::FabroError::Precondition(_))), + "expected Precondition error, got: {result:?}", + result = result.as_ref().map(|_| "Ok"), + ); + } + + #[tokio::test] + async fn resume_errors_when_checkpoint_missing() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + let emitter = Arc::new(EventEmitter::new()); + let registry = Arc::new(test_registry()); + let sandbox: Arc = + Arc::new(LocalSandbox::new(std::env::current_dir().unwrap())); + + persisted_workflow(MINIMAL_DOT, &run_dir); + + let result = resume( + &run_dir, + test_start_options( + &run_dir, + sandbox, + emitter, + registry, + LifecycleOptions { + setup_commands: vec![], + setup_command_timeout_ms: 1_000, + devcontainer_phases: vec![], + }, + false, + ), + ) + .await; + + assert!( + matches!(&result, Err(crate::error::FabroError::Precondition(_))), + "expected Precondition error, got: {result:?}", + result = result.as_ref().map(|_| "Ok"), + ); } }