From cf7306b01114a66b7c53641ef8546ba1f4b31d62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C3=ADas=20D=2E?= <9351115+decode2@users.noreply.github.com> Date: Sat, 18 Jul 2026 07:23:15 +0000 Subject: [PATCH] feat(pty): add platform-neutral session contract --- crates/splice-pty/src/lib.rs | 185 ++++++++++++++++++-- crates/splice-pty/tests/session_contract.rs | 41 +++++ 2 files changed, 209 insertions(+), 17 deletions(-) create mode 100644 crates/splice-pty/tests/session_contract.rs diff --git a/crates/splice-pty/src/lib.rs b/crates/splice-pty/src/lib.rs index c560442..2c5afe8 100644 --- a/crates/splice-pty/src/lib.rs +++ b/crates/splice-pty/src/lib.rs @@ -1,5 +1,86 @@ pub mod flow; +/// A backend-independent event emitted by a PTY session. +/// +/// Output always carries the originating session id, including output that +/// arrives before a frontend listener is ready. A natural exit is distinct +/// from an explicit close so callers do not restart a deliberately closed tab. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PtySessionEvent { + Output { session_id: u64, data: String }, + NaturalExit { session_id: u64 }, +} + +impl PtySessionEvent { + pub fn from_output(session_id: u64, data: String) -> Self { + Self::Output { session_id, data } + } + + pub fn natural_exit(session_id: u64) -> Self { + Self::NaturalExit { session_id } + } + + pub fn session_id(&self) -> u64 { + match self { + Self::Output { session_id, .. } | Self::NaturalExit { session_id } => *session_id, + } + } + + pub fn output(&self) -> Option<&str> { + match self { + Self::Output { data, .. } => Some(data), + Self::NaturalExit { .. } => None, + } + } +} + +/// Shared lifecycle state for target-specific PTY sessions. +/// +/// `begin_close` returns true only for the first close, which gives every +/// backend an exactly-once teardown boundary and suppresses natural-exit +/// notifications caused by an explicit close. +pub struct PtySessionLifecycle { + id: u64, + closing: std::sync::atomic::AtomicBool, +} + +impl PtySessionLifecycle { + pub fn new(id: u64) -> Self { + Self { + id, + closing: std::sync::atomic::AtomicBool::new(false), + } + } + + pub fn id(&self) -> u64 { + self.id + } + + pub fn begin_close(&self) -> bool { + !self.closing.swap(true, std::sync::atomic::Ordering::SeqCst) + } + + pub fn should_emit_natural_exit(&self) -> bool { + !self.closing.load(std::sync::atomic::Ordering::SeqCst) + } +} + +/// Observable methods every target-specific PTY session must provide. +/// +/// Spawning takes output and natural-exit callbacks; implementations must +/// attribute both callbacks with `id`, preserve the shared credit-window ACK +/// semantics, and make `close` idempotent. +pub trait PtySessionContract { + fn id(&self) -> u64; + fn write(&self, data: &str) -> Result<(), PtyError>; + fn interrupt(&self) -> Result<(), PtyError>; + fn resize(&self, size: TerminalSize) -> Result<(), PtyError>; + fn is_running(&self) -> Result; + fn active_process_name(&self) -> Result; + fn active_process_candidates(&self) -> Result, PtyError>; + fn close(&self); +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct TerminalSize { pub columns: u16, @@ -176,6 +257,43 @@ impl PtySession { pub fn active_process_candidates(&self) -> Result, PtyError> { Err(PtyError::UnsupportedPlatform) } + + pub fn close(&self) {} +} + +#[cfg(not(windows))] +impl PtySessionContract for PtySession { + fn id(&self) -> u64 { + self.id() + } + + fn write(&self, data: &str) -> Result<(), PtyError> { + self.write(data) + } + + fn interrupt(&self) -> Result<(), PtyError> { + self.interrupt() + } + + fn resize(&self, size: TerminalSize) -> Result<(), PtyError> { + self.resize(size) + } + + fn is_running(&self) -> Result { + self.is_running() + } + + fn active_process_name(&self) -> Result { + self.active_process_name() + } + + fn active_process_candidates(&self) -> Result, PtyError> { + self.active_process_candidates() + } + + fn close(&self) { + self.close(); + } } #[cfg(windows)] @@ -211,15 +329,50 @@ pub fn run_conpty_command_with_resize( #[cfg(windows)] pub use windows_conpty::PtySession; +#[cfg(windows)] +impl PtySessionContract for PtySession { + fn id(&self) -> u64 { + self.id() + } + + fn write(&self, data: &str) -> Result<(), PtyError> { + self.write(data) + } + + fn interrupt(&self) -> Result<(), PtyError> { + self.interrupt() + } + + fn resize(&self, size: TerminalSize) -> Result<(), PtyError> { + self.resize(size) + } + + fn is_running(&self) -> Result { + self.is_running() + } + + fn active_process_name(&self) -> Result { + self.active_process_name() + } + + fn active_process_candidates(&self) -> Result, PtyError> { + self.active_process_candidates() + } + + fn close(&self) { + self.close(); + } +} + #[cfg(windows)] mod windows_conpty { - use super::{PtyError, TerminalSize}; + use super::{PtyError, PtySessionLifecycle, TerminalSize}; use std::{ ffi::c_void, mem::size_of, ptr::null_mut, sync::{ - atomic::{AtomicBool, AtomicU64, Ordering}, + atomic::{AtomicU64, Ordering}, Arc, Mutex, }, thread::{self, JoinHandle}, @@ -278,15 +431,13 @@ mod windows_conpty { // purpose: the waiter must be able to read it without ever locking the // session mutex, which keeps the waiter off `inner` entirely and makes // a lock cycle with `close()` impossible. - closing: Arc, + lifecycle: Arc, // Join handle for this session's waiter thread. `close()` joins it (via // `Option::take`, so it is idempotent across `close()`/`Drop`) so // neither the thread nor the process handle it waits on outlives the // session. Kept outside `inner` so the join never runs while the // session lock is held. waiter: Mutex>>, - // Monotonic id assigned at spawn (see `SESSION_COUNTER`). - id: u64, // Invoked exactly once, at the very top of `close()`, BEFORE the child // is terminated and long before the reader thread is joined. // @@ -301,7 +452,7 @@ mod windows_conpty { // which makes the flusher exit and drop the channel receiver, which in // turn makes the reader's `send` return `Err`). // - // Fired from `close()` only, guarded by the `closing` flag's + // Fired from `close()` only, guarded by the lifecycle's // false->true transition, so it runs exactly once even though `close()` // is idempotent and also runs from `Drop`. on_closing: Box, @@ -377,14 +528,14 @@ mod windows_conpty { // waiter thread: the waiter is its sole owner and closes it when it // ends, so this handle is deliberately NOT stored in // `PtySessionInner` (exactly one owner). - let closing = Arc::new(AtomicBool::new(false)); - let closing_for_waiter = Arc::clone(&closing); + let lifecycle = Arc::new(PtySessionLifecycle::new(id)); + let lifecycle_for_waiter = Arc::clone(&lifecycle); let waiter_process = duplicate_process_handle(handles.process.raw())?; let waiter_handle = SendHandle(waiter_process.into_raw()); let waiter = thread::spawn(move || { // Sole owner of the duplicated process handle; dropped (and // thus `CloseHandle`d) when this thread ends. This closure - // captures only lightweight data (`id`, an `Arc`, + // captures only lightweight data (`id`, an `Arc`, // the duplicated handle, and `on_exit`) — never `inner` and // never an `Arc` — so it can never form a lock // cycle with `close()`. @@ -393,11 +544,11 @@ mod windows_conpty { WaitForSingleObject(process.raw(), INFINITE); } // Only a NATURAL exit fires the callback. If `close()` - // published `closing` before terminating the child, this wait + // began close before terminating the child, this wait // was released by that intentional teardown, so suppress - // `on_exit` (no spurious frontend restart). SeqCst pairs with - // the SeqCst store in `close()`. - if !closing_for_waiter.load(Ordering::SeqCst) { + // `on_exit` (no spurious frontend restart). The lifecycle's + // SeqCst load pairs with its SeqCst close transition. + if lifecycle_for_waiter.should_emit_natural_exit() { on_exit(id); } }); @@ -434,9 +585,8 @@ mod windows_conpty { root_process_name: program.to_owned(), job: handles.job, })), - closing, + lifecycle, waiter: Mutex::new(Some(waiter)), - id, on_closing: Box::new(on_closing), }) } @@ -444,7 +594,7 @@ mod windows_conpty { /// Monotonic id assigned to this session at spawn. Stable for the /// session's lifetime and carried in the `pty-exit` event payload. pub fn id(&self) -> u64 { - self.id + self.lifecycle.id() } pub fn write(&self, data: &str) -> Result<(), PtyError> { @@ -574,7 +724,7 @@ mod windows_conpty { // `swap` (not `store`) so the false->true transition also fires the // teardown hook EXACTLY once, even though `close()` is idempotent // and runs again from `Drop`. - if !self.closing.swap(true, Ordering::SeqCst) { + if self.lifecycle.begin_close() { // MUST run before the reader join below: with credit-based flow // control the reader can be parked inside a blocking // `on_output`, and this hook is what releases it. Running it @@ -1533,6 +1683,7 @@ mod windows_conpty { #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::AtomicBool; #[test] fn test_create_breakaway_from_job_constant_value() { diff --git a/crates/splice-pty/tests/session_contract.rs b/crates/splice-pty/tests/session_contract.rs new file mode 100644 index 0000000..a291874 --- /dev/null +++ b/crates/splice-pty/tests/session_contract.rs @@ -0,0 +1,41 @@ +use splice_pty::{ + flow::CreditWindow, PtyError, PtySession, PtySessionContract, PtySessionEvent, + PtySessionLifecycle, +}; + +fn requires_platform_neutral_contract() {} + +#[test] +fn session_contract_keeps_identity_and_early_output_attributed() { + requires_platform_neutral_contract::(); + + let early_output = PtySessionEvent::from_output(41, "ready before listener".to_owned()); + let natural_exit = PtySessionEvent::natural_exit(41); + + assert_eq!(early_output.session_id(), 41); + assert_eq!(early_output.output(), Some("ready before listener")); + assert_eq!(natural_exit.session_id(), 41); + assert_eq!(natural_exit.output(), None); +} + +#[test] +fn session_contract_distinguishes_natural_exit_from_idempotent_close() { + let lifecycle = PtySessionLifecycle::new(7); + + assert_eq!(lifecycle.id(), 7); + assert!(lifecycle.should_emit_natural_exit()); + assert!(lifecycle.begin_close()); + assert!(!lifecycle.begin_close()); + assert!(!lifecycle.should_emit_natural_exit()); +} + +#[test] +fn session_contract_preserves_input_error_and_ack_backpressure_boundaries() { + let credit = CreditWindow::new(4); + credit.consume(4); + credit.replenish(2); + credit.replenish(99); + + assert_eq!(credit.available(), 4, "stale ACKs must not exceed capacity"); + assert!(PtyError::SessionClosed.is_terminal_closed()); +}