fix: handle short writes in write_to_tty_stdin to prevent paste truncation (#4818)

* fix: handle short writes in write_to_tty_stdin to prevent paste truncation

On macOS the kernel PTY buffer is only 4-16 KiB, so a single write()
call for a large paste returns a short write count rather than blocking
until all bytes are accepted. The current code returns this partial
count without retrying, silently dropping the remaining bytes.

This wraps the write in a retry loop that handles:
- Short writes (advances the buffer offset and retries)
- EINTR (retries immediately)
- EAGAIN (sleeps 1ms then retries)

Without this fix, pasting ~1 KiB+ of text into a terminal pane loses
everything beyond the first kernel-buffer-full.

Includes a test that exercises the fix using a real PTY pair with
O_NONBLOCK set on the master fd to force short writes.

* fix: cap EAGAIN retries at 50 to prevent infinite busy-wait

Per reviewer feedback, add a retry counter that caps consecutive EAGAIN
retries at 50 and logs an error if the limit is reached. The counter
resets on any successful write progress, so it only triggers when the
PTY is genuinely stuck.

* fix: use poll(POLLOUT) instead of sleep-retry loop for EAGAIN handling

Replace the 50-retry sleep(1ms) busy-wait with poll(POLLOUT, 5s timeout).
This properly waits for fd writability without CPU spinning and returns
an error on timeout instead of silently returning Ok(partial) — which
the caller (pty_writer_main) would ignore, causing silent data loss.

Tests:
- write_to_tty_stdin_delivers_all_bytes: remove flaky reader delay,
  use concurrent reader, move master close after reader join to avoid
  macOS kernel buffer flush race
- write_all_to_fd_errors_on_stuck_pty: new test verifying timeout
  returns Err on stuck PTY (uses 100ms timeout for speed)

* fix: re-queue partial PTY writes instead of blocking PtyWriter thread

Replace the poll(POLLOUT) blocking loop with a non-blocking re-queue
approach as suggested in review. On partial write or EAGAIN:

- write_to_tty_stdin now drains as much as the kernel accepts in one
  pass and returns the count (may be 0 on EAGAIN)
- PtyWriter re-queues unwritten bytes back onto its own channel,
  letting other terminals' writes proceed between retries
- Retry count tracks consecutive zero-progress attempts (reset on
  any forward progress), drops bytes after 50 retries with log::error
- yield_now() before re-queue on zero progress to avoid tight spin
- Guard against write() returning Ok(0) on non-empty buffer
- Wire up PtyWriter's self-sender (was None, now Some(&to_pty_writer))

* fix: per-terminal write buffering to handle short writes and EAGAIN

Replace re-queue approach with per-terminal pending write buffers
(HashMap<u32, VecDeque<PendingWrite>>). Uses recv_timeout(10ms) when
pending data exists, blocking recv when idle. 10MB per-terminal cap
drops buffer on overflow. Terminal errors clear the queue.

Adds Bus::recv_timeout via crossbeam Select::select_timeout.
Removes tcdrain from write path.

* use safe accessor methods and saturating substitutions

* add tests for ptywriter

* rustfmt

* fix tests by not assuming kernel buffer sizes

* rustfmt

* add pr

---------

Co-authored-by: Aram Drevekenin <aram@poor.dev>

5 files changed, 650 insertions, 52 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ae575515..1c35384a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -79,6 +79,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 * fix: allow specifying --ip and --port with `zellij web --status` rather than relying on those in the config (https://github.com/zellij-org/zellij/pull/4825)
 * fix: make sure clients are focused when atomically showing floating panes (https://github.com/zellij-org/zellij/pull/4834)
 * feat: allow configuring focus_follows_mouse and mouse_click_through (https://github.com/zellij-org/zellij/pull/4836)
+* fix: handle very large writes to the pty (https://github.com/zellij-org/zellij/pull/4818)
 
 
 ## [0.43.1] - 2025-08-08
diff --git a/zellij-server/src/os_input_output_unix.rs b/zellij-server/src/os_input_output_unix.rs
index d698fd4c..052636b8 100644
--- a/zellij-server/src/os_input_output_unix.rs
+++ b/zellij-server/src/os_input_output_unix.rs
@@ -278,6 +278,26 @@ pub(crate) struct UnixPtyBackend {
     terminal_id_to_raw_fd: Arc<Mutex<BTreeMap<u32, Option<RawFd>>>>,
 }
 
+/// Try to write as many bytes from `buf` as possible to `fd` without blocking.
+///
+/// Loops on successful short writes and EINTR to drain as much as the kernel
+/// will accept. On EAGAIN (fd buffer full), stops and returns how many bytes
+/// were written so far (which may be 0). The caller is expected to re-queue
+/// any unwritten remainder.
+fn try_write_to_fd(fd: RawFd, buf: &[u8]) -> Result<usize> {
+    let mut written = 0;
+    while written < buf.len() {
+        match unistd::write(fd, &buf[written..]) {
+            Ok(0) => break, // fd returned 0 on non-empty buf; treat like EAGAIN
+            Ok(n) => written += n,
+            Err(nix::errno::Errno::EINTR) => continue,
+            Err(nix::errno::Errno::EAGAIN) => break,
+            Err(e) => return Err(e.into()),
+        }
+    }
+    Ok(written)
+}
+
 impl UnixPtyBackend {
     pub fn new() -> Result<Self, io::Error> {
         let current_termios = termios::tcgetattr(0).ok();
@@ -358,16 +378,20 @@ impl UnixPtyBackend {
     pub fn write_to_tty_stdin(&self, terminal_id: u32, buf: &[u8]) -> Result<usize> {
         let err_context = || format!("failed to write to stdin of TTY ID {}", terminal_id);
 
-        match self
+        let fd = match self
             .terminal_id_to_raw_fd
             .lock()
             .to_anyhow()
             .with_context(err_context)?
             .get(&terminal_id)
         {
-            Some(Some(fd)) => unistd::write(*fd, buf).with_context(err_context),
-            _ => Err(anyhow!("could not find raw file descriptor")).with_context(err_context),
-        }
+            Some(Some(fd)) => *fd,
+            _ => {
+                return Err(anyhow!("could not find raw file descriptor")).with_context(err_context)
+            },
+        };
+
+        try_write_to_fd(fd, buf).with_context(err_context)
     }
 
     pub fn tcdrain(&self, terminal_id: u32) -> Result<()> {
@@ -426,3 +450,110 @@ impl UnixPtyBackend {
             .or(Some(0))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nix::fcntl::{fcntl, FcntlArg, OFlag};
+    use nix::sys::termios;
+    use std::io::Read;
+
+    /// Verify that `try_write_to_fd` writes as many bytes as the kernel will
+    /// accept in one pass and returns a partial count (not an error) when the
+    /// PTY buffer fills up.
+    ///
+    /// A concurrent reader drains the slave side so some bytes are accepted.
+    /// The key assertion: the function returns Ok(n) where n <= buf.len(),
+    /// and the caller (PtyWriter) is responsible for re-queuing the rest.
+    #[test]
+    fn try_write_to_fd_returns_partial_on_full_buffer() {
+        let pty = openpty(None, &None).expect("openpty failed");
+
+        let mut attrs = termios::tcgetattr(pty.slave).expect("tcgetattr failed");
+        termios::cfmakeraw(&mut attrs);
+        termios::tcsetattr(pty.slave, termios::SetArg::TCSANOW, &attrs).expect("tcsetattr failed");
+
+        // O_NONBLOCK so write() returns EAGAIN instead of blocking
+        let flags = fcntl(pty.master, FcntlArg::F_GETFL).expect("F_GETFL");
+        let mut oflags = OFlag::from_bits_truncate(flags);
+        oflags.insert(OFlag::O_NONBLOCK);
+        fcntl(pty.master, FcntlArg::F_SETFL(oflags)).expect("F_SETFL");
+
+        // Fill most of the buffer, leaving some space
+        let chunk = vec![0x42u8; 1024];
+        let mut total_filled = 0;
+        loop {
+            match super::try_write_to_fd(pty.master, &chunk) {
+                Ok(0) => break,
+                Ok(n) => total_filled += n,
+                Err(e) => panic!("unexpected error filling buffer: {e}"),
+            }
+        }
+        assert!(
+            total_filled > 0,
+            "should have written some bytes to fill buffer"
+        );
+
+        // Read a small amount from the slave to free partial space
+        let mut drain = vec![0u8; 512];
+        let slave_file = unsafe { std::fs::File::from_raw_fd(pty.slave) };
+        let mut slave_reader = std::io::BufReader::new(&slave_file);
+        let drained = slave_reader.read(&mut drain).expect("slave read failed");
+        assert!(drained > 0, "should have drained some bytes");
+        // Prevent File from closing the slave fd — we close it manually below
+        std::mem::forget(slave_file);
+
+        // Now write more than the freed space — should get a partial write
+        let size = 128 * 1024;
+        let data: Vec<u8> = (0..size).map(|i| (i % 256) as u8).collect();
+        let written = super::try_write_to_fd(pty.master, &data)
+            .expect("try_write_to_fd should not error on EAGAIN");
+
+        assert!(
+            written > 0 && written < size,
+            "expected partial write, got {written}/{size}",
+        );
+
+        unsafe {
+            libc::close(pty.master);
+            libc::close(pty.slave);
+        }
+    }
+
+    /// Verify that `try_write_to_fd` returns Ok(0) — not an error — when the
+    /// fd is completely full and cannot accept any bytes at all.
+    #[test]
+    fn try_write_to_fd_returns_zero_on_stuck_pty() {
+        let pty = openpty(None, &None).expect("openpty failed");
+
+        let mut attrs = termios::tcgetattr(pty.slave).expect("tcgetattr failed");
+        termios::cfmakeraw(&mut attrs);
+        termios::tcsetattr(pty.slave, termios::SetArg::TCSANOW, &attrs).expect("tcsetattr failed");
+
+        let flags = fcntl(pty.master, FcntlArg::F_GETFL).expect("F_GETFL");
+        let mut oflags = OFlag::from_bits_truncate(flags);
+        oflags.insert(OFlag::O_NONBLOCK);
+        fcntl(pty.master, FcntlArg::F_SETFL(oflags)).expect("F_SETFL");
+
+        // Fill the buffer completely — keep writing until we get Ok(0)
+        let fill = vec![0x42u8; 1024];
+        loop {
+            match super::try_write_to_fd(pty.master, &fill) {
+                Ok(0) => break,
+                Ok(_) => continue,
+                Err(e) => panic!("unexpected error filling buffer: {e}"),
+            }
+        }
+
+        // Now the buffer is full — next write should return Ok(0)
+        let written = super::try_write_to_fd(pty.master, &[0x01, 0x02, 0x03])
+            .expect("try_write_to_fd should not error on EAGAIN");
+
+        assert_eq!(written, 0, "expected zero bytes written on full buffer");
+
+        unsafe {
+            libc::close(pty.master);
+            libc::close(pty.slave);
+        }
+    }
+}
diff --git a/zellij-server/src/pty_writer.rs b/zellij-server/src/pty_writer.rs
index 7cdfaadc..3de9906c 100644
--- a/zellij-server/src/pty_writer.rs
+++ b/zellij-server/src/pty_writer.rs
@@ -1,3 +1,7 @@
+use std::collections::{HashMap, VecDeque};
+use std::time::Duration;
+
+use zellij_utils::channels;
 use zellij_utils::errors::{prelude::*, ContextType, PtyWriteContext};
 
 use crate::route::NotificationEnd;
@@ -28,63 +32,151 @@ impl From<&PtyWriteInstruction> for PtyWriteContext {
     }
 }
 
+/// Maximum pending bytes per terminal before dropping the buffer.
+///
+/// Exposed (but not `pub`) so tests in this module can reference it.
+#[allow(dead_code)]
+const MAX_PENDING_BYTES: usize = 10 * 1024 * 1024;
+
+/// When pending writes exist, poll for new instructions with this timeout
+/// before retrying the drain. Keeps the thread responsive to new instructions
+/// while still retrying stuck terminals.
+const PENDING_DRAIN_TIMEOUT: Duration = Duration::from_millis(10);
+
+/// A chunk of bytes waiting to be written to a terminal's stdin.
+struct PendingWrite {
+    bytes: Vec<u8>,
+    offset: usize,
+    _completion: Option<NotificationEnd>,
+}
+
 pub(crate) fn pty_writer_main(bus: Bus<PtyWriteInstruction>) -> Result<()> {
     let err_context = || "failed to write to pty".to_string();
+    let mut pending: HashMap<u32, VecDeque<PendingWrite>> = HashMap::new();
 
     loop {
-        let (event, mut err_ctx) = bus.recv().with_context(err_context)?;
-        err_ctx.add_call(ContextType::PtyWrite((&event).into()));
+        // If we have pending writes, use a short timeout so we can keep
+        // draining. Otherwise, block until a new instruction arrives.
+        let has_pending = pending.values().any(|q| !q.is_empty());
+        let event = if has_pending {
+            match bus.recv_timeout(PENDING_DRAIN_TIMEOUT) {
+                Ok(pair) => Some(pair),
+                Err(channels::RecvTimeoutError::Timeout) => None,
+                Err(channels::RecvTimeoutError::Disconnected) => return Ok(()),
+            }
+        } else {
+            Some(bus.recv().with_context(err_context)?)
+        };
+
         let mut os_input = bus
             .os_input
             .clone()
             .context("no OS input API found")
             .with_context(err_context)?;
-        match event {
-            PtyWriteInstruction::Write(bytes, terminal_id, _completion) => {
-                os_input
-                    .write_to_tty_stdin(terminal_id, &bytes)
-                    .with_context(err_context)
-                    .non_fatal();
-                os_input
-                    .tcdrain(terminal_id)
-                    .with_context(err_context)
-                    .non_fatal();
-            },
-            PtyWriteInstruction::ResizePty(
-                terminal_id,
-                columns,
-                rows,
-                width_in_pixels,
-                height_in_pixels,
-            ) => {
-                os_input
-                    .set_terminal_size_using_terminal_id(
-                        terminal_id,
-                        columns,
-                        rows,
-                        width_in_pixels,
-                        height_in_pixels,
-                    )
-                    .with_context(err_context)
-                    .non_fatal();
-            },
-            PtyWriteInstruction::StartCachingResizes => {
-                // we do this because there are some logic traps inside the screen/tab/layout code
-                // the cause multiple resizes to be sent to the pty - while the last one is always
-                // the correct one, many programs and shells debounce those (I guess due to the
-                // trauma of dealing with GUI resizes of the controlling terminal window), and this
-                // then causes glitches and missing redraws
-                // so we do this to play nice and always only send the last resize instruction to
-                // each pane
-                // the logic for this happens in the main Screen event loop
-                os_input.cache_resizes();
-            },
-            PtyWriteInstruction::ApplyCachedResizes => {
-                os_input.apply_cached_resizes();
-            },
-            PtyWriteInstruction::Exit => {
-                return Ok(());
-            },
+
+        // Process the instruction if we received one
+        if let Some((event, mut err_ctx)) = event {
+            err_ctx.add_call(ContextType::PtyWrite((&event).into()));
+            match event {
+                PtyWriteInstruction::Write(bytes, terminal_id, completion) => {
+                    let queue = pending.entry(terminal_id).or_default();
+                    let queued: usize = queue
+                        .iter()
+                        .map(|w| w.bytes.len().saturating_sub(w.offset))
+                        .sum();
+                    if queued + bytes.len() > MAX_PENDING_BYTES {
+                        log::error!(
+                            "dropping write buffer for terminal {} \
+                             ({} + {} bytes exceeds {} limit)",
+                            terminal_id,
+                            queued,
+                            bytes.len(),
+                            MAX_PENDING_BYTES,
+                        );
+                        queue.clear();
+                    } else {
+                        queue.push_back(PendingWrite {
+                            bytes,
+                            offset: 0,
+                            _completion: completion,
+                        });
+                    }
+                },
+                PtyWriteInstruction::ResizePty(
+                    terminal_id,
+                    columns,
+                    rows,
+                    width_in_pixels,
+                    height_in_pixels,
+                ) => {
+                    os_input
+                        .set_terminal_size_using_terminal_id(
+                            terminal_id,
+                            columns,
+                            rows,
+                            width_in_pixels,
+                            height_in_pixels,
+                        )
+                        .with_context(err_context)
+                        .non_fatal();
+                },
+                PtyWriteInstruction::StartCachingResizes => {
+                    // we do this because there are some logic traps inside the screen/tab/layout code
+                    // the cause multiple resizes to be sent to the pty - while the last one is always
+                    // the correct one, many programs and shells debounce those (I guess due to the
+                    // trauma of dealing with GUI resizes of the controlling terminal window), and this
+                    // then causes glitches and missing redraws
+                    // so we do this to play nice and always only send the last resize instruction to
+                    // each pane
+                    // the logic for this happens in the main Screen event loop
+                    os_input.cache_resizes();
+                },
+                PtyWriteInstruction::ApplyCachedResizes => {
+                    os_input.apply_cached_resizes();
+                },
+                PtyWriteInstruction::Exit => {
+                    return Ok(());
+                },
+            }
         }
+
+        // Drain pending writes — one pass across all terminals.
+        // For each terminal, write as much as the kernel accepts without
+        // blocking, then move on to the next terminal.
+        let terminal_ids: Vec<u32> = pending.keys().copied().collect();
+        for tid in terminal_ids {
+            let queue = match pending.get_mut(&tid) {
+                Some(q) if !q.is_empty() => q,
+                _ => continue,
+            };
+
+            while let Some(front) = queue.front_mut() {
+                let remaining = front.bytes.get(front.offset..).unwrap_or_default();
+                if remaining.is_empty() {
+                    queue.pop_front();
+                    continue;
+                }
+                match os_input.write_to_tty_stdin(tid, remaining) {
+                    Ok(0) => break, // EAGAIN — move to next terminal
+                    Ok(n) => {
+                        front.offset = front.offset.saturating_add(n);
+                        if front.offset >= front.bytes.len() {
+                            queue.pop_front();
+                        }
+                    },
+                    Err(e) => {
+                        // Terminal error (EBADF, EIO, etc) — clear its queue
+                        Err::<(), _>(e).with_context(err_context).non_fatal();
+                        queue.clear();
+                        break;
+                    },
+                }
+            }
+        }
+        pending.retain(|_, q| !q.is_empty());
     }
 }
+
+#[cfg(test)]
+#[path = "./unit/pty_writer_tests.rs"]
+mod pty_writer_tests;
diff --git a/zellij-server/src/thread_bus.rs b/zellij-server/src/thread_bus.rs
index ce997539..a381cd35 100644
--- a/zellij-server/src/thread_bus.rs
+++ b/zellij-server/src/thread_bus.rs
@@ -222,4 +222,22 @@ impl<T> Bus<T> {
         let idx = oper.index();
         oper.recv(&self.receivers[idx])
     }
+
+    pub fn recv_timeout(
+        &self,
+        timeout: std::time::Duration,
+    ) -> Result<(T, ErrorContext), channels::RecvTimeoutError> {
+        let mut selector = channels::Select::new();
+        self.receivers.iter().for_each(|r| {
+            selector.recv(r);
+        });
+        match selector.select_timeout(timeout) {
+            Ok(oper) => {
+                let idx = oper.index();
+                oper.recv(&self.receivers[idx])
+                    .map_err(|_| channels::RecvTimeoutError::Disconnected)
+            },
+            Err(_) => Err(channels::RecvTimeoutError::Timeout),
+        }
+    }
 }
diff --git a/zellij-server/src/unit/pty_writer_tests.rs b/zellij-server/src/unit/pty_writer_tests.rs
new file mode 100644
index 00000000..4c104063
--- /dev/null
+++ b/zellij-server/src/unit/pty_writer_tests.rs
@@ -0,0 +1,356 @@
+use super::*;
+use crate::os_input_output::{AsyncReader, ServerOsApi};
+use crate::panes::PaneId;
+use crate::thread_bus::Bus;
+use interprocess::local_socket::Stream as LocalSocketStream;
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
+use zellij_utils::channels::SenderWithContext;
+use zellij_utils::data::Palette;
+
+use zellij_utils::input::command::RunCommand;
+use zellij_utils::ipc::{ClientToServerMsg, IpcReceiverWithContext, ServerToClientMsg};
+
+use crate::ClientId;
+
+/// Per-terminal write behavior for the mock.
+#[derive(Clone)]
+enum WriteBehavior {
+    /// Accept all bytes (like a healthy terminal).
+    AcceptAll,
+    /// Accept at most `n` bytes per write call (simulates short writes).
+    AcceptAtMost(usize),
+    /// Accept zero bytes (simulates EAGAIN / full buffer).
+    AcceptNone,
+    /// Return an error (simulates EBADF, EIO, etc).
+    Error,
+}
+
+#[derive(Clone)]
+struct MockServerOsApi {
+    /// Controls how each terminal behaves on write.
+    write_behavior: Arc<Mutex<HashMap<u32, WriteBehavior>>>,
+    /// Log of (terminal_id, bytes) for each successful write.
+    write_log: Arc<Mutex<Vec<(u32, Vec<u8>)>>>,
+}
+
+impl MockServerOsApi {
+    fn new() -> Self {
+        MockServerOsApi {
+            write_behavior: Arc::new(Mutex::new(HashMap::new())),
+            write_log: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+    fn set_behavior(&self, terminal_id: u32, behavior: WriteBehavior) {
+        self.write_behavior
+            .lock()
+            .unwrap()
+            .insert(terminal_id, behavior);
+    }
+    fn get_written_bytes(&self, terminal_id: u32) -> Vec<u8> {
+        self.write_log
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|(tid, _)| *tid == terminal_id)
+            .flat_map(|(_, bytes)| bytes.clone())
+            .collect()
+    }
+}
+
+impl ServerOsApi for MockServerOsApi {
+    fn set_terminal_size_using_terminal_id(
+        &self,
+        _id: u32,
+        _cols: u16,
+        _rows: u16,
+        _width_in_pixels: Option<u16>,
+        _height_in_pixels: Option<u16>,
+    ) -> Result<()> {
+        Ok(())
+    }
+    fn spawn_terminal(
+        &self,
+        _terminal_action: zellij_utils::input::command::TerminalAction,
+        _quit_cb: Box<dyn Fn(PaneId, Option<i32>, RunCommand) + Send>,
+        _default_editor: Option<PathBuf>,
+    ) -> Result<(u32, Box<dyn AsyncReader>, Option<u32>)> {
+        unimplemented!()
+    }
+    fn write_to_tty_stdin(&self, terminal_id: u32, buf: &[u8]) -> Result<usize> {
+        let behavior = self
+            .write_behavior
+            .lock()
+            .unwrap()
+            .get(&terminal_id)
+            .cloned()
+            .unwrap_or(WriteBehavior::AcceptAll);
+        match behavior {
+            WriteBehavior::AcceptAll => {
+                self.write_log
+                    .lock()
+                    .unwrap()
+                    .push((terminal_id, buf.to_vec()));
+                Ok(buf.len())
+            },
+            WriteBehavior::AcceptAtMost(max) => {
+                let n = buf.len().min(max);
+                if n > 0 {
+                    self.write_log
+                        .lock()
+                        .unwrap()
+                        .push((terminal_id, buf[..n].to_vec()));
+                }
+                Ok(n)
+            },
+            WriteBehavior::AcceptNone => Ok(0),
+            WriteBehavior::Error => Err(anyhow::anyhow!(
+                "simulated write error for terminal {}",
+                terminal_id
+            )),
+        }
+    }
+    fn tcdrain(&self, _id: u32) -> Result<()> {
+        Ok(())
+    }
+    fn kill(&self, _pid: u32) -> Result<()> {
+        Ok(())
+    }
+    fn force_kill(&self, _pid: u32) -> Result<()> {
+        Ok(())
+    }
+    fn send_sigint(&self, _pid: u32) -> Result<()> {
+        Ok(())
+    }
+    fn box_clone(&self) -> Box<dyn ServerOsApi> {
+        Box::new((*self).clone())
+    }
+    fn send_to_client(&self, _client_id: ClientId, _msg: ServerToClientMsg) -> Result<()> {
+        Ok(())
+    }
+    fn new_client(
+        &mut self,
+        _client_id: ClientId,
+        _stream: LocalSocketStream,
+    ) -> Result<IpcReceiverWithContext<ClientToServerMsg>> {
+        unimplemented!()
+    }
+    fn new_client_with_reply(
+        &mut self,
+        _client_id: ClientId,
+        _stream: LocalSocketStream,
+        _reply_stream: LocalSocketStream,
+    ) -> Result<IpcReceiverWithContext<ClientToServerMsg>> {
+        unimplemented!()
+    }
+    fn remove_client(&mut self, _client_id: ClientId) -> Result<()> {
+        Ok(())
+    }
+    fn load_palette(&self) -> Palette {
+        Palette::default()
+    }
+    fn get_cwd(&self, _pid: u32) -> Option<PathBuf> {
+        None
+    }
+    fn write_to_file(&mut self, _buf: String, _file: Option<String>) -> Result<()> {
+        Ok(())
+    }
+    fn re_run_command_in_terminal(
+        &self,
+        _terminal_id: u32,
+        _run_command: RunCommand,
+        _quit_cb: Box<dyn Fn(PaneId, Option<i32>, RunCommand) + Send>,
+    ) -> Result<(Box<dyn AsyncReader>, Option<u32>)> {
+        unimplemented!()
+    }
+    fn clear_terminal_id(&self, _terminal_id: u32) -> Result<()> {
+        Ok(())
+    }
+}
+
+/// Helper: create a Bus and sender for PtyWriteInstruction with the given mock.
+fn make_test_bus(
+    mock: MockServerOsApi,
+) -> (
+    Bus<PtyWriteInstruction>,
+    SenderWithContext<PtyWriteInstruction>,
+) {
+    let (sender, receiver) = channels::unbounded();
+    let sender_with_ctx = SenderWithContext::new(sender);
+    let bus = Bus::new(
+        vec![receiver],
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        Some(Box::new(mock)),
+    );
+    (bus, sender_with_ctx)
+}
+
+/// Helper: send instructions and then Exit, run pty_writer_main, return when done.
+fn run_pty_writer(
+    bus: Bus<PtyWriteInstruction>,
+    sender: SenderWithContext<PtyWriteInstruction>,
+    instructions: Vec<PtyWriteInstruction>,
+) {
+    // Send all instructions followed by Exit
+    for instruction in instructions {
+        sender.send(instruction).unwrap();
+    }
+    sender.send(PtyWriteInstruction::Exit).unwrap();
+
+    pty_writer_main(bus).unwrap();
+}
+
+#[test]
+fn full_write_completes_immediately() {
+    let mock = MockServerOsApi::new();
+    mock.set_behavior(1, WriteBehavior::AcceptAll);
+    let mock_clone = mock.clone();
+
+    let (bus, sender) = make_test_bus(mock);
+    run_pty_writer(
+        bus,
+        sender,
+        vec![PtyWriteInstruction::Write(b"hello world".to_vec(), 1, None)],
+    );
+
+    let written = mock_clone.get_written_bytes(1);
+    assert_eq!(written, b"hello world");
+}
+
+#[test]
+fn partial_write_is_buffered_and_drained() {
+    let mock = MockServerOsApi::new();
+    // Accept only 3 bytes at a time
+    mock.set_behavior(1, WriteBehavior::AcceptAtMost(3));
+    let mock_clone = mock.clone();
+
+    let (bus, sender) = make_test_bus(mock);
+    run_pty_writer(
+        bus,
+        sender,
+        vec![PtyWriteInstruction::Write(b"abcdefghij".to_vec(), 1, None)],
+    );
+
+    let written = mock_clone.get_written_bytes(1);
+    assert_eq!(
+        written, b"abcdefghij",
+        "all bytes should eventually be written"
+    );
+}
+
+#[test]
+fn eagain_on_one_terminal_does_not_block_another() {
+    let mock = MockServerOsApi::new();
+    // Terminal 1 is stuck (EAGAIN)
+    mock.set_behavior(1, WriteBehavior::AcceptNone);
+    // Terminal 2 accepts everything
+    mock.set_behavior(2, WriteBehavior::AcceptAll);
+    let mock_clone = mock.clone();
+
+    let (bus, sender) = make_test_bus(mock);
+    run_pty_writer(
+        bus,
+        sender,
+        vec![
+            PtyWriteInstruction::Write(b"stuck data".to_vec(), 1, None),
+            PtyWriteInstruction::Write(b"good data".to_vec(), 2, None),
+        ],
+    );
+
+    // Terminal 2 should have received its bytes despite terminal 1 being stuck
+    let written_2 = mock_clone.get_written_bytes(2);
+    assert_eq!(written_2, b"good data");
+
+    // Terminal 1 data is lost (stuck until Exit, then dropped)
+    // — this is expected behavior; no assertion on terminal 1
+}
+
+#[test]
+fn writes_to_same_terminal_preserve_order() {
+    let mock = MockServerOsApi::new();
+    // Accept only 4 bytes at a time — forces buffering
+    mock.set_behavior(1, WriteBehavior::AcceptAtMost(4));
+    let mock_clone = mock.clone();
+
+    let (bus, sender) = make_test_bus(mock);
+    run_pty_writer(
+        bus,
+        sender,
+        vec![
+            PtyWriteInstruction::Write(b"AAAA".to_vec(), 1, None),
+            PtyWriteInstruction::Write(b"BBBB".to_vec(), 1, None),
+            PtyWriteInstruction::Write(b"CCCC".to_vec(), 1, None),
+        ],
+    );
+
+    let written = mock_clone.get_written_bytes(1);
+    assert_eq!(
+        written, b"AAAABBBBCCCC",
+        "writes must be delivered in order"
+    );
+}
+
+#[test]
+fn memory_cap_drops_buffer_when_exceeded() {
+    let mock = MockServerOsApi::new();
+    // Terminal is stuck — nothing drains
+    mock.set_behavior(1, WriteBehavior::AcceptNone);
+    let mock_clone = mock.clone();
+
+    let (bus, sender) = make_test_bus(mock);
+
+    // Send enough data to exceed MAX_PENDING_BYTES
+    let chunk_size = 1024 * 1024; // 1 MB
+    let mut instructions = Vec::new();
+    for _ in 0..(MAX_PENDING_BYTES / chunk_size + 1) {
+        instructions.push(PtyWriteInstruction::Write(vec![0x42; chunk_size], 1, None));
+    }
+    // After the cap is exceeded, send a write to terminal 2 to verify
+    // the writer thread is still functional
+    instructions.push(PtyWriteInstruction::Write(b"still alive".to_vec(), 2, None));
+    mock_clone.set_behavior(2, WriteBehavior::AcceptAll);
+
+    run_pty_writer(bus, sender, instructions);
+
+    let written_2 = mock_clone.get_written_bytes(2);
+    assert_eq!(
+        written_2, b"still alive",
+        "writer thread should continue functioning after dropping a buffer"
+    );
+}
+
+#[test]
+fn write_error_clears_terminal_queue() {
+    let mock = MockServerOsApi::new();
+    mock.set_behavior(1, WriteBehavior::Error);
+    mock.set_behavior(2, WriteBehavior::AcceptAll);
+    let mock_clone = mock.clone();
+
+    let (bus, sender) = make_test_bus(mock);
+    run_pty_writer(
+        bus,
+        sender,
+        vec![
+            PtyWriteInstruction::Write(b"will fail".to_vec(), 1, None),
+            PtyWriteInstruction::Write(b"also queued".to_vec(), 1, None),
+            PtyWriteInstruction::Write(b"should work".to_vec(), 2, None),
+        ],
+    );
+
+    // Terminal 1 should have nothing written (error cleared queue)
+    let written_1 = mock_clone.get_written_bytes(1);
+    assert!(
+        written_1.is_empty(),
+        "errored terminal should have no written bytes"
+    );
+
+    // Terminal 2 should be unaffected
+    let written_2 = mock_clone.get_written_bytes(2);
+    assert_eq!(written_2, b"should work");
+}