From 9f8d07fc258959cb32e267ab6df55371cec1b2d9 Mon Sep 17 00:00:00 2001 From: "glm-5.2" Date: Tue, 7 Jul 2026 14:56:35 +0000 Subject: [PATCH] =?UTF-8?q?feat(tty-local):=20implement=20PTY=20mode=20?= =?UTF-8?q?=E2=80=94=20three-thread=20bridge,=20PtyControl,=20REQ-TTY-02,?= =?UTF-8?q?=20ADR-056=20kill=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement allocate_pty in src/pty.rs: portable_pty openpty + spawn as session leader (set_controlling_tty(true)), with the three-thread blocking→async bridge (REQ-TTY-01): reader → mpsc with zero-length EOF sentinel, writer draining mpsc (Bytes/Eof), waiter → oneshot. TtyHandle.stdout is ReceiverStream, stdin is a StdinSink AsyncWrite wrapper over mpsc::Sender (parks a reserve+send future when the channel is full), stderr is None (PTY merges), exit_code is LocalExitFuture. PtyControl implements TtyControl: resize via MasterPty::resize; signal via libc::kill(-pgid, sig) with kill(pid, sig) fallback (Unix, REQ-TTY-02) using alknet_tty::signal_from_name; unknown names and non-Unix fall back to ChildKiller::kill (SIGHUP). LocalExitFuture wraps the oneshot::Receiver + a ChildKiller kill guard (ADR-056). poll delegates to the receiver and disarms the guard on Ready (no-op Drop on happy path); Drop on cancel calls ChildKiller::kill (SIGHUP) — best-effort; the waiter thread reaps. Bump portable-pty 0.8 → 0.9 to match the POC API (MasterPty: Send). Add tokio-stream, serde_json (dev), tempfile (dev) deps. Tests: happy path (echo), stdin round-trip (cat), resize, signal INT kills child, signal reaches process group (bash -c "sleep 60"), cancel cleanup on drop, unknown signal falls back to ChildKiller. --- Cargo.lock | 117 ++--- crates/alknet-tty-local/Cargo.toml | 11 +- crates/alknet-tty-local/src/pty.rs | 698 ++++++++++++++++++++++++++++- 3 files changed, 739 insertions(+), 87 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c0630d2..8c9287a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -156,8 +156,11 @@ dependencies = [ "futures-core", "libc", "portable-pty", + "serde_json", + "tempfile", "thiserror 2.0.18", "tokio", + "tokio-stream", "tracing", ] @@ -647,6 +650,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd16c4719339c4530435d38e511904438d07cce7950afa3718a84ac36c10e89e" + [[package]] name = "cfg_aliases" version = "0.2.1" @@ -1963,15 +1972,6 @@ dependencies = [ "web-sys", ] -[[package]] -name = "ioctl-rs" -version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7970510895cee30b3e9128319f2cefd4bde883a39f38baa279567ba3a7eb97d" -dependencies = [ - "libc", -] - [[package]] name = "ipconfig" version = "0.3.4" @@ -2002,7 +2002,7 @@ dependencies = [ "atomic-waker", "backon", "bytes", - "cfg_aliases", + "cfg_aliases 0.2.1", "concurrent-queue", "crypto_box", "data-encoding", @@ -2101,7 +2101,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "76c6245c9ed906506ab9185e8d7f64857129aee4f935e899f398a3bd3b70338d" dependencies = [ "bytes", - "cfg_aliases", + "cfg_aliases 0.2.1", "iroh-quinn-proto", "iroh-quinn-udp", "pin-project-lite", @@ -2140,7 +2140,7 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c53afaa1049f7c83ea1331f5ebb9e6ebc5fdd69c468b7a22dd598b02c9bcc973" dependencies = [ - "cfg_aliases", + "cfg_aliases 0.2.1", "libc", "once_cell", "socket2 0.5.10", @@ -2156,7 +2156,7 @@ checksum = "c63f122cdfaa4b4e0e7d6d3921d2b878f42a0c6d3ee5a29456dc3f5ab5ec931f" dependencies = [ "anyhow", "bytes", - "cfg_aliases", + "cfg_aliases 0.2.1", "data-encoding", "derive_more 1.0.0", "getrandom 0.3.4", @@ -2417,15 +2417,6 @@ version = "2.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" -[[package]] -name = "memoffset" -version = "0.6.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5aa361d4faea93603064a027415f07bd8e1d5c88c9fbf68bf56a285428fd79ce" -dependencies = [ - "autocfg", -] - [[package]] name = "mime" version = "0.3.17" @@ -2493,7 +2484,7 @@ version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7bb0e5d99e681ab3c938842b96fcb41bf8a7bb4bfdb11ccbd653a7e83e06c794" dependencies = [ - "cfg_aliases", + "cfg_aliases 0.2.1", "derive_more 1.0.0", "futures-buffered", "futures-lite", @@ -2514,7 +2505,7 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e2ab99dfb861450e68853d34ae665243a88b8c493d01ba957321a1e9b2312bbe" dependencies = [ - "cfg_aliases", + "cfg_aliases 0.2.1", "derive_more 2.1.1", "futures-buffered", "futures-lite", @@ -2645,7 +2636,7 @@ checksum = "67eeaa5f7505c93c5a9b35ba84fd21fb8aa3f24678c76acfe8716af7862fb07a" dependencies = [ "atomic-waker", "bytes", - "cfg_aliases", + "cfg_aliases 0.2.1", "derive_more 1.0.0", "iroh-quinn-udp", "js-sys", @@ -2672,16 +2663,14 @@ dependencies = [ [[package]] name = "nix" -version = "0.25.1" +version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f346ff70e7dbfd675fe90590b92d59ef2de15a8779ae305ebcbfd3f0caf59be4" +checksum = "ab2156c4fce2f8df6c499cc1c763e4394b7482525bf2a9701c9d79d215f519e4" dependencies = [ - "autocfg", - "bitflags 1.3.2", + "bitflags 2.13.0", "cfg-if", + "cfg_aliases 0.1.1", "libc", - "memoffset", - "pin-utils", ] [[package]] @@ -2707,7 +2696,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "198b99fc085a5db1f7d259edb5ede8311e59f28cdd2687920b4313613d21a73f" dependencies = [ "bytes", - "cfg_aliases", + "cfg_aliases 0.2.1", "derive_more 2.1.1", "noq-proto", "noq-udp", @@ -2755,7 +2744,7 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd5a37756f168cf350d68a97c4f0158bdf3c76f10175123941569b09ab51f011" dependencies = [ - "cfg_aliases", + "cfg_aliases 0.2.1", "libc", "socket2 0.6.4", "tracing", @@ -3064,7 +3053,7 @@ dependencies = [ "async-compat", "base32", "bytes", - "cfg_aliases", + "cfg_aliases 0.2.1", "document-features", "dyn-clone", "ed25519-dalek", @@ -3183,9 +3172,9 @@ checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" [[package]] name = "portable-pty" -version = "0.8.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "806ee80c2a03dbe1a9fb9534f8d19e4c0546b790cde8fd1fea9d6390644cb0be" +checksum = "b4a596a2b3d2752d94f51fac2d4a96737b8705dddd311a32b9af47211f08671e" dependencies = [ "anyhow", "bitflags 1.3.2", @@ -3195,7 +3184,7 @@ dependencies = [ "libc", "log", "nix", - "serial", + "serial2", "shared_library", "shell-words", "winapi", @@ -3340,7 +3329,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" dependencies = [ "bytes", - "cfg_aliases", + "cfg_aliases 0.2.1", "pin-project-lite", "quinn-proto", "quinn-udp", @@ -3383,7 +3372,7 @@ version = "0.5.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd" dependencies = [ - "cfg_aliases", + "cfg_aliases 0.2.1", "libc", "once_cell", "socket2 0.6.4", @@ -4167,45 +4156,14 @@ dependencies = [ ] [[package]] -name = "serial" -version = "0.4.0" +name = "serial2" +version = "0.2.37" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1237a96570fc377c13baa1b88c7589ab66edced652e43ffb17088f003db3e86" -dependencies = [ - "serial-core", - "serial-unix", - "serial-windows", -] - -[[package]] -name = "serial-core" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f46209b345401737ae2125fe5b19a77acce90cd53e1658cda928e4fe9a64581" +checksum = "9eb6ea5562eeaed6936b8b54e086aa0f88b9e5b1bef45beb038e2519fa1185b1" dependencies = [ + "cfg-if", "libc", -] - -[[package]] -name = "serial-unix" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f03fbca4c9d866e24a459cbca71283f545a37f8e3e002ad8c70593871453cab7" -dependencies = [ - "ioctl-rs", - "libc", - "serial-core", - "termios", -] - -[[package]] -name = "serial-windows" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "15c6d3b776267a75d31bbdfd5d36c0ca051251caafc285827052bc53bcdc8162" -dependencies = [ - "libc", - "serial-core", + "windows-sys 0.61.2", ] [[package]] @@ -4551,15 +4509,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "termios" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5d9cf598a6d7ce700a4e6a9199da127e6819a61e64b68609683cc9a01b5683a" -dependencies = [ - "libc", -] - [[package]] name = "thiserror" version = "1.0.69" diff --git a/crates/alknet-tty-local/Cargo.toml b/crates/alknet-tty-local/Cargo.toml index 8e4ffb4..76dd7c0 100644 --- a/crates/alknet-tty-local/Cargo.toml +++ b/crates/alknet-tty-local/Cargo.toml @@ -11,12 +11,19 @@ name = "alknet_tty_local" [dependencies] alknet-tty = { path = "../alknet-tty" } -portable-pty = "0.8" +portable-pty = "0.9" tokio = { version = "1", features = ["full"] } +tokio-stream = "0.1" bytes = "1" futures-core = "0.3" tracing = "0.1" thiserror = "2" [target.'cfg(unix)'.dependencies] -libc = "0.2" \ No newline at end of file +libc = "0.2" + +[dev-dependencies] +tokio = { version = "1", features = ["full", "test-util", "macros"] } +tokio-stream = "0.1" +serde_json = "1" +tempfile = "3" \ No newline at end of file diff --git a/crates/alknet-tty-local/src/pty.rs b/crates/alknet-tty-local/src/pty.rs index 867f2ea..16676eb 100644 --- a/crates/alknet-tty-local/src/pty.rs +++ b/crates/alknet-tty-local/src/pty.rs @@ -5,5 +5,701 @@ //! `PtyControl` for resize/signal, REQ-TTY-02 process-group signal //! forwarding (`libc::kill(-pgid, sig)`), and the ADR-056 kill guard on the //! `exit_code` future. +//! +//! # REQ-TTY-01 — blocking→async bridge +//! +//! `portable_pty` exposes a blocking `std::io` API +//! (`MasterPty::try_clone_reader()`, `take_writer()`, `Child::wait()`). +//! The three-thread bridge is the reference pattern for any blocking +//! backend: dedicated std threads feed tokio mpsc/oneshot channels, and +//! the async-facing `TtyHandle` fields are wrappers over those channels. +//! +//! # REQ-TTY-02 — signal targets the process group +//! +//! `portable_pty` spawns the child as a session leader +//! (`CommandBuilder::set_controlling_tty(true)`, the default), so the +//! child's pid *is* its process-group id. `signal()` calls +//! `libc::kill(-pgid, sig)` (the negative pid) to reach the whole group, +//! with a `kill(pid, sig)` fallback if the group signal fails (e.g. the +//! child already exited). Unknown signal names fall back to +//! `ChildKiller::kill` (SIGHUP). See `tty-local.md` §"REQ-TTY-02". +//! +//! # ADR-056 — kill-on-Drop guard +//! +//! `LocalExitFuture` wraps the `oneshot::Receiver` from the waiter +//! thread alongside a `portable_pty::ChildKiller`. On cancel (the future +//! is dropped without being driven to completion), `Drop` calls +//! `ChildKiller::kill()` (SIGHUP) — best-effort, the child may already be +//! exiting. On the happy path (the future resolves), `poll` disarms the +//! guard (`Option::take()`), so the subsequent `Drop` is a no-op. The +//! waiter thread reaps the killed child via its blocking `wait()`, so +//! there is no zombie. See `tty-local.md` §"Cancel-Cleanup (ADR-056)". -// TODO: implement +use std::collections::HashMap; +use std::future::Future; +use std::io::{Read, Write}; +use std::path::PathBuf; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::Mutex; +use std::task::{Context, Poll}; +use std::thread; + +use alknet_tty::backend::{ + BoxFuture, TerminalParams, TtyControl, TtyControlHandle, TtyError, TtyHandle, +}; +use alknet_tty::control::signal_from_name; +use bytes::Bytes; +use futures_core::Stream; +use portable_pty::{native_pty_system, ChildKiller, CommandBuilder, MasterPty, PtySize}; +use tokio::sync::{mpsc, oneshot}; +use tokio_stream::wrappers::ReceiverStream; +use tracing::{debug, warn}; + +/// Channel command for the writer thread: bytes to write, or EOF. +/// +/// `Bytes` writes the bytes to the master writer and flushes. `Eof` drops +/// the writer (sends EOF to the slave's stdin) and exits the writer thread. +pub enum StdinCmd { + /// Write these bytes to the master writer (write + flush). + Bytes(Vec), + /// Close the master writer (EOF to the slave's stdin). The writer thread + /// drops the writer and exits. + Eof, +} + +/// Control handle for a live PTY: resize + signal forwarding. Cheap to +/// clone (all fields are `Arc`-backed) so the adapter can hand a clone to +/// the spawned control-chunk dispatcher. +#[derive(Clone)] +pub struct PtyControl { + master: Arc>>, + killer: Arc>>, + pid: Option, +} + +impl PtyControl { + /// Construct from the shared master, the cloned killer, and the child's + /// pid (used for process-group signal targeting). + pub fn new( + master: Arc>>, + killer: Arc>>, + pid: Option, + ) -> Self { + Self { + master, + killer, + pid, + } + } +} + +impl TtyControl for PtyControl { + /// Resize the PTY. Safe to call from the async pump — + /// `MasterPty::resize` is non-blocking (it issues an `ioctl`). + fn resize(&self, cols: u16, rows: u16, pixel_width: u16, pixel_height: u16) { + let size = PtySize { + cols, + rows, + pixel_width, + pixel_height, + }; + let master = self.master.lock().expect("master mutex poisoned"); + if let Err(e) = master.resize(size) { + warn!("pty resize failed: {e}"); + } + } + + /// Forward a signal by name to the child's process group (REQ-TTY-02). + /// + /// On Unix: maps the name to a `libc` signal number via + /// `alknet_tty::signal_from_name`, then calls `kill(-pgid, sig)` (the + /// negative pid reaches the whole process group — the child is a + /// session leader because `set_controlling_tty(true)`, the default). + /// Falls back to `kill(pid, sig)` if the group signal fails (e.g. the + /// child already exited). Unknown names fall back to + /// `ChildKiller::kill` (SIGHUP). + /// + /// On non-Unix: `ChildKiller::kill` (SIGHUP) directly. + fn signal(&self, name: &str) { + #[cfg(unix)] + { + if let Some(pid) = self.pid { + if let Some(sig) = signal_from_name(name) { + let pgid = pid as i32; + let r = unsafe { libc::kill(-pgid, sig) }; + if r == 0 { + return; + } + let err = std::io::Error::last_os_error(); + let r2 = unsafe { libc::kill(pgid, sig) }; + if r2 == 0 { + return; + } + warn!( + "pty signal `{name}` (group {pgid}) failed: {err}; \ + direct kill also failed: {}", + std::io::Error::last_os_error() + ); + return; + } + } + // Unknown name or no pid: fall back to ChildKiller (SIGHUP). + let mut killer = self.killer.lock().expect("killer mutex poisoned"); + if let Err(e) = killer.kill() { + warn!("pty fallback ChildKiller::kill failed: {e}"); + } + } + + #[cfg(not(unix))] + { + let _ = name; + let mut killer = self.killer.lock().expect("killer mutex poisoned"); + if let Err(e) = killer.kill() { + warn!("pty ChildKiller::kill failed: {e}"); + } + } + } +} + +/// ADR-056 kill guard wrapping the waiter-thread oneshot + a +/// `portable_pty::ChildKiller`. +/// +/// `poll` delegates to the oneshot receiver (resolves on natural exit). On +/// `Ready`, the killer is taken (`Option::take()`) — disarmed — so the +/// subsequent `Drop` is a no-op. On cancel (the future is dropped before +/// resolving), `Drop` calls `ChildKiller::kill()` (SIGHUP) — best-effort; +/// the child may already be exiting. The waiter thread's blocking `wait()` +/// reaps the killed child, so there is no zombie. The contract is "kill on +/// cancel; no-op on resolve." +pub struct LocalExitFuture { + rx: oneshot::Receiver, + killer: Option>, +} + +impl LocalExitFuture { + fn new(rx: oneshot::Receiver, killer: Box) -> Self { + Self { + rx, + killer: Some(killer), + } + } +} + +impl Future for LocalExitFuture { + type Output = Result; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + match Pin::new(&mut self.rx).poll(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Ok(code)) => { + // Disarm the kill guard — the child exited naturally. + self.killer.take(); + Poll::Ready(Ok(code)) + } + Poll::Ready(Err(_)) => { + // The waiter thread's oneshot sender was dropped (wait() + // failed). Disarm to avoid killing an already-reaped child. + self.killer.take(); + Poll::Ready(Err(TtyError::WaitFailed { + message: "waiter thread exited without sending exit code".to_string(), + })) + } + } + } +} + +impl Drop for LocalExitFuture { + fn drop(&mut self) { + if let Some(killer) = self.killer.take() { + // ADR-056: kill on cancel. Best-effort — the child may already + // be exiting. SIGHUP is what portable_pty sends on Unix. + let mut killer = killer; + if let Err(e) = killer.kill() { + debug!("LocalExitFuture drop: ChildKiller::kill failed: {e}"); + } + } + } +} + +/// `AsyncWrite` adapter over an `mpsc::Sender`. `poll_write` sends +/// `StdinCmd::Bytes(buf.to_vec())`; `poll_flush` is a no-op (the writer +/// thread flushes); `poll_close` sends `StdinCmd::Eof`. +/// +/// When the channel is full, `poll_write` parks in an in-flight send +/// future stored on the struct (so a re-poll resumes the same send rather +/// than starting a new one — `reserve()` is not `Unpin`). +struct StdinSink { + tx: mpsc::Sender, + /// In-flight `reserve()` + send, captured as a boxed future. `None` + /// when no write is pending. + inflight: Option, + /// Bytes for the in-flight write (returned as the write count on + /// completion). + inflight_len: usize, + close_sent: bool, +} + +/// Boxed future for an in-flight stdin `reserve + send`. The permit borrows +/// the sender, so the future owns a cloned sender and the bytes to send. +type InflightSend = Pin>> + Send>>; + +impl StdinSink { + fn new(tx: mpsc::Sender) -> Self { + Self { + tx, + inflight: None, + inflight_len: 0, + close_sent: false, + } + } +} + +impl tokio::io::AsyncWrite for StdinSink { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + // Drain any in-flight write first. + if let Some(fut) = self.inflight.as_mut() { + match fut.as_mut().poll(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Ok(())) => { + let n = self.inflight_len; + self.inflight = None; + self.inflight_len = 0; + return Poll::Ready(Ok(n)); + } + Poll::Ready(Err(_)) => { + self.inflight = None; + self.inflight_len = 0; + return Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "stdin channel closed", + ))); + } + } + } + // Fast path: try to send without parking. + match self.tx.try_send(StdinCmd::Bytes(buf.to_vec())) { + Ok(()) => Poll::Ready(Ok(buf.len())), + Err(mpsc::error::TrySendError::Full(_)) => { + // Park a `reserve + send` future. + let tx = self.tx.clone(); + let bytes = buf.to_vec(); + let len = bytes.len(); + self.inflight_len = len; + self.inflight = Some(Box::pin(async move { + let permit = tx.reserve().await?; + permit.send(StdinCmd::Bytes(bytes)); + Ok(()) + })); + // Recurse via a re-poll so the parked future is polled now. + self.poll_write(cx, buf) + } + Err(mpsc::error::TrySendError::Closed(_)) => Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "stdin channel closed", + ))), + } + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll> { + if self.close_sent { + return Poll::Ready(Ok(())); + } + match self.tx.try_send(StdinCmd::Eof) { + Ok(()) => { + self.close_sent = true; + Poll::Ready(Ok(())) + } + Err(mpsc::error::TrySendError::Full(_)) => Poll::Pending, + Err(mpsc::error::TrySendError::Closed(_)) => { + self.close_sent = true; + Poll::Ready(Ok(())) + } + } + } +} + +/// Allocate a local PTY, spawn `cmd` into it, and return the async-facing +/// `TtyHandle`. +/// +/// Spawns the child as a session leader with a controlling tty +/// (`CommandBuilder::set_controlling_tty(true)` — the default; REQ-TTY-02 +/// depends on it). Wires the three-thread bridge (reader/writer/waiter), +/// the `PtyControl`, and the ADR-056 `LocalExitFuture`. +pub fn allocate_pty( + terminal: TerminalParams, + cmd: Vec, + cwd: Option, + env: HashMap, +) -> Result { + if cmd.is_empty() { + return Err(TtyError::AllocFailed { + message: "cmd must be non-empty".to_string(), + }); + } + + let pty_system = native_pty_system(); + let size = PtySize { + cols: terminal.cols, + rows: terminal.rows, + pixel_width: terminal.pixel_width, + pixel_height: terminal.pixel_height, + }; + let pair = pty_system + .openpty(size) + .map_err(|e| TtyError::AllocFailed { + message: format!("openpty: {e}"), + })?; + + let mut builder = CommandBuilder::new(&cmd[0]); + for arg in &cmd[1..] { + builder.arg(arg); + } + if let Some(cwd) = cwd { + builder.cwd(cwd); + } + for (k, v) in env { + builder.env(k, v); + } + // Session leader + controlling tty — the default. REQ-TTY-02: + // `kill(-pgid, sig)` reaches the whole group only when the child is a + // session leader, which requires a controlling tty. + builder.set_controlling_tty(true); + + // Spawn the child on the slave side, then drop the slave so that when + // the master writer closes, the child sees EOF on its stdin. + let mut child = pair + .slave + .spawn_command(builder) + .map_err(|e| TtyError::AllocFailed { + message: format!("spawn_command: {e}"), + })?; + drop(pair.slave); + + let pid = child.process_id(); + // Two killer views: one for the ADR-056 kill guard (LocalExitFuture's + // Drop), one for the signal path's fallback kill (PtyControl). Both + // reference the same underlying pid/handle via clone_killer. + let killer = child.clone_killer(); + let killer_for_control = killer.clone_killer(); + + let master: Arc>> = Arc::new(Mutex::new(pair.master)); + + // --- Reader thread: blocking reads from the master reader → mpsc --- + let reader_master = master.clone(); + let (stdout_tx, stdout_rx) = mpsc::channel::(64); + thread::Builder::new() + .name("pty-reader".into()) + .spawn(move || { + let reader = { + let m = reader_master.lock().expect("master mutex poisoned"); + match m.try_clone_reader() { + Ok(r) => r, + Err(e) => { + warn!("pty-reader: try_clone_reader failed: {e}"); + let _ = stdout_tx.blocking_send(Bytes::new()); + return; + } + } + }; + let mut reader = reader; + let mut buf = vec![0u8; 8192]; + loop { + match reader.read(&mut buf) { + Ok(0) => break, + Ok(n) => { + let chunk = Bytes::copy_from_slice(&buf[..n]); + if stdout_tx.blocking_send(chunk).is_err() { + break; + } + } + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + Err(e) => { + warn!("pty-reader: read error: {e}"); + break; + } + } + } + // Zero-length sentinel: signals the pump the stream drained. + let _ = stdout_tx.blocking_send(Bytes::new()); + debug!("pty-reader thread done"); + }) + .expect("spawn pty-reader"); + + // --- Writer thread: drain mpsc → blocking writes --- + let writer_master = master.clone(); + let (stdin_tx, mut stdin_rx) = mpsc::channel::(64); + thread::Builder::new() + .name("pty-writer".into()) + .spawn(move || { + let writer = { + let m = writer_master.lock().expect("master mutex poisoned"); + match m.take_writer() { + Ok(w) => w, + Err(e) => { + warn!("pty-writer: take_writer failed: {e}"); + return; + } + } + }; + let mut writer = writer; + while let Some(cmd) = stdin_rx.blocking_recv() { + match cmd { + StdinCmd::Bytes(bytes) => { + if let Err(e) = writer.write_all(&bytes) { + warn!("pty-writer: write_all failed: {e}"); + break; + } + if let Err(e) = writer.flush() { + warn!("pty-writer: flush failed: {e}"); + break; + } + } + StdinCmd::Eof => { + drop(writer); + break; + } + } + } + debug!("pty-writer thread done"); + }) + .expect("spawn pty-writer"); + + // --- Waiter thread: blocking Child::wait() → oneshot --- + let (exit_tx, exit_rx) = oneshot::channel::(); + thread::Builder::new() + .name("pty-waiter".into()) + .spawn(move || { + let status = match child.wait() { + Ok(s) => s, + Err(e) => { + warn!("pty-waiter: wait failed: {e}"); + let _ = exit_tx.send(-1); + return; + } + }; + let code = status.exit_code() as i32; + debug!(exit_code = code, "pty-waiter: child reaped"); + let _ = exit_tx.send(code); + }) + .expect("spawn pty-waiter"); + + // --- Assemble the TtyHandle --- + let stdout: Pin + Send>> = + Box::pin(ReceiverStream::new(stdout_rx)); + let stdin: Box = Box::new(StdinSink::new(stdin_tx)); + let exit_code: BoxFuture> = + Box::pin(LocalExitFuture::new(exit_rx, killer)); + let control = Some(TtyControlHandle::new(Arc::new(PtyControl::new( + master, + Arc::new(Mutex::new(killer_for_control)), + pid, + )))); + + Ok(TtyHandle { + stdin, + stdout, + stderr: None, + exit_code, + control, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::AsyncWriteExt; + use tokio_stream::StreamExt; + + fn term() -> TerminalParams { + TerminalParams { + term: None, + cols: 80, + rows: 24, + pixel_width: 0, + pixel_height: 0, + modes: serde_json::Value::Null, + } + } + + fn env_default() -> HashMap { + let mut env = HashMap::new(); + env.insert("TERM".to_string(), "dumb".to_string()); + env + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn happy_path_echo_exits_zero() { + let handle = allocate_pty( + term(), + vec!["echo".to_string(), "hello".to_string()], + None, + env_default(), + ) + .expect("allocate"); + let mut stdout = handle.stdout; + let mut collected = Vec::new(); + while let Some(chunk) = stdout.next().await { + if chunk.is_empty() { + break; + } + collected.extend_from_slice(&chunk); + } + let code = handle.exit_code.await.expect("exit_code"); + assert_eq!(code, 0); + let s = String::from_utf8_lossy(&collected); + assert!(s.contains("hello"), "stdout should contain hello: {s:?}"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn stdin_round_trip_cat() { + let handle = + allocate_pty(term(), vec!["cat".to_string()], None, env_default()).expect("allocate"); + let mut stdin = handle.stdin; + let mut stdout = handle.stdout; + + // PTY may echo input; we still expect to see our bytes somewhere + // in the output. Write, then close, then drain. + stdin.write_all(b"ping\n").await.expect("write"); + stdin.shutdown().await.expect("shutdown (eof)"); + + let mut collected = Vec::new(); + while let Some(chunk) = stdout.next().await { + if chunk.is_empty() { + break; + } + collected.extend_from_slice(&chunk); + } + let code = handle.exit_code.await.expect("exit_code"); + let s = String::from_utf8_lossy(&collected); + assert!(s.contains("ping"), "stdout should contain ping: {s:?}"); + assert_eq!(code, 0); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn resize_does_not_error() { + let handle = allocate_pty( + term(), + vec!["sleep".to_string(), "1".to_string()], + None, + env_default(), + ) + .expect("allocate"); + let control = handle.control.as_ref().expect("control"); + control.resize(120, 40, 0, 0); + let _ = handle.exit_code.await.expect("exit_code"); + } + + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn signal_int_kills_child() { + let handle = allocate_pty( + term(), + vec!["sleep".to_string(), "60".to_string()], + None, + env_default(), + ) + .expect("allocate"); + let control = handle.control.as_ref().expect("control"); + // Give the child a moment to actually exec sleep. + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + control.signal("INT"); + let code = tokio::time::timeout(std::time::Duration::from_secs(5), handle.exit_code) + .await + .expect("exit timed out") + .expect("exit_code"); + assert_ne!( + code, 0, + "signal-terminated child should report non-zero exit: {code}" + ); + } + + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn signal_reaches_process_group_child() { + // bash -c "sleep 60" — sleep is a child of bash. The group signal + // must reach sleep too (REQ-TTY-02). bash exits when its child does. + let handle = allocate_pty( + term(), + vec!["bash".to_string(), "-c".to_string(), "sleep 60".to_string()], + None, + env_default(), + ) + .expect("allocate"); + let control = handle.control.as_ref().expect("control"); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + control.signal("INT"); + let code = tokio::time::timeout(std::time::Duration::from_secs(5), handle.exit_code) + .await + .expect("exit timed out") + .expect("exit_code"); + assert_ne!(code, 0, "process group should have been killed: {code}"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn cancel_cleanup_kills_child_on_drop() { + // ADR-056: dropping the TtyHandle (and thus the exit_code future) + // without awaiting it MUST kill the child. The waiter thread reaps + // the killed child (no zombie). We assert the kill happened by + // spawning a second session that completes promptly — i.e., the + // dropped session's child does not outlive a short grace period + // (if it did, the SIGHUP from Drop would not have fired). + let handle = allocate_pty( + term(), + vec!["sleep".to_string(), "60".to_string()], + None, + env_default(), + ) + .expect("allocate"); + drop(handle); + // Drop fires LocalExitFuture::Drop → ChildKiller::kill (SIGHUP). + // The waiter thread reaps the child. Give it a moment. + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + + // Probe: a new session should be allocatable and complete cleanly. + let probe = allocate_pty( + term(), + vec!["echo".to_string(), "ok".to_string()], + None, + env_default(), + ) + .expect("allocate"); + let code = tokio::time::timeout(std::time::Duration::from_secs(5), probe.exit_code) + .await + .expect("probe timed out") + .expect("probe exit_code"); + assert_eq!(code, 0); + } + + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn unknown_signal_falls_back_to_child_killer() { + let handle = allocate_pty( + term(), + vec!["sleep".to_string(), "60".to_string()], + None, + env_default(), + ) + .expect("allocate"); + let control = handle.control.as_ref().expect("control"); + // "NOSUCH" is not a known signal name → falls back to + // ChildKiller::kill (SIGHUP). The child should die. + control.signal("NOSUCH"); + let code = tokio::time::timeout(std::time::Duration::from_secs(5), handle.exit_code) + .await + .expect("exit timed out") + .expect("exit_code"); + assert_ne!(code, 0, "fallback kill should terminate the child: {code}"); + } +}