feat(pty): harden Unix PTY lifecycle

crates/splice-pty/src/lib.rs

@@ -215,7 +215,6 @@ pub struct PtySession {
     writer: std::sync::Mutex<Option<Box<dyn std::io::Write + Send>>>,
     killer: std::sync::Mutex<Box<dyn portable_pty::ChildKiller + Send + Sync>>,
     pid: Option<u32>,
-    completed: std::sync::Mutex<std::sync::mpsc::Receiver<()>>,
     reader: std::sync::Mutex<Option<std::thread::JoinHandle<()>>>,
     waiter: std::sync::Mutex<Option<std::thread::JoinHandle<()>>>,
     running: std::sync::Arc<std::sync::atomic::AtomicBool>,
@@ -344,16 +343,20 @@ impl PtySession {
         let id = lifecycle.id();
         let reader_lifecycle = std::sync::Arc::clone(&lifecycle);
 
-        let reader =
-            std::thread::spawn(move || read_output(reader, id, reader_lifecycle, on_output));
+        let (reader_completed_tx, reader_completed) = std::sync::mpsc::channel();
+        let reader = std::thread::spawn(move || {
+            read_output(reader, id, reader_lifecycle, on_output);
+            let _ = reader_completed_tx.send(());
+        });
 
         let waiter_lifecycle = std::sync::Arc::clone(&lifecycle);
         let waiter_running = std::sync::Arc::clone(&running);
-        let (completed_tx, completed) = std::sync::mpsc::channel();
         let waiter = std::thread::spawn(move || {
             let exited = child.wait().is_ok();
             waiter_running.store(false, std::sync::atomic::Ordering::SeqCst);
-            let _ = completed_tx.send(());
+            // A Unix PTY returns EOF/EIO only after the child releases its slave
+            // side. Drain that final output before observers see natural exit.
+            let _ = reader_completed.recv_timeout(std::time::Duration::from_millis(250));
             if exited && waiter_lifecycle.should_emit_natural_exit() {
                 on_exit(waiter_lifecycle.id());
             }
@@ -365,7 +368,6 @@ impl PtySession {
             writer: std::sync::Mutex::new(Some(writer)),
             killer: std::sync::Mutex::new(killer),
             pid,
-            completed: std::sync::Mutex::new(completed),
             reader: std::sync::Mutex::new(Some(reader)),
             waiter: std::sync::Mutex::new(Some(waiter)),
             running,
@@ -395,11 +397,28 @@ impl PtySession {
     }
 
     pub fn interrupt(&self) -> Result<(), PtyError> {
-        self.write("\u{3}")
+        if !self.lifecycle.should_emit_natural_exit() {
+            return Err(PtyError::SessionClosed);
+        }
+        match self.writer.try_lock() {
+            Ok(mut writer) => {
+                let writer = writer.as_mut().ok_or(PtyError::SessionClosed)?;
+                writer.write_all(b"\x03")?;
+                writer.flush()?;
+                Ok(())
+            }
+            Err(std::sync::TryLockError::WouldBlock) if self.signal_process_group(libc::SIGINT) => {
+                Ok(())
+            }
+            Err(_) => Err(PtyError::SessionClosed),
+        }
     }
 
     pub fn resize(&self, size: TerminalSize) -> Result<(), PtyError> {
         use portable_pty::PtySize;
+        if !self.lifecycle.should_emit_natural_exit() {
+            return Err(PtyError::SessionClosed);
+        }
         self.master
             .lock()
             .map_err(|_| PtyError::SessionClosed)?
@@ -427,19 +446,17 @@ impl PtySession {
     pub fn close(&self) {
         if self.lifecycle.begin_close() {
             (self.on_closing)();
-            if let Ok(mut killer) = self.killer.lock() {
-                let _ = killer.kill();
+            for signal in [libc::SIGHUP, libc::SIGTERM, libc::SIGKILL] {
+                let _ = self.signal_process_group(signal);
+                if self.wait_for_teardown(std::time::Duration::from_millis(100)) {
+                    break;
+                }
             }
-            let completed = self.completed.lock().is_ok_and(|completed| {
-                completed
-                    .recv_timeout(std::time::Duration::from_millis(250))
-                    .is_ok()
-            });
-            if !completed {
-                if let Some(pid) = self.pid {
-                    // Root-only escalation; process-group teardown belongs to PR4.
-                    unsafe { libc::kill(pid as libc::pid_t, libc::SIGKILL) };
+            if !self.teardown_complete() {
+                if let Ok(mut killer) = self.killer.lock() {
+                    let _ = killer.kill();
                 }
+                let _ = self.wait_for_teardown(std::time::Duration::from_millis(250));
             }
             if let Ok(mut writer) = self.writer.lock() {
                 writer.take();
@@ -453,6 +470,33 @@ impl PtySession {
             }
         }
     }
+
+    fn signal_process_group(&self, signal: libc::c_int) -> bool {
+        self.pid.is_some_and(|pid| unsafe {
+            libc::kill(-(pid as libc::pid_t), signal) == 0
+                || (self.running.load(std::sync::atomic::Ordering::SeqCst)
+                    && libc::kill(pid as libc::pid_t, signal) == 0)
+        })
+    }
+
+    fn teardown_complete(&self) -> bool {
+        !self.process_group_exists() && !self.running.load(std::sync::atomic::Ordering::SeqCst)
+    }
+
+    fn process_group_exists(&self) -> bool {
+        self.pid.is_some_and(|pid| unsafe {
+            libc::kill(-(pid as libc::pid_t), 0) == 0
+                || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
+        })
+    }
+
+    fn wait_for_teardown(&self, timeout: std::time::Duration) -> bool {
+        let deadline = std::time::Instant::now() + timeout;
+        while !self.teardown_complete() && std::time::Instant::now() < deadline {
+            std::thread::sleep(std::time::Duration::from_millis(10));
+        }
+        self.teardown_complete()
+    }
 }
 
 #[cfg(unix)]
@@ -543,8 +587,12 @@ fn read_output<F>(
     let mut bytes = [0; 4096];
     let mut pending = Vec::new();
     while lifecycle.should_emit_natural_exit() {
-        let Ok(count) = reader.read(&mut bytes) else {
-            break;
+        let count = match reader.read(&mut bytes) {
+            Ok(count) => count,
+            // Linux PTYs report EIO once their slave side closes. It is EOF,
+            // not a session error, and still needs the buffered UTF-8 flush.
+            Err(error) if error.raw_os_error() == Some(libc::EIO) => break,
+            Err(_) => break,
         };
         if count == 0 {
             break;
@@ -572,6 +620,9 @@ fn read_output<F>(
             }
         }
     }
+    if lifecycle.should_emit_natural_exit() && !pending.is_empty() {
+        on_output(id, String::from_utf8_lossy(&pending).into_owned());
+    }
 }
 
 #[cfg(windows)]
@@ -2342,4 +2393,51 @@ mod tests {
             })
         );
     }
+
+    #[cfg(unix)]
+    #[test]
+    fn unix_interrupt_uses_process_group_fallback_while_writer_lock_is_held() {
+        use std::{
+            sync::mpsc,
+            time::{Duration, Instant},
+        };
+
+        let (sender, receiver) = mpsc::channel();
+        let session = PtySession::spawn(
+            "/bin/sh",
+            &[
+                "-c",
+                "trap 'printf interrupted; exit' INT; printf ready; while :; do :; done",
+            ],
+            TerminalSize::new(80, 24).unwrap(),
+            move |_, output| {
+                let _ = sender.send(output);
+            },
+            |_| {},
+        )
+        .unwrap();
+        let deadline = Instant::now() + Duration::from_secs(1);
+        let mut output = String::new();
+        while Instant::now() < deadline && !output.contains("ready") {
+            if let Ok(chunk) = receiver.recv_timeout(Duration::from_millis(50)) {
+                output.push_str(&chunk);
+            }
+        }
+        assert!(output.contains("ready"));
+        assert_eq!(
+            unsafe { libc::getpgid(session.pid.unwrap() as libc::pid_t) },
+            session.pid.unwrap() as libc::pid_t
+        );
+
+        let writer = session.writer.lock().unwrap();
+        let started = Instant::now();
+        session.interrupt().unwrap();
+        assert!(started.elapsed() < Duration::from_millis(100));
+        assert!(receiver
+            .recv_timeout(Duration::from_secs(1))
+            .unwrap()
+            .contains("interrupted"));
+        drop(writer);
+        session.close();
+    }
 }

crates/splice-pty/tests/unix_pty.rs

@@ -61,4 +61,51 @@ fn pid(output: &str) -> String { output.split("pid=").nth(1).unwrap().trim().to_
     assert!(receive_until(&receiver, "24 80").contains("24 80"));
     session.resize(TerminalSize::new(132, 43).unwrap()).unwrap();
     assert!(receive_until(&receiver, "43 132").contains("43 132")); session.close(); }
+#[test] fn unix_pty_close_escalates_for_resistant_descendants() {
+    let (session, receiver) = spawn("trap '' HUP TERM; sh -c 'trap \"\" HUP TERM; echo grandchild=$$; while :; do sleep 1; done' & echo root=$$; wait");
+    let deadline = Instant::now() + Duration::from_secs(5); let mut output = String::new();
+    while Instant::now() < deadline && (!output.contains("root=") || !output.contains("grandchild=")) {
+        if let Ok(chunk) = receiver.recv_timeout(Duration::from_millis(250)) { output.push_str(&chunk); }
+    }
+    let root = output.split("root=").nth(1).unwrap().lines().next().unwrap().trim().to_owned();
+    let grandchild = output.split("grandchild=").nth(1).unwrap().lines().next().unwrap().trim().to_owned();
+    let started = Instant::now(); session.close();
+    assert!(started.elapsed() < Duration::from_secs(2) && !is_alive(&root) && !is_alive(&grandchild)); }
+#[test] fn unix_pty_close_escalates_after_the_leader_exits_on_hup() {
+    let (session, receiver) = spawn("trap 'exit 0' HUP; sh -c 'trap \"\" HUP TERM; echo grandchild=$$; while :; do sleep 1; done' & echo leader=$$; wait");
+    let deadline = Instant::now() + Duration::from_secs(5); let mut output = String::new();
+    while Instant::now() < deadline && (!output.contains("leader=") || !output.contains("grandchild=")) {
+        if let Ok(chunk) = receiver.recv_timeout(Duration::from_millis(250)) { output.push_str(&chunk); }
+    }
+    let leader = output.split("leader=").nth(1).unwrap().lines().next().unwrap().trim().to_owned();
+    let grandchild = output.split("grandchild=").nth(1).unwrap().lines().next().unwrap().trim().to_owned();
+    let started = Instant::now(); session.close();
+    assert!(started.elapsed() < Duration::from_secs(2) && !is_alive(&leader) && !is_alive(&grandchild)); }
+#[test] fn unix_pty_flushes_partial_utf8_before_natural_exit() {
+    let (output_sender, output_receiver) = mpsc::channel(); let (exit_sender, exit_receiver) = mpsc::channel();
+    let session = PtySession::spawn("/bin/sh", &["-c", "printf final; printf '\\303'"], size(),
+        move |_, output| { let _ = output_sender.send(output); }, move |_| { let _ = exit_sender.send(()); }).unwrap();
+    assert!(exit_receiver.recv_timeout(Duration::from_secs(1)).is_ok()); let mut output = String::new();
+    while let Ok(chunk) = output_receiver.recv_timeout(Duration::from_millis(100)) { output.push_str(&chunk); }
+    assert!(output.contains("final�")); session.close(); }
+#[test] fn unix_pty_delivers_final_output_before_natural_exit() {
+    enum Event { Output(String), Exit }
+    let (sender, receiver) = mpsc::channel(); let output_sender = sender.clone();
+    let session = PtySession::spawn("/bin/sh", &["-c", "printf final-output"], size(),
+        move |_, output| { std::thread::sleep(Duration::from_millis(100)); let _ = output_sender.send(Event::Output(output)); }, move |_| { let _ = sender.send(Event::Exit); }).unwrap();
+    let first = receiver.recv_timeout(Duration::from_secs(1)).unwrap();
+    assert!(matches!(first, Event::Output(output) if output.contains("final-output")));
+    assert!(matches!(receiver.recv_timeout(Duration::from_secs(1)).unwrap(), Event::Exit)); session.close(); }
+#[test] fn unix_pty_bounds_natural_exit_drain_when_descendant_retains_slave() {
+    enum Event { Output(String), Exit }
+    let (sender, receiver) = mpsc::channel(); let output_sender = sender.clone();
+    let session = PtySession::spawn("/bin/sh", &["-c", "sh -c 'trap \"\" HUP TERM; echo descendant=$$; while :; do sleep 1; done' & sleep 0.1; printf final-output"], size(),
+        move |_, output| { let _ = output_sender.send(Event::Output(output)); }, move |_| { let _ = sender.send(Event::Exit); }).unwrap();
+    let mut output = String::new();
+    while !output.contains("final-output") { match receiver.recv_timeout(Duration::from_secs(1)).unwrap() {
+        Event::Output(chunk) => output.push_str(&chunk), Event::Exit => panic!("final output must precede exit"),
+    }}
+    let descendant = pid(&output.replace("descendant", "pid").replace("final-output", ""));
+    assert!(matches!(receiver.recv_timeout(Duration::from_secs(1)).unwrap(), Event::Exit));
+    session.close(); assert!(!is_alive(&descendant)); }
 }