feat: add sandbox stdio processes

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,

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,
 };

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

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<String, String>>,
+        _cancel_token: Option<CancellationToken>,
+    ) -> crate::Result<StdioProcess> {
+        Err(crate::Error::message(
+            "ACP backend requires bidirectional stdio; the Daytona sandbox provider does not support it yet",
+        ))
+    }
+
     async fn grep(
         &self,
         pattern: &str,

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<EnsureImageOutcome> {
@@ -804,6 +793,128 @@ exit \"$status\"\
     )
 }
 
+fn docker_stdio_exec_options(
+    command: String,
+    working_dir: String,
+    env: Option<Vec<String>>,
+) -> (CreateExecOptions<String>, 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<Option<CommandTermination>>,
+}
+
+#[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<CommandTermination> {
+        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<String, String>>,
+        cancel_token: Option<CancellationToken>,
+    ) -> crate::Result<StdioProcess> {
+        let effective_dir = working_dir
+            .map(Self::resolve_container_path)
+            .unwrap_or_else(|| WORKING_DIRECTORY.to_string());
+        let env: Option<Vec<String>> =
+            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");

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

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<Child>,
+    termination: tokio::sync::Mutex<Option<CommandTermination>>,
+}
+
+#[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<CommandTermination> {
+        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<String, String>>,
+        cancel_token: Option<CancellationToken>,
+    ) -> crate::Result<StdioProcess> {
+        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();

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())]
+        );
+    }
 }

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<String, String>>,
+                cancel_token: Option<tokio_util::sync::CancellationToken>,
+            ) -> $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<Vec<String>> {
                 self.$field.glob(pattern, path).await
             }
@@ -666,6 +679,90 @@ pub type CommandOutputCallback = Arc<
         + Sync,
 >;
 
+pub struct StdioProcess {
+    pub stdin:  Pin<Box<dyn AsyncWrite + Send>>,
+    pub stdout: Pin<Box<dyn AsyncRead + Send>>,
+    pub stderr: StderrCollector,
+    pub handle: StdioProcessHandle,
+}
+
+#[derive(Debug, Clone)]
+pub struct StderrCollector {
+    inner:     Arc<tokio::sync::Mutex<Vec<u8>>>,
+    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<R>(&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<dyn StdioProcessControl>,
+}
+
+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<CommandTermination> {
+        self.control.wait().await
+    }
+}
+
+#[async_trait]
+pub(crate) trait StdioProcessControl: Send + Sync {
+    async fn terminate(&self) -> crate::Result<()>;
+    async fn wait(&self) -> crate::Result<CommandTermination>;
+}
+
 #[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<String, String>>,
+        _cancel_token: Option<CancellationToken>,
+    ) -> crate::Result<StdioProcess> {
+        Err(crate::Error::message(
+            "ACP backend requires bidirectional stdio; this sandbox provider does not support it",
+        ))
+    }
+
     async fn grep(
         &self,
         pattern: &str,

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<CommandTermination> {
+        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<String, String>>,
+        _cancel_token: Option<CancellationToken>,
+    ) -> crate::Result<StdioProcess> {
+        *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,

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<String, String>>,
+        cancel_token: Option<CancellationToken>,
+    ) -> crate::Result<StdioProcess> {
+        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
     // -----------------------------------------------------------------------