diff --git a/crates/tui/src/tools/shell.rs b/crates/tui/src/tools/shell.rs index 62aa201e55..e4ec8f876b 100644 --- a/crates/tui/src/tools/shell.rs +++ b/crates/tui/src/tools/shell.rs @@ -873,7 +873,23 @@ impl ShellChild { enum StdinWriter { Pipe(ChildStdin), - #[cfg(not(target_env = "ohos"))] + /// Unix PTY lane: a dup of the portable-pty master fd, so interact + /// writes can run on the bounded worker thread exactly like pipe + /// writes. The reader thread holds its own clone of the master fd for + /// the task's whole lifetime, so this dup is NOT the last master-side + /// handle: dropping it delivers nothing to the slave on its own. An + /// explicit `close_stdin` therefore delivers the ^D bytes itself, + /// through the same bounded worker as a payload write (see + /// [`bounded_stdin_eof_delivery`]), and a delivery that fails or + /// stalls claims no end-of-file. + /// The cfg mirrors the only constructor (the unix arm inside the + /// `not(target_env = "ohos")` PTY block — ohos returns from tty mode + /// before PTY setup): keeping the variant off ohos avoids a + /// never-constructed `dead_code` variant there, which `deny(warnings)` + /// makes fatal, and every referencing site carries the same predicate. + #[cfg(all(unix, not(target_env = "ohos")))] + PtyFile(std::fs::File), + #[cfg(all(not(target_env = "ohos"), not(unix)))] Pty(Box), } @@ -881,7 +897,9 @@ impl StdinWriter { fn write_all(&mut self, data: &[u8]) -> std::io::Result<()> { match self { StdinWriter::Pipe(stdin) => stdin.write_all(data), - #[cfg(not(target_env = "ohos"))] + #[cfg(all(unix, not(target_env = "ohos")))] + StdinWriter::PtyFile(file) => file.write_all(data), + #[cfg(all(not(target_env = "ohos"), not(unix)))] StdinWriter::Pty(writer) => writer.write_all(data), } } @@ -889,12 +907,210 @@ impl StdinWriter { fn flush(&mut self) -> std::io::Result<()> { match self { StdinWriter::Pipe(stdin) => stdin.flush(), - #[cfg(not(target_env = "ohos"))] + #[cfg(all(unix, not(target_env = "ohos")))] + StdinWriter::PtyFile(file) => file.flush(), + #[cfg(all(not(target_env = "ohos"), not(unix)))] StdinWriter::Pty(writer) => writer.flush(), } } } +/// Blocks SIGPIPE for the guarded region on unix, restoring it on drop. +/// The process deliberately runs with SIGPIPE at SIG_DFL (`run_with_args` +/// re-arms it so delegated children see conventional pipe semantics), which +/// arms it for in-process threads too. A bounded stdin write that times out +/// deliberately leaves its worker blocked mid-write on the handle, and the +/// documented stall recovery — killing the task, or the stalled command +/// simply exiting later — closes the read end: the resumed write would +/// otherwise kill the whole host process with SIGPIPE instead of failing +/// with EPIPE like every handled write error. +/// +/// Blocking alone is not sufficient: a SIGPIPE generated while blocked is +/// not discarded — it stays PENDING on the worker thread, and unblocking +/// delivers a pending unblocked signal before `pthread_sigmask` returns. +/// With SIGPIPE at SIG_DFL that would kill the process one instruction +/// after the "handled" EPIPE, exactly where the guard ends. The drop +/// therefore checks this thread's pending set and consumes a pending +/// SIGPIPE with `sigwait` BEFORE restoring the mask. A write-failure +/// SIGPIPE is thread-directed, so a pending SIGPIPE inside a guarded +/// region is one this region's write generated; `sigpending` also reports +/// process-directed signals, but nothing in this process raises SIGPIPE +/// that way, and consuming such a signal under SIG_DFL would only spare +/// the process. Test binaries run with std's SIG_IGN, under +/// which a generated SIGPIPE is discarded at generation and never becomes +/// pending — which is why the suite cannot catch a block-and-restore guard; +/// the consume path is pinned by re-arming SIG_DFL inside a helper process +/// (see `sigpipe_guard_survives_a_broken_pipe_restore`). +struct BlockedSigpipeGuard { + #[cfg(unix)] + previous: libc::sigset_t, +} + +impl BlockedSigpipeGuard { + fn new() -> Self { + #[cfg(unix)] + { + // SAFETY: `zeroed` is valid for `sigset_t` (an array of + // integer words) and both pointers below are local stack + // values. A failed `pthread_sigmask` is not expected with + // these local sets; if it did fail, the write would run with + // the process default disposition (the pre-guard behavior) + // and the drop's restore would install the zeroed `previous` + // mask — an empty set, leaving this thread's signals unblocked. + unsafe { + let mut blocked: libc::sigset_t = std::mem::zeroed(); + let mut previous: libc::sigset_t = std::mem::zeroed(); + libc::sigemptyset(&mut blocked); + libc::sigaddset(&mut blocked, libc::SIGPIPE); + // Best-effort: if the mask cannot be set, the write runs + // with the process default and the failure mode is simply + // the pre-guard one. + libc::pthread_sigmask(libc::SIG_BLOCK, &blocked, &mut previous); + Self { previous } + } + } + #[cfg(not(unix))] + Self {} + } +} + +#[cfg(unix)] +impl Drop for BlockedSigpipeGuard { + fn drop(&mut self) { + unsafe { + // A write that ended in EPIPE raised SIGPIPE while it was + // blocked here. Consume it before the restore — otherwise the + // unblock delivers the pending signal at the process + // disposition (SIG_DFL in production) and the guard's own Drop + // kills the process it exists to protect (see the struct doc). + let mut pending: libc::sigset_t = std::mem::zeroed(); + libc::sigpending(&mut pending); + if libc::sigismember(&pending, libc::SIGPIPE) == 1 { + let mut blocked: libc::sigset_t = std::mem::zeroed(); + libc::sigemptyset(&mut blocked); + libc::sigaddset(&mut blocked, libc::SIGPIPE); + let mut delivered: i32 = 0; + // SAFETY: the signal is pending on this thread, so the + // wait returns immediately; the sets are local stack + // values. + libc::sigwait(&blocked, &mut delivered); + } + libc::pthread_sigmask(libc::SIG_SETMASK, &self.previous, std::ptr::null_mut()); + } + } +} + +#[cfg(not(unix))] +impl Drop for BlockedSigpipeGuard { + fn drop(&mut self) {} +} + +/// Outcome of the bounded end-of-file delivery behind an explicit +/// `close_stdin` on a unix terminal task: `Delivered` means the ^D bytes +/// landed, `Failed` means the write errored without landing (typically the +/// child was already gone and the master write failed), and `Stalled` means +/// the tty input queue was still full at the deadline. +#[cfg(all(unix, not(target_env = "ohos")))] +#[derive(Debug)] +enum BoundedEofDelivery { + Delivered, + Failed, + Stalled, +} + +/// Deliver `\n` + VEOF through the given master fd on the same bounded +/// worker a payload write uses ([`bounded_stdin_write_owned`]). The leading +/// newline flushes any pending partial line, because canonical mode only +/// turns VEOF into a read of length zero from an EMPTY line buffer — the +/// same two bytes the old ownership box injected on drop. In raw mode both +/// bytes are ordinary input, exactly like that box drop; the caller records +/// a delivery only when the write really completed, so the flag never +/// claims an end-of-file that did not land. The EOF byte is written +/// literally (`0x04`) rather than read from `c_cc[VEOF]`: portable-pty's +/// unix writer reads the live byte at drop and writes nothing at all when +/// VEOF is disabled (`_POSIX_VDISABLE`), so the two behaviors differ in +/// that state — this close sends the two bytes as ordinary input, exactly +/// like raw mode, and no end-of-file ever arrives. The `close_stdin` +/// schema discloses that; the old box delivered nothing in the same +/// state, so the close still reports honestly what it did. +#[cfg(all(unix, not(target_env = "ohos")))] +fn bounded_stdin_eof_delivery(file: &std::fs::File) -> BoundedEofDelivery { + use std::os::fd::AsFd; + let dup_fd = match file.as_fd().try_clone_to_owned() { + Ok(fd) => fd, + // No clone means no worker can attempt the delivery; report it as + // a failed delivery — a clone failure is a process-fd problem, not + // a queue-full condition, and the flag must stay down either way. + Err(_) => return BoundedEofDelivery::Failed, + }; + match bounded_stdin_write_owned( + StdinWriter::Pipe(std::process::ChildStdin::from(dup_fd)), + b"\n\x04".to_vec(), + ) { + // The clone is dropped on every arm: the dup this function was + // given stays with the task, and only the write's outcome decides + // what the close may claim. + Err(_) => BoundedEofDelivery::Failed, + Ok((_, Ok(Ok(())))) => BoundedEofDelivery::Delivered, + Ok((_, Ok(Err(_)))) => BoundedEofDelivery::Failed, + Ok((_, Err(std::sync::mpsc::RecvTimeoutError::Timeout))) => BoundedEofDelivery::Stalled, + Ok((_, Err(std::sync::mpsc::RecvTimeoutError::Disconnected))) => BoundedEofDelivery::Failed, + } +} + +/// The single bounded stdin write every platform uses: the worker thread +/// takes the writer by value, writes and flushes the payload, and hands the +/// writer back through the channel. Unix reaches it with a clone of the +/// task's handle — pipes clone the `ChildStdin` fd and the PTY lane clones +/// the dup'd master fd (see [`BackgroundShell::write_stdin`] and +/// [`bounded_stdin_eof_delivery`]); Windows and ohos cannot clone an fd out +/// of their handle, so the handoff moves the real writer. +/// `Ok((Some(writer), Ok(io_result)))` returns the writer to the caller. +/// `Ok((None, Err(RecvTimeoutError::Timeout)))` means the deadline hit +/// first: the writer stays with the abandoned worker, the bytes still land +/// if the command resumes reading, and the caller marks the stall. +/// `Err(RecvTimeoutError::Disconnected)` means the worker died before +/// completing (a panic in the write path): the writer is gone, and the +/// caller must NOT mark a stall — a writer failure is not the command's +/// fault. An `Err` from the function itself means the worker thread could +/// not be spawned at all: nothing was written, and the writer is gone with +/// the failed closure (ownership was already handed over), so the caller +/// must report the task as having no writable stdin. A plain +/// `thread::spawn` panic here would unwind through the shell-manager lock +/// and poison it for the rest of the process. +type OwnedBoundedWriteOutcome = ( + Option, + Result, std::sync::mpsc::RecvTimeoutError>, +); + +fn bounded_stdin_write_owned( + writer: StdinWriter, + payload: Vec, +) -> anyhow::Result { + let (sender, receiver) = std::sync::mpsc::channel(); + let worker = std::thread::Builder::new() + .name("stdin-bounded-write".to_string()) + .spawn(move || { + let _sigpipe = BlockedSigpipeGuard::new(); + let mut writer = writer; + let result = writer.write_all(&payload).and_then(|_| writer.flush()); + // The writer moves back out only on the send path; a send failure + // (receiver gone) drops it, which is fine — the caller already + // gave up on the handle at that point. + let _ = sender.send((writer, result)); + }); + let receiver = match worker { + Ok(_) => receiver, + Err(error) => { + return Err(anyhow!(error).context("Failed to spawn the bounded stdin writer thread")); + } + }; + match receiver.recv_timeout(STDIN_WRITE_DEADLINE) { + Ok((writer, result)) => Ok((Some(writer), Ok(result))), + Err(error) => Ok((None, Err(error))), + } +} + fn spawn_reader_thread( mut reader: R, buffer: SharedRawOutput, @@ -1079,6 +1295,22 @@ pub struct BackgroundShell { stderr_cursor: usize, completion_reported: bool, stdin: Option, + /// Set when the child has really been handed end-of-file: either its + /// stdin pipe was closed through `write_stdin`, or an explicit unix + /// terminal close delivered the ^D bytes through the bounded worker. + /// Stays false when the task never exposed a writable stdin (persistent + /// services spawn with a null stdin, PTY lanes keep the slave open + /// through the reader), when the handle was simply released after + /// completion, and when a close delivery failed or stalled — so + /// `interact` can state the true reason stdin is unavailable. + stdin_closed_at_eof: bool, + /// Set when a stdin write outlived [`STDIN_WRITE_DEADLINE`] because the + /// command stopped consuming its pipe, or when a terminal close's own + /// ^D delivery stalled the same way (same deadline, same refusal + /// family). The abandoned bytes keep flowing from the background writer + /// if the child resumes, so further writes are refused (two writers + /// could interleave) instead of queued behind an unknown pipe state. + stdin_write_stalled: bool, child: Option, /// A stop was delivered but the direct child's exit was never observed: /// the bounded reap grace elapsed first (e.g. uninterruptible sleep), so @@ -1411,16 +1643,287 @@ impl BackgroundShell { } } + /// Close the stdin pipe of this task, keeping a PTY writer attached — + /// the foreground close: pipe readers get end-of-file at once, while a + /// terminal command keeps the interact surface a session member can + /// write to while it runs. Only an unstalled PIPE is claimed as + /// end-of-file: after a stalled write the abandoned worker still holds + /// a clone of the handle and the bytes have not finished landing, so + /// the flag stays down and refusals report the stall instead — the + /// same conjunct the `write_stdin` close applies. + fn close_stdin_pipe(&mut self) { + if matches!(self.stdin, Some(StdinWriter::Pipe(_))) { + self.stdin_closed_at_eof = !self.stdin_write_stalled; + self.stdin = None; + } + } + fn write_stdin(&mut self, input: &str, close: bool) -> Result<()> { - if let Some(stdin) = self.stdin.as_mut() { + // Taken owned so the Windows/ohos fallback can move the writer into + // its bounded worker and hand it back. Every unix exit restores the + // slot first, keeping the stall/eof refusal ordering identical with + // the pinned unix behavior; on the ownership-fallback cfgs the + // Timeout, Disconnected, and spawn-failure arms deliberately leave + // the slot empty — the writer is gone with the worker, and the stall + // flag or the returned refusal reports that. + // `mut` is real only on the ownership-fallback cfgs (they re-take and + // reassign the slot); the unix fd-clone path only moves it, so the + // same predicate that strips the fallback also silences `unused_mut`. + #[cfg_attr(all(unix, not(target_env = "ohos")), allow(unused_mut))] + let mut slot = self.stdin.take(); + if slot.is_some() { if !input.is_empty() { - stdin - .write_all(input.as_bytes()) - .context("Failed to write to stdin")?; - stdin.flush().ok(); + if self.stdin_write_stalled { + self.stdin = slot; + return Err(anyhow!( + "stdin writes to task {} are refused: an earlier write did not complete within {STDIN_WRITE_DEADLINE:?} because the command stopped consuming its stdin; the abandoned bytes still land if it resumes. Kill the task and start a new one instead of feeding it further.", + self.id + )); + } + // A blocking write under the shell-manager lock wedges the + // executor thread and every other shell in the session when + // the stdin handle's buffer is full and the command never + // reads (the exact stall the inline-payload cap exists to + // close on the run boundary — interact chunks and + // small-buffer platforms hit the same wall one chunk later). + // Every platform bounds the write now, by what its handle + // offers: on unix the write is bounded for BOTH handle kinds + // by fd clone — pipes clone the `ChildStdin` fd, and the PTY + // lane clones the dup'd master fd it spawned with — a worker + // thread owns the clone and finishes the payload in the + // background (it still lands if the child resumes), the + // caller learns about the stall at the deadline, and the + // stall flag refuses any further write so two writers can + // never interleave on one handle. + // The same predicate as the `PtyFile` variant. ohos strips + // this whole statement (the attribute gates the if-else), so + // ohos takes the owned-writer fallback below — that + // predicate is what keeps ohos compiling with a live write + // and no dead binding, field, or methods under the workspace + // deny(warnings). + #[cfg(all(unix, not(target_env = "ohos")))] + if let Some(writer) = slot.as_ref() { + // Exhaustive by construction: on this cfg `StdinWriter` + // has exactly `Pipe` and `PtyFile`, so a future third + // unix variant fails to COMPILE at the match below + // instead of silently falling into an unbounded blocking + // write under the session lock. The `else` arm only + // covers the `None` slot the take site already ruled + // out. + use std::os::fd::AsFd; + let dup_fd = match writer { + StdinWriter::Pipe(pipe) => pipe.as_fd().try_clone_to_owned(), + StdinWriter::PtyFile(file) => file.as_fd().try_clone_to_owned(), + }; + let dup_fd = match dup_fd { + Ok(fd) => fd, + Err(error) => { + // The clone failed without consuming the + // handle: restore it so the task keeps its + // interact surface instead of later refusals + // misreporting a live task as "a finished + // command has released its stdin". + self.stdin = slot; + return Err(anyhow!(error).context( + "Failed to clone the stdin handle for the bounded write", + )); + } + }; + // The bounded write is the shared helper (the same one + // the owned platforms and the ^D close delivery use): + // the clone rides it by value, so only the ORIGINAL + // slot decides what is restored below — a returned + // clone is dropped, never reattached. + match bounded_stdin_write_owned( + StdinWriter::Pipe(std::process::ChildStdin::from(dup_fd)), + input.as_bytes().to_vec(), + ) { + // The clone came back with a completed write: drop + // it and keep the original attached. + Ok((_, Ok(Ok(())))) => {} + Ok((_, Ok(Err(error)))) => { + // The worker completed: restore the ORIGINAL + // handle so a write error does not silently + // strip the task's interact surface (the base + // behavior kept the handle attached after a + // write error too). + self.stdin = slot; + return Err(anyhow!(error).context("Failed to write to stdin")); + } + // A channel close without a send means the worker + // died before completing (a panic in the write + // path): that is a writer failure, not a stall, so + // the task is not marked stalled — a healthy write + // after this must not be refused by the flag. + Ok((_, Err(std::sync::mpsc::RecvTimeoutError::Disconnected))) => { + self.stdin = slot; + return Err(anyhow!(stdin_writer_failed_refusal(&self.id))); + } + Ok((_, Err(std::sync::mpsc::RecvTimeoutError::Timeout))) => { + // The clone stays with the abandoned worker; + // the original handle is restored, and the + // stall flag refuses any further write so two + // writers can never interleave on one handle. + self.stdin_write_stalled = true; + self.stdin = slot; + return Err(anyhow!(stdin_stalled_write_refusal(&self.id))); + } + Err(error) => { + // The spawn failed before any byte was written + // and the worker holds nothing: restore the + // handle and fail like a plain write error. A + // bare `thread::spawn` panic here would unwind + // through the shell-manager mutex and poison it + // for the rest of the process. + self.stdin = slot; + return Err(anyhow!(error) + .context("Failed to spawn the bounded stdin writer thread")); + } + } + } else { + // The `None` slot the take site already ruled out: + // no handle to write through, nothing to restore. + } + // Windows and ohos cannot clone an fd out of their handle + // (Windows strips the statement above because `unix` is + // false, ohos by the `not(target_env = "ohos")` half — the + // portable-pty writer box exposes no cloneable handle at + // all). They bound by ownership instead: the worker takes + // the writer, finishes the payload, and hands the writer + // back through the channel; a deadline timeout leaves the + // writer with the abandoned worker (the bytes still land if + // the command resumes), marks the stall, and the caller + // returns with the session's shell lock released — cancel, + // kill, and wait all stay live. This is the same wedge the + // unix comment above describes, closed here too. + #[cfg(any(target_env = "ohos", not(unix)))] + { + let writer = slot.take().expect("slot held above"); + let (returned, result) = bounded_stdin_write_owned( + writer, + input.as_bytes().to_vec(), + ) + .map_err(|error| { + // The worker could not be spawned and the writer + // went with the failed closure: `slot` stays None, + // so the task honestly reports no writable stdin + // from here on. + anyhow!(error).context("Failed to spawn the bounded stdin writer thread") + })?; + match result { + Ok(Ok(())) => { + slot = returned; + } + Ok(Err(error)) => { + // The worker completed and handed the writer + // back: keep it attached so a write error does + // not silently strip the task's interact + // surface (same as the unix path above). + self.stdin = returned; + return Err(anyhow!(error).context("Failed to write to stdin")); + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + self.stdin_write_stalled = true; + // `slot` stays None: the abandoned worker owns + // the handle, so the close below cannot claim + // EOF and further writes hit the stall refusal. + return Err(anyhow!(stdin_stalled_write_refusal(&self.id))); + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + // The worker died before completing (a panic in + // the write path): that is a writer failure, not + // a stall — the task is not marked stalled so a + // healthy write after this is not refused, and + // the writer is gone with the worker (nothing + // to restore). + return Err(anyhow!(stdin_writer_failed_refusal(&self.id))); + } + } + } } if close { + // Only a dropped PIPE delivers end-of-file. An explicit + // close on a unix PTY writer must DELIVER it instead: the + // old ownership box injected `\n` + VEOF on drop (a `cat` + // on a pty exited 0 the instant it dropped), and the dup + // the terminal lane now holds delivers nothing on its own + // because the reader holds the master — so write the same + // ^D semantics through the dup, then drop it. The delivery + // runs on the same kind of bounded worker as a payload + // write: two bytes into a tty whose input queue is already + // full (the earlier payload completed inside the deadline, + // so the stall flag is still down) block just as forever + // as a large payload would, and inline under the manager + // lock they wedge every shell in the session. + #[cfg(all(unix, not(target_env = "ohos")))] + if let Some(StdinWriter::PtyFile(file)) = slot.as_ref() { + if !self.stdin_write_stalled { + match bounded_stdin_eof_delivery(file) { + BoundedEofDelivery::Delivered => { + self.stdin = None; + self.stdin_closed_at_eof = true; + return Ok(()); + } + // The close consumed the handle as requested, + // but the bytes did not land (typically the + // child was already gone and the master write + // failed): keep the flag down so later refusals + // do not claim readers saw end-of-file. Ok is + // honest here — the release the caller asked + // for happened, and no delivery is possible + // once the child is gone. + BoundedEofDelivery::Failed => { + self.stdin = None; + return Ok(()); + } + // Queue still full at the deadline: restore the + // handle under the stall flag and report the + // close as FAILED. The schema's close promises a + // delivered end-of-file, so a close that neither + // delivered nor released must not read as + // success — the same disposition the + // payload-timeout arm applies. + BoundedEofDelivery::Stalled => { + self.stdin_write_stalled = true; + self.stdin = slot; + return Err(anyhow!(stdin_stalled_close_refusal(&self.id))); + } + } + } + // A task whose last write stalled gets no delivery + // attempt at all: the queue that stalled the payload + // would stall the two bytes too, so the close is + // refused without waiting and without destroying the + // interact surface — the model is told the close was + // not applied instead of reading a success that + // delivered nothing. + self.stdin = slot; + return Err(anyhow!(stdin_stalled_close_not_applied_refusal(&self.id))); + } + // Windows cannot deliver end-of-file to a console child at + // all (portable-pty's ConPTY writer has no EOF-on-drop — + // that exists only in its unix writer), so an explicit + // close here only releases the handle. Say so instead of + // returning a silent success that the schema's "Close + // stdin" would read as a delivered end-of-file. + #[cfg(all(not(unix), not(target_env = "ohos")))] + if matches!(slot, Some(StdinWriter::Pty(_))) { + self.stdin = None; + return Err(anyhow!( + "stdin close for task {} released the terminal's stdin handle, but this platform cannot deliver end-of-file to a console child; the command keeps running until it finishes or its timeout ends it.", + self.id + )); + } + // A task whose last write STALLED must not claim EOF: the + // abandoned worker still holds a clone of the handle and + // the bytes have not finished landing, so the flag stays + // down and the tail refusal reports the stall instead. + if let Some(StdinWriter::Pipe(_)) = slot { + self.stdin_closed_at_eof = !self.stdin_write_stalled; + } self.stdin = None; + } else { + self.stdin = slot; } return Ok(()); } @@ -1429,7 +1932,23 @@ impl BackgroundShell { return Ok(()); } - Err(anyhow!("stdin is not available for task {}", self.id)) + if self.stdin_write_stalled { + return Err(anyhow!( + "stdin is not available for task {}: its last write stalled because the command stopped consuming stdin, and the abandoned writer may still hold the handle; kill it and start a new one.", + self.id + )); + } + + if self.stdin_closed_at_eof { + return Err(anyhow!( + "stdin is not available for task {}: its stdin was closed, so stdin readers have seen end-of-file (a program that put its terminal into raw mode — or disabled or remapped VEOF — consumes those bytes as ordinary input instead); start input-driven work with background=true and feed it with action=\"interact\"", + self.id + )); + } + Err(anyhow!( + "stdin is not available for task {}: no writable stdin handle is attached to it (a finished command has released its stdin, a persistent command never exposes one, a terminal command exposes one only while it runs and only until its stdin is closed, and a bounded stdin writer that failed to spawn or panicked takes its handle with it); start input-driven work with background=true and feed it with action=\"interact\"", + self.id + )) } fn full_output(&self) -> (String, String, usize, usize) { @@ -1986,6 +2505,8 @@ impl ShellManager { stderr_cursor: 0, completion_reported: false, stdin: None, + stdin_closed_at_eof: false, + stdin_write_stalled: false, child: None, stop_unconfirmed: false, #[cfg(windows)] @@ -2041,6 +2562,8 @@ impl ShellManager { stderr_cursor: 0, completion_reported: false, stdin: None, + stdin_closed_at_eof: false, + stdin_write_stalled: false, child: None, stop_unconfirmed: false, #[cfg(windows)] @@ -2491,6 +3014,13 @@ impl ShellManager { if stdin_data.is_some() { cmd.stdin(Stdio::piped()); + } else { + // std defaults to `Stdio::inherit()`, which on a hosted (non-PTY) + // run hands the child the parent's TTY or an arbitrary pipe it can + // never see EOF on. This lane has no stdin payload and no operator: + // a deterministic null stdin gives stdin-readers an immediate + // end-of-file instead of a hang or a SIGTTIN stop. + cmd.stdin(Stdio::null()); } child_env::apply_to_command(&mut cmd, child_env::string_map_env(&exec_env.env)); @@ -2525,6 +3055,17 @@ impl ShellManager { if let Some(input) = stdin_data && let Some(mut stdin) = child.stdin.take() { + // Same SIGPIPE discipline as the bounded stdin workers (see + // [`BlockedSigpipeGuard`]): the child dying mid-write must + // surface as a handled write error, not kill a process that + // runs with SIGPIPE at its default disposition. Unlike the + // tool surface this manager-API lane has neither + // [`MAX_INLINE_STDIN_BYTES`] nor the bounded writer's + // deadline — every production payload goes through the + // background lane's bounded writer (see + // [`STDIN_WRITE_DEADLINE`]), so a direct caller passing a + // payload here must keep it within what the child reads. + let _sigpipe = BlockedSigpipeGuard::new(); stdin .write_all(input.as_bytes()) .context("Failed to write to stdin")?; @@ -2842,22 +3383,73 @@ impl ShellManager { return Err(err).context("Failed to clone PTY reader"); } }; - let writer = match pair.master.take_writer() { + // The interact writer is acquired BEFORE the reader thread + // spawns, so a writer-acquisition failure aborts with only + // the child to clean up (kill + wait) instead of orphaning + // an already-spawned reader thread. Nothing observes the + // order: `pair.master` is alive across the whole block, so + // the slave sees no end-of-file in either arrangement. + // + // Unix: the interact writer is a dup of the master fd, and + // `take_writer` is NOT called — its box's Drop injects a + // newline plus `VEOF` into the master (that is how "dropping + // the writer sends EOF" is implemented there), which a + // terminal command that never receives an interact would read + // as an immediate end-of-file (probe-pinned: `cat` on a pty + // exited 0 the instant the box was dropped). The dup keeps + // a master-side description open — no EOF for the slave + // while the reader's clone is open too. A plain drop of the + // dup (the spawn-path close never touches this lane, and a + // close whose delivery failed releases it) still shows the + // child nothing; + // an explicit `close_stdin` instead delivers the ^D bytes + // itself through the bounded worker and claims end-of-file + // only when that write really landed; the final master + // handle closes at teardown, after the child can no longer + // observe it. + #[cfg(unix)] + let stdin_writer = { + use std::os::fd::FromRawFd; + match pair.master.as_raw_fd() { + Some(fd) => { + let dup_fd = unsafe { libc::dup(fd) }; + if dup_fd < 0 { + let _ = child.kill(); + let _ = child.wait(); + return Err(std::io::Error::last_os_error()) + .context("Failed to dup the PTY master for the stdin writer"); + } + // SAFETY: `dup_fd` is a freshly duplicated fd owned + // by this call; `File::from_raw_fd` takes ownership. + Some(StdinWriter::PtyFile(unsafe { + std::fs::File::from_raw_fd(dup_fd) + })) + } + None => { + let _ = child.kill(); + let _ = child.wait(); + return Err(anyhow!( + "Failed to obtain the PTY master fd for the stdin writer" + )); + } + } + }; + // Windows has no dup'd master fd to hand out (the + // portable-pty writer box is not fd-exposed there), so the + // lane keeps the box and its blocking write, per the + // interact comment. + #[cfg(not(unix))] + let stdin_writer = Some(StdinWriter::Pty(match pair.master.take_writer() { Ok(writer) => writer, Err(err) => { let _ = child.kill(); let _ = child.wait(); return Err(err).context("Failed to take PTY writer"); } - }; + })); let stdout_thread = Some(spawn_reader_thread(reader, Arc::clone(&stdout_buffer))); - ( - ShellChild::Pty(child), - Some(StdinWriter::Pty(writer)), - stdout_thread, - None, - ) + (ShellChild::Pty(child), stdin_writer, stdout_thread, None) } } else if persist_pending { let mut cmd = Command::new(program); @@ -2980,6 +3572,8 @@ impl ShellManager { stderr_cursor: 0, completion_reported: false, stdin, + stdin_closed_at_eof: false, + stdin_write_stalled: false, child: Some(child), stop_unconfirmed: false, #[cfg(windows)] @@ -3005,8 +3599,14 @@ impl ShellManager { if let Some(input) = stdin_data && let Err(err) = bg_shell.write_stdin(input, false) { + // No task id exists yet — the record is only registered below — + // so `write_stdin`'s "kill it and start a new one" recovery copy + // has no target here. The spawn aborts instead: the child is + // killed and the caller learns the run never started. let _ = bg_shell.kill(); - return Err(err); + return Err(err.context( + "the run never started: delivering the inline stdin payload failed, so the command was killed and NO task was created (the task id the error above mentions was never registered — do not try to cancel or kill it)", + )); } if let Err(err) = bg_shell.publish_lifecycle() { @@ -3105,6 +3705,22 @@ impl ShellManager { Ok(()) } + /// Close a foreground command's stdin PIPE so pipe readers get their + /// end-of-file at once, keeping any PTY writer attached: dropping the + /// PTY writer is invisible to the child (the reader holds the master) + /// but destroys the interact surface a session member can still use + /// while a terminal command runs — the foreground terminal lane's + /// contract. A missing record is a no-op, not an error: only eviction + /// removes a record, and eviction drops the stdin handle with it, so + /// the pipe cannot still be open — and the close runs between spawn + /// and the first caller-visible step, where an error would abort a run + /// whose result is already determined. + fn close_stdin_pipe(&mut self, task_id: &str) { + if let Some(shell) = self.processes.get_mut(task_id) { + shell.close_stdin_pipe(); + } + } + pub fn write_stdin_for_session( &mut self, active_session_id: &str, @@ -4573,20 +5189,50 @@ async fn execute_foreground_via_background( let task_id = spawned .task_id .ok_or_else(|| anyhow!("foreground shell did not return a process id"))?; - if let Some(permit) = heavy_permit { + // The stdin write end must close BEFORE any fallible step between spawn + // and close: a future step that can fail here (the heavy-permit attach + // below) would otherwise return early and leak an open write end, + // wedging stdin-reading commands until the foreground watchdog killed + // them. The task-id extraction above is the one step between spawn and + // close, and `spawn_background_sandboxed` always returns an id on `Ok`, + // so the close reaches the shell layer without a fallible step in + // between (the manager-level `close_stdin_pipe` wrapper treats an + // already-evicted record as success — eviction drops the stdin + // handle with the record, so the invariant that matters holds: no + // later failure can leave the pipe open). + // + // Foreground has no operator at the keyboard, so whatever the command + // will ever read from stdin must be bounded now. With stdin data the + // payload was delivered at spawn and closing the write end turns the + // pipe into a normal end-of-file; with no data, closing immediately + // delivers that EOF at once. Leaving the write end open wedged any + // stdin-reading command (`cat`, a bare `$SHELL`) on a pipe that never + // EOFs until the foreground watchdog killed it. This is the foreground + // stdin contract the tool description states: a command detached to the + // background later (`request_foreground_background`) keeps running, but + // its stdin stays at EOF — `interact` reports the closed stdin and points + // at background=true, which is the lane for input-driven work. The close + // is deliberately pipe-only: a terminal (`combined_output`) foreground + // command's PTY writer must stay attached, because dropping it would + // destroy the interact surface a session member can still write to + // while the terminal command runs (the child sees no difference — the + // reader holds the master — and the terminal lane's blocking is bounded + // by the foreground watchdog, not by a never-EOFing pipe). The + // manager-side close is idempotent-safe after natural completion and on + // lanes where the handle is absent (persistent). + { let mut manager = context .shell_manager .lock() .map_err(|_| anyhow!("shell manager lock poisoned"))?; - manager.attach_heavy_permit(&task_id, permit)?; + manager.close_stdin_pipe(&task_id); } - - if stdin_data.is_some() { + if let Some(permit) = heavy_permit { let mut manager = context .shell_manager .lock() .map_err(|_| anyhow!("shell manager lock poisoned"))?; - manager.write_stdin(&task_id, "", true)?; + manager.attach_heavy_permit(&task_id, permit)?; } let deadline = timeout_ms.map(|timeout| Instant::now() + Duration::from_millis(timeout)); @@ -4657,6 +5303,66 @@ async fn execute_foreground_via_background( const BASH_MAX_TIMEOUT_MS: u64 = i32::MAX as u64; +/// Upper bound for one inline `stdin` payload (counted in UTF-8 bytes). +/// Every payload the tool surface accepts is delivered through the bounded +/// stdin writer (see [`STDIN_WRITE_DEADLINE`]), so a command that stops +/// reading surfaces as a deadline stall rather than a wedged executor; the +/// cap additionally keeps oversized payloads from ever reaching the +/// writer, and on Linux's default 64 KiB pipe buffer a cap-sized payload +/// completes into an empty pipe without a reader, so the common +/// payload-then-close foreground shape delivers without stalling there. +/// Platforms with smaller default pipe buffers lean on the deadline +/// instead. Anything larger is refused with a pointer at background +/// `interact` streaming instead of hanging the session. +const MAX_INLINE_STDIN_BYTES: usize = 64 * 1024; + +/// Wall-clock bound on one stdin write. At the bound an `interact` write is +/// abandoned to its background worker (the bytes still land if the child +/// resumes reading), the task's stdin is marked stalled, and the manager +/// lock is released — so each stalled write costs one bounded wait under +/// the lock, and n stalled tasks serialize into n such waits. A stalled +/// SPAWN payload is different: no task id exists yet, so the spawn aborts +/// and the child is killed instead of running task-less. The abandoned +/// worker is SIGPIPE-blocked (see [`BlockedSigpipeGuard`]): when the task +/// is later killed — the documented stall recovery — the worker observes +/// EPIPE instead of taking the process down. +const STDIN_WRITE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(5); + +/// One shared copy per stdin refusal shape so the unix fd-clone path and +/// the owned-writer path cannot drift apart — the pins assert substrings +/// of these. They are functions rather than consts so the deadline +/// interpolation stays a real format capture. +fn stdin_stalled_write_refusal(task_id: &str) -> String { + format!( + "the write to task {task_id} did not complete within {STDIN_WRITE_DEADLINE:?}: the command stopped consuming its stdin, so its input buffer filled and the rest of the payload was abandoned to a background writer (it still lands if the command resumes reading). Further stdin writes to this task are refused; kill it and start a new one." + ) +} + +fn stdin_writer_failed_refusal(task_id: &str) -> String { + format!( + "the stdin writer for task {task_id} failed: the background write ended without completing or reporting (the writer task panicked); the payload's delivery is unknown. Kill the task and start a new one." + ) +} + +/// Serves only the unix PTY close arm (the Windows/ohos close arm has its +/// own inline copy and never takes the unix delivery path). Same +/// predicate as the `PtyFile` variant, so `deny(warnings)`' fatal +/// `dead_code` cannot fire on platforms where the arm is compiled out. +#[cfg(all(unix, not(target_env = "ohos")))] +fn stdin_stalled_close_refusal(task_id: &str) -> String { + format!( + "the stdin close for task {task_id} did not complete within {STDIN_WRITE_DEADLINE:?}: the terminal's input queue stayed full because the command stopped consuming input, so the end-of-file bytes were not delivered; the close was not applied and the interact surface stays attached. Further stdin writes to this task are refused; kill it and start a new one." + ) +} + +/// See [`stdin_stalled_close_refusal`]: unix-PTY-only, same cfg predicate. +#[cfg(all(unix, not(target_env = "ohos")))] +fn stdin_stalled_close_not_applied_refusal(task_id: &str) -> String { + format!( + "the stdin close for task {task_id} was not applied: its last stdin write stalled because the command stopped consuming input, so the end-of-file was never delivered and no close can deliver it now; the interact surface stays attached and further stdin writes are refused; kill it and start a new one." + ) +} + /// Initial cadence for foreground-via-background completion polling. Fast /// commands dominate real agent traffic; detection latency on `true`-class /// commands drops from ~100ms to ~10ms, while long commands reach the @@ -4775,7 +5481,7 @@ impl ToolSpec for LowercaseBashTool { } fn description(&self) -> &'static str { - "Execute a shell command in the workspace and return stdout and stderr. The command runs in the detected platform shell — `$SHELL` on Unix (any shell it names, falling back to `/bin/sh`); on Windows, bash when the session provides one, otherwise PowerShell (pwsh 7 or Windows PowerShell 5.1) or cmd.exe. Windows PowerShell 5.1 lacks `&&`; prefer forms the detected shell supports. In read-only shell sessions a single approved command runs directly as argv instead (only pipelines still go through the shell); with an external sandbox backend configured this tool is unavailable (use the backend-specific shell tool instead). Output keeps the last 2000 lines or 50KB. An optional timeout is expressed in seconds; when omitted the command is killed after 120 seconds, so pass an explicit timeout for work expected to take longer. In Ask, after a sandbox denial, retry the exact command once with sandbox_permissions (the narrowest wider mode that suffices) and a one-sentence justification; the approval prompt asks the user." + "Execute a shell command in the workspace and return stdout and stderr. The command runs in the detected platform shell — `$SHELL` on Unix (any shell it names, falling back to `/bin/sh`); on Windows, bash when the session provides one, otherwise PowerShell (pwsh 7 or Windows PowerShell 5.1) or cmd.exe. Windows PowerShell 5.1 lacks `&&`; prefer forms the detected shell supports. In read-only shell sessions a single approved command runs directly as argv instead (only pipelines still go through the shell). This foreground surface runs without a terminal stdin: interactive programs and a bare `$SHELL` receive end-of-file and exit. With an external sandbox backend configured this tool is unavailable (use the backend-specific shell tool instead). Output keeps the last 2000 lines or 50KB. An optional timeout is expressed in seconds; when omitted the command is killed after 120 seconds, so pass an explicit timeout for work expected to take longer. In Ask, after a sandbox denial, retry the exact command once with sandbox_permissions (the narrowest wider mode that suffices) and a one-sentence justification; the approval prompt asks the user." } fn input_schema(&self) -> serde_json::Value { @@ -4990,14 +5696,17 @@ impl ToolSpec for BashTool { }, "stdin": { "type": "string", - "description": "Stdin data to send (action=run: before waiting; action=interact: to the background task). Also accepted as `input` or `data` — send only one." + "maxLength": 65536, + "description": "Stdin data to send (action=run: before waiting; action=interact: to the background task). Also accepted as `input` or `data` — send only one. Inline payloads are capped at 64 KiB, counted as UTF-8 bytes (maxLength mirrors the same number): a write to a command that is not reading ends as a bounded deadline-stall refusal instead of blocking the session; for bigger payloads run the command in the background and feed it through `action=interact` in chunks." }, "input": { "type": "string", + "maxLength": 65536, "description": "Alias for `stdin`." }, "data": { "type": "string", + "maxLength": 65536, "description": "Alias for `stdin`." }, "cwd": { @@ -5036,7 +5745,7 @@ impl ToolSpec for BashTool { }, "close_stdin": { "type": "boolean", - "description": "Close stdin after sending (action=interact)" + "description": "Close stdin after sending (action=interact); on a terminal task the close delivers end-of-file, sent as a literal newline followed by the EOF byte — a program that put its terminal into raw mode, or disabled or remapped VEOF, receives those bytes as ordinary input instead, and on Windows the close only releases the terminal handle (a console child has no end-of-file to deliver)" }, "all": { "type": "boolean", @@ -5190,12 +5899,30 @@ impl ToolSpec for BashTool { // for misspelled names. A wrong type is an error, never a no-op. let stdin_data = match first_present_field(&input, &["stdin", "input", "data"]) { None => None, - Some((name, value)) => Some( - value + Some((name, value)) => { + let payload = value .as_str() .ok_or_else(|| type_mismatch(name, value, "a string"))? - .to_string(), - ), + .to_string(); + // The tool surface's inline cap keeps an oversized payload + // from ever reaching the writer; the spawn delivers the + // payload through the bounded stdin writer, so a command + // that never reads surfaces as the writer's 5-second + // deadline stall refusal rather than a wedged executor. On + // Linux the cap equals the default pipe buffer, so a + // cap-sized write into an empty pipe completes without a + // reader there; smaller default buffers lean on the same + // deadline on the bounded lanes. Bigger payloads belong on + // the background lane, fed through `action=interact` in + // chunks. + if payload.len() > MAX_INLINE_STDIN_BYTES { + return Ok(ToolResult::error(format!( + "The inline stdin payload is {} bytes; the shell tool accepts at most {MAX_INLINE_STDIN_BYTES} inline — run the command with background=true and feed it through action=interact in chunks.", + payload.len() + ))); + } + Some(payload) + } }; if interactive && background { @@ -6086,9 +6813,24 @@ impl BashTool { // place whichever spelling the model uses. let interaction_input = match first_present_field(input, &["stdin", "input", "data"]) { None => "", - Some((name, value)) => value - .as_str() - .ok_or_else(|| type_mismatch(name, value, "a string"))?, + Some((name, value)) => { + let payload = value + .as_str() + .ok_or_else(|| type_mismatch(name, value, "a string"))?; + // The same bound the inline spawn payload carries: one + // oversized `interact` payload would push megabytes through + // a bounded write for no benefit — feed ≥64 KiB of input as + // several interact calls (or generate it in the command). + // The deadline below is what actually bounds the stall; this + // keeps each write bounded too. + if payload.len() > MAX_INLINE_STDIN_BYTES { + return Ok(ToolResult::error(format!( + "The interact stdin payload is {} bytes; the shell tool accepts at most {MAX_INLINE_STDIN_BYTES} per interact call — split the feed across interact calls.", + payload.len() + ))); + } + payload + } }; { diff --git a/crates/tui/src/tools/shell/guidance.rs b/crates/tui/src/tools/shell/guidance.rs index 8a565a1205..5ca9ff5aab 100644 --- a/crates/tui/src/tools/shell/guidance.rs +++ b/crates/tui/src/tools/shell/guidance.rs @@ -85,7 +85,20 @@ pub(super) fn description() -> &'static str { \"wait\" blocks for a background task until completion or timeout; \ \"interact\" sends stdin to a background task; \"cancel\" kills a background task. \ Pass wait=false for a nonblocking task snapshot. Foreground mode is for bounded \ - commands; use background=true for work expected to take >5 seconds. \ + commands; use background=true for work expected to take >5 seconds. Foreground \ + commands without `interactive`, `tty`, or `combined_output` run without a \ + terminal stdin — interactive programs and a bare `$SHELL` receive EOF and \ + exit; `tty` always selects a background terminal task, and a foreground \ + command with `combined_output` runs on a terminal lane that nothing \ + writes to unless a session member interacts with the task, so a \ + stdin-reading command waits until the watchdog ends it. A foreground \ + run without `combined_output` may carry one inline stdin payload \ + (delivered to the command, then end-of-file); a `combined_output` \ + foreground run delivers the payload but its terminal lane stays \ + open, so no end-of-file follows it. For multi-write input-driven \ + workflows supply stdin on the background lane: \ + use background=true plus \ + action=\"interact\". \ Output is truncated per stream (~30KB: head/tail kept, middle summarized; \ see metadata flags).", runtime_command_guidance() diff --git a/crates/tui/src/tools/shell/tests.rs b/crates/tui/src/tools/shell/tests.rs index d73535648b..7a9f9e26e4 100644 --- a/crates/tui/src/tools/shell/tests.rs +++ b/crates/tui/src/tools/shell/tests.rs @@ -3697,6 +3697,8 @@ fn killed_shell_does_not_wait_for_blocked_reader_threads() { completion_reported: false, bounded_output: None, stdin: None, + stdin_closed_at_eof: false, + stdin_write_stalled: false, child: None, stop_unconfirmed: false, windows_job: None, @@ -5024,6 +5026,1156 @@ fn bounded_reap_reports_grace_expiry_while_the_child_lives() { let _ = child.wait(); } +/// An inline stdin payload past one pipe buffer is refused before spawn: +/// the payload is delivered with a blocking write while the shell-manager +/// lock is held, and a non-reading child would wedge the executor past any +/// watchdog. The boundary payload (exactly 64 KiB) still runs. The raw +/// lane is the one that carries payloads — the read-only grammar rejects +/// a `stdin` field outright. +// Unix-only: the boundary leg executes `wc -c`, which the sanitized child +// PATH does not resolve on Windows (and the deadline semantics the +// boundary payload rides are unix — see `write_stdin_…` above). +#[cfg(unix)] +#[tokio::test] +async fn inline_stdin_payload_over_the_cap_refuses() { + let workspace = tempdir().expect("workspace"); + let context = ToolContext::new(workspace.path().to_path_buf()); + let oversized = "x".repeat(super::MAX_INLINE_STDIN_BYTES + 1); + let output = BashTool::new("Bash") + .execute(json!({"command": "wc -c", "stdin": oversized}), &context) + .await + .expect("the refusal is a result, not an error"); + assert!(!output.success, "{}", output.content); + assert!( + output.content.contains("action=interact"), + "the refusal must point at the streaming alternative: {}", + output.content + ); + + let boundary = "y".repeat(super::MAX_INLINE_STDIN_BYTES); + let output = BashTool::new("Bash") + .execute(json!({"command": "wc -c", "stdin": boundary}), &context) + .await + .expect("the boundary payload dispatches"); + assert!(output.success, "{}", output.content); + // Exact equality: a `contains` anchor would stay green if the cap + // moved to a value whose spelling embeds this one (655360 contains + // "65536"). + assert_eq!( + output.content.trim(), + super::MAX_INLINE_STDIN_BYTES.to_string(), + "the boundary payload must reach the command whole: {}", + output.content + ); +} +/// Runs a stdin scenario on a worker thread behind a 60 s outer guard. The +/// scenario's own deadline machinery is what should finish it; a mechanism +/// regression that never returns trips this guard instead (the join can +/// extend past the guard while the scenario's child still holds the blocked +/// write open — the child's own watchdog releases it, so the guard +/// terminates, loudly late). A worker panic surfaces as a failed verdict, +/// never as a hung test. +#[cfg(unix)] +fn run_stdin_scenario_with_hang_guard( + name: &'static str, + body: impl FnOnce() -> Result<(), String> + Send + 'static, +) -> Result<(), String> { + let (sender, receiver) = std::sync::mpsc::channel::>(); + let handle = std::thread::spawn(move || { + let verdict = body(); + let _ = sender.send(verdict.clone()); + verdict + }); + match receiver.recv_timeout(Duration::from_secs(60)) { + Ok(verdict) => { + // Reaps the worker; the verdict from the channel is the + // scenario outcome, so the join's own copy adds nothing. + let _ = handle.join().expect("the scenario worker panicked"); + verdict + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + let _ = handle.join(); + Err(format!("{name} hung past its 60 s guard")) + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + // The worker died before its verdict reached the channel. + let _ = handle.join().expect_err("the scenario worker panicked"); + Err(format!("{name} panicked before reporting a verdict")) + } + } +} + +#[cfg(unix)] +const SIGPIPE_GUARD_HELPER_ENV: &str = "CODEWHALE_SIGPIPE_GUARD_HELPER"; + +/// Lets the driver verify the helper actually RAN: libtest exits 0 on an +/// unmatched `--exact` filter, so a future rename of the helper test would +/// otherwise leave the pin silently green while guarding nothing. +#[cfg(unix)] +const SIGPIPE_GUARD_HELPER_CANARY_ENV: &str = "CODEWHALE_SIGPIPE_GUARD_HELPER_CANARY"; + +/// Helper half of `sigpipe_guard_survives_a_broken_pipe_restore`: runs in a +/// process that re-armed SIGPIPE at SIG_DFL (what `run_with_args` does in +/// production), takes the guard, writes into a readerless pipe, checks that +/// the drop restored the prior mask, and only then writes the ran-canary — +/// so the driver fails on a helper that died at signal 13 (consume removed), +/// panicked on a left-blocked mask (restore removed), or never ran (rename +/// drift). +#[cfg(unix)] +#[test] +fn sigpipe_guard_helper_process() { + if std::env::var(SIGPIPE_GUARD_HELPER_ENV).ok().as_deref() != Some("1") { + return; + } + // The suite binary runs with std's SIG_IGN, under which a broken-pipe + // SIGPIPE is discarded at generation and never becomes pending — the + // bug cannot reproduce there, so the helper re-arms the production + // disposition first. + unsafe { + libc::signal(libc::SIGPIPE, libc::SIG_DFL); + let mut fds = [0 as libc::c_int; 2]; + assert_eq!(libc::pipe(fds.as_mut_ptr()), 0, "pipe"); + libc::close(fds[0]); + { + let guard = super::BlockedSigpipeGuard::new(); + let buf = [b'x'; 4096]; + let written = libc::write(fds[1], buf.as_ptr().cast(), buf.len()); + assert_eq!(written, -1, "the readerless write must fail"); + assert_eq!( + std::io::Error::last_os_error().raw_os_error(), + Some(libc::EPIPE), + "the readerless write must fail with EPIPE" + ); + // Dropping the guard restores the mask. A block-and-restore + // guard delivers the pending SIGPIPE here and this process + // dies at signal 13 before reaching the end of the test; the + // consume-before-restore fix reaches it and exits 0. + drop(guard); + } + // The restore half is pinned too: after the drop, SIGPIPE must no + // longer be blocked on this thread. A no-op restore would leave + // every later SIGPIPE pending on this thread forever — invisible + // to the exit-status assertion above, so query the mask directly. + let mut now_blocked: libc::sigset_t = std::mem::zeroed(); + assert_eq!( + libc::pthread_sigmask(libc::SIG_BLOCK, std::ptr::null(), &mut now_blocked), + 0, + "query the signal mask" + ); + assert_eq!( + libc::sigismember(&now_blocked, libc::SIGPIPE), + 0, + "the guard drop must restore the prior mask (SIGPIPE must not stay blocked)" + ); + libc::close(fds[1]); + } + // Ran-canary for the driver: written only after every assertion above + // passed, so an unmatched filter (helper never ran) is distinguishable + // from a green run. + if let Some(canary) = std::env::var_os(SIGPIPE_GUARD_HELPER_CANARY_ENV) { + std::fs::write(&canary, b"ran").expect("write the ran-canary"); + } +} + +/// The SIGPIPE guard must consume the signal it blocked before restoring +/// the mask: a bounded write that ends in EPIPE leaves SIGPIPE pending on +/// the worker thread, and a plain block-and-restore delivers it at the +/// process disposition (SIG_DFL in production, see `run_with_args`) — +/// killing the whole host one instruction after the "handled" EPIPE. The +/// suite cannot see this (test binaries run SIG_IGN, so no SIGPIPE ever +/// becomes pending), so the driver re-execs this binary as a helper that +/// re-arms SIG_DFL first: with the consume the helper exits 0; without it +/// the helper dies at signal 13 and the non-success status fails here. +#[cfg(unix)] +#[test] +fn sigpipe_guard_survives_a_broken_pipe_restore() { + let test_binary = std::env::current_exe().expect("current test binary"); + let canary = std::env::temp_dir().join(format!( + "codewhale-sigpipe-guard-canary-{}", + std::process::id() + )); + let _ = std::fs::remove_file(&canary); + let mut helper = Command::new(&test_binary) + .arg("--exact") + .arg("tools::shell::tests::sigpipe_guard_helper_process") + .env(SIGPIPE_GUARD_HELPER_ENV, "1") + .env(SIGPIPE_GUARD_HELPER_CANARY_ENV, &canary) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn sigpipe guard helper"); + let status = helper.wait().expect("wait sigpipe guard helper"); + assert!( + status.success(), + "the guard's mask restore must not deliver the pending SIGPIPE (a broken guard kills the helper at signal 13): {status}" + ); + assert!( + canary.exists(), + "the helper must actually run — libtest exits 0 on an unmatched --exact filter, so a green status without the canary means this pin guarded nothing (helper renamed?)" + ); + let _ = std::fs::remove_file(&canary); +} + +/// The bounded stdin write: a command that never consumes its pipe fills +/// it, and past [`STDIN_WRITE_DEADLINE`] the write errors and the task's +/// stdin is marked stalled instead of holding the shell-manager lock (and +/// with it every other shell in the session) forever. The stall flag then +/// refuses further writes without another wait — two writers could +/// interleave on one pipe. +// Unix-only: the bounded write (worker + deadline) the test exercises is +// `#[cfg(unix)]`, as are the raw-mode tty semantics it pins; the +// Windows/ohos owned path is bounded by the same deadline through +// `bounded_stdin_write_owned` and covered by its own helper test. +#[cfg(unix)] +#[test] +fn write_stdin_to_a_child_that_stops_consuming_errors_instead_of_wedging() { + let verdict = run_stdin_scenario_with_hang_guard( + "the bounded-write pin", + || -> Result<(), String> { + let tmp = tempdir().map_err(|error| error.to_string())?; + let mut manager = ShellManager::new(tmp.path().to_path_buf()); + let result = manager + .execute_with_options_env( + &sleep_command(60), + None, + 60_000, + true, + None, + false, + None, + HashMap::new(), + ) + .map_err(|error| error.to_string())?; + let task_id = result + .task_id + .ok_or_else(|| "background execution should return task_id".to_string())?; + // Far past any pipe buffer on any platform: the write cannot + // complete unless the child reads, and it never does. + let payload = "x".repeat(200 * 1024); + let started = std::time::Instant::now(); + let error = manager + .write_stdin(&task_id, &payload, false) + .err() + .ok_or_else(|| "the stalled write must error at the deadline".to_string())? + .to_string(); + if !error.contains("did not complete within") { + return Err(error); + } + let elapsed = started.elapsed(); + if elapsed < STDIN_WRITE_DEADLINE { + return Err(format!( + "the write returned before the deadline: {elapsed:?}" + )); + } + // No upper bound on the elapsed time: the load-bearing assertions + // are that the deadline FIRED (elapsed ≥ deadline) and further + // writes are refused. A fixed upper bound is load-fragile at any + // multiple — under a saturated full-suite parallel run the + // deadline worker was observed starving past 30 s — while a + // broken-mechanism mutant (deadline removed) never returns and + // trips the outer hang guard instead. + // Further writes are refused by the stall flag without waiting. + let started = std::time::Instant::now(); + let again = manager.write_stdin(&task_id, "y", false); + if again.is_ok() { + return Err("the stalled task must refuse further writes".to_string()); + } + if started.elapsed() > std::time::Duration::from_millis(500) { + return Err("the stall refusal must not wait on another write".to_string()); + } + // A close after the stall must not claim the end-of-file the + // abandoned writer never delivered: the post-close refusal must + // still name the stall (the `stdin_closed_at_eof` stall conjunct + // and the stall-first check order together keep the EOF claim + // off; a reorder or conjunct drop that surfaced the EOF message + // reddens here). + let _closed = manager.write_stdin(&task_id, "", true); + let after_close = manager + .write_stdin(&task_id, "z", false) + .err() + .ok_or_else(|| "the post-stall close must leave writes refused".to_string())? + .to_string(); + if !after_close.contains("stalled") || after_close.contains("end-of-file") { + return Err(format!( + "a close after a stall must not claim an EOF the abandoned writer has not delivered: {after_close}" + )); + } + manager.kill(&task_id).map_err(|error| error.to_string())?; + Ok(()) + }, + ); + verdict.expect("bounded-write assertions"); +} +/// The `interact` payload carries the same per-call bound as the inline +/// spawn payload: one oversized call is refused with a pointer at chunking +/// before any write is attempted. +#[tokio::test] +async fn interact_payload_over_the_cap_refuses() { + let workspace = tempdir().expect("workspace"); + let context = ToolContext::new(workspace.path().to_path_buf()); + let oversized = "x".repeat(super::MAX_INLINE_STDIN_BYTES + 1); + let output = BashTool::new("Bash") + .execute( + json!({"action": "interact", "command": "pwd", "task_id": "no-such-task", "stdin": oversized}), + &context, + ) + .await + .expect("the refusal is a result, not an error"); + assert!(!output.success, "{}", output.content); + assert!( + output.content.contains("per interact call"), + "the refusal must name the per-call bound: {}", + output.content + ); + // The advertised bound stays in sync with the constant (the copy + // interpolates it; this asserts the value actually reaches the model). + assert!( + output + .content + .contains(&super::MAX_INLINE_STDIN_BYTES.to_string()), + "the refusal must carry the exact bound value: {}", + output.content + ); +} +/// The tool description states the foreground stdin contract; the sentence is +/// the model-facing contract for why stdin readers exit, so it is pinned. +#[test] +fn guidance_states_the_foreground_stdin_contract() { + let description = super::guidance::description(); + assert!( + description.contains("Foreground commands without `interactive`, `tty`, or `combined_output` run without a terminal stdin"), + "guidance must state the bounded foreground stdin: {description}" + ); + assert!( + description.contains("`tty` always selects a background terminal task"), + "guidance must state where tty commands actually run — the run path forces them background: {description}" + ); + assert!( + description.contains( + "a terminal lane that nothing writes to unless a session member interacts with the task" + ), + "guidance must state the combined_output foreground stdin contract accurately: {description}" + ); + assert!( + description.contains("action=\"interact\""), + "guidance must point at background interact for input-driven work: {description}" + ); +} +/// The lowercase tool's own description carries the same foreground stdin +/// contract: its legacy translation accepts only command/timeout/sandbox +/// fields, so that sentence is the whole stdin story on that surface. +#[test] +fn lowercase_bash_description_states_the_foreground_stdin_contract() { + let description = LowercaseBashTool.description(); + assert!( + description.contains("This foreground surface runs without a terminal stdin"), + "the lowercase description must state the bounded foreground stdin: {description}" + ); + assert!( + description.contains("a bare `$SHELL` receive end-of-file"), + "the bare-shell probe must be named as exiting on EOF: {description}" + ); +} +/// A foreground command that reads stdin with NO stdin data used to wedge on +/// a pipe whose write end the tool kept open: the live repro was `$SHELL` +/// holding the turn until the foreground watchdog killed it at the full +/// timeout (`cat` behaves identically). The foreground lane must deliver EOF +/// at spawn instead. Background keeps the pipe open for deferred +/// `interact` streaming — that contract is pinned by +/// `test_write_stdin_streams_output` and must not regress. +#[tokio::test] +async fn foreground_without_stdin_data_gives_stdin_readers_eof() { + let workspace = tempdir().expect("workspace"); + let context = ToolContext::new(workspace.path().to_path_buf()); + // `cat` on Unix, `more` on cmd, `[Console]::In.ReadToEnd()` on PowerShell — + // each blocks until stdin reaches EOF. + let echo_stdin = echo_stdin_command(); + + let started = Instant::now(); + let result = BashTool::new("Bash") + .execute(json!({"command": echo_stdin, "timeout_ms": 3000}), &context) + .await + .expect("foreground stdin-reader must spawn"); + let elapsed = started.elapsed(); + + assert!( + result.success, + "a stdin-reading foreground command with no stdin data must see EOF \ + and exit promptly, not run to the timeout; took {elapsed:?}: {}", + result.content + ); + // The elapsed bound is a unix pin: on Windows a cold PowerShell startup + // can legitimately approach the 3000 ms watchdog under parallel load, + // while `success` above already fails if the watchdog killed the run. + #[cfg(unix)] + assert!( + elapsed < Duration::from_millis(2900), + "EOF must be delivered right after spawn, not discovered at the \ + timeout; took {elapsed:?}: {}", + result.content + ); +} +/// When the inline payload cannot be delivered at spawn, the run must abort +/// with the run-never-started context: no task id exists yet, so the child +/// is killed and the error must say the mentioned task id was never +/// registered — the caller must not try to cancel or kill it. The +/// manager-level lane has no inline cap, so an over-pipe-buffer payload +/// against a non-reading child reaches the bounded writer's deadline here; +/// the tool surface cannot produce this stall on Linux (its 64 KiB cap +/// equals the default pipe buffer), which is why this pin drives the +/// manager API directly. +#[test] +#[cfg(unix)] +fn spawn_aborts_with_run_never_started_context_when_the_payload_stalls() { + let tmp = tempdir().expect("tempdir"); + let mut manager = ShellManager::new(tmp.path().to_path_buf()); + // 2 MiB: far past any default pipe buffer, so the bounded writer's + // deadline is what ends the write. + let payload = "x".repeat(2 * 1024 * 1024); + + let started = Instant::now(); + let err = manager + .execute_with_options_env( + &sleep_command(60), + None, + 60_000, + true, + Some(payload.as_str()), + false, + None, + HashMap::new(), + ) + .expect_err("the spawn must abort when the inline payload stalls"); + let elapsed = started.elapsed(); + + let text = format!("{err:#}"); + assert!( + text.contains("the run never started"), + "error must carry the run-never-started context; got: {text}" + ); + assert!( + text.contains("never registered"), + "error must tell the caller the task id was never registered; got: {text}" + ); + assert!( + !text.contains("Kill the task"), + "the task-targeted recovery copy must not appear on the spawn-abort \ + path; got: {text}" + ); + assert!( + elapsed >= STDIN_WRITE_DEADLINE, + "the abort must come from the bounded writer's deadline, not an \ + instant failure; took {elapsed:?}: {text}" + ); +} +#[cfg(unix)] +const SYNC_STDIN_EOF_HELPER_ENV: &str = "CODEWHALE_SYNC_STDIN_EOF_HELPER"; +#[cfg(unix)] +const SYNC_STDIN_EOF_RESULT_FILE_ENV: &str = "CODEWHALE_SYNC_STDIN_EOF_RESULT_FILE"; +/// Helper half of +/// `sync_lane_without_stdin_data_gets_eof_not_inherited_stdin`: runs one +/// synchronous (background=false) stdin-reading command from a process whose +/// own stdin is a pipe the driver deliberately holds open, and records the +/// resulting status for the driver to judge. +#[cfg(unix)] +#[test] +fn sync_lane_stdin_eof_helper_process() { + if std::env::var(SYNC_STDIN_EOF_HELPER_ENV).ok().as_deref() != Some("1") { + return; + } + let result = ShellManager::new(std::env::temp_dir()) + .execute_with_options_env( + &echo_stdin_command(), + None, + 10_000, + false, + None, + false, + None, + HashMap::new(), + ) + .expect("sync execute in helper"); + std::fs::write( + PathBuf::from( + std::env::var(SYNC_STDIN_EOF_RESULT_FILE_ENV).expect("sync stdin result file"), + ), + format!("{:?}", result.status), + ) + .expect("write sync stdin result file"); +} +/// The sync lane's old default left the child inheriting the host's stdin +/// (std's `Stdio::inherit()`), so on a live TTY a stdin-reading command hung +/// or drew SIGTTIN. Reproduce deterministically: re-exec this test binary as +/// a helper whose own stdin is a pipe this process holds open, and run the +/// sync lane inside it. An inherited pipe can never EOF; `Stdio::null()` +/// gives the child an immediate end-of-file. +#[cfg(unix)] +#[test] +fn sync_lane_without_stdin_data_gets_eof_not_inherited_stdin() { + let tmp = tempdir().expect("tempdir"); + let result_file = tmp.path().join("sync-stdin-result"); + let test_binary = std::env::current_exe().expect("current test binary"); + let mut helper = Command::new(&test_binary) + .arg("--exact") + .arg("tools::shell::tests::sync_lane_stdin_eof_helper_process") + .arg("--nocapture") + .env(SYNC_STDIN_EOF_HELPER_ENV, "1") + .env(SYNC_STDIN_EOF_RESULT_FILE_ENV, &result_file) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn sync stdin helper"); + // Hold the write end for the whole run: while it is open, an inherited + // stdin can never reach EOF, which is exactly the old hang. + let held_stdin = helper.stdin.take().expect("helper stdin pipe"); + let deadline = Instant::now() + Duration::from_secs(60); + let status = loop { + match helper.try_wait().expect("poll sync stdin helper") { + Some(status) => break status, + None => { + assert!( + Instant::now() < deadline, + "sync-lane helper never finished; the lane is inheriting \ + a never-EOF stdin" + ); + std::thread::sleep(Duration::from_millis(50)); + } + } + }; + drop(held_stdin); + assert!(status.success(), "sync stdin helper failed: {status}"); + let outcome = std::fs::read_to_string(&result_file).expect("sync lane result file"); + assert_eq!( + outcome, "Completed", + "a sync command reading stdin with no stdin data must get EOF, not \ + inherited host stdin" + ); +} +/// The foreground stdin contract survives a detach: the command keeps +/// running in the background, but its stdin was put at end-of-file at spawn +/// (so stdin readers exit instead of wedging), and a later `interact` says so +/// and points at the background lane instead of failing opaquely. +#[tokio::test] +async fn detached_foreground_command_reports_its_closed_stdin_on_interact() { + let tmp = tempdir().expect("tempdir"); + let ctx = ToolContext::new(tmp.path()); + let shell_manager = ctx.shell_manager.clone(); + let command = sleep_command(30); + let task_ctx = ctx.clone(); + + let task = tokio::spawn(async move { + BashTool::new("Bash") + .execute( + json!({"command": command, "timeout_ms": 600_000}), + &task_ctx, + ) + .await + .expect("execute") + }); + tokio::time::sleep(Duration::from_millis(150)).await; + shell_manager + .lock() + .expect("shell manager lock") + .request_foreground_background(); + let result = tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("foreground shell should detach") + .expect("task should not panic"); + let task_id = result + .metadata + .as_ref() + .and_then(|meta| meta.get("task_id")) + .and_then(Value::as_str) + .expect("task id") + .to_string(); + + let mut manager = shell_manager.lock().expect("shell manager lock"); + assert_eq!( + manager + .inspect_job(&task_id) + .expect("inspect job") + .snapshot + .status, + ShellStatus::Running, + "the detached command keeps running" + ); + let error = manager + .write_stdin(&task_id, "hello\n", false) + .expect_err("stdin was closed at spawn by the foreground contract") + .to_string(); + assert!(error.contains("end-of-file"), "{error}"); + assert!(error.contains("background=true"), "{error}"); + let killed = manager.kill(&task_id).expect("kill"); + assert_eq!(killed.status, ShellStatus::Killed); +} +/// The foreground stdin-EOF close is deliberately pipe-only: a terminal +/// (`combined_output`) foreground command keeps its PTY writer attached so +/// a session member can interact with it while it runs — the base behavior +/// on that lane, which an unconditional close silently destroyed (and the +/// guidance sentence describes). +#[tokio::test] +#[cfg(all(unix, not(target_env = "ohos")))] +async fn foreground_terminal_command_keeps_its_interact_surface() { + let tmp = tempdir().expect("tempdir"); + let ctx = ToolContext::new(tmp.path()); + let shell_manager = ctx.shell_manager.clone(); + let command = echo_stdin_command(); + let task_ctx = ctx.clone(); + + let task = tokio::spawn(async move { + BashTool::new("Bash") + .execute( + json!({"command": command, "combined_output": true, "timeout_ms": 600_000}), + &task_ctx, + ) + .await + .expect("execute") + }); + tokio::time::sleep(Duration::from_millis(300)).await; + shell_manager + .lock() + .expect("shell manager lock") + .request_foreground_background(); + let result = tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("foreground shell should detach") + .expect("task should not panic"); + let task_id = result + .metadata + .as_ref() + .and_then(|meta| meta.get("task_id")) + .and_then(Value::as_str) + .expect("task id") + .to_string(); + + let mut manager = shell_manager.lock().expect("shell manager lock"); + { + let snapshot = manager.inspect_job(&task_id).expect("inspect job").snapshot; + assert_eq!( + snapshot.status, + ShellStatus::Running, + "the terminal command keeps running (exit {:?}, out {:?}, err {:?})", + snapshot.exit_code, + snapshot.stdout_tail, + snapshot.stderr_tail + ); + } + // The interact surface must still be attached: a session member writes + // to the running terminal command and the input reaches it (the tty + // echoes, and `cat` reads the line and writes it back). + manager + .write_stdin(&task_id, "hello-round14\n", false) + .expect("the terminal lane keeps a writable stdin while the command runs"); + let mut delivered = false; + for _ in 0..40 { + std::thread::sleep(Duration::from_millis(150)); + let snapshot = manager.inspect_job(&task_id).expect("inspect job").snapshot; + if snapshot.stdout_tail.contains("hello-round14") + || snapshot.stderr_tail.contains("hello-round14") + { + delivered = true; + break; + } + } + assert!( + delivered, + "the interact write must reach the terminal command: {:?}", + manager.inspect_job(&task_id).expect("inspect job").snapshot + ); + manager.kill(&task_id).expect("kill"); +} +/// The unix PTY interact write rides the same bounded worker as a pipe +/// write: a tty command that stops consuming input fills its line- +/// discipline buffer, the write hits the deadline, the payload is abandoned +/// to a background writer, and further writes are refused by the stall +/// flag — instead of blocking the session's shell-manager lock forever. +#[test] +#[cfg(all(unix, not(target_env = "ohos")))] +fn interact_write_to_a_stalled_tty_command_is_bounded() { + let verdict = run_stdin_scenario_with_hang_guard( + "the bounded tty-write pin", + || -> Result<(), String> { + let tmp = tempdir().map_err(|error| error.to_string())?; + let mut manager = ShellManager::new(tmp.path().to_path_buf()); + let result = manager + // A RAW-mode tty is the shape whose full input queue + // backpressures the master write (canonical mode discards + // overflow instead): a raw `sleep` that never reads pins the + // writer until the deadline. + .execute_with_options_env( + // The marker prints only after `stty` succeeds, so the + // wait below also turns a stty failure into an honest + // setup error instead of a misleading write success. + "stty raw -echo 2>/dev/null && printf RAW-READY; sleep 60", + None, + 60_000, + true, + None, + true, + None, + HashMap::new(), + ) + .map_err(|error| error.to_string())?; + let task_id = result + .task_id + .ok_or_else(|| "background execution should return task_id".to_string())?; + // Raw mode must be established before the write lands: on a fast + // or loaded host the background shell may not have run `stty` + // yet, and a payload written in canonical mode drains (overflow + // is discarded there) instead of backpressuring — the write then + // completes and the deadline pin never gets to fire. That race + // failed the first CI run of this pin. + let mut raw_ready = false; + for _ in 0..40 { + std::thread::sleep(Duration::from_millis(150)); + let snapshot = manager.inspect_job(&task_id).expect("inspect job").snapshot; + if snapshot.stdout_tail.contains("RAW-READY") { + raw_ready = true; + break; + } + } + assert!( + raw_ready, + "raw mode must be established before the stalled write: {:?}", + manager.inspect_job(&task_id).expect("inspect job").snapshot + ); + // Far past any line-discipline input buffer: the write cannot + // complete unless the child reads, and it never does. + let payload = "x".repeat(200 * 1024); + let started = std::time::Instant::now(); + let error = manager + .write_stdin(&task_id, &payload, false) + .err() + .ok_or_else(|| "the stalled tty write must error at the deadline".to_string())? + .to_string(); + if !error.contains("did not complete within") { + return Err(error); + } + let elapsed = started.elapsed(); + if elapsed < STDIN_WRITE_DEADLINE { + return Err(format!( + "the write returned before the deadline: {elapsed:?}" + )); + } + // No upper bound on the elapsed time: the load-bearing assertions + // are that the deadline FIRED (elapsed ≥ deadline) and further + // writes are refused. A fixed upper bound is load-fragile at any + // multiple — under a saturated full-suite parallel run the + // deadline worker was observed starving past 30 s — while a + // broken-mechanism mutant (deadline removed) never returns and + // trips the outer hang guard instead. + let started = std::time::Instant::now(); + let again = manager.write_stdin(&task_id, "y", false); + if again.is_ok() { + return Err("the stalled tty task must refuse further writes".to_string()); + } + if started.elapsed() > std::time::Duration::from_millis(500) { + return Err("the stall refusal must not wait on another write".to_string()); + } + manager.kill(&task_id).map_err(|error| error.to_string())?; + Ok(()) + }, + ); + assert!(verdict.is_ok(), "{verdict:?}"); +} +/// An explicit `close_stdin` on a terminal task whose last write already +/// STALLED must stay bounded, must not claim an end-of-file the abandoned +/// worker never delivered, and must not read as success either: the stall +/// skip refuses the close outright ("was not applied") and keeps the +/// interact surface attached, and the tail refusal still names the stall. +/// A mutation that reports the stalled close as Ok reddens on the +/// expect_err; a close that wedges on the full input queue trips the outer +/// guard instead. +#[cfg(all(unix, not(target_env = "ohos")))] +#[test] +fn close_stdin_on_a_stalled_terminal_task_is_bounded_and_claims_no_eof() { + let verdict = run_stdin_scenario_with_hang_guard( + "the stalled-terminal-close pin", + || -> Result<(), String> { + let tmp = tempdir().map_err(|error| error.to_string())?; + let mut manager = ShellManager::new(tmp.path().to_path_buf()); + let result = manager + // Same shape as the stalled-write pin above: a raw-mode + // tty whose command never reads backpressures the master + // write, and the marker only prints after `stty` succeeds. + .execute_with_options_env( + "stty raw -echo 2>/dev/null && printf RAW-READY; sleep 60", + None, + 60_000, + true, + None, + true, + None, + HashMap::new(), + ) + .map_err(|error| error.to_string())?; + let task_id = result + .task_id + .ok_or_else(|| "background execution should return task_id".to_string())?; + let mut raw_ready = false; + for _ in 0..40 { + std::thread::sleep(Duration::from_millis(150)); + let snapshot = manager.inspect_job(&task_id).expect("inspect job").snapshot; + if snapshot.stdout_tail.contains("RAW-READY") { + raw_ready = true; + break; + } + } + assert!( + raw_ready, + "raw mode must be established before the stalled write: {:?}", + manager.inspect_job(&task_id).expect("inspect job").snapshot + ); + let payload = "x".repeat(200 * 1024); + let error = manager + .write_stdin(&task_id, &payload, false) + .err() + .ok_or_else(|| "the stalled tty write must error at the deadline".to_string())? + .to_string(); + if !error.contains("did not complete within") { + return Err(error); + } + // The close itself must return promptly (no delivery attempt + // on a stalled queue) and must NOT report success: with the + // end-of-file undelivered, an Ok close would tell the model + // stdin is closed when it is not. + let started = std::time::Instant::now(); + let close_error = manager + .write_stdin(&task_id, "", true) + .expect_err("a close on a stalled terminal task must not claim success") + .to_string(); + if started.elapsed() >= STDIN_WRITE_DEADLINE { + return Err(format!( + "the close on a stalled terminal task must not wait on the queue: {:?}", + started.elapsed() + )); + } + if !close_error.contains("was not applied") { + return Err(format!( + "the stalled-close refusal must say the close was not applied: {close_error}" + )); + } + // And any further stdin traffic must still be refused without + // claiming an EOF: no delivery happened, so "end-of-file" would + // be a lie (the refusal here comes from the write-stall gate — + // the refused close left the handle attached). + let after_close = manager + .write_stdin(&task_id, "y", false) + .err() + .ok_or_else(|| "the post-close write must be refused".to_string())? + .to_string(); + if after_close.contains("end-of-file") { + return Err(format!( + "a terminal close after a stall must not claim an EOF that never landed: {after_close}" + )); + } + manager.kill(&task_id).map_err(|error| error.to_string())?; + Ok(()) + }, + ); + assert!(verdict.is_ok(), "{verdict:?}"); +} +/// The owned-writer bounded stdin write (the Windows/ohos fallback path): +/// a consuming child gets the payload and the writer handed back; a child +/// that never reads fills its pipe buffer and past the deadline the helper +/// reports TimedOut with the writer left to the abandoned worker — the +/// caller's lock is never held across the deadline. Present on every +/// target; production unix keeps the fd-clone path, so this exercise is +/// test-only there. +#[cfg(unix)] +#[test] +fn bounded_stdin_write_owned_returns_the_writer_and_times_out_clean() { + use std::process::{Command, Stdio}; + + let mut child = Command::new("cat") + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .spawn() + .expect("cat spawns"); + let stdin = child.stdin.take().expect("stdin piped"); + let (returned, result) = + super::bounded_stdin_write_owned(super::StdinWriter::Pipe(stdin), b"hello\n".to_vec()) + .expect("the bounded-write worker thread must spawn"); + let Ok(result) = result else { + panic!("a consuming child must complete the write, not lose the worker: {result:?}"); + }; + result.expect("a consuming child accepts the payload"); + assert!(returned.is_some(), "the writer is handed back on success"); + drop(returned); + child.wait().expect("cat exits after its stdin closes"); + + let mut child = Command::new("sleep") + .arg("120") + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .spawn() + .expect("sleep spawns"); + let stdin = child.stdin.take().expect("stdin piped"); + let payload = vec![b'x'; 256 * 1024]; // well past the 64 KiB pipe buffer + let started = std::time::Instant::now(); + let (returned, result) = + super::bounded_stdin_write_owned(super::StdinWriter::Pipe(stdin), payload) + .expect("the bounded-write worker thread must spawn"); + let elapsed = started.elapsed(); + assert!( + matches!(result, Err(std::sync::mpsc::RecvTimeoutError::Timeout)), + "a non-reading child must time the write out: {result:?}" + ); + assert!( + returned.is_none(), + "a timed-out writer stays with the abandoned worker" + ); + assert!( + elapsed >= super::STDIN_WRITE_DEADLINE, + "the timeout must honor the deadline, not fire early: {elapsed:?}" + ); + let _ = child.kill(); + let _ = child.wait(); +} + +/// `bounded_stdin_eof_delivery` — the worker the explicit terminal close +/// rides — must classify its outcome honestly: a write that lands reports +/// `Delivered`, a write into a FULL pipe (nothing consuming) reports +/// `Stalled` only at the deadline, and a write to a readerless pipe reports +/// `Failed`. A Stalled that came back as Delivered would let `write_stdin` +/// claim an end-of-file that never landed. The call is deadline-bounded by +/// construction, so a broken deadline hangs this test into the nextest +/// slow-timeout instead of passing. +// Unix PTY-lane cfg, matching the function it pins. +#[cfg(all(unix, not(target_env = "ohos")))] +#[test] +fn bounded_stdin_eof_delivery_reports_stalled_failed_and_delivered() { + use std::os::fd::FromRawFd; + + // Delivered: /dev/null accepts the two bytes unconditionally. + let null = std::fs::OpenOptions::new() + .write(true) + .open("/dev/null") + .expect("open /dev/null"); + assert!( + matches!( + super::bounded_stdin_eof_delivery(&null), + super::BoundedEofDelivery::Delivered + ), + "a write that lands must report Delivered" + ); + + // Failed: a readerless pipe errors the write without delivering (the + // suite runs with SIG_IGN, so the EPIPE surfaces as an error, not a + // signal — the process-death discipline is pinned by the sigpipe guard + // helper test). + let mut fds = [0 as libc::c_int; 2]; + assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0, "pipe"); + unsafe { libc::close(fds[0]) }; + let orphan = unsafe { std::fs::File::from_raw_fd(fds[1]) }; + assert!( + matches!( + super::bounded_stdin_eof_delivery(&orphan), + super::BoundedEofDelivery::Failed + ), + "a readerless pipe must report Failed" + ); + drop(orphan); + + // Stalled: fill a pipe to capacity without a consumer (O_NONBLOCK so + // the test thread never blocks), restore blocking mode for the worker, + // and the two-byte delivery cannot land until the deadline. + let mut fds = [0 as libc::c_int; 2]; + assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0, "pipe"); + let read_fd = fds[0]; + let full = unsafe { std::fs::File::from_raw_fd(fds[1]) }; + unsafe { libc::fcntl(fds[1], libc::F_SETFL, libc::O_NONBLOCK) }; + let chunk = [b'x'; 4096]; + loop { + let written = unsafe { libc::write(fds[1], chunk.as_ptr().cast(), chunk.len()) }; + if written <= 0 { + break; // EAGAIN: the pipe is full + } + } + unsafe { libc::fcntl(fds[1], libc::F_SETFL, 0) }; + let started = std::time::Instant::now(); + let outcome = super::bounded_stdin_eof_delivery(&full); + drop(full); + assert!( + matches!(outcome, super::BoundedEofDelivery::Stalled), + "a full pipe must stall the delivery: {outcome:?}" + ); + assert!( + started.elapsed() >= super::STDIN_WRITE_DEADLINE, + "the stall must honor the deadline: {:?}", + started.elapsed() + ); + // Unblock the abandoned worker: a reader consumes the queued bytes, the + // worker's write lands, its send fails on the dropped receiver, and the + // thread exits instead of parking for the process lifetime. The drain + // is bounded so the reader cannot hang once the pipe ends are closed. + let drainer = std::thread::spawn(move || { + let mut buf = [0u8; 4096]; + let mut total = 0usize; + while total < 512 * 1024 { + let n = unsafe { libc::read(read_fd, buf.as_mut_ptr().cast(), buf.len()) }; + if n <= 0 { + break; + } + total += n as usize; + } + unsafe { libc::close(read_fd) }; + }); + drainer.join().expect("drain the stalled pipe"); +} + +/// The FRESH-stall close arm — a terminal close whose own ^D delivery +/// stalls with the stall flag still down (no prior write set it) — must +/// wait the bounded deadline, return the close refusal that cites it, +/// restore the handle under the stall flag, and refuse the follow-up write +/// without another deadline wait. A mutant reporting this arm as Ok would +/// read as the schema's delivered end-of-file while the bytes never +/// landed; the not-applied pin next door only reaches the SKIP arm (its +/// scenario stalls a write first), so this record-level pin is the only +/// guard on the arm itself. Drives `write_stdin` on a record whose stdin +/// is a `PtyFile` over a full pipe — exactly what a real PTY task hands +/// the arm. +#[cfg(all(unix, not(target_env = "ohos")))] +#[test] +fn close_stdin_that_stalls_on_fresh_delivery_refuses_and_stays_attached() { + use std::os::fd::FromRawFd; + + // Fill a pipe to capacity without a consumer (O_NONBLOCK so the test + // thread never blocks), restore blocking mode for the worker: the + // two-byte ^D delivery cannot land until the deadline. + let mut fds = [0 as libc::c_int; 2]; + assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0, "pipe"); + let read_fd = fds[0]; + let full = unsafe { std::fs::File::from_raw_fd(fds[1]) }; + unsafe { libc::fcntl(fds[1], libc::F_SETFL, libc::O_NONBLOCK) }; + let chunk = [b'x'; 4096]; + loop { + let written = unsafe { libc::write(fds[1], chunk.as_ptr().cast(), chunk.len()) }; + if written <= 0 { + break; // EAGAIN: the pipe is full + } + } + unsafe { libc::fcntl(fds[1], libc::F_SETFL, 0) }; + + let now = std::time::Instant::now(); + let mut shell = BackgroundShell { + id: "fresh-stall-close".to_string(), + owner_session_id: "stdin-test-session".to_string(), + command: "cat".to_string(), + working_dir: std::path::PathBuf::from("."), + status: ShellStatus::Running, + exit_code: None, + started_at: now, + finished_at: None, + last_output_at: now, + last_observed_output_len: 0, + sandbox_type: SandboxType::None, + ownership: ShellOwnership::Managed, + linked_task_id: None, + owner_agent: None, + origin_tool_call_id: None, + origin_turn_id: None, + stdout_buffer: super::new_shared_raw_output(), + stderr_buffer: None, + bounded_output: None, + heavy_permit: None, + stdout_cursor: 0, + stderr_cursor: 0, + completion_reported: false, + stdin: Some(StdinWriter::PtyFile(full)), + stdin_closed_at_eof: false, + stdin_write_stalled: false, + child: None, + stop_unconfirmed: false, + stdout_thread: None, + stderr_thread: None, + work_lifecycle: None, + lifecycle_seq: 0, + last_lifecycle_status: None, + last_lifecycle_bytes: 0, + }; + + let started = std::time::Instant::now(); + let close_err = shell + .write_stdin("", true) + .expect_err("a close whose own delivery stalls must be refused, not reported Ok"); + assert!( + started.elapsed() >= super::STDIN_WRITE_DEADLINE, + "the fresh-stall close must attempt the bounded delivery first: {:?}", + started.elapsed() + ); + let message = format!("{close_err:#}"); + assert!( + message.starts_with("the stdin close for task fresh-stall-close"), + "the refusal must be the close refusal: {message}" + ); + assert!( + message.contains("did not complete within"), + "the fresh-stall refusal must cite the deadline: {message}" + ); + assert!( + !message.contains("no close can deliver it now"), + "a fresh stall must not take the not-applied skip arm: {message}" + ); + assert!( + !message.contains("stdin writes to task"), + "the close refusal must not be the write-stall gate: {message}" + ); + assert!( + shell.stdin.is_some(), + "the refused close must keep the interact surface attached" + ); + assert!( + shell.stdin_write_stalled, + "the fresh-stall close must mark the stall for later refusals" + ); + assert!( + !shell.stdin_closed_at_eof, + "a stalled delivery must not claim an end-of-file" + ); + + // The follow-up write is refused by the stall gate without another + // bounded wait. + let started = std::time::Instant::now(); + let write_err = shell + .write_stdin("x", false) + .expect_err("a write after a stalled close must be refused"); + assert!( + started.elapsed() < std::time::Duration::from_millis(500), + "the stall flag must refuse the follow-up write without another deadline wait: {:?}", + started.elapsed() + ); + let message = format!("{write_err:#}"); + assert!( + message.contains("stdin writes to task fresh-stall-close are refused"), + "the follow-up refusal must be the stall gate: {message}" + ); + + // Unblock the abandoned worker exactly like the classification pin: a + // bounded reader consumes the queued bytes, the worker's write lands, + // its send fails on the dropped receiver, and the thread exits. The + // record must drop first — it still holds the ORIGINAL pipe write end, + // and until every write end closes the drainer's read never sees EOF, + // so joining before the drop would hang the harness (the classification + // pin avoids this by dropping `full` before its call). + drop(shell); + let drainer = std::thread::spawn(move || { + let mut buf = [0u8; 4096]; + let mut total = 0usize; + while total < 512 * 1024 { + let n = unsafe { libc::read(read_fd, buf.as_mut_ptr().cast(), buf.len()) }; + if n <= 0 { + break; + } + total += n as usize; + } + unsafe { libc::close(read_fd) }; + }); + drainer.join().expect("drain the stalled pipe"); +} + /// An unconfirmed stop keeps its write scope across every coordination /// query — until a later poll reaps the real child and clears the /// uncertainty. The flag models the un-reaped state; the confirmation cycle @@ -5290,3 +6442,100 @@ fn dropping_the_manager_stops_a_child_with_an_unconfirmed_stop() { std::thread::sleep(Duration::from_millis(20)); } } + +#[test] +fn bash_schema_stdin_max_length_matches_the_enforced_inline_cap() { + // The schema literal and the enforced byte cap are one contract; this + // keeps the two hand-maintained numbers from drifting apart. + let schema = BashTool::new("Bash").input_schema(); + assert!( + !schema["properties"]["stdin"].is_null(), + "the full schema must declare stdin at its root properties" + ); + for key in ["stdin", "input", "data"] { + assert_eq!( + schema["properties"][key]["maxLength"], + json!(super::MAX_INLINE_STDIN_BYTES), + "schema {key} maxLength must equal the enforced inline byte cap" + ); + } +} + +// The ^D close delivery is compiled only for the unix PTY lane (the same +// predicate as the `PtyFile` variant): Windows has no console EOF to +// deliver and ohos has no tty lane at all, so both the exit and the +// end-of-file refusal assertions below are meaningless there. +#[cfg(all(unix, not(target_env = "ohos")))] +#[tokio::test] +async fn interact_close_stdin_delivers_eof_to_a_terminal_command() { + // Base delivered end-of-file to a terminal task through the ownership + // box's drop (`cat` on a pty exited 0 the instant it dropped); the + // dup'd-master rewrite made that drop invisible, so an explicit + // `close_stdin: true` must write the ^D semantics itself. Without it + // the command below runs until the watchdog and this pin reddens. + let tmp = tempdir().expect("tempdir"); + let ctx = ToolContext::new(tmp.path()); + let shell_manager = ctx.shell_manager.clone(); + let command = echo_stdin_command(); + let task_ctx = ctx.clone(); + + let task = tokio::spawn(async move { + BashTool::new("Bash") + .execute( + json!({"command": command, "combined_output": true, "timeout_ms": 600_000}), + &task_ctx, + ) + .await + .expect("execute") + }); + tokio::time::sleep(Duration::from_millis(300)).await; + shell_manager + .lock() + .expect("shell manager lock") + .request_foreground_background(); + let result = tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("foreground shell should detach") + .expect("task should not panic"); + let task_id = result + .metadata + .as_ref() + .and_then(|meta| meta.get("task_id")) + .and_then(Value::as_str) + .expect("task id") + .to_string(); + + let mut manager = shell_manager.lock().expect("shell manager lock"); + // Feed one line first so the close proves itself against a command that + // is genuinely mid-read, then deliver end-of-file. + manager + .write_stdin(&task_id, "close-me\n", false) + .expect("the terminal lane keeps a writable stdin while the command runs"); + manager + .write_stdin(&task_id, "", true) + .expect("an explicit stdin close on the terminal lane must succeed"); + let mut exited = false; + for _ in 0..40 { + std::thread::sleep(Duration::from_millis(150)); + let snapshot = manager.inspect_job(&task_id).expect("inspect job").snapshot; + if snapshot.exit_code.is_some() { + exited = true; + break; + } + } + assert!( + exited, + "closing stdin must end the terminal command (base parity with the \ + ownership-box drop): {:?}", + manager.inspect_job(&task_id).expect("inspect job").snapshot + ); + // The post-close refusal must tell the truth: stdin readers have really + // seen end-of-file now. + let error = manager + .write_stdin(&task_id, "again\n", false) + .expect_err("writes after the close must be refused"); + assert!( + error.to_string().contains("end-of-file"), + "the post-close refusal must cite end-of-file: {error}" + ); +} diff --git a/crates/tui/src/tools/verifier.rs b/crates/tui/src/tools/verifier.rs index aba4d275b2..ed686eab87 100644 --- a/crates/tui/src/tools/verifier.rs +++ b/crates/tui/src/tools/verifier.rs @@ -568,7 +568,7 @@ fn start_background_gates( .shell_manager .lock() .map_err(|_| ToolError::execution_failed("shell manager lock poisoned"))?; - manager.execute_with_options_env_for_session( + let spawned = manager.execute_with_options_env_for_session( &command, Some(&cwd), BACKGROUND_GATE_TIMEOUT_MS, @@ -578,7 +578,25 @@ fn start_background_gates( context.elevated_sandbox_policy.clone(), env, &context.state_namespace, - ) + ); + // A gate job is machine-launched with nobody to feed its stdin, + // so hand the child end-of-file at once: a gate that reads stdin + // then exits instead of idling on an open pipe until the + // background timeout kills it. Idempotent-safe if the job already + // finished. + if let Some(task_id) = spawned + .as_ref() + .ok() + .and_then(|result| result.task_id.as_deref()) + && let Err(error) = manager.write_stdin(task_id, "", true) + { + // Unreachable today (the job was just inserted under the + // held lock and the close writes nothing), but a silent + // `.ok()` here would quietly revert gates to idling on + // the open pipe until the background gate timeout. + tracing::warn!(%task_id, %error, "closing the gate job's stdin failed; the gate may idle until its timeout"); + } + spawned }; match spawn_result { @@ -1588,4 +1606,59 @@ mod tests { output.stdout ); } + + /// A machine-launched gate job has nobody to feed its stdin, so the + /// spawn closes it at once: a gate that reads stdin (`cat`) must + /// complete on the delivered end-of-file instead of idling on the + /// held-open pipe until the background gate timeout kills it. This + /// pins the `write_stdin(task_id, "", true)` close in the + /// background-gate spawn loop; without it this test reddens after the + /// completion wait finds the job still running. + #[cfg(unix)] + #[tokio::test] + async fn run_verifiers_background_gate_jobs_get_stdin_eof_at_spawn() { + let tmp = tempdir().expect("tempdir"); + let ctx = ToolContext::new(tmp.path()); + let tool = RunVerifiersTool; + let result = tool + .execute( + json!({ + "profile": "auto", + "background": true, + "commands": [ + {"name": "stdin-reader", "program": "cat", "args": []} + ] + }), + &ctx, + ) + .await + .expect("execute"); + + let parsed: RunVerifiersBackgroundOutput = + serde_json::from_str(&result.content).expect("background verifier output json"); + assert!(parsed.success, "result: {}", result.content); + assert_eq!(parsed.started, 1); + let task_id = parsed.jobs[0] + .task_id + .as_deref() + .expect("background task id"); + + let output = wait_for_completed_shell( + &mut ctx.shell_manager.lock().expect("shell manager"), + task_id, + ); + if output.status == ShellStatus::Running { + // Clean up the mutated behavior's wedged job before failing. + let _ = ctx + .shell_manager + .lock() + .expect("shell manager") + .kill(task_id); + } + assert_eq!( + output.status, + ShellStatus::Completed, + "a stdin-reading gate must exit on the end-of-file delivered at spawn" + ); + } }