From 9c9ce8547e5a88cbe834a53e9a6d83a789d08e47 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Tue, 3 Nov 2026 00:52:41 +0100 Subject: [PATCH] feat(env): run programs with plain pipes through the daemon The daemon's process operation runs a long-lived program with piped stdin, stdout and stderr, for programs that speak a protocol on stdin and stdout such as MCP servers. processWrite and processCloseStdin feed stdin in arrival order; output arrives as raw bytes per stream; cancelling or a lost connection kills the process group. hello reports it in features, and openProcess() is the client. On Windows, readers and writers of child pipes now run their own overlapped operations and look at stop every 100 ms. The standard library opens these pipes for overlapped I/O, so CancelSynchronousIo never reached a blocked read: exec's readers stayed blocked while a descendant held a pipe open. Spawning, reaping, argv parsing and target lookup are shared by exec, pty and process. --- packages/env/CHANGELOG.md | 8 + packages/env/README.md | 16 ++ packages/env/daemon/src/exec.rs | 147 +++++++---- packages/env/daemon/src/main.rs | 37 ++- packages/env/daemon/src/process.rs | 257 ++++++++++++++++++++ packages/env/daemon/src/pty.rs | 48 +--- packages/env/daemon/src/sys/unix.rs | 43 +++- packages/env/daemon/src/sys/windows.rs | 157 ++++++++++-- packages/env/docs/protocol.md | 27 ++- packages/env/src/index.ts | 1 + packages/env/src/process.ts | 116 +++++++++ packages/env/test/process.test.ts | 321 +++++++++++++++++++++++++ 12 files changed, 1057 insertions(+), 121 deletions(-) create mode 100644 packages/env/daemon/src/process.rs create mode 100644 packages/env/src/process.ts create mode 100644 packages/env/test/process.test.ts diff --git a/packages/env/CHANGELOG.md b/packages/env/CHANGELOG.md index 6c101bdd4..7f16fe545 100644 --- a/packages/env/CHANGELOG.md +++ b/packages/env/CHANGELOG.md @@ -2,6 +2,14 @@ ## [Unreleased] +### Added + +- The daemon's `process` operation runs a program with plain pipes for stdin, stdout and stderr, for programs that speak a protocol on stdin and stdout such as MCP servers: `processWrite` and `processCloseStdin` feed its stdin in arrival order, output arrives as raw bytes per stream, and cancelling or a lost connection kills its process group. `hello` reports it in `features`. `openProcess()` returns a `RemoteProcess`; daemons released before it fail the open with `not_supported`. Unlike terminals, it works on Windows too. + +### Fixed + +- On Windows, an `exec` command's output readers stop once the command has settled, also while a descendant keeps a pipe open without writing; they stayed blocked until it exited, because cancelling a blocked read did not reach the standard library's overlapped pipe I/O. + ## [1.5.0] - 2026-11-02 ### Breaking Changes diff --git a/packages/env/README.md b/packages/env/README.md index a112dba56..0ff5c819f 100644 --- a/packages/env/README.md +++ b/packages/env/README.md @@ -55,6 +55,22 @@ daemon: raw output, input, resizing and its exit code, for interactive terminals machine, for terminals without SSH (`new Connection({ command: [localDaemon()] })`). Terminals are not supported on Windows yet. +`openProcess(connection, options)` runs a program with plain pipes, for programs that speak a protocol on stdin and +stdout, such as an MCP server: `write()` and `closeStdin()` for its input, raw stdout and stderr bytes through +`onStdout` and `onStderr`, `close()` to kill its process group, and `exited` for its exit code. Nothing is echoed, +translated or merged, unlike a terminal; it works on every supported system. A daemon released before it fails the open +with `not_supported`. `argv` runs as an `exec` argv command does: on Windows, batch files such as `npx.cmd` are refused, +as Node refuses them without a shell, so callers start them through `cmd.exe`. + +```ts +const server = await openProcess(connection, { + cwd: "/home/me/project", + argv: ["npx", "-y", "@modelcontextprotocol/server-filesystem", "."], + onStdout: (bytes) => parser.push(bytes), +}); +server.write(`${JSON.stringify(initialize)}\n`); +``` + Supported remote systems: Linux, macOS, Android (Termux) and Windows, on x86-64 and arm64. On Windows, string commands run through Git Bash as `NodeExecutionEnv` runs them there; argv commands run directly. diff --git a/packages/env/daemon/src/exec.rs b/packages/env/daemon/src/exec.rs index d92c51279..072a5ad70 100644 --- a/packages/env/daemon/src/exec.rs +++ b/packages/env/daemon/src/exec.rs @@ -1,4 +1,5 @@ -//! `exec`: Durable's `NodeExecutionEnv.exec` semantics (docs/semantics.md), output streamed as `output` events. +//! `exec`: Durable's `NodeExecutionEnv.exec` semantics (docs/semantics.md), output streamed as `output` events; and what +//! the commands of `exec`, `pty` and `process` share: cancellation, starting, reading pipes and reaping. use crate::decode::StreamDecoder; use crate::errors::Failure; @@ -7,7 +8,7 @@ use crate::output::Output; use crate::sys; use crate::window::{Pending, Window}; use serde_json::{Map, Value, json}; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::fs::{File, OpenOptions}; use std::io::Write; use std::path::Path; @@ -188,30 +189,94 @@ impl Spill { } } -fn spawn( - request: &ExecRequest, +/// Where and with which environment a program starts. +pub struct Launch<'a> { + pub cwd: &'a str, + pub env: &'a Map, + pub inherit_env: bool, +} + +/// Start `program` with piped stdout and stderr, in a process group of its own, as libuv starts commands. +pub fn spawn( program: &str, args: &[String], - piped_stdin: bool, + launch: &Launch, + stdin: Stdio, ) -> std::io::Result { let mut command = Command::new(program); command - .current_dir(&request.cwd) - .stdin(if piped_stdin { - Stdio::piped() - } else { - Stdio::null() - }) + .current_dir(launch.cwd) + .stdin(stdin) .stdout(Stdio::piped()) .stderr(Stdio::piped()); - sys::configure(&mut command, args, &request.env, request.inherit_env); + sys::configure(&mut command, args, launch.env, launch.inherit_env); let child = command.spawn()?; sys::after_spawn(&child); Ok(child) } -/// Read one pipe on its own thread. Without an output window, reading waits while too much output is unsent. -fn spawn_reader( +/// Reap `child` on a thread of its own, which sends its exit to `sender`. A thread that cannot start hands the child +/// back with the error, so the caller kills and reaps it. +pub fn start_waiter(child: Child, sender: Sender) -> Result<(), (Child, std::io::Error)> { + // The waiter receives the child once it runs, so a waiter that cannot start leaves the child here to reap. + let (give, take) = mpsc::channel::(); + let started = thread::Builder::new().spawn(move || { + let Ok(mut child) = take.recv() else { + return; + }; + let _ = sender.send(match child.wait() { + Ok(status) => Message::Exit(status), + Err(error) => Message::WaitFailed(error.to_string()), + }); + }); + match started { + Ok(_) => { + let _ = give.send(child); + Ok(()) + } + Err(error) => Err((child, error)), + } +} + +/// The `argv` of a request: `None` when absent, else a non-empty array of strings. +pub fn argv(json: &Value) -> Result>, Failure> { + match &json["argv"] { + Value::Null => Ok(None), + value => value + .as_array() + .and_then(|items| { + items + .iter() + .map(|item| item.as_str().map(str::to_string)) + .collect::>>() + }) + .filter(|argv| !argv.is_empty()) + .map(Some) + .ok_or_else(|| Failure::new("EINVAL", "argv must be a non-empty array of strings")), + } +} + +/// The running request a follow-up request names in `target` (`ptyWrite`, `processWrite`, ...); `what` names its kind +/// in the error. +pub fn running( + requests: &Mutex>>, + json: &Value, + what: &str, +) -> Result, Failure> { + let target = json["target"] + .as_u64() + .ok_or_else(|| Failure::new("EINVAL", "missing field target"))?; + requests + .lock() + .unwrap() + .get(&(target as u32)) + .cloned() + .ok_or_else(|| Failure::new("EBADF", format!("no running {what} with this id"))) +} + +/// Read one pipe on its own thread until it ends or `stop` is set. With `output`, reading waits while too much output +/// is unsent. +pub fn spawn_reader( stream: usize, pipe: impl sys::Pipe + Send + 'static, sender: Sender, @@ -283,7 +348,17 @@ pub fn run( ), )); } - let mut child = spawn(&request, &program, &args, stdin_command.is_some()) + let launch = Launch { + cwd: &request.cwd, + env: &request.env, + inherit_env: request.inherit_env, + }; + let stdin = if stdin_command.is_some() { + Stdio::piped() + } else { + Stdio::null() + }; + let mut child = spawn(&program, &args, &launch, stdin) .map_err(|error| Failure::new("spawn_error", error.to_string()))?; if let (Some(command), Some(mut stdin)) = (stdin_command, child.stdin.take()) { let _ = stdin.write_all(command.as_bytes()); @@ -298,12 +373,10 @@ pub fn run( } else { None }; - // A thread that cannot start (a task limit) fails the command; the started child is killed and reaped. - let abandon = |mut child: Child, readers: &[JoinHandle<()>], error: std::io::Error| { + // A thread that cannot start (a task limit) fails the command; the started child is killed and reaped. Readers + // already running see `stop` within 100 ms. + let abandon = |mut child: Child, error: std::io::Error| { stop.store(true, Ordering::SeqCst); - for reader in readers { - sys::interrupt_reader(reader); - } kill_group(pid); // On Windows the group kill spawns taskkill, which can fail under the same limit; kill the child directly. let _ = child.kill(); @@ -311,33 +384,17 @@ pub fn run( groups.lock().unwrap().remove(&pid); Failure::new("spawn_error", error.to_string()) }; - let mut readers = Vec::with_capacity(2); let stdout = child.stdout.take().unwrap(); - match spawn_reader(0, stdout, sender.clone(), stop.clone(), gate.clone()) { - Ok(reader) => readers.push(reader), - Err(error) => return Err(abandon(child, &readers, error)), + if let Err(error) = spawn_reader(0, stdout, sender.clone(), stop.clone(), gate.clone()) { + return Err(abandon(child, error)); } let stderr = child.stderr.take().unwrap(); - match spawn_reader(1, stderr, sender.clone(), stop.clone(), gate) { - Ok(reader) => readers.push(reader), - Err(error) => return Err(abandon(child, &readers, error)), + if let Err(error) = spawn_reader(1, stderr, sender.clone(), stop.clone(), gate) { + return Err(abandon(child, error)); } - // The waiter receives the child once it runs, so a waiter that cannot start leaves the child here to reap. - let (give, take) = mpsc::channel::(); - let waiter = sender.clone(); - let started = thread::Builder::new().spawn(move || { - let Ok(mut child) = take.recv() else { - return; - }; - let _ = waiter.send(match child.wait() { - Ok(status) => Message::Exit(status), - Err(error) => Message::WaitFailed(error.to_string()), - }); - }); - if let Err(error) = started { - return Err(abandon(child, &readers, error)); + if let Err((child, error)) = start_waiter(child, sender.clone()) { + return Err(abandon(child, error)); } - let _ = give.send(child); drop(sender); let mut decoders = [StreamDecoder::new(), StreamDecoder::new()]; @@ -452,11 +509,9 @@ pub fn run( Err(RecvTimeoutError::Disconnected) => break, } } - // Like Node destroying its streams: stop reading, so the pipes close even if a descendant still holds them. + // Like Node destroying its streams: readers stop within 100 ms, so the pipes close even if a descendant still holds + // them. stop.store(true, Ordering::SeqCst); - for reader in &readers { - sys::interrupt_reader(reader); - } groups.lock().unwrap().remove(&pid); emit(&mut pending, 0, decoders[0].decode(&[], true)); emit(&mut pending, 1, decoders[1].decode(&[], true)); diff --git a/packages/env/daemon/src/main.rs b/packages/env/daemon/src/main.rs index ea720eabb..d567463e2 100644 --- a/packages/env/daemon/src/main.rs +++ b/packages/env/daemon/src/main.rs @@ -8,6 +8,7 @@ mod frame; mod fs; mod list; mod output; +mod process; mod pty; mod scan; mod sys; @@ -31,7 +32,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; const PROTOCOL: u64 = 1; /// Operations added after protocol 1's first release; clients check `hello.features` before using them. -const FEATURES: &[&str] = &["listFiles"]; +const FEATURES: &[&str] = &["listFiles", "process"]; /// The npm package version this daemon ships with (build.rs). const VERSION: &str = env!("PI_ENV_VERSION"); const PING_INTERVAL: Duration = Duration::from_secs(5); @@ -79,6 +80,7 @@ struct Server { controls: Mutex>>, groups: exec::Groups, terminals: pty::Terminals, + processes: process::Processes, } fn now_ms() -> u64 { @@ -268,6 +270,14 @@ impl Server { &self.groups, &self.terminals, )), + "process" => plain(process::run( + id, + process::ProcessRequest::from_json(json)?, + &self.output, + control, + &self.groups, + &self.processes, + )), _ => Err(Failure::new("EINVAL", format!("unknown operation {op}"))), } } @@ -360,7 +370,10 @@ fn start_workers() -> Sender { } fn respond(server: &Server, id: u32, request: Frame, control: Arc) { - let is_exec = matches!(request.json["op"].as_str(), Some("exec" | "pty")); + let is_exec = matches!( + request.json["op"].as_str(), + Some("exec" | "pty" | "process") + ); let reply = match server.dispatch(id, &request.json, request.payload, &control) { Ok((json, payload)) => Frame::with_payload(RESULT, id, json, payload), Err(failure) => Frame::new(ERROR, id, failure.to_json()), @@ -408,7 +421,13 @@ fn serve_frames(server: &Arc, input: &mut impl Read) -> io::Result<()> { REQUEST if matches!( request.json["op"].as_str(), - Some("ptyWrite" | "ptyResize" | "ptyForeground") + Some( + "ptyWrite" + | "ptyResize" + | "ptyForeground" + | "processWrite" + | "processCloseStdin" + ) ) => { // Answered here, in arrival order: through the worker pool, two writes could swap. @@ -417,6 +436,12 @@ fn serve_frames(server: &Arc, input: &mut impl Read) -> io::Result<()> { pty::write(&server.terminals, &request.json, request.payload) } Some("ptyResize") => pty::resize(&server.terminals, &request.json), + Some("processWrite") => { + process::write(&server.processes, &request.json, request.payload) + } + Some("processCloseStdin") => { + process::close_stdin(&server.processes, &request.json) + } _ => pty::foreground(&server.terminals, &request.json), }; server.output.control(match reply { @@ -430,7 +455,10 @@ fn serve_frames(server: &Arc, input: &mut impl Read) -> io::Result<()> { let control = Arc::new(Control::default()); server.controls.lock().unwrap().insert(id, control.clone()); // Commands and watchers last; they get their own threads (watchers recurse through trees). - if matches!(request.json["op"].as_str(), Some("exec" | "watch" | "pty")) { + if matches!( + request.json["op"].as_str(), + Some("exec" | "watch" | "pty" | "process") + ) { let worker = server.clone(); let spawned = thread::Builder::new() .stack_size(16 * 1024 * 1024) @@ -499,6 +527,7 @@ fn serve(token: &str) -> io::Result<()> { controls: Mutex::new(HashMap::new()), groups, terminals: Arc::new(Mutex::new(HashMap::new())), + processes: Arc::new(Mutex::new(HashMap::new())), }); let last_seen = Arc::new(AtomicU64::new(now_ms())); { diff --git a/packages/env/daemon/src/process.rs b/packages/env/daemon/src/process.rs new file mode 100644 index 000000000..82f4da32f --- /dev/null +++ b/packages/env/daemon/src/process.rs @@ -0,0 +1,257 @@ +//! `process`: a long-lived program with plain pipes, for programs that speak a protocol on stdin and stdout (an MCP +//! server). Unlike `exec`, the program reads input (`processWrite`, `processCloseStdin`, addressed by the `process` +//! request's id) and output is raw bytes per stream; unlike `pty`, nothing is echoed, translated or merged. + +use crate::errors::Failure; +use crate::exec::{ + Control, Groups, Launch, Message, kill_group, running, spawn, spawn_reader, start_waiter, +}; +use crate::frame::{EVENT, Frame}; +use crate::output::Output; +use crate::sys; +use serde_json::{Map, Value, json}; +use std::collections::HashMap; +use std::path::Path; +use std::process::{Child, ExitStatus, Stdio}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{self, RecvTimeoutError, Sender}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +/// After the program exits, output still arriving is collected until this much silence, as for `exec`. +const EXIT_STDIO_GRACE: Duration = Duration::from_millis(100); + +/// Input for a program's stdin, in arrival order. +enum Input { + Data(Vec), + Close, +} + +/// What `processWrite` and `processCloseStdin` reach: the stdin of one running `process` request. +pub struct Process { + input: Sender, +} + +/// Running programs by the id of their `process` request. +pub type Processes = Arc>>>; + +pub struct ProcessRequest { + argv: Vec, + cwd: String, + env: Map, + inherit_env: bool, +} + +impl ProcessRequest { + pub fn from_json(json: &Value) -> Result { + Ok(ProcessRequest { + argv: crate::exec::argv(json)? + .ok_or_else(|| Failure::new("EINVAL", "process needs argv"))?, + cwd: json["cwd"] + .as_str() + .ok_or_else(|| Failure::new("EINVAL", "process needs cwd"))? + .to_string(), + env: json["env"].as_object().cloned().unwrap_or_default(), + inherit_env: json["inheritEnv"].as_bool().unwrap_or(true), + }) + } +} + +/// Queue input for the program's stdin; it is written in the order it arrives, without holding up other requests. +pub fn write(processes: &Processes, json: &Value, payload: Vec) -> Result { + running(processes, json, "process")? + .input + .send(Input::Data(payload)) + .map_err(|_| Failure::new("EBADF", "the program's stdin is closed"))?; + Ok(json!({})) +} + +/// Close the program's stdin once the input queued before has been written. +pub fn close_stdin(processes: &Processes, json: &Value) -> Result { + running(processes, json, "process")? + .input + .send(Input::Close) + .map_err(|_| Failure::new("EBADF", "the program's stdin is closed"))?; + Ok(json!({})) +} + +/// Run the program, sending a `started` event and then its output, until it exits. +pub fn run( + id: u32, + request: ProcessRequest, + output: &Arc, + control: &Control, + groups: &Groups, + processes: &Processes, +) -> Result { + let (sender, receiver) = mpsc::channel(); + control.attach(sender.clone()); + if control.aborted.load(Ordering::SeqCst) { + return Err(Failure::new("aborted", "aborted")); + } + let program = sys::resolve_program(&request.argv[0], &request.cwd, &request.env)?; + if !Path::new(&request.cwd).exists() { + return Err(Failure::new( + "spawn_error", + format!("Working directory does not exist: {}", request.cwd), + )); + } + let launch = Launch { + cwd: &request.cwd, + env: &request.env, + inherit_env: request.inherit_env, + }; + let mut child = spawn(&program, &request.argv[1..], &launch, Stdio::piped()) + .map_err(|error| Failure::new("spawn_error", error.to_string()))?; + let pid = child.id(); + groups.lock().unwrap().insert(pid); + let stop = Arc::new(AtomicBool::new(false)); + // A thread that cannot start (a task limit) fails the request; the started program is killed and reaped. Threads + // already running see `stop` within 100 ms. + let abandon = |mut child: Child, error: std::io::Error| { + stop.store(true, Ordering::SeqCst); + kill_group(pid); + // On Windows the group kill spawns taskkill, which can fail under the same limit; kill the child directly. + let _ = child.kill(); + let _ = child.wait(); + groups.lock().unwrap().remove(&pid); + Failure::new("spawn_error", error.to_string()) + }; + // Like a full pipe: a program that writes faster than the link waits. + let stdout = child.stdout.take().unwrap(); + if let Err(error) = spawn_reader( + 0, + stdout, + sender.clone(), + stop.clone(), + Some(output.clone()), + ) { + return Err(abandon(child, error)); + } + let stderr = child.stderr.take().unwrap(); + if let Err(error) = spawn_reader( + 1, + stderr, + sender.clone(), + stop.clone(), + Some(output.clone()), + ) { + return Err(abandon(child, error)); + } + let (input, inputs) = mpsc::channel::(); + let mut stdin = child.stdin.take().unwrap(); + let started = { + let stop = stop.clone(); + thread::Builder::new().spawn(move || { + for input in inputs { + match input { + Input::Data(bytes) => { + if !sys::write_pipe(&mut stdin, &bytes, &stop) { + return; + } + } + // Returning drops stdin, which closes it; later input fails with `EBADF`. + Input::Close => return, + } + } + }) + }; + if let Err(error) = started { + return Err(abandon(child, error)); + } + if let Err((child, error)) = start_waiter(child, sender.clone()) { + return Err(abandon(child, error)); + } + drop(sender); + // Registered before `started` goes out, so input sent after it always finds the program. + processes + .lock() + .unwrap() + .insert(id, Arc::new(Process { input })); + output.control(Frame::new( + EVENT, + id, + json!({ "kind": "started", "pid": pid }), + )); + + let names = ["stdout", "stderr"]; + let mut ended = [false, false]; + let mut status: Option> = None; + let mut idle_until: Option = None; + let mut aborted = false; + let mut killed = false; + let kill = |killed: &mut bool| { + if !*killed { + *killed = true; + kill_group(pid); + } + }; + if control.killed.load(Ordering::SeqCst) { + kill(&mut killed); + } + loop { + if status.is_some() && ended[0] && ended[1] { + break; + } + let wait = match idle_until { + Some(idle) => { + let now = Instant::now(); + if now >= idle { + // A descendant may hold the pipes open after the program exited; stop waiting for it. + break; + } + idle - now + } + None => Duration::from_secs(3600), + }; + match receiver.recv_timeout(wait) { + Ok(Message::Data(stream, bytes)) => { + output.bulk( + Frame::with_payload( + EVENT, + id, + json!({ "kind": "output", "stream": names[stream] }), + bytes, + ), + None, + ); + if status.is_some() { + idle_until = Some(Instant::now() + EXIT_STDIO_GRACE); + } + } + Ok(Message::Eof(stream)) => ended[stream] = true, + Ok(Message::Exit(exit)) => { + status = Some(Ok(exit)); + idle_until = Some(Instant::now() + EXIT_STDIO_GRACE); + } + Ok(Message::WaitFailed(error)) => { + status = Some(Err(error)); + idle_until = Some(Instant::now()); + } + Ok(Message::Cancel) => { + aborted = true; + kill(&mut killed); + } + Ok(Message::Kill) => kill(&mut killed), + Err(RecvTimeoutError::Timeout) => {} + Err(RecvTimeoutError::Disconnected) => break, + } + } + // Readers and the writer stop within 100 ms, also when a descendant still holds a pipe; removing the program + // drops its input channel, which ends a writer waiting for input. + stop.store(true, Ordering::SeqCst); + processes.lock().unwrap().remove(&id); + groups.lock().unwrap().remove(&pid); + if aborted { + return Err(Failure::new("aborted", "aborted")); + } + match status { + Some(Ok(status)) => Ok(json!({ "exitCode": sys::exit_code(status) })), + Some(Err(error)) => Err(Failure::new("spawn_error", error)), + None => Err(Failure::new( + "spawn_error", + "process did not report its exit", + )), + } +} diff --git a/packages/env/daemon/src/pty.rs b/packages/env/daemon/src/pty.rs index e8be9e222..9df9dd9fd 100644 --- a/packages/env/daemon/src/pty.rs +++ b/packages/env/daemon/src/pty.rs @@ -3,9 +3,9 @@ //! its id. use crate::errors::Failure; -use crate::exec::{Control, Groups}; +use crate::exec::{Control, Groups, running}; #[cfg(unix)] -use crate::exec::{Message, kill_group}; +use crate::exec::{Message, kill_group, start_waiter}; #[cfg(unix)] use crate::frame::{EVENT, Frame}; use crate::output::Output; @@ -55,25 +55,8 @@ fn size(json: &Value, key: &str) -> Result { impl PtyRequest { pub fn from_json(json: &Value) -> Result { - let argv = match &json["argv"] { - Value::Null => None, - value => Some( - value - .as_array() - .and_then(|items| { - items - .iter() - .map(|item| item.as_str().map(str::to_string)) - .collect() - }) - .filter(|argv: &Vec| !argv.is_empty()) - .ok_or_else(|| { - Failure::new("EINVAL", "argv must be a non-empty array of strings") - })?, - ), - }; Ok(PtyRequest { - argv, + argv: crate::exec::argv(json)?, cwd: json["cwd"] .as_str() .ok_or_else(|| Failure::new("EINVAL", "pty needs cwd"))? @@ -87,15 +70,7 @@ impl PtyRequest { } fn find(terminals: &Terminals, json: &Value) -> Result, Failure> { - let target = json["target"] - .as_u64() - .ok_or_else(|| Failure::new("EINVAL", "missing field target"))?; - terminals - .lock() - .unwrap() - .get(&(target as u32)) - .cloned() - .ok_or_else(|| Failure::new("EBADF", "no running terminal with this id")) + running(terminals, json, "terminal") } /// Queue input for the terminal; it is written in the order it arrives, without holding up other requests. @@ -245,22 +220,9 @@ pub fn run( if let Err(error) = started { return Err(abandon(child, error)); } - // The waiter receives the child once it runs, so a waiter that cannot start leaves the child here to reap. - let (give, take) = mpsc::channel::(); - let waiter = sender.clone(); - let started = thread::Builder::new().spawn(move || { - let Ok(mut child) = take.recv() else { - return; - }; - let _ = waiter.send(match child.wait() { - Ok(status) => Message::Exit(status), - Err(error) => Message::WaitFailed(error.to_string()), - }); - }); - if let Err(error) = started { + if let Err((child, error)) = start_waiter(child, sender.clone()) { return Err(abandon(child, error)); } - let _ = give.send(child); drop(sender); let mut status = None; diff --git a/packages/env/daemon/src/sys/unix.rs b/packages/env/daemon/src/sys/unix.rs index 4b0637f0e..a8564857a 100644 --- a/packages/env/daemon/src/sys/unix.rs +++ b/packages/env/daemon/src/sys/unix.rs @@ -153,7 +153,8 @@ pub fn remove_file(path: &str) -> io::Result<()> { pub trait Pipe: io::Read + std::os::fd::AsRawFd {} impl Pipe for T {} -/// Read `pipe` until it ends, `on_data` returns false, or `stop` is set; the pipe closes when this returns. +/// Read `pipe` until it ends, `on_data` returns false, or `stop` is set (looked at every 100 ms, also while no output +/// arrives); the pipe closes when this returns. pub fn read_pipe(mut pipe: impl Pipe, stop: &AtomicBool, mut on_data: impl FnMut(&[u8]) -> bool) { let fd = pipe.as_raw_fd(); // SAFETY: fcntl on a descriptor this thread owns. @@ -195,8 +196,44 @@ pub fn read_pipe(mut pipe: impl Pipe, stop: &AtomicBool, mut on_data: impl FnMut } } -/// Readers poll `stop` every 100 ms. -pub fn interrupt_reader(_reader: &std::thread::JoinHandle<()>) {} +/// A pipe a writer thread can poll, so that it stops when the program settles. +pub trait WritePipe: io::Write + std::os::fd::AsRawFd {} +impl WritePipe for T {} + +/// Write all of `bytes`, waiting while the program does not read; false once the pipe is closed or `stop` is set +/// (looked at every 100 ms, also while the write waits). +pub fn write_pipe(pipe: &mut impl WritePipe, mut bytes: &[u8], stop: &AtomicBool) -> bool { + let fd = pipe.as_raw_fd(); + // SAFETY: fcntl on a descriptor this thread owns; only the daemon's end of the pipe becomes nonblocking. + unsafe { + let flags = libc::fcntl(fd, libc::F_GETFL); + libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK); + } + while !bytes.is_empty() { + if stop.load(Ordering::SeqCst) { + return false; + } + match pipe.write(bytes) { + Ok(0) => return false, + Ok(written) => bytes = &bytes[written..], + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + let mut poll = libc::pollfd { + fd, + events: libc::POLLOUT, + revents: 0, + }; + // SAFETY: one valid pollfd. + let ready = unsafe { libc::poll(&mut poll, 1, 100) }; + if ready < 0 && io::Error::last_os_error().kind() != io::ErrorKind::Interrupted { + return false; + } + } + Err(error) if error.kind() == io::ErrorKind::Interrupted => {} + Err(_) => return false, + } + } + true +} /// The program an argv command runs; `Command` searches `PATH` like `execvp`, as libuv does. pub fn resolve_program( diff --git a/packages/env/daemon/src/sys/windows.rs b/packages/env/daemon/src/sys/windows.rs index 401719369..f2a7516d0 100644 --- a/packages/env/daemon/src/sys/windows.rs +++ b/packages/env/daemon/src/sys/windows.rs @@ -10,29 +10,33 @@ use std::hash::{BuildHasher, Hasher}; use std::io; use std::os::windows::ffi::OsStrExt; use std::os::windows::fs::FileExt; -use std::os::windows::io::{AsRawHandle, FromRawHandle}; +use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; use std::os::windows::process::CommandExt; use std::path::Path; use std::process::{Child, Command, ExitStatus, Stdio}; use std::sync::OnceLock; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::UNIX_EPOCH; -use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE}; +use windows_sys::Win32::Foundation::{ + CloseHandle, ERROR_IO_PENDING, FALSE, HANDLE, INVALID_HANDLE_VALUE, TRUE, WAIT_OBJECT_0, + WAIT_TIMEOUT, +}; use windows_sys::Win32::Storage::FileSystem::{ BY_HANDLE_FILE_INFORMATION, CREATE_ALWAYS, CreateFileW, FILE_APPEND_DATA, FILE_ATTRIBUTE_NORMAL, FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, FILE_GENERIC_READ, FILE_GENERIC_WRITE, FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE, FILE_WRITE_DATA, GetFileInformationByHandle, MOVEFILE_REPLACE_EXISTING, MoveFileExW, - OPEN_ALWAYS, OPEN_EXISTING, + OPEN_ALWAYS, OPEN_EXISTING, ReadFile, WriteFile, }; -use windows_sys::Win32::System::IO::CancelSynchronousIo; +use windows_sys::Win32::System::IO::{CancelIoEx, GetOverlappedResult, OVERLAPPED}; use windows_sys::Win32::System::JobObjects::{ AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_BREAKAWAY_OK, JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JobObjectExtendedLimitInformation, SetInformationJobObject, }; -use windows_sys::Win32::System::Threading::CREATE_NO_WINDOW; +use windows_sys::Win32::System::Threading::{CREATE_NO_WINDOW, CreateEventW, WaitForSingleObject}; +use windows_sys::core::BOOL; /// libuv's `uv_translate_sys_error` for the errors file and process operations produce. pub fn os_error_name(error: &io::Error) -> &'static str { @@ -253,31 +257,146 @@ pub fn remove_file(path: &str) -> io::Result<()> { } } -/// A pipe a reader thread reads with blocking calls; `interrupt_reader` cancels a blocked read. -pub trait Pipe: io::Read {} -impl Pipe for T {} +/// The daemon's end of a child pipe. The standard library opens it for overlapped I/O, so a blocked read or write can only +/// be cancelled with `CancelIoEx`, not `CancelSynchronousIo`; readers and writers issue their own overlapped operations +/// and wait for them 100 ms at a time instead, like the Unix side's `poll`. +pub trait Pipe: AsRawHandle {} +impl Pipe for T {} -/// Read `pipe` until it ends, `on_data` returns false, or `stop` is set; the pipe closes when this returns. -pub fn read_pipe(mut pipe: impl Pipe, stop: &AtomicBool, mut on_data: impl FnMut(&[u8]) -> bool) { +/// The writing end, as `Pipe`. +pub trait WritePipe: AsRawHandle {} +impl WritePipe for T {} + +/// How long one wait for an overlapped operation lasts before `stop` is looked at again. +const STOP_POLL_MS: u32 = 100; + +/// Overlapped operations on one pipe handle, each completed or cancelled by the thread that started it, so nothing is +/// left pending on a buffer that goes away. +struct Overlapped { + handle: HANDLE, + event: OwnedHandle, +} + +impl Overlapped { + fn new(pipe: &impl AsRawHandle) -> io::Result { + // SAFETY: a manual-reset, unnamed event; ReadFile and WriteFile reset it when they start. + let event = unsafe { CreateEventW(std::ptr::null(), TRUE, FALSE, std::ptr::null()) }; + if event.is_null() { + return Err(io::Error::last_os_error()); + } + Ok(Overlapped { + handle: pipe.as_raw_handle() as HANDLE, + // SAFETY: the event created above, owned from here on. + event: unsafe { OwnedHandle::from_raw_handle(event as _) }, + }) + } + + /// Run one operation `start` begins with the `OVERLAPPED` it is given; the bytes it transferred, or `None` once + /// `stop` was set while it waited, after cancelling it. + fn run( + &self, + stop: &AtomicBool, + start: impl FnOnce(*mut OVERLAPPED) -> BOOL, + ) -> io::Result> { + // SAFETY: a zeroed OVERLAPPED is valid; only its event is set. + let mut overlapped: OVERLAPPED = unsafe { std::mem::zeroed() }; + overlapped.hEvent = self.event.as_raw_handle() as HANDLE; + if start(&mut overlapped) == FALSE { + let error = io::Error::last_os_error(); + if error.raw_os_error() != Some(ERROR_IO_PENDING as i32) { + return Err(error); + } + } + loop { + // SAFETY: the event outlives the wait. + let waited = unsafe { WaitForSingleObject(overlapped.hEvent, STOP_POLL_MS) }; + if waited == WAIT_OBJECT_0 { + return self.result(&overlapped, FALSE).map(Some); + } + if waited == WAIT_TIMEOUT && !stop.load(Ordering::SeqCst) { + continue; + } + let failure = (waited != WAIT_TIMEOUT).then(io::Error::last_os_error); + // SAFETY: cancels only this operation, then waits for it to finish before `overlapped` and the caller's + // buffer go away. + unsafe { CancelIoEx(self.handle, &overlapped) }; + let _ = self.result(&overlapped, TRUE); + return failure.map_or(Ok(None), Err); + } + } + + fn result(&self, overlapped: &OVERLAPPED, wait: BOOL) -> io::Result { + let mut transferred = 0u32; + // SAFETY: the OVERLAPPED of an operation started on this handle. + if unsafe { GetOverlappedResult(self.handle, overlapped, &mut transferred, wait) } == FALSE + { + return Err(io::Error::last_os_error()); + } + Ok(transferred) + } +} + +/// Read `pipe` until it ends, `on_data` returns false, or `stop` is set (looked at every 100 ms, also while no output +/// arrives); the pipe closes when this returns. +pub fn read_pipe(pipe: impl Pipe, stop: &AtomicBool, mut on_data: impl FnMut(&[u8]) -> bool) { + let Ok(io) = Overlapped::new(&pipe) else { + return; + }; let mut buffer = vec![0u8; 64 * 1024]; while !stop.load(Ordering::SeqCst) { - match pipe.read(&mut buffer) { - Ok(0) | Err(_) => return, - Ok(read) => { - if !on_data(&buffer[..read]) { + let read = io.run(stop, |overlapped| { + // SAFETY: the buffer outlives the operation, which `run` completes or cancels before returning. + unsafe { + ReadFile( + io.handle, + buffer.as_mut_ptr(), + buffer.len() as u32, + std::ptr::null_mut(), + overlapped, + ) + } + }); + // A broken pipe (every writer closed) is the end, as for the standard library's reads. + match read { + Ok(Some(read)) if read > 0 => { + if !on_data(&buffer[..read as usize]) { return; } } + _ => return, } } } -/// Cancel a reader blocked in `ReadFile`, so it sees `stop` and closes its pipe. -pub fn interrupt_reader(reader: &std::thread::JoinHandle<()>) { - // SAFETY: the handle of a live or finished thread; cancelling no pending I/O is harmless. - unsafe { - CancelSynchronousIo(reader.as_raw_handle() as _); +/// Write all of `bytes`, waiting while the program does not read; false once the pipe is closed or `stop` is set +/// (looked at every 100 ms, also while the write waits). +pub fn write_pipe(pipe: &mut impl WritePipe, mut bytes: &[u8], stop: &AtomicBool) -> bool { + let Ok(io) = Overlapped::new(pipe) else { + return false; + }; + while !bytes.is_empty() { + if stop.load(Ordering::SeqCst) { + return false; + } + let length = bytes.len().min(u32::MAX as usize) as u32; + let written = io.run(stop, |overlapped| { + // SAFETY: `bytes` outlives the operation, which `run` completes or cancels before returning. + unsafe { + WriteFile( + io.handle, + bytes.as_ptr(), + length, + std::ptr::null_mut(), + overlapped, + ) + } + }); + match written { + Ok(Some(written)) if written > 0 => bytes = &bytes[written as usize..], + _ => return false, + } } + true } /// libuv's `search_path` for an argv command: a name with a directory is tried as given, otherwise the working diff --git a/packages/env/docs/protocol.md b/packages/env/docs/protocol.md index c5e482856..b83d876ef 100644 --- a/packages/env/docs/protocol.md +++ b/packages/env/docs/protocol.md @@ -30,7 +30,7 @@ does not parse gets an `EINVAL` error; framing errors end the connection. | 1 | request | client → daemon | `json.op` names the operation | | 2 | result | daemon → client | success of request `id`; the request is finished | | 3 | error | daemon → client | failure of request `id`; the request is finished | -| 4 | event | daemon → client | progress of a running request (`exec` and `pty` output, `watch` changes), not finishing it | +| 4 | event | daemon → client | progress of a running request (`exec`, `pty` and `process` output, `watch` changes), not finishing it | | 5 | cancel | client → daemon | cancel request `id`; see below | | 6 | ping | both | liveness; the receiver ignores it | @@ -45,16 +45,16 @@ contents and output travel as raw payload bytes. ## Cancellation A cancel names one request. The daemon registers each request before running it, so a cancel sent right after its -request is never lost. `exec` and `pty` kill the command's process group and finish with `aborted`; `{ mode: "kill" }` kills +request is never lost. `exec`, `pty` and `process` kill the command's process group and finish with `aborted`; `{ mode: "kill" }` kills without aborting, so the command finishes with the killed process's status, as `cleanup()` does. `scanLines` stops between chunks and `listFiles` between entries, with `aborted`. `watch` stops and finishes with `{}`. Other operations are short and finish normally. ## Scheduling -File operations run on a fixed pool of worker threads; `exec`, `pty` and `watch` run on their own threads. `ptyWrite`, -`ptyResize` and `ptyForeground` are answered in arrival order by the thread that reads frames, so keystrokes never -swap. Chunk writes to one handle run in arrival order. Results, errors, pings and `watch` events are sent before queued +File operations run on a fixed pool of worker threads; `exec`, `pty`, `process` and `watch` run on their own threads. +`ptyWrite`, `ptyResize`, `ptyForeground`, `processWrite` and `processCloseStdin` are answered in arrival order by the +thread that reads frames, so keystrokes and protocol messages never swap. Chunk writes to one handle run in arrival order. Results, errors, pings and `watch` events are sent before queued command output, and results with payloads above 64 KiB queue behind output too, so a slow link delays bulk data, not replies. Output of a command without an output window waits while more than 4 MiB of output is unsent, which slows the command as a full pipe would. @@ -75,7 +75,7 @@ request. `version` is the npm package version the daemon shipped with; `os` and values (`linux`, `android`, `macos`, `windows`; `x86_64`, `aarch64`), which the client's `nodePlatform` maps to Node's names; `tmpdir` follows the remote Node's `os.tmpdir()` rules; `cwd` and `driveCwds` (Windows' `=C:` variables) are what Node's `path.resolve` falls back to for drive-relative paths. `features` lists operations added to protocol 1 -after its first release (`listFiles`); older daemons omit the field and refuse those operations as unknown, so clients +after its first release (`listFiles`, `process`); older daemons omit the field and refuse those operations as unknown, so clients check it first. File operations take `{ path }` (and the listed fields) and return `{}` unless noted: @@ -151,3 +151,18 @@ Once the terminal has ended, they fail with `EBADF`. `{ foreground: { pid, name } }`, where `pid` is the group's id and `name` its leader's process name (`null` if unknown), or `{ foreground: null }` while the program itself is in front. For a shell this tells a running job (an editor, a build) from the prompt; background jobs are not reported. It fails with `EBADF` once the terminal has ended. + +`process { argv, cwd, env, inheritEnv }` runs a program with plain pipes for stdin, stdout and stderr (feature +`process`), for programs that speak a protocol on stdin and stdout, such as MCP servers. `argv` is required and runs +directly, without a shell, resolved and started as an `exec` argv command is (its own process group, the same +environment rules, a job object on Windows). Once the program runs, an event `{ kind: "started", pid }` comes first; +then `{ kind: "output", stream }` events carry raw bytes of `stdout` or `stderr` as payload (not decoded, not +translated). Output waits while more than 4 MiB of output is unsent, like a full pipe. After the program exits, output +is collected until both pipes close or 100 ms pass without output. The result is `{ exitCode }`; a program that cannot +start fails with `spawn_error`. + +While a `process` request runs, `processWrite { target }` writes its payload to the program's stdin and +`processCloseStdin { target }` closes stdin after the input queued before it; `target` is the id of the `process` +request. Both return `{}` at once; input is queued and written in order, waiting while the program does not read it. +Input queued after `processCloseStdin` is dropped; once stdin is closed, and once the program has ended, both fail +with `EBADF`. diff --git a/packages/env/src/index.ts b/packages/env/src/index.ts index 0b22358f6..32b7904cf 100644 --- a/packages/env/src/index.ts +++ b/packages/env/src/index.ts @@ -9,6 +9,7 @@ export { type RequestOptions, } from "./connection.ts"; export { LocalExecutionEnv, type LocalExecutionEnvOptions } from "./list-files.ts"; +export { openProcess, type ProcessOptions, RemoteProcess } from "./process.ts"; export { RemoteExecutionEnv, type RemoteExecutionEnvOptions } from "./remote-env.ts"; export { acceptHostKey, diff --git a/packages/env/src/process.ts b/packages/env/src/process.ts new file mode 100644 index 000000000..473e7d4a6 --- /dev/null +++ b/packages/env/src/process.ts @@ -0,0 +1,116 @@ +import { type Connection, RemoteError } from "./connection.ts"; + +/** Input larger than this goes to the daemon in several writes; frames are at most 16 MiB. */ +const WRITE_CHUNK = 1024 * 1024; + +export interface ProcessOptions { + /** Working directory of the program; it must exist. */ + cwd: string; + /** Program and arguments, run directly without a shell. */ + argv: readonly string[]; + /** Added to the daemon's environment, or the whole environment with `inheritEnv: false`. */ + env?: Record; + inheritEnv?: boolean; + /** The program's stdout, raw bytes in order. */ + onStdout: (bytes: Uint8Array) => void; + /** The program's stderr, raw bytes in order. */ + onStderr?: (bytes: Uint8Array) => void; +} + +/** + * A program running with plain pipes in a pi-env daemon, for programs that speak a protocol on stdin and stdout. It + * ends when the program exits, `close()` kills it, or the connection is lost (the daemon then kills it). + */ +export class RemoteProcess { + readonly pid: number; + /** The exit code; rejects with the daemon's error, with code `aborted` after `close()`. */ + readonly exited: Promise; + readonly #connection: Connection; + readonly #id: number; + readonly #session: number; + readonly #abort: AbortController; + + constructor( + connection: Connection, + id: number, + session: number, + pid: number, + exited: Promise, + abort: AbortController, + ) { + this.#connection = connection; + this.#id = id; + this.#session = session; + this.pid = pid; + this.exited = exited; + this.#abort = abort; + } + + /** Write to the program's stdin, in order; once stdin is closed or the program has ended, input is dropped. */ + write(data: Uint8Array | string): void { + const bytes = typeof data === "string" ? new TextEncoder().encode(data) : data; + for (let offset = 0; offset < bytes.length; offset += WRITE_CHUNK) { + this.#request("processWrite", bytes.subarray(offset, offset + WRITE_CHUNK)); + } + } + + /** Close the program's stdin after the input written before it. */ + closeStdin(): void { + this.#request("processCloseStdin"); + } + + /** Kill the program and its process group. */ + close(): void { + this.#abort.abort(); + } + + #request(op: string, payload?: Uint8Array): void { + this.#connection + .request(op, { target: this.#id }, { session: this.#session, ...(payload ? { payload } : {}) }) + // Stdin already closed, the program ended, or a lost connection: `exited` reports the end. + .catch(() => {}); + } +} + +/** + * Start a program with plain pipes in the daemon behind `connection`. Fails with `not_supported` when the daemon was + * released before the `process` operation. + */ +export async function openProcess(connection: Connection, options: ProcessOptions): Promise { + if (!(await connection.info()).features?.includes("process")) { + throw new RemoteError({ code: "not_supported", message: "The pi-env daemon cannot run processes; update it" }); + } + const abort = new AbortController(); + const started = Promise.withResolvers(); + let request: { id: number; session: number } | undefined; + const exited = connection + .request( + "process", + { + cwd: options.cwd, + argv: options.argv, + ...(options.env === undefined ? {} : { env: options.env }), + ...(options.inheritEnv === undefined ? {} : { inheritEnv: options.inheritEnv }), + }, + { + signal: abort.signal, + onStart: (id, session) => { + request = { id, session }; + }, + onEvent: (json, payload) => { + if (json.kind === "output") { + if (json.stream === "stderr") options.onStderr?.(payload); + else options.onStdout(payload); + } else if (json.kind === "started" && request) { + started.resolve( + new RemoteProcess(connection, request.id, request.session, Number(json.pid), exited, abort), + ); + } + }, + }, + ) + .then((reply) => Number(reply.json.exitCode)); + // Failing before it started rejects the open; afterwards, `exited` carries it. + exited.catch(started.reject); + return started.promise; +} diff --git a/packages/env/test/process.test.ts b/packages/env/test/process.test.ts new file mode 100644 index 000000000..db1bbb505 --- /dev/null +++ b/packages/env/test/process.test.ts @@ -0,0 +1,321 @@ +import { createHash } from "node:crypto"; +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, describe, expect, it } from "vitest"; +import { Connection, RemoteError } from "../src/connection.ts"; +import { openProcess, type ProcessOptions } from "../src/process.ts"; +import { daemon } from "./daemon.ts"; + +const connections: Connection[] = []; +const dirs: string[] = []; +afterAll(() => { + for (const connection of connections) connection.close(); + for (const dir of dirs) rmSync(dir, { recursive: true, force: true }); +}); + +function connect(): { connection: Connection; cwd: string } { + const cwd = mkdtempSync(join(tmpdir(), "pi-env-process-")); + dirs.push(cwd); + const connection = new Connection({ command: [daemon] }); + connections.push(connection); + return { connection, cwd }; +} + +/** A Node script as the program, so the tests run the same on every system. */ +function node(script: string): string[] { + return [process.execPath, "-e", script]; +} + +/** Collects one stream's bytes and waits for text to appear in it. */ +function collector(): { + onData: (bytes: Uint8Array) => void; + bytes: () => Buffer; + waitFor: (text: string) => Promise; +} { + const chunks: Buffer[] = []; + const waiting: { text: string; resolve: () => void }[] = []; + const bytes = () => Buffer.concat(chunks); + return { + bytes, + onData: (data) => { + chunks.push(Buffer.from(data)); + const all = bytes().toString("latin1"); + for (const wait of waiting.splice(0)) { + if (all.includes(wait.text)) wait.resolve(); + else waiting.push(wait); + } + }, + waitFor: (text) => + bytes().toString("latin1").includes(text) + ? Promise.resolve() + : new Promise((resolve) => waiting.push({ text, resolve })), + }; +} + +function alive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +async function waitUntilGone(pid: number): Promise { + const deadline = Date.now() + 5000; + while (alive(pid) && Date.now() < deadline) await new Promise((done) => setTimeout(done, 50)); + return !alive(pid); +} + +/** Echoes each stdin chunk to stdout, and `err:` lines to stderr; exits with code 7 at the end of stdin. */ +const ECHO = ` +process.stdin.on("data", (chunk) => { + const text = chunk.toString("latin1"); + if (text.startsWith("err:")) process.stderr.write(text.slice(4)); + else process.stdout.write(chunk); +}); +process.stdin.on("end", () => process.exit(7)); +`; + +describe("openProcess", () => { + it("pipes stdin to the program and its stdout and stderr back separately", async () => { + const { connection, cwd } = connect(); + const stdout = collector(); + const stderr = collector(); + const program = await openProcess(connection, { + cwd, + argv: node(ECHO), + onStdout: stdout.onData, + onStderr: stderr.onData, + }); + expect(program.pid).toBeGreaterThan(0); + program.write('{"jsonrpc":"2.0","id":1}\n'); + await stdout.waitFor('{"jsonrpc":"2.0","id":1}\n'); + program.write("err:diagnostic\n"); + await stderr.waitFor("diagnostic\n"); + expect(stdout.bytes().toString("utf8")).toBe('{"jsonrpc":"2.0","id":1}\n'); + program.closeStdin(); + expect(await program.exited).toBe(7); + }); + + it("passes bytes through unchanged: no echo, no newline translation, no decoding", async () => { + const { connection, cwd } = connect(); + const stdout = collector(); + const program = await openProcess(connection, { + cwd, + argv: node(ECHO), + onStdout: stdout.onData, + }); + program.write(new Uint8Array([0x61, 0x0a, 0xff, 0x0d, 0x0a, 0x62])); + program.closeStdin(); + expect(await program.exited).toBe(7); + expect([...stdout.bytes()]).toEqual([0x61, 0x0a, 0xff, 0x0d, 0x0a, 0x62]); + }); + + it("passes the working directory and environment", async () => { + const { connection, cwd } = connect(); + const stdout = collector(); + const program = await openProcess(connection, { + cwd, + argv: node("process.stdout.write(JSON.stringify([process.cwd(), process.env.PI_ENV_PROCESS_TEST]))"), + env: { PI_ENV_PROCESS_TEST: "set" }, + onStdout: stdout.onData, + }); + expect(await program.exited).toBe(0); + const [where, variable] = JSON.parse(stdout.bytes().toString("utf8")) as [string, string]; + expect(where.toLowerCase()).toContain("pi-env-process-"); + expect(variable).toBe("set"); + }); +}); + +describe("input and output", () => { + it("keeps input in order across many writes and writes larger than one frame chunk", async () => { + const { connection, cwd } = connect(); + const stdout = collector(); + const program = await openProcess(connection, { + cwd, + // Reports the length and hash of everything it read once stdin ends. + argv: node(` + const hash = require("node:crypto").createHash("sha256"); + let length = 0; + process.stdin.on("data", (chunk) => { hash.update(chunk); length += chunk.length; }); + process.stdin.on("end", () => process.stdout.write(length + " " + hash.digest("hex"))); + `), + onStdout: stdout.onData, + }); + const expected = createHash("sha256"); + let length = 0; + for (let index = 0; index < 4000; index++) { + const piece = Buffer.from(`${String(index).padStart(9, "0")}\n`); + expected.update(piece); + length += piece.length; + program.write(piece); + } + const big = new Uint8Array(2_500_003); + for (let index = 0; index < big.length; index++) big[index] = (index * 7919) % 251; + expected.update(big); + length += big.length; + program.write(big); + program.closeStdin(); + expect(await program.exited).toBe(0); + expect(stdout.bytes().toString("utf8")).toBe(`${length} ${expected.digest("hex")}`); + }, 30_000); + + it("delivers large output completely", async () => { + const { connection, cwd } = connect(); + const stdout = collector(); + const program = await openProcess(connection, { + cwd, + argv: node(`process.stdout.write(Buffer.alloc(12 * 1024 * 1024, 120))`), + onStdout: stdout.onData, + }); + expect(await program.exited).toBe(0); + const output = stdout.bytes(); + expect(output.length).toBe(12 * 1024 * 1024); + expect(output.every((byte) => byte === 120)).toBe(true); + }, 30_000); + + it("refuses input after stdin was closed while the program keeps running", async () => { + const { connection, cwd } = connect(); + const started = Promise.withResolvers(); + let target = 0; + const running = connection.request( + "process", + { cwd, argv: node("process.stdin.resume(); setTimeout(() => {}, 30000)") }, + { + onStart: (id) => { + target = id; + }, + onEvent: (json) => { + if (json.kind === "started") started.resolve(); + }, + }, + ); + running.catch(() => {}); + await started.promise; + await connection.request("processWrite", { target }, { payload: new Uint8Array([1]) }); + await connection.request("processCloseStdin", { target }); + // The writer closes stdin after the input queued before it; from then on input is refused. + let refused: unknown; + for (let tries = 0; refused === undefined && tries < 100; tries++) { + try { + await connection.request("processWrite", { target }, { payload: new Uint8Array([2]) }); + await new Promise((done) => setTimeout(done, 10)); + } catch (error) { + refused = error; + } + } + expect(refused).toMatchObject({ code: "EBADF" }); + connection.close(); + }); + + it("refuses input for programs that do not run", async () => { + const { connection } = connect(); + await expect(connection.request("processWrite", { target: 999 })).rejects.toMatchObject({ code: "EBADF" }); + await expect(connection.request("processCloseStdin", { target: 999 })).rejects.toMatchObject({ code: "EBADF" }); + }); +}); + +describe("ending", () => { + it("reports the exit code of a program that exits on its own", async () => { + const { connection, cwd } = connect(); + const program = await openProcess(connection, { cwd, argv: node("process.exit(3)"), onStdout: () => {} }); + expect(await program.exited).toBe(3); + }); + + it("kills the program and what it started on close", async () => { + const { connection, cwd } = connect(); + const stdout = collector(); + const program = await openProcess(connection, { + cwd, + // A wrapper that starts the real server, as npx and uvx do. + argv: node(` + const child = require("node:child_process").spawn(process.execPath, ["-e", "setTimeout(() => {}, 30000)"], { stdio: "ignore" }); + process.stdout.write(child.pid + "\\n"); + setTimeout(() => {}, 30000); + `), + onStdout: stdout.onData, + }); + await stdout.waitFor("\n"); + const child = Number.parseInt(stdout.bytes().toString("utf8"), 10); + expect(alive(child)).toBe(true); + program.close(); + await expect(program.exited).rejects.toMatchObject({ code: "aborted" }); + expect(await waitUntilGone(program.pid)).toBe(true); + expect(await waitUntilGone(child)).toBe(true); + }); + + // On Windows the daemon's pipe ends are overlapped; a writer blocked in the standard library's write could not be + // cancelled, so it kept stdin open, still writing, for as long as a descendant held it. + it("stops writing to stdin a descendant still holds once the program has exited", async () => { + const { connection, cwd } = connect(); + const count = join(cwd, "count.txt"); + // The descendant inherits stdin and reads it only after the program exited, until its end. + const descendant = ` + setTimeout(() => { + let length = 0; + process.stdin.on("data", (chunk) => { length += chunk.length; }); + process.stdin.on("end", () => { require("node:fs").writeFileSync(${JSON.stringify(count)}, String(length)); process.exit(0); }); + }, 1500); + setTimeout(() => process.exit(1), 15000); + `; + const program = await openProcess(connection, { + cwd, + argv: node(` + require("node:child_process").spawn(process.execPath, ["-e", ${JSON.stringify(descendant)}], { stdio: ["inherit", "ignore", "ignore"], detached: true }).unref(); + setTimeout(() => process.exit(0), 300); + `), + onStdout: () => {}, + }); + // Far more than the pipe holds: the writer blocks until it is stopped. + program.write(new Uint8Array(4 * 1024 * 1024)); + expect(await program.exited).toBe(0); + const deadline = Date.now() + 8000; + while (!existsSync(count) && Date.now() < deadline) await new Promise((done) => setTimeout(done, 50)); + const read = Number(existsSync(count) ? readFileSync(count, "utf8") : Number.NaN); + expect(read).toBeLessThan(4 * 1024 * 1024); + }, 20_000); + + it("kills the program when the connection closes", async () => { + const { connection, cwd } = connect(); + const program = await openProcess(connection, { + cwd, + argv: node("setTimeout(() => {}, 30000)"), + onStdout: () => {}, + }); + connection.close(); + await expect(program.exited).rejects.toBeInstanceOf(RemoteError); + expect(await waitUntilGone(program.pid)).toBe(true); + }); +}); + +describe("failures to start", () => { + const start = (options: Partial) => { + const { connection, cwd } = connect(); + return openProcess(connection, { cwd, argv: node(""), onStdout: () => {}, ...options }); + }; + + it("fails in a missing directory", async () => { + const { cwd } = connect(); + await expect(start({ cwd: join(cwd, "missing") })).rejects.toMatchObject({ code: "spawn_error" }); + }); + + it("fails for a program that does not exist", async () => { + await expect(start({ argv: ["pi-env-no-such-program"] })).rejects.toMatchObject({ code: "spawn_error" }); + }); + + it("rejects an empty argv", async () => { + await expect(start({ argv: [] })).rejects.toMatchObject({ code: "EINVAL" }); + }); + + it("reports daemons released before the operation as not supported", async () => { + const cwd = mkdtempSync(join(tmpdir(), "pi-env-process-")); + dirs.push(cwd); + const connection = new Connection({ command: [process.execPath, join(import.meta.dirname, "old-daemon.ts")] }); + connections.push(connection); + await expect(openProcess(connection, { cwd, argv: node(""), onStdout: () => {} })).rejects.toMatchObject({ + code: "not_supported", + }); + }); +});