diff --git a/lib/crates/fabro-agent/src/lib.rs b/lib/crates/fabro-agent/src/lib.rs index 9f75a68e7..db293eda5 100644 --- a/lib/crates/fabro-agent/src/lib.rs +++ b/lib/crates/fabro-agent/src/lib.rs @@ -41,8 +41,9 @@ pub use profiles::{AnthropicProfile, EnvContext, GeminiProfile, OpenAiProfile}; pub use read_before_write_sandbox::ReadBeforeWriteSandbox; pub use sandbox::{ CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox, - SandboxEvent, SandboxEventCallback, WorktreeEvent, WorktreeEventCallback, WorktreeOptions, - WorktreeSandbox, format_lines_numbered, shell_quote, + SandboxEvent, SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle, + WorktreeEvent, WorktreeEventCallback, WorktreeOptions, WorktreeSandbox, format_lines_numbered, + shell_quote, }; pub use session::{ CompletionCoordinator, Session, SessionControlHandle, StaticEnvProvider, SteeringItem, diff --git a/lib/crates/fabro-agent/src/sandbox.rs b/lib/crates/fabro-agent/src/sandbox.rs index 0a7335364..60ad25581 100644 --- a/lib/crates/fabro-agent/src/sandbox.rs +++ b/lib/crates/fabro-agent/src/sandbox.rs @@ -3,6 +3,7 @@ // `crate::delegate_sandbox!` invocations continue to work. pub use fabro_sandbox::{ CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox, - SandboxEvent, SandboxEventCallback, WorktreeEvent, WorktreeEventCallback, WorktreeOptions, - WorktreeSandbox, delegate_sandbox, format_lines_numbered, shell_quote, + SandboxEvent, SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle, + WorktreeEvent, WorktreeEventCallback, WorktreeOptions, WorktreeSandbox, delegate_sandbox, + format_lines_numbered, shell_quote, }; diff --git a/lib/crates/fabro-sandbox/Cargo.toml b/lib/crates/fabro-sandbox/Cargo.toml index b7d3f51f5..0f6c3f874 100644 --- a/lib/crates/fabro-sandbox/Cargo.toml +++ b/lib/crates/fabro-sandbox/Cargo.toml @@ -24,7 +24,7 @@ anyhow.workspace = true async-trait.workspace = true thiserror.workspace = true tokio.workspace = true -tokio-util.workspace = true +tokio-util = { workspace = true, features = ["compat"] } serde.workspace = true serde_json.workspace = true strum.workspace = true diff --git a/lib/crates/fabro-sandbox/src/daytona/mod.rs b/lib/crates/fabro-sandbox/src/daytona/mod.rs index 572b3f422..646cae16c 100644 --- a/lib/crates/fabro-sandbox/src/daytona/mod.rs +++ b/lib/crates/fabro-sandbox/src/daytona/mod.rs @@ -29,7 +29,7 @@ use crate::redact::redact_auth_url; use crate::sandbox::{optional_timeout, resolve_path}; use crate::{ CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox, - SandboxEvent, SandboxEventCallback, format_lines_numbered, shell_quote, + SandboxEvent, SandboxEventCallback, StdioProcess, format_lines_numbered, shell_quote, }; const WORKING_DIRECTORY: &str = "/home/daytona/workspace"; @@ -1535,6 +1535,18 @@ impl Sandbox for DaytonaSandbox { }) } + async fn spawn_stdio_process( + &self, + _command: &str, + _working_dir: Option<&str>, + _env_vars: Option<&HashMap>, + _cancel_token: Option, + ) -> crate::Result { + Err(crate::Error::message( + "ACP backend requires bidirectional stdio; the Daytona sandbox provider does not support it yet", + )) + } + async fn grep( &self, pattern: &str, diff --git a/lib/crates/fabro-sandbox/src/docker.rs b/lib/crates/fabro-sandbox/src/docker.rs index b465a2eb6..df5898d92 100644 --- a/lib/crates/fabro-sandbox/src/docker.rs +++ b/lib/crates/fabro-sandbox/src/docker.rs @@ -12,13 +12,14 @@ use bollard::container::{ UploadToContainerOptions, }; use bollard::errors::Error as DockerError; -use bollard::exec::{CreateExecOptions, StartExecResults}; +use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults}; use bollard::image::CreateImageOptions; use bollard::models::HostConfig; use fabro_github::GitHubCredentials; use fabro_types::{CommandOutputStream, CommandTermination, RunId}; use fabro_util::time::elapsed_ms; use futures::StreamExt; +use tokio::io::AsyncWriteExt; use tokio::sync::OnceCell; use tokio::{fs, time}; use tokio_util::sync::CancellationToken; @@ -27,8 +28,9 @@ use crate::clone_source::{self, CloneDecision, EmptyWorkspaceReason}; use crate::redact::redact_auth_url; use crate::sandbox::{optional_timeout, resolve_path}; use crate::{ - CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox, - SandboxEvent, SandboxEventCallback, format_lines_numbered, shell_quote, + CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, + ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, + StderrCollector, StdioProcess, StdioProcessHandle, format_lines_numbered, shell_quote, }; const WORKING_DIRECTORY: &str = "/workspace"; @@ -451,20 +453,7 @@ impl DockerSandbox { } async fn request_docker_exec_stop(&self, stop_file: &str) -> crate::Result<()> { - let command = format!("touch {}", shell_quote(stop_file)); - let (stdout, stderr, exit_code) = self - .docker_exec( - vec!["/bin/bash".to_string(), "-lc".to_string(), command.clone()], - Some("/"), - None, - ) - .await?; - if exit_code != 0 { - return Err(crate::Error::message(format!( - "Failed to request Docker exec stop (exit {exit_code}): {stderr}{stdout}" - ))); - } - Ok(()) + request_docker_exec_stop_with(&self.docker, self.container_id()?, stop_file).await } async fn ensure_image(&self) -> crate::Result { @@ -804,6 +793,128 @@ exit \"$status\"\ ) } +fn docker_stdio_exec_options( + command: String, + working_dir: String, + env: Option>, +) -> (CreateExecOptions, StartExecOptions) { + ( + CreateExecOptions { + attach_stdin: Some(true), + attach_stdout: Some(true), + attach_stderr: Some(true), + tty: Some(false), + cmd: Some(vec!["/bin/bash".to_string(), "-lc".to_string(), command]), + working_dir: Some(working_dir), + env, + ..Default::default() + }, + StartExecOptions { + detach: false, + tty: false, + output_capacity: None, + }, + ) +} + +async fn request_docker_exec_stop_with( + docker: &Docker, + container_id: &str, + stop_file: &str, +) -> crate::Result<()> { + let command = format!("touch {}", shell_quote(stop_file)); + let exec_opts = CreateExecOptions { + cmd: Some(vec![ + "/bin/bash".to_string(), + "-lc".to_string(), + command, + ]), + attach_stdout: Some(true), + attach_stderr: Some(true), + working_dir: Some("/".to_string()), + ..Default::default() + }; + let exec_instance = docker + .create_exec(container_id, exec_opts) + .await + .map_err(|e| crate::Error::context("Failed to create Docker exec stop request", e))?; + let start_result = docker + .start_exec(&exec_instance.id, None) + .await + .map_err(|e| crate::Error::context("Failed to start Docker exec stop request", e))?; + + let mut stdout = String::new(); + let mut stderr = String::new(); + if let StartExecResults::Attached { mut output, .. } = start_result { + while let Some(chunk) = output.next().await { + match chunk { + Ok(LogOutput::StdOut { message }) => { + stdout.push_str(&String::from_utf8_lossy(&message)); + } + Ok(LogOutput::StdErr { message }) => { + stderr.push_str(&String::from_utf8_lossy(&message)); + } + Ok(_) => {} + Err(e) => return Err(crate::Error::context("Error reading stop request output", e)), + } + } + } + + let inspect = docker + .inspect_exec(&exec_instance.id) + .await + .map_err(|e| crate::Error::context("Failed to inspect Docker exec stop request", e))?; + let exit_code = inspect + .exit_code + .and_then(|code| i32::try_from(code).ok()) + .unwrap_or(-1); + if exit_code != 0 { + return Err(crate::Error::message(format!( + "Failed to request Docker exec stop (exit {exit_code}): {stderr}{stdout}" + ))); + } + Ok(()) +} + +struct DockerStdioProcessControl { + docker: Docker, + container_id: String, + exec_id: String, + stop_file: String, + termination: tokio::sync::Mutex>, +} + +#[async_trait] +impl crate::sandbox::StdioProcessControl for DockerStdioProcessControl { + async fn terminate(&self) -> crate::Result<()> { + if self.termination.lock().await.is_some() { + return Ok(()); + } + request_docker_exec_stop_with(&self.docker, &self.container_id, &self.stop_file).await?; + *self.termination.lock().await = Some(CommandTermination::Cancelled); + Ok(()) + } + + async fn wait(&self) -> crate::Result { + if let Some(termination) = *self.termination.lock().await { + return Ok(termination); + } + + loop { + let inspect = self + .docker + .inspect_exec(&self.exec_id) + .await + .map_err(|e| crate::Error::context("Failed to inspect Docker stdio exec", e))?; + if inspect.running != Some(true) { + *self.termination.lock().await = Some(CommandTermination::Exited); + return Ok(CommandTermination::Exited); + } + time::sleep(std::time::Duration::from_millis(50)).await; + } + } +} + fn git_clone_command(clone_url: &str, branch: Option<&str>) -> String { let mut command = "git -c maintenance.auto=0 -c gc.auto=0 clone".to_string(); if let Some(branch) = branch { @@ -1333,6 +1444,93 @@ impl Sandbox for DockerSandbox { .await } + async fn spawn_stdio_process( + &self, + command: &str, + working_dir: Option<&str>, + env_vars: Option<&HashMap>, + cancel_token: Option, + ) -> crate::Result { + let effective_dir = working_dir + .map(Self::resolve_container_path) + .unwrap_or_else(|| WORKING_DIRECTORY.to_string()); + let env: Option> = + env_vars.map(|vars| vars.iter().map(|(k, v)| format!("{k}={v}")).collect()); + let (stop_file, pid_file) = docker_exec_control_paths(); + let controlled_command = docker_controlled_shell_command(command, &stop_file, &pid_file); + let (create_opts, start_opts) = + docker_stdio_exec_options(controlled_command, effective_dir, env); + + let container_id = self.container_id()?.to_string(); + let exec_instance = self + .docker + .create_exec(&container_id, create_opts) + .await + .map_err(|e| crate::Error::context("Failed to create Docker stdio exec", e))?; + let exec_id = exec_instance.id.clone(); + let start_result = self + .docker + .start_exec(&exec_id, Some(start_opts)) + .await + .map_err(|e| crate::Error::context("Failed to start Docker stdio exec", e))?; + + let StartExecResults::Attached { mut output, input } = start_result else { + return Err(crate::Error::message( + "Docker stdio exec started detached unexpectedly", + )); + }; + + let stderr_collector = StderrCollector::new(DEFAULT_EXEC_OUTPUT_TAIL_BYTES); + let stderr_for_output = stderr_collector.clone(); + let (mut stdout_writer, stdout_reader) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + while let Some(chunk) = output.next().await { + match chunk { + Ok(LogOutput::StdOut { message }) => { + if let Err(err) = stdout_writer.write_all(&message).await { + tracing::warn!(error = %err, "Failed to forward Docker stdio stdout"); + return; + } + } + Ok(LogOutput::StdErr { message }) => { + stderr_for_output.push(&message).await; + } + Ok(_) => {} + Err(err) => { + let message = format!("Docker stdio output stream error: {err}"); + stderr_for_output.push(message.as_bytes()).await; + return; + } + } + } + }); + + let handle = StdioProcessHandle::new(DockerStdioProcessControl { + docker: self.docker.clone(), + container_id, + exec_id, + stop_file, + termination: tokio::sync::Mutex::new(None), + }); + + if let Some(token) = cancel_token { + let handle_for_cancel = handle.clone(); + tokio::spawn(async move { + token.cancelled().await; + if let Err(err) = handle_for_cancel.terminate().await { + tracing::warn!(error = %err, "Failed to terminate cancelled Docker stdio exec"); + } + }); + } + + Ok(StdioProcess { + stdin: input, + stdout: Box::pin(stdout_reader), + stderr: stderr_collector, + handle, + }) + } + async fn read_file( &self, path: &str, @@ -1744,6 +1942,33 @@ mod tests { ); } + #[test] + fn stdio_exec_options_attach_streams_without_tty() { + let (create, start) = docker_stdio_exec_options( + "python fake_agent.py".to_string(), + WORKING_DIRECTORY.to_string(), + Some(vec!["MODE=test".to_string()]), + ); + + assert_eq!(create.attach_stdin, Some(true)); + assert_eq!(create.attach_stdout, Some(true)); + assert_eq!(create.attach_stderr, Some(true)); + assert_eq!(create.tty, Some(false)); + assert_eq!(create.working_dir.as_deref(), Some(WORKING_DIRECTORY)); + assert_eq!(create.env, Some(vec!["MODE=test".to_string()])); + assert_eq!( + create.cmd, + Some(vec![ + "/bin/bash".to_string(), + "-lc".to_string(), + "python fake_agent.py".to_string() + ]) + ); + assert!(!start.detach); + assert!(!start.tty); + assert_eq!(start.output_capacity, None); + } + #[tokio::test] async fn controlled_shell_command_honors_stop_requested_before_pid_file_exists() { let tempdir = tempfile::tempdir().expect("tempdir should be created"); diff --git a/lib/crates/fabro-sandbox/src/lib.rs b/lib/crates/fabro-sandbox/src/lib.rs index a0902a46f..9d3a419e1 100644 --- a/lib/crates/fabro-sandbox/src/lib.rs +++ b/lib/crates/fabro-sandbox/src/lib.rs @@ -41,8 +41,9 @@ pub use reconnect::{reconnect, reconnect_for_run, reconnect_for_run_with_callbac pub use sandbox::{ CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, ExecStreamingResult, GitRunInfo, GitSetupIntent, GrepOptions, Sandbox, SandboxEvent, - SandboxEventCallback, format_lines_numbered, git_push_via_exec, redacted_output_tail, - setup_git_via_exec, shell_quote, + SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle, + format_lines_numbered, git_push_via_exec, redacted_output_tail, setup_git_via_exec, + shell_quote, }; pub use sandbox_spec::SandboxSpec; pub use terminal::{TerminalSession, TerminalSize, open_terminal_for_run}; diff --git a/lib/crates/fabro-sandbox/src/local.rs b/lib/crates/fabro-sandbox/src/local.rs index 9b5091a61..bba0c041d 100644 --- a/lib/crates/fabro-sandbox/src/local.rs +++ b/lib/crates/fabro-sandbox/src/local.rs @@ -13,8 +13,9 @@ use tokio_util::sync::CancellationToken; use crate::sandbox::optional_timeout; use crate::{ - CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox, - SandboxEvent, SandboxEventCallback, format_lines_numbered, + CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, + ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, + StderrCollector, StdioProcess, StdioProcessHandle, format_lines_numbered, }; pub struct LocalSandbox { @@ -141,6 +142,39 @@ where buf } +struct LocalStdioProcessControl { + child: tokio::sync::Mutex, + termination: tokio::sync::Mutex>, +} + +#[async_trait] +impl crate::sandbox::StdioProcessControl for LocalStdioProcessControl { + async fn terminate(&self) -> crate::Result<()> { + if self.termination.lock().await.is_some() { + return Ok(()); + } + + let mut child = self.child.lock().await; + sigterm_then_kill(&mut child).await; + *self.termination.lock().await = Some(CommandTermination::Cancelled); + Ok(()) + } + + async fn wait(&self) -> crate::Result { + if let Some(termination) = *self.termination.lock().await { + return Ok(termination); + } + + let mut child = self.child.lock().await; + child + .wait() + .await + .map_err(|e| crate::Error::context("Failed to wait for stdio process", e))?; + *self.termination.lock().await = Some(CommandTermination::Exited); + Ok(CommandTermination::Exited) + } +} + #[async_trait] impl Sandbox for LocalSandbox { async fn read_file( @@ -424,6 +458,85 @@ impl Sandbox for LocalSandbox { }) } + async fn spawn_stdio_process( + &self, + command: &str, + working_dir: Option<&str>, + env_vars: Option<&std::collections::HashMap>, + cancel_token: Option, + ) -> crate::Result { + let mut filtered_env: Vec<(String, String)> = process_env_vars() + .into_iter() + .filter(|(key, _)| !Self::should_filter_env_var(key)) + .collect(); + + if let Some(extra) = env_vars { + for (k, v) in extra { + if !Self::should_filter_env_var(k) { + filtered_env.push((k.clone(), v.clone())); + } + } + } + + let effective_dir = + working_dir.map_or_else(|| self.working_directory.clone(), std::path::PathBuf::from); + + let mut cmd = Command::new("/bin/bash"); + cmd.arg("-lc") + .arg(command) + .current_dir(&effective_dir) + .env_clear() + .envs(filtered_env) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()); + + #[cfg(unix)] + fabro_proc::pre_exec_setpgid(cmd.as_std_mut()); + + let mut child = cmd + .spawn() + .map_err(|e| crate::Error::context("Failed to spawn stdio process", e))?; + + let stdin = child + .stdin + .take() + .ok_or_else(|| crate::Error::message("Failed to open stdio process stdin"))?; + let stdout = child + .stdout + .take() + .ok_or_else(|| crate::Error::message("Failed to open stdio process stdout"))?; + let stderr = child + .stderr + .take() + .ok_or_else(|| crate::Error::message("Failed to open stdio process stderr"))?; + + let stderr_collector = StderrCollector::new(DEFAULT_EXEC_OUTPUT_TAIL_BYTES); + stderr_collector.spawn_reader(stderr); + + let handle = StdioProcessHandle::new(LocalStdioProcessControl { + child: tokio::sync::Mutex::new(child), + termination: tokio::sync::Mutex::new(None), + }); + + if let Some(token) = cancel_token { + let handle_for_cancel = handle.clone(); + tokio::spawn(async move { + token.cancelled().await; + if let Err(err) = handle_for_cancel.terminate().await { + tracing::warn!(error = %err, "Failed to terminate cancelled stdio process"); + } + }); + } + + Ok(StdioProcess { + stdin: Box::pin(stdin), + stdout: Box::pin(stdout), + stderr: stderr_collector, + handle, + }) + } + async fn grep( &self, pattern: &str, @@ -740,7 +853,7 @@ mod tests { use std::pin::Pin; use std::task::{Context as TaskContext, Poll}; - use tokio::io::ReadBuf; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadBuf}; use super::*; @@ -876,6 +989,34 @@ mod tests { std::fs::remove_dir_all(&dir).unwrap(); } + #[tokio::test] + async fn stdio_process_round_trips_lines() { + let dir = temp_dir(); + let sandbox = LocalSandbox::new(dir.clone()); + let process = sandbox + .spawn_stdio_process( + "python3 -u -c 'import sys; [print(line.strip()[::-1], flush=True) for line in sys.stdin]'", + None, + None, + None, + ) + .await + .unwrap(); + + let mut stdin = process.stdin; + let mut stdout = BufReader::new(process.stdout); + + stdin.write_all(b"abc\n").await.unwrap(); + stdin.flush().await.unwrap(); + + let mut line = String::new(); + stdout.read_line(&mut line).await.unwrap(); + assert_eq!(line.trim_end(), "cba"); + + process.handle.terminate().await.unwrap(); + std::fs::remove_dir_all(&dir).unwrap(); + } + #[tokio::test] async fn exec_command_exit_code() { let dir = temp_dir(); diff --git a/lib/crates/fabro-sandbox/src/read_guard.rs b/lib/crates/fabro-sandbox/src/read_guard.rs index 74b6d0c29..34669801c 100644 --- a/lib/crates/fabro-sandbox/src/read_guard.rs +++ b/lib/crates/fabro-sandbox/src/read_guard.rs @@ -277,4 +277,23 @@ mod tests { assert!(result.is_ok()); } + + #[tokio::test] + async fn stdio_process_forwards_to_inner_sandbox() { + let mock = Arc::new(MockSandbox::linux()); + let env = ReadBeforeWriteSandbox::new(mock.clone()); + + env.spawn_stdio_process("python fake_agent.py", Some("/work/sub"), None, None) + .await + .unwrap(); + + assert_eq!( + *mock.captured_command.lock().unwrap(), + Some("python fake_agent.py".to_string()) + ); + assert_eq!( + *mock.captured_working_dirs.lock().unwrap(), + vec![Some("/work/sub".to_string())] + ); + } } diff --git a/lib/crates/fabro-sandbox/src/sandbox.rs b/lib/crates/fabro-sandbox/src/sandbox.rs index ef7ae3b90..14b424b88 100644 --- a/lib/crates/fabro-sandbox/src/sandbox.rs +++ b/lib/crates/fabro-sandbox/src/sandbox.rs @@ -9,6 +9,7 @@ use std::time::Duration; use async_trait::async_trait; use fabro_types::{CommandOutputStream, CommandTermination}; use serde::{Deserialize, Serialize}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite}; use tokio::time; use tokio_util::sync::CancellationToken; @@ -121,6 +122,18 @@ macro_rules! delegate_sandbox { .await } + async fn spawn_stdio_process( + &self, + command: &str, + working_dir: Option<&str>, + env_vars: Option<&std::collections::HashMap>, + cancel_token: Option, + ) -> $crate::Result<$crate::StdioProcess> { + self.$field + .spawn_stdio_process(command, working_dir, env_vars, cancel_token) + .await + } + async fn glob(&self, pattern: &str, path: Option<&str>) -> $crate::Result> { self.$field.glob(pattern, path).await } @@ -666,6 +679,90 @@ pub type CommandOutputCallback = Arc< + Sync, >; +pub struct StdioProcess { + pub stdin: Pin>, + pub stdout: Pin>, + pub stderr: StderrCollector, + pub handle: StdioProcessHandle, +} + +#[derive(Debug, Clone)] +pub struct StderrCollector { + inner: Arc>>, + max_bytes: usize, +} + +impl StderrCollector { + #[must_use] + pub fn new(max_bytes: usize) -> Self { + Self { + inner: Arc::new(tokio::sync::Mutex::new(Vec::new())), + max_bytes, + } + } + + pub async fn push(&self, bytes: &[u8]) { + let mut tail = self.inner.lock().await; + tail.extend_from_slice(bytes); + if tail.len() > self.max_bytes { + let excess = tail.len() - self.max_bytes; + tail.drain(..excess); + } + } + + pub async fn tail_string(&self) -> String { + let tail = self.inner.lock().await; + String::from_utf8_lossy(&tail).into_owned() + } + + pub fn spawn_reader(&self, mut reader: R) -> tokio::task::JoinHandle<()> + where + R: AsyncRead + Unpin + Send + 'static, + { + let collector = self.clone(); + tokio::spawn(async move { + let mut buf = [0_u8; 8192]; + loop { + match reader.read(&mut buf).await { + Ok(0) => return, + Ok(read) => collector.push(&buf[..read]).await, + Err(err) => { + tracing::warn!(error = %err, "Failed to read stdio process stderr"); + return; + } + } + } + }) + } +} + +#[derive(Clone)] +pub struct StdioProcessHandle { + control: Arc, +} + +impl StdioProcessHandle { + pub(crate) fn new(control: impl StdioProcessControl + 'static) -> Self { + Self { + control: Arc::new(control), + } + } + + pub async fn terminate(&self) -> crate::Result<()> { + self.control.terminate().await + } + + pub async fn wait(&self) -> crate::Result { + self.control.wait().await + } +} + +#[async_trait] +pub(crate) trait StdioProcessControl: Send + Sync { + async fn terminate(&self) -> crate::Result<()>; + async fn wait(&self) -> crate::Result; +} + #[derive(Debug, Clone)] pub struct DirEntry { pub name: String, @@ -752,6 +849,19 @@ pub trait Sandbox: Send + Sync { live_streaming: false, }) } + + async fn spawn_stdio_process( + &self, + _command: &str, + _working_dir: Option<&str>, + _env_vars: Option<&HashMap>, + _cancel_token: Option, + ) -> crate::Result { + Err(crate::Error::message( + "ACP backend requires bidirectional stdio; this sandbox provider does not support it", + )) + } + async fn grep( &self, pattern: &str, diff --git a/lib/crates/fabro-sandbox/src/test_support.rs b/lib/crates/fabro-sandbox/src/test_support.rs index 0c7af3dd9..09df77fbe 100644 --- a/lib/crates/fabro-sandbox/src/test_support.rs +++ b/lib/crates/fabro-sandbox/src/test_support.rs @@ -6,7 +6,10 @@ use fabro_types::CommandTermination; use tokio::fs; use tokio_util::sync::CancellationToken; -use crate::{DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback}; +use crate::{ + DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, + SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle, +}; // --- MockSandbox --- @@ -104,6 +107,19 @@ impl Default for MockSandbox { } } +struct MockStdioProcessControl; + +#[async_trait] +impl crate::sandbox::StdioProcessControl for MockStdioProcessControl { + async fn terminate(&self) -> crate::Result<()> { + Ok(()) + } + + async fn wait(&self) -> crate::Result { + Ok(CommandTermination::Exited) + } +} + #[async_trait] impl Sandbox for MockSandbox { async fn read_file( @@ -184,6 +200,40 @@ impl Sandbox for MockSandbox { Ok(self.exec_result.clone()) } + async fn spawn_stdio_process( + &self, + command: &str, + working_dir: Option<&str>, + env_vars: Option<&std::collections::HashMap>, + _cancel_token: Option, + ) -> crate::Result { + *self + .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() + .expect("captured_env_vars lock poisoned") = env_vars.cloned(); + + let (stdin, _stdin_read) = tokio::io::duplex(1024); + let (_stdout_write, stdout) = tokio::io::duplex(1024); + Ok(StdioProcess { + stdin: Box::pin(stdin), + stdout: Box::pin(stdout), + stderr: StderrCollector::new(DEFAULT_EXEC_OUTPUT_TAIL_BYTES), + handle: StdioProcessHandle::new(MockStdioProcessControl), + }) + } + async fn grep( &self, _pattern: &str, diff --git a/lib/crates/fabro-sandbox/src/worktree.rs b/lib/crates/fabro-sandbox/src/worktree.rs index 87a31db31..39ef168c7 100644 --- a/lib/crates/fabro-sandbox/src/worktree.rs +++ b/lib/crates/fabro-sandbox/src/worktree.rs @@ -8,7 +8,7 @@ use tokio_util::sync::CancellationToken; use crate::sandbox::fetch_source_run_ref; use crate::{ CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GitRunInfo, GitSetupIntent, - GrepOptions, Sandbox, shell_quote, + GrepOptions, Sandbox, StdioProcess, shell_quote, }; /// Git command prefix that disables background maintenance. @@ -267,6 +267,19 @@ impl Sandbox for WorktreeSandbox { .await } + async fn spawn_stdio_process( + &self, + command: &str, + working_dir: Option<&str>, + env_vars: Option<&HashMap>, + cancel_token: Option, + ) -> crate::Result { + let wd = working_dir.unwrap_or(&self.config.worktree_path); + self.inner + .spawn_stdio_process(command, Some(wd), env_vars, cancel_token) + .await + } + // --- Delegated methods --- async fn read_file( @@ -676,6 +689,23 @@ mod tests { ); } + #[tokio::test] + async fn stdio_process_none_working_dir_defaults_to_worktree_path() { + let (inner, mock) = make_mock(); + let wt = WorktreeSandbox::new(inner, make_config("/tmp/wt")); + + wt.spawn_stdio_process("python fake_agent.py", 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" + ); + } + // ----------------------------------------------------------------------- // Accessors // -----------------------------------------------------------------------