From fc7ecde7562b822bde3fef79300be92fa500926a Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 9 Mar 2026 01:54:49 -0400 Subject: [PATCH] Add exe.dev cancellation, git cloning, and checkpointing support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Wire up cancel_token in ExeSandbox::exec_command via tokio::select! - Clone the local git repo into exe.dev VMs after initialization - Add GitCredentialSandbox decorator for push credential refresh - Enable GitCheckpointMode::Remote for exe.dev (rename daytona_* → remote_*) - Rename setup_daytona_git → setup_remote_git (already sandbox-agnostic) Co-Authored-By: Claude Opus 4.6 --- crates/arc-exe/src/lib.rs | 66 +++++- crates/arc-workflows/src/cli/run.rs | 214 +++++++++++++++--- .../src/git_credential_sandbox.rs | 163 +++++++++++++ crates/arc-workflows/src/lib.rs | 1 + 4 files changed, 414 insertions(+), 30 deletions(-) create mode 100644 crates/arc-workflows/src/git_credential_sandbox.rs diff --git a/crates/arc-exe/src/lib.rs b/crates/arc-exe/src/lib.rs index df3e26878..2ea9095f7 100644 --- a/crates/arc-exe/src/lib.rs +++ b/crates/arc-exe/src/lib.rs @@ -246,7 +246,7 @@ impl Sandbox for ExeSandbox { timeout_ms: u64, working_dir: Option<&str>, env_vars: Option<&HashMap>, - _cancel_token: Option, + cancel_token: Option, ) -> Result { let ssh = self.data_ssh()?; let start = Instant::now(); @@ -273,7 +273,20 @@ impl Sandbox for ExeSandbox { let full_cmd = parts.join(" "); let timeout = std::time::Duration::from_millis(timeout_ms); - let output = ssh.run_command_with_timeout(&full_cmd, timeout).await; + let token = cancel_token.unwrap_or_default(); + let output = tokio::select! { + res = ssh.run_command_with_timeout(&full_cmd, timeout) => res, + () = token.cancelled() => { + let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX); + return Ok(ExecResult { + stdout: String::new(), + stderr: "Command cancelled".to_string(), + exit_code: -1, + timed_out: true, + duration_ms, + }); + } + }; let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX); @@ -684,7 +697,7 @@ mod tests { } /// Helper: create an ExeSandbox with mock data SSH already initialized (skipping lifecycle). - fn sandbox_with_mock_data(data_ssh: MockSshRunner) -> ExeSandbox { + fn sandbox_with_mock_data(data_ssh: impl SshRunner + 'static) -> ExeSandbox { let mgmt = MockSshRunner::new(); let sandbox = ExeSandbox::new(Box::new(mgmt)); let _ = sandbox.vm_name.set("test-vm".to_string()); @@ -796,6 +809,53 @@ mod tests { assert_eq!(result.exit_code, -1); } + /// SSH runner that never completes — blocks forever on a Notify. + struct HangingSshRunner; + + #[async_trait] + impl SshRunner for HangingSshRunner { + async fn run_command(&self, _command: &str) -> Result { + std::future::pending().await + } + + async fn run_command_with_timeout( + &self, + _command: &str, + _timeout: std::time::Duration, + ) -> Result { + std::future::pending().await + } + + async fn upload_file(&self, _path: &str, _content: &[u8]) -> Result<(), String> { + Ok(()) + } + + async fn download_file(&self, _path: &str) -> Result, String> { + Ok(Vec::new()) + } + } + + #[tokio::test] + async fn exec_command_cancelled() { + let sandbox = sandbox_with_mock_data(HangingSshRunner); + + let token = CancellationToken::new(); + let token_clone = token.clone(); + + // Cancel immediately so the select! picks it up + token_clone.cancel(); + + let result = sandbox + .exec_command("sleep 999", 60_000, None, None, Some(token)) + .await + .unwrap(); + + assert!(result.timed_out); + assert_eq!(result.exit_code, -1); + assert_eq!(result.stderr, "Command cancelled"); + assert!(result.stdout.is_empty()); + } + // ---- Step 3: read_file ---- #[tokio::test] diff --git a/crates/arc-workflows/src/cli/run.rs b/crates/arc-workflows/src/cli/run.rs index 6cb1d1757..1285a477b 100644 --- a/crates/arc-workflows/src/cli/run.rs +++ b/crates/arc-workflows/src/cli/run.rs @@ -526,7 +526,7 @@ pub async fn run_command( env.set_event_callback(Arc::new(move |event| { emitter_cb.emit(&crate::event::WorkflowRunEvent::Sandbox { event }); })); - Arc::new(env) + Arc::new(env) as Arc } SandboxProvider::Local => { let mut env = LocalSandbox::new(cwd); @@ -538,15 +538,33 @@ pub async fn run_command( } }; - // Wrap with ReadBeforeWriteSandbox to enforce read-before-write guard - let sandbox: Arc = Arc::new(arc_agent::ReadBeforeWriteSandbox::new(sandbox)); - // Initialize sandbox (creates sandbox/container once for the whole run) sandbox .initialize() .await .map_err(|e| anyhow::anyhow!("Failed to initialize sandbox: {e}"))?; + // Clone repo into exe.dev VM after initialization + let exe_origin_url = if sandbox_provider == SandboxProvider::Exe { + clone_repo_into_exe(&*sandbox, &emitter, &original_cwd, &github_app).await? + } else { + None + }; + + // Wrap exe.dev sandbox with GitCredentialSandbox for push credential refresh + let sandbox: Arc = if sandbox_provider == SandboxProvider::Exe { + Arc::new(crate::git_credential_sandbox::GitCredentialSandbox::new( + sandbox, + exe_origin_url, + github_app.clone(), + )) + } else { + sandbox + }; + + // Wrap with ReadBeforeWriteSandbox to enforce read-before-write guard + let sandbox: Arc = Arc::new(arc_agent::ReadBeforeWriteSandbox::new(sandbox)); + // Safety net: if we panic or return early, best-effort cleanup via spawn. let sandbox_for_cleanup = Arc::clone(&sandbox); let cleanup_guard = scopeguard::guard((), move |()| { @@ -561,22 +579,24 @@ pub async fn run_command( } }); - // Set up git inside Daytona sandbox (if applicable) - let (daytona_base_sha, daytona_branch, daytona_base_branch) = - if sandbox_provider == SandboxProvider::Daytona { - match setup_daytona_git(&*sandbox, &run_id).await { - Ok((base, branch, base_br)) => (Some(base), Some(branch), base_br), - Err(e) => { - eprintln!( - "{} Daytona git setup failed ({e}), running without git checkpoints.", - styles.yellow.apply_to("Warning:"), - ); - (None, None, None) - } + // Set up git inside remote sandbox (Daytona or exe.dev) for checkpoint commits + let (remote_base_sha, remote_branch, remote_base_branch) = if matches!( + sandbox_provider, + SandboxProvider::Daytona | SandboxProvider::Exe + ) { + match setup_remote_git(&*sandbox, &run_id).await { + Ok((base, branch, base_br)) => (Some(base), Some(branch), base_br), + Err(e) => { + eprintln!( + "{} Remote git setup failed ({e}), running without git checkpoints.", + styles.yellow.apply_to("Warning:"), + ); + (None, None, None) } - } else { - (None, None, None) - }; + } + } else { + (None, None, None) + }; // Create SSH access if requested if args.ssh { @@ -730,7 +750,7 @@ pub async fn run_command( // 7. Execute // Set up metadata branch for git checkpointing (host or remote) - let meta_branch = if worktree_work_dir.is_some() || daytona_base_sha.is_some() { + let meta_branch = if worktree_work_dir.is_some() || remote_base_sha.is_some() { Some(crate::git::MetadataStore::branch_name(&run_id)) } else { None @@ -748,13 +768,12 @@ pub async fn run_command( SandboxProvider::Local | SandboxProvider::Docker => { worktree_work_dir.map(GitCheckpointMode::Host) } - SandboxProvider::Daytona => daytona_base_sha + SandboxProvider::Daytona | SandboxProvider::Exe => remote_base_sha .as_ref() .map(|_| GitCheckpointMode::Remote(original_cwd.clone())), - SandboxProvider::Exe => None, }, - base_sha: worktree_base_sha.or(daytona_base_sha), - run_branch: worktree_branch.or(daytona_branch), + base_sha: worktree_base_sha.or(remote_base_sha), + run_branch: worktree_branch.or(remote_branch), meta_branch, labels: args .label @@ -765,7 +784,7 @@ pub async fn run_command( checkpoint_exclude_globs, github_app: github_app.clone(), git_author, - base_branch: detected_base_branch.or(daytona_base_branch), + base_branch: detected_base_branch.or(remote_base_branch), pull_request_enabled: run_cfg .as_ref() .and_then(|c| c.pull_request.as_ref()) @@ -1003,9 +1022,150 @@ fn setup_worktree( Ok((worktree_path.clone(), worktree_path, branch_name, base_sha)) } -/// Set up git inside a Daytona sandbox for checkpoint commits. +/// Clone the local git repo into an exe.dev VM. +/// Returns the HTTPS origin URL on success, or None if no git repo was detected. +async fn clone_repo_into_exe( + sandbox: &dyn arc_agent::Sandbox, + emitter: &std::sync::Arc, + cwd: &std::path::Path, + github_app: &Option, +) -> anyhow::Result> { + use arc_agent::sandbox::SandboxEvent; + + let (detected_url, branch) = match crate::daytona_sandbox::detect_repo_info(cwd) { + Ok(info) => info, + Err(_) => return Ok(None), // no git repo — continue without cloning + }; + + let url = crate::github_app::ssh_url_to_https(&detected_url); + emitter.emit(&crate::event::WorkflowRunEvent::Sandbox { + event: SandboxEvent::GitCloneStarted { + url: url.clone(), + branch: branch.clone(), + }, + }); + let clone_start = std::time::Instant::now(); + + // Resolve clone credentials via GitHub App or fall back to no auth + let token = match github_app { + Some(creds) => { + let (owner, repo) = crate::github_app::parse_github_owner_repo(&url).map_err(|e| { + emitter.emit(&crate::event::WorkflowRunEvent::Sandbox { + event: SandboxEvent::GitCloneFailed { + url: url.clone(), + error: e.clone(), + }, + }); + anyhow::anyhow!("Failed to parse GitHub URL for clone: {e}") + })?; + let (_username, password) = + crate::github_app::resolve_clone_credentials(creds, &owner, &repo) + .await + .map_err(|e| { + emitter.emit(&crate::event::WorkflowRunEvent::Sandbox { + event: SandboxEvent::GitCloneFailed { + url: url.clone(), + error: e.clone(), + }, + }); + anyhow::anyhow!("Failed to get GitHub App credentials for clone: {e}") + })?; + password + } + None => None, + }; + + let auth_url = match &token { + Some(t) => url.replacen("https://", &format!("https://x-access-token:{t}@"), 1), + None => url.clone(), + }; + + let working_dir = sandbox.working_directory(); + let branch_flag = branch + .as_deref() + .map(|b| format!(" --branch '{}'", b.replace('\'', "'\\''"))) + .unwrap_or_default(); + + // Try cloning directly into the working directory + let clone_cmd = format!( + "git clone{branch_flag} '{}' '{working_dir}'", + auth_url.replace('\'', "'\\''"), + ); + let clone_result = sandbox + .exec_command(&clone_cmd, 300_000, Some("/tmp"), None, None) + .await + .map_err(|e| anyhow::anyhow!("git clone failed: {e}"))?; + + if clone_result.exit_code != 0 { + // Fall back to init + fetch + checkout if directory is not empty + if clone_result.stderr.contains("not an empty directory") + || clone_result + .stderr + .contains("already exists and is not an empty") + { + let init_cmds = [ + "git init", + &format!( + "git remote add origin '{}'", + auth_url.replace('\'', "'\\''") + ), + "git fetch origin", + &format!("git checkout {}", branch.as_deref().unwrap_or("main")), + ]; + for cmd in &init_cmds { + let r = sandbox + .exec_command(cmd, 300_000, None, None, None) + .await + .map_err(|e| anyhow::anyhow!("git fallback command failed: {e}"))?; + if r.exit_code != 0 { + let err = format!("git fallback failed on '{cmd}': {}", r.stderr); + emitter.emit(&crate::event::WorkflowRunEvent::Sandbox { + event: SandboxEvent::GitCloneFailed { + url: url.clone(), + error: err.clone(), + }, + }); + anyhow::bail!("{err}"); + } + } + } else { + let err = format!( + "git clone failed (exit {}): {}", + clone_result.exit_code, clone_result.stderr + ); + emitter.emit(&crate::event::WorkflowRunEvent::Sandbox { + event: SandboxEvent::GitCloneFailed { + url: url.clone(), + error: err.clone(), + }, + }); + anyhow::bail!("{err}"); + } + } + + // Set clean push credentials on the remote + let set_url_cmd = format!( + "git remote set-url origin '{}'", + auth_url.replace('\'', "'\\''"), + ); + let _ = sandbox + .exec_command(&set_url_cmd, 10_000, None, None, None) + .await; + + let duration_ms = u64::try_from(clone_start.elapsed().as_millis()).unwrap_or(u64::MAX); + emitter.emit(&crate::event::WorkflowRunEvent::Sandbox { + event: SandboxEvent::GitCloneCompleted { + url: url.clone(), + duration_ms, + }, + }); + + Ok(Some(url)) +} + +/// Set up git inside a remote sandbox (Daytona or exe.dev) for checkpoint commits. /// Returns (base_sha, branch_name, base_branch) on success. -async fn setup_daytona_git( +async fn setup_remote_git( sandbox: &dyn arc_agent::Sandbox, run_id: &str, ) -> anyhow::Result<(String, String, Option)> { diff --git a/crates/arc-workflows/src/git_credential_sandbox.rs b/crates/arc-workflows/src/git_credential_sandbox.rs new file mode 100644 index 000000000..78eb261df --- /dev/null +++ b/crates/arc-workflows/src/git_credential_sandbox.rs @@ -0,0 +1,163 @@ +use std::collections::HashMap; +use std::sync::Arc; + +use arc_agent::sandbox::{DirEntry, ExecResult, GrepOptions, Sandbox}; +use async_trait::async_trait; +use tokio_util::sync::CancellationToken; + +use crate::github_app::GitHubAppCredentials; + +/// Sandbox decorator that overrides `refresh_push_credentials` to rotate +/// GitHub App tokens on the git remote URL. +/// +/// This lets `ExeSandbox` (in `arc-exe`) get credential refresh behaviour +/// without depending on `github_app` (which lives in `arc-workflows`). +pub struct GitCredentialSandbox { + inner: Arc, + origin_url: Option, + github_app: Option, +} + +impl GitCredentialSandbox { + pub fn new( + inner: Arc, + origin_url: Option, + github_app: Option, + ) -> Self { + Self { + inner, + origin_url, + github_app, + } + } +} + +#[async_trait] +impl Sandbox for GitCredentialSandbox { + async fn initialize(&self) -> Result<(), String> { + self.inner.initialize().await + } + + async fn cleanup(&self) -> Result<(), String> { + self.inner.cleanup().await + } + + async fn exec_command( + &self, + command: &str, + timeout_ms: u64, + working_dir: Option<&str>, + env_vars: Option<&HashMap>, + cancel_token: Option, + ) -> Result { + self.inner + .exec_command(command, timeout_ms, working_dir, env_vars, cancel_token) + .await + } + + 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 grep( + &self, + pattern: &str, + path: &str, + options: &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 + } + + fn working_directory(&self) -> &str { + self.inner.working_directory() + } + + 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> { + let origin_url = match &self.origin_url { + Some(url) => url, + None => return Ok(()), + }; + let creds = match &self.github_app { + Some(c) => c, + None => return Ok(()), + }; + + let (owner, repo) = crate::github_app::parse_github_owner_repo(origin_url) + .map_err(|e| format!("Failed to parse origin URL for credential refresh: {e}"))?; + + let (_username, password) = + crate::github_app::resolve_clone_credentials(creds, &owner, &repo) + .await + .map_err(|e| format!("Failed to refresh GitHub App token: {e}"))?; + + if let Some(token) = password { + let auth_url = + origin_url.replacen("https://", &format!("https://x-access-token:{token}@"), 1); + let cmd = format!( + "git -c maintenance.auto=0 remote set-url origin '{}'", + auth_url.replace('\'', "'\\''"), + ); + self.inner + .exec_command(&cmd, 10_000, None, None, None) + .await + .map_err(|e| format!("Failed to set refreshed push credentials: {e}"))?; + } + + Ok(()) + } + + async fn set_autostop_interval(&self, minutes: i32) -> Result<(), String> { + self.inner.set_autostop_interval(minutes).await + } +} diff --git a/crates/arc-workflows/src/lib.rs b/crates/arc-workflows/src/lib.rs index 7151697cd..c62751175 100644 --- a/crates/arc-workflows/src/lib.rs +++ b/crates/arc-workflows/src/lib.rs @@ -37,6 +37,7 @@ pub mod engine; pub mod error; pub mod event; pub mod git; +pub(crate) mod git_credential_sandbox; pub mod github_app; pub mod graph; pub mod handler;