diff --git a/docs/reference/sdk.mdx b/docs/reference/sdk.mdx index 2990fb03f..d2afde968 100644 --- a/docs/reference/sdk.mdx +++ b/docs/reference/sdk.mdx @@ -143,7 +143,7 @@ pub trait Sandbox: Send + Sync { fn working_directory(&self) -> &str; fn platform(&self) -> &str; fn os_version(&self) -> String; - // ... optional methods with defaults: is_remote(), refresh_push_credentials(), etc. + // ... optional methods with defaults: setup_git_for_run(), git_push_branch(), etc. } ``` diff --git a/lib/crates/fabro-agent/src/lib.rs b/lib/crates/fabro-agent/src/lib.rs index 614e6a3a3..e150b663e 100644 --- a/lib/crates/fabro-agent/src/lib.rs +++ b/lib/crates/fabro-agent/src/lib.rs @@ -40,8 +40,8 @@ pub use profiles::{AnthropicProfile, EnvContext, GeminiProfile, OpenAiProfile}; pub use provider_profile::{ProfileCapabilities, ProviderProfile}; pub use read_before_write_sandbox::ReadBeforeWriteSandbox; pub use sandbox::{ - format_lines_numbered, DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, - SandboxEventCallback, + format_lines_numbered, shell_quote, DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, + SandboxEventCallback, WorktreeConfig, WorktreeEvent, WorktreeEventCallback, WorktreeSandbox, }; pub use session::Session; pub use skills::Skill; diff --git a/lib/crates/fabro-agent/src/sandbox.rs b/lib/crates/fabro-agent/src/sandbox.rs index 887a5682e..706ec106c 100644 --- a/lib/crates/fabro-agent/src/sandbox.rs +++ b/lib/crates/fabro-agent/src/sandbox.rs @@ -1,7 +1,7 @@ // Re-export all sandbox types from fabro-sandbox. pub use fabro_sandbox::{ format_lines_numbered, shell_quote, DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, - SandboxEventCallback, + SandboxEventCallback, WorktreeConfig, WorktreeEvent, WorktreeEventCallback, WorktreeSandbox, }; // Re-export the delegate_sandbox! macro at crate root so existing diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index a5ba4dc05..b15ff7681 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -7,7 +7,9 @@ use std::time::Instant; use anyhow::{bail, Context}; use chrono::{Local, Utc}; use clap::{Args, ValueEnum}; -use fabro_agent::{DockerSandbox, DockerSandboxConfig, LocalSandbox, Sandbox}; +use fabro_agent::{ + DockerSandbox, DockerSandboxConfig, LocalSandbox, Sandbox, WorktreeConfig, WorktreeSandbox, +}; use fabro_config::run::{RunDefaults, WorkflowRunConfig}; use fabro_config::{project as project_config, run as run_config, sandbox as sandbox_config}; use fabro_interview::{AutoApproveInterviewer, ConsoleInterviewer, FileInterviewer, Interviewer}; @@ -433,6 +435,15 @@ struct CostAccumulator { has_pricing: bool, } +/// Create a [`LocalSandbox`] wired to emit [`WorkflowRunEvent::Sandbox`] events. +fn local_sandbox_with_callback(cwd: PathBuf, emitter: Arc) -> Arc { + let mut env = LocalSandbox::new(cwd); + env.set_event_callback(Arc::new(move |event| { + emitter.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); + })); + Arc::new(env) +} + /// Result of workflow preparation (shared between `create` and `run` commands). pub(crate) struct PreparedWorkflow { pub source: String, @@ -792,29 +803,33 @@ pub async fn run_command( }; // Determine the working directory strategy. - // Remote sandboxes clone from origin; local runs may use a git worktree. - let workdir_strategy = if sandbox_provider.is_remote() { - WorkdirStrategy::Cloud - } else { - let worktree_mode = resolve_worktree_mode(run_cfg.as_ref(), &run_defaults); - match worktree_mode { - sandbox_config::WorktreeMode::Always => WorkdirStrategy::LocalWorktree, - sandbox_config::WorktreeMode::Clean => { - if git_status.is_clean() { - WorkdirStrategy::LocalWorktree - } else { - WorkdirStrategy::LocalDirectory + // Only the Local provider supports git worktrees on the host. + // Remote sandboxes (Daytona, Exe, SSH) clone from origin inside the sandbox. + // Docker uses the bind-mounted host directory as-is. + let workdir_strategy = match sandbox_provider { + SandboxProvider::Local => { + let worktree_mode = resolve_worktree_mode(run_cfg.as_ref(), &run_defaults); + match worktree_mode { + sandbox_config::WorktreeMode::Always => WorkdirStrategy::LocalWorktree, + sandbox_config::WorktreeMode::Clean => { + if git_status.is_clean() { + WorkdirStrategy::LocalWorktree + } else { + WorkdirStrategy::LocalDirectory + } } - } - sandbox_config::WorktreeMode::Dirty => { - if git_status.is_clean() { - WorkdirStrategy::LocalDirectory - } else { - WorkdirStrategy::LocalWorktree + sandbox_config::WorktreeMode::Dirty => { + if git_status.is_clean() { + WorkdirStrategy::LocalDirectory + } else { + WorkdirStrategy::LocalWorktree + } } + sandbox_config::WorktreeMode::Never => WorkdirStrategy::LocalDirectory, } - sandbox_config::WorktreeMode::Never => WorkdirStrategy::LocalDirectory, } + SandboxProvider::Docker => WorkdirStrategy::LocalDirectory, + _ => WorkdirStrategy::Cloud, }; debug!( ?workdir_strategy, @@ -895,22 +910,29 @@ pub async fn run_command( } } - // Set up git worktree for local isolation. - let (worktree_work_dir, worktree_path, worktree_branch, worktree_base_sha) = - if workdir_strategy == WorkdirStrategy::LocalWorktree { - match setup_worktree(&original_cwd, &run_dir, &run_id) { - Ok((wd, wt, branch, base)) => (Some(wd), Some(wt), Some(branch), Some(base)), - Err(e) => { - eprintln!( - "{} Git worktree setup failed ({e}), running without worktree.", - styles.yellow.apply_to("Warning:"), - ); - (None, None, None, None) - } + // Compute worktree configuration for local isolation. + // The actual git setup (branch, worktree add, reset) happens inside the sandbox + // creation block for SandboxProvider::Local below. + let (mut worktree_path, mut worktree_branch, mut worktree_base_sha) = if workdir_strategy + == WorkdirStrategy::LocalWorktree + { + match fabro_workflows::git::head_sha(&original_cwd) { + Ok(base_sha) => { + let branch_name = format!("{}{run_id}", fabro_workflows::git::RUN_BRANCH_PREFIX); + let wt_path = run_dir.join("worktree"); + (Some(wt_path), Some(branch_name), Some(base_sha)) } - } else { - (None, None, None, None) - }; + Err(e) => { + eprintln!( + "{} Git worktree setup failed ({e}), running without worktree.", + styles.yellow.apply_to("Warning:"), + ); + (None, None, None) + } + } + } else { + (None, None, None) + }; if let Some(ref wt) = worktree_path { progress_ui @@ -1183,12 +1205,43 @@ pub async fn run_command( Arc::new(env) } SandboxProvider::Local => { - let mut env = LocalSandbox::new(cwd.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) + if let (Some(base_sha), Some(branch_name), Some(wt_path)) = ( + worktree_base_sha.as_ref(), + worktree_branch.as_ref(), + worktree_path.as_ref(), + ) { + // Set up a WorktreeSandbox for git-isolated local execution. + let wt_path_str = wt_path.to_string_lossy().into_owned(); + let inner = local_sandbox_with_callback(original_cwd.clone(), Arc::clone(&emitter)); + let wt_config = WorktreeConfig { + branch_name: branch_name.clone(), + base_sha: base_sha.clone(), + worktree_path: wt_path_str, + skip_branch_creation: false, + }; + let mut wt_sandbox = WorktreeSandbox::new(inner, wt_config); + wt_sandbox.set_event_callback(Arc::clone(&emitter).worktree_callback()); + + match wt_sandbox.initialize().await { + Ok(()) => { + std::env::set_current_dir(wt_path)?; + Arc::new(wt_sandbox) as Arc + } + Err(e) => { + eprintln!( + "{} Git worktree setup failed ({e}), running without worktree.", + styles.yellow.apply_to("Warning:"), + ); + // Reset so RunConfig does not enable git checkpointing + worktree_path = None; + worktree_branch = None; + worktree_base_sha = None; + local_sandbox_with_callback(cwd.clone(), Arc::clone(&emitter)) + } + } + } else { + local_sandbox_with_callback(cwd.clone(), Arc::clone(&emitter)) + } } }; @@ -1336,7 +1389,7 @@ pub async fn run_command( // 7. Execute // Set up metadata branch for git checkpointing (host or remote — engine fills remote) - let meta_branch = if worktree_work_dir.is_some() { + let meta_branch = if worktree_path.is_some() { Some(fabro_workflows::git::MetadataStore::branch_name(&run_id)) } else { None @@ -1350,7 +1403,7 @@ pub async fn run_command( cancel_token: None, dry_run: dry_run_mode, run_id: run_id.clone(), - git_checkpoint_enabled: worktree_work_dir.is_some(), + git_checkpoint_enabled: worktree_path.is_some(), host_repo_path: Some(original_cwd.clone()), base_sha: worktree_base_sha, run_branch: worktree_branch, @@ -1750,29 +1803,6 @@ pub async fn run_command( } } -/// Set up a git worktree for an isolated workflow run. -/// Caller must have already verified the repo is clean via `git::ensure_clean`. -/// Returns (work_dir, worktree_path, branch_name, base_sha) on success. -fn setup_worktree( - original_cwd: &std::path::Path, - run_dir: &std::path::Path, - run_id: &str, -) -> anyhow::Result<(PathBuf, PathBuf, String, String)> { - let base_sha = - fabro_workflows::git::head_sha(original_cwd).map_err(|e| anyhow::anyhow!("{e}"))?; - let branch_name = format!("{}{run_id}", fabro_workflows::git::RUN_BRANCH_PREFIX); - fabro_workflows::git::create_branch(original_cwd, &branch_name) - .map_err(|e| anyhow::anyhow!("{e}"))?; - - let worktree_path = run_dir.join("worktree"); - fabro_workflows::git::replace_worktree(original_cwd, &worktree_path, &branch_name) - .map_err(|e| anyhow::anyhow!("{e}"))?; - - std::env::set_current_dir(&worktree_path)?; - - Ok((worktree_path.clone(), worktree_path, branch_name, base_sha)) -} - /// Resume a workflow run from a git run branch. /// /// Reads the checkpoint, manifest, and graph DOT from the metadata branch @@ -1865,21 +1895,31 @@ async fn run_from_branch( }; let emitter = Arc::new(EventEmitter::new()); - let (sandbox, worktree_path): (Arc, Option) = + let (sandbox, _worktree_path): (Arc, Option) = match sandbox_provider { SandboxProvider::Local | SandboxProvider::Docker => { - // Re-attach worktree to the existing run branch + // Re-attach worktree to the existing run branch via WorktreeSandbox. let wt = run_dir.join("worktree"); - fabro_workflows::git::replace_worktree(&original_cwd, &wt, run_branch).map_err( - |e| anyhow::anyhow!("failed to attach worktree to {run_branch}: {e}"), - )?; + let wt_str = wt.to_string_lossy().into_owned(); + + let inner = local_sandbox_with_callback(original_cwd.clone(), Arc::clone(&emitter)); + let wt_config = WorktreeConfig { + branch_name: run_branch.to_string(), + 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.initialize().await.map_err(|e| { + anyhow::anyhow!("failed to attach worktree to {run_branch}: {e}") + })?; std::env::set_current_dir(&wt)?; - let mut env = fabro_agent::LocalSandbox::new(wt.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), Some(wt)) + ( + Arc::new(wt_sandbox) as Arc, + Some(wt), + ) } #[cfg(feature = "exedev")] SandboxProvider::Exe => { @@ -1928,14 +1968,8 @@ async fn run_from_branch( 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() - }; + // Let the sandbox provide any commands needed to resume on the existing run branch + let resume_setup_commands: Vec = sandbox.resume_setup_commands(run_branch); // Build interviewer let interviewer: Arc = if args.auto_approve { @@ -1990,11 +2024,7 @@ async fn run_from_branch( cancel_token: None, dry_run: dry_run_mode, run_id: run_id.clone(), - git_checkpoint_enabled: if sandbox.is_remote() { - true - } else { - worktree_path.is_some() - }, + git_checkpoint_enabled: true, // always true for resume (worktree or sandbox git is set up) host_repo_path: Some(original_cwd.clone()), base_sha, run_branch: Some(run_branch.to_string()), diff --git a/lib/crates/fabro-sandbox/src/daytona/mod.rs b/lib/crates/fabro-sandbox/src/daytona/mod.rs index ab8c9b140..3ee5c4bf2 100644 --- a/lib/crates/fabro-sandbox/src/daytona/mod.rs +++ b/lib/crates/fabro-sandbox/src/daytona/mod.rs @@ -692,8 +692,34 @@ impl Sandbox for DaytonaSandbox { .unwrap_or_default() } - fn is_remote(&self) -> bool { - true + async fn setup_git_for_run(&self, run_id: &str) -> Result, String> { + crate::setup_git_via_exec(self, run_id).await.map(Some) + } + + fn resume_setup_commands(&self, run_branch: &str) -> Vec { + vec![format!( + "git fetch origin {run_branch} && git checkout {run_branch}" + )] + } + + async fn git_push_branch(&self, branch: &str) -> bool { + crate::git_push_via_exec(self, branch).await + } + + fn parallel_worktree_path( + &self, + _run_dir: &std::path::Path, + run_id: &str, + node_id: &str, + key: &str, + ) -> String { + format!( + "{}/.fabro/runs/{}/parallel/{}/{}", + self.working_directory(), + run_id, + node_id, + key + ) } async fn ssh_access_command(&self) -> Result, String> { diff --git a/lib/crates/fabro-sandbox/src/exe/mod.rs b/lib/crates/fabro-sandbox/src/exe/mod.rs index 7a4360f31..da1a96b5b 100644 --- a/lib/crates/fabro-sandbox/src/exe/mod.rs +++ b/lib/crates/fabro-sandbox/src/exe/mod.rs @@ -688,8 +688,34 @@ impl Sandbox for ExeSandbox { Ok(()) } - fn is_remote(&self) -> bool { - true + async fn setup_git_for_run(&self, run_id: &str) -> Result, String> { + crate::setup_git_via_exec(self, run_id).await.map(Some) + } + + fn resume_setup_commands(&self, run_branch: &str) -> Vec { + vec![format!( + "git fetch origin {run_branch} && git checkout {run_branch}" + )] + } + + async fn git_push_branch(&self, branch: &str) -> bool { + crate::git_push_via_exec(self, branch).await + } + + fn parallel_worktree_path( + &self, + _run_dir: &std::path::Path, + run_id: &str, + node_id: &str, + key: &str, + ) -> String { + format!( + "{}/.fabro/runs/{}/parallel/{}/{}", + self.working_directory(), + run_id, + node_id, + key + ) } async fn ssh_access_command(&self) -> Result, String> { diff --git a/lib/crates/fabro-sandbox/src/lib.rs b/lib/crates/fabro-sandbox/src/lib.rs index b54a8d703..a1d31fb07 100644 --- a/lib/crates/fabro-sandbox/src/lib.rs +++ b/lib/crates/fabro-sandbox/src/lib.rs @@ -2,6 +2,8 @@ pub mod sandbox; pub mod read_guard; +pub mod worktree; + #[cfg(feature = "ssh")] pub(crate) mod ssh_common; @@ -27,12 +29,14 @@ pub mod daytona; pub mod test_support; pub use sandbox::{ - format_lines_numbered, shell_quote, DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, - SandboxEventCallback, + format_lines_numbered, git_push_via_exec, setup_git_via_exec, shell_quote, DirEntry, + ExecResult, GitRunInfo, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, }; pub use read_guard::ReadBeforeWriteSandbox; +pub use worktree::{WorktreeConfig, WorktreeEvent, WorktreeEventCallback, WorktreeSandbox}; + #[cfg(feature = "local")] pub use local::LocalSandbox; diff --git a/lib/crates/fabro-sandbox/src/sandbox.rs b/lib/crates/fabro-sandbox/src/sandbox.rs index 6c12d7c6c..915a116ae 100644 --- a/lib/crates/fabro-sandbox/src/sandbox.rs +++ b/lib/crates/fabro-sandbox/src/sandbox.rs @@ -6,6 +6,16 @@ use std::path::Path; use std::sync::Arc; use tokio_util::sync::CancellationToken; +/// Git command prefix that disables background maintenance. +const GIT: &str = "git -c maintenance.auto=0 -c gc.auto=0"; + +/// Information returned when a sandbox sets up git for a workflow run. +pub struct GitRunInfo { + pub base_sha: String, + pub run_branch: String, + pub base_branch: Option, +} + /// Generates an `#[async_trait] impl Sandbox` block for a decorator type /// that wraps an `Arc`. The caller provides custom method /// implementations; all remaining trait methods delegate to the inner field. @@ -110,8 +120,30 @@ macro_rules! delegate_sandbox { self.$field.set_autostop_interval(minutes).await } - fn is_remote(&self) -> bool { - self.$field.is_remote() + async fn setup_git_for_run(&self, run_id: &str) -> Result, String> { + self.$field.setup_git_for_run(run_id).await + } + + fn resume_setup_commands(&self, run_branch: &str) -> Vec { + self.$field.resume_setup_commands(run_branch) + } + + async fn git_push_branch(&self, branch: &str) -> bool { + self.$field.git_push_branch(branch).await + } + + fn host_git_dir(&self) -> Option<&str> { + self.$field.host_git_dir() + } + + fn parallel_worktree_path( + &self, + run_dir: &std::path::Path, + run_id: &str, + node_id: &str, + key: &str, + ) -> String { + self.$field.parallel_worktree_path(run_dir, run_id, node_id, key) } async fn ssh_access_command(&self) -> Result, String> { @@ -400,11 +432,48 @@ pub trait Sandbox: Send + Sync { Ok(()) } - /// Whether this sandbox runs on a remote machine (e.g. Daytona, exe.dev). - fn is_remote(&self) -> bool { + /// Set up git state for a new workflow run. + /// Sandboxes that manage their own git clone (e.g., remote VMs) should create + /// a run branch and return the git info. Local sandboxes return `None`. + async fn setup_git_for_run(&self, _run_id: &str) -> Result, String> { + Ok(None) + } + + /// Commands to run inside the sandbox when resuming on an existing run branch. + fn resume_setup_commands(&self, _run_branch: &str) -> Vec { + Vec::new() + } + + /// Push a run branch to origin from inside the sandbox. + /// Returns `true` if the push was handled. When `false`, the engine will + /// attempt a host-side push instead. + async fn git_push_branch(&self, _branch: &str) -> bool { false } + /// The host-accessible path to this sandbox's git worktree, if applicable. + /// When `Some`, the engine runs git operations (add, commit) from the host. + fn host_git_dir(&self) -> Option<&str> { + None + } + + /// Compute the filesystem path for a parallel branch worktree. + fn parallel_worktree_path( + &self, + run_dir: &std::path::Path, + _run_id: &str, + node_id: &str, + key: &str, + ) -> String { + run_dir + .join("parallel") + .join(node_id) + .join(key) + .join("worktree") + .to_string_lossy() + .into_owned() + } + /// Return an SSH command string for connecting to this sandbox, if supported. async fn ssh_access_command(&self) -> Result, String> { Ok(None) @@ -455,6 +524,83 @@ pub fn shell_quote(s: &str) -> String { ) } +/// Helper for sandbox implementations that manage git internally. +/// Executes git commands inside the sandbox to create a run branch. +pub async fn setup_git_via_exec(sandbox: &dyn Sandbox, run_id: &str) -> Result { + // Get current branch name + 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!("fabro/run/{run_id}"); + + // 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(GitRunInfo { + base_sha, + run_branch: branch_name, + base_branch, + }) +} + +/// Helper for sandbox implementations that manage git internally. +/// Pushes a branch to origin via exec_command inside the sandbox. +pub async fn git_push_via_exec(sandbox: &dyn Sandbox, branch: &str) -> bool { + if let Err(e) = sandbox.refresh_push_credentials().await { + tracing::warn!(error = %e, "Failed to refresh push credentials"); + } + let cmd = format!("{GIT} push origin {branch}"); + match sandbox.exec_command(&cmd, 60_000, None, None, None).await { + Ok(r) if r.exit_code == 0 => { + tracing::info!(branch, "Pushed run branch to origin"); + true + } + Ok(r) => { + tracing::warn!(branch, exit_code = r.exit_code, "Failed to push run branch"); + false + } + Err(e) => { + tracing::warn!(branch, error = %e, "Failed to push run branch"); + false + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/lib/crates/fabro-sandbox/src/ssh/mod.rs b/lib/crates/fabro-sandbox/src/ssh/mod.rs index f2546fd15..31ca93a95 100644 --- a/lib/crates/fabro-sandbox/src/ssh/mod.rs +++ b/lib/crates/fabro-sandbox/src/ssh/mod.rs @@ -558,8 +558,34 @@ impl Sandbox for SshSandbox { Ok(()) } - fn is_remote(&self) -> bool { - true + async fn setup_git_for_run(&self, run_id: &str) -> Result, String> { + crate::setup_git_via_exec(self, run_id).await.map(Some) + } + + fn resume_setup_commands(&self, run_branch: &str) -> Vec { + vec![format!( + "git fetch origin {run_branch} && git checkout {run_branch}" + )] + } + + async fn git_push_branch(&self, branch: &str) -> bool { + crate::git_push_via_exec(self, branch).await + } + + fn parallel_worktree_path( + &self, + _run_dir: &std::path::Path, + run_id: &str, + node_id: &str, + key: &str, + ) -> String { + format!( + "{}/.fabro/runs/{}/parallel/{}/{}", + self.working_directory(), + run_id, + node_id, + key + ) } async fn ssh_access_command(&self) -> Result, String> { diff --git a/lib/crates/fabro-sandbox/src/test_support.rs b/lib/crates/fabro-sandbox/src/test_support.rs index e8b01b1e2..79cbb65d0 100644 --- a/lib/crates/fabro-sandbox/src/test_support.rs +++ b/lib/crates/fabro-sandbox/src/test_support.rs @@ -20,8 +20,12 @@ pub struct MockSandbox { pub written_files: Mutex>, /// Captures the `timeout_ms` argument from `exec_command` calls. pub captured_timeout: Mutex>, - /// Captures the `command` argument from `exec_command` calls. + /// Captures the `command` argument from `exec_command` calls (last only). pub captured_command: Mutex>, + /// Captures all `command` arguments from `exec_command` calls in order. + pub captured_commands: Mutex>, + /// Captures all `working_dir` arguments from `exec_command` calls in order. + pub captured_working_dirs: Mutex>>, /// Captures the `env_vars` argument from `exec_command` calls. pub captured_env_vars: Mutex>>, pub event_callback: Option, @@ -67,6 +71,8 @@ impl Default for MockSandbox { written_files: Mutex::new(Vec::new()), captured_timeout: Mutex::new(None), captured_command: Mutex::new(None), + captured_commands: Mutex::new(Vec::new()), + captured_working_dirs: Mutex::new(Vec::new()), captured_env_vars: Mutex::new(None), event_callback: None, } @@ -126,7 +132,7 @@ impl Sandbox for MockSandbox { &self, command: &str, timeout_ms: u64, - _working_dir: Option<&str>, + working_dir: Option<&str>, env_vars: Option<&std::collections::HashMap>, _cancel_token: Option, ) -> Result { @@ -138,6 +144,14 @@ impl Sandbox for MockSandbox { .captured_command .lock() .expect("captured_command lock poisoned") = Some(command.to_string()); + self.captured_commands + .lock() + .expect("captured_commands lock poisoned") + .push(command.to_string()); + self.captured_working_dirs + .lock() + .expect("captured_working_dirs lock poisoned") + .push(working_dir.map(String::from)); *self .captured_env_vars .lock() diff --git a/lib/crates/fabro-sandbox/src/worktree.rs b/lib/crates/fabro-sandbox/src/worktree.rs new file mode 100644 index 000000000..2f53e10a7 --- /dev/null +++ b/lib/crates/fabro-sandbox/src/worktree.rs @@ -0,0 +1,747 @@ +use async_trait::async_trait; +use std::collections::HashMap; +use std::path::Path; +use std::sync::Arc; +use tokio_util::sync::CancellationToken; + +use crate::{shell_quote, DirEntry, ExecResult, GrepOptions, Sandbox}; + +/// Git command prefix that disables background maintenance. +const GIT: &str = "git -c maintenance.auto=0 -c gc.auto=0"; + +// --------------------------------------------------------------------------- +// Public types +// --------------------------------------------------------------------------- + +/// Events emitted during worktree lifecycle operations. +pub enum WorktreeEvent { + BranchCreated { branch: String, sha: String }, + WorktreeAdded { path: String, branch: String }, + WorktreeRemoved { path: String }, +} + +/// Callback type for worktree lifecycle events. +pub type WorktreeEventCallback = Arc; + +/// Configuration for a `WorktreeSandbox`. +pub struct WorktreeConfig { + pub branch_name: String, + pub base_sha: String, + pub worktree_path: String, + /// Skip branch creation and hard reset (for resume, where branch already exists). + pub skip_branch_creation: bool, +} + +/// Wraps any `Sandbox`, manages a git worktree lifecycle in `initialize()`/`cleanup()`, +/// and overrides `working_directory()` and `exec_command()` to use the worktree path. +/// +/// `initialize()` and `cleanup()` do NOT call the inner sandbox's lifecycle methods. +/// The inner sandbox's lifecycle is managed separately by the caller. +pub struct WorktreeSandbox { + inner: Arc, + config: WorktreeConfig, + event_callback: Option, + initialized: std::sync::atomic::AtomicBool, +} + +impl WorktreeSandbox { + /// Create a new `WorktreeSandbox` wrapping `inner` with the given configuration. + pub fn new(inner: Arc, config: WorktreeConfig) -> Self { + Self { + inner, + config, + event_callback: None, + initialized: std::sync::atomic::AtomicBool::new(false), + } + } + + /// Set the callback to receive worktree lifecycle events. + pub fn set_event_callback(&mut self, cb: WorktreeEventCallback) { + self.event_callback = Some(cb); + } + + /// The git branch name managed by this sandbox. + pub fn branch_name(&self) -> &str { + &self.config.branch_name + } + + /// The base commit SHA used when initializing the worktree. + pub fn base_sha(&self) -> &str { + &self.config.base_sha + } + + /// The filesystem path to the worktree directory. + pub fn worktree_path(&self) -> &str { + &self.config.worktree_path + } + + fn emit(&self, event: WorktreeEvent) { + if let Some(ref cb) = self.event_callback { + cb(event); + } + } + + fn resolve_path(&self, path: &str) -> String { + if std::path::Path::new(path).is_absolute() { + path.to_string() + } else { + format!("{}/{path}", self.config.worktree_path) + } + } +} + +// --------------------------------------------------------------------------- +// Sandbox implementation +// --------------------------------------------------------------------------- + +#[async_trait] +impl Sandbox for WorktreeSandbox { + // --- Lifecycle --- + + /// Set up the git worktree: + /// 1. Best-effort remove any stale worktree at `path` (so the branch is free to be updated). + /// 2. Unless `skip_branch_creation`: force-create the branch at `base_sha`, emit `BranchCreated`. + /// 3. Add the worktree, emit `WorktreeAdded`. + /// + /// Does NOT call `inner.initialize()`. + async fn initialize(&self) -> Result<(), String> { + if self + .initialized + .swap(true, std::sync::atomic::Ordering::Relaxed) + { + return Ok(()); + } + let path = shell_quote(&self.config.worktree_path); + let branch = shell_quote(&self.config.branch_name); + let sha = shell_quote(&self.config.base_sha); + + // Best-effort remove any stale worktree registration + directory first, + // so that the branch is not "in use" when we try to force-update it. + let rm_cmd = format!("{GIT} worktree remove --force {path}"); + let _ = self + .inner + .exec_command(&rm_cmd, 30_000, None, None, None) + .await; + + if !self.config.skip_branch_creation { + let cmd = format!("{GIT} branch --force {branch} {sha}"); + let result = self + .inner + .exec_command(&cmd, 30_000, None, None, None) + .await?; + if result.exit_code != 0 { + return Err(format!( + "git branch --force failed (exit {}): {}", + result.exit_code, + result.stderr.trim() + )); + } + self.emit(WorktreeEvent::BranchCreated { + branch: self.config.branch_name.clone(), + sha: self.config.base_sha.clone(), + }); + } + + let add_cmd = format!("{GIT} worktree add {path} {branch}"); + let result = self + .inner + .exec_command(&add_cmd, 30_000, None, None, None) + .await?; + if result.exit_code != 0 { + return Err(format!( + "git worktree add failed (exit {}): {}", + result.exit_code, + result.stderr.trim() + )); + } + self.emit(WorktreeEvent::WorktreeAdded { + path: self.config.worktree_path.clone(), + branch: self.config.branch_name.clone(), + }); + + Ok(()) + } + + /// No-op — the worktree must survive cleanup for `fabro cp` access. + /// Worktrees are pruned separately by `system prune`. + async fn cleanup(&self) -> Result<(), String> { + Ok(()) + } + + fn working_directory(&self) -> &str { + &self.config.worktree_path + } + + /// Execute a command, defaulting `working_dir` to the worktree path when `None`. + async fn exec_command( + &self, + command: &str, + timeout_ms: u64, + working_dir: Option<&str>, + env_vars: Option<&HashMap>, + cancel_token: Option, + ) -> Result { + let wd = working_dir.unwrap_or(&self.config.worktree_path); + self.inner + .exec_command(command, timeout_ms, Some(wd), env_vars, cancel_token) + .await + } + + // --- Delegated methods --- + + async fn read_file( + &self, + path: &str, + offset: Option, + limit: Option, + ) -> Result { + let resolved = self.resolve_path(path); + self.inner.read_file(&resolved, offset, limit).await + } + + async fn write_file(&self, path: &str, content: &str) -> Result<(), String> { + let resolved = self.resolve_path(path); + self.inner.write_file(&resolved, content).await + } + + async fn delete_file(&self, path: &str) -> Result<(), String> { + let resolved = self.resolve_path(path); + self.inner.delete_file(&resolved).await + } + + async fn file_exists(&self, path: &str) -> Result { + let resolved = self.resolve_path(path); + self.inner.file_exists(&resolved).await + } + + async fn list_directory( + &self, + path: &str, + depth: Option, + ) -> Result, String> { + let resolved = self.resolve_path(path); + self.inner.list_directory(&resolved, depth).await + } + + async fn grep( + &self, + pattern: &str, + path: &str, + options: &GrepOptions, + ) -> Result, String> { + let resolved = self.resolve_path(path); + self.inner.grep(pattern, &resolved, options).await + } + + async fn glob(&self, pattern: &str, path: Option<&str>) -> Result, String> { + let resolved = path.map(|p| self.resolve_path(p)); + let glob_path = resolved.as_deref().unwrap_or(&self.config.worktree_path); + self.inner.glob(pattern, Some(glob_path)).await + } + + async fn download_file_to_local( + &self, + remote_path: &str, + local_path: &Path, + ) -> Result<(), String> { + let resolved = self.resolve_path(remote_path); + self.inner + .download_file_to_local(&resolved, local_path) + .await + } + + async fn upload_file_from_local( + &self, + local_path: &Path, + remote_path: &str, + ) -> Result<(), String> { + let resolved = self.resolve_path(remote_path); + self.inner + .upload_file_from_local(local_path, &resolved) + .await + } + + fn platform(&self) -> &str { + self.inner.platform() + } + + fn os_version(&self) -> String { + self.inner.os_version() + } + + fn sandbox_info(&self) -> String { + self.inner.sandbox_info() + } + + async fn refresh_push_credentials(&self) -> Result<(), String> { + self.inner.refresh_push_credentials().await + } + + async fn set_autostop_interval(&self, minutes: i32) -> Result<(), String> { + self.inner.set_autostop_interval(minutes).await + } + + fn host_git_dir(&self) -> Option<&str> { + Some(&self.config.worktree_path) + } + + async fn setup_git_for_run(&self, run_id: &str) -> Result, String> { + self.inner.setup_git_for_run(run_id).await + } + + fn resume_setup_commands(&self, run_branch: &str) -> Vec { + self.inner.resume_setup_commands(run_branch) + } + + async fn git_push_branch(&self, branch: &str) -> bool { + self.inner.git_push_branch(branch).await + } + + fn parallel_worktree_path( + &self, + run_dir: &Path, + run_id: &str, + node_id: &str, + key: &str, + ) -> String { + self.inner + .parallel_worktree_path(run_dir, run_id, node_id, key) + } + + async fn ssh_access_command(&self) -> Result, String> { + self.inner.ssh_access_command().await + } + + fn origin_url(&self) -> Option<&str> { + self.inner.origin_url() + } + + async fn get_preview_url( + &self, + port: u16, + ) -> Result)>, String> { + self.inner.get_preview_url(port).await + } + + fn mark_agent_read(&self, path: &str) { + let resolved = self.resolve_path(path); + self.inner.mark_agent_read(&resolved); + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::MockSandbox; + use std::sync::Mutex; + + fn make_config(wt_path: &str) -> WorktreeConfig { + WorktreeConfig { + branch_name: "fabro/run/test-branch".to_string(), + base_sha: "abc123def456".to_string(), + worktree_path: wt_path.to_string(), + skip_branch_creation: false, + } + } + + fn make_config_skip(wt_path: &str) -> WorktreeConfig { + WorktreeConfig { + branch_name: "fabro/run/test-branch".to_string(), + base_sha: "abc123def456".to_string(), + worktree_path: wt_path.to_string(), + skip_branch_creation: true, + } + } + + /// Create a shared mock and return both the `Arc` (passed to WorktreeSandbox) + /// and the `Arc` (used to assert captured state). + fn make_mock() -> (Arc, Arc) { + let mock = Arc::new(MockSandbox::linux()); + let as_sandbox: Arc = mock.clone(); + (as_sandbox, mock) + } + + // ----------------------------------------------------------------------- + // initialize() — full setup (skip_branch_creation = false) + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn initialize_issues_correct_git_commands() { + let (inner, mock) = make_mock(); + let wt = WorktreeSandbox::new(inner, make_config("/tmp/wt")); + + wt.initialize().await.unwrap(); + + let cmds = mock.captured_commands.lock().unwrap().clone(); + // worktree remove (best-effort), branch --force, worktree add + assert_eq!(cmds.len(), 3, "expected 3 git commands, got: {cmds:?}"); + assert!( + cmds[0].contains("worktree remove --force"), + "cmd[0]: {}", + cmds[0] + ); + assert!(cmds[1].contains("branch --force"), "cmd[1]: {}", cmds[1]); + assert!(cmds[2].contains("worktree add"), "cmd[2]: {}", cmds[2]); + } + + #[tokio::test] + async fn initialize_emits_branch_and_worktree_events() { + let (inner, _mock) = make_mock(); + let mut wt = WorktreeSandbox::new(inner, make_config("/tmp/wt")); + + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let events_clone = Arc::clone(&events); + wt.set_event_callback(Arc::new(move |event| { + let label = match &event { + WorktreeEvent::BranchCreated { .. } => "BranchCreated", + WorktreeEvent::WorktreeAdded { .. } => "WorktreeAdded", + WorktreeEvent::WorktreeRemoved { .. } => "WorktreeRemoved", + }; + events_clone.lock().unwrap().push(label.to_string()); + })); + + wt.initialize().await.unwrap(); + + let captured = events.lock().unwrap(); + assert_eq!(*captured, vec!["BranchCreated", "WorktreeAdded"]); + } + + #[tokio::test] + async fn initialize_uses_shell_quoted_values_in_commands() { + let (inner, mock) = make_mock(); + let config = WorktreeConfig { + branch_name: "fabro/run/my-branch".to_string(), + base_sha: "deadbeef".to_string(), + worktree_path: "/tmp/my worktree".to_string(), // path with space + skip_branch_creation: false, + }; + let wt = WorktreeSandbox::new(inner, config); + + wt.initialize().await.unwrap(); + + let cmds = mock.captured_commands.lock().unwrap().clone(); + // The path "/tmp/my worktree" should be quoted in the worktree remove command (cmd[0]) + assert!( + cmds[0].contains("'/tmp/my worktree'") || cmds[0].contains("\"/tmp/my worktree\""), + "worktree path should be shell-quoted: {}", + cmds[0] + ); + } + + // ----------------------------------------------------------------------- + // initialize() — skip_branch_creation = true + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn initialize_skip_branch_creation_issues_only_worktree_commands() { + let (inner, mock) = make_mock(); + let wt = WorktreeSandbox::new(inner, make_config_skip("/tmp/wt")); + + wt.initialize().await.unwrap(); + + let cmds = mock.captured_commands.lock().unwrap().clone(); + // Only worktree remove (best-effort) and worktree add + assert_eq!(cmds.len(), 2, "expected 2 git commands, got: {cmds:?}"); + assert!( + cmds[0].contains("worktree remove --force"), + "cmd[0]: {}", + cmds[0] + ); + assert!(cmds[1].contains("worktree add"), "cmd[1]: {}", cmds[1]); + } + + #[tokio::test] + async fn initialize_skip_branch_creation_emits_only_worktree_added() { + let (inner, _mock) = make_mock(); + let mut wt = WorktreeSandbox::new(inner, make_config_skip("/tmp/wt")); + + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let events_clone = Arc::clone(&events); + wt.set_event_callback(Arc::new(move |event| { + let label = match &event { + WorktreeEvent::BranchCreated { .. } => "BranchCreated", + WorktreeEvent::WorktreeAdded { .. } => "WorktreeAdded", + WorktreeEvent::WorktreeRemoved { .. } => "WorktreeRemoved", + }; + events_clone.lock().unwrap().push(label.to_string()); + })); + + wt.initialize().await.unwrap(); + + let captured = events.lock().unwrap(); + assert_eq!(*captured, vec!["WorktreeAdded"]); + } + + // ----------------------------------------------------------------------- + // initialize() — error propagation + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn initialize_propagates_error_on_nonzero_exit() { + let inner: Arc = Arc::new(MockSandbox { + exec_result: ExecResult { + stdout: String::new(), + stderr: "fatal: not a git repo".to_string(), + exit_code: 128, + timed_out: false, + duration_ms: 5, + }, + ..MockSandbox::linux() + }); + let wt = WorktreeSandbox::new(inner, make_config("/tmp/wt")); + + let result = wt.initialize().await; + + assert!(result.is_err(), "should return Err on non-zero exit"); + let err = result.unwrap_err(); + assert!( + err.contains("branch --force failed") || err.contains("128"), + "error should mention the failure: {err}" + ); + } + + // ----------------------------------------------------------------------- + // cleanup() + // ----------------------------------------------------------------------- + + // ----------------------------------------------------------------------- + // working_directory() + // ----------------------------------------------------------------------- + + #[test] + fn working_directory_returns_worktree_path() { + let (inner, _mock) = make_mock(); + let wt = WorktreeSandbox::new(inner, make_config("/tmp/my_worktree")); + + assert_eq!(wt.working_directory(), "/tmp/my_worktree"); + } + + // ----------------------------------------------------------------------- + // exec_command() working_dir defaulting + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn exec_command_none_working_dir_defaults_to_worktree_path() { + let (inner, mock) = make_mock(); + let wt = WorktreeSandbox::new(inner, make_config("/tmp/wt")); + + wt.exec_command("echo hello", 5000, None, None, None) + .await + .unwrap(); + + let wdirs = mock.captured_working_dirs.lock().unwrap().clone(); + assert_eq!( + wdirs.last(), + Some(&Some("/tmp/wt".to_string())), + "None working_dir should be replaced with worktree path" + ); + } + + #[tokio::test] + async fn exec_command_explicit_working_dir_passes_through() { + let (inner, mock) = make_mock(); + let wt = WorktreeSandbox::new(inner, make_config("/tmp/wt")); + + wt.exec_command("echo hello", 5000, Some("/explicit/path"), None, None) + .await + .unwrap(); + + let wdirs = mock.captured_working_dirs.lock().unwrap().clone(); + assert_eq!( + wdirs.last(), + Some(&Some("/explicit/path".to_string())), + "explicit working_dir should be passed through unchanged" + ); + } + + // ----------------------------------------------------------------------- + // Accessors + // ----------------------------------------------------------------------- + + // ----------------------------------------------------------------------- + // Bug: cleanup() destroys worktree, breaking `fabro cp` + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn cleanup_should_preserve_worktree_for_post_run_access() { + // The worktree directory must survive cleanup() so that `fabro cp` can + // access run artifacts afterward. It is pruned separately by `system prune`. + // LocalSandbox.cleanup() was a no-op; WorktreeSandbox should match. + let (inner, mock) = make_mock(); + let wt = WorktreeSandbox::new(inner, make_config("/tmp/wt")); + + wt.cleanup().await.unwrap(); + + let cmds = mock.captured_commands.lock().unwrap().clone(); + assert!( + cmds.is_empty(), + "cleanup should not issue destructive git commands \ + (worktree must be preserved for fabro cp), but got: {cmds:?}" + ); + } + + // ----------------------------------------------------------------------- + // Bug: initialize() is not idempotent — double call destroys worktree + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn initialize_is_idempotent_on_second_call() { + // engine.run_with_lifecycle() calls sandbox.initialize() unconditionally, + // even when run.rs already called it during sandbox construction. + // The second call must be a no-op; it must NOT re-run + // `git worktree remove --force` which would destroy the worktree. + let (inner, mock) = make_mock(); + let wt = WorktreeSandbox::new(inner, make_config("/tmp/wt")); + + wt.initialize().await.unwrap(); + let first_count = mock.captured_commands.lock().unwrap().len(); + + wt.initialize().await.unwrap(); + let second_count = mock.captured_commands.lock().unwrap().len(); + + assert_eq!( + first_count, + second_count, + "second initialize() should be a no-op, but it issued {} additional commands", + second_count - first_count + ); + } + + // ----------------------------------------------------------------------- + // Bug: file operations resolve against inner working_directory, not worktree + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn grep_should_search_worktree_not_inner_working_directory() { + // WorktreeSandbox delegates grep() to the inner sandbox without path + // adjustment. When the inner LocalSandbox was created with original_cwd, + // grep("pattern", ".") searches the original repo instead of the worktree. + let original = + std::env::temp_dir().join(format!("fabro-test-original-{}", uuid::Uuid::new_v4())); + let worktree = + std::env::temp_dir().join(format!("fabro-test-worktree-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&original).unwrap(); + std::fs::create_dir_all(&worktree).unwrap(); + + // Put a marker file ONLY in the worktree directory + std::fs::write(worktree.join("marker.txt"), "UNIQUE_WORKTREE_MARKER").unwrap(); + + let inner: Arc = Arc::new(crate::local::LocalSandbox::new(original.clone())); + let config = WorktreeConfig { + branch_name: "test-branch".into(), + base_sha: "abc123".into(), + worktree_path: worktree.to_string_lossy().to_string(), + skip_branch_creation: false, + }; + let wt = WorktreeSandbox::new(inner, config); + + // working_directory() correctly returns the worktree path + assert_eq!(wt.working_directory(), worktree.to_string_lossy().as_ref()); + + // grep with "." should search the worktree, not the original repo + let results = wt + .grep("UNIQUE_WORKTREE_MARKER", ".", &GrepOptions::default()) + .await + .unwrap(); + assert!( + !results.is_empty(), + "grep(\".\") should search the worktree directory, not the inner sandbox's working directory" + ); + + std::fs::remove_dir_all(&original).ok(); + std::fs::remove_dir_all(&worktree).ok(); + } + + #[tokio::test] + async fn glob_should_search_worktree_when_path_is_none() { + // WorktreeSandbox delegates glob() to the inner sandbox without path + // adjustment. LocalSandbox::glob(pattern, None) defaults to + // self.working_directory, which is the original repo path. + let original = + std::env::temp_dir().join(format!("fabro-test-original-{}", uuid::Uuid::new_v4())); + let worktree = + std::env::temp_dir().join(format!("fabro-test-worktree-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&original).unwrap(); + std::fs::create_dir_all(&worktree).unwrap(); + + // Put a file ONLY in the worktree directory + std::fs::write(worktree.join("worktree_only.txt"), "content").unwrap(); + + let inner: Arc = Arc::new(crate::local::LocalSandbox::new(original.clone())); + let config = WorktreeConfig { + branch_name: "test-branch".into(), + base_sha: "abc123".into(), + worktree_path: worktree.to_string_lossy().to_string(), + skip_branch_creation: false, + }; + let wt = WorktreeSandbox::new(inner, config); + + let results = wt.glob("*.txt", None).await.unwrap(); + assert!( + results.iter().any(|r| r.contains("worktree_only.txt")), + "glob(pattern, None) should search the worktree directory, not the inner sandbox's working directory. Got: {results:?}" + ); + + std::fs::remove_dir_all(&original).ok(); + std::fs::remove_dir_all(&worktree).ok(); + } + + #[tokio::test] + async fn read_file_relative_should_resolve_against_worktree() { + // WorktreeSandbox delegates read_file() to the inner sandbox without + // path adjustment. Relative paths resolve against the inner + // LocalSandbox's working_directory (original repo), not the worktree. + let original = + std::env::temp_dir().join(format!("fabro-test-original-{}", uuid::Uuid::new_v4())); + let worktree = + std::env::temp_dir().join(format!("fabro-test-worktree-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&original).unwrap(); + std::fs::create_dir_all(&worktree).unwrap(); + + // Put the file ONLY in the worktree directory + std::fs::write(worktree.join("only_in_worktree.txt"), "worktree content").unwrap(); + + let inner: Arc = Arc::new(crate::local::LocalSandbox::new(original.clone())); + let config = WorktreeConfig { + branch_name: "test-branch".into(), + base_sha: "abc123".into(), + worktree_path: worktree.to_string_lossy().to_string(), + skip_branch_creation: false, + }; + let wt = WorktreeSandbox::new(inner, config); + + let result = wt.read_file("only_in_worktree.txt", None, None).await; + assert!( + result.is_ok(), + "read_file with relative path should resolve against worktree, not inner sandbox's working directory. Error: {}", + result.unwrap_err() + ); + + std::fs::remove_dir_all(&original).ok(); + std::fs::remove_dir_all(&worktree).ok(); + } + + // ----------------------------------------------------------------------- + // Accessors + // ----------------------------------------------------------------------- + + #[test] + fn accessors_return_config_values() { + let (inner, _mock) = make_mock(); + let config = WorktreeConfig { + branch_name: "my-branch".to_string(), + base_sha: "sha123".to_string(), + worktree_path: "/path/to/wt".to_string(), + skip_branch_creation: false, + }; + let wt = WorktreeSandbox::new(inner, config); + + assert_eq!(wt.branch_name(), "my-branch"); + assert_eq!(wt.base_sha(), "sha123"); + assert_eq!(wt.worktree_path(), "/path/to/wt"); + } +} diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index d1292c53b..2348f9d8a 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -762,28 +762,6 @@ pub async fn git_push_host( } } -/// Push the run branch to origin inside a remote sandbox (best-effort). -async fn git_push_remote(sandbox: &dyn Sandbox, branch: &str) -> bool { - if let Err(e) = sandbox.refresh_push_credentials().await { - tracing::warn!(error = %e, "Failed to refresh push credentials"); - } - let cmd = format!("{GIT_REMOTE} push origin {branch}"); - match sandbox.exec_command(&cmd, 60_000, None, None, None).await { - Ok(r) if r.exit_code == 0 => { - tracing::info!(branch, "Pushed run branch to origin"); - true - } - Ok(r) => { - tracing::warn!(branch, exit_code = r.exit_code, "Failed to push run branch"); - false - } - Err(e) => { - tracing::warn!(branch, error = %e, "Failed to push run branch"); - false - } - } -} - /// Run a git diff via the sandbox. async fn git_diff(sandbox: &dyn Sandbox, base: &str) -> Option { let cmd = format!("{GIT_REMOTE} diff {base} HEAD"); @@ -837,59 +815,6 @@ 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, @@ -1304,7 +1229,7 @@ impl WorkflowRunEngine { /// 1. Initialize sandbox /// 2. Fire `SandboxReady` hook (blocking — can abort run) /// 3. Emit `SandboxInitialized` event - /// 4. Remote git setup if `sandbox.is_remote()` + /// 4. Sandbox git setup via `sandbox.setup_git_for_run()` /// 5. Run setup commands /// 6. Run devcontainer lifecycle phases /// 7. Execute the workflow graph via `run_internal` @@ -1349,23 +1274,30 @@ impl WorkflowRunEngine { 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 + // 4. Sandbox git setup — let the sandbox set up its own git state if needed // (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)) => { + if config.run_branch.is_none() { + match self + .services + .sandbox + .setup_git_for_run(&config.run_id) + .await + { + Ok(Some(info)) => { config.git_checkpoint_enabled = true; - config.base_sha = Some(base_sha); - config.run_branch = Some(run_branch); + config.base_sha = Some(info.base_sha); + config.run_branch = Some(info.run_branch); if config.base_branch.is_none() { - config.base_branch = base_branch; + config.base_branch = info.base_branch; } config.meta_branch = Some(crate::git::MetadataStore::branch_name(&config.run_id)); } + Ok(None) => { + // Sandbox does not manage git internally (e.g. local sandbox) + } 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) + tracing::warn!(error = %e, "Sandbox git setup failed, running without git checkpoints"); } } } @@ -1538,9 +1470,9 @@ impl WorkflowRunEngine { }; self.services.set_git_state(git_state); - // Local git checkpoint: sandbox is local and checkpointing is enabled + // Host-side git checkpoint: sandbox has a host-accessible worktree let local_git_checkpoint = - config.git_checkpoint_enabled && !self.services.sandbox.is_remote(); + config.git_checkpoint_enabled && self.services.sandbox.host_git_dir().is_some(); self.services .emitter @@ -2246,8 +2178,9 @@ impl WorkflowRunEngine { // Push run branch (skip in dry-run mode) if !config.dry_run { if let Some(ref branch) = config.run_branch { - let push_ok = if self.services.sandbox.is_remote() { - git_push_remote(&*self.services.sandbox, branch).await + let push_ok = if self.services.sandbox.git_push_branch(branch).await + { + true } else if let Some(ref repo_path) = config.host_repo_path { let refspec = format!("refs/heads/{branch}"); git_push_host( diff --git a/lib/crates/fabro-workflows/src/event.rs b/lib/crates/fabro-workflows/src/event.rs index 227f18fae..03c7d3bd0 100644 --- a/lib/crates/fabro-workflows/src/event.rs +++ b/lib/crates/fabro-workflows/src/event.rs @@ -1,9 +1,10 @@ use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::Arc; use serde::{Deserialize, Serialize}; use crate::outcome::StageUsage; -use fabro_agent::{AgentEvent, SandboxEvent}; +use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback}; /// Events emitted during workflow run execution for observability. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -1070,6 +1071,22 @@ impl EventEmitter { pub fn touch(&self) { self.last_event_at.store(epoch_millis(), Ordering::Relaxed); } + + /// Build a [`WorktreeEventCallback`] that forwards worktree lifecycle events as + /// [`WorkflowRunEvent`]s on this emitter. + pub fn worktree_callback(self: Arc) -> WorktreeEventCallback { + Arc::new(move |event| match event { + WorktreeEvent::BranchCreated { branch, sha } => { + self.emit(&WorkflowRunEvent::GitBranch { branch, sha }); + } + WorktreeEvent::WorktreeAdded { path, branch } => { + self.emit(&WorkflowRunEvent::GitWorktreeAdd { path, branch }); + } + WorktreeEvent::WorktreeRemoved { path } => { + self.emit(&WorkflowRunEvent::GitWorktreeRemove { path }); + } + }) + } } #[cfg(test)] diff --git a/lib/crates/fabro-workflows/src/handler/parallel.rs b/lib/crates/fabro-workflows/src/handler/parallel.rs index b1e73dc55..b8545f646 100644 --- a/lib/crates/fabro-workflows/src/handler/parallel.rs +++ b/lib/crates/fabro-workflows/src/handler/parallel.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use std::time::Instant; use async_trait::async_trait; -use fabro_agent::Sandbox; +use fabro_agent::{Sandbox, WorktreeConfig, WorktreeSandbox}; use tokio::sync::Semaphore; use crate::context::keys; @@ -13,118 +13,11 @@ use crate::error::FabroError; use crate::event::WorkflowRunEvent; use crate::millis_u64; use crate::outcome::{Outcome, StageStatus}; -use fabro_agent::LocalSandbox; use fabro_graphviz::graph::{Graph, Node}; use fabro_hooks::{HookContext, HookEvent}; use super::{EngineServices, Handler}; -// --------------------------------------------------------------------------- -// WorktreeSandbox — decorates a Sandbox with a custom working dir -// --------------------------------------------------------------------------- - -/// Wraps an existing `Sandbox` so that all operations use a -/// different working directory (the worktree path inside a remote sandbox). -struct WorktreeSandbox { - inner: Arc, - worktree_dir: String, -} - -#[async_trait] -impl Sandbox for WorktreeSandbox { - async fn read_file( - &self, - path: &str, - offset: Option, - limit: Option, - ) -> Result { - self.inner.read_file(path, offset, limit).await - } - async fn write_file(&self, path: &str, content: &str) -> Result<(), String> { - self.inner.write_file(path, content).await - } - async fn delete_file(&self, path: &str) -> Result<(), String> { - self.inner.delete_file(path).await - } - async fn file_exists(&self, path: &str) -> Result { - self.inner.file_exists(path).await - } - async fn list_directory( - &self, - path: &str, - depth: Option, - ) -> Result, String> { - self.inner.list_directory(path, depth).await - } - async fn exec_command( - &self, - command: &str, - timeout_ms: u64, - working_dir: Option<&str>, - env_vars: Option<&std::collections::HashMap>, - cancel_token: Option, - ) -> Result { - // Default to worktree dir when no explicit working_dir is given - let wd = working_dir.unwrap_or(&self.worktree_dir); - self.inner - .exec_command(command, timeout_ms, Some(wd), env_vars, cancel_token) - .await - } - async fn grep( - &self, - pattern: &str, - path: &str, - options: &fabro_agent::sandbox::GrepOptions, - ) -> Result, String> { - self.inner.grep(pattern, path, options).await - } - async fn glob(&self, pattern: &str, path: Option<&str>) -> Result, String> { - self.inner.glob(pattern, path).await - } - async fn download_file_to_local( - &self, - remote_path: &str, - local_path: &std::path::Path, - ) -> Result<(), String> { - self.inner - .download_file_to_local(remote_path, local_path) - .await - } - async fn upload_file_from_local( - &self, - local_path: &std::path::Path, - remote_path: &str, - ) -> Result<(), String> { - self.inner - .upload_file_from_local(local_path, remote_path) - .await - } - async fn initialize(&self) -> Result<(), String> { - self.inner.initialize().await - } - async fn cleanup(&self) -> Result<(), String> { - self.inner.cleanup().await - } - fn working_directory(&self) -> &str { - &self.worktree_dir - } - fn platform(&self) -> &str { - self.inner.platform() - } - fn os_version(&self) -> String { - self.inner.os_version() - } - fn is_remote(&self) -> bool { - self.inner.is_remote() - } - async fn ssh_access_command(&self) -> Result, String> { - self.inner.ssh_access_command().await - } - fn origin_url(&self) -> Option<&str> { - self.inner.origin_url() - } -} - /// Fans out execution to multiple branches concurrently. /// Each branch gets an isolated context clone and runs independently. pub struct ParallelHandler; @@ -374,79 +267,30 @@ impl Handler for ParallelHandler { crate::git::sanitize_ref_component(branch_key), ); - // Compute worktree path - let wt_path_str = if services.sandbox.is_remote() { - format!( - "{}/.fabro/runs/{}/parallel/{}/{}", - services.sandbox.working_directory(), - gs.run_id, - node.id, - branch_key - ) - } else { - run_dir - .join("parallel") - .join(&node.id) - .join(branch_key) - .join("worktree") - .to_string_lossy() - .to_string() - }; + // Compute worktree path (each sandbox type knows its own path scheme) + let wt_path_str = services + .sandbox + .parallel_worktree_path(run_dir, &gs.run_id, &node.id, branch_key); tracing::debug!(branch = %branch_name, path = %wt_path_str, "Creating worktree for parallel branch"); - // Create branch + worktree + reset via sandbox - if !crate::engine::git_create_branch_at(&*services.sandbox, &branch_name, bsha) + // Set up worktree via WorktreeSandbox + let wt_config = WorktreeConfig { + branch_name: branch_name.clone(), + base_sha: bsha.clone(), + worktree_path: wt_path_str.clone(), + skip_branch_creation: false, + }; + let mut wt_sandbox = WorktreeSandbox::new(Arc::clone(&services.sandbox), wt_config); + wt_sandbox.set_event_callback(Arc::clone(&services.emitter).worktree_callback()); + wt_sandbox + .initialize() .await - { - return Err(FabroError::handler(format!( - "failed to create branch {branch_name}" - ))); - } - services.emitter.emit(&WorkflowRunEvent::GitBranch { - branch: branch_name.clone(), - sha: bsha.clone(), - }); - if !crate::engine::git_replace_worktree( - &*services.sandbox, - &wt_path_str, - &branch_name, - ) - .await - { - return Err(FabroError::handler(format!( - "failed to add worktree {wt_path_str}" - ))); - } - services.emitter.emit(&WorkflowRunEvent::GitWorktreeAdd { - path: wt_path_str.clone(), - branch: branch_name.clone(), - }); - let reset_cmd = format!("{} reset --hard {bsha}", crate::engine::GIT_REMOTE); - let reset_result = services - .sandbox - .exec_command(&reset_cmd, 30_000, Some(&wt_path_str), None, None) - .await; - if !matches!(reset_result, Ok(ref r) if r.exit_code == 0) { - return Err(FabroError::handler(format!( - "failed to reset worktree {wt_path_str}" - ))); - } - services - .emitter - .emit(&WorkflowRunEvent::GitReset { sha: bsha.clone() }); + .map_err(|e| FabroError::handler(format!("worktree setup failed: {e}")))?; branch_context.set(keys::INTERNAL_WORK_DIR, serde_json::json!(&wt_path_str)); - // Create appropriate sandbox wrapper let wt_path = PathBuf::from(&wt_path_str); - let env: Arc = if services.sandbox.is_remote() { - Arc::new(WorktreeSandbox { - inner: Arc::clone(&services.sandbox), - worktree_dir: wt_path_str, - }) - } else { - Arc::new(LocalSandbox::new(wt_path.clone())) - }; + let env: Arc = Arc::new(wt_sandbox); (env, Some(wt_path)) } else { (Arc::clone(&services.sandbox), None) @@ -660,7 +504,7 @@ impl Handler for ParallelHandler { // Clean up worktrees first for result in &results { if let Some(ref wt_path) = result.worktree_path { - let wt_str = wt_path.to_string_lossy().to_string(); + let wt_str = wt_path.to_string_lossy().into_owned(); crate::engine::git_remove_worktree(&*services.sandbox, &wt_str).await; services .emitter diff --git a/lib/crates/fabro-workflows/src/sandbox_provider.rs b/lib/crates/fabro-workflows/src/sandbox_provider.rs index d8bc872f6..6205e2224 100644 --- a/lib/crates/fabro-workflows/src/sandbox_provider.rs +++ b/lib/crates/fabro-workflows/src/sandbox_provider.rs @@ -18,18 +18,7 @@ pub enum SandboxProvider { Ssh, } -impl SandboxProvider { - #[must_use] - pub fn is_remote(&self) -> bool { - match self { - Self::Daytona => true, - #[cfg(feature = "exedev")] - Self::Exe => true, - Self::Ssh => true, - _ => false, - } - } -} +impl SandboxProvider {} impl fmt::Display for SandboxProvider { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {