fix(terminal): write process input off the UI thread (#2183)

A child that stopped reading its input could freeze the whole window, because terminal input was written synchronously on the UI thread. Terminal, Run and debug stdin now share a per-session ordered writer queue bounded to 4 MiB and 1,024 writes. An overflow rejects the whole write and shows an error; closing or killing a session discards pending input without waiting for the child, and an explicit stdin EOF drains accepted writes first.

Refs #2160

crates/ide-core/Cargo.toml

@@ -22,4 +22,4 @@ tempfile.workspace = true
 libc = "0.2"
 
 [target.'cfg(windows)'.dependencies]
-windows-sys = { version = "0.61", features = ["Win32_Globalization", "Win32_Security", "Win32_System_Registry", "Win32_Foundation", "Win32_System_Diagnostics_ToolHelp", "Win32_System_Threading"] }
+windows-sys = { version = "0.61", features = ["Win32_Globalization", "Win32_Security", "Win32_System_Registry", "Win32_Foundation", "Win32_System_Diagnostics_ToolHelp", "Win32_System_Threading", "Win32_System_IO"] }

crates/ide-core/src/input.rs

@@ -0,0 +1,285 @@
+//! Bounded, ordered process input. Only the worker touches the OS writer.
+
+use std::collections::VecDeque;
+use std::io::{self, Write};
+use std::sync::{Arc, Condvar, Mutex};
+use std::thread::JoinHandle;
+use std::time::Duration;
+
+const MAX_BYTES: usize = 4 * 1024 * 1024;
+const MAX_WRITES: usize = 1024;
+const RETRY: Duration = Duration::from_millis(4);
+
+#[derive(Default)]
+struct State {
+    queue: VecDeque<Vec<u8>>,
+    bytes: usize,
+    finished: bool,
+    aborted: bool,
+    error: Option<String>,
+}
+
+/// One worker per pipe/PTY. The byte budget includes the in-flight write.
+/// Rejected writes enqueue nothing. Dropping this handle discards pending input
+/// without joining the worker; `finish` instead drains accepted input before EOF.
+pub struct InputWriter {
+    shared: Arc<(Mutex<State>, Condvar)>,
+    worker: Option<JoinHandle<()>>,
+}
+
+impl InputWriter {
+    /// Unix writers must first be made nonblocking with `nonblocking`.
+    pub fn new(
+        name: String,
+        mut writer: Box<dyn Write + Send>,
+        on_error: impl FnOnce(io::Error) + Send + 'static,
+    ) -> io::Result<Self> {
+        let shared = Arc::new((Mutex::new(State::default()), Condvar::new()));
+        let worker_shared = shared.clone();
+        let worker = std::thread::Builder::new().name(name).spawn(move || {
+            let (lock, wake) = &*worker_shared;
+            let result = (|| {
+                loop {
+                    let data = {
+                        let mut state = lock.lock().unwrap();
+                        while state.queue.is_empty() && !state.finished && !state.aborted {
+                            state = wake.wait(state).unwrap();
+                        }
+                        if state.aborted {
+                            return Ok(());
+                        }
+                        let Some(data) = state.queue.pop_front() else { return Ok(()) };
+                        data
+                    };
+                    let mut offset = 0;
+                    loop {
+                        if lock.lock().unwrap().aborted {
+                            return Ok(());
+                        }
+                        let flushing = offset == data.len();
+                        let result = if !flushing {
+                            writer.write(&data[offset..]).and_then(|n| {
+                                if n == 0 {
+                                    Err(io::ErrorKind::WriteZero.into())
+                                } else {
+                                    offset += n;
+                                    Ok(())
+                                }
+                            })
+                        } else {
+                            writer.flush()
+                        };
+                        match result {
+                            Ok(()) if flushing => break,
+                            Ok(()) => {}
+                            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
+                            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
+                                let state = lock.lock().unwrap();
+                                if state.aborted {
+                                    return Ok(());
+                                }
+                                drop(wake.wait_timeout(state, RETRY).unwrap());
+                            }
+                            Err(e) => return Err(e),
+                        }
+                    }
+                    let len = data.len();
+                    drop(data);
+                    let mut state = lock.lock().unwrap();
+                    state.bytes = state.bytes.saturating_sub(len);
+                }
+            })();
+            if let Err(e) = result {
+                let mut state = lock.lock().unwrap();
+                if state.aborted {
+                    return;
+                }
+                state.error = Some(e.to_string());
+                state.queue.clear();
+                state.bytes = 0;
+                drop(state);
+                on_error(e);
+            }
+        })?;
+        Ok(Self { shared, worker: Some(worker) })
+    }
+
+    pub fn write(&self, data: &[u8]) -> io::Result<()> {
+        let (lock, wake) = &*self.shared;
+        let mut state = lock.lock().unwrap();
+        if let Some(error) = &state.error {
+            return Err(io::Error::new(io::ErrorKind::BrokenPipe, error.clone()));
+        }
+        if state.finished || state.aborted {
+            return Err(io::Error::new(io::ErrorKind::BrokenPipe, "stdin is closed"));
+        }
+        if data.is_empty() {
+            return Ok(());
+        }
+        if data.len() > MAX_BYTES.saturating_sub(state.bytes) || state.queue.len() >= MAX_WRITES {
+            return Err(io::Error::new(
+                io::ErrorKind::WouldBlock,
+                "Input queue is full (4 MiB or 1024 writes); input was not sent.",
+            ));
+        }
+        state.bytes += data.len();
+        state.queue.push_back(data.to_vec());
+        wake.notify_one();
+        Ok(())
+    }
+
+    /// Send EOF after the accepted writes, without waiting for the child to read.
+    pub fn finish(&self) {
+        let (lock, wake) = &*self.shared;
+        lock.lock().unwrap().finished = true;
+        wake.notify_one();
+    }
+
+    pub fn close(&self) {
+        let (lock, wake) = &*self.shared;
+        let mut state = lock.lock().unwrap();
+        state.aborted = true;
+        state.queue.clear();
+        wake.notify_one();
+    }
+}
+
+impl Drop for InputWriter {
+    fn drop(&mut self) {
+        self.close();
+        let Some(worker) = self.worker.take() else { return };
+        #[cfg(windows)]
+        {
+            use std::os::windows::io::AsRawHandle;
+            // A cancellation can race with entry into WriteFile, so retry off
+            // the caller's thread until the worker observes the closed queue.
+            let _ = std::thread::Builder::new().name("openidea-input-cancel".into()).spawn(move || {
+                while !worker.is_finished() {
+                    unsafe {
+                        windows_sys::Win32::System::IO::CancelSynchronousIo(worker.as_raw_handle());
+                    }
+                    std::thread::sleep(RETRY);
+                }
+                let _ = worker.join();
+            });
+        }
+        #[cfg(not(windows))]
+        drop(worker);
+    }
+}
+
+/// PTY clones share file status flags, so readers must also handle WouldBlock.
+#[cfg(unix)]
+pub fn nonblocking(fd: std::os::fd::RawFd) -> io::Result<()> {
+    let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
+    if flags == -1 || unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } == -1 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::mpsc;
+    #[cfg(unix)]
+    use std::time::Instant;
+
+    struct SlowWriter {
+        ready: Arc<AtomicBool>,
+        output: Arc<Mutex<Vec<u8>>>,
+        dropped: mpsc::Sender<()>,
+    }
+
+    impl Write for SlowWriter {
+        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+            if !self.ready.load(Ordering::Acquire) {
+                return Err(io::ErrorKind::WouldBlock.into());
+            }
+            let n = bytes.len().min(7);
+            self.output.lock().unwrap().extend_from_slice(&bytes[..n]);
+            Ok(n)
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl Drop for SlowWriter {
+        fn drop(&mut self) {
+            let _ = self.dropped.send(());
+        }
+    }
+
+    #[test]
+    fn bounds_include_inflight_bytes_and_reject_whole_writes() {
+        let ready = Arc::new(AtomicBool::new(false));
+        let output = Arc::new(Mutex::new(Vec::new()));
+        let (tx, rx) = mpsc::channel();
+        let writer = InputWriter::new(
+            "test-input".into(),
+            Box::new(SlowWriter { ready: ready.clone(), output: output.clone(), dropped: tx }),
+            |e| panic!("{e}"),
+        )
+        .unwrap();
+        let bytes = vec![b'x'; MAX_BYTES];
+        writer.write(&bytes).unwrap();
+        std::thread::sleep(RETRY * 2);
+        assert_eq!(writer.write(b"rejected").unwrap_err().kind(), io::ErrorKind::WouldBlock);
+        writer.close();
+        rx.recv_timeout(Duration::from_secs(1)).expect("worker stops while peer still refuses input");
+        assert!(output.lock().unwrap().is_empty());
+        assert!(writer.write(b"closed").is_err());
+        assert!(writer.shared.0.lock().unwrap().queue.is_empty());
+    }
+
+    #[test]
+    fn partial_writes_preserve_order_and_finish_drains_before_eof() {
+        let ready = Arc::new(AtomicBool::new(false));
+        let output = Arc::new(Mutex::new(Vec::new()));
+        let (tx, rx) = mpsc::channel();
+        let writer = InputWriter::new(
+            "test-input".into(),
+            Box::new(SlowWriter { ready: ready.clone(), output: output.clone(), dropped: tx }),
+            |e| panic!("{e}"),
+        )
+        .unwrap();
+        let paste = "abé".repeat(1024);
+        writer.write(paste.as_bytes()).unwrap();
+        writer.write(b"tail").unwrap();
+        writer.finish();
+        assert!(writer.write(b"after EOF").is_err());
+        ready.store(true, Ordering::Release);
+        rx.recv_timeout(Duration::from_secs(1)).unwrap();
+        assert_eq!(*output.lock().unwrap(), format!("{paste}tail").as_bytes());
+    }
+
+    #[test]
+    fn tiny_writes_are_also_bounded() {
+        let shared = Arc::new((Mutex::new(State::default()), Condvar::new()));
+        let writer = InputWriter { shared, worker: None };
+        for _ in 0..MAX_WRITES {
+            writer.write(b"x").unwrap();
+        }
+        assert_eq!(writer.write(b"x").unwrap_err().kind(), io::ErrorKind::WouldBlock);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn close_interrupts_a_full_os_pipe_while_the_peer_stays_open() {
+        use std::os::fd::AsRawFd;
+        use std::os::unix::net::UnixStream;
+        let (pipe, _peer) = UnixStream::pair().unwrap();
+        nonblocking(pipe.as_raw_fd()).unwrap();
+        let mut writer = InputWriter::new("test-pipe-input".into(), Box::new(pipe), |e| panic!("{e}")).unwrap();
+        writer.write(&vec![b'x'; MAX_BYTES]).unwrap();
+        std::thread::sleep(Duration::from_millis(50));
+        assert!(!writer.worker.as_ref().unwrap().is_finished());
+        let started = Instant::now();
+        writer.close();
+        writer.worker.take().unwrap().join().unwrap();
+        assert!(started.elapsed() < Duration::from_millis(250));
+    }
+}

crates/ide-core/src/lib.rs

@@ -10,6 +10,7 @@ pub mod event;
 pub mod go;
 pub mod ids;
 pub mod ignored;
+pub mod input;
 pub mod inspection_settings;
 pub mod ipc_path;
 pub mod java_runtime;

crates/ide-dap/src/session.rs

@@ -600,15 +600,14 @@ fn on_reverse_request(mgr: &Arc<DebugManager>, s: &Arc<Session>, seq: i64, comma
 
 /// A `runInTerminal` program and the pseudo-terminal it runs in.
 pub(crate) struct ProgramTerminal {
-    writer: Box<dyn std::io::Write + Send>,
+    writer: ide_core::input::InputWriter,
     child: Box<dyn portable_pty::Child + Send + Sync>,
     _master: Box<dyn portable_pty::MasterPty + Send>,
 }
 
 impl ProgramTerminal {
     pub(crate) fn write(&mut self, text: &str) -> std::io::Result<()> {
-        self.writer.write_all(text.as_bytes())?;
-        self.writer.flush()
+        self.writer.write(text.as_bytes())
     }
 }
 
@@ -628,6 +627,10 @@ pub(crate) fn run_in_terminal(s: &Arc<Session>, args: &Value) -> Result<Value, S
     let pair = native_pty_system()
         .openpty(PtySize { rows: 24, cols: 200, pixel_width: 0, pixel_height: 0 })
         .map_err(|e| format!("cannot open a terminal for {program}: {e}"))?;
+    #[cfg(unix)]
+    if let Some(fd) = pair.master.as_raw_fd() {
+        ide_core::input::nonblocking(fd).map_err(|e| format!("terminal of {program}: {e}"))?;
+    }
     let mut cmd = CommandBuilder::new(program);
     // CodeLLDB's launcher clears the terminal first; the console keeps the session's earlier output.
     cmd.args(rest.iter().filter(|a| *a != "--clear-screen"));
@@ -653,10 +656,18 @@ pub(crate) fn run_in_terminal(s: &Arc<Session>, args: &Value) -> Result<Value, S
             };
         }
     }
-    let child = pair.slave.spawn_command(cmd).map_err(|e| format!("failed to start {program}: {e}"))?;
+    let mut child = pair.slave.spawn_command(cmd).map_err(|e| format!("failed to start {program}: {e}"))?;
     drop(pair.slave);
     let reader = pair.master.try_clone_reader().map_err(|e| format!("terminal of {program}: {e}"))?;
     let writer = pair.master.take_writer().map_err(|e| format!("terminal of {program}: {e}"))?;
+    let writer = ide_core::input::InputWriter::new(format!("openidea-debug-input-{}", s.id.0), writer, |e| {
+        tracing::warn!("debug input: {e}");
+    })
+    .inspect_err(|_| {
+        let _ = child.kill();
+        let _ = child.wait();
+    })
+    .map_err(|e| format!("terminal input worker: {e}"))?;
     let pid = child.process_id();
     forward_terminal_output(s, reader);
     if let Some(mut previous) = s.program.lock().replace(ProgramTerminal { writer, child, _master: pair.master }) {
@@ -675,8 +686,19 @@ fn forward_terminal_output(s: &Session, mut reader: Box<dyn std::io::Read + Send
         let mut pending: Vec<u8> = Vec::new();
         loop {
             match reader.read(&mut buf) {
-                Ok(0) | Err(_) => break,
+                Ok(0) => break,
                 Ok(n) => pending.extend_from_slice(&buf[..n]),
+                Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                    std::thread::sleep(std::time::Duration::from_millis(4));
+                    continue;
+                }
+                #[cfg(unix)]
+                Err(e) if e.raw_os_error() == Some(libc::EIO) => break,
+                Err(e) => {
+                    tracing::warn!(%session, "debug terminal read: {e}");
+                    break;
+                }
             }
             // Keep incomplete UTF-8 sequences, and a `\r` that may start a `\r\n`, for the next chunk.
             let mut valid = match std::str::from_utf8(&pending) {

crates/ide-dap/src/tests.rs

@@ -554,6 +554,22 @@ async fn run_in_terminal_program_reads_typed_input() {
     assert!(h.manager.send_input(session, "x\n").is_err());
 }
 
+#[cfg(unix)]
+#[tokio::test]
+async fn blocked_program_input_does_not_hold_up_shutdown() {
+    let h = harness(Script {
+        run_in_terminal: Some(vec!["sh".into(), "-c".into(), "stty raw -echo; printf READY; exec sleep 60".into()]),
+        ..Script::default()
+    });
+    let session = h.start(config()).await.unwrap();
+    h.wait_event(|e| matches!(e, DebugEvent::Output { output, .. } if output.contains("READY"))).await;
+    let started = std::time::Instant::now();
+    h.manager.send_input(session, &"x".repeat(512 * 1024)).unwrap();
+    assert!(started.elapsed() < std::time::Duration::from_millis(250));
+    h.manager.stop(session).await.unwrap();
+    assert!(h.manager.send_input(session, "later").is_err());
+}
+
 #[test]
 fn launch_arguments_merge_and_defaults() {
     let mut c = config();

crates/ide-run/src/process.rs

@@ -51,7 +51,7 @@ struct Entry {
     config_id: String,
     name: String,
     pid: Mutex<Option<u32>>,
-    stdin: Mutex<Option<Box<dyn Write + Send>>>,
+    stdin: Mutex<Option<ide_core::input::InputWriter>>,
     /// The PTY master of a terminal-emulated run (resize; dropped on exit).
     pty: Option<Mutex<Option<Box<dyn MasterPty + Send>>>>,
     stop: Option<StopCommand>,
@@ -422,10 +422,23 @@ impl ProcessManager {
         })?;
 
         let id = RunId::next();
-        let stdin: Option<Box<dyn Write + Send>> = match child.stdin.take() {
-            Some(pipe) if !redirected => Some(Box::new(pipe)),
-            _ => None,
-        };
+        let stdin = (|| -> std::io::Result<_> {
+            Ok(match child.stdin.take() {
+                Some(pipe) if !redirected => {
+                    #[cfg(unix)]
+                    {
+                        use std::os::fd::AsRawFd;
+                        ide_core::input::nonblocking(pipe.as_raw_fd())?;
+                    }
+                    Some(input_writer(id, Box::new(pipe))?)
+                }
+                _ => None,
+            })
+        })()
+        .inspect_err(|_| {
+            let _ = child.kill();
+            let _ = child.wait();
+        })?;
         let entry = Arc::new(Entry {
             project_root: root.to_path_buf(),
             id,
@@ -524,6 +537,10 @@ impl ProcessManager {
         for (k, v) in &env.vars {
             cmd.env(k, v);
         }
+        #[cfg(unix)]
+        if let Some(fd) = pair.master.as_raw_fd() {
+            ide_core::input::nonblocking(fd)?;
+        }
         let mut child = pair.slave.spawn_command(cmd).map_err(|e| {
             let text = e.to_string();
             if text.contains("No such file") || text.contains("not found") {
@@ -538,6 +555,10 @@ impl ProcessManager {
         let writer = pair.master.take_writer().map_err(|e| IdeError::internal(format!("pty writer: {e}")))?;
 
         let id = RunId::next();
+        let writer = input_writer(id, writer).inspect_err(|_| {
+            let _ = child.kill();
+            let _ = child.wait();
+        })?;
         let entry = Arc::new(Entry {
             project_root: root.to_path_buf(),
             id,
@@ -609,16 +630,9 @@ impl ProcessManager {
         if entry.exited.load(Ordering::Acquire) {
             return Err(IdeError::invalid("process has exited"));
         }
-        let mut stdin = entry.stdin.lock();
-        let pipe = stdin.as_mut().ok_or_else(|| IdeError::invalid("stdin is closed"))?;
-        let res = pipe.write_all(data.as_bytes()).and_then(|_| pipe.flush());
-        if let Err(e) = res {
-            if e.kind() == std::io::ErrorKind::BrokenPipe {
-                *stdin = None;
-            }
-            return Err(e.into());
-        }
-        Ok(())
+        let stdin = entry.stdin.lock();
+        let pipe = stdin.as_ref().ok_or_else(|| IdeError::invalid("stdin is closed"))?;
+        pipe.write(data.as_bytes()).map_err(Into::into)
     }
 
     /// Closes the process's stdin (sends EOF; on a terminal, types ^D).
@@ -627,7 +641,9 @@ impl ProcessManager {
         if entry.pty.is_some() {
             return self.write_stdin(id, "\u{4}");
         }
-        entry.stdin.lock().take();
+        if let Some(stdin) = entry.stdin.lock().as_ref() {
+            stdin.finish();
+        }
         Ok(())
     }
 
@@ -769,7 +785,23 @@ fn run_followup(
     (entry.output)(OutputChunk::System { text: format!("{}\n", inv.display) });
     let mut child = cmd.spawn()?;
     *pid = Some(child.id());
-    *entry.stdin.lock() = child.stdin.take().map(|p| Box::new(p) as Box<dyn Write + Send>);
+    *entry.stdin.lock() = (|| -> std::io::Result<_> {
+        Ok(match child.stdin.take() {
+            Some(pipe) => {
+                #[cfg(unix)]
+                {
+                    use std::os::fd::AsRawFd;
+                    ide_core::input::nonblocking(pipe.as_raw_fd())?;
+                }
+                Some(input_writer(entry.id, Box::new(pipe))?)
+            }
+            None => None,
+        })
+    })()
+    .inspect_err(|_| {
+        let _ = child.kill();
+        let _ = child.wait();
+    })?;
     drop(pid);
     let readers = (|| {
         if let Some(out) = child.stdout.take() {
@@ -963,7 +995,13 @@ fn spawn_reader(
                     }
                 }
                 Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
-                Err(_) => break,
+                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(BATCH_INTERVAL),
+                #[cfg(unix)]
+                Err(e) if e.raw_os_error() == Some(libc::EIO) => break,
+                Err(e) => {
+                    tracing::warn!(%id, "run read: {e}");
+                    break;
+                }
             }
         }
     })?;
@@ -1180,6 +1218,12 @@ fn signal_number(_description: &str) -> Option<i32> {
     None
 }
 
+fn input_writer(id: RunId, writer: Box<dyn std::io::Write + Send>) -> std::io::Result<ide_core::input::InputWriter> {
+    ide_core::input::InputWriter::new(format!("openidea-run-input-{}", id.0), writer, move |e| {
+        tracing::warn!(%id, "run input: {e}");
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/ide-run/tests/process.rs

@@ -150,6 +150,24 @@ fn stdin_echo_with_cat() {
     assert_eq!(wait_exit(&h, id, Duration::from_secs(10)), Some(0));
 }
 
+#[test]
+fn full_stdin_does_not_block_input_or_stop_for_pipes_and_ptys() {
+    for pty in [false, true] {
+        let h = harness();
+        let mut config =
+            cfg("t:blocked-input", "/bin/sh", &["-c", "stty raw -echo 2>/dev/null; printf READY; exec sleep 60"]);
+        config.extras.emulate_terminal = pty;
+        let (id, chunks) = start(&h, &config);
+        wait_for(&chunks, "READY");
+        let started = Instant::now();
+        h.pm.write_stdin(id, &"x".repeat(512 * 1024)).unwrap();
+        assert!(started.elapsed() < Duration::from_millis(250));
+        h.pm.stop(id).unwrap();
+        h.pm.stop(id).unwrap();
+        wait_exit(&h, id, Duration::from_secs(10));
+    }
+}
+
 #[test]
 fn missing_command_and_cwd_are_errors() {
     let h = harness();

crates/ide-terminal/src/lib.rs

@@ -11,7 +11,7 @@ use portable_pty::{ChildKiller, CommandBuilder, MasterPty, PtySize, native_pty_s
 use serde::{Deserialize, Serialize};
 use specta::Type;
 use std::collections::{BTreeMap, HashMap};
-use std::io::{Read, Write};
+use std::io::Read;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -56,8 +56,9 @@ struct Session {
     /// The window that opened it: closing that window's project closes its terminals.
     key: String,
     info: TerminalInfo,
+    // Cancel input before dropping the master (ClosePseudoConsole on Windows).
+    writer: ide_core::input::InputWriter,
     master: Mutex<Box<dyn MasterPty + Send>>,
-    writer: Mutex<Box<dyn Write + Send>>,
     killer: Mutex<Box<dyn ChildKiller + Send + Sync>>,
     exited: Arc<AtomicBool>,
     started: Instant,
@@ -161,7 +162,12 @@ impl TerminalManager {
         }
         apply_powershell_env(&shell, &mut cmd);
 
-        let child = pair
+        #[cfg(unix)]
+        if let Some(fd) = pair.master.as_raw_fd() {
+            ide_core::input::nonblocking(fd)?;
+        }
+
+        let mut child = pair
             .slave
             .spawn_command(cmd)
             .map_err(|e| IdeError::unavailable(format!("failed to start shell `{shell}`: {e}")))?;
@@ -174,11 +180,18 @@ impl TerminalManager {
         let id = TerminalId::next();
         let info = TerminalInfo { id, title: shell_title(&shell), pid: child.process_id() };
         let exited = Arc::new(AtomicBool::new(false));
+        let writer = ide_core::input::InputWriter::new(format!("openidea-term-write-{}", id.0), writer, move |e| {
+            tracing::warn!(%id, "terminal input: {e}");
+        })
+        .inspect_err(|_| {
+            let _ = child.kill();
+            let _ = child.wait();
+        })?;
         let session = Arc::new(Session {
             key,
             info: info.clone(),
             master: Mutex::new(pair.master),
-            writer: Mutex::new(writer),
+            writer,
             killer: Mutex::new(child.clone_killer()),
             exited: exited.clone(),
             started: Instant::now(),
@@ -209,9 +222,7 @@ impl TerminalManager {
             let ms = session.started.elapsed().as_millis();
             tracing::info!(%id, ms, bytes = data.len(), cursor_report, "terminal first input");
         }
-        let mut w = session.writer.lock();
-        w.write_all(data.as_bytes())?;
-        w.flush()?;
+        session.writer.write(data.as_bytes())?;
         Ok(())
     }
 
@@ -364,6 +375,7 @@ mod win {
 
 /// Hangs up the shell (and its process group), escalating to SIGKILL if it lingers.
 fn terminate(session: &Session) {
+    session.writer.close();
     if session.exited.load(Ordering::Acquire) {
         return;
     }
@@ -407,8 +419,14 @@ fn spawn_reader(id: TerminalId, mut reader: Box<dyn Read + Send>, tx: mpsc::Send
                     }
                 }
                 Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(BATCH_INTERVAL),
                 // EIO is how Linux/macOS report "slave side closed".
-                Err(_) => break,
+                #[cfg(unix)]
+                Err(e) if e.raw_os_error() == Some(libc::EIO) => break,
+                Err(e) => {
+                    tracing::warn!(%id, "terminal read: {e}");
+                    break;
+                }
             }
         }
     })?;
@@ -809,6 +827,40 @@ mod tests {
         all
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn blocked_input_returns_promptly_and_kill_does_not_wait_for_it() {
+        let h = harness();
+        let (info, output) = spawn_collect(&h, sh("stty raw -echo; printf READY; exec sleep 60"));
+        read_until(&output, "READY");
+        let started = Instant::now();
+        h.manager.write(info.id, &"x".repeat(512 * 1024)).unwrap();
+        h.manager.write(info.id, "\u{3}").unwrap();
+        assert!(started.elapsed() < Duration::from_millis(250));
+        std::thread::sleep(Duration::from_millis(50));
+        let started = Instant::now();
+        h.manager.kill(info.id).unwrap();
+        assert!(started.elapsed() < Duration::from_millis(250));
+        wait_exit(&h, info.id);
+        assert!(h.manager.write(info.id, "later").is_err());
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn input_keeps_large_pastes_and_following_keystrokes_in_order() {
+        let h = harness();
+        let (info, output) = spawn_collect(&h, sh("stty raw -echo; printf READY; sleep 1; exec cat"));
+        read_until(&output, "READY");
+        let paste = "abé".repeat(128 * 1024);
+        let started = Instant::now();
+        h.manager.write(info.id, &paste).unwrap();
+        h.manager.write(info.id, "END-OF-INPUT").unwrap();
+        assert!(started.elapsed() < Duration::from_millis(250));
+        let actual = read_until(&output, "END-OF-INPUT");
+        h.manager.kill(info.id).unwrap();
+        assert_eq!(actual, format!("{paste}END-OF-INPUT"));
+    }
+
     #[test]
     fn program_sessions_never_fall_back_to_another_executable_or_cwd() {
         let h = harness();

src/features/run/store.ts

@@ -1257,7 +1257,7 @@ export async function sendStdin(sessionId: string, text: string): Promise<void>
   } catch (e) {
     enqueue(sessionId, {
       kind: "system",
-      text: tx`Cannot write to stdin: ${e instanceof Error ? e.message : String(e)}\n`,
+      text: tx`Cannot write to stdin: ${e instanceof Error ? tx(e.message) : String(e)}\n`,
     });
   }
 }
@@ -1266,7 +1266,12 @@ export async function sendStdin(sessionId: string, text: string): Promise<void>
 export function writeTerminal(sessionId: string, data: string) {
   const session = getSession(sessionId);
   if (!session || session.runId === null || session.status !== "running") return;
-  void call(commands.runStdin(session.runId, data)).catch(() => undefined);
+  void call(commands.runStdin(session.runId, data)).catch((e) => {
+    enqueue(sessionId, {
+      kind: "system",
+      text: tx`Cannot write to stdin: ${e instanceof Error ? tx(e.message) : String(e)}\n`,
+    });
+  });
 }
 
 /** The terminal console's size in cells; the process's terminal follows it. */

src/features/terminal/session.test.ts

@@ -16,6 +16,7 @@ const ipc = vi.hoisted(() => ({
     termKill: vi.fn(),
     termResize: vi.fn(),
     termIsBusy: vi.fn(),
+    termWrite: vi.fn(),
   },
 }));
 const workbench = vi.hoisted(() => ({
@@ -122,6 +123,27 @@ describe("resolveStartCwd", () => {
 describe("terminal start directory fallback", () => {
   beforeEach(() => vi.clearAllMocks());
 
+  it("shows rejected terminal input instead of silently discarding it", async () => {
+    ipc.commands.termSpawn.mockResolvedValue({ id: 9, title: "zsh" });
+    ipc.commands.termWrite.mockRejectedValueOnce(new Error("input queue full"));
+    const session = new TerminalSession({
+      key: "s-input-error",
+      cwd: "/repo",
+      onChange: () => undefined,
+      onFocus: () => undefined,
+    });
+    await session.start();
+    session.input("paste");
+    await vi.waitFor(() =>
+      expect(workbench.notify).toHaveBeenCalledWith({
+        level: "error",
+        title: "Cannot write to terminal",
+        body: "input queue full",
+      }),
+    );
+    session.dispose();
+  });
+
   it("spawns a restored session in the project root when its saved directory is gone", async () => {
     ipc.commands.fsEntryKind.mockRejectedValue(new Error("gone"));
     ipc.commands.termSpawn.mockResolvedValue({ id: 7, title: "zsh" });

src/features/terminal/session.ts

@@ -9,6 +9,7 @@ import { SearchAddon, type ISearchOptions } from "@xterm/addon-search";
 import "@xterm/xterm/css/xterm.css";
 import { Channel } from "@tauri-apps/api/core";
 import { call, commands, isTauri, type TerminalId } from "@/platform/ipc";
+import { tx } from "@/platform/i18n";
 import { editorService, notifications, registerFontZoomSurface } from "@/workbench/contributions";
 import { fitScheduler, settledResize } from "./fit";
 import { installTerminalDrop } from "./drop";
@@ -658,14 +659,20 @@ export class TerminalSession {
       this.pendingInput += data;
       return;
     }
-    void commands.termWrite(this.ptyId, data);
+    void call(commands.termWrite(this.ptyId, data)).catch((error) => {
+      notifications().notify({
+        level: "error",
+        title: tx("Cannot write to terminal"),
+        body: error instanceof Error ? tx(error.message) : String(error),
+      });
+    });
   }
 
   /** Sends keystrokes buffered while the PTY was starting. */
   private flushPendingInput() {
     const queued = this.pendingInput;
     this.pendingInput = "";
-    if (queued && this.ptyId !== null) void commands.termWrite(this.ptyId, queued);
+    if (queued && this.ptyId !== null) this.input(queued);
   }
 
   /** Writes input immediately (shortcuts translated outside xterm). */

src/platform/locales/text/en.json

@@ -1,4 +1,6 @@
 {
+  "Input queue is full (4 MiB or 1024 writes); input was not sent.": "Input queue is full (4 MiB or 1024 writes); input was not sent.",
+  "Cannot write to terminal": "Cannot write to terminal",
   "Docker": "Docker",
   "Shell": "Shell",
   "C/C++": "C/C++",

src/platform/locales/text/zh.json

@@ -1,4 +1,6 @@
 {
+  "Input queue is full (4 MiB or 1024 writes); input was not sent.": "输入队列已满（4 MiB 或 1024 次写入）；本次输入未发送。",
+  "Cannot write to terminal": "无法写入终端",
   "Docker": "Docker",
   "Shell": "Shell",
   "C/C++": "C/C++",