feat: add native windows beta support

5 files changed, 1460 insertions, 1191 deletions

diff --git a/src/pty/actor.rs b/src/pty/actor.rs
index 70634475..8c3a6b52 100644
--- a/src/pty/actor.rs
+++ b/src/pty/actor.rs
@@ -1,1172 +1,231 @@
-use std::{
-    collections::VecDeque,
-    io::{Read, Write},
-    os::fd::{AsRawFd, OwnedFd, RawFd},
-    sync::{mpsc as std_mpsc, Arc, Mutex},
-    time::{Duration, Instant},
-};
+#[cfg(unix)]
+mod unix;
 
-use bytes::Bytes;
-use tokio::sync::mpsc::{self, error::TryRecvError as DataTryRecvError};
-use tracing::{debug, warn};
+#[cfg(unix)]
+pub(crate) use unix::*;
 
-use crate::pty::fd;
+#[cfg(windows)]
+mod windows {
+    use std::io::{Read, Write};
+    use std::sync::{mpsc as std_mpsc, Arc, Mutex};
+    use std::time::Duration;
 
-#[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);
+    use bytes::Bytes;
+    use portable_pty::{MasterPty, PtySize};
+    use tokio::sync::mpsc;
+    use tracing::{debug, warn};
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum ActorState {
-    Running,
-    Quiesced,
-    Released,
-}
-
-pub(crate) struct PtyReadResult {
-    pub terminal_responses: Vec<Bytes>,
-}
-
-impl PtyReadResult {
-    #[cfg(test)]
-    pub(crate) fn empty() -> Self {
-        Self {
-            terminal_responses: Vec::new(),
-        }
-    }
-}
-
-type ReadCallback = Box<dyn FnMut(&[u8]) -> PtyReadResult + Send + 'static>;
-type ReaderExitCallback = Box<dyn FnOnce() + Send + 'static>;
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-struct PtyResize {
-    rows: u16,
-    cols: u16,
-    cell_width_px: u32,
-    cell_height_px: u32,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-struct PtyResizeRequest {
-    resize: PtyResize,
-    terminal_responses: Vec<Bytes>,
-}
-
-#[derive(Default)]
-struct SharedPtyControls {
-    resize: Option<PtyResizeRequest>,
-    nudge: Option<PtyResize>,
-}
-
-pub(crate) struct PtyIoActorConfig {
-    pub pane_id: u32,
-    pub master_fd: OwnedFd,
-    pub initially_quiesced: bool,
-    pub on_read: ReadCallback,
-    pub on_reader_exit: Option<ReaderExitCallback>,
-}
-
-enum PtyIoDataCommand {
-    WriteUserInput(Bytes),
-}
-
-enum PtyIoControlCommand {
-    BeginHandoff(std_mpsc::Sender<std::io::Result<()>>),
-    DuplicateForHandoff(std_mpsc::Sender<std::io::Result<RawFd>>),
-    ForegroundProcessGroup(std_mpsc::Sender<Option<u32>>),
-    RollbackHandoff(std_mpsc::Sender<std::io::Result<()>>),
-    ReleaseAfterCommit(std_mpsc::Sender<std::io::Result<()>>),
-    Shutdown,
-}
-
-#[derive(Clone)]
-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>>,
-}
-
-#[derive(Debug)]
-struct UserWriteGate {
-    accepting: bool,
-}
-
-impl PtyIoActorHandle {
-    pub(crate) async fn write_user_input(
-        &self,
-        bytes: Bytes,
-    ) -> Result<(), mpsc::error::SendError<Bytes>> {
-        {
-            let user_writes = self
-                .user_writes
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            if !user_writes.accepting {
-                return Err(mpsc::error::SendError(bytes));
-            }
-        }
-
-        let permit = match self.data_tx.reserve().await {
-            Ok(permit) => permit,
-            Err(_) => return Err(mpsc::error::SendError(bytes)),
-        };
-
-        let user_writes = self
-            .user_writes
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
-        if !user_writes.accepting {
-            return Err(mpsc::error::SendError(bytes));
-        }
-        permit.send(PtyIoDataCommand::WriteUserInput(bytes));
-        self.wake_actor();
-        Ok(())
+    pub(crate) struct PtyReadResult {
+        pub terminal_responses: Vec<Bytes>,
     }
 
-    pub(crate) fn try_write_user_input(
-        &self,
-        bytes: Bytes,
-    ) -> Result<(), mpsc::error::TrySendError<Bytes>> {
-        let user_writes = self
-            .user_writes
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
-        if !user_writes.accepting {
-            return Err(mpsc::error::TrySendError::Closed(bytes));
-        }
-        match self
-            .data_tx
-            .try_send(PtyIoDataCommand::WriteUserInput(bytes))
-        {
-            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))
-            }
-        }
-    }
+    type ReadCallback = Box<dyn FnMut(&[u8]) -> PtyReadResult + Send + 'static>;
+    type ReaderExitCallback = Box<dyn FnOnce() + Send + 'static>;
 
-    pub(crate) fn resize(
-        &self,
+    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    struct PtyResize {
         rows: u16,
         cols: u16,
         cell_width_px: u32,
         cell_height_px: u32,
-        terminal_responses: Vec<Bytes>,
-    ) {
-        {
-            let mut controls = self
-                .controls
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            controls.resize = Some(PtyResizeRequest {
-                resize: PtyResize {
-                    rows,
-                    cols,
-                    cell_width_px,
-                    cell_height_px,
-                },
-                terminal_responses,
-            });
-        }
-        self.wake_actor();
-    }
-
-    pub(crate) fn nudge_child_redraw_after_handoff(
-        &self,
-        rows: u16,
-        cols: u16,
-        cell_width_px: u32,
-        cell_height_px: u32,
-    ) {
-        {
-            let mut controls = self
-                .controls
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            controls.nudge = Some(PtyResize {
-                rows,
-                cols,
-                cell_width_px,
-                cell_height_px,
-            });
-        }
-        self.wake_actor();
     }
 
-    pub(crate) fn begin_handoff(&self, timeout: Duration) -> std::io::Result<()> {
-        let (reply_tx, reply_rx) = std_mpsc::channel();
-        {
-            let mut user_writes = self
-                .user_writes
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            user_writes.accepting = false;
-            if self
-                .control_tx
-                .send(PtyIoControlCommand::BeginHandoff(reply_tx))
-                .is_err()
-            {
-                user_writes.accepting = true;
-                return Err(std::io::Error::new(
-                    std::io::ErrorKind::BrokenPipe,
-                    "pty actor closed",
-                ));
-            }
-            self.wake_actor();
-        }
-        match reply_rx.recv_timeout(timeout) {
-            Ok(Ok(())) => Ok(()),
-            Ok(Err(err)) => {
-                let _ = self.rollback_handoff();
-                Err(err)
-            }
-            Err(_) => {
-                let _ = self.rollback_handoff();
-                Err(std::io::Error::new(
-                    std::io::ErrorKind::TimedOut,
-                    "timed out waiting for PTY actor to quiesce",
-                ))
-            }
-        }
+    struct PtyResizeRequest {
+        resize: PtyResize,
+        terminal_responses: Vec<Bytes>,
     }
 
-    pub(crate) fn duplicate_for_handoff(&self) -> std::io::Result<RawFd> {
-        let (reply_tx, reply_rx) = std_mpsc::channel();
-        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,
-                "timed out waiting for PTY handoff duplicate",
-            )
-        })?
+    pub(crate) struct PtyIoActorConfig {
+        pub pane_id: u32,
+        pub master: Box<dyn MasterPty + Send>,
+        pub initially_quiesced: bool,
+        pub on_read: ReadCallback,
+        pub on_reader_exit: Option<ReaderExitCallback>,
     }
 
-    pub(crate) fn foreground_process_group_id(&self) -> Option<u32> {
-        let (reply_tx, reply_rx) = std_mpsc::channel();
-        self.control_tx
-            .send(PtyIoControlCommand::ForegroundProcessGroup(reply_tx))
-            .ok()?;
-        self.wake_actor();
-        reply_rx.recv_timeout(Duration::from_secs(1)).ok()?
+    enum PtyIoControlCommand {
+        Resize(PtyResizeRequest),
+        Shutdown,
     }
 
-    pub(crate) fn rollback_handoff(&self) -> std::io::Result<()> {
-        let (reply_tx, reply_rx) = std_mpsc::channel();
-        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,
-                "timed out waiting for PTY handoff rollback",
-            )
-        })?;
-        if result.is_ok() {
-            let mut user_writes = self
-                .user_writes
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            user_writes.accepting = true;
-        }
-        result
+    #[derive(Clone)]
+    pub(crate) struct PtyIoActorHandle {
+        data_tx: mpsc::Sender<Bytes>,
+        control_tx: std_mpsc::Sender<PtyIoControlCommand>,
+        accepting: Arc<Mutex<bool>>,
     }
 
-    pub(crate) fn release_after_commit(&self) -> std::io::Result<()> {
-        {
-            let mut user_writes = self
-                .user_writes
+    impl PtyIoActorHandle {
+        pub(crate) async fn write_user_input(
+            &self,
+            bytes: Bytes,
+        ) -> Result<(), mpsc::error::SendError<Bytes>> {
+            if !*self
+                .accepting
                 .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            user_writes.accepting = false;
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+            {
+                return Err(mpsc::error::SendError(bytes));
+            }
+            self.data_tx.send(bytes).await
         }
-        let (reply_tx, reply_rx) = std_mpsc::channel();
-        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,
-                "timed out waiting for PTY actor release",
-            )
-        })?
-    }
 
-    pub(crate) fn shutdown(&self) {
-        {
-            let mut user_writes = self
-                .user_writes
+        pub(crate) fn try_write_user_input(
+            &self,
+            bytes: Bytes,
+        ) -> Result<(), mpsc::error::TrySendError<Bytes>> {
+            if !*self
+                .accepting
                 .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            user_writes.accepting = false;
-        }
-        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");
-        }
-    }
-}
-
-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,
-        }));
-        let controls = Arc::new(Mutex::new(SharedPtyControls::default()));
-        let handle = PtyIoActorHandle {
-            data_tx,
-            control_tx,
-            wake: wake_pipe.writer,
-            user_writes,
-            controls: Arc::clone(&controls),
-        };
-
-        let mut runner = PtyIoActorRunner {
-            pane_id: config.pane_id,
-            file: std::fs::File::from(config.master_fd),
-            data_rx,
-            control_rx,
-            state: if config.initially_quiesced {
-                ActorState::Quiesced
-            } else {
-                ActorState::Running
-            },
-            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))
-            .spawn(move || runner.run())
-            .map_err(|err| std::io::Error::other(err.to_string()))?;
-
-        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 {
-    pane_id: u32,
-    file: std::fs::File,
-    data_rx: mpsc::Receiver<PtyIoDataCommand>,
-    control_rx: std_mpsc::Receiver<PtyIoControlCommand>,
-    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 {
-    fn run(&mut self) {
-        let mut should_exit = false;
-        while !should_exit {
-            should_exit = self.drain_commands();
-            if should_exit || self.state == ActorState::Released {
-                break;
-            }
-
-            self.apply_pending_controls();
-
-            if !self.pending_writes.is_empty() {
-                self.flush_pending_writes_once();
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+            {
+                return Err(mpsc::error::TrySendError::Closed(bytes));
             }
+            self.data_tx.try_send(bytes)
+        }
 
-            if let Some(poll_observer) = &self.poll_observer {
-                let _ = poll_observer.send(());
-            }
+        pub(crate) fn resize(
+            &self,
+            rows: u16,
+            cols: u16,
+            cell_width_px: u32,
+            cell_height_px: u32,
+            terminal_responses: Vec<Bytes>,
+        ) {
+            let _ = self
+                .control_tx
+                .send(PtyIoControlCommand::Resize(PtyResizeRequest {
+                    resize: PtyResize {
+                        rows,
+                        cols,
+                        cell_width_px,
+                        cell_height_px,
+                    },
+                    terminal_responses,
+                }));
+        }
+
+        pub(crate) fn shutdown(&self) {
+            if let Ok(mut accepting) = self.accepting.lock() {
+                *accepting = false;
+            }
+            let _ = self.control_tx.send(PtyIoControlCommand::Shutdown);
+        }
+    }
+
+    pub(crate) struct PtyIoActor;
+
+    impl PtyIoActor {
+        pub(crate) fn spawn(config: PtyIoActorConfig) -> std::io::Result<PtyIoActorHandle> {
+            let PtyIoActorConfig {
+                pane_id,
+                master,
+                initially_quiesced,
+                mut on_read,
+                on_reader_exit,
+            } = config;
+
+            let mut reader = master
+                .try_clone_reader()
+                .map_err(|err| std::io::Error::other(err.to_string()))?;
+            let writer = master
+                .take_writer()
+                .map_err(|err| std::io::Error::other(err.to_string()))?;
+            let writer = Arc::new(Mutex::new(writer));
+            let (data_tx, mut data_rx) = mpsc::channel::<Bytes>(1024);
+            let (control_tx, control_rx) = std_mpsc::channel::<PtyIoControlCommand>();
+            let accepting = Arc::new(Mutex::new(!initially_quiesced));
 
-            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");
+            {
+                let writer = Arc::clone(&writer);
+                std::thread::spawn(move || {
+                    while let Some(bytes) = data_rx.blocking_recv() {
+                        if write_all_locked(&writer, &bytes).is_err() {
                             break;
                         }
-                        continue;
-                    }
-                    if readiness.pty_write_ready && !self.pending_writes.is_empty() {
-                        self.flush_pending_writes_once();
-                    }
-                    if self.state == ActorState::Running
-                        && readiness.pty_read_ready
-                        && !self.read_once()
-                    {
-                        break;
                     }
-                }
-                Err(err) => {
-                    debug!(pane = self.pane_id, err = %err, "PTY actor poll failed");
-                    break;
-                }
+                    debug!(pane_id, "windows pty writer thread exiting");
+                });
             }
-        }
 
-        if let Some(on_reader_exit) = self.on_reader_exit.take() {
-            on_reader_exit();
-        }
-        debug!(pane = self.pane_id, "PTY actor exiting");
-    }
-
-    fn drain_commands(&mut self) -> bool {
-        if self.drain_control_commands() {
-            return true;
-        }
-        self.drain_data_commands()
-    }
-
-    fn drain_control_commands(&mut self) -> bool {
-        let mut should_exit = false;
-        loop {
-            match self.control_rx.try_recv() {
-                Ok(command) => {
-                    if self.handle_control_command(command) {
-                        should_exit = true;
-                        break;
+            {
+                let writer = Arc::clone(&writer);
+                std::thread::spawn(move || {
+                    let mut buf = [0u8; 8192];
+                    loop {
+                        match reader.read(&mut buf) {
+                            Ok(0) => break,
+                            Ok(n) => {
+                                let result = on_read(&buf[..n]);
+                                for response in result.terminal_responses {
+                                    if write_all_locked(&writer, &response).is_err() {
+                                        break;
+                                    }
+                                }
+                            }
+                            Err(err) => {
+                                debug!(pane_id, err = %err, "windows pty reader failed");
+                                break;
+                            }
+                        }
                     }
-                }
-                Err(std_mpsc::TryRecvError::Empty) => break,
-                Err(std_mpsc::TryRecvError::Disconnected) => {
-                    should_exit = true;
-                    break;
-                }
-            }
-        }
-        should_exit
-    }
-
-    fn drain_data_commands(&mut self) -> bool {
-        let mut should_exit = false;
-        loop {
-            match self.data_rx.try_recv() {
-                Ok(command) => {
-                    if self.handle_data_command(command) {
-                        should_exit = true;
-                        break;
+                    if let Some(on_reader_exit) = on_reader_exit {
+                        on_reader_exit();
                     }
-                }
-                Err(DataTryRecvError::Empty) => break,
-                Err(DataTryRecvError::Disconnected) => {
-                    should_exit = true;
-                    break;
-                }
+                    debug!(pane_id, "windows pty reader thread exiting");
+                });
             }
-        }
-        should_exit
-    }
 
-    fn handle_data_command(&mut self, command: PtyIoDataCommand) -> bool {
-        match command {
-            PtyIoDataCommand::WriteUserInput(bytes) => {
-                if self.state == ActorState::Running {
-                    self.pending_writes.push_back(bytes);
-                }
-            }
-        }
-        false
-    }
-
-    fn handle_control_command(&mut self, command: PtyIoControlCommand) -> bool {
-        match command {
-            PtyIoControlCommand::BeginHandoff(reply) => {
-                let result = self.begin_handoff();
-                let _ = reply.send(result);
-            }
-            PtyIoControlCommand::DuplicateForHandoff(reply) => {
-                let result = if self.state == ActorState::Quiesced {
-                    fd::duplicate_cloexec_fd(self.file.as_raw_fd())
-                } else {
-                    Err(std::io::Error::other(
-                        "PTY actor must be quiesced before handoff duplication",
-                    ))
-                };
-                let _ = reply.send(result);
-            }
-            PtyIoControlCommand::ForegroundProcessGroup(reply) => {
-                let result =
-                    crate::platform::foreground_process_group_id_for_tty_fd(self.file.as_raw_fd());
-                let _ = reply.send(result);
-            }
-            PtyIoControlCommand::RollbackHandoff(reply) => {
-                let result = if self.state == ActorState::Released {
-                    Err(std::io::Error::new(
-                        std::io::ErrorKind::BrokenPipe,
-                        "PTY actor was released before handoff rollback",
-                    ))
-                } else {
-                    self.state = ActorState::Running;
-                    Ok(())
-                };
-                let _ = reply.send(result);
-            }
-            PtyIoControlCommand::ReleaseAfterCommit(reply) => {
-                self.state = ActorState::Released;
-                self.pending_writes.clear();
-                let _ = reply.send(Ok(()));
-                return true;
-            }
-            PtyIoControlCommand::Shutdown => return true,
-        }
-        false
-    }
-
-    fn begin_handoff(&mut self) -> std::io::Result<()> {
-        self.drain_pre_quiesce_commands();
-        self.apply_pending_controls();
-        if self.state == ActorState::Released {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::BrokenPipe,
-                "PTY actor was released before handoff quiesce",
-            ));
-        }
-        let deadline = Instant::now() + HANDOFF_DRAIN_TIMEOUT;
-        while !self.pending_writes.is_empty() {
-            if Instant::now() >= deadline {
-                return Err(std::io::Error::new(
-                    std::io::ErrorKind::TimedOut,
-                    "timed out draining PTY writes before handoff",
-                ));
-            }
-            self.flush_pending_writes_once();
-        }
-        self.state = ActorState::Quiesced;
-        Ok(())
-    }
-
-    fn drain_pre_quiesce_commands(&mut self) {
-        while let Ok(PtyIoDataCommand::WriteUserInput(bytes)) = self.data_rx.try_recv() {
-            if self.state != ActorState::Released {
-                self.pending_writes.push_back(bytes);
-            }
-        }
-    }
-
-    fn apply_pending_controls(&mut self) {
-        let (resize, nudge) = {
-            let mut controls = self
-                .controls
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            (controls.resize.take(), controls.nudge.take())
-        };
-        if self.state == ActorState::Released {
-            return;
-        }
-        if let Some(request) = resize {
-            self.resize(request.resize);
-            self.enqueue_terminal_responses(request.terminal_responses);
-        }
-        if let Some(nudge) = nudge {
-            self.nudge(nudge);
-        }
-    }
-
-    fn read_once(&mut self) -> bool {
-        let mut buf = [0u8; 8192];
-        match self.file.read(&mut buf) {
-            Ok(0) => false,
-            Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => true,
-            Err(err) if err.kind() == std::io::ErrorKind::Interrupted => true,
-            Err(err) => {
-                debug!(pane = self.pane_id, err = %err, "PTY actor read failed");
-                false
-            }
-            Ok(n) => {
-                let result = (self.on_read)(&buf[..n]);
-                self.enqueue_terminal_responses(result.terminal_responses);
-                true
-            }
-        }
-    }
-
-    fn enqueue_terminal_responses(&mut self, terminal_responses: Vec<Bytes>) {
-        if self.state == ActorState::Released {
-            return;
-        }
-        self.pending_writes.extend(terminal_responses);
-    }
-
-    fn flush_pending_writes_once(&mut self) {
-        while let Some(bytes) = self.pending_writes.front() {
-            let chunk = &bytes[self.current_write_offset..];
-            match self.file.write(chunk) {
-                Ok(0) => {
-                    warn!(pane = self.pane_id, "PTY actor write returned zero bytes");
-                    return;
-                }
-                Ok(written) => {
-                    self.current_write_offset += written;
-                    if self.current_write_offset >= bytes.len() {
-                        self.pending_writes.pop_front();
-                        self.current_write_offset = 0;
+            {
+                let writer = Arc::clone(&writer);
+                std::thread::spawn(move || {
+                    for command in control_rx {
+                        match command {
+                            PtyIoControlCommand::Resize(request) => {
+                                let size = request.resize;
+                                if let Err(err) = master.resize(PtySize {
+                                    rows: size.rows,
+                                    cols: size.cols,
+                                    pixel_width: size.cell_width_px.min(u16::MAX as u32) as u16,
+                                    pixel_height: size.cell_height_px.min(u16::MAX as u32) as u16,
+                                }) {
+                                    warn!(pane_id, err = %err, "windows pty resize failed");
+                                }
+                                for response in request.terminal_responses {
+                                    if write_all_locked(&writer, &response).is_err() {
+                                        break;
+                                    }
+                                }
+                            }
+                            PtyIoControlCommand::Shutdown => break,
+                        }
                     }
-                }
-                Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
-                    let _ = fd::poll_write_ready(self.file.as_raw_fd(), ACTOR_POLL_MS);
-                    return;
-                }
-                Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return,
-                Err(err) => {
-                    warn!(pane = self.pane_id, err = %err, "PTY actor write failed");
-                    self.pending_writes.clear();
-                    self.current_write_offset = 0;
-                    return;
-                }
+                    debug!(pane_id, "windows pty control thread exiting");
+                });
             }
-        }
-        let _ = self.file.flush();
-    }
 
-    fn resize(&self, resize: PtyResize) {
-        self.log_resize_result(fd::resize_pty_fd(
-            self.file.as_raw_fd(),
-            resize.rows,
-            resize.cols,
-            resize.cell_width_px,
-            resize.cell_height_px,
-        ));
-    }
-
-    fn nudge(&mut self, resize: PtyResize) {
-        if self.state == ActorState::Released {
-            return;
-        }
-        let nudge = if resize.rows > 2 {
-            (
-                resize.rows - 1,
-                resize.cols,
-                resize.cell_width_px,
-                resize.cell_height_px,
-            )
-        } else {
-            (
-                resize.rows,
-                resize.cols.saturating_sub(1).max(4),
-                resize.cell_width_px,
-                resize.cell_height_px,
-            )
-        };
-        if nudge
-            == (
-                resize.rows,
-                resize.cols,
-                resize.cell_width_px,
-                resize.cell_height_px,
-            )
-        {
-            return;
-        }
-        self.log_resize_result(fd::resize_pty_fd(
-            self.file.as_raw_fd(),
-            nudge.0,
-            nudge.1,
-            nudge.2,
-            nudge.3,
-        ));
-        std::thread::sleep(Duration::from_millis(30));
-        self.log_resize_result(fd::resize_pty_fd(
-            self.file.as_raw_fd(),
-            resize.rows,
-            resize.cols,
-            resize.cell_width_px,
-            resize.cell_height_px,
-        ));
-    }
-
-    fn log_resize_result(&self, result: std::io::Result<()>) {
-        if let Err(err) = result {
-            debug!(pane = self.pane_id, err = %err, "PTY resize failed");
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::{
-        io::{Read, Write},
-        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)
-            .expect("actor socket nonblocking");
-        peer.set_read_timeout(Some(Duration::from_secs(1)))
-            .expect("peer timeout");
-        let owned = unsafe { OwnedFd::from_raw_fd(actor_socket.into_raw_fd()) };
-        let (read_tx, read_rx) = std_mpsc::channel();
-        let config = PtyIoActorConfig {
-            pane_id: 1,
-            master_fd: owned,
-            initially_quiesced,
-            on_read: Box::new(move |bytes| {
-                read_tx
-                    .send(Bytes::copy_from_slice(bytes))
-                    .expect("read callback receiver alive");
-                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)
-    }
-
-    #[test]
-    fn actor_writes_user_input_to_owned_fd() {
-        let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
-
-        handle
-            .try_write_user_input(Bytes::from_static(b"hello"))
-            .expect("write command accepted");
-
-        let mut buf = [0u8; 5];
-        peer.read_exact(&mut buf).expect("peer receives write");
-        assert_eq!(&buf, b"hello");
-        handle.shutdown();
-    }
-
-    #[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);
-
-        peer.write_all(b"from-peer").expect("peer write");
-
-        let read = read_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("actor read callback");
-        assert_eq!(read, Bytes::from_static(b"from-peer"));
-        handle.shutdown();
-    }
-
-    #[test]
-    fn begin_handoff_stops_reads_and_rejects_user_writes_until_rollback() {
-        let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
-
-        handle
-            .begin_handoff(Duration::from_secs(1))
-            .expect("handoff quiesced");
-        assert!(handle
-            .try_write_user_input(Bytes::from_static(b"blocked"))
-            .is_err());
-
-        peer.write_all(b"held").expect("peer write during quiesce");
-        assert!(
-            read_rx.recv_timeout(Duration::from_millis(150)).is_err(),
-            "actor must not read while quiesced"
-        );
-
-        handle.rollback_handoff().expect("rollback resumes actor");
-        let read = read_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("actor reads held bytes after rollback");
-        assert_eq!(read, Bytes::from_static(b"held"));
-
-        handle
-            .try_write_user_input(Bytes::from_static(b"after"))
-            .expect("write accepted after rollback");
-        let mut buf = [0u8; 5];
-        peer.read_exact(&mut buf).expect("peer receives after");
-        assert_eq!(&buf, b"after");
-        handle.shutdown();
-    }
-
-    #[test]
-    fn duplicate_for_handoff_requires_quiesced_actor() {
-        let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
-
-        assert!(handle.duplicate_for_handoff().is_err());
-        handle
-            .begin_handoff(Duration::from_secs(1))
-            .expect("handoff quiesced");
-        let duplicate = handle
-            .duplicate_for_handoff()
-            .expect("handoff duplicate created");
-        assert!(duplicate >= 0);
-        unsafe {
-            libc::close(duplicate);
-        }
-        handle.rollback_handoff().expect("rollback resumes actor");
-
-        peer.write_all(b"still-live").expect("peer write");
-        let read = read_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("actor still reads after duplicate closes");
-        assert_eq!(read, Bytes::from_static(b"still-live"));
-        handle.shutdown();
-    }
-
-    #[test]
-    fn resize_and_nudge_keep_latest_request_when_command_queue_is_full() {
-        let (data_tx, _data_rx) = mpsc::channel(1);
-        let (control_tx, _control_rx) = std_mpsc::channel();
-        data_tx
-            .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),
-        };
-
-        handle.resize(20, 80, 8, 16, vec![Bytes::from_static(b"old")]);
-        handle.resize(40, 120, 9, 18, vec![Bytes::from_static(b"new")]);
-        handle.nudge_child_redraw_after_handoff(41, 121, 10, 20);
-
-        let controls = controls.lock().expect("controls lock");
-        assert_eq!(
-            controls.resize,
-            Some(PtyResizeRequest {
-                resize: PtyResize {
-                    rows: 40,
-                    cols: 120,
-                    cell_width_px: 9,
-                    cell_height_px: 18,
-                },
-                terminal_responses: vec![Bytes::from_static(b"new")],
+            Ok(PtyIoActorHandle {
+                data_tx,
+                control_tx,
+                accepting,
             })
-        );
-        assert_eq!(
-            controls.nudge,
-            Some(PtyResize {
-                rows: 41,
-                cols: 121,
-                cell_width_px: 10,
-                cell_height_px: 20,
-            })
-        );
-    }
-
-    #[test]
-    fn resize_writes_terminal_responses_after_applying_resize() {
-        let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
-        let response = Bytes::from_static(b"\x1B[48;40;100;720;900t");
-
-        handle.resize(40, 100, 9, 18, vec![response.clone()]);
-
-        let mut buf = vec![0; response.len()];
-        peer.read_exact(&mut buf)
-            .expect("peer receives resize response");
-        assert_eq!(Bytes::from(buf), response);
-        handle.shutdown();
-    }
-
-    #[tokio::test]
-    async fn async_user_input_waits_for_queue_capacity() {
-        let (data_tx, mut data_rx) = mpsc::channel(1);
-        let (control_tx, _control_rx) = std_mpsc::channel();
-        data_tx
-            .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())),
-        };
-
-        let write = tokio::spawn(async move {
-            handle
-                .write_user_input(Bytes::from_static(b"wait-for-capacity"))
-                .await
-        });
-        tokio::time::sleep(Duration::from_millis(50)).await;
-        assert!(
-            !write.is_finished(),
-            "async input should wait for queue capacity"
-        );
-
-        assert!(matches!(
-            data_rx.recv().await,
-            Some(PtyIoDataCommand::WriteUserInput(_))
-        ));
-        write
-            .await
-            .expect("write task joins")
-            .expect("write succeeds after capacity opens");
-        match data_rx.recv().await {
-            Some(PtyIoDataCommand::WriteUserInput(bytes)) => {
-                assert_eq!(bytes, Bytes::from_static(b"wait-for-capacity"));
-            }
-            _ => panic!("expected queued user input"),
-        }
-    }
-
-    #[tokio::test]
-    async fn async_user_input_waiting_for_capacity_is_rejected_after_handoff_begins() {
-        let (data_tx, mut data_rx) = mpsc::channel(1);
-        let (control_tx, control_rx) = std_mpsc::channel();
-        data_tx
-            .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())),
-        };
-        let write_handle = handle.clone();
-        let write = tokio::spawn(async move {
-            write_handle
-                .write_user_input(Bytes::from_static(b"after-handoff-start"))
-                .await
-        });
-        tokio::time::sleep(Duration::from_millis(50)).await;
-
-        let handoff = std::thread::spawn(move || handle.begin_handoff(Duration::from_secs(1)));
-        match control_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("handoff control command")
-        {
-            PtyIoControlCommand::BeginHandoff(reply) => {
-                reply.send(Ok(())).expect("handoff waiter alive");
-            }
-            _ => panic!("expected begin handoff command"),
-        }
-        handoff
-            .join()
-            .expect("handoff thread joins")
-            .expect("handoff succeeds");
-        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",
-        );
-        assert_eq!(err.0, Bytes::from_static(b"after-handoff-start"));
-        match tokio::time::timeout(Duration::from_millis(50), data_rx.recv()).await {
-            Err(_) | Ok(None) => {}
-            Ok(Some(_)) => panic!("rejected write must not be queued"),
         }
     }
 
-    #[test]
-    fn handoff_control_is_not_blocked_by_full_data_queue() {
-        let (data_tx, _data_rx) = mpsc::channel(1);
-        let (control_tx, control_rx) = std_mpsc::channel();
-        data_tx
-            .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())),
-        };
-
-        let handoff = std::thread::spawn(move || handle.begin_handoff(Duration::from_secs(1)));
-        match control_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("handoff control command")
-        {
-            PtyIoControlCommand::BeginHandoff(reply) => {
-                reply.send(Ok(())).expect("handoff waiter alive");
-            }
-            _ => panic!("expected begin handoff command"),
-        }
-
-        handoff
-            .join()
-            .expect("handoff thread joins")
-            .expect("handoff succeeds despite full data queue");
-    }
-
-    #[test]
-    fn begin_handoff_drains_user_writes_already_in_command_queue() {
-        let (actor_socket, mut peer) = UnixStream::pair().expect("socket pair");
-        actor_socket
-            .set_nonblocking(true)
-            .expect("actor socket nonblocking");
-        peer.set_read_timeout(Some(Duration::from_secs(1)))
-            .expect("peer timeout");
-        let (data_tx, data_rx) = mpsc::channel(ACTOR_COMMAND_BUFFER);
-        let (_control_tx, control_rx) = std_mpsc::channel();
-        data_tx
-            .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
-                b"queued-before-ack",
-            )))
-            .expect("queued write");
-        let mut runner = PtyIoActorRunner {
-            pane_id: 1,
-            file: std::fs::File::from(unsafe { OwnedFd::from_raw_fd(actor_socket.into_raw_fd()) }),
-            data_rx,
-            control_rx,
-            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");
-
-        let mut buf = [0u8; 17];
-        peer.read_exact(&mut buf)
-            .expect("queued write reaches peer before quiesce ack");
-        assert_eq!(&buf, b"queued-before-ack");
-        assert_eq!(runner.state, ActorState::Quiesced);
+    fn write_all_locked(
+        writer: &Arc<Mutex<Box<dyn Write + Send>>>,
+        bytes: &[u8],
+    ) -> std::io::Result<()> {
+        let mut writer = writer
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        writer.write_all(bytes)?;
+        writer.flush()
     }
 
-    #[test]
-    fn release_after_commit_prevents_further_io() {
-        let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
-
-        handle.release_after_commit().expect("actor released");
-        assert!(handle
-            .try_write_user_input(Bytes::from_static(b"blocked"))
-            .is_err());
-
-        let _ = peer.write_all(b"ignored");
-        assert!(read_rx.recv_timeout(Duration::from_millis(150)).is_err());
-    }
+    #[allow(dead_code)]
+    fn _assert_duration_send(_: Duration) {}
 }
+
+#[cfg(windows)]
+pub(crate) use windows::*;
diff --git a/src/pty/actor/unix.rs b/src/pty/actor/unix.rs
new file mode 100644
index 00000000..70634475
--- /dev/null
+++ b/src/pty/actor/unix.rs
@@ -0,0 +1,1172 @@
+use std::{
+    collections::VecDeque,
+    io::{Read, Write},
+    os::fd::{AsRawFd, OwnedFd, RawFd},
+    sync::{mpsc as std_mpsc, Arc, Mutex},
+    time::{Duration, Instant},
+};
+
+use bytes::Bytes;
+use tokio::sync::mpsc::{self, error::TryRecvError as DataTryRecvError};
+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);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum ActorState {
+    Running,
+    Quiesced,
+    Released,
+}
+
+pub(crate) struct PtyReadResult {
+    pub terminal_responses: Vec<Bytes>,
+}
+
+impl PtyReadResult {
+    #[cfg(test)]
+    pub(crate) fn empty() -> Self {
+        Self {
+            terminal_responses: Vec::new(),
+        }
+    }
+}
+
+type ReadCallback = Box<dyn FnMut(&[u8]) -> PtyReadResult + Send + 'static>;
+type ReaderExitCallback = Box<dyn FnOnce() + Send + 'static>;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PtyResize {
+    rows: u16,
+    cols: u16,
+    cell_width_px: u32,
+    cell_height_px: u32,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct PtyResizeRequest {
+    resize: PtyResize,
+    terminal_responses: Vec<Bytes>,
+}
+
+#[derive(Default)]
+struct SharedPtyControls {
+    resize: Option<PtyResizeRequest>,
+    nudge: Option<PtyResize>,
+}
+
+pub(crate) struct PtyIoActorConfig {
+    pub pane_id: u32,
+    pub master_fd: OwnedFd,
+    pub initially_quiesced: bool,
+    pub on_read: ReadCallback,
+    pub on_reader_exit: Option<ReaderExitCallback>,
+}
+
+enum PtyIoDataCommand {
+    WriteUserInput(Bytes),
+}
+
+enum PtyIoControlCommand {
+    BeginHandoff(std_mpsc::Sender<std::io::Result<()>>),
+    DuplicateForHandoff(std_mpsc::Sender<std::io::Result<RawFd>>),
+    ForegroundProcessGroup(std_mpsc::Sender<Option<u32>>),
+    RollbackHandoff(std_mpsc::Sender<std::io::Result<()>>),
+    ReleaseAfterCommit(std_mpsc::Sender<std::io::Result<()>>),
+    Shutdown,
+}
+
+#[derive(Clone)]
+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>>,
+}
+
+#[derive(Debug)]
+struct UserWriteGate {
+    accepting: bool,
+}
+
+impl PtyIoActorHandle {
+    pub(crate) async fn write_user_input(
+        &self,
+        bytes: Bytes,
+    ) -> Result<(), mpsc::error::SendError<Bytes>> {
+        {
+            let user_writes = self
+                .user_writes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if !user_writes.accepting {
+                return Err(mpsc::error::SendError(bytes));
+            }
+        }
+
+        let permit = match self.data_tx.reserve().await {
+            Ok(permit) => permit,
+            Err(_) => return Err(mpsc::error::SendError(bytes)),
+        };
+
+        let user_writes = self
+            .user_writes
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if !user_writes.accepting {
+            return Err(mpsc::error::SendError(bytes));
+        }
+        permit.send(PtyIoDataCommand::WriteUserInput(bytes));
+        self.wake_actor();
+        Ok(())
+    }
+
+    pub(crate) fn try_write_user_input(
+        &self,
+        bytes: Bytes,
+    ) -> Result<(), mpsc::error::TrySendError<Bytes>> {
+        let user_writes = self
+            .user_writes
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if !user_writes.accepting {
+            return Err(mpsc::error::TrySendError::Closed(bytes));
+        }
+        match self
+            .data_tx
+            .try_send(PtyIoDataCommand::WriteUserInput(bytes))
+        {
+            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,
+        terminal_responses: Vec<Bytes>,
+    ) {
+        {
+            let mut controls = self
+                .controls
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            controls.resize = Some(PtyResizeRequest {
+                resize: PtyResize {
+                    rows,
+                    cols,
+                    cell_width_px,
+                    cell_height_px,
+                },
+                terminal_responses,
+            });
+        }
+        self.wake_actor();
+    }
+
+    pub(crate) fn nudge_child_redraw_after_handoff(
+        &self,
+        rows: u16,
+        cols: u16,
+        cell_width_px: u32,
+        cell_height_px: u32,
+    ) {
+        {
+            let mut controls = self
+                .controls
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            controls.nudge = Some(PtyResize {
+                rows,
+                cols,
+                cell_width_px,
+                cell_height_px,
+            });
+        }
+        self.wake_actor();
+    }
+
+    pub(crate) fn begin_handoff(&self, timeout: Duration) -> std::io::Result<()> {
+        let (reply_tx, reply_rx) = std_mpsc::channel();
+        {
+            let mut user_writes = self
+                .user_writes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            user_writes.accepting = false;
+            if self
+                .control_tx
+                .send(PtyIoControlCommand::BeginHandoff(reply_tx))
+                .is_err()
+            {
+                user_writes.accepting = true;
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::BrokenPipe,
+                    "pty actor closed",
+                ));
+            }
+            self.wake_actor();
+        }
+        match reply_rx.recv_timeout(timeout) {
+            Ok(Ok(())) => Ok(()),
+            Ok(Err(err)) => {
+                let _ = self.rollback_handoff();
+                Err(err)
+            }
+            Err(_) => {
+                let _ = self.rollback_handoff();
+                Err(std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    "timed out waiting for PTY actor to quiesce",
+                ))
+            }
+        }
+    }
+
+    pub(crate) fn duplicate_for_handoff(&self) -> std::io::Result<RawFd> {
+        let (reply_tx, reply_rx) = std_mpsc::channel();
+        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,
+                "timed out waiting for PTY handoff duplicate",
+            )
+        })?
+    }
+
+    pub(crate) fn foreground_process_group_id(&self) -> Option<u32> {
+        let (reply_tx, reply_rx) = std_mpsc::channel();
+        self.control_tx
+            .send(PtyIoControlCommand::ForegroundProcessGroup(reply_tx))
+            .ok()?;
+        self.wake_actor();
+        reply_rx.recv_timeout(Duration::from_secs(1)).ok()?
+    }
+
+    pub(crate) fn rollback_handoff(&self) -> std::io::Result<()> {
+        let (reply_tx, reply_rx) = std_mpsc::channel();
+        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,
+                "timed out waiting for PTY handoff rollback",
+            )
+        })?;
+        if result.is_ok() {
+            let mut user_writes = self
+                .user_writes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            user_writes.accepting = true;
+        }
+        result
+    }
+
+    pub(crate) fn release_after_commit(&self) -> std::io::Result<()> {
+        {
+            let mut user_writes = self
+                .user_writes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            user_writes.accepting = false;
+        }
+        let (reply_tx, reply_rx) = std_mpsc::channel();
+        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,
+                "timed out waiting for PTY actor release",
+            )
+        })?
+    }
+
+    pub(crate) fn shutdown(&self) {
+        {
+            let mut user_writes = self
+                .user_writes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            user_writes.accepting = false;
+        }
+        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");
+        }
+    }
+}
+
+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,
+        }));
+        let controls = Arc::new(Mutex::new(SharedPtyControls::default()));
+        let handle = PtyIoActorHandle {
+            data_tx,
+            control_tx,
+            wake: wake_pipe.writer,
+            user_writes,
+            controls: Arc::clone(&controls),
+        };
+
+        let mut runner = PtyIoActorRunner {
+            pane_id: config.pane_id,
+            file: std::fs::File::from(config.master_fd),
+            data_rx,
+            control_rx,
+            state: if config.initially_quiesced {
+                ActorState::Quiesced
+            } else {
+                ActorState::Running
+            },
+            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))
+            .spawn(move || runner.run())
+            .map_err(|err| std::io::Error::other(err.to_string()))?;
+
+        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 {
+    pane_id: u32,
+    file: std::fs::File,
+    data_rx: mpsc::Receiver<PtyIoDataCommand>,
+    control_rx: std_mpsc::Receiver<PtyIoControlCommand>,
+    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 {
+    fn run(&mut self) {
+        let mut should_exit = false;
+        while !should_exit {
+            should_exit = self.drain_commands();
+            if should_exit || self.state == ActorState::Released {
+                break;
+            }
+
+            self.apply_pending_controls();
+
+            if !self.pending_writes.is_empty() {
+                self.flush_pending_writes_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();
+                    }
+                    if self.state == ActorState::Running
+                        && readiness.pty_read_ready
+                        && !self.read_once()
+                    {
+                        break;
+                    }
+                }
+                Err(err) => {
+                    debug!(pane = self.pane_id, err = %err, "PTY actor poll failed");
+                    break;
+                }
+            }
+        }
+
+        if let Some(on_reader_exit) = self.on_reader_exit.take() {
+            on_reader_exit();
+        }
+        debug!(pane = self.pane_id, "PTY actor exiting");
+    }
+
+    fn drain_commands(&mut self) -> bool {
+        if self.drain_control_commands() {
+            return true;
+        }
+        self.drain_data_commands()
+    }
+
+    fn drain_control_commands(&mut self) -> bool {
+        let mut should_exit = false;
+        loop {
+            match self.control_rx.try_recv() {
+                Ok(command) => {
+                    if self.handle_control_command(command) {
+                        should_exit = true;
+                        break;
+                    }
+                }
+                Err(std_mpsc::TryRecvError::Empty) => break,
+                Err(std_mpsc::TryRecvError::Disconnected) => {
+                    should_exit = true;
+                    break;
+                }
+            }
+        }
+        should_exit
+    }
+
+    fn drain_data_commands(&mut self) -> bool {
+        let mut should_exit = false;
+        loop {
+            match self.data_rx.try_recv() {
+                Ok(command) => {
+                    if self.handle_data_command(command) {
+                        should_exit = true;
+                        break;
+                    }
+                }
+                Err(DataTryRecvError::Empty) => break,
+                Err(DataTryRecvError::Disconnected) => {
+                    should_exit = true;
+                    break;
+                }
+            }
+        }
+        should_exit
+    }
+
+    fn handle_data_command(&mut self, command: PtyIoDataCommand) -> bool {
+        match command {
+            PtyIoDataCommand::WriteUserInput(bytes) => {
+                if self.state == ActorState::Running {
+                    self.pending_writes.push_back(bytes);
+                }
+            }
+        }
+        false
+    }
+
+    fn handle_control_command(&mut self, command: PtyIoControlCommand) -> bool {
+        match command {
+            PtyIoControlCommand::BeginHandoff(reply) => {
+                let result = self.begin_handoff();
+                let _ = reply.send(result);
+            }
+            PtyIoControlCommand::DuplicateForHandoff(reply) => {
+                let result = if self.state == ActorState::Quiesced {
+                    fd::duplicate_cloexec_fd(self.file.as_raw_fd())
+                } else {
+                    Err(std::io::Error::other(
+                        "PTY actor must be quiesced before handoff duplication",
+                    ))
+                };
+                let _ = reply.send(result);
+            }
+            PtyIoControlCommand::ForegroundProcessGroup(reply) => {
+                let result =
+                    crate::platform::foreground_process_group_id_for_tty_fd(self.file.as_raw_fd());
+                let _ = reply.send(result);
+            }
+            PtyIoControlCommand::RollbackHandoff(reply) => {
+                let result = if self.state == ActorState::Released {
+                    Err(std::io::Error::new(
+                        std::io::ErrorKind::BrokenPipe,
+                        "PTY actor was released before handoff rollback",
+                    ))
+                } else {
+                    self.state = ActorState::Running;
+                    Ok(())
+                };
+                let _ = reply.send(result);
+            }
+            PtyIoControlCommand::ReleaseAfterCommit(reply) => {
+                self.state = ActorState::Released;
+                self.pending_writes.clear();
+                let _ = reply.send(Ok(()));
+                return true;
+            }
+            PtyIoControlCommand::Shutdown => return true,
+        }
+        false
+    }
+
+    fn begin_handoff(&mut self) -> std::io::Result<()> {
+        self.drain_pre_quiesce_commands();
+        self.apply_pending_controls();
+        if self.state == ActorState::Released {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                "PTY actor was released before handoff quiesce",
+            ));
+        }
+        let deadline = Instant::now() + HANDOFF_DRAIN_TIMEOUT;
+        while !self.pending_writes.is_empty() {
+            if Instant::now() >= deadline {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    "timed out draining PTY writes before handoff",
+                ));
+            }
+            self.flush_pending_writes_once();
+        }
+        self.state = ActorState::Quiesced;
+        Ok(())
+    }
+
+    fn drain_pre_quiesce_commands(&mut self) {
+        while let Ok(PtyIoDataCommand::WriteUserInput(bytes)) = self.data_rx.try_recv() {
+            if self.state != ActorState::Released {
+                self.pending_writes.push_back(bytes);
+            }
+        }
+    }
+
+    fn apply_pending_controls(&mut self) {
+        let (resize, nudge) = {
+            let mut controls = self
+                .controls
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            (controls.resize.take(), controls.nudge.take())
+        };
+        if self.state == ActorState::Released {
+            return;
+        }
+        if let Some(request) = resize {
+            self.resize(request.resize);
+            self.enqueue_terminal_responses(request.terminal_responses);
+        }
+        if let Some(nudge) = nudge {
+            self.nudge(nudge);
+        }
+    }
+
+    fn read_once(&mut self) -> bool {
+        let mut buf = [0u8; 8192];
+        match self.file.read(&mut buf) {
+            Ok(0) => false,
+            Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => true,
+            Err(err) if err.kind() == std::io::ErrorKind::Interrupted => true,
+            Err(err) => {
+                debug!(pane = self.pane_id, err = %err, "PTY actor read failed");
+                false
+            }
+            Ok(n) => {
+                let result = (self.on_read)(&buf[..n]);
+                self.enqueue_terminal_responses(result.terminal_responses);
+                true
+            }
+        }
+    }
+
+    fn enqueue_terminal_responses(&mut self, terminal_responses: Vec<Bytes>) {
+        if self.state == ActorState::Released {
+            return;
+        }
+        self.pending_writes.extend(terminal_responses);
+    }
+
+    fn flush_pending_writes_once(&mut self) {
+        while let Some(bytes) = self.pending_writes.front() {
+            let chunk = &bytes[self.current_write_offset..];
+            match self.file.write(chunk) {
+                Ok(0) => {
+                    warn!(pane = self.pane_id, "PTY actor write returned zero bytes");
+                    return;
+                }
+                Ok(written) => {
+                    self.current_write_offset += written;
+                    if self.current_write_offset >= bytes.len() {
+                        self.pending_writes.pop_front();
+                        self.current_write_offset = 0;
+                    }
+                }
+                Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
+                    let _ = fd::poll_write_ready(self.file.as_raw_fd(), ACTOR_POLL_MS);
+                    return;
+                }
+                Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return,
+                Err(err) => {
+                    warn!(pane = self.pane_id, err = %err, "PTY actor write failed");
+                    self.pending_writes.clear();
+                    self.current_write_offset = 0;
+                    return;
+                }
+            }
+        }
+        let _ = self.file.flush();
+    }
+
+    fn resize(&self, resize: PtyResize) {
+        self.log_resize_result(fd::resize_pty_fd(
+            self.file.as_raw_fd(),
+            resize.rows,
+            resize.cols,
+            resize.cell_width_px,
+            resize.cell_height_px,
+        ));
+    }
+
+    fn nudge(&mut self, resize: PtyResize) {
+        if self.state == ActorState::Released {
+            return;
+        }
+        let nudge = if resize.rows > 2 {
+            (
+                resize.rows - 1,
+                resize.cols,
+                resize.cell_width_px,
+                resize.cell_height_px,
+            )
+        } else {
+            (
+                resize.rows,
+                resize.cols.saturating_sub(1).max(4),
+                resize.cell_width_px,
+                resize.cell_height_px,
+            )
+        };
+        if nudge
+            == (
+                resize.rows,
+                resize.cols,
+                resize.cell_width_px,
+                resize.cell_height_px,
+            )
+        {
+            return;
+        }
+        self.log_resize_result(fd::resize_pty_fd(
+            self.file.as_raw_fd(),
+            nudge.0,
+            nudge.1,
+            nudge.2,
+            nudge.3,
+        ));
+        std::thread::sleep(Duration::from_millis(30));
+        self.log_resize_result(fd::resize_pty_fd(
+            self.file.as_raw_fd(),
+            resize.rows,
+            resize.cols,
+            resize.cell_width_px,
+            resize.cell_height_px,
+        ));
+    }
+
+    fn log_resize_result(&self, result: std::io::Result<()>) {
+        if let Err(err) = result {
+            debug!(pane = self.pane_id, err = %err, "PTY resize failed");
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::{
+        io::{Read, Write},
+        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)
+            .expect("actor socket nonblocking");
+        peer.set_read_timeout(Some(Duration::from_secs(1)))
+            .expect("peer timeout");
+        let owned = unsafe { OwnedFd::from_raw_fd(actor_socket.into_raw_fd()) };
+        let (read_tx, read_rx) = std_mpsc::channel();
+        let config = PtyIoActorConfig {
+            pane_id: 1,
+            master_fd: owned,
+            initially_quiesced,
+            on_read: Box::new(move |bytes| {
+                read_tx
+                    .send(Bytes::copy_from_slice(bytes))
+                    .expect("read callback receiver alive");
+                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)
+    }
+
+    #[test]
+    fn actor_writes_user_input_to_owned_fd() {
+        let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
+
+        handle
+            .try_write_user_input(Bytes::from_static(b"hello"))
+            .expect("write command accepted");
+
+        let mut buf = [0u8; 5];
+        peer.read_exact(&mut buf).expect("peer receives write");
+        assert_eq!(&buf, b"hello");
+        handle.shutdown();
+    }
+
+    #[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);
+
+        peer.write_all(b"from-peer").expect("peer write");
+
+        let read = read_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("actor read callback");
+        assert_eq!(read, Bytes::from_static(b"from-peer"));
+        handle.shutdown();
+    }
+
+    #[test]
+    fn begin_handoff_stops_reads_and_rejects_user_writes_until_rollback() {
+        let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
+
+        handle
+            .begin_handoff(Duration::from_secs(1))
+            .expect("handoff quiesced");
+        assert!(handle
+            .try_write_user_input(Bytes::from_static(b"blocked"))
+            .is_err());
+
+        peer.write_all(b"held").expect("peer write during quiesce");
+        assert!(
+            read_rx.recv_timeout(Duration::from_millis(150)).is_err(),
+            "actor must not read while quiesced"
+        );
+
+        handle.rollback_handoff().expect("rollback resumes actor");
+        let read = read_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("actor reads held bytes after rollback");
+        assert_eq!(read, Bytes::from_static(b"held"));
+
+        handle
+            .try_write_user_input(Bytes::from_static(b"after"))
+            .expect("write accepted after rollback");
+        let mut buf = [0u8; 5];
+        peer.read_exact(&mut buf).expect("peer receives after");
+        assert_eq!(&buf, b"after");
+        handle.shutdown();
+    }
+
+    #[test]
+    fn duplicate_for_handoff_requires_quiesced_actor() {
+        let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
+
+        assert!(handle.duplicate_for_handoff().is_err());
+        handle
+            .begin_handoff(Duration::from_secs(1))
+            .expect("handoff quiesced");
+        let duplicate = handle
+            .duplicate_for_handoff()
+            .expect("handoff duplicate created");
+        assert!(duplicate >= 0);
+        unsafe {
+            libc::close(duplicate);
+        }
+        handle.rollback_handoff().expect("rollback resumes actor");
+
+        peer.write_all(b"still-live").expect("peer write");
+        let read = read_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("actor still reads after duplicate closes");
+        assert_eq!(read, Bytes::from_static(b"still-live"));
+        handle.shutdown();
+    }
+
+    #[test]
+    fn resize_and_nudge_keep_latest_request_when_command_queue_is_full() {
+        let (data_tx, _data_rx) = mpsc::channel(1);
+        let (control_tx, _control_rx) = std_mpsc::channel();
+        data_tx
+            .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),
+        };
+
+        handle.resize(20, 80, 8, 16, vec![Bytes::from_static(b"old")]);
+        handle.resize(40, 120, 9, 18, vec![Bytes::from_static(b"new")]);
+        handle.nudge_child_redraw_after_handoff(41, 121, 10, 20);
+
+        let controls = controls.lock().expect("controls lock");
+        assert_eq!(
+            controls.resize,
+            Some(PtyResizeRequest {
+                resize: PtyResize {
+                    rows: 40,
+                    cols: 120,
+                    cell_width_px: 9,
+                    cell_height_px: 18,
+                },
+                terminal_responses: vec![Bytes::from_static(b"new")],
+            })
+        );
+        assert_eq!(
+            controls.nudge,
+            Some(PtyResize {
+                rows: 41,
+                cols: 121,
+                cell_width_px: 10,
+                cell_height_px: 20,
+            })
+        );
+    }
+
+    #[test]
+    fn resize_writes_terminal_responses_after_applying_resize() {
+        let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
+        let response = Bytes::from_static(b"\x1B[48;40;100;720;900t");
+
+        handle.resize(40, 100, 9, 18, vec![response.clone()]);
+
+        let mut buf = vec![0; response.len()];
+        peer.read_exact(&mut buf)
+            .expect("peer receives resize response");
+        assert_eq!(Bytes::from(buf), response);
+        handle.shutdown();
+    }
+
+    #[tokio::test]
+    async fn async_user_input_waits_for_queue_capacity() {
+        let (data_tx, mut data_rx) = mpsc::channel(1);
+        let (control_tx, _control_rx) = std_mpsc::channel();
+        data_tx
+            .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())),
+        };
+
+        let write = tokio::spawn(async move {
+            handle
+                .write_user_input(Bytes::from_static(b"wait-for-capacity"))
+                .await
+        });
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        assert!(
+            !write.is_finished(),
+            "async input should wait for queue capacity"
+        );
+
+        assert!(matches!(
+            data_rx.recv().await,
+            Some(PtyIoDataCommand::WriteUserInput(_))
+        ));
+        write
+            .await
+            .expect("write task joins")
+            .expect("write succeeds after capacity opens");
+        match data_rx.recv().await {
+            Some(PtyIoDataCommand::WriteUserInput(bytes)) => {
+                assert_eq!(bytes, Bytes::from_static(b"wait-for-capacity"));
+            }
+            _ => panic!("expected queued user input"),
+        }
+    }
+
+    #[tokio::test]
+    async fn async_user_input_waiting_for_capacity_is_rejected_after_handoff_begins() {
+        let (data_tx, mut data_rx) = mpsc::channel(1);
+        let (control_tx, control_rx) = std_mpsc::channel();
+        data_tx
+            .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())),
+        };
+        let write_handle = handle.clone();
+        let write = tokio::spawn(async move {
+            write_handle
+                .write_user_input(Bytes::from_static(b"after-handoff-start"))
+                .await
+        });
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        let handoff = std::thread::spawn(move || handle.begin_handoff(Duration::from_secs(1)));
+        match control_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("handoff control command")
+        {
+            PtyIoControlCommand::BeginHandoff(reply) => {
+                reply.send(Ok(())).expect("handoff waiter alive");
+            }
+            _ => panic!("expected begin handoff command"),
+        }
+        handoff
+            .join()
+            .expect("handoff thread joins")
+            .expect("handoff succeeds");
+        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",
+        );
+        assert_eq!(err.0, Bytes::from_static(b"after-handoff-start"));
+        match tokio::time::timeout(Duration::from_millis(50), data_rx.recv()).await {
+            Err(_) | Ok(None) => {}
+            Ok(Some(_)) => panic!("rejected write must not be queued"),
+        }
+    }
+
+    #[test]
+    fn handoff_control_is_not_blocked_by_full_data_queue() {
+        let (data_tx, _data_rx) = mpsc::channel(1);
+        let (control_tx, control_rx) = std_mpsc::channel();
+        data_tx
+            .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())),
+        };
+
+        let handoff = std::thread::spawn(move || handle.begin_handoff(Duration::from_secs(1)));
+        match control_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("handoff control command")
+        {
+            PtyIoControlCommand::BeginHandoff(reply) => {
+                reply.send(Ok(())).expect("handoff waiter alive");
+            }
+            _ => panic!("expected begin handoff command"),
+        }
+
+        handoff
+            .join()
+            .expect("handoff thread joins")
+            .expect("handoff succeeds despite full data queue");
+    }
+
+    #[test]
+    fn begin_handoff_drains_user_writes_already_in_command_queue() {
+        let (actor_socket, mut peer) = UnixStream::pair().expect("socket pair");
+        actor_socket
+            .set_nonblocking(true)
+            .expect("actor socket nonblocking");
+        peer.set_read_timeout(Some(Duration::from_secs(1)))
+            .expect("peer timeout");
+        let (data_tx, data_rx) = mpsc::channel(ACTOR_COMMAND_BUFFER);
+        let (_control_tx, control_rx) = std_mpsc::channel();
+        data_tx
+            .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
+                b"queued-before-ack",
+            )))
+            .expect("queued write");
+        let mut runner = PtyIoActorRunner {
+            pane_id: 1,
+            file: std::fs::File::from(unsafe { OwnedFd::from_raw_fd(actor_socket.into_raw_fd()) }),
+            data_rx,
+            control_rx,
+            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");
+
+        let mut buf = [0u8; 17];
+        peer.read_exact(&mut buf)
+            .expect("queued write reaches peer before quiesce ack");
+        assert_eq!(&buf, b"queued-before-ack");
+        assert_eq!(runner.state, ActorState::Quiesced);
+    }
+
+    #[test]
+    fn release_after_commit_prevents_further_io() {
+        let (handle, mut peer, read_rx) = actor_with_socket_pair(false);
+
+        handle.release_after_commit().expect("actor released");
+        assert!(handle
+            .try_write_user_input(Bytes::from_static(b"blocked"))
+            .is_err());
+
+        let _ = peer.write_all(b"ignored");
+        assert!(read_rx.recv_timeout(Duration::from_millis(150)).is_err());
+    }
+}
diff --git a/src/pty/backend.rs b/src/pty/backend.rs
index 5871004a..d09922b7 100644
--- a/src/pty/backend.rs
+++ b/src/pty/backend.rs
@@ -1,14 +1,19 @@
-use std::os::fd::{FromRawFd, OwnedFd};
+#[cfg(unix)]
+mod unix;
 
-use portable_pty::{native_pty_system, Child, CommandBuilder, PtySize};
+#[cfg(unix)]
+pub(crate) use unix::*;
 
-use crate::pty::fd;
+#[cfg(windows)]
+use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize};
 
+#[cfg(windows)]
 pub(crate) struct SpawnedPty {
-    pub master_fd: OwnedFd,
+    pub master: Box<dyn MasterPty + Send>,
     pub child: Box<dyn Child + Send + Sync>,
 }
 
+#[cfg(windows)]
 pub(crate) fn spawn_with_portable_pty(
     rows: u16,
     cols: u16,
@@ -23,72 +28,13 @@ pub(crate) fn spawn_with_portable_pty(
             pixel_height: 0,
         })
         .map_err(|err| std::io::Error::other(err.to_string()))?;
-    let master_fd = pair
-        .master
-        .as_raw_fd()
-        .ok_or_else(|| std::io::Error::other("pty master fd is unavailable"))?;
-    let actor_fd = fd::duplicate_cloexec_fd(master_fd)?;
-    let actor_fd = unsafe { OwnedFd::from_raw_fd(actor_fd) };
     let child = pair
         .slave
         .spawn_command(cmd)
         .map_err(|err| std::io::Error::other(err.to_string()))?;
-    drop(pair);
 
     Ok(SpawnedPty {
-        master_fd: actor_fd,
+        master: pair.master,
         child,
     })
 }
-
-#[cfg(all(test, target_os = "linux"))]
-mod tests {
-    use super::*;
-    use std::sync::{Mutex, OnceLock};
-
-    fn pty_fd_test_lock() -> &'static Mutex<()> {
-        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
-        LOCK.get_or_init(|| Mutex::new(()))
-    }
-
-    fn parent_pty_fd_targets() -> Vec<String> {
-        let Ok(entries) = std::fs::read_dir("/proc/self/fd") else {
-            return Vec::new();
-        };
-        let mut targets: Vec<String> = entries
-            .filter_map(Result::ok)
-            .filter_map(|entry| std::fs::read_link(entry.path()).ok())
-            .map(|target| target.to_string_lossy().into_owned())
-            .filter(|target| target.starts_with("/dev/pts/") || target == "/dev/ptmx")
-            .collect();
-        targets.sort();
-        targets
-    }
-
-    fn parent_pty_fd_count() -> usize {
-        parent_pty_fd_targets().len()
-    }
-
-    #[test]
-    fn portable_pty_setup_leaves_one_parent_pty_fd() {
-        let _guard = pty_fd_test_lock().lock().expect("pty fd test lock");
-        let before = parent_pty_fd_count();
-        let mut cmd = CommandBuilder::new("/bin/cat");
-        cmd.env(crate::HERDR_ENV_VAR, crate::HERDR_ENV_VALUE);
-
-        let mut spawned =
-            spawn_with_portable_pty(24, 80, cmd).expect("portable pty setup succeeds");
-        let after_spawn = parent_pty_fd_count();
-
-        assert_eq!(
-            after_spawn,
-            before + 1,
-            "portable-pty setup should leave only the Herdr-owned master fd in the parent: {:?}",
-            parent_pty_fd_targets()
-        );
-
-        let _ = spawned.child.kill();
-        let _ = spawned.child.wait();
-        drop(spawned.master_fd);
-    }
-}
diff --git a/src/pty/backend/unix.rs b/src/pty/backend/unix.rs
new file mode 100644
index 00000000..5871004a
--- /dev/null
+++ b/src/pty/backend/unix.rs
@@ -0,0 +1,94 @@
+use std::os::fd::{FromRawFd, OwnedFd};
+
+use portable_pty::{native_pty_system, Child, CommandBuilder, PtySize};
+
+use crate::pty::fd;
+
+pub(crate) struct SpawnedPty {
+    pub master_fd: OwnedFd,
+    pub child: Box<dyn Child + Send + Sync>,
+}
+
+pub(crate) fn spawn_with_portable_pty(
+    rows: u16,
+    cols: u16,
+    cmd: CommandBuilder,
+) -> std::io::Result<SpawnedPty> {
+    let pty_system = native_pty_system();
+    let pair = pty_system
+        .openpty(PtySize {
+            rows,
+            cols,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .map_err(|err| std::io::Error::other(err.to_string()))?;
+    let master_fd = pair
+        .master
+        .as_raw_fd()
+        .ok_or_else(|| std::io::Error::other("pty master fd is unavailable"))?;
+    let actor_fd = fd::duplicate_cloexec_fd(master_fd)?;
+    let actor_fd = unsafe { OwnedFd::from_raw_fd(actor_fd) };
+    let child = pair
+        .slave
+        .spawn_command(cmd)
+        .map_err(|err| std::io::Error::other(err.to_string()))?;
+    drop(pair);
+
+    Ok(SpawnedPty {
+        master_fd: actor_fd,
+        child,
+    })
+}
+
+#[cfg(all(test, target_os = "linux"))]
+mod tests {
+    use super::*;
+    use std::sync::{Mutex, OnceLock};
+
+    fn pty_fd_test_lock() -> &'static Mutex<()> {
+        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+        LOCK.get_or_init(|| Mutex::new(()))
+    }
+
+    fn parent_pty_fd_targets() -> Vec<String> {
+        let Ok(entries) = std::fs::read_dir("/proc/self/fd") else {
+            return Vec::new();
+        };
+        let mut targets: Vec<String> = entries
+            .filter_map(Result::ok)
+            .filter_map(|entry| std::fs::read_link(entry.path()).ok())
+            .map(|target| target.to_string_lossy().into_owned())
+            .filter(|target| target.starts_with("/dev/pts/") || target == "/dev/ptmx")
+            .collect();
+        targets.sort();
+        targets
+    }
+
+    fn parent_pty_fd_count() -> usize {
+        parent_pty_fd_targets().len()
+    }
+
+    #[test]
+    fn portable_pty_setup_leaves_one_parent_pty_fd() {
+        let _guard = pty_fd_test_lock().lock().expect("pty fd test lock");
+        let before = parent_pty_fd_count();
+        let mut cmd = CommandBuilder::new("/bin/cat");
+        cmd.env(crate::HERDR_ENV_VAR, crate::HERDR_ENV_VALUE);
+
+        let mut spawned =
+            spawn_with_portable_pty(24, 80, cmd).expect("portable pty setup succeeds");
+        let after_spawn = parent_pty_fd_count();
+
+        assert_eq!(
+            after_spawn,
+            before + 1,
+            "portable-pty setup should leave only the Herdr-owned master fd in the parent: {:?}",
+            parent_pty_fd_targets()
+        );
+
+        let _ = spawned.child.kill();
+        let _ = spawned.child.wait();
+        drop(spawned.master_fd);
+    }
+}
diff --git a/src/pty/mod.rs b/src/pty/mod.rs
index 74bbb124..ce838679 100644
--- a/src/pty/mod.rs
+++ b/src/pty/mod.rs
@@ -1,6 +1,4 @@
-#[cfg(unix)]
 pub(crate) mod actor;
-#[cfg(unix)]
 pub(crate) mod backend;
 #[cfg(unix)]
 pub(crate) mod fd;