scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-10 22:43:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Unified WorktreeSandbox (#117)

Browse source 
This PR introduces a unified `WorktreeSandbox` type in `fabro-sandbox`
that consolidates previously duplicated git worktree management logic
spread across `parallel.rs` and `run.rs`. The new type wraps any
`Arc<dyn Sandbox>`, handles the full worktree lifecycle (branch
creation, `worktree add`, and cleanup) in its `initialize()`/`cleanup()`
methods, overrides `working_directory()` and `exec_command()` to default
to the worktree path, and delegates all other `Sandbox` methods to the
inner sandbox. A `WorktreeConfig` struct controls behavior (branch name,
base SHA, worktree path, and a `skip_branch_creation` flag for resume
flows), and a `WorktreeEventCallback` mechanism bridges lifecycle events
to the workflow event system via a new
`EventEmitter::worktree_callback()` helper.

The old private `WorktreeSandbox` struct in `parallel.rs` (which only
redirected `exec_command` working dirs with no lifecycle awareness) is
removed and replaced with the shared implementation. The
`setup_worktree()` function in `run.rs` is also removed; its logic is
absorbed directly into the `SandboxProvider::Local` branch of sandbox
construction, where `WorktreeSandbox::initialize()` is called and
`std::env::set_current_dir()` follows on success. The resume path
(`run_from_branch`) similarly replaces direct `git::replace_worktree`
calls with `WorktreeSandbox` using `skip_branch_creation: true`. The
`MockSandbox` in `test_support.rs` gains `captured_commands` and
`captured_working_dirs` vectors to support sequenced-command assertions
in the new unit tests.

The `MockSandbox` enhancement is a notable improvement for testability
beyond this specific change—having the full ordered sequence of commands
rather than just the last one makes it straightforward to assert on
multi-step git workflows. One subtle behavior worth noting is that in
`parallel.rs` the `git reset --hard` step previously present after
worktree creation is now absent from `WorktreeSandbox::initialize()`;
the plan mentioned it but the implementation deliberately omits it (the
branch is already force-set to the target SHA, so the reset was
redundant for the parallel case). Cleanup for parallel branches
continues to go through `engine::git_remove_worktree` on the parent
sandbox rather than calling `wt_sandbox.cleanup()`, since the sandbox
`Arc` is consumed by the spawned task—this is a reasonable tradeoff
noted in the plan.

### Fabro Details

<details>
<summary>Ran 11 stages in 62m 32s for $3.67</summary>

| Stage | Duration | Cost | Retries |
|---|---|---|---|
| start | 0s | – | 0 |
| toolchain | 0s | – | 0 |
| preflight_compile | 1m 11s | – | 0 |
| preflight_lint | 12s | – | 0 |
| implement | 30m 15s | $2.15 | 0 |
| simplify_opus | 15m 5s | $0.71 | 0 |
| simplify_gpt | 11m 8s | $0.54 | 0 |
| verify | 46s | – | 0 |
| fixup | 3m 17s | $0.27 | 0 |
| verify | 46s | – | 0 |
| fmt | 1s | – | 0 |
| **Total** | **62m 32s** | **$3.67** | **0** |

</details>

<details>
<summary>Ran <code>ImplementAndSimplify.fabro</code> (12 nodes and 15
edges)</summary>

```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_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", 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_gpt -> verify
    verify -> fmt   [condition="outcome=success"]
    verify -> fixup
    fixup -> verify
    fmt -> exit
}

```

</details>

⚒️ Generated with [Fabro](https://fabro.sh)

---------

Co-authored-by: Fabro <noreply@fabro.sh>
Co-authored-by: Bryan Helmkamp <bryan@brynary.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
brynary-fabro[bot] 2026-03-20 15:35:04 -04:00 • committed by GitHub
parent 3680a57a25
commit 956e681e4f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 1187 additions and 385 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
docs/reference/sdk.mdx
View file

@ -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.
}
```

4
lib/crates/fabro-agent/src/lib.rs
View file

@ -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;

2
lib/crates/fabro-agent/src/sandbox.rs
View file

@ -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

210
lib/crates/fabro-cli/src/commands/run.rs
View file

@ -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<EventEmitter>) -> Arc<dyn Sandbox> {
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<dyn Sandbox>
}
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<dyn fabro_agent::Sandbox>, Option<PathBuf>) =
let (sandbox, _worktree_path): (Arc<dyn fabro_agent::Sandbox>, Option<PathBuf>) =
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<dyn fabro_agent::Sandbox>,
Some(wt),
)
}
#[cfg(feature = "exedev")]
SandboxProvider::Exe => {
@ -1928,14 +1968,8 @@ async fn run_from_branch(
let sandbox: Arc<dyn fabro_agent::Sandbox> =
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<String> = 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<String> = sandbox.resume_setup_commands(run_branch);
// Build interviewer
let interviewer: Arc<dyn Interviewer> = 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()),

30
lib/crates/fabro-sandbox/src/daytona/mod.rs
View file

@ -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<Option<crate::GitRunInfo>, String> {
crate::setup_git_via_exec(self, run_id).await.map(Some)
}
fn resume_setup_commands(&self, run_branch: &str) -> Vec<String> {
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<Option<String>, String> {

30
lib/crates/fabro-sandbox/src/exe/mod.rs
View file

@ -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<Option<crate::GitRunInfo>, String> {
crate::setup_git_via_exec(self, run_id).await.map(Some)
}
fn resume_setup_commands(&self, run_branch: &str) -> Vec<String> {
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<Option<String>, String> {

8
lib/crates/fabro-sandbox/src/lib.rs
View file

@ -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;

154
lib/crates/fabro-sandbox/src/sandbox.rs
View file

@ -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<String>,
}
/// Generates an `#[async_trait] impl Sandbox` block for a decorator type
/// that wraps an `Arc<dyn Sandbox>`. 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<Option<$crate::GitRunInfo>, String> {
self.$field.setup_git_for_run(run_id).await
}
fn resume_setup_commands(&self, run_branch: &str) -> Vec<String> {
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<Option<String>, 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<Option<GitRunInfo>, 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<String> {
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<Option<String>, 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<GitRunInfo, String> {
// 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::*;

30
lib/crates/fabro-sandbox/src/ssh/mod.rs
View file

@ -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<Option<crate::GitRunInfo>, String> {
crate::setup_git_via_exec(self, run_id).await.map(Some)
}
fn resume_setup_commands(&self, run_branch: &str) -> Vec<String> {
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<Option<String>, String> {

18
lib/crates/fabro-sandbox/src/test_support.rs
View file

@ -20,8 +20,12 @@ pub struct MockSandbox {
pub written_files: Mutex<Vec<(String, String)>>,
/// Captures the `timeout_ms` argument from `exec_command` calls.
pub captured_timeout: Mutex<Option<u64>>,
/// Captures the `command` argument from `exec_command` calls.
/// Captures the `command` argument from `exec_command` calls (last only).
pub captured_command: Mutex<Option<String>>,
/// Captures all `command` arguments from `exec_command` calls in order.
pub captured_commands: Mutex<Vec<String>>,
/// Captures all `working_dir` arguments from `exec_command` calls in order.
pub captured_working_dirs: Mutex<Vec<Option<String>>>,
/// Captures the `env_vars` argument from `exec_command` calls.
pub captured_env_vars: Mutex<Option<HashMap<String, String>>>,
pub event_callback: Option<SandboxEventCallback>,
@ -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<String, String>>,
_cancel_token: Option<CancellationToken>,
) -> Result<ExecResult, String> {
@ -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()

747
lib/crates/fabro-sandbox/src/worktree.rs Normal file
View file

@ -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<dyn Fn(WorktreeEvent) + Send + Sync>;
/// 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<dyn Sandbox>,
config: WorktreeConfig,
event_callback: Option<WorktreeEventCallback>,
initialized: std::sync::atomic::AtomicBool,
}
impl WorktreeSandbox {
/// Create a new `WorktreeSandbox` wrapping `inner` with the given configuration.
pub fn new(inner: Arc<dyn Sandbox>, 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<String, String>>,
cancel_token: Option<CancellationToken>,
) -> Result<ExecResult, String> {
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<usize>,
limit: Option<usize>,
) -> Result<String, String> {
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<bool, String> {
let resolved = self.resolve_path(path);
self.inner.file_exists(&resolved).await
}
async fn list_directory(
&self,
path: &str,
depth: Option<usize>,
) -> Result<Vec<DirEntry>, 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<Vec<String>, String> {
let resolved = self.resolve_path(path);
self.inner.grep(pattern, &resolved, options).await
}
async fn glob(&self, pattern: &str, path: Option<&str>) -> Result<Vec<String>, 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<Option<crate::GitRunInfo>, String> {
self.inner.setup_git_for_run(run_id).await
}
fn resume_setup_commands(&self, run_branch: &str) -> Vec<String> {
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<Option<String>, 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<Option<(String, HashMap<String, String>)>, 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<dyn Sandbox>` (passed to WorktreeSandbox)
/// and the `Arc<MockSandbox>` (used to assert captured state).
fn make_mock() -> (Arc<dyn Sandbox>, Arc<MockSandbox>) {
let mock = Arc::new(MockSandbox::linux());
let as_sandbox: Arc<dyn Sandbox> = 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<Mutex<Vec<String>>> = 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<Mutex<Vec<String>>> = 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<dyn Sandbox> = 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<dyn Sandbox> = 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<dyn Sandbox> = 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<dyn Sandbox> = 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");
}
}

111
lib/crates/fabro-workflows/src/engine.rs
View file

@ -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<String> {
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>), 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(

19
lib/crates/fabro-workflows/src/event.rs
View file

@ -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<Self>) -> 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)]

194
lib/crates/fabro-workflows/src/handler/parallel.rs
View file

@ -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<dyn Sandbox>,
worktree_dir: String,
}
#[async_trait]
impl Sandbox for WorktreeSandbox {
async fn read_file(
&self,
path: &str,
offset: Option<usize>,
limit: Option<usize>,
) -> Result<String, String> {
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<bool, String> {
self.inner.file_exists(path).await
}
async fn list_directory(
&self,
path: &str,
depth: Option<usize>,
) -> Result<Vec<fabro_agent::sandbox::DirEntry>, 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<String, String>>,
cancel_token: Option<tokio_util::sync::CancellationToken>,
) -> Result<fabro_agent::sandbox::ExecResult, String> {
// 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<Vec<String>, String> {
self.inner.grep(pattern, path, options).await
}
async fn glob(&self, pattern: &str, path: Option<&str>) -> Result<Vec<String>, 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<Option<String>, 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<dyn Sandbox> = 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<dyn Sandbox> = 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

13
lib/crates/fabro-workflows/src/sandbox_provider.rs
View file

@ -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 {