From e1610c2825e660c2960cbf65747708cc5beb5ff6 Mon Sep 17 00:00:00 2001 From: "glm-5.2" Date: Mon, 17 Aug 2026 10:19:22 +0000 Subject: [PATCH] phase 3: port alknet-tty-local behind the local feature MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Folds the alknet-tty-local crate into alktty as a feature-gated local submodule (the single-crate + local-feature design from ADR-054; the cyclic-dep workaround the mono-repo needed doesn't apply to one crate). src/local/mod.rs: - Module root, re-exports LocalTtyBackend, declares backend/pipe/pty - Doc-commented as feature-gated + non-wasm by design (portable-pty + tokio::process need a real OS; enabling local on a wasm target is a build error by design) src/local/backend.rs: - LocalTtyBackend: implements crate::backend::TtyBackend - allocate() dispatches on params.terminal: Some -> pty::allocate_pty, None -> pipe::allocate_pipe (ADR-054) - 7 tests: PTY/pipe dispatch, empty-cmd rejection (both modes), resource_id returns None, new() constructs src/local/pty.rs: - portable_pty + 3 std threads (reader/writer/waiter) feeding tokio mpsc/oneshot (REQ-TTY-01 blocking->async bridge) - PtyControl: resize + REQ-TTY-02 process-group signal forwarding (libc::kill(-pgid, sig) + kill(pid, sig) fallback + ChildKiller::kill for unknown names) - LocalExitFuture: ADR-056 kill-on-Drop guard (ChildKiller on cancel, Option::take disarms on resolve) - StdinSink: AsyncWrite over mpc::Sender with in-flight reserve+send parking for full-channel backpressure - 7 tests incl. process-group reach (bash -c sleep), cancel-cleanup, unknown-signal fallback src/local/pipe.rs: - tokio::process::Command + tokio_util::io::ReaderStream - PipeControl: no-op resize, libc::kill(pid, sig) with SIGKILL fallback for unknown names (pid-only, no process group - documented limitation of the runner case) - PipeExitFuture: in-place Child::wait() poll (avoids self-referential borrow) + ADR-056 kill-on-Drop guard (start_kill on cancel) - BytesStream: wraps ReaderStream, strips io::Error to EOF - 9 tests incl. separate stderr, SIGTERM=-15, SIGKILL=-9 fallback, cancel-cleanup pid probe (kill(pid,0) returns ESRCH) Import migration: alknet_tty::backend::{...} -> crate::backend::{...}, alknet_tty::control::signal_from_name -> crate::control::signal_from_name. Cargo.toml: no changes needed (portable-pty + tokio-util optional, libc under cfg(unix), and the local feature wiring tokio/process + tokio/rt-multi-thread were already in the scaffold per Phase 0). Also fixes pre-existing rustfmt drift in adapter.rs/channels.rs/ session.rs/lib.rs left by the phase 2 commit (cargo fmt --check without --features local reported 28 diffs; cargo fmt does not accept --features, so the earlier 'clean' check was a false negative — the check errored on the unknown flag and grepped an empty stdout). Lesson: run cargo fmt --check with no feature flags; cargo fmt doesn't gate on features. Verification: - cargo test --features local -> 103/103 pass (was 80 at phase 2 end; +23 new tests across the 3 local modules: 7 backend, 7 pty, 9 pipe) - cargo test (no features) -> 80/80 pass (local module not compiled) - cargo clippy --features local --all-targets -> clean - cargo clippy --all-targets (no features) -> clean - cargo check --target wasm32-unknown-unknown -> clean (default crate stays wasm-clean; local is feature-gated and non-wasm by design) - cargo fmt --check -> clean --- src/adapter.rs | 3 +- src/channels.rs | 176 +++++------ src/lib.rs | 2 +- src/local/backend.rs | 179 +++++++++++ src/local/mod.rs | 20 ++ src/local/pipe.rs | 469 ++++++++++++++++++++++++++++ src/local/pty.rs | 705 +++++++++++++++++++++++++++++++++++++++++++ src/session.rs | 96 +++--- 8 files changed, 1487 insertions(+), 163 deletions(-) create mode 100644 src/local/backend.rs create mode 100644 src/local/mod.rs create mode 100644 src/local/pipe.rs create mode 100644 src/local/pty.rs diff --git a/src/adapter.rs b/src/adapter.rs index fa9c5d1..c9fa651 100644 --- a/src/adapter.rs +++ b/src/adapter.rs @@ -76,8 +76,7 @@ use tracing::{debug, warn}; use crate::backend::{TtyBackend, TtyHandle}; use crate::control::ControlMessage; use crate::negotiation::{ - error_response_bytes, NegotiateRequest, NegotiationError, NegotiationReader, - NegotiationWriter, + error_response_bytes, NegotiateRequest, NegotiationError, NegotiationReader, NegotiationWriter, }; use crate::wire::{Chunk, ChunkReader, ChunkWriter, RawError, STREAM_CTRL_IN, STREAM_STDIN}; diff --git a/src/channels.rs b/src/channels.rs index 6e7a487..25062cf 100644 --- a/src/channels.rs +++ b/src/channels.rs @@ -182,33 +182,35 @@ fn make_tty_open_handler( ownership: Option>, identity: Option, ) -> OpenHandler { - Arc::new(move |input: Value, channel_conn: Connection, auth: AuthContext| { - let backends = Arc::clone(&backends); - let ownership = ownership.clone(); - let identity = identity.clone().or_else(|| auth.identity.clone()); - let _ = input; + Arc::new( + move |input: Value, channel_conn: Connection, auth: AuthContext| { + let backends = Arc::clone(&backends); + let ownership = ownership.clone(); + let identity = identity.clone().or_else(|| auth.identity.clone()); + let _ = input; - tokio::spawn(async move { - let stream = match channel_conn.accept_bi().await { - Ok(s) => s, - Err(e) => { - debug!("tty: channels open: accept_bi failed: {e}"); - return; - } - }; - let (client_read, client_write) = tokio::io::split(stream); - drive_session(client_write, client_read, backends, ownership, identity).await; - }) - }) + tokio::spawn(async move { + let stream = match channel_conn.accept_bi().await { + Ok(s) => s, + Err(e) => { + debug!("tty: channels open: accept_bi failed: {e}"); + return; + } + }; + let (client_read, client_write) = tokio::io::split(stream); + drive_session(client_write, client_read, backends, ownership, identity).await; + }) + }, + ) } #[cfg(test)] mod tests { use super::*; use crate::backend::{MockBackend, TtyError}; + use alkcall::channels::client::ChannelClient; use alkcall::channels::operations::ChannelCore; use alkcall::channels::policy::default_policy; - use alkcall::channels::client::ChannelClient; use alkcall::core::auth::Identity; use alkcall::core::types::Connection as CoreConnection; use alkcall::registry::registration::OperationRegistry; @@ -230,19 +232,14 @@ mod tests { ownership: Option>, identity: Option, ) -> ChannelClient { - use alkcall::channels::adapter::{ - ChannelsAdapter, InstallChannelZero, - }; + use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; use alkcall::core::auth::IdentityProvider; use alkcall::protocol::connection::split_single_stream; use alkcall::protocol::dispatch::Dispatcher; struct NoopIdProvider; impl IdentityProvider for NoopIdProvider { - fn resolve_from_fingerprint( - &self, - _: &str, - ) -> Option { + fn resolve_from_fingerprint(&self, _: &str) -> Option { None } fn resolve_from_token( @@ -257,65 +254,56 @@ mod tests { let policy_for_hook = Arc::clone(&policy); let identity_for_conn = identity.clone(); - let install_hook: InstallChannelZero = - Arc::new(move |manager, channel0_conn, auth| { - let backends = Arc::clone(&backends); - let ownership = ownership.clone(); - let _identity = identity.clone(); - let policy = Arc::clone(&policy_for_hook); - tokio::spawn(async move { - let channel0_bidi = match channel0_conn.accept_bi().await { - Ok(s) => s, - Err(_) => return, - }; - let (writer, reader) = split_single_stream(channel0_bidi); + let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, auth| { + let backends = Arc::clone(&backends); + let ownership = ownership.clone(); + let _identity = identity.clone(); + let policy = Arc::clone(&policy_for_hook); + tokio::spawn(async move { + let channel0_bidi = match channel0_conn.accept_bi().await { + Ok(s) => s, + Err(_) => return, + }; + let (writer, reader) = split_single_stream(channel0_bidi); - // Propagate the identity to the channel-0 - // connection so the call dispatch's - // `resolve_identity` sees it. The - // `ChannelsAdapter` builds `channel0_conn` fresh - // from `channel_source` and does NOT inherit the - // outer connection's identity — we set it here - // so the `AccessControl` scope-gate can check it. - if let Some(id) = _identity { - let _ = channel0_conn.set_identity(id); - } + // Propagate the identity to the channel-0 + // connection so the call dispatch's + // `resolve_identity` sees it. The + // `ChannelsAdapter` builds `channel0_conn` fresh + // from `channel_source` and does NOT inherit the + // outer connection's identity — we set it here + // so the `AccessControl` scope-gate can check it. + if let Some(id) = _identity { + let _ = channel0_conn.set_identity(id); + } - let core = ChannelCore::new(manager, policy); - let mut registry = OperationRegistry::new(); - register_openable( - &core, - Arc::clone(&backends), - ownership.clone(), - &mut registry, - auth.clone(), - ) - .expect("register_openable"); - let registry = Arc::new(registry); - let provider: Arc = Arc::new(NoopIdProvider); - let call_connection = Arc::new( - alkcall::protocol::connection::CallConnection::new_single_stream( - channel0_conn, - Arc::clone(&writer), - ), - ); - let dp = Dispatcher::new(registry, provider); - dp.run_loop_single_stream(call_connection, reader, writer) - .await; - }) - }); + let core = ChannelCore::new(manager, policy); + let mut registry = OperationRegistry::new(); + register_openable( + &core, + Arc::clone(&backends), + ownership.clone(), + &mut registry, + auth.clone(), + ) + .expect("register_openable"); + let registry = Arc::new(registry); + let provider: Arc = Arc::new(NoopIdProvider); + let call_connection = Arc::new( + alkcall::protocol::connection::CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + ), + ); + let dp = Dispatcher::new(registry, provider); + dp.run_loop_single_stream(call_connection, reader, writer) + .await; + }) + }); let (client_end, server_end) = duplex(64 * 1024); - let client_conn = CoreConnection::from_bidi( - client_end, - b"alk/channels".to_vec(), - None, - ); - let server_conn = CoreConnection::from_bidi( - server_end, - b"alk/channels".to_vec(), - None, - ); + let client_conn = CoreConnection::from_bidi(client_end, b"alk/channels".to_vec(), None); + let server_conn = CoreConnection::from_bidi(server_end, b"alk/channels".to_vec(), None); // Set the identity on the server connection so the call // dispatch sees it (the ACL check runs against the // connection's identity, not the `install_channel_zero` @@ -327,10 +315,8 @@ mod tests { let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap)); let auth = AuthContext::anonymous(b"alk/channels"); let _server_handle = tokio::spawn(async move { - let _ = alkcall::core::types::ProtocolHandler::handle( - &adapter, server_conn, &auth, - ) - .await; + let _ = + alkcall::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await; }); ChannelClient::from_connection(client_conn) @@ -376,8 +362,7 @@ mod tests { .get("required") .and_then(|v| v.as_array()) .expect("required array"); - let required_names: Vec<&str> = - required.iter().filter_map(|v| v.as_str()).collect(); + let required_names: Vec<&str> = required.iter().filter_map(|v| v.as_str()).collect(); assert!(required_names.contains(&"carriage")); assert!(required_names.contains(&"backend")); assert!(required_names.contains(&"cmd")); @@ -393,8 +378,7 @@ mod tests { let manager = alkcall::channels::manager::ChannelManager::with_defaults(handle, None); let core = ChannelCore::new(manager, default_policy()); - let backends: Arc>> = - Arc::new(HashMap::new()); + let backends: Arc>> = Arc::new(HashMap::new()); let mut registry = OperationRegistry::new(); register_openable( &core, @@ -415,10 +399,7 @@ mod tests { #[tokio::test] async fn end_to_end_open_op_returns_channel_id() { let mut backends: HashMap> = HashMap::new(); - backends.insert( - "mock".to_string(), - Arc::new(MockBackend::with_exit_code(0)), - ); + backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0))); let backends = Arc::new(backends); let identity = Identity { @@ -459,10 +440,7 @@ mod tests { #[tokio::test] async fn end_to_end_open_op_denies_without_tty_open_scope() { let mut backends: HashMap> = HashMap::new(); - backends.insert( - "mock".to_string(), - Arc::new(MockBackend::with_exit_code(0)), - ); + backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0))); let backends = Arc::new(backends); let identity = Identity { @@ -488,7 +466,9 @@ mod tests { let err = response.result.expect_err("open op should be denied"); assert!( - err.message.contains("scope") || err.code.contains("FORBIDDEN") || err.code.contains("AUTH"), + err.message.contains("scope") + || err.code.contains("FORBIDDEN") + || err.code.contains("AUTH"), "error should mention scope/forbidden/auth, got code={} message={}", err.code, err.message @@ -514,4 +494,4 @@ mod tests { assert_eq!(code, 7); let _: Result = Ok(code); } -} \ No newline at end of file +} diff --git a/src/lib.rs b/src/lib.rs index ea89d37..ff33ca8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,4 +55,4 @@ pub mod session; pub mod wire; #[cfg(feature = "local")] -pub mod local; \ No newline at end of file +pub mod local; diff --git a/src/local/backend.rs b/src/local/backend.rs new file mode 100644 index 0000000..1b3e285 --- /dev/null +++ b/src/local/backend.rs @@ -0,0 +1,179 @@ +//! `LocalTtyBackend`: the [`crate::backend::TtyBackend`] implementation +//! for local processes. +//! +//! `allocate()` branches on `TtyParams::terminal`: `Some(TerminalParams)` +//! selects PTY mode (this crate's `pty` module); `None` selects pipe mode +//! (the runner case, this crate's `pipe` module). One backend serves both +//! (ADR-054). + +use crate::backend::{TtyBackend, TtyError, TtyHandle, TtyParams}; +use async_trait::async_trait; + +/// Local TTY backend: implements [`crate::backend::TtyBackend`] for local +/// processes. +/// +/// `allocate()` branches on `TtyParams::terminal`: `Some(TerminalParams)` +/// selects PTY mode (this crate's `pty` module); `None` selects pipe mode +/// (the runner case, this crate's `pipe` module). One backend serves both +/// (ADR-054). +/// +/// Takes no constructor dependencies — unlike `DockerTtyBackend` (wraps a +/// `bollard::Docker` client) or `SshTtyBackend` (wraps an SSH session), the +/// `portable_pty` system is process-global. The assembly layer constructs one +/// `LocalTtyBackend` and registers it as `"local"`. +pub struct LocalTtyBackend; + +impl LocalTtyBackend { + /// Construct a new `LocalTtyBackend`. The backend is dependency-free; + /// `portable_pty`'s native system is process-global. + pub fn new() -> Self { + Self + } +} + +impl Default for LocalTtyBackend { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl TtyBackend for LocalTtyBackend { + /// Allocate a terminal/process session, branching on + /// `params.terminal`. + /// + /// `Some(TerminalParams)` dispatches to [`crate::local::pty::allocate_pty`] + /// (PTY mode, real terminal semantics — resize, process-group signal + /// forwarding, merged stdout/stderr). `None` dispatches to + /// [`crate::local::pipe::allocate_pipe`] (pipe mode, the runner case — + /// separate stdout/stderr, no-op resize, pid-only signal). + /// + /// `backend_params` is ignored: the local backend has no + /// backend-specific selector fields. An empty `cmd` returns + /// [`TtyError::AllocFailed`] (the adapter already checks this at + /// negotiation, but the backend fails gracefully if called directly). + async fn allocate(&self, params: &TtyParams) -> Result { + if params.cmd.is_empty() { + return Err(TtyError::AllocFailed { + message: "cmd must be non-empty".to_string(), + }); + } + match ¶ms.terminal { + Some(terminal) => crate::local::pty::allocate_pty( + terminal.clone(), + params.cmd.clone(), + params.cwd.clone(), + params.env.clone(), + ), + None => crate::local::pipe::allocate_pipe( + params.cmd.clone(), + params.cwd.clone(), + params.env.clone(), + ), + } + } + + /// The local backend creates its own resource (a process), so there is + /// no pre-existing resource for the ownership check. Returns `None`. + fn resource_id(&self, _params: &TtyParams) -> Option<(&'static str, String)> { + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + fn term() -> crate::backend::TerminalParams { + crate::backend::TerminalParams { + term: None, + cols: 80, + rows: 24, + pixel_width: 0, + pixel_height: 0, + modes: serde_json::Value::Null, + } + } + + fn params(terminal: Option, cmd: Vec) -> TtyParams { + TtyParams { + terminal, + cmd, + cwd: None, + env: HashMap::new(), + backend_params: serde_json::Map::new(), + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn pty_dispatch_yields_no_stderr() { + let backend = LocalTtyBackend::new(); + let handle = backend + .allocate(¶ms( + Some(term()), + vec!["echo".to_string(), "hi".to_string()], + )) + .await + .expect("allocate"); + assert!(handle.stderr.is_none(), "PTY backends merge stdout/stderr"); + assert!(handle.control.is_some(), "PTY backends provide control"); + let _ = handle.exit_code.await; + } + + #[tokio::test] + async fn pipe_dispatch_yields_some_stderr() { + let backend = LocalTtyBackend::new(); + let handle = backend + .allocate(¶ms(None, vec!["echo".to_string(), "hi".to_string()])) + .await + .expect("allocate"); + assert!( + handle.stderr.is_some(), + "pipe backends have separate stderr" + ); + assert!(handle.control.is_some(), "pipe backends provide control"); + let _ = handle.exit_code.await; + } + + #[tokio::test] + async fn empty_cmd_returns_alloc_failed() { + let backend = LocalTtyBackend::new(); + let result = backend.allocate(¶ms(None, vec![])).await; + assert!( + matches!(result, Err(TtyError::AllocFailed { .. })), + "expected AllocFailed for empty cmd" + ); + } + + #[tokio::test] + async fn empty_cmd_with_terminal_returns_alloc_failed() { + let backend = LocalTtyBackend::new(); + let result = backend.allocate(¶ms(Some(term()), vec![])).await; + assert!( + matches!(result, Err(TtyError::AllocFailed { .. })), + "expected AllocFailed for empty cmd even with terminal set" + ); + } + + #[test] + fn resource_id_returns_none() { + let backend = LocalTtyBackend::new(); + let params = params(None, vec!["true".to_string()]); + assert!(backend.resource_id(¶ms).is_none()); + } + + #[test] + fn resource_id_returns_none_with_terminal() { + let backend = LocalTtyBackend::new(); + let params = params(Some(term()), vec!["true".to_string()]); + assert!(backend.resource_id(¶ms).is_none()); + } + + #[test] + fn new_constructs() { + let backend = LocalTtyBackend::new(); + let params = params(None, vec!["true".to_string()]); + assert!(backend.resource_id(¶ms).is_none()); + } +} diff --git a/src/local/mod.rs b/src/local/mod.rs new file mode 100644 index 0000000..40670c5 --- /dev/null +++ b/src/local/mod.rs @@ -0,0 +1,20 @@ +//! Local TTY backend for alktty: `LocalTtyBackend` implements +//! [`crate::backend::TtyBackend`] via `portable_pty` (PTY mode, terminal +//! semantics) and `tokio::process::Command` (pipe mode, the runner case). +//! +//! The blocking→async bridge for PTY mode uses three dedicated std +//! threads feeding tokio mpsc/oneshot channels (REQ-TTY-01). Signal +//! forwarding targets the foreground process group (REQ-TTY-02). +//! +//! This module is feature-gated behind the `local` feature and is +//! non-wasm by design (`portable_pty` + `tokio::process` need a real +//! OS). The default crate (no features) compiles to +//! `wasm32-unknown-unknown`; enabling `local` on a wasm target is a +//! build error by design — the local-process backend runs on a real +//! OS (Linux, macOS, Windows), never in a sandbox. + +pub mod backend; +pub mod pipe; +pub mod pty; + +pub use backend::LocalTtyBackend; diff --git a/src/local/pipe.rs b/src/local/pipe.rs new file mode 100644 index 0000000..d26f786 --- /dev/null +++ b/src/local/pipe.rs @@ -0,0 +1,469 @@ +//! Pipe mode for `LocalTtyBackend`: `tokio::process`-backed runner sessions. +//! +//! Spawns the command with `Stdio::piped()` for stdin/stdout/stderr, exposes +//! tokio-native `AsyncRead`/`AsyncWrite` (no std-thread bridge needed), +//! `PipeControl` (no-op resize, `libc::kill(pid, sig)` signal), and the +//! ADR-056 kill guard on the `exit_code` future. + +use std::collections::HashMap; +use std::future::Future; +use std::path::PathBuf; +use std::pin::Pin; +use std::process::ExitStatus; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use crate::backend::{BoxFuture, TtyControl, TtyControlHandle, TtyError, TtyHandle}; +use crate::control::signal_from_name; +use bytes::Bytes; +use futures_core::Stream; +use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command}; +use tokio_util::io::ReaderStream; + +/// Allocate a pipe-mode session: spawn `cmd` with `Stdio::piped()` for +/// stdin/stdout/stderr and return a `TtyHandle` whose `stderr` is `Some` +/// (separate streams). The `exit_code` future carries the ADR-056 kill +/// guard (see [`PipeExitFuture`]). `control` is a [`PipeControl`] +/// (no-op resize, `libc::kill(pid, sig)` signal). +pub fn allocate_pipe( + cmd: Vec, + cwd: Option, + env: HashMap, +) -> Result { + if cmd.is_empty() { + return Err(TtyError::AllocFailed { + message: "empty command vector".to_string(), + }); + } + + let mut command = Command::new(&cmd[0]); + command.args(&cmd[1..]); + command.stdin(std::process::Stdio::piped()); + command.stdout(std::process::Stdio::piped()); + command.stderr(std::process::Stdio::piped()); + command.kill_on_drop(true); + if let Some(cwd) = cwd { + command.current_dir(cwd); + } + for (k, v) in env { + command.env(k, v); + } + + let mut child = command.spawn().map_err(|e| TtyError::AllocFailed { + message: format!("spawn failed: {e}"), + })?; + + let pid = child.id(); + + let stdin: Box = child + .stdin + .take() + .map(|s: ChildStdin| Box::new(s) as Box) + .ok_or_else(|| TtyError::AllocFailed { + message: "child stdin not piped".to_string(), + })?; + + let stdout: ChildStdout = child.stdout.take().ok_or_else(|| TtyError::AllocFailed { + message: "child stdout not piped".to_string(), + })?; + let stderr: ChildStderr = child.stderr.take().ok_or_else(|| TtyError::AllocFailed { + message: "child stderr not piped".to_string(), + })?; + + let stdout: Pin + Send>> = + Box::pin(BytesStream::wrap(ReaderStream::new(stdout))); + let stderr: Pin + Send>> = + Box::pin(BytesStream::wrap(ReaderStream::new(stderr))); + + let control = Some(TtyControlHandle::new(Arc::new(PipeControl::new(pid)))); + + let exit_code: BoxFuture> = Box::pin(PipeExitFuture::new(child)); + + Ok(TtyHandle { + stdin, + stdout, + stderr: Some(stderr), + exit_code, + control, + }) +} + +/// Adapter wrapping `tokio_util::io::ReaderStream` (which yields +/// `Result`) into a `Stream`, dropping the +/// `io::Error` info by ending the stream on read error. The adapter pump +/// treats a stream end as EOF; a read error is indistinguishable from EOF +/// at the wire level (the `exit_code` future surfaces the failure if the +/// child died abnormally). +struct BytesStream { + inner: S, +} + +impl BytesStream { + fn wrap(inner: S) -> Self { + Self { inner } + } +} + +impl Stream for BytesStream +where + S: Stream> + Unpin, +{ + type Item = Bytes; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match Pin::new(&mut self.inner).poll_next(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(None) => Poll::Ready(None), + Poll::Ready(Some(Ok(bytes))) => Poll::Ready(Some(bytes)), + // Read error: end the stream (EOF to the adapter). + Poll::Ready(Some(Err(_))) => Poll::Ready(None), + } + } +} + +/// Control handle for pipe mode: no-op resize and `libc::kill(pid, sig)` +/// signal forwarding (the runner case — ADR-054). +/// +/// # Signal scope — no process group (documented limitation) +/// +/// Pipe mode has no session leader / controlling tty, so `signal()` targets +/// the direct child pid only via `libc::kill(pid, sig)`. If the child has +/// spawned its own children, they will NOT receive the signal — this is a +/// known limitation of the runner case. A runner that needs process-group +/// signal delivery uses the PTY case, not the pipe case. See +/// `tty-local.md` §"Pipe Mode". +pub struct PipeControl { + pid: Option, +} + +impl PipeControl { + fn new(pid: Option) -> Self { + Self { pid } + } + + #[cfg(test)] + pub(crate) fn pid(&self) -> Option { + self.pid + } +} + +impl TtyControl for PipeControl { + fn resize(&self, _cols: u16, _rows: u16, _pixel_width: u16, _pixel_height: u16) { + // No-op — no PTY in pipe mode. + } + + fn signal(&self, name: &str) { + // Targets the direct child pid only (no process group — see the + // type-level doc comment on `PipeControl`). + #[cfg(unix)] + { + if let Some(pid) = self.pid { + let pid = pid as libc::pid_t; + match signal_from_name(name) { + Some(sig) => { + // SAFETY: libc::kill on a known pid with a valid + // signal number is the documented signal-forwarding + // path. Best-effort: ignore errors (child may be gone). + let _ = unsafe { libc::kill(pid, sig) }; + } + None => { + // Unknown name → SIGKILL. On Unix, `libc::kill(pid, + // SIGKILL)` is exactly what + // `tokio::process::Child::start_kill()` does + // internally; the `Child` handle lives in the + // `PipeExitFuture` and isn't accessible here, so we + // kill by pid instead. + let _ = unsafe { libc::kill(pid, libc::SIGKILL) }; + } + } + } else { + tracing::warn!(name = name, "PipeControl::signal: no pid; cannot signal"); + } + } + #[cfg(not(unix))] + { + // Non-Unix: no `libc::kill` available, and the `Child` handle + // lives in the `PipeExitFuture` (not accessible from here). + // Dropping the `TtyHandle` triggers the ADR-056 kill guard. + let _ = name; + tracing::warn!( + name = name, + "PipeControl::signal: non-Unix; drop TtyHandle to kill" + ); + } + } +} + +/// The `exit_code` future for pipe mode: wraps `tokio::process::Child::wait` +/// plus the ADR-056 kill guard. +/// +/// # ADR-056 — kill-on-Drop contract +/// +/// On cancel (the future is dropped without being driven to completion), +/// `Drop` calls `Child::start_kill()` (SIGKILL); the child exits and the +/// inner `wait` future reaps it. On the happy path, `poll` resolves the +/// `wait` future, then **disarms** the guard by taking the `Child` out of +/// `self.child` — so the subsequent `Drop` is a no-op. The guard is the +/// spec-compliant mechanism; `kill_on_drop(true)` on the `Command` is a +/// defense-in-depth backstop, not the primary path (the `Child` is moved +/// into the future and the future's `Drop` is the cancel path, not the +/// `Child`'s). +/// +/// # Implementation note +/// +/// `tokio::process::Child::wait()` borrows `&mut self`, so the `wait` future +/// cannot be held alongside the `Child` in the same struct without a +/// self-referential borrow. Instead, the `Child` lives in an `Option` +/// and is polled in place: on each `poll`, we `as_mut()` the `Option`, +/// construct a fresh `Child::wait()` borrow-future, pin it locally, and drive +/// it. The borrow ends when the local future is dropped at the end of the +/// `poll` call, so the `Child` remains free to be `take`n (disarmed or +/// killed) outside the borrow. This is the same pattern the standard +/// library's `tokio::process` examples use for "wait with a kill guard." +struct PipeExitFuture { + child: Option, +} + +impl PipeExitFuture { + fn new(child: Child) -> Self { + Self { child: Some(child) } + } +} + +impl Future for PipeExitFuture { + type Output = Result; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + // Poll the wait future in an inner scope so the borrow of + // `self.child` ends before we `take()` the child to disarm/kill. + enum Step { + Pending, + Exited(ExitStatus), + Failed(String), + Taken, + } + let step = match self.child.as_mut() { + None => Step::Taken, + Some(child) => { + let wait = child.wait(); + tokio::pin!(wait); + match Future::poll(Pin::new(&mut wait), cx) { + Poll::Pending => Step::Pending, + Poll::Ready(Ok(status)) => Step::Exited(status), + Poll::Ready(Err(e)) => Step::Failed(format!("wait failed: {e}")), + } + } + }; + // The borrow of `self.child` ended with `step`; now we can `take()`. + match step { + Step::Pending => Poll::Pending, + Step::Exited(status) => { + self.child.take(); + Poll::Ready(Ok(exit_code_from(status))) + } + Step::Failed(msg) => { + self.child.take(); + Poll::Ready(Err(TtyError::WaitFailed { message: msg })) + } + Step::Taken => Poll::Ready(Err(TtyError::WaitFailed { + message: "child already taken".to_string(), + })), + } + } +} + +impl Drop for PipeExitFuture { + fn drop(&mut self) { + if let Some(mut child) = self.child.take() { + // Best-effort kill; the child may already be exiting. + let _ = child.start_kill(); + } + } +} + +/// Map an `ExitStatus` to the adapter's `i32` exit code: `code()` on normal +/// exit; on Unix, the negative signal number on signal-terminated exit +/// (matches `std` convention and ADR-055 §4). Falls back to `-1` (the +/// adapter's "could not determine" sentinel — ADR-055 §4). +fn exit_code_from(status: ExitStatus) -> i32 { + if let Some(code) = status.code() { + return code; + } + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + if let Some(sig) = status.signal() { + return -sig; + } + } + -1 +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::AsyncWriteExt; + use tokio_stream::StreamExt; + + fn sh_cmd(argv: &[&str]) -> Vec { + argv.iter().map(|s| s.to_string()).collect() + } + + /// Drain a `Stream` into a `Vec`. + async fn drain(mut s: Pin + Send>>) -> Vec { + let mut out = Vec::new(); + while let Some(chunk) = s.next().await { + out.extend_from_slice(&chunk); + } + out + } + + /// Swap a `Pin + Send>>` out for an empty + /// stream, returning the original. (`TtyHandle.stdout` is not `Option`, + /// so `.take()` isn't available; this is the equivalent.) + fn swap_stdout( + slot: &mut Pin + Send>>, + ) -> Pin + Send>> { + std::mem::replace(slot, Box::pin(tokio_stream::empty())) + } + + #[tokio::test] + async fn happy_path_echo() { + let mut handle = + allocate_pipe(sh_cmd(&["echo", "hello"]), None, HashMap::new()).expect("allocate"); + let stderr = handle.stderr.take().expect("stderr present"); + let stdout = swap_stdout(&mut handle.stdout); + let out = drain(stdout).await; + let err = drain(stderr).await; + assert!(err.is_empty(), "stderr empty"); + let code = handle.exit_code.await.expect("exit"); + assert_eq!(code, 0); + assert_eq!(String::from_utf8_lossy(&out), "hello\n"); + } + + #[tokio::test] + async fn stdin_round_trip_cat() { + let mut handle = allocate_pipe(sh_cmd(&["cat"]), None, HashMap::new()).expect("allocate"); + handle + .stdin + .write_all(b"round-trip\n") + .await + .expect("write stdin"); + // Close stdin so `cat` sees EOF and exits. Replace with a sink + // (`TtyHandle.stdin` is not `Option`; this drops the child stdin). + handle.stdin = Box::new(tokio::io::sink()); + let stdout = swap_stdout(&mut handle.stdout); + let out = drain(stdout).await; + let code = handle.exit_code.await.expect("exit"); + assert_eq!(code, 0); + assert_eq!(String::from_utf8_lossy(&out), "round-trip\n"); + } + + #[tokio::test] + async fn separate_stderr() { + let mut handle = allocate_pipe(sh_cmd(&["sh", "-c", "echo err >&2"]), None, HashMap::new()) + .expect("allocate"); + let stderr = handle.stderr.take().expect("stderr present"); + let stdout = swap_stdout(&mut handle.stdout); + let out = drain(stdout).await; + assert!(out.is_empty(), "stdout empty"); + let err = drain(stderr).await; + let code = handle.exit_code.await.expect("exit"); + assert_eq!(code, 0); + assert_eq!(String::from_utf8_lossy(&err), "err\n"); + } + + #[cfg(unix)] + #[tokio::test] + async fn signal_term_kills_child() { + let handle = + allocate_pipe(sh_cmd(&["sleep", "60"]), None, HashMap::new()).expect("allocate"); + let control = handle.control.clone().expect("control present"); + control.signal("TERM"); + let code = handle.exit_code.await.expect("exit"); + assert_eq!(code, -15, "SIGTERM = -15"); + } + + #[cfg(unix)] + #[tokio::test] + async fn cancel_cleanup_drops_kill_child() { + // Use a temp file to capture the child's pid, then drop the handle + // (which drops exit_code → the ADR-056 guard sends SIGKILL) and + // assert the process no longer exists via `kill(pid, 0)`. + let pid_file = std::env::temp_dir().join(format!( + "alktty_pipe_cancel_pid_{}_{}.txt", + std::process::id(), + rand_seed() + )); + let cmd = sh_cmd(&[ + "sh", + "-c", + &format!("echo $$ > '{}'; exec sleep 60", pid_file.display()), + ]); + let handle = allocate_pipe(cmd, None, HashMap::new()).expect("allocate"); + // Wait for the shell to write its pid. + for _ in 0..100 { + if pid_file.exists() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + let pid_str = std::fs::read_to_string(&pid_file).expect("pid file written"); + let pid: i32 = pid_str.trim().parse().expect("pid parses"); + let _ = std::fs::remove_file(&pid_file); + // Drop the handle without awaiting exit_code — the ADR-056 guard + // must kill the child. + drop(handle); + // Give the kernel a moment to deliver SIGKILL and reap. + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + // kill(pid, 0) returns ESRCH (no such process) when the child is gone. + let alive = unsafe { libc::kill(pid, 0) } == 0; + assert!(!alive, "child (pid={pid}) should be killed after drop"); + } + + #[tokio::test] + async fn resize_is_noop() { + let handle = allocate_pipe(sh_cmd(&["true"]), None, HashMap::new()).expect("allocate"); + let control = handle.control.as_ref().expect("control present"); + control.resize(120, 40, 800, 600); + control.resize(0, 0, 0, 0); + } + + #[cfg(unix)] + #[tokio::test] + async fn unknown_signal_falls_back_to_sigkill() { + let handle = + allocate_pipe(sh_cmd(&["sleep", "60"]), None, HashMap::new()).expect("allocate"); + let control = handle.control.clone().expect("control present"); + // Unknown name → SIGKILL fallback (equivalent to start_kill()). + control.signal("NOSUCH"); + let code = handle.exit_code.await.expect("exit"); + assert_eq!(code, -9, "SIGKILL = -9"); + } + + #[cfg(unix)] + #[tokio::test] + async fn pipe_control_pid_recorded() { + let ctrl = PipeControl::new(Some(42)); + assert_eq!(ctrl.pid(), Some(42)); + let ctrl_none = PipeControl::new(None); + assert_eq!(ctrl_none.pid(), None); + } + + #[test] + fn empty_command_returns_alloc_failed() { + let result = allocate_pipe(vec![], None, HashMap::new()); + assert!( + matches!(result, Err(TtyError::AllocFailed { .. })), + "expected AllocFailed" + ); + } + + fn rand_seed() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0) + } +} diff --git a/src/local/pty.rs b/src/local/pty.rs new file mode 100644 index 0000000..365820f --- /dev/null +++ b/src/local/pty.rs @@ -0,0 +1,705 @@ +//! PTY mode for `LocalTtyBackend`: `portable_pty`-backed terminal sessions. +//! +//! Implements the blocking→async bridge (REQ-TTY-01) via three dedicated +//! std threads (reader, writer, waiter) feeding tokio mpsc/oneshot channels, +//! `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)". + +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 crate::backend::{ + BoxFuture, TerminalParams, TtyControl, TtyControlHandle, TtyError, TtyHandle, +}; +use crate::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 + /// `crate::control::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}"); + } +} diff --git a/src/session.rs b/src/session.rs index 501e4f2..e361638 100644 --- a/src/session.rs +++ b/src/session.rs @@ -87,7 +87,10 @@ pub enum TtySessionError { /// the stream is a length-prefixed JSON `{"error":"..."}` rather /// than a raw chunk). #[error("negotiation rejected: {error}")] - NegotiationRejected { error: String, fields: HashMap }, + NegotiationRejected { + error: String, + fields: HashMap, + }, /// The session ended (server closed the stream) before an `Exit` /// control chunk arrived. `wait()` returns this when the /// stdout/stderr pumps drain and no exit chunk was observed. @@ -174,17 +177,19 @@ impl TtySession { params: serde_json::Value, ) -> Result { let (channel_id, send, recv) = client - .open_channel(crate::channels::OP_TTY_OPEN, params.clone(), crate::channels::TTY_ALPN) + .open_channel( + crate::channels::OP_TTY_OPEN, + params.clone(), + crate::channels::TTY_ALPN, + ) .await .map_err(TtySessionError::ChannelsOpen)?; debug!("tty: opened channel {channel_id} via channels"); let remote_addr = client.manager().remote_addr(); let source = alkcall::channels::source::channel_source(recv, send, remote_addr); - let channel_conn = Connection::from_source( - source, - crate::channels::TTY_ALPN.as_bytes().to_vec(), - ); + let channel_conn = + Connection::from_source(source, crate::channels::TTY_ALPN.as_bytes().to_vec()); let negotiate: NegotiateRequest = serde_json::from_value(params).map_err(TtySessionError::NegotiationSerialize)?; @@ -237,12 +242,7 @@ impl TtySession { let (exit_tx, exit_rx) = tokio::sync::watch::channel::>>(None); - let read_pump = tokio::spawn(read_pump( - read, - stdout_tx, - stderr_tx, - exit_tx, - )); + let read_pump = tokio::spawn(read_pump(read, stdout_tx, stderr_tx, exit_tx)); Ok(Self { writer: Mutex::new(writer), @@ -327,10 +327,9 @@ impl TtySession { pub async fn recv_stdout(&self) -> Pin + Send>> { let mut guard = self.stdout_rx.lock().await; if let Some(rx) = guard.take() { - return Box::pin(futures::stream::unfold( - rx, - |mut rx| async move { rx.recv().await.map(|bytes| (bytes, rx)) }, - )); + return Box::pin(futures::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|bytes| (bytes, rx)) + })); } // Already taken — return an empty stream. Box::pin(futures::stream::empty()) @@ -343,10 +342,9 @@ impl TtySession { pub async fn recv_stderr(&self) -> Option + Send>>> { let mut guard = self.stderr_rx.lock().await; let rx = guard.take()?; - Some(Box::pin(futures::stream::unfold( - rx, - |mut rx| async move { rx.recv().await.map(|bytes| (bytes, rx)) }, - ))) + Some(Box::pin(futures::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|bytes| (bytes, rx)) + }))) } /// Await the `Exit` control chunk and return the process exit @@ -516,18 +514,11 @@ mod tests { let (client, server) = duplex(64 * 1024); let (server_read, server_write) = tokio::io::split(server); let server_task = tokio::spawn(async move { - crate::adapter::drive_session( - server_write, - server_read, - backends, - None, - identity, - ) - .await; + crate::adapter::drive_session(server_write, server_read, backends, None, identity) + .await; }); - let client_conn = - Connection::from_bidi(client, b"alk/tty".to_vec(), None); + let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None); let session = TtySession::connect_direct(client_conn, test_negotiate("mock")) .await .expect("connect_direct"); @@ -547,13 +538,10 @@ mod tests { resources: HashMap::new(), }); let (session, _server) = wire_session_and_server(backend, identity).await; - let code = tokio::time::timeout( - std::time::Duration::from_secs(5), - session.wait(), - ) - .await - .expect("wait didn't time out") - .expect("wait returns exit code"); + let code = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()) + .await + .expect("wait didn't time out") + .expect("wait returns exit code"); assert_eq!(code, 0); } @@ -574,15 +562,8 @@ mod tests { .send_stdin(Bytes::from_static(b"hello")) .await .expect("send_stdin"); - session - .close_stdin() - .await - .expect("close_stdin"); - let _ = tokio::time::timeout( - std::time::Duration::from_secs(5), - session.wait(), - ) - .await; + session.close_stdin().await.expect("close_stdin"); + let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()).await; } #[tokio::test] @@ -600,11 +581,7 @@ mod tests { let (session, _server) = wire_session_and_server(backend, identity).await; session.resize(80, 24, 0, 0).await.expect("resize"); session.signal("INT").await.expect("signal"); - let _ = tokio::time::timeout( - std::time::Duration::from_secs(5), - session.wait(), - ) - .await; + let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()).await; } #[tokio::test] @@ -650,17 +627,13 @@ mod tests { let _ = server.read_exact(&mut body).await; // Drop `server` — the client's read pump hits EOF. }); - let client_conn = - Connection::from_bidi(client, b"alk/tty".to_vec(), None); + let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None); let session = TtySession::connect_direct(client_conn, test_negotiate("mock")) .await .expect("connect_direct"); - let result = tokio::time::timeout( - std::time::Duration::from_secs(5), - session.wait(), - ) - .await - .expect("wait didn't time out"); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()) + .await + .expect("wait didn't time out"); assert!(matches!(result, Err(TtySessionError::NoExitChunk))); let _ = server_handle.await; } @@ -672,12 +645,11 @@ mod tests { async fn connect_direct_errors_when_stream_is_broken() { let (client, server) = duplex(64); drop(server); - let client_conn = - Connection::from_bidi(client, b"alk/tty".to_vec(), None); + let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None); let result = TtySession::connect_direct(client_conn, test_negotiate("mock")).await; assert!( result.is_err(), "construction should fail when the negotiation write hits a broken pipe" ); } -} \ No newline at end of file +}