fix(env): kill what a process program left running when it exits

When a program started by the process operation exits on its own, the
daemon now ends its process group (Unix) or its own job object (Windows)
before reaping it, so a wrapper like npx does not leave its server
running. The pid leaves the tracked groups before it is reaped, so a
racing close cannot signal a reused pid.

packages/env/CHANGELOG.md

@@ -4,7 +4,7 @@
 
 ### 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.
+- 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, a lost connection or the program's own exit kills its process group (on Windows, its job object), so a wrapper such as `npx` that exits leaves nothing running. `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
 

packages/env/README.md

@@ -57,10 +57,11 @@ 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`.
+`onStdout` and `onStderr`, `close()` to kill its process group, and `exited` for its exit code. When the program exits,
+what it started in its process group is killed too, so a wrapper such as `npx` leaves nothing running. 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, {

packages/env/daemon/src/exec.rs

@@ -215,16 +215,39 @@ pub fn spawn(
     Ok(child)
 }
 
-/// 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<Message>) -> Result<(), (Child, std::io::Error)> {
+/// A `process` program's group, killed once the program exits, and the daemon's groups, which forget the program
+/// before it is reaped.
+pub struct Ending {
+    pub group: sys::Group,
+    pub groups: Groups,
+}
+
+/// Reap `child` on a thread of its own, which sends its exit to `sender`. With `ending`, what the child started in its
+/// group is killed once it exits, before the exit is sent. 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,
+    ending: Option<Ending>,
+    sender: Sender<Message>,
+) -> 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::<Child>();
     let started = thread::Builder::new().spawn(move || {
         let Ok(mut child) = take.recv() else {
             return;
         };
-        let _ = sender.send(match child.wait() {
+        let status = match &ending {
+            None => child.wait(),
+            Some(ending) => {
+                // After a failed wait, ending the group kills the program, so the reap below still returns.
+                let exited = ending.group.wait_for_exit(&mut child);
+                ending.group.end();
+                // Forgotten before it is reaped: a kill that checks `groups` under its lock never reaches a reused pid.
+                ending.groups.lock().unwrap().remove(&child.id());
+                exited.and(child.wait())
+            }
+        };
+        let _ = sender.send(match status {
             Ok(status) => Message::Exit(status),
             Err(error) => Message::WaitFailed(error.to_string()),
         });
@@ -392,7 +415,7 @@ pub fn run(
     if let Err(error) = spawn_reader(1, stderr, sender.clone(), stop.clone(), gate) {
         return Err(abandon(child, error));
     }
-    if let Err((child, error)) = start_waiter(child, sender.clone()) {
+    if let Err((child, error)) = start_waiter(child, None, sender.clone()) {
         return Err(abandon(child, error));
     }
     drop(sender);

packages/env/daemon/src/process.rs

@@ -4,7 +4,8 @@
 
 use crate::errors::Failure;
 use crate::exec::{
-    Control, Groups, Launch, Message, kill_group, running, spawn, spawn_reader, start_waiter,
+    Control, Ending, Groups, Launch, Message, kill_group, running, spawn, spawn_reader,
+    start_waiter,
 };
 use crate::frame::{EVENT, Frame};
 use crate::output::Output;
@@ -118,6 +119,12 @@ pub fn run(
         groups.lock().unwrap().remove(&pid);
         Failure::new("spawn_error", error.to_string())
     };
+    // What the program starts in its group dies when it exits: a wrapper such as `npx` that exits does not leave the
+    // server it started running, out of reach of `close()` and of a lost connection.
+    let group = match sys::group(&child) {
+        Ok(group) => group,
+        Err(error) => return Err(abandon(child, error)),
+    };
     // 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(
@@ -160,7 +167,14 @@ pub fn run(
     if let Err(error) = started {
         return Err(abandon(child, error));
     }
-    if let Err((child, error)) = start_waiter(child, sender.clone()) {
+    if let Err((child, error)) = start_waiter(
+        child,
+        Some(Ending {
+            group,
+            groups: groups.clone(),
+        }),
+        sender.clone(),
+    ) {
         return Err(abandon(child, error));
     }
     drop(sender);
@@ -184,7 +198,11 @@ pub fn run(
     let kill = |killed: &mut bool| {
         if !*killed {
             *killed = true;
-            kill_group(pid);
+            // Only while the program is in `groups`: the waiter removes it under this lock before reaping it.
+            let groups = groups.lock().unwrap();
+            if groups.contains(&pid) {
+                kill_group(pid);
+            }
         }
     };
     if control.killed.load(Ordering::SeqCst) {
@@ -222,6 +240,8 @@ pub fn run(
             }
             Ok(Message::Eof(stream)) => ended[stream] = true,
             Ok(Message::Exit(exit)) => {
+                // The waiter already ended the group; nothing is left to kill.
+                killed = true;
                 status = Some(Ok(exit));
                 idle_until = Some(Instant::now() + EXIT_STDIO_GRACE);
             }

packages/env/daemon/src/pty.rs

@@ -220,7 +220,7 @@ pub fn run(
     if let Err(error) = started {
         return Err(abandon(child, error));
     }
-    if let Err((child, error)) = start_waiter(child, sender.clone()) {
+    if let Err((child, error)) = start_waiter(child, None, sender.clone()) {
         return Err(abandon(child, error));
     }
     drop(sender);

packages/env/daemon/src/sys/unix.rs

@@ -354,6 +354,47 @@ pub fn kill_tree(pid: u32) {
     }
 }
 
+/// The process group of a `process` program: what the program started there is killed once it exits.
+pub struct Group {
+    pid: u32,
+}
+
+/// The group of a program just started; `configure` already made it a process group of its own.
+pub fn group(child: &Child) -> io::Result<Group> {
+    Ok(Group { pid: child.id() })
+}
+
+impl Group {
+    /// Wait until the program has exited, leaving it unreaped: its pid, which is the group's id, names no other
+    /// process until it is reaped.
+    pub fn wait_for_exit(&self, child: &mut Child) -> io::Result<()> {
+        loop {
+            // SAFETY: a zeroed siginfo_t for waitid to fill; WNOWAIT leaves the program unreaped.
+            let waited = unsafe {
+                let mut info: libc::siginfo_t = std::mem::zeroed();
+                libc::waitid(
+                    libc::P_PID,
+                    child.id() as libc::id_t,
+                    &mut info,
+                    libc::WEXITED | libc::WNOWAIT,
+                )
+            };
+            if waited == 0 {
+                return Ok(());
+            }
+            let error = io::Error::last_os_error();
+            if error.kind() != io::ErrorKind::Interrupted {
+                return Err(error);
+            }
+        }
+    }
+
+    /// Kill what is left in the group, the program too if it still runs.
+    pub fn end(&self) {
+        kill_tree(self.pid);
+    }
+}
+
 /// A process killed by a signal has no exit code; report 128 + the signal, as the shell does.
 pub fn exit_code(status: ExitStatus) -> i64 {
     i64::from(
@@ -381,3 +422,45 @@ pub fn read_dir(path: &str) -> io::Result<std::fs::ReadDir> {
 pub fn path_identity(_path: &str, metadata: &Metadata, _follow: bool) -> (u64, u64) {
     identity(metadata)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Read;
+    use std::time::{Duration, Instant};
+
+    #[test]
+    fn ending_a_group_kills_what_the_program_left_running() {
+        // A wrapper that starts a background process in its group, prints its pid and exits, as `npx` can.
+        let mut command = Command::new("/bin/sh");
+        command.stdout(Stdio::piped());
+        let script = "sleep 30 </dev/null >/dev/null 2>&1 & echo $!";
+        configure(
+            &mut command,
+            &["-c".to_string(), script.to_string()],
+            &Map::new(),
+            true,
+        );
+        let mut child = command.spawn().unwrap();
+        let group = group(&child).unwrap();
+        let mut stdout = String::new();
+        child
+            .stdout
+            .take()
+            .unwrap()
+            .read_to_string(&mut stdout)
+            .unwrap();
+        let sleeper: i32 = stdout.trim().parse().unwrap();
+        group.wait_for_exit(&mut child).unwrap();
+        group.end();
+        assert!(child.wait().unwrap().success());
+        // The killed sleeper is an orphan; it is gone once init has reaped it.
+        let deadline = Instant::now() + Duration::from_secs(5);
+        // SAFETY: signal 0 only checks that the process exists.
+        while unsafe { libc::kill(sleeper, 0) } == 0 && Instant::now() < deadline {
+            std::thread::sleep(Duration::from_millis(20));
+        }
+        // SAFETY: as above.
+        assert_ne!(unsafe { libc::kill(sleeper, 0) }, 0);
+    }
+}

packages/env/daemon/src/sys/windows.rs

@@ -33,7 +33,7 @@ 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,
+    JobObjectExtendedLimitInformation, SetInformationJobObject, TerminateJobObject,
 };
 use windows_sys::Win32::System::Threading::{CREATE_NO_WINDOW, CreateEventW, WaitForSingleObject};
 use windows_sys::core::BOOL;
@@ -685,6 +685,57 @@ pub fn kill_tree(pid: u32) {
         .spawn();
 }
 
+/// The job object of a `process` program: what is left in it is killed once the program exits, and when the daemon
+/// ends (the job is closed with it). It is nested in the daemon's job.
+pub struct Group {
+    job: OwnedHandle,
+}
+
+/// Put a program just started into a job object of its own. The program already runs: a child it starts before this
+/// assignment stays outside the job, as one started before `after_spawn` stays outside the daemon's job. Starting it
+/// suspended would close that gap.
+pub fn group(child: &Child) -> io::Result<Group> {
+    // SAFETY: plain Win32 calls with a zeroed, correctly sized information struct; the job handle is owned from here.
+    unsafe {
+        let job = CreateJobObjectW(std::ptr::null(), std::ptr::null());
+        if job.is_null() {
+            return Err(io::Error::last_os_error());
+        }
+        let job = OwnedHandle::from_raw_handle(job as _);
+        let mut information: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = std::mem::zeroed();
+        // No breakaway: a child started after the assignment stays in this job, even if the daemon's job allows
+        // breakaway.
+        information.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
+        if SetInformationJobObject(
+            job.as_raw_handle() as _,
+            JobObjectExtendedLimitInformation,
+            (&information as *const JOBOBJECT_EXTENDED_LIMIT_INFORMATION).cast(),
+            std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
+        ) == FALSE
+            || AssignProcessToJobObject(job.as_raw_handle() as _, child.as_raw_handle() as _)
+                == FALSE
+        {
+            return Err(io::Error::last_os_error());
+        }
+        Ok(Group { job })
+    }
+}
+
+impl Group {
+    /// Wait until the program has exited; the child's open handle keeps its pid from naming another process.
+    pub fn wait_for_exit(&self, child: &mut Child) -> io::Result<()> {
+        child.wait().map(drop)
+    }
+
+    /// Kill what is left in the job, the program too if it still runs.
+    pub fn end(&self) {
+        // SAFETY: the job handle is valid while `self` lives.
+        unsafe {
+            TerminateJobObject(self.job.as_raw_handle() as _, 1);
+        }
+    }
+}
+
 /// Exit codes are unsigned 32-bit values on Windows (an access violation is 3221225477).
 pub fn exit_code(status: ExitStatus) -> i64 {
     i64::from(status.code().unwrap_or(1) as u32)

packages/env/docs/protocol.md

@@ -64,9 +64,10 @@ command as a full pipe would.
 Each side sends `ping` every 5 seconds. The daemon kills every process group it started and exits after 30 seconds
 without receiving any bytes, when stdin ends, or on `SIGTERM`, `SIGINT` or `SIGHUP`. A dropped connection or a
 stopped daemon therefore stops remote commands. On Windows, a terminated daemon cannot clean up: a job object kills
-each command's own process, but not processes that one started. The client fails requests of a lost connection with
-`{ code: "unknown", lost: true }` and starts a new daemon on the next request; handles belong to the daemon that opened
-them, so requests on them fail instead of reaching the new one.
+each command's own process, but not processes that one started, except for `process` programs, whose own job object
+holds what they start (see `process` for the gap right after the start). The client fails requests of a lost
+connection with `{ code: "unknown", lost: true }` and starts a new daemon on the next request; handles belong to the
+daemon that opened them, so requests on them fail instead of reaching the new one.
 
 ## Operations
 
@@ -155,11 +156,15 @@ build) from the prompt; background jobs are not reported. It fails with `EBADF`
 `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`.
+environment rules), and on Windows also in a job object of its own. When the program exits, whatever it started in its
+process group (on Windows, its job object) is killed: a wrapper such as `npx` that exits does not leave the server it
+started running. On Unix, a process that moved to a process group or session of its own (`setpgid`, `setsid`) has left
+the group and keeps running. On Windows, the program runs before the daemon assigns it to its job objects: a child it
+starts in that moment stays outside them and keeps running; children started later cannot leave the job. 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`

packages/env/src/process.ts

@@ -19,7 +19,9 @@ export interface ProcessOptions {
 
 /**
  * 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).
+ * ends when the program exits, `close()` kills it, or the connection is lost (the daemon then kills it). When the
+ * program exits, the daemon kills whatever it started in its process group, so a wrapper such as `npx` that exits
+ * leaves nothing running.
  */
 export class RemoteProcess {
 	readonly pid: number;

packages/env/test/process.test.ts

@@ -246,13 +246,36 @@ describe("ending", () => {
 		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 () => {
+	it("kills what the program started when it exits", 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 = `
+		const stdout = collector();
+		const program = await openProcess(connection, {
+			cwd,
+			// A wrapper that exits and leaves the server it started running, as npx can once stdin closes. On Windows the
+			// server is detached, so libuv's own job object does not take it down with the wrapper: only the daemon's can.
+			argv: node(`
+				const child = require("node:child_process").spawn(process.execPath, ["-e", "setTimeout(() => {}, 30000)"], { stdio: "ignore", detached: ${process.platform === "win32"} });
+				process.stdout.write(child.pid + "\\n", () => process.exit(0));
+			`),
+			onStdout: stdout.onData,
+		});
+		expect(await program.exited).toBe(0);
+		const child = Number.parseInt(stdout.bytes().toString("utf8"), 10);
+		expect(Number.isInteger(child)).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. There a descendant now dies
+	// with the program's job object; only one started before the program was assigned to the job (see the protocol)
+	// can still hold stdin, which a test cannot arrange reliably. On Unix the descendant leaves the process group.
+	it.skipIf(process.platform === "win32")(
+		"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; });
@@ -260,22 +283,24 @@ describe("ending", () => {
 			}, 1500);
 			setTimeout(() => process.exit(1), 15000);
 		`;
-		const program = await openProcess(connection, {
-			cwd,
-			argv: node(`
+			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);
+				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();