fix: wake pty actor for queued input

refs #379

2 files changed, 325 insertions, 56 deletions

diff --git a/src/pty/actor.rs b/src/pty/actor.rs
index 660a7d1c..cce58a75 100644
--- a/src/pty/actor.rs
+++ b/src/pty/actor.rs
@@ -12,7 +12,10 @@ use tracing::{debug, warn};
 
 use crate::pty::fd;
 
+#[cfg(not(test))]
 const ACTOR_POLL_MS: i32 = 50;
+#[cfg(test)]
+const ACTOR_POLL_MS: i32 = 1000;
 const ACTOR_COMMAND_BUFFER: usize = 1024;
 const HANDOFF_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
 
@@ -62,7 +65,6 @@ pub(crate) struct PtyIoActorConfig {
 }
 
 enum PtyIoDataCommand {
-    Wake,
     WriteUserInput(Bytes),
 }
 
@@ -79,6 +81,7 @@ enum PtyIoControlCommand {
 pub(crate) struct PtyIoActorHandle {
     data_tx: mpsc::Sender<PtyIoDataCommand>,
     control_tx: std_mpsc::Sender<PtyIoControlCommand>,
+    wake: fd::WakeWriter,
     user_writes: Arc<Mutex<UserWriteGate>>,
     controls: Arc<Mutex<SharedPtyControls>>,
 }
@@ -116,6 +119,7 @@ impl PtyIoActorHandle {
             return Err(mpsc::error::SendError(bytes));
         }
         permit.send(PtyIoDataCommand::WriteUserInput(bytes));
+        self.wake_actor();
         Ok(())
     }
 
@@ -130,22 +134,21 @@ impl PtyIoActorHandle {
         if !user_writes.accepting {
             return Err(mpsc::error::TrySendError::Closed(bytes));
         }
-        self.data_tx
+        match self
+            .data_tx
             .try_send(PtyIoDataCommand::WriteUserInput(bytes))
-            .map_err(|err| match err {
-                mpsc::error::TrySendError::Full(PtyIoDataCommand::WriteUserInput(bytes)) => {
-                    mpsc::error::TrySendError::Full(bytes)
-                }
-                mpsc::error::TrySendError::Closed(PtyIoDataCommand::WriteUserInput(bytes)) => {
-                    mpsc::error::TrySendError::Closed(bytes)
-                }
-                mpsc::error::TrySendError::Full(PtyIoDataCommand::Wake) => {
-                    mpsc::error::TrySendError::Full(Bytes::new())
-                }
-                mpsc::error::TrySendError::Closed(PtyIoDataCommand::Wake) => {
-                    mpsc::error::TrySendError::Closed(Bytes::new())
-                }
-            })
+        {
+            Ok(()) => {
+                self.wake_actor();
+                Ok(())
+            }
+            Err(mpsc::error::TrySendError::Full(PtyIoDataCommand::WriteUserInput(bytes))) => {
+                Err(mpsc::error::TrySendError::Full(bytes))
+            }
+            Err(mpsc::error::TrySendError::Closed(PtyIoDataCommand::WriteUserInput(bytes))) => {
+                Err(mpsc::error::TrySendError::Closed(bytes))
+            }
+        }
     }
 
     pub(crate) fn resize(&self, rows: u16, cols: u16, cell_width_px: u32, cell_height_px: u32) {
@@ -161,7 +164,7 @@ impl PtyIoActorHandle {
                 cell_height_px,
             });
         }
-        let _ = self.data_tx.try_send(PtyIoDataCommand::Wake);
+        self.wake_actor();
     }
 
     pub(crate) fn nudge_child_redraw_after_handoff(
@@ -183,7 +186,7 @@ impl PtyIoActorHandle {
                 cell_height_px,
             });
         }
-        let _ = self.data_tx.try_send(PtyIoDataCommand::Wake);
+        self.wake_actor();
     }
 
     pub(crate) fn begin_handoff(&self, timeout: Duration) -> std::io::Result<()> {
@@ -205,6 +208,7 @@ impl PtyIoActorHandle {
                     "pty actor closed",
                 ));
             }
+            self.wake_actor();
         }
         match reply_rx.recv_timeout(timeout) {
             Ok(Ok(())) => Ok(()),
@@ -227,6 +231,7 @@ impl PtyIoActorHandle {
         self.control_tx
             .send(PtyIoControlCommand::DuplicateForHandoff(reply_tx))
             .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed"))?;
+        self.wake_actor();
         reply_rx.recv_timeout(Duration::from_secs(1)).map_err(|_| {
             std::io::Error::new(
                 std::io::ErrorKind::TimedOut,
@@ -240,6 +245,7 @@ impl PtyIoActorHandle {
         self.control_tx
             .send(PtyIoControlCommand::ForegroundProcessGroup(reply_tx))
             .ok()?;
+        self.wake_actor();
         reply_rx.recv_timeout(Duration::from_secs(1)).ok()?
     }
 
@@ -248,6 +254,7 @@ impl PtyIoActorHandle {
         self.control_tx
             .send(PtyIoControlCommand::RollbackHandoff(reply_tx))
             .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed"))?;
+        self.wake_actor();
         let result = reply_rx.recv_timeout(Duration::from_secs(1)).map_err(|_| {
             std::io::Error::new(
                 std::io::ErrorKind::TimedOut,
@@ -276,6 +283,7 @@ impl PtyIoActorHandle {
         self.control_tx
             .send(PtyIoControlCommand::ReleaseAfterCommit(reply_tx))
             .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed"))?;
+        self.wake_actor();
         reply_rx.recv_timeout(Duration::from_secs(1)).map_err(|_| {
             std::io::Error::new(
                 std::io::ErrorKind::TimedOut,
@@ -292,7 +300,15 @@ impl PtyIoActorHandle {
                 .unwrap_or_else(|poisoned| poisoned.into_inner());
             user_writes.accepting = false;
         }
-        let _ = self.control_tx.send(PtyIoControlCommand::Shutdown);
+        if self.control_tx.send(PtyIoControlCommand::Shutdown).is_ok() {
+            self.wake_actor();
+        }
+    }
+
+    fn wake_actor(&self) {
+        if let Err(err) = self.wake.wake() {
+            debug!(err = %err, "failed to wake PTY actor");
+        }
     }
 }
 
@@ -300,11 +316,19 @@ pub(crate) struct PtyIoActor;
 
 impl PtyIoActor {
     pub(crate) fn spawn(config: PtyIoActorConfig) -> std::io::Result<PtyIoActorHandle> {
+        Self::spawn_inner(config, None)
+    }
+
+    fn spawn_inner(
+        config: PtyIoActorConfig,
+        poll_observer: Option<std_mpsc::Sender<()>>,
+    ) -> std::io::Result<PtyIoActorHandle> {
         fd::set_cloexec(config.master_fd.as_raw_fd())?;
         fd::set_nonblocking(config.master_fd.as_raw_fd())?;
 
         let (data_tx, data_rx) = mpsc::channel(ACTOR_COMMAND_BUFFER);
         let (control_tx, control_rx) = std_mpsc::channel();
+        let wake_pipe = fd::create_wake_pipe()?;
         let user_writes = Arc::new(Mutex::new(UserWriteGate {
             accepting: !config.initially_quiesced,
         }));
@@ -312,6 +336,7 @@ impl PtyIoActor {
         let handle = PtyIoActorHandle {
             data_tx,
             control_tx,
+            wake: wake_pipe.writer,
             user_writes,
             controls: Arc::clone(&controls),
         };
@@ -328,9 +353,11 @@ impl PtyIoActor {
             },
             pending_writes: VecDeque::new(),
             current_write_offset: 0,
+            wake_read_fd: wake_pipe.read_fd,
             controls,
             on_read: config.on_read,
             on_reader_exit: config.on_reader_exit,
+            poll_observer,
         };
         std::thread::Builder::new()
             .name(format!("herdr-pty-{}", config.pane_id))
@@ -339,6 +366,14 @@ impl PtyIoActor {
 
         Ok(handle)
     }
+
+    #[cfg(test)]
+    fn spawn_with_poll_observer(
+        config: PtyIoActorConfig,
+        poll_observer: std_mpsc::Sender<()>,
+    ) -> std::io::Result<PtyIoActorHandle> {
+        Self::spawn_inner(config, Some(poll_observer))
+    }
 }
 
 struct PtyIoActorRunner {
@@ -349,9 +384,11 @@ struct PtyIoActorRunner {
     state: ActorState,
     pending_writes: VecDeque<Bytes>,
     current_write_offset: usize,
+    wake_read_fd: OwnedFd,
     controls: Arc<Mutex<SharedPtyControls>>,
     on_read: ReadCallback,
     on_reader_exit: Option<ReaderExitCallback>,
+    poll_observer: Option<std_mpsc::Sender<()>>,
 }
 
 impl PtyIoActorRunner {
@@ -369,21 +406,39 @@ impl PtyIoActorRunner {
                 self.flush_pending_writes_once();
             }
 
-            if self.state == ActorState::Running {
-                match fd::poll_read_ready(self.file.as_raw_fd(), ACTOR_POLL_MS) {
-                    Ok(true) => {
-                        if !self.read_once() {
+            if let Some(poll_observer) = &self.poll_observer {
+                let _ = poll_observer.send(());
+            }
+
+            match fd::poll_pty_and_wake(
+                self.file.as_raw_fd(),
+                self.wake_read_fd.as_raw_fd(),
+                self.state == ActorState::Running,
+                !self.pending_writes.is_empty(),
+                ACTOR_POLL_MS,
+            ) {
+                Ok(readiness) => {
+                    if readiness.wake_ready {
+                        if let Err(err) = fd::drain_wake_fd(self.wake_read_fd.as_raw_fd()) {
+                            debug!(pane = self.pane_id, err = %err, "PTY actor wake drain failed");
                             break;
                         }
+                        continue;
+                    }
+                    if readiness.pty_write_ready && !self.pending_writes.is_empty() {
+                        self.flush_pending_writes_once();
                     }
-                    Ok(false) => {}
-                    Err(err) => {
-                        debug!(pane = self.pane_id, err = %err, "PTY actor poll failed");
+                    if self.state == ActorState::Running
+                        && readiness.pty_read_ready
+                        && !self.read_once()
+                    {
                         break;
                     }
                 }
-            } else {
-                std::thread::sleep(Duration::from_millis(ACTOR_POLL_MS as u64));
+                Err(err) => {
+                    debug!(pane = self.pane_id, err = %err, "PTY actor poll failed");
+                    break;
+                }
             }
         }
 
@@ -442,7 +497,6 @@ impl PtyIoActorRunner {
 
     fn handle_data_command(&mut self, command: PtyIoDataCommand) -> bool {
         match command {
-            PtyIoDataCommand::Wake => {}
             PtyIoDataCommand::WriteUserInput(bytes) => {
                 if self.state == ActorState::Running {
                     self.pending_writes.push_back(bytes);
@@ -520,15 +574,9 @@ impl PtyIoActorRunner {
     }
 
     fn drain_pre_quiesce_commands(&mut self) {
-        loop {
-            match self.data_rx.try_recv() {
-                Ok(PtyIoDataCommand::Wake) => {}
-                Ok(PtyIoDataCommand::WriteUserInput(bytes)) => {
-                    if self.state != ActorState::Released {
-                        self.pending_writes.push_back(bytes);
-                    }
-                }
-                Err(DataTryRecvError::Empty | DataTryRecvError::Disconnected) => break,
+        while let Ok(PtyIoDataCommand::WriteUserInput(bytes)) = self.data_rx.try_recv() {
+            if self.state != ActorState::Released {
+                self.pending_writes.push_back(bytes);
             }
         }
     }
@@ -673,13 +721,25 @@ mod tests {
     use super::*;
     use std::{
         io::{Read, Write},
-        os::fd::{FromRawFd, IntoRawFd},
+        os::fd::{AsRawFd, FromRawFd, IntoRawFd},
         os::unix::net::UnixStream,
     };
 
+    fn test_wake_pair() -> (fd::WakeWriter, OwnedFd) {
+        let pipe = fd::create_wake_pipe().expect("wake pipe");
+        (pipe.writer, pipe.read_fd)
+    }
+
     fn actor_with_socket_pair(
         initially_quiesced: bool,
     ) -> (PtyIoActorHandle, UnixStream, std_mpsc::Receiver<Bytes>) {
+        actor_with_socket_pair_and_poll_observer(initially_quiesced, None)
+    }
+
+    fn actor_with_socket_pair_and_poll_observer(
+        initially_quiesced: bool,
+        poll_observer: Option<std_mpsc::Sender<()>>,
+    ) -> (PtyIoActorHandle, UnixStream, std_mpsc::Receiver<Bytes>) {
         let (actor_socket, peer) = UnixStream::pair().expect("socket pair");
         actor_socket
             .set_nonblocking(true)
@@ -688,7 +748,7 @@ mod tests {
             .expect("peer timeout");
         let owned = unsafe { OwnedFd::from_raw_fd(actor_socket.into_raw_fd()) };
         let (read_tx, read_rx) = std_mpsc::channel();
-        let handle = PtyIoActor::spawn(PtyIoActorConfig {
+        let config = PtyIoActorConfig {
             pane_id: 1,
             master_fd: owned,
             initially_quiesced,
@@ -699,7 +759,12 @@ mod tests {
                 PtyReadResult::empty()
             }),
             on_reader_exit: None,
-        })
+        };
+        let handle = if let Some(poll_observer) = poll_observer {
+            PtyIoActor::spawn_with_poll_observer(config, poll_observer)
+        } else {
+            PtyIoActor::spawn(config)
+        }
         .expect("actor spawn");
         (handle, peer, read_rx)
     }
@@ -719,6 +784,56 @@ mod tests {
     }
 
     #[test]
+    fn actor_wakes_idle_poll_for_user_input() {
+        let (poll_tx, poll_rx) = std_mpsc::channel();
+        let (handle, mut peer, _read_rx) =
+            actor_with_socket_pair_and_poll_observer(false, Some(poll_tx));
+        peer.set_read_timeout(Some(Duration::from_millis(500)))
+            .expect("peer timeout");
+        poll_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("actor entered idle poll");
+
+        let start = Instant::now();
+        handle
+            .try_write_user_input(Bytes::from_static(b"x"))
+            .expect("write command accepted");
+
+        let mut buf = [0u8; 1];
+        peer.read_exact(&mut buf)
+            .expect("peer receives write without waiting for actor poll timeout");
+        assert_eq!(&buf, b"x");
+        assert!(
+            start.elapsed() < Duration::from_millis(500),
+            "actor write should be driven by wake fd, not the idle poll timeout"
+        );
+        handle.shutdown();
+    }
+
+    #[test]
+    fn poll_ignores_pty_hup_without_pty_interest() {
+        let (actor_socket, peer) = UnixStream::pair().expect("socket pair");
+        actor_socket
+            .set_nonblocking(true)
+            .expect("actor socket nonblocking");
+        drop(peer);
+        let wake_pipe = fd::create_wake_pipe().expect("wake pipe");
+
+        let readiness = fd::poll_pty_and_wake(
+            actor_socket.as_raw_fd(),
+            wake_pipe.read_fd.as_raw_fd(),
+            false,
+            false,
+            10,
+        )
+        .expect("poll succeeds");
+
+        assert!(!readiness.pty_read_ready);
+        assert!(!readiness.pty_write_ready);
+        assert!(!readiness.wake_ready);
+    }
+
+    #[test]
     fn actor_delivers_fd_reads_to_callback() {
         let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
 
@@ -793,12 +908,16 @@ mod tests {
         let (data_tx, _data_rx) = mpsc::channel(1);
         let (control_tx, _control_rx) = std_mpsc::channel();
         data_tx
-            .try_send(PtyIoDataCommand::Wake)
+            .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
+                b"fill",
+            )))
             .expect("fill command queue");
         let controls = Arc::new(Mutex::new(SharedPtyControls::default()));
+        let (wake, _wake_read_fd) = test_wake_pair();
         let handle = PtyIoActorHandle {
             data_tx,
             control_tx,
+            wake,
             user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
             controls: Arc::clone(&controls),
         };
@@ -833,11 +952,15 @@ mod tests {
         let (data_tx, mut data_rx) = mpsc::channel(1);
         let (control_tx, _control_rx) = std_mpsc::channel();
         data_tx
-            .try_send(PtyIoDataCommand::Wake)
+            .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
+                b"fill",
+            )))
             .expect("fill data queue");
+        let (wake, _wake_read_fd) = test_wake_pair();
         let handle = PtyIoActorHandle {
             data_tx,
             control_tx,
+            wake,
             user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
         };
@@ -853,7 +976,10 @@ mod tests {
             "async input should wait for queue capacity"
         );
 
-        assert!(matches!(data_rx.recv().await, Some(PtyIoDataCommand::Wake)));
+        assert!(matches!(
+            data_rx.recv().await,
+            Some(PtyIoDataCommand::WriteUserInput(_))
+        ));
         write
             .await
             .expect("write task joins")
@@ -871,11 +997,15 @@ mod tests {
         let (data_tx, mut data_rx) = mpsc::channel(1);
         let (control_tx, control_rx) = std_mpsc::channel();
         data_tx
-            .try_send(PtyIoDataCommand::Wake)
+            .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
+                b"fill",
+            )))
             .expect("fill data queue");
+        let (wake, _wake_read_fd) = test_wake_pair();
         let handle = PtyIoActorHandle {
             data_tx,
             control_tx,
+            wake,
             user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
         };
@@ -901,7 +1031,10 @@ mod tests {
             .join()
             .expect("handoff thread joins")
             .expect("handoff succeeds");
-        assert!(matches!(data_rx.recv().await, Some(PtyIoDataCommand::Wake)));
+        assert!(matches!(
+            data_rx.recv().await,
+            Some(PtyIoDataCommand::WriteUserInput(_))
+        ));
 
         let err = write.await.expect("write task joins").expect_err(
             "write waiting for capacity must be rejected after handoff closes the input gate",
@@ -918,11 +1051,15 @@ mod tests {
         let (data_tx, _data_rx) = mpsc::channel(1);
         let (control_tx, control_rx) = std_mpsc::channel();
         data_tx
-            .try_send(PtyIoDataCommand::Wake)
+            .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
+                b"fill",
+            )))
             .expect("fill data queue");
+        let (wake, _wake_read_fd) = test_wake_pair();
         let handle = PtyIoActorHandle {
             data_tx,
             control_tx,
+            wake,
             user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
         };
@@ -967,9 +1104,11 @@ mod tests {
             state: ActorState::Running,
             pending_writes: VecDeque::new(),
             current_write_offset: 0,
+            wake_read_fd: fd::create_wake_pipe().expect("wake pipe").read_fd,
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
             on_read: Box::new(|_| PtyReadResult::empty()),
             on_reader_exit: None,
+            poll_observer: None,
         };
 
         runner.begin_handoff().expect("handoff drains queued write");
diff --git a/src/pty/fd.rs b/src/pty/fd.rs
index a8bf021e..cc0edadd 100644
--- a/src/pty/fd.rs
+++ b/src/pty/fd.rs
@@ -1,5 +1,8 @@
 #[cfg(unix)]
-use std::os::fd::RawFd;
+use std::{
+    os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd},
+    sync::Arc,
+};
 
 #[cfg(unix)]
 pub(crate) fn duplicate_fd(fd: RawFd) -> std::io::Result<RawFd> {
@@ -45,14 +48,121 @@ pub(crate) fn duplicate_cloexec_fd(fd: RawFd) -> std::io::Result<RawFd> {
 }
 
 #[cfg(unix)]
-pub(crate) fn poll_read_ready(fd: RawFd, timeout_ms: i32) -> std::io::Result<bool> {
-    let mut poll_fd = libc::pollfd {
-        fd,
-        events: libc::POLLIN,
-        revents: 0,
-    };
+#[derive(Clone)]
+pub(crate) struct WakeWriter {
+    fd: Arc<OwnedFd>,
+}
+
+#[cfg(unix)]
+impl WakeWriter {
+    pub(crate) fn wake(&self) -> std::io::Result<()> {
+        loop {
+            let byte = [1u8];
+            let written =
+                unsafe { libc::write(self.fd.as_raw_fd(), byte.as_ptr().cast(), byte.len()) };
+            if written >= 0 {
+                return Ok(());
+            }
+
+            let err = std::io::Error::last_os_error();
+            if err.kind() == std::io::ErrorKind::WouldBlock {
+                return Ok(());
+            }
+            if err.kind() != std::io::ErrorKind::Interrupted {
+                return Err(err);
+            }
+        }
+    }
+}
+
+#[cfg(unix)]
+pub(crate) struct WakePipe {
+    pub(crate) read_fd: OwnedFd,
+    pub(crate) writer: WakeWriter,
+}
+
+#[cfg(unix)]
+pub(crate) fn create_wake_pipe() -> std::io::Result<WakePipe> {
+    let mut fds = [-1; 2];
+    if unsafe { libc::pipe(fds.as_mut_ptr()) } < 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+
+    let read_fd = unsafe { OwnedFd::from_raw_fd(fds[0]) };
+    let write_fd = unsafe { OwnedFd::from_raw_fd(fds[1]) };
+    for fd in [read_fd.as_raw_fd(), write_fd.as_raw_fd()] {
+        set_cloexec(fd).and_then(|_| set_nonblocking(fd))?;
+    }
+
+    Ok(WakePipe {
+        read_fd,
+        writer: WakeWriter {
+            fd: Arc::new(write_fd),
+        },
+    })
+}
+
+#[cfg(unix)]
+pub(crate) fn drain_wake_fd(fd: RawFd) -> std::io::Result<()> {
+    let mut buf = [0u8; 64];
     loop {
-        let result = unsafe { libc::poll(&mut poll_fd, 1, timeout_ms) };
+        let read = unsafe { libc::read(fd, buf.as_mut_ptr().cast(), buf.len()) };
+        if read == 0 {
+            return Ok(());
+        }
+        if read > 0 {
+            continue;
+        }
+
+        let err = std::io::Error::last_os_error();
+        if err.kind() == std::io::ErrorKind::WouldBlock {
+            return Ok(());
+        }
+        if err.kind() != std::io::ErrorKind::Interrupted {
+            return Err(err);
+        }
+    }
+}
+
+#[cfg(unix)]
+pub(crate) struct PtyWakeReadiness {
+    pub(crate) pty_read_ready: bool,
+    pub(crate) pty_write_ready: bool,
+    pub(crate) wake_ready: bool,
+}
+
+#[cfg(unix)]
+pub(crate) fn poll_pty_and_wake(
+    pty_fd: RawFd,
+    wake_fd: RawFd,
+    poll_pty_read: bool,
+    poll_pty_write: bool,
+    timeout_ms: i32,
+) -> std::io::Result<PtyWakeReadiness> {
+    let poll_pty = poll_pty_read || poll_pty_write;
+    let mut pty_events = 0;
+    if poll_pty_read {
+        pty_events |= libc::POLLIN;
+    }
+    if poll_pty_write {
+        pty_events |= libc::POLLOUT;
+    }
+
+    let mut poll_fds = [
+        libc::pollfd {
+            fd: if poll_pty { pty_fd } else { -1 },
+            events: pty_events,
+            revents: 0,
+        },
+        libc::pollfd {
+            fd: wake_fd,
+            events: libc::POLLIN,
+            revents: 0,
+        },
+    ];
+
+    loop {
+        let result = unsafe { libc::poll(poll_fds.as_mut_ptr(), poll_fds.len() as _, timeout_ms) };
         if result < 0 {
             let err = std::io::Error::last_os_error();
             if err.kind() == std::io::ErrorKind::Interrupted {
@@ -60,7 +170,27 @@ pub(crate) fn poll_read_ready(fd: RawFd, timeout_ms: i32) -> std::io::Result<boo
             }
             return Err(err);
         }
-        return Ok(result > 0 && (poll_fd.revents & (libc::POLLIN | libc::POLLHUP)) != 0);
+
+        let pty_revents = if poll_pty { poll_fds[0].revents } else { 0 };
+        let wake_revents = poll_fds[1].revents;
+        if (pty_revents | wake_revents) & libc::POLLNVAL != 0 {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                "poll encountered invalid PTY actor fd",
+            ));
+        }
+        if pty_revents & libc::POLLERR != 0 {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                "poll encountered PTY fd error",
+            ));
+        }
+
+        return Ok(PtyWakeReadiness {
+            pty_read_ready: pty_revents & (libc::POLLIN | libc::POLLHUP) != 0,
+            pty_write_ready: pty_revents & (libc::POLLOUT | libc::POLLHUP) != 0,
+            wake_ready: wake_revents & (libc::POLLIN | libc::POLLHUP | libc::POLLERR) != 0,
+        });
     }
 }