diff --git a/Cargo.lock b/Cargo.lock index c987b3a88..2d12412f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2496,6 +2496,7 @@ dependencies = [ "clap", "libc", "os_pipe", + "portable-pty", "regex", "serde", "serde_json", diff --git a/crates/pi-natives/src/shell.rs b/crates/pi-natives/src/shell.rs index a81f75ce7..576d69bef 100644 --- a/crates/pi-natives/src/shell.rs +++ b/crates/pi-natives/src/shell.rs @@ -84,6 +84,8 @@ pub struct ShellRunOptions<'env> { pub cwd: Option, /// Environment variables to apply for this command only. pub env: Option>, + /// Run the command attached to a PTY. + pub pty: Option, /// Timeout in milliseconds before cancelling the command. pub timeout_ms: Option, /// Abort signal for cancelling the operation. @@ -101,6 +103,8 @@ pub struct ShellExecuteOptions<'env> { pub env: Option>, /// Environment variables to apply once per session. pub session_env: Option>, + /// Run the command attached to a PTY. + pub pty: Option, /// Timeout in milliseconds before cancelling the command. pub timeout_ms: Option, /// Optional snapshot file to source on session creation. @@ -209,7 +213,7 @@ impl Shell { command: options.command, cwd: options.cwd, env: options.env, - pty: false, + pty: options.pty.unwrap_or(false), timeout_ms: options.timeout_ms, }; task::future(env, "shell.run", async move { @@ -253,7 +257,7 @@ pub fn execute_shell<'env>( timeout_ms: options.timeout_ms, snapshot_path: options.snapshot_path, minimizer: options.minimizer.map(Into::into), - pty: false, + pty: options.pty.unwrap_or(false), }; task::future(env, "shell.execute", async move { let chunk_tx = bridge_chunks(on_chunk); @@ -369,6 +373,34 @@ mod tests { assert_eq!(child_sid, child_pid); } + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread")] + async fn pty_run_attaches_stdio_to_terminal() { + let shell = CoreShell::new(None); + let (tx, mut rx) = mpsc::unbounded_channel::(); + let result = shell + .run( + CoreShellRunOptions { + command: "test -t 0 && test -t 1 && tty".to_string(), + cwd: None, + env: None, + pty: true, + timeout_ms: Some(5_000), + }, + Some(tx), + CancelToken::default(), + ) + .await + .expect("shell run"); + let mut output = String::new(); + while let Ok(chunk) = rx.try_recv() { + output.push_str(&chunk); + } + + assert_eq!(result.exit_code, Some(0), "output: {output:?}"); + assert!(output.contains("/dev/"), "tty output should include a terminal path: {output:?}"); + } + #[tokio::test] async fn read_output_stops_when_cancelled_before_pipe_eof() { let shell = CoreShell::new(None); diff --git a/crates/pi-shell/Cargo.toml b/crates/pi-shell/Cargo.toml index 053033f17..c86ad6aac 100644 --- a/crates/pi-shell/Cargo.toml +++ b/crates/pi-shell/Cargo.toml @@ -18,6 +18,7 @@ brush-builtins = { version = "0.2.0", path = "../brush-builtins-vendored" } brush-parser = "0.3" clap = { version = "4", features = ["derive"] } os_pipe = "1" +portable-pty = "0.9" serde = { version = "1.0", features = ["derive"] } serde_json = { version = "1.0", features = ["preserve_order"] } toml = "1.1" diff --git a/crates/pi-shell/src/shell.rs b/crates/pi-shell/src/shell.rs index 6ec446058..b5d906723 100644 --- a/crates/pi-shell/src/shell.rs +++ b/crates/pi-shell/src/shell.rs @@ -4,9 +4,9 @@ use std::{ collections::{HashMap, HashSet}, fs, future::Future, - io::{self, Write}, + io::{self, Read, Write}, str, - sync::Arc, + sync::{Arc, mpsc as std_mpsc}, time::Duration, }; @@ -20,6 +20,7 @@ use brush_core::{ openfiles::{self, OpenFile, OpenFiles}, }; use clap::Parser; +use portable_pty::{Child, CommandBuilder, PtySize, native_pty_system}; #[cfg(not(unix))] use tokio::io::AsyncReadExt as _; use tokio::{ @@ -121,6 +122,38 @@ pub struct ShellExecuteOptions { pub type ShellExecuteResult = ShellRunResult; +const PTY_DEFAULT_COLS: u16 = 120; +const PTY_DEFAULT_ROWS: u16 = 40; +const PTY_LOOP_INTERVAL: Duration = Duration::from_millis(16); +const PTY_POST_CANCEL_DRAIN_TIMEOUT: Duration = Duration::from_millis(300); +const PTY_POST_EXIT_DRAIN_TIMEOUT: Duration = Duration::from_millis(300); +const PTY_FINAL_READER_DRAIN_TIMEOUT: Duration = Duration::from_millis(50); +const PTY_READER_EVENTS_PER_TICK: usize = 256; + +struct PtyShellConfig { + command: String, + cwd: String, + env: Vec<(String, String)>, + capture_output: bool, + max_capture_bytes: usize, +} + +struct PtyShellOutput { + exit_code: i32, + output: Option, +} + +enum PtyReaderEvent { + Chunk(String), + Done, +} + +struct CaptureState { + text: String, + exceeded: bool, + max_bytes: usize, +} + pub struct Shell { session: Arc>>, abort_state: ShellAbortState, @@ -492,11 +525,6 @@ async fn run_shell_command( on_chunk: Option>, cancel_token: CancellationToken, ) -> Result<(ExecutionResult, Option)> { - if options.pty { - // TODO: The shared brush path is currently pipe-based. PTY execution - // still falls back to the non-PTY brush path until a native PTY-backed - // brush integration lands. - } if let Some(cwd) = options.cwd.as_deref() { session .shell @@ -504,22 +532,6 @@ async fn run_shell_command( .map_err(|err| Error::msg(format!("Failed to set cwd: {err}")))?; } - let (reader_file, writer_file) = pipe_to_files("output")?; - - let stdout_file = OpenFile::from( - writer_file - .try_clone() - .map_err(|err| Error::msg(format!("Failed to clone pipe: {err}")))?, - ); - let stderr_file = OpenFile::from(writer_file); - - let mut params = session.shell.default_exec_params(); - params.set_fd(OpenFiles::STDIN_FD, null_file()?); - params.set_fd(OpenFiles::STDOUT_FD, stdout_file); - params.set_fd(OpenFiles::STDERR_FD, stderr_file); - params.process_group_policy = ProcessGroupPolicy::NewProcessGroup; - params.set_cancel_token(cancel_token.clone()); - let mut env_scope_pushed = false; if let Some(env) = options.env.as_ref() { session @@ -557,6 +569,42 @@ async fn run_shell_command( } else { 0 }; + + if options.pty { + let result = run_pty_shell_command( + session, + options, + on_chunk, + cancel_token, + minimizer_mode, + max_capture_bytes, + ) + .await; + if env_scope_pushed { + session + .shell + .env_mut() + .pop_scope(EnvironmentScope::Command) + .map_err(|err| Error::msg(format!("Failed to pop env scope: {err}")))?; + } + return result; + } + + let (reader_file, writer_file) = pipe_to_files("output")?; + + let stdout_file = OpenFile::from( + writer_file + .try_clone() + .map_err(|err| Error::msg(format!("Failed to clone pipe: {err}")))?, + ); + let stderr_file = OpenFile::from(writer_file); + + let mut params = session.shell.default_exec_params(); + params.set_fd(OpenFiles::STDIN_FD, null_file()?); + params.set_fd(OpenFiles::STDOUT_FD, stdout_file); + params.set_fd(OpenFiles::STDERR_FD, stderr_file); + params.process_group_policy = ProcessGroupPolicy::NewProcessGroup; + params.set_cancel_token(cancel_token.clone()); let baseline_descendants = process::current_descendant_pids(); let tracked_targets = Arc::new(TokioMutex::new(process::TerminationTargets::new())); let tracking_cancel = CancellationToken::new(); @@ -729,6 +777,358 @@ async fn run_shell_command( Ok((result, minimized_out)) } +async fn run_pty_shell_command( + session: &ShellSessionCore, + options: &ShellRunConfig, + on_chunk: Option>, + cancel_token: CancellationToken, + minimizer_mode: minimizer::engine::MinimizerMode, + max_capture_bytes: usize, +) -> Result<(ExecutionResult, Option)> { + let pty_config = PtyShellConfig { + command: options.command.clone(), + cwd: session.shell.working_dir().to_string_lossy().into_owned(), + env: exported_env(&session.shell), + capture_output: !matches!(minimizer_mode, minimizer::engine::MinimizerMode::None), + max_capture_bytes, + }; + + let pty_output = tokio::task::spawn_blocking(move || { + run_pty_shell_command_sync(pty_config, on_chunk, cancel_token) + }) + .await + .map_err(|err| Error::msg(format!("PTY execution task failed: {err}")))??; + + let result = execution_result_from_exit_code(pty_output.exit_code); + let minimized_out = pty_output.output.and_then(|output| { + minimize_buffered_output( + &options.command, + output, + pty_output.exit_code, + options.minimizer.as_ref(), + minimizer_mode, + ) + }); + Ok((result, minimized_out)) +} + +fn run_pty_shell_command_sync( + config: PtyShellConfig, + on_chunk: Option>, + cancel_token: CancellationToken, +) -> Result { + let pty_system = native_pty_system(); + let pair = pty_system + .openpty(PtySize { + rows: PTY_DEFAULT_ROWS, + cols: PTY_DEFAULT_COLS, + pixel_width: 0, + pixel_height: 0, + }) + .map_err(|err| Error::msg(format!("Failed to open PTY: {err}")))?; + + let mut cmd = CommandBuilder::new("sh"); + cmd.arg("-lc"); + cmd.arg(&config.command); + cmd.cwd(config.cwd.as_str()); + cmd.env_clear(); + for (key, value) in &config.env { + cmd.env(key, value); + } + + let mut child = pair + .slave + .spawn_command(cmd) + .map_err(|err| Error::msg(format!("Failed to spawn PTY command: {err}")))?; + drop(pair.slave); + + let master = pair.master; + let mut reader = master + .try_clone_reader() + .map_err(|err| Error::msg(format!("Failed to create PTY reader: {err}")))?; + let (reader_tx, reader_rx) = std_mpsc::channel::(); + let reader_thread = std::thread::spawn(move || read_pty_output(&mut reader, reader_tx)); + + let child_pid = child + .process_id() + .and_then(|value| i32::try_from(value).ok()); + #[cfg(unix)] + let process_group_id = master.process_group_leader().filter(|pgid| *pgid > 0); + #[cfg(not(unix))] + let process_group_id: Option = None; + + let mut capture = config.capture_output.then(|| CaptureState { + text: String::new(), + exceeded: false, + max_bytes: config.max_capture_bytes, + }); + let mut reader_done = false; + let mut exit_code: Option = None; + let mut terminate_requested = false; + let mut reader_drain_deadline = None; + + while exit_code.is_none() || !reader_done { + if !terminate_requested && cancel_token.is_cancelled() { + terminate_pty_processes(&mut child, child_pid, process_group_id); + terminate_requested = true; + reader_drain_deadline = Some(std::time::Instant::now() + PTY_POST_CANCEL_DRAIN_TIMEOUT); + } + + drain_pty_reader_events(&reader_rx, on_chunk.as_ref(), &mut capture, &mut reader_done); + + if exit_code.is_none() + && let Some(status) = child + .try_wait() + .map_err(|err| Error::msg(format!("Failed checking PTY status: {err}")))? + { + exit_code = Some(i32::try_from(status.exit_code()).unwrap_or(i32::MAX)); + if !reader_done && reader_drain_deadline.is_none() { + reader_drain_deadline = Some(std::time::Instant::now() + PTY_POST_EXIT_DRAIN_TIMEOUT); + } + } + + if let Some(deadline) = reader_drain_deadline + && std::time::Instant::now() >= deadline + { + break; + } + + if exit_code.is_none() || !reader_done { + let wait_duration = reader_drain_deadline.map_or(PTY_LOOP_INTERVAL, |deadline| { + deadline + .saturating_duration_since(std::time::Instant::now()) + .min(PTY_LOOP_INTERVAL) + }); + match reader_rx.recv_timeout(wait_duration) { + Ok(event) => { + handle_pty_reader_event(event, on_chunk.as_ref(), &mut capture, &mut reader_done) + }, + Err(std_mpsc::RecvTimeoutError::Timeout) => {}, + Err(std_mpsc::RecvTimeoutError::Disconnected) => reader_done = true, + } + } + } + + if exit_code.is_none() { + if terminate_requested { + if let Some(status) = child + .try_wait() + .map_err(|err| Error::msg(format!("Failed checking PTY status: {err}")))? + { + exit_code = Some(i32::try_from(status.exit_code()).unwrap_or(i32::MAX)); + } + } else { + let status = child + .wait() + .map_err(|err| Error::msg(format!("Failed waiting PTY process: {err}")))?; + exit_code = Some(i32::try_from(status.exit_code()).unwrap_or(i32::MAX)); + } + } + + drop(master); + if !reader_done { + let finalize_deadline = std::time::Instant::now() + PTY_FINAL_READER_DRAIN_TIMEOUT; + while std::time::Instant::now() < finalize_deadline { + let wait_duration = finalize_deadline + .saturating_duration_since(std::time::Instant::now()) + .min(Duration::from_millis(5)); + match reader_rx.recv_timeout(wait_duration) { + Ok(event) => { + handle_pty_reader_event(event, on_chunk.as_ref(), &mut capture, &mut reader_done) + }, + Err(std_mpsc::RecvTimeoutError::Timeout) => {}, + Err(std_mpsc::RecvTimeoutError::Disconnected) => { + reader_done = true; + break; + }, + } + } + } + if reader_done { + let _ = reader_thread.join(); + } + + Ok(PtyShellOutput { + exit_code: exit_code.unwrap_or(1), + output: capture.map(CaptureState::finish), + }) +} + +fn exported_env(shell: &BrushShell) -> Vec<(String, String)> { + shell + .env() + .iter_exported() + .filter_map(|(key, value)| { + value + .value() + .try_get_cow_str(shell) + .map(|value| (key.clone(), value.into_owned())) + }) + .collect() +} + +fn execution_result_from_exit_code(exit_code: i32) -> ExecutionResult { + let code = u8::try_from(exit_code).unwrap_or(if exit_code < 0 { 1 } else { u8::MAX }); + ExecutionResult::new(code) +} + +fn minimize_buffered_output( + command: &str, + output: BufferedOutput, + exit_code: i32, + config: Option<&minimizer::MinimizerConfig>, + minimizer_mode: minimizer::engine::MinimizerMode, +) -> Option { + let config = config.filter(|_| !output.exceeded)?; + let minimized = match minimizer_mode { + minimizer::engine::MinimizerMode::WholeCommand => { + minimizer::apply(command, &output.text, exit_code, config) + }, + minimizer::engine::MinimizerMode::None => { + minimizer::MinimizerOutput::passthrough(&output.text) + }, + }; + if !minimized.changed { + return None; + } + let original = minimized.original_text?; + let output_bytes = u32::try_from(minimized.text.len()).unwrap_or(u32::MAX); + Some(MinimizerResult { + filter: minimized.filter.to_string(), + text: minimized.text, + original_text: original, + input_bytes: u32::try_from(minimized.input_bytes).unwrap_or(u32::MAX), + output_bytes, + }) +} + +fn terminate_pty_processes( + child: &mut Box, + child_pid: Option, + process_group_id: Option, +) { + let mut targets = process::TerminationTargets::new(); + if let Some(pgid) = process_group_id { + targets.add_pgid(pgid); + } + if let Some(pid) = child_pid { + targets.add_pid(pid); + } + targets.signal(process::TERM_SIGNAL); + let _ = child.kill(); + targets.signal(process::KILL_SIGNAL); +} + +fn drain_pty_reader_events( + reader_rx: &std_mpsc::Receiver, + on_chunk: Option<&mpsc::UnboundedSender>, + capture: &mut Option, + reader_done: &mut bool, +) { + for _ in 0..PTY_READER_EVENTS_PER_TICK { + match reader_rx.try_recv() { + Ok(event) => handle_pty_reader_event(event, on_chunk, capture, reader_done), + Err(std_mpsc::TryRecvError::Empty) => break, + Err(std_mpsc::TryRecvError::Disconnected) => { + *reader_done = true; + break; + }, + } + } +} + +fn handle_pty_reader_event( + event: PtyReaderEvent, + on_chunk: Option<&mpsc::UnboundedSender>, + capture: &mut Option, + reader_done: &mut bool, +) { + match event { + PtyReaderEvent::Chunk(chunk) => { + if let Some(capture) = capture.as_mut() { + capture.push(&chunk); + } + if let Some(on_chunk) = on_chunk { + let _ = on_chunk.send(chunk); + } + }, + PtyReaderEvent::Done => *reader_done = true, + } +} + +fn read_pty_output(reader: &mut Box, reader_tx: std_mpsc::Sender) { + const REPLACEMENT: &str = "\u{FFFD}"; + const BUF: usize = 65536; + let mut buf = vec![0_u8; BUF + 4]; + let mut it = 0; + loop { + match reader.read(&mut buf[it..BUF]) { + Ok(0) => break, + Ok(n) => { + it += n; + while it > 0 { + let pending = &buf[..it]; + match str::from_utf8(pending) { + Ok(text) => { + let _ = reader_tx.send(PtyReaderEvent::Chunk(text.to_string())); + it = 0; + break; + }, + Err(err) => { + let valid_up_to = err.valid_up_to(); + if valid_up_to > 0 { + if let Ok(text) = str::from_utf8(&pending[..valid_up_to]) { + let _ = reader_tx.send(PtyReaderEvent::Chunk(text.to_string())); + } + buf.copy_within(valid_up_to..it, 0); + it -= valid_up_to; + } + match err.error_len() { + Some(invalid_len) => { + let _ = reader_tx.send(PtyReaderEvent::Chunk(REPLACEMENT.to_string())); + let drop_len = invalid_len.min(it); + buf.copy_within(drop_len..it, 0); + it -= drop_len; + }, + None => break, + } + }, + } + } + }, + Err(_) => break, + } + } + for chunk in buf[..it].utf8_chunks() { + let valid = chunk.valid(); + if !valid.is_empty() { + let _ = reader_tx.send(PtyReaderEvent::Chunk(valid.to_string())); + } + if !chunk.invalid().is_empty() { + let _ = reader_tx.send(PtyReaderEvent::Chunk(REPLACEMENT.to_string())); + } + } + let _ = reader_tx.send(PtyReaderEvent::Done); +} + +impl CaptureState { + fn push(&mut self, chunk: &str) { + if self.exceeded { + return; + } + if self.text.len().saturating_add(chunk.len()) > self.max_bytes { + self.exceeded = true; + self.text.clear(); + return; + } + self.text.push_str(chunk); + } + + fn finish(self) -> BufferedOutput { + BufferedOutput { text: self.text, exceeded: self.exceeded } + } +} + fn terminate_background_jobs(shell: &BrushShell, baseline_descendants: &HashSet) { let mut targets = process::TerminationTargets::new(); for job in &shell.jobs().jobs { diff --git a/packages/coding-agent/src/prompts/system/system-prompt.md b/packages/coding-agent/src/prompts/system/system-prompt.md index 998bd258e..f68d208ed 100644 --- a/packages/coding-agent/src/prompts/system/system-prompt.md +++ b/packages/coding-agent/src/prompts/system/system-prompt.md @@ -38,9 +38,9 @@ Assumptions you didn't validate: incidents to debug. -- You **MUST NOT** narrate about or even consider, session limits, token/tool budgets, effort estimates, or how much of the task you think you can finish. These are not your concern: +- You **MUST NOT** narrate about or even consider, session limits, token/tool budgets, effort estimates, or how much of the task you think you can finish. These are not your concern: - Even if it was true, start, as if it was not. It's the only way to make progress. - - Execute the work or delegate it. + - Execute the work or delegate it. - You **MUST NOT** speculate about scope inflation ("this is actually a multi-week effort"). You have no comprehension of time, so stop pretending. diff --git a/packages/natives/CHANGELOG.md b/packages/natives/CHANGELOG.md index 9f40f13c6..96c6d659c 100644 --- a/packages/natives/CHANGELOG.md +++ b/packages/natives/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed shell `pty` execution so `executeShell` and persistent `Shell.run` attach command stdio to a real PTY when requested. + ## [14.9.3] - 2026-05-10 ### Added diff --git a/packages/natives/native/index.d.ts b/packages/natives/native/index.d.ts index 74ac86dc8..32e56d684 100644 --- a/packages/natives/native/index.d.ts +++ b/packages/natives/native/index.d.ts @@ -1150,6 +1150,8 @@ export interface ShellExecuteOptions { env?: Record /** Environment variables to apply once per session. */ sessionEnv?: Record + /** Run the command attached to a PTY. */ + pty?: boolean /** Timeout in milliseconds before cancelling the command. */ timeoutMs?: number /** Optional snapshot file to source on session creation. */ @@ -1178,6 +1180,8 @@ export interface ShellRunOptions { cwd?: string /** Environment variables to apply for this command only. */ env?: Record + /** Run the command attached to a PTY. */ + pty?: boolean /** Timeout in milliseconds before cancelling the command. */ timeoutMs?: number /** Abort signal for cancelling the operation. */ diff --git a/packages/natives/test/native.test.ts b/packages/natives/test/native.test.ts index b67bf3325..cf114d53a 100644 --- a/packages/natives/test/native.test.ts +++ b/packages/natives/test/native.test.ts @@ -513,6 +513,28 @@ describe("pi-natives", () => { }); describe("shell", () => { + it("runs pty commands with terminal stdio", async () => { + if (process.platform === "win32") { + return; + } + + let output = ""; + const result = await executeShell( + { + command: "test -t 0 && test -t 1 && tty", + cwd: testDir, + pty: true, + timeoutMs: 5_000, + }, + (_err, chunk) => { + output += chunk; + }, + ); + + expect(result.exitCode).toBe(0); + expect(output).toContain("/dev/"); + }); + it("should time out background workloads without leaving delayed writers behind", async () => { if (process.platform === "win32") { return;