From c12cf7aad0625ffe1ebf03397b91e44fd083dc24 Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 3 Sep 2026 16:22:07 +0000 Subject: [PATCH 1/5] test(pty): cover exit ordering and nonreading child cleanup --- Cargo.toml | 2 +- tests/runtime.rs | 149 ++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 149 insertions(+), 2 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 4fee6af..50f8354 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,7 +36,7 @@ windows-sys = { version = "0.61.2", features = [ ] } [target.'cfg(windows)'.dev-dependencies] -windows-sys = { version = "0.61.2", features = ["Win32_System_Console"] } +windows-sys = { version = "0.61.2", features = ["Win32_System_Console", "Win32_System_Threading"] } [profile.release] codegen-units = 1 diff --git a/tests/runtime.rs b/tests/runtime.rs index 8257c73..f1e3328 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -7,7 +7,9 @@ use std::thread; use std::time::{Duration, Instant}; use opencode_pty::protocol::AttachmentRole; -use opencode_pty::service::{StreamEvent, TerminalId, TerminalService, TerminalSnapshot}; +use opencode_pty::service::{ + StreamEvent, TerminalId, TerminalLifecycle, TerminalService, TerminalSnapshot, +}; use terminal_fixture::{Command, Deadline, Fixture, TempDir}; fn wait_text(service: &TerminalService, id: TerminalId, expected: &str) -> TerminalSnapshot { @@ -241,3 +243,148 @@ fn independent_terminals_keep_bounded_replay() { service.terminate(beta.id).unwrap(); assert!(service.list().unwrap().is_empty()); } + +#[test] +fn normal_exit_follows_final_output_and_preserves_exit_code() { + let _deadline = Deadline::new(); + let fixture = Fixture::new(); + let service = TerminalService::default(); + let info = service.create(fixture.request()).unwrap(); + let mut child = fixture.connect(); + let observer = service + .attach( + info.id, + 0, + "exit-observer".into(), + AttachmentRole::Observer, + false, + ) + .unwrap(); + let mut bytes = observer.replay.bytes.clone(); + let mut offset = observer.replay.end_offset; + child.command(Command::Output("\r\nFINAL_OUTPUT_BEFORE_EXIT\r\n".into())); + child.command(Command::Exit(23)); + loop { + match observer + .events + .recv_timeout(Duration::from_secs(15)) + .unwrap_or_else(|error| { + panic!( + "missing exit event: {error}; snapshot={:?}", + service.snapshot(info.id) + ) + }) { + StreamEvent::Output { + start, + end, + bytes: output, + } => { + assert_eq!(start, offset); + assert_eq!(end - start, output.len() as u64); + bytes.extend(output); + offset = end; + } + StreamEvent::Exited { + exit_code, + final_offset, + } => { + assert_eq!(exit_code, Some(23)); + assert_eq!(final_offset, offset); + break; + } + _ => {} + } + } + assert_eq!(service.replay(info.id, 0).unwrap().bytes, bytes); + let snapshot = service.snapshot(info.id).unwrap(); + assert!( + snapshot.text.contains("FINAL_OUTPUT_BEFORE_EXIT"), + "{snapshot:?}" + ); + assert_eq!( + snapshot.info.lifecycle, + TerminalLifecycle::Exited { + exit_code: Some(23) + } + ); + assert_eq!(snapshot.info.output_tail, offset); + assert!(observer.events.try_recv().is_err()); + drop(observer); + service.terminate(info.id).unwrap(); +} + +#[test] +fn repeated_termination_and_drop_release_nonreading_children() { + let _deadline = Deadline::new(); + let service = TerminalService::default(); + for _ in 0..3 { + let fixture = Fixture::new(); + let info = service.create(fixture.request()).unwrap(); + let _child = fixture.connect(); + let process = ChildProcess::open(info.pid.unwrap()); + // Child waits on the private control channel, never draining PTY stdin. + service.write(info.id, vec![b'x'; 256 * 1024]).unwrap(); + service.terminate(info.id).unwrap(); + process.assert_exited(); + assert!(service.list().unwrap().is_empty()); + assert!(service.terminate(info.id).is_err()); + } + let fixture = Fixture::new(); + let info = service.create(fixture.request()).unwrap(); + let _child = fixture.connect(); + let process = ChildProcess::open(info.pid.unwrap()); + service.write(info.id, vec![b'x'; 256 * 1024]).unwrap(); + drop(service); + process.assert_exited(); +} + +struct ChildProcess { + #[cfg(unix)] + pid: u32, + #[cfg(windows)] + handle: std::os::windows::io::OwnedHandle, +} + +impl ChildProcess { + fn open(pid: u32) -> Self { + #[cfg(unix)] + { + Self { pid } + } + #[cfg(windows)] + { + use std::os::windows::io::FromRawHandle; + use windows_sys::Win32::System::Threading::{OpenProcess, PROCESS_SYNCHRONIZE}; + // SAFETY: request only wait access to this known-live test child. + let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE, 0, pid) }; + assert!(!handle.is_null(), "{}", std::io::Error::last_os_error()); + Self { + handle: unsafe { std::os::windows::io::OwnedHandle::from_raw_handle(handle) }, + } + } + } + + fn assert_exited(&self) { + #[cfg(unix)] + { + // SAFETY: signal zero checks liveness without signalling a process. + assert_eq!(unsafe { libc::kill(self.pid as i32, 0) }, -1); + assert_eq!( + std::io::Error::last_os_error().raw_os_error(), + Some(libc::ESRCH) + ); + } + #[cfg(windows)] + { + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::{ + Foundation::WAIT_OBJECT_0, System::Threading::WaitForSingleObject, + }; + // SAFETY: the owned process handle has wait access and stays live. + assert_eq!( + unsafe { WaitForSingleObject(self.handle.as_raw_handle(), 0) }, + WAIT_OBJECT_0 + ); + } + } +} From 2addb50db0e777f930f73d85383d0e69c348bfa2 Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 3 Sep 2026 16:33:30 +0000 Subject: [PATCH 2/5] fix(pty): interrupt shutdown queue backpressure --- README.md | 9 + src/service.rs | 413 +++++++++++++++++++++++++----- tests/runtime.rs | 79 +++++- tests/support/terminal_fixture.rs | 11 + 4 files changed, 453 insertions(+), 59 deletions(-) diff --git a/README.md b/README.md index 65c7a85..16890bb 100644 --- a/README.md +++ b/README.md @@ -284,6 +284,15 @@ multiple connections, namespace ownership, cancellation, and final-frame completion. These checks do not yet establish Windows daemon support or complete ConPTY shutdown behavior. +### Known Linux cleanup limitation + +A child that exits without consuming a large queued PTY write can leave the +Linux master write blocked after the child, actor, reader, and waiter have +already stopped. `TerminalService::terminate` or service drop can then wait +indefinitely for the writer thread. This pre-existing Unix I/O limitation is +not repaired by the Windows cleanup work. A test watchdog or the daemon's +forced-exit deadline does not prove that those worker threads were joined. + Once the workflow is on `master`, it can also be run manually against a branch: ```sh diff --git a/src/service.rs b/src/service.rs index 5478922..289580b 100644 --- a/src/service.rs +++ b/src/service.rs @@ -400,6 +400,7 @@ impl Drop for TerminalService { struct TerminalHandle { actor_tx: Sender, + shutdown_tx: Mutex>>, killer: Mutex>, info: Arc>, joins: Mutex>>>, @@ -461,10 +462,14 @@ impl TerminalHandle { let (actor_tx, actor_rx) = bounded(ACTOR_QUEUE_CAPACITY); let (writer_tx, writer_rx) = bounded(WRITER_QUEUE_CAPACITY); + // Disconnecting this channel wakes every cancellable queue operation, + // even when the actor cannot consume an ordinary shutdown message. + let (shutdown_tx, shutdown) = bounded(0); let (init_tx, init_rx) = std_mpsc::sync_channel(1); let actor_info = Arc::clone(&info); let actor_writer_tx = writer_tx.clone(); + let actor_shutdown = shutdown.clone(); let actor_thread = thread::Builder::new() .name(format!("opencode-pty-actor-{id}")) .spawn(move || { @@ -472,6 +477,7 @@ impl TerminalHandle { master: pair.master, messages: actor_rx, writes: actor_writer_tx, + shutdown: actor_shutdown, info: actor_info, replay_capacity, cols: request.cols, @@ -500,7 +506,7 @@ impl TerminalHandle { let writer_actor_tx = actor_tx.clone(); let writer_thread = thread::Builder::new() .name(format!("opencode-pty-writer-{id}")) - .spawn(move || run_writer(writer, writer_rx, writer_actor_tx)) + .spawn(move || run_writer(writer, writer_rx, writer_actor_tx, shutdown)) .context("failed to spawn terminal writer")?; let reader_actor_tx = actor_tx.clone(); @@ -508,6 +514,7 @@ impl TerminalHandle { .name(format!("opencode-pty-reader-{id}")) .spawn(move || { let mut buffer = [0_u8; 8192]; + let mut forwarding = true; loop { match reader.read(&mut buffer) { Ok(0) => { @@ -515,12 +522,18 @@ impl TerminalHandle { break; } Ok(length) => { - if reader_actor_tx - .send(ActorMessage::Output(buffer[..length].to_vec())) - .is_err() - { - break; + if forwarding { + forwarding = reader_actor_tx + .send(ActorMessage::Output(buffer[..length].to_vec())) + .is_ok(); + #[cfg(unix)] + if !forwarding { + break; + } } + // On Windows, ClosePseudoConsole can wait for its + // output pipe to drain. Once the actor stops, keep + // this sole reader draining without forwarding. } Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, #[cfg(unix)] @@ -551,6 +564,7 @@ impl TerminalHandle { Ok(Self { actor_tx, + shutdown_tx: Mutex::new(Some(shutdown_tx)), killer: Mutex::new(killer), info, joins: Mutex::new(Some(vec![ @@ -581,17 +595,23 @@ impl TerminalHandle { } fn shutdown(&self) -> Result<()> { + // Keep this guard through the joins: concurrent shutdown callers must + // not return early or signal the child again after it has been reaped. + let mut joins = self + .joins + .lock() + .map_err(|_| anyhow!("terminal join lock poisoned"))?; + let Some(handles) = joins.take() else { + return Ok(()); + }; + self.shutdown_tx + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); if let Ok(mut killer) = self.killer.lock() { let _ = killer.kill(); } - let _ = self.actor_tx.send(ActorMessage::Shutdown); - let joins = self - .joins - .lock() - .map_err(|_| anyhow!("terminal join lock poisoned"))? - .take() - .unwrap_or_default(); - for join in joins { + for join in handles { let _ = join.join(); } Ok(()) @@ -646,7 +666,6 @@ enum ActorMessage { Detach { attachment_id: String, }, - Shutdown, } struct Attached { @@ -669,6 +688,7 @@ struct ActorConfig { master: Box, messages: Receiver, writes: Sender, + shutdown: Receiver<()>, info: Arc>, replay_capacity: usize, cols: u16, @@ -681,6 +701,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { master, messages, writes, + shutdown, info, replay_capacity, cols, @@ -703,10 +724,11 @@ fn run_actor(config: ActorConfig) -> Result<()> { let mut pending_message = None; let _ = init_tx.send(Ok(())); - loop { - match receive_actor_message(&messages, &mut pending_message) { + let result = loop { + match receive_actor_message(&messages, &mut pending_message, &shutdown) { Ok(ActorMessage::RefreshForegroundProcess { reply }) => { publish_foreground_process( + &shutdown, &*master, &info, &mut subscribers, @@ -718,9 +740,12 @@ fn run_actor(config: ActorConfig) -> Result<()> { Ok(ActorMessage::Output(bytes)) => { let (start, end) = replay.append(&bytes); terminal.vt_write(&bytes); - forward_replies(&mut terminal, &writes)?; + if let Err(error) = forward_replies(&mut terminal, &writes, &shutdown) { + break Err(error); + } update_offsets(&info, &replay); broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -739,6 +764,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { }; if changed { broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -752,12 +778,8 @@ fn run_actor(config: ActorConfig) -> Result<()> { bytes, reply, }) => { - let result = - authorize_controller(&controller, attachment_id.as_deref()).and_then(|_| { - writes - .send(WriterMessage::Bytes(bytes)) - .map_err(|_| anyhow!("terminal writer stopped")) - }); + let result = authorize_controller(&controller, attachment_id.as_deref()) + .and_then(|_| queue_write(&writes, &shutdown, bytes)); let _ = reply.send(result); } Ok(ActorMessage::Resize { @@ -775,7 +797,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { pixel_height: 0, })?; let resized = terminal.resize(cols, rows); - forward_replies(&mut terminal, &writes)?; + forward_replies(&mut terminal, &writes, &shutdown)?; resized?; if let Ok(mut value) = info.write() { value.cols = cols; @@ -784,6 +806,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let generation = controller_generation; let checkpoint = format_terminal(&terminal, Format::Vt)?.into_bytes(); broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -823,6 +846,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { } let generation = controller_generation; broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -844,7 +868,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { pixel_height: 0, })?; let resized = terminal.resize(cols, rows); - forward_replies(&mut terminal, &writes)?; + forward_replies(&mut terminal, &writes, &shutdown)?; resized?; if let Ok(mut value) = info.write() { value.cols = cols; @@ -853,6 +877,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let generation = controller_generation; let checkpoint = format_terminal(&terminal, Format::Vt)?.into_bytes(); broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -865,9 +890,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { ); } if let Some(bytes) = bytes { - writes - .send(WriterMessage::Bytes(bytes)) - .map_err(|_| anyhow!("terminal writer stopped"))?; + queue_write(&writes, &shutdown, bytes)?; } Ok(()) })(); @@ -935,6 +958,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let controller_id = controller.clone(); let generation = controller_generation; broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -960,6 +984,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let generation = controller_generation; let attachment_id = controller.clone(); broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -979,8 +1004,8 @@ fn run_actor(config: ActorConfig) -> Result<()> { } } Ok(ActorMessage::ReaderEof) => reader_eof = true, - Ok(ActorMessage::Shutdown) | Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { - break; + Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { + break Ok(()); } Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue, } @@ -996,6 +1021,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { }; } broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -1006,22 +1032,41 @@ fn run_actor(config: ActorConfig) -> Result<()> { ); exit_published = true; } - } + }; + // Release senders blocked on the actor queue before closing ConPTY. Its + // synchronous close may need the reader to continue draining final output. + drop(messages); drop(terminal); drop(master); drop(writes); - Ok(()) + if matches!( + shutdown.try_recv(), + Err(crossbeam_channel::TryRecvError::Disconnected) + ) { + // Interrupted terminal-generated replies are expected during shutdown. + Ok(()) + } else { + result + } } fn receive_actor_message( messages: &Receiver, pending: &mut Option, + shutdown: &Receiver<()>, ) -> std::result::Result { + if matches!( + shutdown.try_recv(), + Err(crossbeam_channel::TryRecvError::Disconnected) + ) { + return Err(crossbeam_channel::RecvTimeoutError::Disconnected); + } let message = match pending.take() { Some(message) => message, - None => messages - .recv() - .map_err(|_| crossbeam_channel::RecvTimeoutError::Disconnected)?, + None => crossbeam_channel::select_biased! { + recv(shutdown) -> _ => return Err(crossbeam_channel::RecvTimeoutError::Disconnected), + recv(messages) -> message => message.map_err(|_| crossbeam_channel::RecvTimeoutError::Disconnected)?, + }, }; let ActorMessage::Output(mut bytes) = message else { return Ok(message); @@ -1042,6 +1087,7 @@ fn receive_actor_message( } fn publish_foreground_process( + shutdown: &Receiver<()>, master: &dyn MasterPty, info: &Arc>, subscribers: &mut HashMap, @@ -1061,6 +1107,7 @@ fn publish_foreground_process( }; if changed { broadcast( + shutdown, subscribers, controller, controller_generation, @@ -1157,21 +1204,42 @@ fn process_depth(pid: i32, parents: &HashMap) -> usize { // Forward effects immediately after each native mutation, even if it reports an // error. In particular, resize notifications must precede any accompanying input. // Actual PTY I/O remains on the single writer thread, never inside a callback. -fn forward_replies(terminal: &mut Terminal, writes: &Sender) -> Result<()> { +fn forward_replies( + terminal: &mut Terminal, + writes: &Sender, + shutdown: &Receiver<()>, +) -> Result<()> { for response in terminal.take_writes() { - writes - .send(WriterMessage::Bytes(response)) - .map_err(|_| anyhow!("terminal writer stopped"))?; + queue_write(writes, shutdown, response)?; } Ok(()) } +fn queue_write( + writes: &Sender, + shutdown: &Receiver<()>, + bytes: Vec, +) -> Result<()> { + crossbeam_channel::select_biased! { + recv(shutdown) -> _ => bail!("terminal is stopping"), + send(writes, WriterMessage::Bytes(bytes)) -> result => result.map_err(|_| anyhow!("terminal writer stopped")), + } +} + fn run_writer( mut writer: Box, writer_rx: Receiver, actor_tx: Sender, + shutdown: Receiver<()>, ) { - for message in writer_rx { + loop { + let message = crossbeam_channel::select_biased! { + recv(shutdown) -> _ => break, + recv(writer_rx) -> message => match message { + Ok(message) => message, + Err(_) => break, + }, + }; let result = match message { WriterMessage::Bytes(bytes) => writer.write_all(&bytes).and_then(|_| writer.flush()), }; @@ -1306,21 +1374,33 @@ impl ReplayBuffer { } fn broadcast( + shutdown: &Receiver<()>, subscribers: &mut HashMap, controller: &mut Option, controller_generation: &mut u64, event: StreamEvent, ) { - let dropped = subscribers - .iter() - .filter_map(|(id, subscriber)| { - let failed = match subscriber.role { - AttachmentRole::Controller => subscriber.events.send(event.clone()).is_err(), - AttachmentRole::Observer => subscriber.events.try_send(event.clone()).is_err(), - }; - failed.then(|| id.clone()) - }) - .collect::>(); + let mut dropped = Vec::new(); + for (id, subscriber) in subscribers.iter() { + let failed = match subscriber.role { + AttachmentRole::Controller => crossbeam_channel::select_biased! { + // Quitting is not a subscriber disconnect: do not promote and + // recursively notify controllers while tearing the actor down. + recv(shutdown) -> _ => return, + send(subscriber.events, event.clone()) -> result => result.is_err(), + }, + AttachmentRole::Observer => subscriber.events.try_send(event.clone()).is_err(), + }; + if failed { + dropped.push(id.clone()); + } + } + if matches!( + shutdown.try_recv(), + Err(crossbeam_channel::TryRecvError::Disconnected) + ) { + return; + } for id in dropped { subscribers.remove(&id); if controller.as_ref() == Some(&id) { @@ -1329,6 +1409,7 @@ fn broadcast( let generation = *controller_generation; let attachment_id = controller.clone(); broadcast( + shutdown, subscribers, controller, controller_generation, @@ -1390,6 +1471,218 @@ fn default_shell() -> String { mod tests { use super::*; + struct TestMaster; + + impl MasterPty for TestMaster { + fn resize(&self, _: PtySize) -> Result<()> { + Ok(()) + } + fn get_size(&self) -> Result { + Ok(PtySize::default()) + } + fn try_clone_reader(&self) -> Result> { + unreachable!() + } + fn take_writer(&self) -> Result> { + unreachable!() + } + #[cfg(unix)] + fn process_group_leader(&self) -> Option { + None + } + #[cfg(unix)] + fn as_raw_fd(&self) -> Option { + None + } + #[cfg(unix)] + fn tty_name(&self) -> Option { + None + } + } + + #[derive(Debug)] + struct TestKiller; + + impl ChildKiller for TestKiller { + fn kill(&mut self) -> io::Result<()> { + Ok(()) + } + fn clone_killer(&self) -> Box { + Box::new(Self) + } + } + + // Real actor/parser and bounded channels, without OS pipe capacity or a + // large output flood. Native child/handle coverage lives in tests/runtime.rs. + fn test_actor() -> (TerminalHandle, Receiver) { + let (actor_tx, messages) = bounded(ACTOR_QUEUE_CAPACITY); + let (writes, writer_rx) = bounded(1); + let (shutdown_tx, shutdown) = bounded(0); + let (init_tx, init_rx) = std_mpsc::sync_channel(1); + let info = Arc::new(RwLock::new(TerminalInfo { + id: 1, + pid: None, + title: "test".into(), + foreground_process: None, + group_id: "test".into(), + command: vec![], + cwd: PathBuf::new(), + cols: 80, + rows: 24, + lifecycle: TerminalLifecycle::Running, + output_head: 0, + output_tail: 0, + })); + let actor_info = Arc::clone(&info); + let actor = thread::spawn(move || { + run_actor(ActorConfig { + master: Box::new(TestMaster), + messages, + writes, + shutdown, + info: actor_info, + replay_capacity: 4096, + cols: 80, + rows: 24, + init_tx, + }) + .unwrap() + }); + init_rx.recv().unwrap().unwrap(); + ( + TerminalHandle { + actor_tx, + shutdown_tx: Mutex::new(Some(shutdown_tx)), + killer: Mutex::new(Box::new(TestKiller)), + info, + joins: Mutex::new(Some(vec![actor])), + }, + writer_rx, + ) + } + + #[test] + fn child_exit_and_reader_eof_are_distinct_actor_signals() { + for eof_first in [false, true] { + let (actor, _writer) = test_actor(); + actor + .actor_tx + .send(if eof_first { + ActorMessage::ReaderEof + } else { + ActorMessage::ChildExited(Ok(Some(23))) + }) + .unwrap(); + let snapshot = actor + .request(|reply| ActorMessage::Snapshot { reply }) + .unwrap(); + assert_eq!(snapshot.info.lifecycle, TerminalLifecycle::Running); + if !eof_first { + actor + .actor_tx + .send(ActorMessage::Output(b"final".to_vec())) + .unwrap(); + } + actor + .actor_tx + .send(if eof_first { + ActorMessage::ChildExited(Ok(Some(23))) + } else { + ActorMessage::ReaderEof + }) + .unwrap(); + let snapshot = actor + .request(|reply| ActorMessage::Snapshot { reply }) + .unwrap(); + assert_eq!( + snapshot.info.lifecycle, + TerminalLifecycle::Exited { + exit_code: Some(23) + } + ); + if !eof_first { + assert_eq!(snapshot.text, "final"); + } + actor.shutdown().unwrap(); + actor.shutdown().unwrap(); + } + } + + #[test] + fn shutdown_interrupts_controller_backpressure() { + let (actor, _writer) = test_actor(); + let attached = actor + .request(|reply| ActorMessage::Attach { + offset: 0, + attachment_id: "controller".into(), + role: AttachmentRole::Controller, + takeover: false, + reply, + }) + .unwrap(); + // Attach already queued ControllerChanged. Force one output per batch + // with a following actor request, filling the queue with only 1 KiB. + for _ in 1..SUBSCRIBER_QUEUE_CAPACITY { + actor + .actor_tx + .send(ActorMessage::Output(b"x".to_vec())) + .unwrap(); + actor + .request(|reply| ActorMessage::Replay { offset: 0, reply }) + .unwrap(); + } + assert!(attached.events.is_full()); + actor + .actor_tx + .send(ActorMessage::Output(b"!".to_vec())) + .unwrap(); + let (done, finished) = std_mpsc::channel(); + let shutdown = thread::spawn(move || { + actor.shutdown().unwrap(); + done.send(()).unwrap(); + }); + let stopped = finished.recv_timeout(Duration::from_secs(1)).is_ok(); + // Release the test consumer even on failure so a red regression never + // leaves a thread stuck or depends on the runtime-test watchdog. + drop(attached); + shutdown.join().unwrap(); + assert!( + stopped, + "shutdown waited for a controller to consume output" + ); + } + + #[test] + fn shutdown_interrupts_actor_to_writer_backpressure() { + let (actor, writer) = test_actor(); + actor + .request(|reply| ActorMessage::Write { + attachment_id: None, + bytes: b"first".to_vec(), + reply, + }) + .unwrap(); + assert!(writer.is_full()); + let (reply, _reply_rx) = std_mpsc::sync_channel(1); + actor + .actor_tx + .send(ActorMessage::Write { + attachment_id: None, + bytes: b"blocked".to_vec(), + reply, + }) + .unwrap(); + let (done, finished) = std_mpsc::channel(); + let shutdown = thread::spawn(move || { + actor.shutdown().unwrap(); + done.send(()).unwrap(); + }); + let stopped = finished.recv_timeout(Duration::from_secs(1)).is_ok(); + drop(writer); + shutdown.join().unwrap(); + assert!(stopped, "shutdown waited for the full input queue"); + } + fn rows_terminal(cols: u16, rows: u16, input: &str) -> Terminal { let mut terminal = Terminal::new(TerminalOptions { cols, @@ -1405,7 +1698,8 @@ mod tests { fn forwarding_replies_drains_once_and_reports_writer_failure() { let mut terminal = rows_terminal(10, 3, "\x1b[5n\x1b[6n"); let (writes, reader) = bounded(4); - forward_replies(&mut terminal, &writes).unwrap(); + let (_shutdown_tx, shutdown) = bounded(0); + forward_replies(&mut terminal, &writes, &shutdown).unwrap(); for expected in [b"\x1b[0n".as_slice(), b"\x1b[1;1R".as_slice()] { let WriterMessage::Bytes(bytes) = reader.try_recv().unwrap(); assert_eq!(bytes, expected); @@ -1416,7 +1710,7 @@ mod tests { terminal.vt_write(b"\x1b[5n"); drop(reader); assert_eq!( - forward_replies(&mut terminal, &writes) + forward_replies(&mut terminal, &writes, &shutdown) .unwrap_err() .to_string(), "terminal writer stopped", @@ -1652,6 +1946,7 @@ mod tests { #[test] fn controller_backpressures_instead_of_disconnecting() { + let (_shutdown_tx, shutdown) = bounded(0); let (events_tx, events) = bounded(1); events_tx .send(StreamEvent::Output { @@ -1672,6 +1967,7 @@ mod tests { let mut controller = Some("controller".to_string()); let mut generation = 1; broadcast( + &shutdown, &mut subscribers, &mut controller, &mut generation, @@ -1697,6 +1993,7 @@ mod tests { #[test] fn slow_observer_is_disconnected() { + let (_shutdown_tx, shutdown) = bounded(0); let (events_tx, _events) = bounded(1); events_tx .send(StreamEvent::Output { @@ -1715,6 +2012,7 @@ mod tests { )]); broadcast( + &shutdown, &mut subscribers, &mut None, &mut 0, @@ -1730,6 +2028,7 @@ mod tests { #[test] fn output_is_batched_without_reordering_other_messages() { + let (_shutdown_tx, shutdown) = bounded(0); let (messages_tx, messages) = bounded(4); messages_tx .send(ActorMessage::Output(b"abc".to_vec())) @@ -1737,16 +2036,16 @@ mod tests { messages_tx .send(ActorMessage::Output(b"def".to_vec())) .unwrap(); - messages_tx.send(ActorMessage::Shutdown).unwrap(); + messages_tx.send(ActorMessage::ReaderEof).unwrap(); let mut pending = None; assert!(matches!( - receive_actor_message(&messages, &mut pending).unwrap(), + receive_actor_message(&messages, &mut pending, &shutdown).unwrap(), ActorMessage::Output(bytes) if bytes == b"abcdef" )); assert!(matches!( - receive_actor_message(&messages, &mut pending).unwrap(), - ActorMessage::Shutdown + receive_actor_message(&messages, &mut pending, &shutdown).unwrap(), + ActorMessage::ReaderEof )); } } diff --git a/tests/runtime.rs b/tests/runtime.rs index f1e3328..98dbc18 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -314,7 +314,19 @@ fn normal_exit_follows_final_output_and_preserves_exit_code() { } #[test] +fn repeated_termination_and_drop_release_children() { + termination_cycles(false); +} + +// Linux has a pre-existing portable-pty blocking-write hang after child exit; +// that repro is retained outside this Windows cleanup suite, not claimed fixed. +#[test] +#[cfg(windows)] fn repeated_termination_and_drop_release_nonreading_children() { + termination_cycles(true); +} + +fn termination_cycles(block_input: bool) { let _deadline = Deadline::new(); let service = TerminalService::default(); for _ in 0..3 { @@ -323,7 +335,9 @@ fn repeated_termination_and_drop_release_nonreading_children() { let _child = fixture.connect(); let process = ChildProcess::open(info.pid.unwrap()); // Child waits on the private control channel, never draining PTY stdin. - service.write(info.id, vec![b'x'; 256 * 1024]).unwrap(); + if block_input { + service.write(info.id, vec![b'x'; 256 * 1024]).unwrap(); + } service.terminate(info.id).unwrap(); process.assert_exited(); assert!(service.list().unwrap().is_empty()); @@ -333,7 +347,9 @@ fn repeated_termination_and_drop_release_nonreading_children() { let info = service.create(fixture.request()).unwrap(); let _child = fixture.connect(); let process = ChildProcess::open(info.pid.unwrap()); - service.write(info.id, vec![b'x'; 256 * 1024]).unwrap(); + if block_input { + service.write(info.id, vec![b'x'; 256 * 1024]).unwrap(); + } drop(service); process.assert_exited(); } @@ -345,6 +361,50 @@ struct ChildProcess { handle: std::os::windows::io::OwnedHandle, } +#[test] +#[cfg(windows)] +fn terminating_a_shell_releases_its_console_child() { + use base64::Engine; + + let _deadline = Deadline::new(); + let fixture = Fixture::new(); + let service = TerminalService::default(); + let mut request = fixture.request(); + request + .env + .insert("PTY_FIXTURE_EXE".into(), request.program); + request.program = std::path::PathBuf::from(std::env::var_os("SystemRoot").unwrap()) + .join("System32/WindowsPowerShell/v1.0/powershell.exe") + .to_str() + .unwrap() + .to_owned(); + // EncodedCommand avoids imposing cmd.exe's distinct quoting rules on the + // portable-pty CommandBuilder's normal Windows argv quoting. + let script = "& $env:PTY_FIXTURE_EXE --ignored --exact terminal_fixture::child --nocapture --test-threads=1; exit $LASTEXITCODE"; + let script = script + .encode_utf16() + .flat_map(u16::to_le_bytes) + .collect::>(); + request.args = vec![ + "-NoLogo".into(), + "-NoProfile".into(), + "-NonInteractive".into(), + "-EncodedCommand".into(), + base64::engine::general_purpose::STANDARD.encode(script), + ]; + let info = service.create(request).unwrap(); + let root = ChildProcess::open(info.pid.unwrap()); + let mut child = fixture.connect(); + let pid = u32::try_from(child.command(Command::Context)["pid"].as_u64().unwrap()).unwrap(); + assert_ne!(pid, info.pid.unwrap()); + let descendant = ChildProcess::open(pid); + child.command(Command::Output("\r\nSHELL_DESCENDANT_READY".into())); + wait_text(&service, info.id, "SHELL_DESCENDANT_READY"); + service.terminate(info.id).unwrap(); + root.assert_exited(); + descendant.assert_exited(); +} + impl ChildProcess { fn open(pid: u32) -> Self { #[cfg(unix)] @@ -388,3 +448,18 @@ impl ChildProcess { } } } + +#[test] +#[cfg(windows)] +fn shutdown_drains_conpty_while_a_child_is_producing_output() { + let _deadline = Deadline::new(); + let fixture = Fixture::new(); + let service = TerminalService::default(); + let info = service.create(fixture.request()).unwrap(); + let mut child = fixture.connect(); + let process = ChildProcess::open(info.pid.unwrap()); + child.command(Command::Flood); + wait_text(&service, info.id, "FLOOD_OUTPUT"); + drop(service); + process.assert_exited(); +} diff --git a/tests/support/terminal_fixture.rs b/tests/support/terminal_fixture.rs index fbe5ef8..cdea300 100644 --- a/tests/support/terminal_fixture.rs +++ b/tests/support/terminal_fixture.rs @@ -26,6 +26,7 @@ pub enum Command { Size, Context, Exit(i32), + Flood, } pub struct Fixture { @@ -167,6 +168,7 @@ fn child() { } Command::Size => serde_json::json!(console_size()), Command::Context => serde_json::json!({ + "pid": std::process::id(), "cwd": env::current_dir().unwrap(), "args": env::args().skip(1).collect::>(), "value": env::var("PTY_FIXTURE_VALUE").ok(), @@ -177,6 +179,15 @@ fn child() { send(channel.get_mut(), &serde_json::Value::Null); std::process::exit(code); } + Command::Flood => { + send(channel.get_mut(), &serde_json::Value::Null); + let mut stdout = io::stdout().lock(); + let chunk = "FLOOD_OUTPUT\r\n".repeat(256); + loop { + stdout.write_all(chunk.as_bytes()).unwrap(); + stdout.flush().unwrap(); + } + } }; send(channel.get_mut(), &result); } From 031c6d3bbad2f39418c9c3ad7149f8933c811c4d Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 3 Sep 2026 16:40:10 +0000 Subject: [PATCH 3/5] fix(pty): close ConPTY off actor before awaiting EOF --- README.md | 8 +++ src/service.rs | 130 ++++++++++++++++++++++++++++++++++------- src/service/windows.rs | 63 ++++++++++++++++++++ tests/runtime.rs | 39 ++++++++++++- 4 files changed, 215 insertions(+), 25 deletions(-) create mode 100644 src/service/windows.rs diff --git a/README.md b/README.md index 16890bb..10b56f8 100644 --- a/README.md +++ b/README.md @@ -284,6 +284,14 @@ multiple connections, namespace ownership, cancellation, and final-frame completion. These checks do not yet establish Windows daemon support or complete ConPTY shutdown behavior. +On Windows, normal root-child exit hands the ConPTY master to the existing +child-wait worker for closing. The actor and reader continue draining until the +real output-pipe EOF; only then is the final exit event published. Detecting the +root's exit adds at most 10 ms of polling latency. After the master is handed +off, Windows input and resize requests fail with a child-exited error; final +snapshots, rows, and replay remain readable until the terminal is removed. +Unix post-exit PTY operations retain their existing behavior. + ### Known Linux cleanup limitation A child that exits without consuming a large queued PTY write can leave the diff --git a/src/service.rs b/src/service.rs index 289580b..11b93a2 100644 --- a/src/service.rs +++ b/src/service.rs @@ -16,6 +16,9 @@ use serde::{Deserialize, Serialize}; use crate::ghostty::{Format, Terminal, TerminalOptions}; use crate::protocol::AttachmentRole; +#[cfg(windows)] +mod windows; + const DEFAULT_COLS: u16 = 100; const DEFAULT_ROWS: u16 = 30; const REPLAY_CAPACITY: usize = 2 * 1024 * 1024; @@ -427,7 +430,7 @@ impl TerminalHandle { command.env("TERM", "xterm-256color"); command.env("COLORTERM", "truecolor"); - let mut child = pair + let child = pair .slave .spawn_command(command) .with_context(|| format!("failed to spawn {}", request.program))?; @@ -465,6 +468,8 @@ impl TerminalHandle { // Disconnecting this channel wakes every cancellable queue operation, // even when the actor cannot consume an ordinary shutdown message. let (shutdown_tx, shutdown) = bounded(0); + #[cfg(windows)] + let (close_tx, close_rx) = bounded(1); let (init_tx, init_rx) = std_mpsc::sync_channel(1); let actor_info = Arc::clone(&info); @@ -474,7 +479,11 @@ impl TerminalHandle { .name(format!("opencode-pty-actor-{id}")) .spawn(move || { let result = run_actor(ActorConfig { - master: pair.master, + master: ActorMaster { + pty: Some(pair.master), + #[cfg(windows)] + close_tx, + }, messages: actor_rx, writes: actor_writer_tx, shutdown: actor_shutdown, @@ -555,10 +564,16 @@ impl TerminalHandle { let wait_thread = thread::Builder::new() .name(format!("opencode-pty-wait-{id}")) .spawn(move || { - let result = child.wait().map(|status| Some(status.exit_code())); - let _ = wait_actor_tx.send(ActorMessage::ChildExited( - result.map_err(|error| error.to_string()), - )); + #[cfg(windows)] + windows::wait_and_close(child, wait_actor_tx, close_rx); + #[cfg(unix)] + { + let mut child = child; + let result = child.wait().map(|status| Some(status.exit_code())); + let _ = wait_actor_tx.send(ActorMessage::ChildExited( + result.map_err(|error| error.to_string()), + )); + } }) .context("failed to spawn terminal child waiter")?; @@ -684,8 +699,38 @@ enum WriterMessage { Bytes(Vec), } +struct ActorMaster { + pty: Option>, + #[cfg(windows)] + close_tx: Sender>, +} + +impl ActorMaster { + fn get(&self) -> Result<&(dyn MasterPty + Send)> { + self.pty + .as_deref() + .ok_or_else(|| anyhow!("terminal child has exited")) + } + + #[cfg(windows)] + fn close(&mut self) { + if let Some(master) = self.pty.take() { + // A capacity-one channel receives this unique master exactly once; + // handoff cannot wait on ClosePseudoConsole or PTY output drainage. + let _ = self.close_tx.send(master); + } + } +} + +impl Drop for ActorMaster { + fn drop(&mut self) { + #[cfg(windows)] + self.close(); + } +} + struct ActorConfig { - master: Box, + master: ActorMaster, messages: Receiver, writes: Sender, shutdown: Receiver<()>, @@ -708,6 +753,8 @@ fn run_actor(config: ActorConfig) -> Result<()> { rows, init_tx, } = config; + #[cfg(windows)] + let mut master = master; let mut terminal = Terminal::new(TerminalOptions { cols, rows, @@ -727,14 +774,16 @@ fn run_actor(config: ActorConfig) -> Result<()> { let result = loop { match receive_actor_message(&messages, &mut pending_message, &shutdown) { Ok(ActorMessage::RefreshForegroundProcess { reply }) => { - publish_foreground_process( - &shutdown, - &*master, - &info, - &mut subscribers, - &mut controller, - &mut controller_generation, - ); + if let Ok(master) = master.get() { + publish_foreground_process( + &shutdown, + master, + &info, + &mut subscribers, + &mut controller, + &mut controller_generation, + ); + } let _ = reply.send(Ok(())); } Ok(ActorMessage::Output(bytes)) => { @@ -778,8 +827,12 @@ fn run_actor(config: ActorConfig) -> Result<()> { bytes, reply, }) => { - let result = authorize_controller(&controller, attachment_id.as_deref()) - .and_then(|_| queue_write(&writes, &shutdown, bytes)); + let result = + authorize_controller(&controller, attachment_id.as_deref()).and_then(|_| { + #[cfg(windows)] + master.get()?; + queue_write(&writes, &shutdown, bytes) + }); let _ = reply.send(result); } Ok(ActorMessage::Resize { @@ -790,7 +843,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { }) => { let result = (|| { authorize_controller(&controller, attachment_id.as_deref())?; - master.resize(PtySize { + master.get()?.resize(PtySize { rows, cols, pixel_width: 0, @@ -861,7 +914,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { .map(|value| value.cols != cols || value.rows != rows) .unwrap_or(true); if changed { - master.resize(PtySize { + master.get()?.resize(PtySize { rows, cols, pixel_width: 0, @@ -890,6 +943,8 @@ fn run_actor(config: ActorConfig) -> Result<()> { ); } if let Some(bytes) = bytes { + #[cfg(windows)] + master.get()?; queue_write(&writes, &shutdown, bytes)?; } Ok(()) @@ -997,9 +1052,16 @@ fn run_actor(config: ActorConfig) -> Result<()> { } Ok(ActorMessage::ChildExited(result)) => { child_exit = Some(result); + // ConPTY keeps its output pipe open until HPCON is closed. + // The wait worker closes it while this actor and the reader + // keep processing output, including the real final EOF. + #[cfg(windows)] + master.close(); } Ok(ActorMessage::ReaderFailed(message)) | Ok(ActorMessage::WriterFailed(message)) => { - if let Ok(mut value) = info.write() { + // Closing ConPTY can finish a pending write with BrokenPipe + // after the final exit event. Never overwrite a published exit. + if !exit_published && let Ok(mut value) = info.write() { value.lifecycle = TerminalLifecycle::Failed { message }; } } @@ -1518,6 +1580,14 @@ mod tests { let (actor_tx, messages) = bounded(ACTOR_QUEUE_CAPACITY); let (writes, writer_rx) = bounded(1); let (shutdown_tx, shutdown) = bounded(0); + #[cfg(windows)] + let (close_tx, close_rx) = bounded(1); + #[cfg(windows)] + let closer = thread::spawn(move || { + if let Ok(master) = close_rx.recv() { + drop(master); + } + }); let (init_tx, init_rx) = std_mpsc::sync_channel(1); let info = Arc::new(RwLock::new(TerminalInfo { id: 1, @@ -1536,7 +1606,11 @@ mod tests { let actor_info = Arc::clone(&info); let actor = thread::spawn(move || { run_actor(ActorConfig { - master: Box::new(TestMaster), + master: ActorMaster { + pty: Some(Box::new(TestMaster)), + #[cfg(windows)] + close_tx, + }, messages, writes, shutdown, @@ -1555,7 +1629,11 @@ mod tests { shutdown_tx: Mutex::new(Some(shutdown_tx)), killer: Mutex::new(Box::new(TestKiller)), info, - joins: Mutex::new(Some(vec![actor])), + joins: Mutex::new(Some(vec![ + actor, + #[cfg(windows)] + closer, + ])), }, writer_rx, ) @@ -1603,6 +1681,14 @@ mod tests { if !eof_first { assert_eq!(snapshot.text, "final"); } + actor + .actor_tx + .send(ActorMessage::WriterFailed("late broken pipe".into())) + .unwrap(); + let after_write_error = actor + .request(|reply| ActorMessage::Snapshot { reply }) + .unwrap(); + assert_eq!(after_write_error.info.lifecycle, snapshot.info.lifecycle); actor.shutdown().unwrap(); actor.shutdown().unwrap(); } diff --git a/src/service/windows.rs b/src/service/windows.rs new file mode 100644 index 0000000..332bd7b --- /dev/null +++ b/src/service/windows.rs @@ -0,0 +1,63 @@ +use std::time::Duration; + +use crossbeam_channel::{Receiver, RecvTimeoutError, Sender}; +use portable_pty::{Child, MasterPty}; +use windows_sys::Win32::{Foundation::WAIT_TIMEOUT, System::Threading::WaitForSingleObject}; + +use super::ActorMessage; + +/// Keep blocking ClosePseudoConsole off the actor/reader without introducing +/// another worker. The actor hands its unique master here on root exit or on +/// teardown; closing it then produces the real output-pipe EOF. +/// Root-exit detection has at most 10 ms polling latency. After reaping, the +/// worker blocks on the close channel rather than continuing to poll. +pub(super) fn wait_and_close( + mut child: Box, + events: Sender, + close: Receiver>, +) { + let mut reported = false; + loop { + if !reported { + // SAFETY: the child owns this process handle throughout this call; + // cloned killer handles do not close or replace it. Poll the wait + // handle rather than GetExitCodeProcess so exit code 259 is not + // confused with STILL_ACTIVE by portable-pty's try_wait. + let ready = child + .as_raw_handle() + .is_none_or(|handle| unsafe { WaitForSingleObject(handle, 0) != WAIT_TIMEOUT }); + if ready { + report_exit(&mut *child, &events); + reported = true; + } + } + let request = if reported { + close.recv().map_err(|_| RecvTimeoutError::Disconnected) + } else { + close.recv_timeout(Duration::from_millis(10)) + }; + match request { + Ok(master) => { + if !reported { + // Also cover actor failure/early teardown, not just normal + // exit or the service's already-issued termination request. + let _ = child.kill(); + } + drop(master); + break; + } + Err(RecvTimeoutError::Disconnected) => break, + Err(RecvTimeoutError::Timeout) => {} + } + } + if !reported { + report_exit(&mut *child, &events); + } +} + +fn report_exit(child: &mut dyn Child, events: &Sender) { + let result = child.wait().map(|status| Some(status.exit_code())); + let _ = events.send(ActorMessage::ChildExited( + result.map_err(|error| error.to_string()), + )); +} diff --git a/tests/runtime.rs b/tests/runtime.rs index 98dbc18..4efdb66 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -246,6 +246,16 @@ fn independent_terminals_keep_bounded_replay() { #[test] fn normal_exit_follows_final_output_and_preserves_exit_code() { + check_normal_exit(23); +} + +#[test] +#[cfg(windows)] +fn windows_exit_code_259_is_not_mistaken_for_a_running_child() { + check_normal_exit(259); +} + +fn check_normal_exit(code: u32) { let _deadline = Deadline::new(); let fixture = Fixture::new(); let service = TerminalService::default(); @@ -263,7 +273,7 @@ fn normal_exit_follows_final_output_and_preserves_exit_code() { let mut bytes = observer.replay.bytes.clone(); let mut offset = observer.replay.end_offset; child.command(Command::Output("\r\nFINAL_OUTPUT_BEFORE_EXIT\r\n".into())); - child.command(Command::Exit(23)); + child.command(Command::Exit(code as i32)); loop { match observer .events @@ -288,7 +298,7 @@ fn normal_exit_follows_final_output_and_preserves_exit_code() { exit_code, final_offset, } => { - assert_eq!(exit_code, Some(23)); + assert_eq!(exit_code, Some(code)); assert_eq!(final_offset, offset); break; } @@ -304,11 +314,34 @@ fn normal_exit_follows_final_output_and_preserves_exit_code() { assert_eq!( snapshot.info.lifecycle, TerminalLifecycle::Exited { - exit_code: Some(23) + exit_code: Some(code) } ); assert_eq!(snapshot.info.output_tail, offset); assert!(observer.events.try_recv().is_err()); + #[cfg(windows)] + { + assert!( + service + .resize(info.id, 100, 40) + .unwrap_err() + .to_string() + .contains("child has exited") + ); + assert!(service.write(info.id, b"late input".to_vec()).is_err()); + assert_eq!( + service.snapshot(info.id).unwrap().info.lifecycle, + snapshot.info.lifecycle + ); + assert!( + service + .read_rows(info.id, None) + .unwrap() + .lines + .iter() + .any(|line| line.contains("FINAL_OUTPUT_BEFORE_EXIT")) + ); + } drop(observer); service.terminate(info.id).unwrap(); } From 46c89c6952da684811981ddd52f8d7dca2dfda23 Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 3 Sep 2026 16:46:04 +0000 Subject: [PATCH 4/5] fix(pty): await Windows process teardown before joining --- README.md | 4 ++-- src/service.rs | 29 +++++++++++++++++++++++++---- src/service/windows.rs | 20 +++++++++++++++++--- 3 files changed, 44 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 10b56f8..994dc56 100644 --- a/README.md +++ b/README.md @@ -286,8 +286,8 @@ complete ConPTY shutdown behavior. On Windows, normal root-child exit hands the ConPTY master to the existing child-wait worker for closing. The actor and reader continue draining until the -real output-pipe EOF; only then is the final exit event published. Detecting the -root's exit adds at most 10 ms of polling latency. After the master is handed +real output-pipe EOF; only then is the final exit event published. The worker +checks the root's wait handle at 10 ms intervals. After the master is handed off, Windows input and resize requests fail with a child-exited error; final snapshots, rows, and replay remain readable until the terminal is removed. Unix post-exit PTY operations retain their existing behavior. diff --git a/src/service.rs b/src/service.rs index 11b93a2..92c71f7 100644 --- a/src/service.rs +++ b/src/service.rs @@ -713,19 +713,25 @@ impl ActorMaster { } #[cfg(windows)] - fn close(&mut self) { + fn close(&mut self) -> Result<()> { if let Some(master) = self.pty.take() { // A capacity-one channel receives this unique master exactly once; // handoff cannot wait on ClosePseudoConsole or PTY output drainage. - let _ = self.close_tx.send(master); + if let Err(error) = self.close_tx.send(master) { + // Keep ownership on failure. The actor stops receiving before + // dropping this fallback master, so its reader can still drain. + self.pty = Some(error.0); + bail!("terminal close worker stopped"); + } } + Ok(()) } } impl Drop for ActorMaster { fn drop(&mut self) { #[cfg(windows)] - self.close(); + let _ = self.close(); } } @@ -1056,7 +1062,9 @@ fn run_actor(config: ActorConfig) -> Result<()> { // The wait worker closes it while this actor and the reader // keep processing output, including the real final EOF. #[cfg(windows)] - master.close(); + if let Err(error) = master.close() { + break Err(error); + } } Ok(ActorMessage::ReaderFailed(message)) | Ok(ActorMessage::WriterFailed(message)) => { // Closing ConPTY can finish a pending write with BrokenPipe @@ -1694,6 +1702,19 @@ mod tests { } } + #[test] + #[cfg(windows)] + fn failed_close_handoff_retains_master_ownership() { + let (close_tx, close_rx) = bounded(1); + drop(close_rx); + let mut master = ActorMaster { + pty: Some(Box::new(TestMaster)), + close_tx, + }; + assert!(master.close().is_err()); + assert!(master.get().is_ok()); + } + #[test] fn shutdown_interrupts_controller_backpressure() { let (actor, _writer) = test_actor(); diff --git a/src/service/windows.rs b/src/service/windows.rs index 332bd7b..81c8cd4 100644 --- a/src/service/windows.rs +++ b/src/service/windows.rs @@ -2,14 +2,17 @@ use std::time::Duration; use crossbeam_channel::{Receiver, RecvTimeoutError, Sender}; use portable_pty::{Child, MasterPty}; -use windows_sys::Win32::{Foundation::WAIT_TIMEOUT, System::Threading::WaitForSingleObject}; +use windows_sys::Win32::{ + Foundation::{WAIT_OBJECT_0, WAIT_TIMEOUT}, + System::Threading::{INFINITE, WaitForSingleObject}, +}; use super::ActorMessage; /// Keep blocking ClosePseudoConsole off the actor/reader without introducing /// another worker. The actor hands its unique master here on root exit or on /// teardown; closing it then produces the real output-pipe EOF. -/// Root-exit detection has at most 10 ms polling latency. After reaping, the +/// Root-exit detection uses a 10 ms poll interval. After reaping, the /// worker blocks on the close channel rather than continuing to poll. pub(super) fn wait_and_close( mut child: Box, @@ -56,7 +59,18 @@ pub(super) fn wait_and_close( } fn report_exit(child: &mut dyn Child, events: &Sender) { - let result = child.wait().map(|status| Some(status.exit_code())); + let result = (|| { + if let Some(handle) = child.as_raw_handle() { + // SAFETY: this worker owns the child and therefore its wait handle. + // GetExitCodeProcess can expose the termination status before the + // process object is signaled; portable-pty's wait fast path is not + // sufficient to prove that TerminateProcess teardown has finished. + if unsafe { WaitForSingleObject(handle, INFINITE) } != WAIT_OBJECT_0 { + return Err(std::io::Error::last_os_error()); + } + } + child.wait().map(|status| Some(status.exit_code())) + })(); let _ = events.send(ActorMessage::ChildExited( result.map_err(|error| error.to_string()), )); From 3f2a883738f9b96eb00c8431377afd1da1eda575 Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 8 Oct 2026 18:47:34 +0000 Subject: [PATCH 5/5] fix(pty): fail queued requests when the actor stops Shutdown now overtakes queued actor work, so a request accepted before shutdown could wait forever: a stopped actor's buffered messages, and their reply senders, stay alive while other handles hold queue senders. Replies now use crossbeam channels and waiters also observe the actor thread's lifetime. Also discard terminal replies after the Windows master handoff, so a writer stopped by ClosePseudoConsole cannot end the actor before the real EOF, and give the descendant-exit assertion a bounded wait. --- src/service.rs | 91 ++++++++++++++++++++++++++++++++++++------------ tests/runtime.rs | 4 ++- 2 files changed, 71 insertions(+), 24 deletions(-) diff --git a/src/service.rs b/src/service.rs index 92c71f7..1afe4a5 100644 --- a/src/service.rs +++ b/src/service.rs @@ -328,7 +328,7 @@ impl TerminalService { takeover: bool, ) -> Result { let terminal = self.get(id)?; - let (reply_tx, reply_rx) = std_mpsc::sync_channel(1); + let (reply_tx, reply_rx) = bounded(1); terminal .actor_tx .send(ActorMessage::Attach { @@ -339,9 +339,7 @@ impl TerminalService { reply: reply_tx, }) .map_err(|_| anyhow!("terminal actor stopped"))?; - let attached = reply_rx - .recv() - .context("terminal actor dropped attach response")??; + let attached = await_reply(&reply_rx, &terminal.actor_done)?; Ok(TerminalAttachment { terminal: terminal.info(), role, @@ -403,6 +401,8 @@ impl Drop for TerminalService { struct TerminalHandle { actor_tx: Sender, + /// Disconnects when the actor thread exits; never carries a message. + actor_done: Receiver<()>, shutdown_tx: Mutex>>, killer: Mutex>, info: Arc>, @@ -475,9 +475,11 @@ impl TerminalHandle { let actor_info = Arc::clone(&info); let actor_writer_tx = writer_tx.clone(); let actor_shutdown = shutdown.clone(); + let (actor_done_tx, actor_done) = bounded::<()>(0); let actor_thread = thread::Builder::new() .name(format!("opencode-pty-actor-{id}")) .spawn(move || { + let _actor_done = actor_done_tx; let result = run_actor(ActorConfig { master: ActorMaster { pty: Some(pair.master), @@ -579,6 +581,7 @@ impl TerminalHandle { Ok(Self { actor_tx, + actor_done, shutdown_tx: Mutex::new(Some(shutdown_tx)), killer: Mutex::new(killer), info, @@ -598,15 +601,12 @@ impl TerminalHandle { .unwrap_or_else(|poisoned| poisoned.into_inner().clone()) } - fn request( - &self, - make: impl FnOnce(std_mpsc::SyncSender>) -> ActorMessage, - ) -> Result { - let (reply_tx, reply_rx) = std_mpsc::sync_channel(1); + fn request(&self, make: impl FnOnce(Reply>) -> ActorMessage) -> Result { + let (reply_tx, reply_rx) = bounded(1); self.actor_tx .send(make(reply_tx)) .map_err(|_| anyhow!("terminal actor stopped"))?; - reply_rx.recv().context("terminal actor dropped response")? + await_reply(&reply_rx, &self.actor_done) } fn shutdown(&self) -> Result<()> { @@ -633,6 +633,21 @@ impl TerminalHandle { } } +type Reply = Sender; + +/// A request accepted into the actor queue may never be handled: shutdown +/// overtakes queued work, and a stopped actor's buffered messages (with their +/// reply senders) stay alive while other handles hold queue senders. Waiting on +/// the actor's lifetime as well as the reply keeps callers from blocking forever. +fn await_reply(reply: &Receiver>, actor_done: &Receiver<()>) -> Result { + crossbeam_channel::select_biased! { + recv(reply) -> result => result.map_err(|_| anyhow!("terminal actor dropped response"))?, + recv(actor_done) -> _ => reply + .try_recv() + .map_err(|_| anyhow!("terminal actor stopped"))?, + } +} + enum ActorMessage { Output(Vec), ReaderEof, @@ -640,43 +655,43 @@ enum ActorMessage { WriterFailed(String), ChildExited(Result, String>), RefreshForegroundProcess { - reply: std_mpsc::SyncSender>, + reply: Reply>, }, Write { attachment_id: Option, bytes: Vec, - reply: std_mpsc::SyncSender>, + reply: Reply>, }, Resize { attachment_id: Option, cols: u16, rows: u16, - reply: std_mpsc::SyncSender>, + reply: Reply>, }, Control { attachment_id: String, cols: u16, rows: u16, bytes: Option>, - reply: std_mpsc::SyncSender>, + reply: Reply>, }, Snapshot { - reply: std_mpsc::SyncSender>, + reply: Reply>, }, ReadRows { rows: Option, - reply: std_mpsc::SyncSender>, + reply: Reply>, }, Replay { offset: u64, - reply: std_mpsc::SyncSender>, + reply: Reply>, }, Attach { offset: u64, attachment_id: String, role: AttachmentRole, takeover: bool, - reply: std_mpsc::SyncSender>, + reply: Reply>, }, Detach { attachment_id: String, @@ -795,7 +810,16 @@ fn run_actor(config: ActorConfig) -> Result<()> { Ok(ActorMessage::Output(bytes)) => { let (start, end) = replay.append(&bytes); terminal.vt_write(&bytes); - if let Err(error) = forward_replies(&mut terminal, &writes, &shutdown) { + // Once the Windows master is handed off for closing, the writer + // may already have stopped; replies to the closing console are + // discarded so final output still reaches the real EOF. + #[cfg(windows)] + let closing = master.get().is_err(); + #[cfg(not(windows))] + let closing = false; + if closing { + terminal.take_writes(); + } else if let Err(error) = forward_replies(&mut terminal, &writes, &shutdown) { break Err(error); } update_offsets(&info, &replay); @@ -1612,7 +1636,9 @@ mod tests { output_tail: 0, })); let actor_info = Arc::clone(&info); + let (actor_done_tx, actor_done) = bounded::<()>(0); let actor = thread::spawn(move || { + let _actor_done = actor_done_tx; run_actor(ActorConfig { master: ActorMaster { pty: Some(Box::new(TestMaster)), @@ -1634,6 +1660,7 @@ mod tests { ( TerminalHandle { actor_tx, + actor_done, shutdown_tx: Mutex::new(Some(shutdown_tx)), killer: Mutex::new(Box::new(TestKiller)), info, @@ -1770,7 +1797,7 @@ mod tests { }) .unwrap(); assert!(writer.is_full()); - let (reply, _reply_rx) = std_mpsc::sync_channel(1); + let (reply, _reply_rx) = bounded(1); actor .actor_tx .send(ActorMessage::Write { @@ -1779,15 +1806,33 @@ mod tests { reply, }) .unwrap(); + // Shutdown overtakes this queued request; its caller must still return. + let actor = Arc::new(actor); + let (queued_tx, queued) = std_mpsc::channel(); + let requester = thread::spawn({ + let actor = Arc::clone(&actor); + move || { + let _ = queued_tx.send(actor.request(|reply| ActorMessage::Snapshot { reply })); + } + }); let (done, finished) = std_mpsc::channel(); - let shutdown = thread::spawn(move || { - actor.shutdown().unwrap(); - done.send(()).unwrap(); + let shutdown = thread::spawn({ + let actor = Arc::clone(&actor); + move || { + actor.shutdown().unwrap(); + done.send(()).unwrap(); + } }); let stopped = finished.recv_timeout(Duration::from_secs(1)).is_ok(); + let queued = queued.recv_timeout(Duration::from_secs(1)); drop(writer); shutdown.join().unwrap(); assert!(stopped, "shutdown waited for the full input queue"); + assert!( + queued.is_ok_and(|result| result.is_err()), + "a request queued before shutdown never returned" + ); + requester.join().unwrap(); } fn rows_terminal(cols: u16, rows: u16, input: &str) -> Terminal { diff --git a/tests/runtime.rs b/tests/runtime.rs index 4efdb66..27eab0d 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -473,9 +473,11 @@ impl ChildProcess { use windows_sys::Win32::{ Foundation::WAIT_OBJECT_0, System::Threading::WaitForSingleObject, }; + // Descendants exit asynchronously once their console closes, so allow + // a bounded wait. The root is already signaled when terminate returns. // SAFETY: the owned process handle has wait access and stays live. assert_eq!( - unsafe { WaitForSingleObject(self.handle.as_raw_handle(), 0) }, + unsafe { WaitForSingleObject(self.handle.as_raw_handle(), 5000) }, WAIT_OBJECT_0 ); }