From c76b6aab7a9084d3a4a989a0952b938da1e9a45e Mon Sep 17 00:00:00 2001 From: "glm-5.2" Date: Tue, 7 Jul 2026 15:04:30 +0000 Subject: [PATCH] =?UTF-8?q?feat(tty-local):=20implement=20pipe=20mode=20?= =?UTF-8?q?=E2=80=94=20tokio::process=20spawn,=20PipeControl,=20ADR-056=20?= =?UTF-8?q?kill=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the runner case (terminal: None, no PTY) in pipe.rs: - allocate_pipe spawns via tokio::process::Command with Stdio::piped(), returning a TtyHandle with separate stdout/stderr (stderr=Some). - stdin: ChildStdin boxed as Box. - stdout/stderr: ChildStdout/ChildStderr wrapped via tokio_util::io::ReaderStream into Pin + Send>> (errors map to stream EOF). - exit_code: PipeExitFuture holding the Child; poll drives Child::wait() and disarms the guard on resolve; Drop calls Child::start_kill() on cancel (ADR-056 spec-compliant kill-on-Drop, not kill_on_drop(true) alone). - control: PipeControl (no-op resize, libc::kill(pid, sig) signal on Unix, SIGKILL fallback for unknown names; non-Unix logs and relies on Drop guard). Doc comment documents the no-process-group limitation of pipe mode. Adds tokio-util (io) and tokio-stream (dev) deps. 9 unit tests cover the happy path, stdin round-trip, separate stderr, SIGTERM signal, cancel cleanup (asserts child killed via kill(pid,0)), resize no-op, unknown signal SIGKILL fallback, pid recording, and empty-command error. --- Cargo.lock | 2 + crates/alknet-tty-local/Cargo.toml | 4 + crates/alknet-tty-local/src/pipe.rs | 463 +++++++++++++++++++++++++++- 3 files changed, 468 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index c0630d2..9842a8b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -158,6 +158,8 @@ dependencies = [ "portable-pty", "thiserror 2.0.18", "tokio", + "tokio-stream", + "tokio-util", "tracing", ] diff --git a/crates/alknet-tty-local/Cargo.toml b/crates/alknet-tty-local/Cargo.toml index 8e4ffb4..6d1ba9a 100644 --- a/crates/alknet-tty-local/Cargo.toml +++ b/crates/alknet-tty-local/Cargo.toml @@ -13,10 +13,14 @@ name = "alknet_tty_local" alknet-tty = { path = "../alknet-tty" } portable-pty = "0.8" tokio = { version = "1", features = ["full"] } +tokio-util = { version = "0.7", features = ["io"] } bytes = "1" futures-core = "0.3" tracing = "0.1" thiserror = "2" +[dev-dependencies] +tokio-stream = "0.1" + [target.'cfg(unix)'.dependencies] libc = "0.2" \ No newline at end of file diff --git a/crates/alknet-tty-local/src/pipe.rs b/crates/alknet-tty-local/src/pipe.rs index 36c82e7..67c7d71 100644 --- a/crates/alknet-tty-local/src/pipe.rs +++ b/crates/alknet-tty-local/src/pipe.rs @@ -5,4 +5,465 @@ //! `PipeControl` (no-op resize, `libc::kill(pid, sig)` signal), and the //! ADR-056 kill guard on the `exit_code` future. -// TODO: implement +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 alknet_tty::backend::{BoxFuture, TtyControl, TtyControlHandle, TtyError, TtyHandle}; +use alknet_tty::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!( + "alknet_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) + } +}