feat(pi-shell): enabled PTY shell runs in pi-shell via PTY runner

- Added PTY flags to ShellRunOptions and ShellExecuteOptions and defaulted pty to false when unset.
- Routed PTY-enabled shell runs to a dedicated PTY runner and executed commands via `sh -lc` in a PTY.
- Handled PTY output with a dedicated thread, Chunk/Done events, child PGID cancellation, and drain-timeout cleanup.
- Added Unix-only PTY assertions and non-Windows native coverage for `executeShell` with `pty: true` and `/dev/` output checks.
- Added `portable-pty = "0.9"` (and lockfile update) to enable PTY execution in pi-shell.
This commit is contained in:
can1357
2026-05-12 00:54:55 +02:00
parent 7e05a2fef4
commit 530942c0b2
8 changed files with 491 additions and 27 deletions
Generated
+1
View File
@@ -2496,6 +2496,7 @@ dependencies = [
"clap",
"libc",
"os_pipe",
"portable-pty",
"regex",
"serde",
"serde_json",
+34 -2
View File
@@ -84,6 +84,8 @@ pub struct ShellRunOptions<'env> {
pub cwd: Option<String>,
/// Environment variables to apply for this command only.
pub env: Option<HashMap<String, String>>,
/// Run the command attached to a PTY.
pub pty: Option<bool>,
/// Timeout in milliseconds before cancelling the command.
pub timeout_ms: Option<u32>,
/// Abort signal for cancelling the operation.
@@ -101,6 +103,8 @@ pub struct ShellExecuteOptions<'env> {
pub env: Option<HashMap<String, String>>,
/// Environment variables to apply once per session.
pub session_env: Option<HashMap<String, String>>,
/// Run the command attached to a PTY.
pub pty: Option<bool>,
/// Timeout in milliseconds before cancelling the command.
pub timeout_ms: Option<u32>,
/// 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::<String>();
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);
+1
View File
@@ -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"
+423 -23
View File
@@ -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<BufferedOutput>,
}
enum PtyReaderEvent {
Chunk(String),
Done,
}
struct CaptureState {
text: String,
exceeded: bool,
max_bytes: usize,
}
pub struct Shell {
session: Arc<TokioMutex<Option<ShellSessionCore>>>,
abort_state: ShellAbortState,
@@ -492,11 +525,6 @@ async fn run_shell_command(
on_chunk: Option<mpsc::UnboundedSender<String>>,
cancel_token: CancellationToken,
) -> Result<(ExecutionResult, Option<MinimizerResult>)> {
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<mpsc::UnboundedSender<String>>,
cancel_token: CancellationToken,
minimizer_mode: minimizer::engine::MinimizerMode,
max_capture_bytes: usize,
) -> Result<(ExecutionResult, Option<MinimizerResult>)> {
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<mpsc::UnboundedSender<String>>,
cancel_token: CancellationToken,
) -> Result<PtyShellOutput> {
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::<PtyReaderEvent>();
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<i32> = 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<i32> = 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<MinimizerResult> {
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<dyn Child + Send + Sync>,
child_pid: Option<i32>,
process_group_id: Option<i32>,
) {
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<PtyReaderEvent>,
on_chunk: Option<&mpsc::UnboundedSender<String>>,
capture: &mut Option<CaptureState>,
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<String>>,
capture: &mut Option<CaptureState>,
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<dyn Read + Send>, reader_tx: std_mpsc::Sender<PtyReaderEvent>) {
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<i32>) {
let mut targets = process::TerminationTargets::new();
for job in &shell.jobs().jobs {
@@ -38,9 +38,9 @@ Assumptions you didn't validate: incidents to debug.
</communication>
<critical>
- 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.
</critical>
+4
View File
@@ -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
+4
View File
@@ -1150,6 +1150,8 @@ export interface ShellExecuteOptions {
env?: Record<string, string>
/** Environment variables to apply once per session. */
sessionEnv?: Record<string, string>
/** 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<string, string>
/** Run the command attached to a PTY. */
pty?: boolean
/** Timeout in milliseconds before cancelling the command. */
timeoutMs?: number
/** Abort signal for cancelling the operation. */
+22
View File
@@ -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;