From 139bf5cbef70b08387f0780f0a11bc60d000cdd1 Mon Sep 17 00:00:00 2001 From: "brynary-fabro[bot]" <265161896+brynary-fabro[bot]@users.noreply.github.com> Date: Mon, 16 Mar 2026 22:00:05 -0400 Subject: [PATCH] Move sandbox lifecycle into the engine (#22) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This PR moves sandbox lifecycle management (initialization, setup commands, devcontainer phases, and cleanup) from the CLI's `run_command` god function into the workflow engine. Two new engine methods are introduced: `run_with_lifecycle()` orchestrates sandbox init, fires the `SandboxReady` hook (now blocking by default), emits a new `SandboxInitialized` event, handles remote git setup, runs setup commands and devcontainer lifecycle phases, then delegates to the existing `run_internal()` graph execution. `cleanup_sandbox()` fires the `SandboxCleanup` hook and optionally tears down the sandbox. Both `SandboxReady` and `SandboxCleanup` hook events were previously defined but never fired — they now fire naturally within the engine alongside all other hooks. The CLI is simplified significantly: sandbox record persistence and progress UI updates are handled via an event listener for `SandboxInitialized` rather than inline code. The `run_from_branch` resume path also benefits, gaining hook support and proper cleanup for free. A new `LifecycleConfig` struct captures setup commands, timeouts, and devcontainer phases, keeping the engine's API clean. The existing `run()` method is unchanged, so API server and integration tests continue working with pre-initialized sandboxes. The `setup_remote_git` helper is moved from `cli/run.rs` into `engine.rs` since it only depends on sandbox exec. Config is now passed by mutable reference to `run_with_lifecycle` so the engine can fill in remote git fields (base SHA, run branch) that downstream code needs. Comprehensive tests verify event emission ordering, setup command execution/failure, and cleanup behavior. ### Fabro Details
Ran 10 stages in 53m 27s for $14.42 | Stage | Duration | Cost | Retries | |---|---|---|---| | start | 0s | – | 0 | | toolchain | 0s | – | 0 | | preflight_compile | 0s | – | 0 | | preflight_lint | 0s | – | 0 | | implement | 0s | $6.22 | 0 | | simplify_opus | 0s | $2.63 | 0 | | simplify_gemini | 0s | $2.73 | 0 | | simplify_gpt | 0s | $2.84 | 0 | | verify | 0s | – | 0 | | fmt | 0s | – | 0 | | **Total** | **53m 27s** | **$14.42** | **0** |
Ran ImplementAndSimplify.fabro (13 nodes and 16 edges) ```dot digraph ImplementAndSimplify { graph [ goal="Implement and simplify", model_stylesheet=" * { backend: api; model: claude-opus-4-6;} " ] rankdir=LR start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] toolchain [label="Toolchain", shape=parallelogram, script="command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1", max_retries=0] preflight_compile [label="Preflight Compile", shape=parallelogram, script="cargo check -q --workspace 2>&1", max_retries=0] preflight_lint [label="Preflight Lint", shape=parallelogram, script="cargo clippy -q --workspace -- -D warnings 2>&1", max_retries=0] fix_lints [label="Fix Lints", prompt="The preflight lint step failed. Read the build output from context and fix all clippy lint warnings.", max_visits=3] implement [label="Implement", prompt="Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD."] simplify_opus [label="Simplify (Opus)", prompt="@prompts/simplify.md"] simplify_gemini [label="Simplify (Gemini)", prompt="@prompts/simplify.md", model="gemini-3.1-pro-preview-customtools"] simplify_gpt [label="Simplify (GPT-54)", prompt="@prompts/simplify.md", model="gpt-54"] verify [label="Verify", shape=parallelogram, script="cargo clippy -q --workspace -- -D warnings 2>&1 && cargo nextest run --cargo-quiet --workspace --status-level fail 2>&1", goal_gate=true, retry_target="fixup"] fixup [label="Fixup", prompt="The verify step failed. Read the build output from context and fix all clippy lint warnings and test failures.", max_visits=3] fmt [label="Format", shape=parallelogram, script="cargo fmt --all 2>&1", goal_gate=true, max_retries=0] start -> toolchain toolchain -> preflight_compile [condition="outcome=success"] toolchain -> exit preflight_compile -> preflight_lint [condition="outcome=success"] preflight_compile -> exit preflight_lint -> implement [condition="outcome=success"] preflight_lint -> fix_lints fix_lints -> preflight_lint implement -> simplify_opus -> simplify_gemini -> simplify_gpt -> verify verify -> fmt [condition="outcome=success"] verify -> fixup fixup -> verify fmt -> exit } ```
⚒️ Generated with [Fabro](https://fabro.sh) --------- Co-authored-by: Fabro Co-authored-by: Bryan Helmkamp Co-authored-by: Claude Opus 4.6 (1M context) --- docs/agents/hooks.mdx | 6 +- lib/crates/fabro-workflows/src/cli/run.rs | 464 +++++++------------ lib/crates/fabro-workflows/src/engine.rs | 415 +++++++++++++++++ lib/crates/fabro-workflows/src/event.rs | 36 ++ lib/crates/fabro-workflows/src/hook/types.rs | 8 +- 5 files changed, 621 insertions(+), 308 deletions(-) diff --git a/docs/agents/hooks.mdx b/docs/agents/hooks.mdx index 5310dba6d..82a7a7308 100644 --- a/docs/agents/hooks.mdx +++ b/docs/agents/hooks.mdx @@ -94,8 +94,8 @@ Each hook fires on a specific lifecycle event: | `edge_selected` | After an edge is chosen for traversal | Yes | | `parallel_start` | Before parallel branches fan out | No | | `parallel_complete` | After parallel branches merge | No | -| `sandbox_ready` | After the sandbox environment is created (reserved — not yet wired) | No | -| `sandbox_cleanup` | Before the sandbox is torn down (reserved — not yet wired) | No | +| `sandbox_ready` | After the sandbox is initialized and ready | Yes | +| `sandbox_cleanup` | Before the sandbox is torn down | No | | `checkpoint_saved` | After a checkpoint is written to disk | No | | `pre_tool_use` | Before an agent tool call executes | Yes | | `post_tool_use` | After an agent tool call succeeds | No | @@ -137,7 +137,7 @@ sandbox = false Blocking hooks can affect workflow execution. Non-blocking hooks run for side effects only — their decisions are ignored. -**Blocking by default:** `run_start`, `stage_start`, `edge_selected`, `pre_tool_use`. These events represent decision points where a hook can prevent or redirect execution. +**Blocking by default:** `run_start`, `stage_start`, `edge_selected`, `pre_tool_use`, `sandbox_ready`. These events represent decision points where a hook can prevent or redirect execution. **Non-blocking by default:** All other events. Override with `blocking = true` if needed. diff --git a/lib/crates/fabro-workflows/src/cli/run.rs b/lib/crates/fabro-workflows/src/cli/run.rs index 68e9fa5f1..860fc5263 100644 --- a/lib/crates/fabro-workflows/src/cli/run.rs +++ b/lib/crates/fabro-workflows/src/cli/run.rs @@ -798,7 +798,92 @@ pub async fn run_command( None }; - // Wrap emitter in Fabro now so we can share it with exec env callbacks + // Deferred sandbox reference — filled after sandbox creation, consumed by event listeners. + let deferred_sandbox: Arc>>> = Arc::new(Mutex::new(None)); + + // Register SandboxInitialized listener (must happen before emitter is wrapped in Arc) + { + let run_dir_for_listener = run_dir.clone(); + let progress_for_listener = Arc::clone(&progress_ui); + let cwd_for_listener = cwd.to_string_lossy().to_string(); + let ssh_data_host = ssh_config.as_ref().map(|c| c.destination.clone()); + let deferred_sb = Arc::clone(&deferred_sandbox); + let provider = sandbox_provider; // Copy — captured by move closure + emitter.on_event(move |event| { + if let crate::event::WorkflowRunEvent::SandboxInitialized { working_directory } = event + { + progress_for_listener + .lock() + .expect("progress lock poisoned") + .set_working_directory(working_directory.clone()); + + // Build sandbox record from template + let sandbox_info_opt = deferred_sb.lock().unwrap().as_ref().and_then(|sb| { + let info = sb.sandbox_info(); + if info.is_empty() { + None + } else { + Some(info) + } + }); + + let is_docker = provider == SandboxProvider::Docker; + let record = crate::sandbox_record::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_data_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"); + } + } + }); + } + + // Register SSH access listener + if args.ssh { + let deferred_sb_ssh = Arc::clone(&deferred_sandbox); + emitter.on_event(move |event| { + if let crate::event::WorkflowRunEvent::SandboxInitialized { .. } = event { + if let Ok(rt) = tokio::runtime::Handle::try_current() { + let sb_lock = deferred_sb_ssh.lock().unwrap(); + if let Some(ref sb) = *sb_lock { + let sb = Arc::clone(sb); + rt.spawn(async move { + match sb.ssh_access_command().await { + Ok(Some(ssh_command)) => { + // Note: we can't emit from here since emitter is shared; + // SSH access info is logged via tracing. + tracing::info!(ssh_command, "SSH access ready"); + } + Ok(None) => {} + Err(e) => { + tracing::warn!(error = %e, "Failed to create SSH access"); + } + } + }); + } + } + } + }); + } + + // Wrap emitter in Arc so we can share it with exec env callbacks let emitter = Arc::new(emitter); let sandbox: Arc = match sandbox_provider { @@ -881,214 +966,12 @@ pub async fn run_command( } }; - // Initialize sandbox (creates sandbox/container once for the whole run) - sandbox - .initialize() - .await - .map_err(|e| anyhow::anyhow!("Failed to initialize sandbox: {e}"))?; - - progress_ui - .lock() - .expect("progress lock poisoned") - .set_working_directory(sandbox.working_directory().to_string()); - - // Persist sandbox connection info for `fabro cp` - { - let sandbox_info_opt = { - let info = sandbox.sandbox_info(); - if info.is_empty() { - None - } else { - Some(info) - } - }; - let record = match sandbox_provider { - SandboxProvider::Local => crate::sandbox_record::SandboxRecord { - provider: "local".to_string(), - working_directory: sandbox.working_directory().to_string(), - identifier: None, - host_working_directory: None, - container_mount_point: None, - data_host: None, - }, - SandboxProvider::Docker => crate::sandbox_record::SandboxRecord { - provider: "docker".to_string(), - working_directory: sandbox.working_directory().to_string(), - identifier: sandbox_info_opt, - host_working_directory: Some(cwd.to_string_lossy().to_string()), - container_mount_point: Some(sandbox.working_directory().to_string()), - data_host: None, - }, - SandboxProvider::Daytona => crate::sandbox_record::SandboxRecord { - provider: "daytona".to_string(), - working_directory: sandbox.working_directory().to_string(), - identifier: sandbox_info_opt, - host_working_directory: None, - container_mount_point: None, - data_host: None, - }, - #[cfg(feature = "exedev")] - SandboxProvider::Exe => { - // Extract data_host from the ssh access command ("ssh ") - let data_host = sandbox - .ssh_access_command() - .await - .ok() - .flatten() - .and_then(|cmd| cmd.strip_prefix("ssh ").map(String::from)); - crate::sandbox_record::SandboxRecord { - provider: "exe".to_string(), - working_directory: sandbox.working_directory().to_string(), - identifier: sandbox_info_opt, - host_working_directory: None, - container_mount_point: None, - data_host, - } - } - SandboxProvider::Ssh => { - let data_host = ssh_config.as_ref().map(|c| c.destination.clone()); - crate::sandbox_record::SandboxRecord { - provider: "ssh".to_string(), - working_directory: sandbox.working_directory().to_string(), - identifier: sandbox_info_opt, - host_working_directory: None, - container_mount_point: None, - data_host, - } - } - }; - if let Err(e) = record.save(&run_dir.join("sandbox.json")) { - tracing::warn!(error = %e, "Failed to save sandbox record"); - } - } - // Wrap with ReadBeforeWriteSandbox to enforce read-before-write guard + // (delegate_sandbox! macro delegates initialize/cleanup) let sandbox: Arc = Arc::new(fabro_agent::ReadBeforeWriteSandbox::new(sandbox)); - // 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; - }); - } - }); - - // Set up git inside remote sandbox (Daytona or exe.dev) for checkpoint commits - let (remote_base_sha, remote_branch, remote_base_branch) = if sandbox.is_remote() { - match setup_remote_git(&*sandbox, &run_id).await { - Ok((base, branch, base_br)) => (Some(base), Some(branch), base_br), - Err(e) => { - eprintln!( - "{} Remote git setup failed ({e}), running without git checkpoints.", - styles.yellow.apply_to("Warning:"), - ); - (None, None, None) - } - } - } else { - (None, None, None) - }; - - if worktree_base_sha.is_none() { - if let Some(ref sha) = remote_base_sha { - let branch = detected_base_branch - .as_deref() - .or(remote_base_branch.as_deref()); - progress_ui - .lock() - .expect("progress lock poisoned") - .show_base_info(branch, sha); - } - } - - // Create SSH access if requested - if args.ssh { - match sandbox.ssh_access_command().await { - Ok(Some(ssh_command)) => { - emitter.emit(&crate::event::WorkflowRunEvent::SshAccessReady { ssh_command }); - } - Ok(None) => { - eprintln!( - "{} --ssh only works with --sandbox daytona, exe, or ssh, skipping.", - styles.yellow.apply_to("Warning:"), - ); - } - Err(e) => { - eprintln!( - "{} Failed to create SSH access: {e}", - styles.yellow.apply_to("Warning:"), - ); - } - } - } - - // Run setup commands inside the sandbox (once, not per-stage) - if !setup_commands.is_empty() { - emitter.emit(&crate::event::WorkflowRunEvent::SetupStarted { - command_count: setup_commands.len(), - }); - let setup_start = Instant::now(); - for (index, cmd) in setup_commands.iter().enumerate() { - emitter.emit(&crate::event::WorkflowRunEvent::SetupCommandStarted { - command: cmd.clone(), - index, - }); - let cmd_start = Instant::now(); - let result = sandbox - .exec_command(cmd, 300_000, None, None, None) - .await - .map_err(|e| anyhow::anyhow!("Setup command failed: {e}"))?; - let cmd_duration = crate::millis_u64(cmd_start.elapsed()); - if result.exit_code != 0 { - emitter.emit(&crate::event::WorkflowRunEvent::SetupFailed { - command: cmd.clone(), - index, - exit_code: result.exit_code, - stderr: result.stderr.clone(), - }); - anyhow::bail!( - "Setup command failed (exit code {}): {cmd}\n{}", - result.exit_code, - result.stderr, - ); - } - emitter.emit(&crate::event::WorkflowRunEvent::SetupCommandCompleted { - command: cmd.clone(), - index, - exit_code: result.exit_code, - duration_ms: cmd_duration, - }); - } - let setup_duration = crate::millis_u64(setup_start.elapsed()); - emitter.emit(&crate::event::WorkflowRunEvent::SetupCompleted { - duration_ms: setup_duration, - }); - } - - // Run devcontainer lifecycle hooks inside the sandbox - if let Some(ref dc) = devcontainer_config { - let phases: &[(&str, &[fabro_devcontainer::Command])] = &[ - ("on_create", &dc.on_create_commands), - ("post_create", &dc.post_create_commands), - ("post_start", &dc.post_start_commands), - ]; - for (phase, commands) in phases { - devcontainer_bridge::run_devcontainer_lifecycle( - sandbox.as_ref(), - &emitter, - phase, - commands, - 300_000, - ) - .await?; - } - } + // Fill deferred sandbox reference for event listeners registered above + *deferred_sandbox.lock().unwrap() = Some(Arc::clone(&sandbox)); // 6. Resolve backend, model, and provider let (dry_run_mode, llm_client) = if args.dry_run { @@ -1234,8 +1117,8 @@ pub async fn run_command( } // 7. Execute - // Set up metadata branch for git checkpointing (host or remote) - let meta_branch = if worktree_work_dir.is_some() || remote_base_sha.is_some() { + // Set up metadata branch for git checkpointing (host or remote — engine fills remote) + let meta_branch = if worktree_work_dir.is_some() { Some(crate::git::MetadataStore::branch_name(&run_id)) } else { None @@ -1244,19 +1127,15 @@ pub async fn run_command( .as_ref() .map(|c| c.checkpoint.exclude_globs.clone()) .unwrap_or_default(); - let config = RunConfig { + let mut config = RunConfig { run_dir: run_dir.clone(), cancel_token: None, dry_run: dry_run_mode, run_id: run_id.clone(), - git_checkpoint_enabled: if sandbox.is_remote() { - remote_base_sha.is_some() - } else { - worktree_work_dir.is_some() - }, + git_checkpoint_enabled: worktree_work_dir.is_some(), host_repo_path: Some(original_cwd.clone()), - base_sha: worktree_base_sha.or(remote_base_sha), - run_branch: worktree_branch.or(remote_branch), + base_sha: worktree_base_sha, + run_branch: worktree_branch, meta_branch, labels: args .label @@ -1267,7 +1146,7 @@ pub async fn run_command( checkpoint_exclude_globs, github_app: github_app.clone(), git_author, - base_branch: detected_base_branch.or(remote_base_branch), + base_branch: detected_base_branch, pull_request: run_cfg .as_ref() .and_then(|c| c.pull_request.as_ref()) @@ -1281,17 +1160,48 @@ pub async fn run_command( workflow_slug: workflow_slug.clone(), }; + // Build lifecycle config for sandbox init, setup commands, and devcontainer phases + let lifecycle = crate::engine::LifecycleConfig { + setup_commands, + setup_command_timeout_ms: 300_000, + 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() + }, + }; + // Defuse the status guard — engine.run() will write "running" and conclusion handles "concluded" scopeguard::ScopeGuard::into_inner(status_guard); + // 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 = if let Some(ref checkpoint_path) = args.resume { let checkpoint = Checkpoint::load(checkpoint_path)?; engine - .run_from_checkpoint(&graph, &config, &checkpoint) + .run_with_lifecycle(&graph, &mut config, lifecycle, Some(&checkpoint)) .await } else { - engine.run(&graph, &config).await + engine + .run_with_lifecycle(&graph, &mut config, lifecycle, None) + .await }; let run_duration_ms = run_start.elapsed().as_millis() as u64; @@ -1300,13 +1210,13 @@ pub async fn run_command( { let (status, failure_reason) = match &engine_result { - Ok(o) => (o.status.clone(), o.failure_reason().map(String::from)), + Ok(ref o) => (o.status.clone(), o.failure_reason().map(String::from)), Err(e) => (crate::outcome::StageStatus::Fail, Some(e.to_string())), }; // Map engine result to RunStatus + StatusReason let (run_status, status_reason) = match &engine_result { - Ok(o) => match o.status { + Ok(ref o) => match o.status { StageStatus::Success | StageStatus::Skipped => ( crate::run_status::RunStatus::Succeeded, Some(crate::run_status::StatusReason::Completed), @@ -1384,7 +1294,7 @@ pub async fn run_command( // Auto-derive retro (always, cheap) and optionally run retro agent if !args.no_retro && super::project_config::is_retro_enabled() { let (failed, failure_reason) = match &engine_result { - Ok(o) => ( + Ok(ref o) => ( o.status == StageStatus::Fail, o.failure_reason().map(String::from), ), @@ -1597,7 +1507,11 @@ pub async fn run_command( } else { eprintln!("\n{} sandbox preserved", styles.bold.apply_to("Info:")); } - } else if let Err(e) = sandbox.cleanup().await { + } + if let Err(e) = engine + .cleanup_sandbox(&run_id, &graph.name, preserve_sandbox) + .await + { tracing::warn!(error = %e, "Sandbox cleanup failed"); eprintln!( "\n{} sandbox cleanup failed: {e}", @@ -1636,61 +1550,6 @@ fn setup_worktree( Ok((worktree_path.clone(), worktree_path, branch_name, base_sha)) } -/// Set up git inside a remote sandbox (Daytona or exe.dev) for checkpoint commits. -/// Returns (base_sha, branch_name, base_branch) on success. -async fn setup_remote_git( - sandbox: &dyn fabro_agent::Sandbox, - run_id: &str, -) -> anyhow::Result<(String, String, Option)> { - // Get current branch name before creating the run branch - let branch_result = sandbox - .exec_command("git rev-parse --abbrev-ref HEAD", 10_000, None, None, None) - .await - .map_err(|e| anyhow::anyhow!("git rev-parse --abbrev-ref HEAD failed: {e}"))?; - let base_branch = if branch_result.exit_code == 0 { - let name = branch_result.stdout.trim().to_string(); - if name.is_empty() || name == "HEAD" { - None - } else { - Some(name) - } - } else { - None - }; - - // Get current HEAD as base SHA - let sha_result = sandbox - .exec_command("git rev-parse HEAD", 10_000, None, None, None) - .await - .map_err(|e| anyhow::anyhow!("git rev-parse HEAD failed: {e}"))?; - if sha_result.exit_code != 0 { - anyhow::bail!( - "git rev-parse HEAD failed (exit {}): {}", - sha_result.exit_code, - sha_result.stderr - ); - } - let base_sha = sha_result.stdout.trim().to_string(); - - let branch_name = format!("{}{run_id}", crate::git::RUN_BRANCH_PREFIX); - - // Create and checkout a run branch - let checkout_cmd = format!("git checkout -b {branch_name}"); - let checkout_result = sandbox - .exec_command(&checkout_cmd, 10_000, None, None, None) - .await - .map_err(|e| anyhow::anyhow!("git checkout failed: {e}"))?; - if checkout_result.exit_code != 0 { - anyhow::bail!( - "git checkout -b failed (exit {}): {}", - checkout_result.exit_code, - checkout_result.stderr - ); - } - - Ok((base_sha, branch_name, base_branch)) -} - /// Resume a workflow run from a git run branch. /// /// Reads the checkpoint, manifest, and graph DOT from the metadata branch @@ -1845,32 +1704,19 @@ async fn run_from_branch( } }; - // Initialize remote sandboxes and checkout the run branch - if sandbox.is_remote() { - sandbox - .initialize() - .await - .map_err(|e| anyhow::anyhow!("Failed to initialize sandbox: {e}"))?; - - // Fetch and checkout the run branch inside the sandbox - let fetch_cmd = format!("git fetch origin {run_branch} && git checkout {run_branch}"); - let result = sandbox - .exec_command(&fetch_cmd, 60_000, None, None, None) - .await - .map_err(|e| anyhow::anyhow!("Failed to checkout run branch in sandbox: {e}"))?; - if result.exit_code != 0 { - bail!( - "Failed to checkout run branch in sandbox (exit {}): {}", - result.exit_code, - result.stderr - ); - } - } - // Wrap with ReadBeforeWriteSandbox to enforce read-before-write guard let sandbox: Arc = Arc::new(fabro_agent::ReadBeforeWriteSandbox::new(sandbox)); + // For remote sandboxes, prepare a setup command to fetch+checkout the existing run branch + let resume_setup_commands: Vec = if sandbox.is_remote() { + vec![format!( + "git fetch origin {run_branch} && git checkout {run_branch}" + )] + } else { + Vec::new() + }; + // Build interviewer let interviewer: Arc = if args.auto_approve { Arc::new(crate::interviewer::auto_approve::AutoApproveInterviewer) @@ -1917,7 +1763,7 @@ async fn run_from_branch( } let meta_branch = Some(crate::git::MetadataStore::branch_name(&run_id)); - let config = RunConfig { + let mut config = RunConfig { run_dir: run_dir.clone(), cancel_token: None, dry_run: dry_run_mode, @@ -1941,20 +1787,25 @@ async fn run_from_branch( workflow_slug: None, }; + let lifecycle = crate::engine::LifecycleConfig { + setup_commands: resume_setup_commands, + setup_command_timeout_ms: 60_000, + devcontainer_phases: Vec::new(), + }; + let run_start = Instant::now(); let engine_result = engine - .run_from_checkpoint(&graph, &config, &checkpoint) + .run_with_lifecycle(&graph, &mut config, lifecycle, Some(&checkpoint)) .await; let run_duration_ms = run_start.elapsed().as_millis() as u64; // Restore cwd (worktree is kept for `fabro cp` access; pruned separately) let _ = std::env::set_current_dir(&original_cwd); - let _ = sandbox.cleanup().await; // Auto-derive retro if !args.no_retro && super::project_config::is_retro_enabled() { let (failed, failure_reason) = match &engine_result { - Ok(o) => ( + Ok(ref o) => ( o.status == StageStatus::Fail, o.failure_reason().map(String::from), ), @@ -1989,6 +1840,11 @@ async fn run_from_branch( // Write finalize commit with retro.json + final node files (captures last diff.patch) write_finalize_commit(&config, &run_dir).await; + // Cleanup sandbox via engine (fires SandboxCleanup hook) + let _ = engine + .cleanup_sandbox(&config.run_id, &graph.name, false) + .await; + let outcome = engine_result?; eprintln!("\n{}", styles.bold.apply_to("=== Run Result ==="),); diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index 3c147e028..9bc44c091 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -830,6 +830,59 @@ pub async fn git_replace_worktree(sandbox: &dyn Sandbox, path: &str, branch: &st git_add_worktree(sandbox, path, branch).await } +/// Set up git inside a remote sandbox for checkpoint commits. +/// Returns `(base_sha, branch_name, base_branch)` on success. +pub async fn setup_remote_git( + sandbox: &dyn Sandbox, + run_id: &str, +) -> std::result::Result<(String, String, Option), String> { + // Get current branch name before creating the run branch + let branch_result = sandbox + .exec_command("git rev-parse --abbrev-ref HEAD", 10_000, None, None, None) + .await + .map_err(|e| format!("git rev-parse --abbrev-ref HEAD failed: {e}"))?; + let base_branch = if branch_result.exit_code == 0 { + let name = branch_result.stdout.trim().to_string(); + if name.is_empty() || name == "HEAD" { + None + } else { + Some(name) + } + } else { + None + }; + + // Get current HEAD as base SHA + let sha_result = sandbox + .exec_command("git rev-parse HEAD", 10_000, None, None, None) + .await + .map_err(|e| format!("git rev-parse HEAD failed: {e}"))?; + if sha_result.exit_code != 0 { + return Err(format!( + "git rev-parse HEAD failed (exit {}): {}", + sha_result.exit_code, sha_result.stderr + )); + } + let base_sha = sha_result.stdout.trim().to_string(); + + let branch_name = format!("{}{run_id}", crate::git::RUN_BRANCH_PREFIX); + + // Create and checkout a run branch + let checkout_cmd = format!("git checkout -b {branch_name}"); + let checkout_result = sandbox + .exec_command(&checkout_cmd, 10_000, None, None, None) + .await + .map_err(|e| format!("git checkout failed: {e}"))?; + if checkout_result.exit_code != 0 { + return Err(format!( + "git checkout -b failed (exit {}): {}", + checkout_result.exit_code, checkout_result.stderr + )); + } + + Ok((base_sha, branch_name, base_branch)) +} + /// Configuration for a workflow run. pub struct RunConfig { pub run_dir: PathBuf, @@ -866,6 +919,16 @@ pub struct RunConfig { pub workflow_slug: Option, } +/// Configuration for sandbox lifecycle management within the engine. +pub struct LifecycleConfig { + /// Setup commands to run inside the sandbox after initialization. + pub setup_commands: Vec, + /// Timeout in milliseconds for each setup command. + pub setup_command_timeout_ms: u64, + /// Devcontainer lifecycle phases and their commands. + pub devcontainer_phases: Vec<(String, Vec)>, +} + /// The workflow run execution engine. pub struct WorkflowRunEngine { services: EngineServices, @@ -1189,6 +1252,172 @@ impl WorkflowRunEngine { Ok(outcome) } + /// Run a workflow with full sandbox lifecycle management. + /// + /// 1. Initialize sandbox + /// 2. Fire `SandboxReady` hook (blocking — can abort run) + /// 3. Emit `SandboxInitialized` event + /// 4. Remote git setup if `sandbox.is_remote()` + /// 5. Run setup commands + /// 6. Run devcontainer lifecycle phases + /// 7. Execute the workflow graph via `run_internal` + /// + /// The sandbox is left alive after return so the caller can run retro, PR creation, etc. + /// Call `cleanup_sandbox()` when done. + /// + /// The config is taken by mutable reference so the caller retains ownership + /// and can read any fields mutated by remote git setup after the call. + pub async fn run_with_lifecycle( + &self, + graph: &Graph, + config: &mut RunConfig, + lifecycle: LifecycleConfig, + checkpoint: Option<&Checkpoint>, + ) -> Result { + // 1. Initialize sandbox + self.services + .sandbox + .initialize() + .await + .map_err(|e| FabroError::engine(format!("Failed to initialize sandbox: {e}")))?; + + // 2. Fire SandboxReady hook (blocking — can abort run) + { + let hook_ctx = HookContext::new( + HookEvent::SandboxReady, + config.run_id.clone(), + graph.name.clone(), + ); + let decision = self.run_hooks(&hook_ctx, None).await; + if let HookDecision::Block { reason } = decision { + let msg = reason.unwrap_or_else(|| "blocked by SandboxReady hook".into()); + return Err(FabroError::engine(msg)); + } + } + + // 3. Emit SandboxInitialized event + self.services + .emitter + .emit(&WorkflowRunEvent::SandboxInitialized { + working_directory: self.services.sandbox.working_directory().to_string(), + }); + + // 4. Remote git setup if sandbox is remote and config doesn't already have git info + // (skip when resuming from an existing branch — caller sets run_branch/base_sha) + if self.services.sandbox.is_remote() && config.run_branch.is_none() { + match setup_remote_git(self.services.sandbox.as_ref(), &config.run_id).await { + Ok((base_sha, run_branch, base_branch)) => { + config.git_checkpoint_enabled = true; + config.base_sha = Some(base_sha); + config.run_branch = Some(run_branch); + if config.base_branch.is_none() { + config.base_branch = base_branch; + } + config.meta_branch = + Some(crate::git::MetadataStore::branch_name(&config.run_id)); + } + Err(e) => { + tracing::warn!(error = %e, "Remote git setup failed, running without git checkpoints"); + // Leave config.git_checkpoint_enabled as-is (false for remote when no base_sha) + } + } + } + + // 5. Run setup commands + if !lifecycle.setup_commands.is_empty() { + self.services.emitter.emit(&WorkflowRunEvent::SetupStarted { + command_count: lifecycle.setup_commands.len(), + }); + let setup_start = Instant::now(); + for (index, cmd) in lifecycle.setup_commands.iter().enumerate() { + self.services + .emitter + .emit(&WorkflowRunEvent::SetupCommandStarted { + command: cmd.clone(), + index, + }); + let cmd_start = Instant::now(); + let result = self + .services + .sandbox + .exec_command(cmd, lifecycle.setup_command_timeout_ms, None, None, None) + .await + .map_err(|e| FabroError::engine(format!("Setup command failed: {e}")))?; + let cmd_duration = crate::millis_u64(cmd_start.elapsed()); + if result.exit_code != 0 { + self.services.emitter.emit(&WorkflowRunEvent::SetupFailed { + command: cmd.clone(), + index, + exit_code: result.exit_code, + stderr: result.stderr.clone(), + }); + return Err(FabroError::engine(format!( + "Setup command failed (exit code {}): {cmd}\n{}", + result.exit_code, result.stderr, + ))); + } + self.services + .emitter + .emit(&WorkflowRunEvent::SetupCommandCompleted { + command: cmd.clone(), + index, + exit_code: result.exit_code, + duration_ms: cmd_duration, + }); + } + let setup_duration = crate::millis_u64(setup_start.elapsed()); + self.services + .emitter + .emit(&WorkflowRunEvent::SetupCompleted { + duration_ms: setup_duration, + }); + } + + // 6. Run devcontainer lifecycle phases + for (phase, commands) in &lifecycle.devcontainer_phases { + crate::devcontainer_bridge::run_devcontainer_lifecycle( + self.services.sandbox.as_ref(), + &self.services.emitter, + phase, + commands, + lifecycle.setup_command_timeout_ms, + ) + .await + .map_err(|e| FabroError::engine(e.to_string()))?; + } + + // 7. Execute the workflow graph + if let Some(cp) = checkpoint { + self.run_from_checkpoint(graph, config, cp).await + } else { + self.run(graph, config).await + } + } + + /// Fire the `SandboxCleanup` hook and optionally clean up the sandbox. + /// + /// Call this after the retro/PR work is done. The hook fires even when + /// `preserve` is true (observability), but the actual cleanup is skipped. + pub async fn cleanup_sandbox( + &self, + run_id: &str, + workflow_name: &str, + preserve: bool, + ) -> std::result::Result<(), String> { + // Fire SandboxCleanup hook (non-blocking) + let hook_ctx = HookContext::new( + HookEvent::SandboxCleanup, + run_id.to_string(), + workflow_name.to_string(), + ); + let _ = self.run_hooks(&hook_ctx, None).await; + + if !preserve { + self.services.sandbox.cleanup().await?; + } + Ok(()) + } + /// Run a workflow seeded with an existing context. Returns both the outcome /// and the final context so the caller can diff changes. pub async fn run_with_context( @@ -5388,4 +5617,190 @@ mod tests { "work node should have a git checkpoint, but found: {git_checkpoint_node_ids:?}" ); } + + fn test_run_config(run_dir: &std::path::Path, run_id: &str) -> RunConfig { + RunConfig { + run_dir: run_dir.to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: run_id.into(), + git_checkpoint_enabled: false, + host_repo_path: None, + base_sha: None, + run_branch: None, + meta_branch: None, + labels: HashMap::new(), + checkpoint_exclude_globs: Vec::new(), + github_app: None, + git_author: crate::git::GitAuthor::default(), + base_branch: None, + pull_request: None, + asset_globs: Vec::new(), + workflow_slug: None, + } + } + + fn test_lifecycle(setup_commands: Vec) -> LifecycleConfig { + LifecycleConfig { + setup_commands, + setup_command_timeout_ms: 300_000, + devcontainer_phases: Vec::new(), + } + } + + #[tokio::test] + async fn run_with_lifecycle_fires_sandbox_initialized_event() { + let dir = tempfile::tempdir().unwrap(); + let g = simple_graph(); + + let events = Arc::new(std::sync::Mutex::new(Vec::::new())); + let events_clone = events.clone(); + let mut emitter = EventEmitter::new(); + emitter.on_event(move |event| { + events_clone.lock().unwrap().push(event.clone()); + }); + + let engine = WorkflowRunEngine::new(make_registry(), Arc::new(emitter), local_env()); + let mut config = test_run_config(dir.path(), "lifecycle-test"); + let outcome = engine + .run_with_lifecycle(&g, &mut config, test_lifecycle(Vec::new()), None) + .await + .unwrap(); + assert_eq!(outcome.status, StageStatus::Success); + + let collected = events.lock().unwrap(); + let sandbox_init_count = collected + .iter() + .filter(|e| matches!(e, WorkflowRunEvent::SandboxInitialized { .. })) + .count(); + assert_eq!( + sandbox_init_count, 1, + "expected exactly one SandboxInitialized event" + ); + } + + #[tokio::test] + async fn run_with_lifecycle_runs_setup_commands() { + let dir = tempfile::tempdir().unwrap(); + let g = simple_graph(); + + let events = Arc::new(std::sync::Mutex::new(Vec::::new())); + let events_clone = events.clone(); + let mut emitter = EventEmitter::new(); + emitter.on_event(move |event| { + events_clone.lock().unwrap().push(event.clone()); + }); + + let engine = WorkflowRunEngine::new(make_registry(), Arc::new(emitter), local_env()); + let mut config = test_run_config(dir.path(), "setup-test"); + let outcome = engine + .run_with_lifecycle( + &g, + &mut config, + test_lifecycle(vec!["echo hello".to_string()]), + None, + ) + .await + .unwrap(); + assert_eq!(outcome.status, StageStatus::Success); + + let collected = events.lock().unwrap(); + let setup_started = collected + .iter() + .any(|e| matches!(e, WorkflowRunEvent::SetupStarted { .. })); + let setup_completed = collected + .iter() + .any(|e| matches!(e, WorkflowRunEvent::SetupCompleted { .. })); + assert!(setup_started, "expected SetupStarted event"); + assert!(setup_completed, "expected SetupCompleted event"); + } + + #[tokio::test] + async fn run_with_lifecycle_setup_failure_aborts_run() { + let dir = tempfile::tempdir().unwrap(); + let g = simple_graph(); + + let engine = + WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env()); + let mut config = test_run_config(dir.path(), "setup-fail-test"); + let result = engine + .run_with_lifecycle( + &g, + &mut config, + test_lifecycle(vec!["exit 1".to_string()]), + None, + ) + .await; + assert!(result.is_err()); + let err = result.err().unwrap().to_string(); + assert!( + err.contains("Setup command failed"), + "expected setup failure error, got: {err}" + ); + } + + #[tokio::test] + async fn cleanup_sandbox_fires_hook() { + let engine = + WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env()); + // With preserve=true, cleanup should succeed without error + let result = engine.cleanup_sandbox("test-run", "test-wf", true).await; + assert!(result.is_ok()); + } + + #[tokio::test] + async fn run_with_lifecycle_emits_events_in_order() { + let dir = tempfile::tempdir().unwrap(); + let g = simple_graph(); + + let event_names = Arc::new(std::sync::Mutex::new(Vec::::new())); + let names_clone = event_names.clone(); + let mut emitter = EventEmitter::new(); + emitter.on_event(move |event| { + let name = match event { + WorkflowRunEvent::SandboxInitialized { .. } => "SandboxInitialized", + WorkflowRunEvent::SetupStarted { .. } => "SetupStarted", + WorkflowRunEvent::SetupCompleted { .. } => "SetupCompleted", + WorkflowRunEvent::WorkflowRunStarted { .. } => "WorkflowRunStarted", + WorkflowRunEvent::WorkflowRunCompleted { .. } => "WorkflowRunCompleted", + _ => return, + }; + names_clone.lock().unwrap().push(name.to_string()); + }); + + let engine = WorkflowRunEngine::new(make_registry(), Arc::new(emitter), local_env()); + let mut config = test_run_config(dir.path(), "order-test"); + engine + .run_with_lifecycle( + &g, + &mut config, + test_lifecycle(vec!["echo ok".to_string()]), + None, + ) + .await + .unwrap(); + + let names = event_names.lock().unwrap(); + // SandboxInitialized must come before SetupStarted which comes before WorkflowRunStarted + let sandbox_idx = names + .iter() + .position(|n| n == "SandboxInitialized") + .expect("SandboxInitialized not found"); + let setup_idx = names + .iter() + .position(|n| n == "SetupStarted") + .expect("SetupStarted not found"); + let run_started_idx = names + .iter() + .position(|n| n == "WorkflowRunStarted") + .expect("WorkflowRunStarted not found"); + assert!( + sandbox_idx < setup_idx, + "SandboxInitialized ({sandbox_idx}) should come before SetupStarted ({setup_idx})" + ); + assert!( + setup_idx < run_started_idx, + "SetupStarted ({setup_idx}) should come before WorkflowRunStarted ({run_started_idx})" + ); + } } diff --git a/lib/crates/fabro-workflows/src/event.rs b/lib/crates/fabro-workflows/src/event.rs index 66e2d9849..227f18fae 100644 --- a/lib/crates/fabro-workflows/src/event.rs +++ b/lib/crates/fabro-workflows/src/event.rs @@ -201,6 +201,10 @@ pub enum WorkflowRunEvent { Sandbox { event: SandboxEvent, }, + /// Emitted after the sandbox has been initialized (by engine lifecycle). + SandboxInitialized { + working_directory: String, + }, SetupStarted { command_count: usize, }, @@ -533,6 +537,11 @@ impl WorkflowRunEvent { } Self::Agent { .. } => {} Self::Sandbox { .. } => {} + Self::SandboxInitialized { + working_directory, .. + } => { + info!(working_directory, "Sandbox initialized"); + } Self::ParallelEarlyTermination { reason, completed_count, @@ -2500,4 +2509,31 @@ mod tests { assert_eq!(events[0], "started"); assert_eq!(events[1], "completed"); } + + #[test] + fn sandbox_initialized_event_serialization() { + let event = WorkflowRunEvent::SandboxInitialized { + working_directory: "/workspace/project".to_string(), + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("SandboxInitialized")); + assert!(json.contains("/workspace/project")); + let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap(); + assert!(matches!( + deserialized, + WorkflowRunEvent::SandboxInitialized { + working_directory + } if working_directory == "/workspace/project" + )); + } + + #[test] + fn flatten_sandbox_initialized() { + let event = WorkflowRunEvent::SandboxInitialized { + working_directory: "/workspace".to_string(), + }; + let (name, fields) = flatten_event(&event); + assert_eq!(name, "SandboxInitialized"); + assert_eq!(fields["working_directory"], "/workspace"); + } } diff --git a/lib/crates/fabro-workflows/src/hook/types.rs b/lib/crates/fabro-workflows/src/hook/types.rs index 5712a1b17..04ea3782b 100644 --- a/lib/crates/fabro-workflows/src/hook/types.rs +++ b/lib/crates/fabro-workflows/src/hook/types.rs @@ -30,7 +30,11 @@ impl HookEvent { pub fn is_blocking_by_default(self) -> bool { matches!( self, - Self::RunStart | Self::StageStart | Self::EdgeSelected | Self::PreToolUse + Self::RunStart + | Self::StageStart + | Self::EdgeSelected + | Self::PreToolUse + | Self::SandboxReady ) } } @@ -240,6 +244,8 @@ mod tests { assert!(HookEvent::RunStart.is_blocking_by_default()); assert!(HookEvent::StageStart.is_blocking_by_default()); assert!(HookEvent::EdgeSelected.is_blocking_by_default()); + assert!(HookEvent::SandboxReady.is_blocking_by_default()); + assert!(!HookEvent::SandboxCleanup.is_blocking_by_default()); assert!(!HookEvent::RunComplete.is_blocking_by_default()); assert!(!HookEvent::StageFailed.is_blocking_by_default()); assert!(!HookEvent::CheckpointSaved.is_blocking_by_default());