Skip to content

fix(cli): start sandbox exec without waiting for piped stdin EOF #4006

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
johntmyers merged 1 commit into from
Oct 1, 2026
Merged

fix(cli): start sandbox exec without waiting for piped stdin EOF #4006

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
307 changes: 258 additions & 49 deletions crates/openshell-cli/src/run.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1852,6 +1852,86 @@ fn local_terminal_size() -> Option<(u32, u32)> {
const MAX_EXEC_REQUEST_BYTES: usize = 1024 * 1024;
const MAX_EXEC_STDIN_BYTES: usize = 4 * 1024 * 1024;

/// How long `sandbox exec` waits for piped stdin to reach EOF before it starts
/// the command and streams the rest of the input as it arrives.
///
/// Small pipes such as `echo x | openshell sandbox exec ...` close well within
/// this window and keep using the single-request path that older gateways
/// need. A pipe that stays open (a CI runner, a supervisor, a harness that
/// never closes stdin) must not block the command: after the grace period the
/// command starts and stdin is forwarded until EOF or until the command exits.
const EXEC_STDIN_UNARY_GRACE: Duration = Duration::from_millis(200);

/// Piped stdin as collected by [`collect_piped_stdin`].
enum PipedStdin {
/// stdin reached EOF within the grace period; `prefix` holds all of it.
Complete(Vec<u8>),
/// stdin is still open. `prefix` is what arrived so far; `rest` delivers
/// the remaining chunks until EOF.
Open {
prefix: Vec<u8>,
rest: tokio::sync::mpsc::Receiver<std::io::Result<Vec<u8>>>,
},
}

/// Read a pipe on a detached OS thread and hand its bytes over in chunks.
///
/// A plain `std::thread` rather than `spawn_blocking`, so runtime shutdown
/// never waits on a thread parked in `read(2)`. The thread exits at EOF, on a
/// read error, or when the receiver is dropped.
fn spawn_piped_stdin_reader(
mut reader: impl Read + Send + 'static,
) -> tokio::sync::mpsc::Receiver<std::io::Result<Vec<u8>>> {
let (tx, rx) = tokio::sync::mpsc::channel::<std::io::Result<Vec<u8>>>(64);
std::thread::spawn(move || {
let mut buf = [0u8; 4096];
loop {
match reader.read(&mut buf) {
Ok(0) => return,
Err(error) if error.kind() == ErrorKind::Interrupted => {}
Err(error) => {
let _ = tx.blocking_send(Err(error));
return;
}
Ok(n) => {
if tx.blocking_send(Ok(buf[..n].to_vec())).is_err() {
return;
}
}
}
}
});
rx
}

/// Collect piped stdin until EOF or until `grace` elapses, whichever comes
/// first. Input beyond `limit` bytes is rejected with the upload hint.
async fn collect_piped_stdin(
mut rx: tokio::sync::mpsc::Receiver<std::io::Result<Vec<u8>>>,
grace: Duration,
limit: usize,
) -> Result<PipedStdin> {
let deadline = tokio::time::Instant::now() + grace;
let mut prefix = Vec::new();
loop {
match tokio::time::timeout_at(deadline, rx.recv()).await {
Ok(Some(Ok(chunk))) => {
prefix.extend_from_slice(&chunk);
if prefix.len() > limit {
return Err(piped_stdin_limit_error());
}
}
Ok(Some(Err(error))) => return Err(error).into_diagnostic(),
Ok(None) => return Ok(PipedStdin::Complete(prefix)),
Err(_elapsed) => return Ok(PipedStdin::Open { prefix, rest: rx }),
}
}
}

fn piped_stdin_limit_error() -> miette::Report {
miette::miette!("piped stdin exceeds the 4 MiB limit; use `sandbox upload` for larger input")
}

/// Execute a command in a running sandbox via gRPC, streaming output to the terminal.
///
/// Returns the remote command's exit code, or an error if the event stream
Expand Down Expand Up @@ -1902,24 +1982,24 @@ pub async fn sandbox_exec_grpc(
// interactive RPC closes the SSH channel when stdin reaches EOF. Retain
// the existing 4 MiB input cap because the supervisor's process stdin
// queue is unbounded; larger input should use file upload instead.
let stdin_prefix = if stdin_is_terminal {
Vec::new()
//
// Never block on stdin EOF before starting the command: a pipe that stays
// open (CI runners, harnesses) would otherwise hang the exec forever
// without the gateway ever seeing the request. After a short grace period
// the command starts and the remaining input streams until EOF.
let (stdin_prefix, stdin_rest) = if stdin_is_terminal {
(Vec::new(), None)
} else {
tokio::task::spawn_blocking(|| {
let mut prefix = Vec::new();
std::io::stdin()
.take((MAX_EXEC_STDIN_BYTES + 1) as u64)
.read_to_end(&mut prefix)
.into_diagnostic()?;
if prefix.len() > MAX_EXEC_STDIN_BYTES {
return Err(miette::miette!(
"piped stdin exceeds the 4 MiB limit; use `sandbox upload` for larger input"
));
}
Ok::<_, miette::Report>(prefix)
})
.await
.into_diagnostic()??
match collect_piped_stdin(
spawn_piped_stdin_reader(std::io::stdin()),
EXEC_STDIN_UNARY_GRACE,
MAX_EXEC_STDIN_BYTES,
)
.await?
{
PipedStdin::Complete(prefix) => (prefix, None),
PipedStdin::Open { prefix, rest } => (prefix, Some(rest)),
}
};

let (cols, rows) = if tty {
Expand Down Expand Up @@ -1952,7 +2032,10 @@ pub async fn sandbox_exec_grpc(
"exec command or environment exceeds the gateway's 1 MiB message limit"
));
}
if (tty && stdin_is_terminal) || request.encoded_len() > MAX_EXEC_REQUEST_BYTES {
if (tty && stdin_is_terminal)
|| stdin_rest.is_some()
|| request.encoded_len() > MAX_EXEC_REQUEST_BYTES
{
return sandbox_exec_streaming_grpc(
client,
&sandbox,
Expand All @@ -1964,6 +2047,7 @@ pub async fn sandbox_exec_grpc(
tty,
stdin_is_terminal,
std::mem::take(&mut request.stdin),
stdin_rest,
)
.await;
}
Expand Down Expand Up @@ -2350,6 +2434,7 @@ async fn sandbox_exec_streaming_grpc(
tty: bool,
stdin_is_terminal: bool,
stdin_prefix: Vec<u8>,
stdin_rest: Option<tokio::sync::mpsc::Receiver<std::io::Result<Vec<u8>>>>,
) -> Result<i32> {
#[cfg(unix)]
use openshell_core::proto::ExecSandboxWindowResize;
Expand Down Expand Up @@ -2407,42 +2492,88 @@ async fn sandbox_exec_streaming_grpc(
// closes (blocking_send returns Err) or stdin hits EOF.
let stdin_tx = input_tx.clone();
let (stdin_result_tx, mut stdin_result_rx) = tokio::sync::oneshot::channel();
std::thread::spawn(move || {
let mut stdin = std::io::stdin().lock();
let mut buf = [0u8; 4096];
let result = (|| {
for chunk in stdin_prefix.chunks(buf.len()) {
if stdin_tx
.blocking_send(ExecSandboxInput {
payload: Some(exec_sandbox_input::Payload::Stdin(chunk.to_vec())),
})
.is_err()
{
return Ok(());
if let Some(mut rest) = stdin_rest {
// Piped stdin that was still open when the command started: the
// reader thread from `spawn_piped_stdin_reader` already owns stdin,
// so forward its chunks here after the collected prefix. The 4 MiB
// cap covers the prefix and the streamed remainder together.
tokio::spawn(async move {
let mut forwarded = 0usize;
let result = async {
for chunk in stdin_prefix.chunks(4096) {
forwarded += chunk.len();
if stdin_tx
.send(ExecSandboxInput {
payload: Some(exec_sandbox_input::Payload::Stdin(chunk.to_vec())),
})
.await
.is_err()
{
return Ok(());
}
}
while let Some(chunk) = rest.recv().await {
let chunk = chunk?;
forwarded += chunk.len();
if forwarded > MAX_EXEC_STDIN_BYTES {
return Err(std::io::Error::new(
ErrorKind::InvalidInput,
piped_stdin_limit_error().to_string(),
));
}
if stdin_tx
.send(ExecSandboxInput {
payload: Some(exec_sandbox_input::Payload::Stdin(chunk)),
})
.await
.is_err()
{
return Ok(());
}
}
Ok(())
}
loop {
match stdin.read(&mut buf) {
Ok(0) => return Ok(()),
Err(error) if error.kind() == ErrorKind::Interrupted => {}
Err(error) => return Err(error),
Ok(n) => {
if stdin_tx
.blocking_send(ExecSandboxInput {
payload: Some(exec_sandbox_input::Payload::Stdin(
buf[..n].to_vec(),
)),
})
.is_err()
{
return Ok(());
.await;
let _ = stdin_result_tx.send(result);
});
} else {
std::thread::spawn(move || {
let mut stdin = std::io::stdin().lock();
let mut buf = [0u8; 4096];
let result = (|| {
for chunk in stdin_prefix.chunks(buf.len()) {
if stdin_tx
.blocking_send(ExecSandboxInput {
payload: Some(exec_sandbox_input::Payload::Stdin(chunk.to_vec())),
})
.is_err()
{
return Ok(());
}
}
loop {
match stdin.read(&mut buf) {
Ok(0) => return Ok(()),
Err(error) if error.kind() == ErrorKind::Interrupted => {}
Err(error) => return Err(error),
Ok(n) => {
if stdin_tx
.blocking_send(ExecSandboxInput {
payload: Some(exec_sandbox_input::Payload::Stdin(
buf[..n].to_vec(),
)),
})
.is_err()
{
return Ok(());
}
}
}
}
}
})();
let _ = stdin_result_tx.send(result);
});
})();
let _ = stdin_result_tx.send(result);
});
}

// SIGWINCH handler: forward terminal resize events.
#[cfg(unix)]
Expand Down Expand Up @@ -8206,4 +8337,82 @@ mod tests {
let log = log_line("OCSF", "ocsf", message, "sandbox", &[]);
assert!(format_log_line(&log).ends_with(message));
}

use std::io::Write as _;
use std::time::{Duration, Instant};

fn exec_stdin_runtime() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime")
}

#[test]
fn piped_stdin_that_closes_quickly_goes_in_one_request() {
let (reader, mut writer) = std::io::pipe().expect("pipe");
writer.write_all(b"hi\n").unwrap();
drop(writer);
let collected = exec_stdin_runtime().block_on(super::collect_piped_stdin(
super::spawn_piped_stdin_reader(reader),
Duration::from_secs(5),
super::MAX_EXEC_STDIN_BYTES,
));
match collected.expect("collect") {
super::PipedStdin::Complete(prefix) => assert_eq!(prefix, b"hi\n"),
super::PipedStdin::Open { .. } => panic!("closed pipe must complete"),
}
}

#[test]
fn piped_stdin_that_stays_open_does_not_block_the_command() {
let (reader, mut writer) = std::io::pipe().expect("pipe");
writer.write_all(b"early").unwrap();
let grace = Duration::from_millis(100);
let started = Instant::now();
let runtime = exec_stdin_runtime();
let collected = runtime.block_on(super::collect_piped_stdin(
super::spawn_piped_stdin_reader(reader),
grace,
super::MAX_EXEC_STDIN_BYTES,
));
let elapsed = started.elapsed();
assert!(elapsed >= grace, "must wait the grace period: {elapsed:?}");
assert!(
elapsed < Duration::from_secs(3),
"must not wait for EOF: {elapsed:?}"
);
let super::PipedStdin::Open { prefix, mut rest } = collected.expect("collect") else {
panic!("an open pipe must start the command before EOF");
};
assert_eq!(prefix, b"early");
// The remainder keeps flowing after the command has started.
writer.write_all(b"late").unwrap();
drop(writer);
let next = runtime
.block_on(rest.recv())
.expect("late chunk")
.expect("read");
assert_eq!(next, b"late");
assert!(
runtime.block_on(rest.recv()).is_none(),
"EOF closes the channel"
);
}

#[test]
fn piped_stdin_over_the_limit_is_rejected_with_the_upload_hint() {
let (reader, mut writer) = std::io::pipe().expect("pipe");
writer.write_all(&[0u8; 8]).unwrap();
drop(writer);
let error = exec_stdin_runtime()
.block_on(super::collect_piped_stdin(
super::spawn_piped_stdin_reader(reader),
Duration::from_secs(5),
4,
))
.err()
.expect("over the limit must fail");
assert!(error.to_string().contains("sandbox upload"), "{error}");
}
}
8 changes: 6 additions & 2 deletions docs/how-it-works/sandboxes/overview.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,8 +325,12 @@ The CLI sends small piped input in one request for compatibility with older
gateways. It streams larger input in bounded frames, including for commands
without a TTY, so input is not limited by the gateway's per-message request
size. Piped input is limited to 4 MiB; use `sandbox upload` for larger files.
The CLI closes remote stdin when the pipe reaches EOF and continues
reading command output until the command finishes.
The CLI waits at most 200 ms for piped stdin to reach EOF. If the pipe is
still open after that, for example under a CI runner or a harness that never
closes stdin, the command starts anyway and the remaining input streams to it
as it arrives. The CLI closes remote stdin when the pipe reaches EOF and
continues reading command output until the command finishes. Redirect stdin
from `/dev/null` when a command needs no input.

The command's exit code is propagated to the CLI, so `exec` works in scripts that check return codes.
Large stdout and stderr streams are delivered before a successful exit. A slow
Expand Down
Loading