From 5ac6b1fbe59caeb55f0289ed63a5166c35da86ca Mon Sep 17 00:00:00 2001 From: Luke Hoban Date: Thu, 6 Aug 2026 22:41:36 -0700 Subject: [PATCH 1/5] fix(rust): reap spawned process trees Bind each spawned CLI transport to an SDK-owned process tree before it can create descendants. Use a kill-on-close Job Object on Windows and a process group on Unix, and carry the RAII owner through startup, stop, force-stop, and drop paths. Cover grandchild teardown and startup-failure cleanup without changing the public client API. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1090dd5b-e149-4d9d-9cc3-67f26e11ad06 --- rust/Cargo.lock | 2 + rust/Cargo.toml | 10 + rust/src/errors.rs | 12 +- rust/src/lib.rs | 322 ++++++++++++++++---- rust/src/process_tree.rs | 628 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 914 insertions(+), 60 deletions(-) create mode 100644 rust/src/process_tree.rs diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 8de679798..b4445c0b6 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -434,6 +434,7 @@ dependencies = [ "getrandom 0.2.17", "http", "indexmap", + "libc", "libloading", "native-tls", "parking_lot", @@ -454,6 +455,7 @@ dependencies = [ "tracing", "ureq", "uuid", + "windows-sys 0.61.2", "zip", ] diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 0f18a9b15..3fdee0f2e 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -69,8 +69,18 @@ reqwest = { version = "0.12", default-features = false, features = ["stream", "h tokio-tungstenite = { version = "0.24", default-features = false, features = ["connect", "native-tls"] } [target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_System_Diagnostics_ToolHelp", + "Win32_System_JobObjects", + "Win32_System_Threading", +] } zip = { version = "2", default-features = false, features = ["deflate"], optional = true } +[target.'cfg(unix)'.dependencies] +libc = "0.2" + [dev-dependencies] rusqlite = { version = "0.35", features = ["bundled"] } schemars = "1" diff --git a/rust/src/errors.rs b/rust/src/errors.rs index 6e05bbfae..ddd57d854 100644 --- a/rust/src/errors.rs +++ b/rust/src/errors.rs @@ -396,11 +396,11 @@ fn capture_backtrace() -> Option> { /// Aggregate of errors collected during [`crate::Client::stop`]. /// /// `Client::stop` performs cooperative shutdown across every active -/// session before killing the CLI child process. Errors from any -/// per-session `session.destroy` RPC and from the terminal child-kill -/// step are collected here rather than short-circuiting on the first -/// failure, so callers see the full picture of what went wrong during -/// teardown. +/// session before terminating and reaping the SDK-owned CLI process tree. +/// Errors from any per-session `session.destroy` RPC and from the terminal +/// process-tree teardown are collected here rather than short-circuiting on +/// the first failure, so callers see the full picture of what went wrong +/// during teardown. /// /// Implements [`std::error::Error`] and forwards to `Display` for the /// first error, with a count suffix when there are more. @@ -409,7 +409,7 @@ pub struct StopErrors(pub(crate) Vec); impl StopErrors { /// Borrow the collected errors as a slice, in the order they - /// occurred (per-session destroys first, then child-kill last). + /// occurred (per-session destroys first, then process-tree teardown). pub fn errors(&self) -> &[Error] { &self.0 } diff --git a/rust/src/lib.rs b/rust/src/lib.rs index cafa3c596..12bf6c563 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -29,6 +29,7 @@ pub mod hooks; mod jsonrpc; /// Permission-policy helpers that produce a [`handler::PermissionHandler`]. pub mod permission; +mod process_tree; /// BYOK bearer-token provider callbacks. pub mod provider_token; mod provider_token_dispatch; @@ -101,7 +102,7 @@ pub mod test_support { use serde::{Deserialize, Serialize}; use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader}; use tokio::net::TcpStream; -use tokio::process::{Child, Command}; +use tokio::process::Command; use tokio::sync::{broadcast, mpsc, oneshot}; use tracing::{Instrument, debug, error, info, warn}; pub use types::*; @@ -971,7 +972,7 @@ fn validate_inprocess_options(options: &ClientOptions) -> Result<()> { /// Connection to a GitHub Copilot CLI server (stdio, TCP, or external). /// /// Cheaply cloneable — cloning shares the underlying connection. -/// The child process (if any) is killed when the last clone drops. +/// The SDK-owned process tree (if any) is terminated when the last clone drops. #[derive(Clone)] pub struct Client { inner: Arc, @@ -987,7 +988,7 @@ impl std::fmt::Debug for Client { } struct ClientInner { - child: parking_lot::Mutex>, + child: parking_lot::Mutex>, #[cfg(feature = "bundled-in-process")] /// In-process FFI runtime host, set only for [`Transport::InProcess`]. /// Closing it tears down the native runtime connection. @@ -1241,8 +1242,8 @@ impl Client { let (mut child, spawn_elapsed) = Self::spawn_stdio(&program, &options, &working_directory)?; timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed)); - let stdin = child.stdin.take().expect("stdin is piped"); - let stdout = child.stdout.take().expect("stdout is piped"); + let stdin = child.child_mut().stdin.take().expect("stdin is piped"); + let stdout = child.child_mut().stdout.take().expect("stdout is piped"); Self::drain_stderr(&mut child); Self::from_transport( stdout, @@ -1525,7 +1526,7 @@ impl Client { fn from_transport( reader: impl AsyncRead + Unpin + Send + 'static, writer: impl AsyncWrite + Unpin + Send + 'static, - child: Option, + child: Option, cwd: PathBuf, on_list_models: Option>, session_fs_configured: bool, @@ -1586,8 +1587,8 @@ impl Client { /// notifications via [`ClientInner::lifecycle_tx`] to subscribers /// returned by [`Self::subscribe_lifecycle`]. fn spawn_lifecycle_dispatcher(&self) { - let inner = Arc::clone(&self.inner); - let mut notif_rx = inner.notification_tx.subscribe(); + let mut notif_rx = self.inner.notification_tx.subscribe(); + let lifecycle_tx = self.inner.lifecycle_tx.clone(); tokio::spawn(async move { loop { match notif_rx.recv().await { @@ -1611,7 +1612,7 @@ impl Client { }; // `send` only errors when there are no subscribers — that's // the normal case before any consumer calls subscribe_lifecycle. - let _ = inner.lifecycle_tx.send(event); + let _ = lifecycle_tx.send(event); } Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { warn!(missed = n, "lifecycle dispatcher lagged"); @@ -1684,13 +1685,6 @@ impl Client { .stdout(Stdio::piped()) .stderr(Stdio::piped()); - #[cfg(windows)] - { - use std::os::windows::process::CommandExt; - const CREATE_NO_WINDOW: u32 = 0x08000000; - command.as_std_mut().creation_flags(CREATE_NO_WINDOW); - } - command } @@ -1747,7 +1741,7 @@ impl Client { program: &Path, options: &ClientOptions, working_directory: &Path, - ) -> Result<(Child, Duration)> { + ) -> Result<(process_tree::ManagedChild, Duration)> { info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)"); let mut command = Self::build_command(program, options, working_directory); command @@ -1759,7 +1753,7 @@ impl Client { .args(&options.extra_args) .stdin(Stdio::piped()); let spawn_start = Instant::now(); - let child = command.spawn()?; + let child = process_tree::ManagedChild::spawn(command)?; let spawn_elapsed = spawn_start.elapsed(); debug!( elapsed_ms = spawn_elapsed.as_millis(), @@ -1773,7 +1767,7 @@ impl Client { options: &ClientOptions, working_directory: &Path, port: u16, - ) -> Result<(Child, u16, Duration, Duration)> { + ) -> Result<(process_tree::ManagedChild, u16, Duration, Duration)> { info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)"); let mut command = Self::build_command(program, options, working_directory); command @@ -1785,13 +1779,13 @@ impl Client { .args(&options.extra_args) .stdin(Stdio::null()); let spawn_start = Instant::now(); - let mut child = command.spawn()?; + let mut child = process_tree::ManagedChild::spawn(command)?; let spawn_elapsed = spawn_start.elapsed(); debug!( elapsed_ms = spawn_elapsed.as_millis(), "Client::spawn_tcp subprocess spawned" ); - let stdout = child.stdout.take().expect("stdout is piped"); + let stdout = child.child_mut().stdout.take().expect("stdout is piped"); let (port_tx, port_rx) = oneshot::channel::(); let span = tracing::error_span!("copilot_cli_port_scan"); @@ -1835,8 +1829,8 @@ impl Client { Ok((child, actual_port, spawn_elapsed, port_wait_elapsed)) } - fn drain_stderr(child: &mut Child) { - if let Some(stderr) = child.stderr.take() { + fn drain_stderr(child: &mut process_tree::ManagedChild) { + if let Some(stderr) = child.child_mut().stderr.take() { let span = tracing::error_span!("copilot_cli"); tokio::spawn( async move { @@ -2342,21 +2336,26 @@ impl Client { /// Return the OS process ID of the CLI child process, if one was spawned. pub fn pid(&self) -> Option { - self.inner.child.lock().as_ref().and_then(|c| c.id()) + self.inner + .child + .lock() + .as_ref() + .and_then(process_tree::ManagedChild::id) } - /// Cooperatively shut down the client and the CLI child process. + /// Cooperatively shut down the client and its SDK-owned process tree. /// /// Walks every still-registered session and sends `session.destroy` - /// for each one, asks SDK-owned runtimes to shut down, then kills the - /// CLI child. Errors from per-session destroys, runtime shutdown, and - /// the final child-kill are collected into + /// for each one, asks SDK-owned runtimes to shut down, then terminates and + /// reaps the complete spawned process tree. Errors from per-session + /// destroys, runtime shutdown, tree termination, and process reaping are + /// collected into /// [`StopErrors`] rather than short-circuiting on the first failure /// — so callers see the full picture of teardown. /// /// If you have already called [`Session::disconnect`] on every /// session this client created, the per-session destroy step is a - /// no-op (the router map is empty); only the child-kill remains. + /// no-op (the router map is empty); only process-tree teardown remains. /// /// [`Session::disconnect`]: crate::session::Session::disconnect /// @@ -2442,20 +2441,17 @@ impl Client { *self.inner.state.lock() = ConnectionState::Disconnected; *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new()); if let Some(mut child) = child { - match child.try_wait() { - Ok(Some(_status)) => {} - Ok(None) => { - // The runtime completes all cleanup before responding to - // runtime.shutdown and then leaves termination to us; it - // deliberately keeps its JSON-RPC server alive to send the - // response and never self-exits. Waiting for a self-exit - // that will never come just wastes time, so terminate the - // child immediately. - if let Err(e) = child.kill().await { - errors.push(e.into()); - } - } - Err(e) => errors.push(e.into()), + // The runtime completes all cleanup before responding to + // runtime.shutdown and deliberately leaves process termination to + // its owner so it can send the response first. + if let Err(e) = child.terminate() { + errors.push(e.into()); + } + if let Err(e) = child.wait().await { + errors.push(e.into()); + } + if let Err(e) = child.wait_for_tree_exit(RUNTIME_SHUTDOWN_TIMEOUT).await { + errors.push(e.into()); } } @@ -2477,14 +2473,14 @@ impl Client { } } - /// Forcibly stop the CLI process without waiting for it to exit. + /// Forcibly stop the CLI process tree. /// /// Synchronous fallback when [`stop`](Self::stop) is unsuitable — for /// example when the awaiting tokio runtime is shutting down or the - /// process is wedged on I/O. Sends a kill signal without awaiting - /// reaper completion and immediately drops all per-session router - /// state so dependent tasks observe a closed channel rather than a - /// hang. + /// process is wedged on I/O. Terminates the complete tree, briefly polls + /// the direct child for exit, and transfers any slow exit to a dedicated + /// finite reaper thread. It immediately drops all per-session router state + /// so dependent tasks observe a closed channel rather than a hang. /// /// # Cancel safety /// @@ -2510,9 +2506,9 @@ impl Client { let pid = self.pid(); info!(pid = ?pid, "force-stopping CLI process"); if let Some(mut child) = self.inner.child.lock().take() - && let Err(e) = child.start_kill() + && let Err(e) = child.terminate() { - error!(pid = ?pid, error = %e, "failed to send kill signal"); + error!(pid = ?pid, error = %e, "failed to terminate CLI process tree"); } self.inner.rpc.force_close(); #[cfg(feature = "bundled-in-process")] @@ -2569,12 +2565,12 @@ impl Client { impl Drop for ClientInner { fn drop(&mut self) { - if let Some(ref mut child) = *self.child.lock() { + if let Some(mut child) = self.child.lock().take() { let pid = child.id(); - if let Err(e) = child.start_kill() { - error!(pid = ?pid, error = %e, "failed to kill CLI process on drop"); + if let Err(e) = child.terminate() { + error!(pid = ?pid, error = %e, "failed to terminate CLI process tree on drop"); } else { - info!(pid = ?pid, "kill signal sent for CLI process on drop"); + info!(pid = ?pid, "CLI process tree terminated on drop"); } } #[cfg(feature = "bundled-in-process")] @@ -2589,6 +2585,13 @@ impl Drop for ClientInner { #[cfg(test)] mod tests { + #[cfg(unix)] + use std::os::unix::fs::PermissionsExt; + + use serial_test::serial; + use tempfile::{TempDir, tempdir}; + use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; + use super::*; #[test] @@ -3215,6 +3218,217 @@ mod tests { client.force_stop(); } + #[tokio::test] + #[serial] + async fn client_process_tree_stop_reaps_grandchild() { + let baseline = process_tree::active_tree_count(); + let (client, direct_pid, grandchild_pid, _temp, mut server_read, mut server_write) = + managed_test_client().await; + let server = tokio::spawn(async move { + let request = read_framed_json(&mut server_read).await; + assert_eq!(request["method"], "runtime.shutdown"); + let response = serde_json::json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": null, + }); + write_framed_json(&mut server_write, &response).await; + }); + + client.stop().await.expect("stop client"); + server.await.expect("runtime shutdown server"); + + assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; + } + + #[tokio::test] + #[serial] + async fn client_process_tree_force_stop_reaps_grandchild() { + let baseline = process_tree::active_tree_count(); + let (client, direct_pid, grandchild_pid, _temp, _server_read, _server_write) = + managed_test_client().await; + + client.force_stop(); + + assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; + } + + #[tokio::test] + #[serial] + async fn client_process_tree_drop_reaps_grandchild() { + let baseline = process_tree::active_tree_count(); + let (client, direct_pid, grandchild_pid, _temp, _server_read, _server_write) = + managed_test_client().await; + + drop(client); + + assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; + } + + #[cfg(unix)] + #[tokio::test] + #[serial] + async fn client_process_tree_tcp_startup_failure_reaps_tree() { + let baseline = process_tree::active_tree_count(); + let temp = tempdir().expect("create temp directory"); + let pid_file = temp.path().join("startup.pids"); + let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("reserve port"); + let unused_port = listener.local_addr().expect("reserved address").port(); + drop(listener); + let script = executable_script( + &temp, + "fake-tcp-cli.sh", + "#!/bin/sh\nsleep 60 /dev/null 2>&1 &\necho \"$$ $!\" > \"$PID_FILE\"\necho \"listening on port $FAKE_PORT\"\nwait\n", + ); + let options = ClientOptions::new() + .with_program(CliProgram::Path(script)) + .with_transport(Transport::Tcp { + port: 0, + connection_token: None, + }) + .with_env([ + ("PID_FILE", pid_file.as_os_str()), + ("FAKE_PORT", std::ffi::OsStr::new(&unused_port.to_string())), + ]); + + Client::start(options) + .await + .expect_err("TCP connect should fail"); + let (direct_pid, grandchild_pid) = read_test_pids(&pid_file).await; + + assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; + } + + #[cfg(unix)] + #[tokio::test] + #[serial] + async fn client_process_tree_handshake_failure_reaps_tree() { + let baseline = process_tree::active_tree_count(); + let temp = tempdir().expect("create temp directory"); + let pid_file = temp.path().join("startup.pids"); + let script = executable_script( + &temp, + "fake-stdio-cli.sh", + "#!/bin/sh\nsleep 60 /dev/null 2>&1 &\necho \"$$ $!\" > \"$PID_FILE\"\nsleep 0.1\nexit 0\n", + ); + let options = ClientOptions::new() + .with_program(CliProgram::Path(script)) + .with_env([("PID_FILE", pid_file.as_os_str())]); + + Client::start(options) + .await + .expect_err("protocol handshake should fail"); + let (direct_pid, grandchild_pid) = read_test_pids(&pid_file).await; + + assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; + } + + async fn managed_test_client() -> (Client, u32, u32, TempDir, DuplexStream, DuplexStream) { + let temp = tempdir().expect("create temp directory"); + let pid_file = temp.path().join("grandchild.pid"); + let child = process_tree::ManagedChild::spawn(process_tree::test_tree_command(&pid_file)) + .expect("spawn managed process tree"); + let direct_pid = child.id().expect("direct child pid"); + let grandchild_pid = process_tree::wait_for_test_pid(&pid_file).await; + let (client_write, server_read) = tokio::io::duplex(8192); + let (server_write, client_read) = tokio::io::duplex(8192); + let client = Client::from_transport( + client_read, + client_write, + Some(child), + temp.path().to_path_buf(), + None, + false, + false, + None, + None, + None, + ClientMode::default(), + ) + .expect("create client"); + ( + client, + direct_pid, + grandchild_pid, + temp, + server_read, + server_write, + ) + } + + async fn assert_test_tree_gone(direct_pid: u32, grandchild_pid: u32, baseline: usize) { + assert!( + process_tree::wait_for_test_condition(Duration::from_secs(10), || { + !process_tree::test_process_exists(direct_pid) + && !process_tree::test_process_exists(grandchild_pid) + && process_tree::active_tree_count() == baseline + }) + .await, + "managed process tree or guard survived teardown" + ); + } + + #[cfg(unix)] + fn executable_script(temp: &TempDir, name: &str, contents: &str) -> PathBuf { + let path = temp.path().join(name); + std::fs::write(&path, contents).expect("write test script"); + let mut permissions = std::fs::metadata(&path) + .expect("read test script metadata") + .permissions(); + permissions.set_mode(0o755); + std::fs::set_permissions(&path, permissions).expect("make test script executable"); + path + } + + #[cfg(unix)] + async fn read_test_pids(path: &Path) -> (u32, u32) { + let found = + process_tree::wait_for_test_condition(Duration::from_secs(10), || path.exists()).await; + assert!(found, "startup helper pid file was not created"); + let contents = std::fs::read_to_string(path).expect("read startup helper pids"); + let mut pids = contents.split_whitespace().map(|value| { + value + .parse::() + .expect("startup helper pid should be numeric") + }); + ( + pids.next().expect("direct child pid"), + pids.next().expect("grandchild pid"), + ) + } + + async fn read_framed_json(reader: &mut DuplexStream) -> serde_json::Value { + let mut header = Vec::new(); + while !header.ends_with(b"\r\n\r\n") { + let mut byte = [0u8; 1]; + reader + .read_exact(&mut byte) + .await + .expect("read frame header"); + header.push(byte[0]); + } + let header = String::from_utf8(header).expect("frame header is UTF-8"); + let length = header + .lines() + .find_map(|line| line.strip_prefix("Content-Length: ")) + .expect("content length header") + .parse::() + .expect("content length is numeric"); + let mut body = vec![0u8; length]; + reader.read_exact(&mut body).await.expect("read frame body"); + serde_json::from_slice(&body).expect("parse framed JSON") + } + + async fn write_framed_json(writer: &mut DuplexStream, value: &serde_json::Value) { + let body = serde_json::to_vec(value).expect("serialize framed JSON"); + writer + .write_all(format!("Content-Length: {}\r\n\r\n", body.len()).as_bytes()) + .await + .expect("write frame header"); + writer.write_all(&body).await.expect("write frame body"); + writer.flush().await.expect("flush frame"); + } + fn client_with_list_models_handler(handler: Arc) -> Client { Client { inner: Arc::new(ClientInner { diff --git a/rust/src/process_tree.rs b/rust/src/process_tree.rs new file mode 100644 index 000000000..01d0a7e3e --- /dev/null +++ b/rust/src/process_tree.rs @@ -0,0 +1,628 @@ +//! Ownership and teardown for SDK-spawned process trees. + +use std::io; +use std::process::ExitStatus; +use std::time::{Duration, Instant}; + +use tokio::process::{Child, Command}; +use tracing::{error, warn}; + +const TREE_EXIT_POLL_INTERVAL: Duration = Duration::from_millis(10); +const SYNC_REAP_GRACE: Duration = Duration::from_millis(250); + +/// Owns a direct child and the platform primitive that contains its descendants. +pub(crate) struct ManagedChild { + child: Option, + tree: Option, + tree_terminated: bool, +} + +impl ManagedChild { + /// Spawn a child into a process tree before it can create descendants. + pub(crate) fn spawn(mut command: Command) -> io::Result { + command.kill_on_drop(true); + platform::configure_command(&mut command); + + let mut child = command.spawn()?; + match platform::ProcessTree::attach_and_start(&mut child) { + Ok(tree) => Ok(Self { + child: Some(child), + tree: Some(tree), + tree_terminated: false, + }), + Err(error) => { + reap_failed_spawn(&mut child); + Err(error) + } + } + } + + pub(crate) fn child_mut(&mut self) -> &mut Child { + self.child.as_mut().expect("managed child is present") + } + + pub(crate) fn id(&self) -> Option { + self.child.as_ref().and_then(Child::id) + } + + /// Terminate the complete tree. If the tree primitive fails, still signal + /// the direct child so teardown never regresses to doing nothing. + pub(crate) fn terminate(&mut self) -> io::Result<()> { + if self.tree_terminated { + return Ok(()); + } + let result = self + .tree + .as_ref() + .expect("managed process tree is present") + .terminate(); + if result.is_ok() { + self.tree_terminated = true; + } + if result.is_err() + && let Some(child) = self.child.as_mut() + { + let _ = child.start_kill(); + } + result + } + + /// Wait for and reap the direct child through Tokio's sole child owner. + pub(crate) async fn wait(&mut self) -> io::Result { + self.child_mut().wait().await + } + + /// Verify that no process remains in the platform tree. + pub(crate) async fn wait_for_tree_exit(&mut self, timeout: Duration) -> io::Result<()> { + let started = Instant::now(); + loop { + let tree = self.tree.as_ref().expect("managed process tree is present"); + tree.reap_adopted()?; + if tree.is_empty()? { + self.tree.take(); + return Ok(()); + } + if started.elapsed() >= timeout { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "timed out waiting for CLI process tree to exit", + )); + } + tokio::time::sleep(TREE_EXIT_POLL_INTERVAL).await; + } + } +} + +impl Drop for ManagedChild { + fn drop(&mut self) { + let Some(mut child) = self.child.take() else { + return; + }; + let tree = self.tree.take(); + let pid = child.id(); + + if !self.tree_terminated + && let Some(tree) = tree.as_ref() + && let Err(error) = tree.terminate() + { + warn!(pid = ?pid, %error, "failed to terminate CLI process tree on drop"); + } + if let Err(error) = child.start_kill() + && child.try_wait().ok().flatten().is_none() + { + warn!(pid = ?pid, %error, "failed to terminate direct CLI child on drop"); + } + + if reap_for(&mut child, SYNC_REAP_GRACE) { + return; + } + + let result = std::thread::Builder::new() + .name("copilot-cli-reaper".to_string()) + .spawn(move || { + loop { + match child.try_wait() { + Ok(Some(_)) => break, + Ok(None) => std::thread::sleep(TREE_EXIT_POLL_INTERVAL), + Err(error) => { + warn!(pid = ?pid, %error, "failed to reap CLI child"); + break; + } + } + } + drop(tree); + }); + if let Err(error) = result { + error!(pid = ?pid, %error, "failed to start CLI child reaper thread"); + } + } +} + +fn reap_failed_spawn(child: &mut Child) { + let pid = child.id(); + if let Err(error) = child.start_kill() { + warn!(pid = ?pid, %error, "failed to terminate CLI after process-tree setup failure"); + } + if !reap_for(child, SYNC_REAP_GRACE) { + warn!(pid = ?pid, "CLI did not exit promptly after process-tree setup failure"); + } +} + +fn reap_for(child: &mut Child, timeout: Duration) -> bool { + let started = Instant::now(); + loop { + match child.try_wait() { + Ok(Some(_)) => return true, + Ok(None) if started.elapsed() < timeout => { + std::thread::sleep(TREE_EXIT_POLL_INTERVAL); + } + Ok(None) | Err(_) => return false, + } + } +} + +#[cfg(test)] +pub(crate) fn active_tree_count() -> usize { + platform::active_tree_count() +} + +#[cfg(test)] +pub(crate) async fn wait_for_test_pid(path: &std::path::Path) -> u32 { + let found = wait_for_test_condition(Duration::from_secs(10), || path.exists()).await; + assert!(found, "grandchild pid file was not created"); + std::fs::read_to_string(path) + .expect("read grandchild pid") + .trim() + .parse() + .expect("parse grandchild pid") +} + +#[cfg(test)] +pub(crate) async fn wait_for_test_condition( + timeout: Duration, + mut predicate: impl FnMut() -> bool, +) -> bool { + let started = Instant::now(); + loop { + if predicate() { + return true; + } + if started.elapsed() >= timeout { + return false; + } + tokio::time::sleep(TREE_EXIT_POLL_INTERVAL).await; + } +} + +#[cfg(all(test, unix))] +pub(crate) fn test_tree_command(pid_file: &std::path::Path) -> Command { + let mut command = Command::new("sh"); + command + .args(["-c", "sleep 60 & echo \"$!\" > \"$PID_FILE\"; wait"]) + .env("PID_FILE", pid_file) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()); + command +} + +#[cfg(all(test, windows))] +pub(crate) fn test_tree_command(pid_file: &std::path::Path) -> Command { + let script = concat!( + "$child = Start-Process powershell.exe ", + "-ArgumentList @('-NoLogo','-NoProfile','-NonInteractive','-Command',", + "'Start-Sleep -Seconds 60') -PassThru; ", + "Set-Content -LiteralPath $env:PID_FILE -Value $child.Id; ", + "Wait-Process -Id $child.Id" + ); + let mut command = Command::new("powershell.exe"); + command + .args([ + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + script, + ]) + .env("PID_FILE", pid_file) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()); + command +} + +#[cfg(all(test, unix))] +pub(crate) fn test_process_exists(pid: u32) -> bool { + // SAFETY: signal 0 only probes process existence. + (unsafe { libc::kill(pid as i32, 0) }) == 0 +} + +#[cfg(all(test, windows))] +pub(crate) fn test_process_exists(pid: u32) -> bool { + use windows_sys::Win32::Foundation::{CloseHandle, WAIT_TIMEOUT}; + use windows_sys::Win32::System::Threading::{ + OpenProcess, PROCESS_SYNCHRONIZE, WaitForSingleObject, + }; + + // SAFETY: the process handle is closed before returning. + unsafe { + let process = OpenProcess(PROCESS_SYNCHRONIZE, 0, pid); + if process.is_null() { + return false; + } + let exists = WaitForSingleObject(process, 0) == WAIT_TIMEOUT; + CloseHandle(process); + exists + } +} + +#[cfg(unix)] +mod platform { + use std::io; + #[cfg(test)] + use std::sync::atomic::{AtomicUsize, Ordering}; + + use tokio::process::{Child, Command}; + + #[cfg(test)] + static ACTIVE_TREES: AtomicUsize = AtomicUsize::new(0); + + pub(super) struct ProcessTree { + pgid: i32, + } + + impl ProcessTree { + pub(super) fn attach_and_start(child: &mut Child) -> io::Result { + let pid = child.id().ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + "CLI exited before process-group ownership was established", + ) + })?; + #[cfg(test)] + ACTIVE_TREES.fetch_add(1, Ordering::Relaxed); + Ok(Self { pgid: pid as i32 }) + } + + pub(super) fn terminate(&self) -> io::Result<()> { + // SAFETY: `pgid` is the dedicated group created for this child. + if unsafe { libc::killpg(self.pgid, libc::SIGKILL) } == 0 { + return Ok(()); + } + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(libc::ESRCH) { + Ok(()) + } else { + Err(error) + } + } + + pub(super) fn is_empty(&self) -> io::Result { + // SAFETY: signal 0 only probes the dedicated process group. + if unsafe { libc::killpg(self.pgid, 0) } == 0 { + return Ok(false); + } + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(libc::ESRCH) { + Ok(true) + } else { + Err(error) + } + } + + pub(super) fn reap_adopted(&self) -> io::Result<()> { + loop { + let mut status = 0; + // SAFETY: a negative pid selects children in this dedicated + // process group. WNOHANG keeps the async caller non-blocking. + let result = unsafe { libc::waitpid(-self.pgid, &mut status, libc::WNOHANG) }; + if result > 0 { + continue; + } + if result == 0 { + return Ok(()); + } + let error = io::Error::last_os_error(); + match error.raw_os_error() { + Some(libc::ECHILD) => return Ok(()), + Some(libc::EINTR) => continue, + _ => return Err(error), + } + } + } + } + + impl Drop for ProcessTree { + fn drop(&mut self) { + #[cfg(test)] + ACTIVE_TREES.fetch_sub(1, Ordering::Relaxed); + } + } + + pub(super) fn configure_command(command: &mut Command) { + command.process_group(0); + } + + #[cfg(test)] + pub(super) fn active_tree_count() -> usize { + ACTIVE_TREES.load(Ordering::Relaxed) + } +} + +#[cfg(windows)] +mod platform { + use std::io; + use std::ptr; + #[cfg(test)] + use std::sync::atomic::{AtomicUsize, Ordering}; + + use tokio::process::{Child, Command}; + use windows_sys::Win32::Foundation::{CloseHandle, HANDLE, INVALID_HANDLE_VALUE}; + use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, TH32CS_SNAPTHREAD, THREADENTRY32, Thread32First, Thread32Next, + }; + use windows_sys::Win32::System::JobObjects::{ + AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, + JOBOBJECT_BASIC_ACCOUNTING_INFORMATION, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, + JobObjectBasicAccountingInformation, JobObjectExtendedLimitInformation, + QueryInformationJobObject, SetInformationJobObject, TerminateJobObject, + }; + use windows_sys::Win32::System::Threading::{ + CREATE_NO_WINDOW, CREATE_SUSPENDED, OpenThread, ResumeThread, THREAD_SUSPEND_RESUME, + }; + + #[cfg(test)] + static ACTIVE_TREES: AtomicUsize = AtomicUsize::new(0); + + struct OwnedHandle(HANDLE); + + // SAFETY: Windows kernel handles can be used and closed from any thread. + unsafe impl Send for OwnedHandle {} + + impl Drop for OwnedHandle { + fn drop(&mut self) { + // SAFETY: this type owns the valid handle and closes it exactly once. + unsafe { + CloseHandle(self.0); + } + } + } + + pub(super) struct ProcessTree { + job: OwnedHandle, + } + + impl ProcessTree { + pub(super) fn attach_and_start(child: &mut Child) -> io::Result { + let raw_process = child.raw_handle().ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + "CLI exited before Job Object ownership was established", + ) + })?; + + // SAFETY: null attributes and name create a private, non-inheritable Job Object. + let raw_job = unsafe { CreateJobObjectW(ptr::null(), ptr::null()) }; + if raw_job.is_null() { + return Err(io::Error::last_os_error()); + } + let job = OwnedHandle(raw_job); + + let mut limits = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default(); + limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + // SAFETY: `job` and `raw_process` are live handles, and `limits` + // has the exact layout required by JobObjectExtendedLimitInformation. + if unsafe { + SetInformationJobObject( + job.0, + JobObjectExtendedLimitInformation, + ptr::from_ref(&limits).cast(), + size_of::() as u32, + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + if unsafe { AssignProcessToJobObject(job.0, raw_process.cast()) } == 0 { + return Err(io::Error::last_os_error()); + } + + resume_primary_thread(child.id().ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + "CLI exited before its primary thread could be resumed", + ) + })?)?; + + #[cfg(test)] + ACTIVE_TREES.fetch_add(1, Ordering::Relaxed); + Ok(Self { job }) + } + + pub(super) fn terminate(&self) -> io::Result<()> { + // SAFETY: `self.job` is a live Job Object handle owned by this guard. + if unsafe { TerminateJobObject(self.job.0, 1) } != 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } + } + + pub(super) fn is_empty(&self) -> io::Result { + let mut accounting = JOBOBJECT_BASIC_ACCOUNTING_INFORMATION::default(); + // SAFETY: `accounting` has the exact layout requested by the query. + if unsafe { + QueryInformationJobObject( + self.job.0, + JobObjectBasicAccountingInformation, + ptr::from_mut(&mut accounting).cast(), + size_of::() as u32, + ptr::null_mut(), + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Ok(accounting.ActiveProcesses == 0) + } + + pub(super) fn reap_adopted(&self) -> io::Result<()> { + Ok(()) + } + } + + impl Drop for ProcessTree { + fn drop(&mut self) { + #[cfg(test)] + ACTIVE_TREES.fetch_sub(1, Ordering::Relaxed); + } + } + + pub(super) fn configure_command(command: &mut Command) { + use std::os::windows::process::CommandExt; + + command + .as_std_mut() + .creation_flags(CREATE_NO_WINDOW | CREATE_SUSPENDED); + } + + fn resume_primary_thread(pid: u32) -> io::Result<()> { + // The child was created suspended and therefore still has exactly one + // thread. Enumerating by owner PID recovers the primary thread handle + // that `std::process::Command` closes after CreateProcessW returns. + // SAFETY: the snapshot and thread handles are wrapped immediately. + let raw_snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) }; + if raw_snapshot == INVALID_HANDLE_VALUE { + return Err(io::Error::last_os_error()); + } + let snapshot = OwnedHandle(raw_snapshot); + let mut entry = THREADENTRY32 { + dwSize: size_of::() as u32, + ..Default::default() + }; + + // SAFETY: `entry` has the required size and remains live for iteration. + let mut has_entry = unsafe { Thread32First(snapshot.0, &mut entry) } != 0; + while has_entry { + if entry.th32OwnerProcessID == pid { + // SAFETY: the thread id came from the live system snapshot. + let raw_thread = + unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, entry.th32ThreadID) }; + if raw_thread.is_null() { + return Err(io::Error::last_os_error()); + } + let thread = OwnedHandle(raw_thread); + // SAFETY: this is the suspended primary thread of our child. + if unsafe { ResumeThread(thread.0) } == u32::MAX { + return Err(io::Error::last_os_error()); + } + return Ok(()); + } + // SAFETY: continue iterating the same valid snapshot and entry. + has_entry = unsafe { Thread32Next(snapshot.0, &mut entry) } != 0; + } + + Err(io::Error::new( + io::ErrorKind::NotFound, + "suspended CLI primary thread was not found", + )) + } + + #[cfg(test)] + pub(super) fn active_tree_count() -> usize { + ACTIVE_TREES.load(Ordering::Relaxed) + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use serial_test::serial; + use tempfile::tempdir; + + use super::*; + + const TEST_TIMEOUT: Duration = Duration::from_secs(10); + + #[tokio::test] + #[serial] + async fn terminate_kills_grandchild_and_reaps_leader() { + let baseline = active_tree_count(); + let temp = tempdir().expect("create temp directory"); + let pid_file = temp.path().join("grandchild.pid"); + let mut child = + ManagedChild::spawn(test_tree_command(&pid_file)).expect("spawn managed process tree"); + let direct_pid = child.id().expect("direct child pid"); + let grandchild_pid = wait_for_test_pid(&pid_file).await; + + assert!(test_process_exists(direct_pid)); + assert!(test_process_exists(grandchild_pid)); + assert_eq!(active_tree_count(), baseline + 1); + + child.terminate().expect("terminate process tree"); + child.wait().await.expect("reap direct child"); + child + .wait_for_tree_exit(TEST_TIMEOUT) + .await + .expect("wait for process tree exit"); + drop(child); + + assert!(!test_process_exists(direct_pid)); + assert!(!test_process_exists(grandchild_pid)); + assert_eq!(active_tree_count(), baseline); + } + + #[tokio::test] + #[serial] + async fn drop_kills_grandchild_and_reaps_leader() { + let baseline = active_tree_count(); + let temp = tempdir().expect("create temp directory"); + let pid_file = temp.path().join("grandchild.pid"); + let child = + ManagedChild::spawn(test_tree_command(&pid_file)).expect("spawn managed process tree"); + let direct_pid = child.id().expect("direct child pid"); + let grandchild_pid = wait_for_test_pid(&pid_file).await; + + drop(child); + + assert!( + wait_for_test_condition(TEST_TIMEOUT, || { + !test_process_exists(direct_pid) && !test_process_exists(grandchild_pid) + }) + .await, + "process tree survived managed-child drop" + ); + assert!( + wait_for_test_condition(TEST_TIMEOUT, || active_tree_count() == baseline).await, + "process-tree guard survived managed-child drop" + ); + } + + #[cfg(windows)] + #[tokio::test] + #[serial] + async fn job_handle_close_kills_grandchild() { + let baseline = active_tree_count(); + let temp = tempdir().expect("create temp directory"); + let pid_file = temp.path().join("grandchild.pid"); + let mut child = + ManagedChild::spawn(test_tree_command(&pid_file)).expect("spawn managed process tree"); + let direct_pid = child.id().expect("direct child pid"); + let grandchild_pid = wait_for_test_pid(&pid_file).await; + + drop(child.tree.take().expect("Windows Job Object")); + child.wait().await.expect("reap direct child"); + drop(child); + + assert!( + wait_for_test_condition(TEST_TIMEOUT, || { + !test_process_exists(direct_pid) + && !test_process_exists(grandchild_pid) + && active_tree_count() == baseline + }) + .await, + "process tree survived KILL_ON_JOB_CLOSE" + ); + } +} From 0f899159bc81263876175e55df646b0606be198c Mon Sep 17 00:00:00 2001 From: Luke Hoban Date: Thu, 6 Aug 2026 22:49:48 -0700 Subject: [PATCH 2/5] style(rust): group Windows process imports Match the repository's nightly rustfmt configuration on Linux. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1090dd5b-e149-4d9d-9cc3-67f26e11ad06 --- rust/src/process_tree.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/rust/src/process_tree.rs b/rust/src/process_tree.rs index 01d0a7e3e..88c0ad849 100644 --- a/rust/src/process_tree.rs +++ b/rust/src/process_tree.rs @@ -351,10 +351,9 @@ mod platform { #[cfg(windows)] mod platform { - use std::io; - use std::ptr; #[cfg(test)] use std::sync::atomic::{AtomicUsize, Ordering}; + use std::{io, ptr}; use tokio::process::{Child, Command}; use windows_sys::Win32::Foundation::{CloseHandle, HANDLE, INVALID_HANDLE_VALUE}; From 05544d9270e0820c2ddd17de34bfedfa5afa521a Mon Sep 17 00:00:00 2001 From: Stephen Toub Date: Fri, 14 Aug 2026 12:16:42 -0400 Subject: [PATCH 3/5] fix(rust): prevent orphaned CLI processes Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- rust/Cargo.lock | 2 - rust/Cargo.toml | 10 - rust/src/errors.rs | 12 +- rust/src/lib.rs | 354 +++++++--------------- rust/src/process_tree.rs | 627 --------------------------------------- 5 files changed, 109 insertions(+), 896 deletions(-) delete mode 100644 rust/src/process_tree.rs diff --git a/rust/Cargo.lock b/rust/Cargo.lock index b4445c0b6..8de679798 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -434,7 +434,6 @@ dependencies = [ "getrandom 0.2.17", "http", "indexmap", - "libc", "libloading", "native-tls", "parking_lot", @@ -455,7 +454,6 @@ dependencies = [ "tracing", "ureq", "uuid", - "windows-sys 0.61.2", "zip", ] diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 3fdee0f2e..0f18a9b15 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -69,18 +69,8 @@ reqwest = { version = "0.12", default-features = false, features = ["stream", "h tokio-tungstenite = { version = "0.24", default-features = false, features = ["connect", "native-tls"] } [target.'cfg(windows)'.dependencies] -windows-sys = { version = "0.61", features = [ - "Win32_Foundation", - "Win32_Security", - "Win32_System_Diagnostics_ToolHelp", - "Win32_System_JobObjects", - "Win32_System_Threading", -] } zip = { version = "2", default-features = false, features = ["deflate"], optional = true } -[target.'cfg(unix)'.dependencies] -libc = "0.2" - [dev-dependencies] rusqlite = { version = "0.35", features = ["bundled"] } schemars = "1" diff --git a/rust/src/errors.rs b/rust/src/errors.rs index ddd57d854..6e05bbfae 100644 --- a/rust/src/errors.rs +++ b/rust/src/errors.rs @@ -396,11 +396,11 @@ fn capture_backtrace() -> Option> { /// Aggregate of errors collected during [`crate::Client::stop`]. /// /// `Client::stop` performs cooperative shutdown across every active -/// session before terminating and reaping the SDK-owned CLI process tree. -/// Errors from any per-session `session.destroy` RPC and from the terminal -/// process-tree teardown are collected here rather than short-circuiting on -/// the first failure, so callers see the full picture of what went wrong -/// during teardown. +/// session before killing the CLI child process. Errors from any +/// per-session `session.destroy` RPC and from the terminal child-kill +/// step are collected here rather than short-circuiting on the first +/// failure, so callers see the full picture of what went wrong during +/// teardown. /// /// Implements [`std::error::Error`] and forwards to `Display` for the /// first error, with a count suffix when there are more. @@ -409,7 +409,7 @@ pub struct StopErrors(pub(crate) Vec); impl StopErrors { /// Borrow the collected errors as a slice, in the order they - /// occurred (per-session destroys first, then process-tree teardown). + /// occurred (per-session destroys first, then child-kill last). pub fn errors(&self) -> &[Error] { &self.0 } diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 12bf6c563..2f44c41d7 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -29,7 +29,6 @@ pub mod hooks; mod jsonrpc; /// Permission-policy helpers that produce a [`handler::PermissionHandler`]. pub mod permission; -mod process_tree; /// BYOK bearer-token provider callbacks. pub mod provider_token; mod provider_token_dispatch; @@ -102,7 +101,7 @@ pub mod test_support { use serde::{Deserialize, Serialize}; use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader}; use tokio::net::TcpStream; -use tokio::process::Command; +use tokio::process::{Child, Command}; use tokio::sync::{broadcast, mpsc, oneshot}; use tracing::{Instrument, debug, error, info, warn}; pub use types::*; @@ -972,7 +971,7 @@ fn validate_inprocess_options(options: &ClientOptions) -> Result<()> { /// Connection to a GitHub Copilot CLI server (stdio, TCP, or external). /// /// Cheaply cloneable — cloning shares the underlying connection. -/// The SDK-owned process tree (if any) is terminated when the last clone drops. +/// The child process (if any) is killed when the last clone drops. #[derive(Clone)] pub struct Client { inner: Arc, @@ -988,7 +987,7 @@ impl std::fmt::Debug for Client { } struct ClientInner { - child: parking_lot::Mutex>, + child: parking_lot::Mutex>, #[cfg(feature = "bundled-in-process")] /// In-process FFI runtime host, set only for [`Transport::InProcess`]. /// Closing it tears down the native runtime connection. @@ -1242,8 +1241,8 @@ impl Client { let (mut child, spawn_elapsed) = Self::spawn_stdio(&program, &options, &working_directory)?; timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed)); - let stdin = child.child_mut().stdin.take().expect("stdin is piped"); - let stdout = child.child_mut().stdout.take().expect("stdout is piped"); + let stdin = child.stdin.take().expect("stdin is piped"); + let stdout = child.stdout.take().expect("stdout is piped"); Self::drain_stderr(&mut child); Self::from_transport( stdout, @@ -1526,7 +1525,7 @@ impl Client { fn from_transport( reader: impl AsyncRead + Unpin + Send + 'static, writer: impl AsyncWrite + Unpin + Send + 'static, - child: Option, + child: Option, cwd: PathBuf, on_list_models: Option>, session_fs_configured: bool, @@ -1625,6 +1624,7 @@ impl Client { fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command { let mut command = Command::new(program); + command.kill_on_drop(true); for arg in &options.prefix_args { command.arg(arg); } @@ -1685,6 +1685,13 @@ impl Client { .stdout(Stdio::piped()) .stderr(Stdio::piped()); + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + const CREATE_NO_WINDOW: u32 = 0x08000000; + command.as_std_mut().creation_flags(CREATE_NO_WINDOW); + } + command } @@ -1741,7 +1748,7 @@ impl Client { program: &Path, options: &ClientOptions, working_directory: &Path, - ) -> Result<(process_tree::ManagedChild, Duration)> { + ) -> Result<(Child, Duration)> { info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)"); let mut command = Self::build_command(program, options, working_directory); command @@ -1753,7 +1760,7 @@ impl Client { .args(&options.extra_args) .stdin(Stdio::piped()); let spawn_start = Instant::now(); - let child = process_tree::ManagedChild::spawn(command)?; + let child = command.spawn()?; let spawn_elapsed = spawn_start.elapsed(); debug!( elapsed_ms = spawn_elapsed.as_millis(), @@ -1767,7 +1774,7 @@ impl Client { options: &ClientOptions, working_directory: &Path, port: u16, - ) -> Result<(process_tree::ManagedChild, u16, Duration, Duration)> { + ) -> Result<(Child, u16, Duration, Duration)> { info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)"); let mut command = Self::build_command(program, options, working_directory); command @@ -1779,13 +1786,13 @@ impl Client { .args(&options.extra_args) .stdin(Stdio::null()); let spawn_start = Instant::now(); - let mut child = process_tree::ManagedChild::spawn(command)?; + let mut child = command.spawn()?; let spawn_elapsed = spawn_start.elapsed(); debug!( elapsed_ms = spawn_elapsed.as_millis(), "Client::spawn_tcp subprocess spawned" ); - let stdout = child.child_mut().stdout.take().expect("stdout is piped"); + let stdout = child.stdout.take().expect("stdout is piped"); let (port_tx, port_rx) = oneshot::channel::(); let span = tracing::error_span!("copilot_cli_port_scan"); @@ -1829,8 +1836,8 @@ impl Client { Ok((child, actual_port, spawn_elapsed, port_wait_elapsed)) } - fn drain_stderr(child: &mut process_tree::ManagedChild) { - if let Some(stderr) = child.child_mut().stderr.take() { + fn drain_stderr(child: &mut Child) { + if let Some(stderr) = child.stderr.take() { let span = tracing::error_span!("copilot_cli"); tokio::spawn( async move { @@ -2336,26 +2343,21 @@ impl Client { /// Return the OS process ID of the CLI child process, if one was spawned. pub fn pid(&self) -> Option { - self.inner - .child - .lock() - .as_ref() - .and_then(process_tree::ManagedChild::id) + self.inner.child.lock().as_ref().and_then(|c| c.id()) } - /// Cooperatively shut down the client and its SDK-owned process tree. + /// Cooperatively shut down the client and the CLI child process. /// /// Walks every still-registered session and sends `session.destroy` - /// for each one, asks SDK-owned runtimes to shut down, then terminates and - /// reaps the complete spawned process tree. Errors from per-session - /// destroys, runtime shutdown, tree termination, and process reaping are - /// collected into + /// for each one, asks SDK-owned runtimes to shut down, then kills the + /// CLI child. Errors from per-session destroys, runtime shutdown, and + /// the final child-kill are collected into /// [`StopErrors`] rather than short-circuiting on the first failure /// — so callers see the full picture of teardown. /// /// If you have already called [`Session::disconnect`] on every /// session this client created, the per-session destroy step is a - /// no-op (the router map is empty); only process-tree teardown remains. + /// no-op (the router map is empty); only the child-kill remains. /// /// [`Session::disconnect`]: crate::session::Session::disconnect /// @@ -2441,17 +2443,20 @@ impl Client { *self.inner.state.lock() = ConnectionState::Disconnected; *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new()); if let Some(mut child) = child { - // The runtime completes all cleanup before responding to - // runtime.shutdown and deliberately leaves process termination to - // its owner so it can send the response first. - if let Err(e) = child.terminate() { - errors.push(e.into()); - } - if let Err(e) = child.wait().await { - errors.push(e.into()); - } - if let Err(e) = child.wait_for_tree_exit(RUNTIME_SHUTDOWN_TIMEOUT).await { - errors.push(e.into()); + match child.try_wait() { + Ok(Some(_status)) => {} + Ok(None) => { + // The runtime completes all cleanup before responding to + // runtime.shutdown and then leaves termination to us; it + // deliberately keeps its JSON-RPC server alive to send the + // response and never self-exits. Waiting for a self-exit + // that will never come just wastes time, so terminate the + // child immediately. + if let Err(e) = child.kill().await { + errors.push(e.into()); + } + } + Err(e) => errors.push(e.into()), } } @@ -2473,14 +2478,14 @@ impl Client { } } - /// Forcibly stop the CLI process tree. + /// Forcibly stop the CLI process without waiting for it to exit. /// /// Synchronous fallback when [`stop`](Self::stop) is unsuitable — for /// example when the awaiting tokio runtime is shutting down or the - /// process is wedged on I/O. Terminates the complete tree, briefly polls - /// the direct child for exit, and transfers any slow exit to a dedicated - /// finite reaper thread. It immediately drops all per-session router state - /// so dependent tasks observe a closed channel rather than a hang. + /// process is wedged on I/O. Sends a kill signal without awaiting + /// reaper completion and immediately drops all per-session router + /// state so dependent tasks observe a closed channel rather than a + /// hang. /// /// # Cancel safety /// @@ -2506,9 +2511,9 @@ impl Client { let pid = self.pid(); info!(pid = ?pid, "force-stopping CLI process"); if let Some(mut child) = self.inner.child.lock().take() - && let Err(e) = child.terminate() + && let Err(e) = child.start_kill() { - error!(pid = ?pid, error = %e, "failed to terminate CLI process tree"); + error!(pid = ?pid, error = %e, "failed to send kill signal"); } self.inner.rpc.force_close(); #[cfg(feature = "bundled-in-process")] @@ -2565,12 +2570,12 @@ impl Client { impl Drop for ClientInner { fn drop(&mut self) { - if let Some(mut child) = self.child.lock().take() { + if let Some(ref mut child) = *self.child.lock() { let pid = child.id(); - if let Err(e) = child.terminate() { - error!(pid = ?pid, error = %e, "failed to terminate CLI process tree on drop"); + if let Err(e) = child.start_kill() { + error!(pid = ?pid, error = %e, "failed to kill CLI process on drop"); } else { - info!(pid = ?pid, "CLI process tree terminated on drop"); + info!(pid = ?pid, "kill signal sent for CLI process on drop"); } } #[cfg(feature = "bundled-in-process")] @@ -2585,13 +2590,6 @@ impl Drop for ClientInner { #[cfg(test)] mod tests { - #[cfg(unix)] - use std::os::unix::fs::PermissionsExt; - - use serial_test::serial; - use tempfile::{TempDir, tempdir}; - use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; - use super::*; #[test] @@ -3219,216 +3217,70 @@ mod tests { } #[tokio::test] - #[serial] - async fn client_process_tree_stop_reaps_grandchild() { - let baseline = process_tree::active_tree_count(); - let (client, direct_pid, grandchild_pid, _temp, mut server_read, mut server_write) = - managed_test_client().await; - let server = tokio::spawn(async move { - let request = read_framed_json(&mut server_read).await; - assert_eq!(request["method"], "runtime.shutdown"); - let response = serde_json::json!({ - "jsonrpc": "2.0", - "id": request["id"], - "result": null, - }); - write_framed_json(&mut server_write, &response).await; - }); - - client.stop().await.expect("stop client"); - server.await.expect("runtime shutdown server"); - - assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; - } - - #[tokio::test] - #[serial] - async fn client_process_tree_force_stop_reaps_grandchild() { - let baseline = process_tree::active_tree_count(); - let (client, direct_pid, grandchild_pid, _temp, _server_read, _server_write) = - managed_test_client().await; - - client.force_stop(); - - assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; - } - - #[tokio::test] - #[serial] - async fn client_process_tree_drop_reaps_grandchild() { - let baseline = process_tree::active_tree_count(); - let (client, direct_pid, grandchild_pid, _temp, _server_read, _server_write) = - managed_test_client().await; + async fn lifecycle_dispatcher_does_not_keep_client_alive() { + let (client_write, _server_read) = tokio::io::duplex(64); + let (_server_write, client_read) = tokio::io::duplex(64); + let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap(); + let inner = Arc::downgrade(&client.inner); drop(client); - assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; + assert!(inner.upgrade().is_none()); } - #[cfg(unix)] + #[cfg(any(unix, windows))] #[tokio::test] - #[serial] - async fn client_process_tree_tcp_startup_failure_reaps_tree() { - let baseline = process_tree::active_tree_count(); - let temp = tempdir().expect("create temp directory"); - let pid_file = temp.path().join("startup.pids"); - let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("reserve port"); - let unused_port = listener.local_addr().expect("reserved address").port(); - drop(listener); - let script = executable_script( - &temp, - "fake-tcp-cli.sh", - "#!/bin/sh\nsleep 60 /dev/null 2>&1 &\necho \"$$ $!\" > \"$PID_FILE\"\necho \"listening on port $FAKE_PORT\"\nwait\n", - ); - let options = ClientOptions::new() - .with_program(CliProgram::Path(script)) - .with_transport(Transport::Tcp { - port: 0, - connection_token: None, - }) - .with_env([ - ("PID_FILE", pid_file.as_os_str()), - ("FAKE_PORT", std::ffi::OsStr::new(&unused_port.to_string())), + async fn spawned_child_is_killed_when_dropped() { + let temp = tempfile::tempdir().unwrap(); + let ready = temp.path().join("ready"); + let survived = temp.path().join("survived"); + + #[cfg(unix)] + let mut command = { + let mut command = + Client::build_command(Path::new("sh"), &ClientOptions::default(), temp.path()); + command.args([ + "-c", + "printf ready > \"$READY\"; sleep 1; printf survived > \"$SURVIVED\"", ]); + command + }; + #[cfg(windows)] + let mut command = { + let mut command = Client::build_command( + Path::new("powershell.exe"), + &ClientOptions::default(), + temp.path(), + ); + command.args([ + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + "Set-Content -LiteralPath $env:READY ready; Start-Sleep -Seconds 1; Set-Content -LiteralPath $env:SURVIVED survived", + ]); + command + }; + command.env("READY", &ready).env("SURVIVED", &survived); + let child = command.spawn().unwrap(); - Client::start(options) - .await - .expect_err("TCP connect should fail"); - let (direct_pid, grandchild_pid) = read_test_pids(&pid_file).await; - - assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; - } - - #[cfg(unix)] - #[tokio::test] - #[serial] - async fn client_process_tree_handshake_failure_reaps_tree() { - let baseline = process_tree::active_tree_count(); - let temp = tempdir().expect("create temp directory"); - let pid_file = temp.path().join("startup.pids"); - let script = executable_script( - &temp, - "fake-stdio-cli.sh", - "#!/bin/sh\nsleep 60 /dev/null 2>&1 &\necho \"$$ $!\" > \"$PID_FILE\"\nsleep 0.1\nexit 0\n", - ); - let options = ClientOptions::new() - .with_program(CliProgram::Path(script)) - .with_env([("PID_FILE", pid_file.as_os_str())]); - - Client::start(options) - .await - .expect_err("protocol handshake should fail"); - let (direct_pid, grandchild_pid) = read_test_pids(&pid_file).await; - - assert_test_tree_gone(direct_pid, grandchild_pid, baseline).await; - } - - async fn managed_test_client() -> (Client, u32, u32, TempDir, DuplexStream, DuplexStream) { - let temp = tempdir().expect("create temp directory"); - let pid_file = temp.path().join("grandchild.pid"); - let child = process_tree::ManagedChild::spawn(process_tree::test_tree_command(&pid_file)) - .expect("spawn managed process tree"); - let direct_pid = child.id().expect("direct child pid"); - let grandchild_pid = process_tree::wait_for_test_pid(&pid_file).await; - let (client_write, server_read) = tokio::io::duplex(8192); - let (server_write, client_read) = tokio::io::duplex(8192); - let client = Client::from_transport( - client_read, - client_write, - Some(child), - temp.path().to_path_buf(), - None, - false, - false, - None, - None, - None, - ClientMode::default(), - ) - .expect("create client"); - ( - client, - direct_pid, - grandchild_pid, - temp, - server_read, - server_write, - ) - } + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while !ready.exists() { + assert!( + tokio::time::Instant::now() < deadline, + "child did not report readiness" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + drop(child); + tokio::time::sleep(Duration::from_millis(1500)).await; - async fn assert_test_tree_gone(direct_pid: u32, grandchild_pid: u32, baseline: usize) { assert!( - process_tree::wait_for_test_condition(Duration::from_secs(10), || { - !process_tree::test_process_exists(direct_pid) - && !process_tree::test_process_exists(grandchild_pid) - && process_tree::active_tree_count() == baseline - }) - .await, - "managed process tree or guard survived teardown" + !survived.exists(), + "child survived after its owner was dropped" ); } - #[cfg(unix)] - fn executable_script(temp: &TempDir, name: &str, contents: &str) -> PathBuf { - let path = temp.path().join(name); - std::fs::write(&path, contents).expect("write test script"); - let mut permissions = std::fs::metadata(&path) - .expect("read test script metadata") - .permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&path, permissions).expect("make test script executable"); - path - } - - #[cfg(unix)] - async fn read_test_pids(path: &Path) -> (u32, u32) { - let found = - process_tree::wait_for_test_condition(Duration::from_secs(10), || path.exists()).await; - assert!(found, "startup helper pid file was not created"); - let contents = std::fs::read_to_string(path).expect("read startup helper pids"); - let mut pids = contents.split_whitespace().map(|value| { - value - .parse::() - .expect("startup helper pid should be numeric") - }); - ( - pids.next().expect("direct child pid"), - pids.next().expect("grandchild pid"), - ) - } - - async fn read_framed_json(reader: &mut DuplexStream) -> serde_json::Value { - let mut header = Vec::new(); - while !header.ends_with(b"\r\n\r\n") { - let mut byte = [0u8; 1]; - reader - .read_exact(&mut byte) - .await - .expect("read frame header"); - header.push(byte[0]); - } - let header = String::from_utf8(header).expect("frame header is UTF-8"); - let length = header - .lines() - .find_map(|line| line.strip_prefix("Content-Length: ")) - .expect("content length header") - .parse::() - .expect("content length is numeric"); - let mut body = vec![0u8; length]; - reader.read_exact(&mut body).await.expect("read frame body"); - serde_json::from_slice(&body).expect("parse framed JSON") - } - - async fn write_framed_json(writer: &mut DuplexStream, value: &serde_json::Value) { - let body = serde_json::to_vec(value).expect("serialize framed JSON"); - writer - .write_all(format!("Content-Length: {}\r\n\r\n", body.len()).as_bytes()) - .await - .expect("write frame header"); - writer.write_all(&body).await.expect("write frame body"); - writer.flush().await.expect("flush frame"); - } - fn client_with_list_models_handler(handler: Arc) -> Client { Client { inner: Arc::new(ClientInner { diff --git a/rust/src/process_tree.rs b/rust/src/process_tree.rs deleted file mode 100644 index 88c0ad849..000000000 --- a/rust/src/process_tree.rs +++ /dev/null @@ -1,627 +0,0 @@ -//! Ownership and teardown for SDK-spawned process trees. - -use std::io; -use std::process::ExitStatus; -use std::time::{Duration, Instant}; - -use tokio::process::{Child, Command}; -use tracing::{error, warn}; - -const TREE_EXIT_POLL_INTERVAL: Duration = Duration::from_millis(10); -const SYNC_REAP_GRACE: Duration = Duration::from_millis(250); - -/// Owns a direct child and the platform primitive that contains its descendants. -pub(crate) struct ManagedChild { - child: Option, - tree: Option, - tree_terminated: bool, -} - -impl ManagedChild { - /// Spawn a child into a process tree before it can create descendants. - pub(crate) fn spawn(mut command: Command) -> io::Result { - command.kill_on_drop(true); - platform::configure_command(&mut command); - - let mut child = command.spawn()?; - match platform::ProcessTree::attach_and_start(&mut child) { - Ok(tree) => Ok(Self { - child: Some(child), - tree: Some(tree), - tree_terminated: false, - }), - Err(error) => { - reap_failed_spawn(&mut child); - Err(error) - } - } - } - - pub(crate) fn child_mut(&mut self) -> &mut Child { - self.child.as_mut().expect("managed child is present") - } - - pub(crate) fn id(&self) -> Option { - self.child.as_ref().and_then(Child::id) - } - - /// Terminate the complete tree. If the tree primitive fails, still signal - /// the direct child so teardown never regresses to doing nothing. - pub(crate) fn terminate(&mut self) -> io::Result<()> { - if self.tree_terminated { - return Ok(()); - } - let result = self - .tree - .as_ref() - .expect("managed process tree is present") - .terminate(); - if result.is_ok() { - self.tree_terminated = true; - } - if result.is_err() - && let Some(child) = self.child.as_mut() - { - let _ = child.start_kill(); - } - result - } - - /// Wait for and reap the direct child through Tokio's sole child owner. - pub(crate) async fn wait(&mut self) -> io::Result { - self.child_mut().wait().await - } - - /// Verify that no process remains in the platform tree. - pub(crate) async fn wait_for_tree_exit(&mut self, timeout: Duration) -> io::Result<()> { - let started = Instant::now(); - loop { - let tree = self.tree.as_ref().expect("managed process tree is present"); - tree.reap_adopted()?; - if tree.is_empty()? { - self.tree.take(); - return Ok(()); - } - if started.elapsed() >= timeout { - return Err(io::Error::new( - io::ErrorKind::TimedOut, - "timed out waiting for CLI process tree to exit", - )); - } - tokio::time::sleep(TREE_EXIT_POLL_INTERVAL).await; - } - } -} - -impl Drop for ManagedChild { - fn drop(&mut self) { - let Some(mut child) = self.child.take() else { - return; - }; - let tree = self.tree.take(); - let pid = child.id(); - - if !self.tree_terminated - && let Some(tree) = tree.as_ref() - && let Err(error) = tree.terminate() - { - warn!(pid = ?pid, %error, "failed to terminate CLI process tree on drop"); - } - if let Err(error) = child.start_kill() - && child.try_wait().ok().flatten().is_none() - { - warn!(pid = ?pid, %error, "failed to terminate direct CLI child on drop"); - } - - if reap_for(&mut child, SYNC_REAP_GRACE) { - return; - } - - let result = std::thread::Builder::new() - .name("copilot-cli-reaper".to_string()) - .spawn(move || { - loop { - match child.try_wait() { - Ok(Some(_)) => break, - Ok(None) => std::thread::sleep(TREE_EXIT_POLL_INTERVAL), - Err(error) => { - warn!(pid = ?pid, %error, "failed to reap CLI child"); - break; - } - } - } - drop(tree); - }); - if let Err(error) = result { - error!(pid = ?pid, %error, "failed to start CLI child reaper thread"); - } - } -} - -fn reap_failed_spawn(child: &mut Child) { - let pid = child.id(); - if let Err(error) = child.start_kill() { - warn!(pid = ?pid, %error, "failed to terminate CLI after process-tree setup failure"); - } - if !reap_for(child, SYNC_REAP_GRACE) { - warn!(pid = ?pid, "CLI did not exit promptly after process-tree setup failure"); - } -} - -fn reap_for(child: &mut Child, timeout: Duration) -> bool { - let started = Instant::now(); - loop { - match child.try_wait() { - Ok(Some(_)) => return true, - Ok(None) if started.elapsed() < timeout => { - std::thread::sleep(TREE_EXIT_POLL_INTERVAL); - } - Ok(None) | Err(_) => return false, - } - } -} - -#[cfg(test)] -pub(crate) fn active_tree_count() -> usize { - platform::active_tree_count() -} - -#[cfg(test)] -pub(crate) async fn wait_for_test_pid(path: &std::path::Path) -> u32 { - let found = wait_for_test_condition(Duration::from_secs(10), || path.exists()).await; - assert!(found, "grandchild pid file was not created"); - std::fs::read_to_string(path) - .expect("read grandchild pid") - .trim() - .parse() - .expect("parse grandchild pid") -} - -#[cfg(test)] -pub(crate) async fn wait_for_test_condition( - timeout: Duration, - mut predicate: impl FnMut() -> bool, -) -> bool { - let started = Instant::now(); - loop { - if predicate() { - return true; - } - if started.elapsed() >= timeout { - return false; - } - tokio::time::sleep(TREE_EXIT_POLL_INTERVAL).await; - } -} - -#[cfg(all(test, unix))] -pub(crate) fn test_tree_command(pid_file: &std::path::Path) -> Command { - let mut command = Command::new("sh"); - command - .args(["-c", "sleep 60 & echo \"$!\" > \"$PID_FILE\"; wait"]) - .env("PID_FILE", pid_file) - .stdin(std::process::Stdio::null()) - .stdout(std::process::Stdio::null()) - .stderr(std::process::Stdio::null()); - command -} - -#[cfg(all(test, windows))] -pub(crate) fn test_tree_command(pid_file: &std::path::Path) -> Command { - let script = concat!( - "$child = Start-Process powershell.exe ", - "-ArgumentList @('-NoLogo','-NoProfile','-NonInteractive','-Command',", - "'Start-Sleep -Seconds 60') -PassThru; ", - "Set-Content -LiteralPath $env:PID_FILE -Value $child.Id; ", - "Wait-Process -Id $child.Id" - ); - let mut command = Command::new("powershell.exe"); - command - .args([ - "-NoLogo", - "-NoProfile", - "-NonInteractive", - "-Command", - script, - ]) - .env("PID_FILE", pid_file) - .stdin(std::process::Stdio::null()) - .stdout(std::process::Stdio::null()) - .stderr(std::process::Stdio::null()); - command -} - -#[cfg(all(test, unix))] -pub(crate) fn test_process_exists(pid: u32) -> bool { - // SAFETY: signal 0 only probes process existence. - (unsafe { libc::kill(pid as i32, 0) }) == 0 -} - -#[cfg(all(test, windows))] -pub(crate) fn test_process_exists(pid: u32) -> bool { - use windows_sys::Win32::Foundation::{CloseHandle, WAIT_TIMEOUT}; - use windows_sys::Win32::System::Threading::{ - OpenProcess, PROCESS_SYNCHRONIZE, WaitForSingleObject, - }; - - // SAFETY: the process handle is closed before returning. - unsafe { - let process = OpenProcess(PROCESS_SYNCHRONIZE, 0, pid); - if process.is_null() { - return false; - } - let exists = WaitForSingleObject(process, 0) == WAIT_TIMEOUT; - CloseHandle(process); - exists - } -} - -#[cfg(unix)] -mod platform { - use std::io; - #[cfg(test)] - use std::sync::atomic::{AtomicUsize, Ordering}; - - use tokio::process::{Child, Command}; - - #[cfg(test)] - static ACTIVE_TREES: AtomicUsize = AtomicUsize::new(0); - - pub(super) struct ProcessTree { - pgid: i32, - } - - impl ProcessTree { - pub(super) fn attach_and_start(child: &mut Child) -> io::Result { - let pid = child.id().ok_or_else(|| { - io::Error::new( - io::ErrorKind::NotFound, - "CLI exited before process-group ownership was established", - ) - })?; - #[cfg(test)] - ACTIVE_TREES.fetch_add(1, Ordering::Relaxed); - Ok(Self { pgid: pid as i32 }) - } - - pub(super) fn terminate(&self) -> io::Result<()> { - // SAFETY: `pgid` is the dedicated group created for this child. - if unsafe { libc::killpg(self.pgid, libc::SIGKILL) } == 0 { - return Ok(()); - } - let error = io::Error::last_os_error(); - if error.raw_os_error() == Some(libc::ESRCH) { - Ok(()) - } else { - Err(error) - } - } - - pub(super) fn is_empty(&self) -> io::Result { - // SAFETY: signal 0 only probes the dedicated process group. - if unsafe { libc::killpg(self.pgid, 0) } == 0 { - return Ok(false); - } - let error = io::Error::last_os_error(); - if error.raw_os_error() == Some(libc::ESRCH) { - Ok(true) - } else { - Err(error) - } - } - - pub(super) fn reap_adopted(&self) -> io::Result<()> { - loop { - let mut status = 0; - // SAFETY: a negative pid selects children in this dedicated - // process group. WNOHANG keeps the async caller non-blocking. - let result = unsafe { libc::waitpid(-self.pgid, &mut status, libc::WNOHANG) }; - if result > 0 { - continue; - } - if result == 0 { - return Ok(()); - } - let error = io::Error::last_os_error(); - match error.raw_os_error() { - Some(libc::ECHILD) => return Ok(()), - Some(libc::EINTR) => continue, - _ => return Err(error), - } - } - } - } - - impl Drop for ProcessTree { - fn drop(&mut self) { - #[cfg(test)] - ACTIVE_TREES.fetch_sub(1, Ordering::Relaxed); - } - } - - pub(super) fn configure_command(command: &mut Command) { - command.process_group(0); - } - - #[cfg(test)] - pub(super) fn active_tree_count() -> usize { - ACTIVE_TREES.load(Ordering::Relaxed) - } -} - -#[cfg(windows)] -mod platform { - #[cfg(test)] - use std::sync::atomic::{AtomicUsize, Ordering}; - use std::{io, ptr}; - - use tokio::process::{Child, Command}; - use windows_sys::Win32::Foundation::{CloseHandle, HANDLE, INVALID_HANDLE_VALUE}; - use windows_sys::Win32::System::Diagnostics::ToolHelp::{ - CreateToolhelp32Snapshot, TH32CS_SNAPTHREAD, THREADENTRY32, Thread32First, Thread32Next, - }; - use windows_sys::Win32::System::JobObjects::{ - AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, - JOBOBJECT_BASIC_ACCOUNTING_INFORMATION, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, - JobObjectBasicAccountingInformation, JobObjectExtendedLimitInformation, - QueryInformationJobObject, SetInformationJobObject, TerminateJobObject, - }; - use windows_sys::Win32::System::Threading::{ - CREATE_NO_WINDOW, CREATE_SUSPENDED, OpenThread, ResumeThread, THREAD_SUSPEND_RESUME, - }; - - #[cfg(test)] - static ACTIVE_TREES: AtomicUsize = AtomicUsize::new(0); - - struct OwnedHandle(HANDLE); - - // SAFETY: Windows kernel handles can be used and closed from any thread. - unsafe impl Send for OwnedHandle {} - - impl Drop for OwnedHandle { - fn drop(&mut self) { - // SAFETY: this type owns the valid handle and closes it exactly once. - unsafe { - CloseHandle(self.0); - } - } - } - - pub(super) struct ProcessTree { - job: OwnedHandle, - } - - impl ProcessTree { - pub(super) fn attach_and_start(child: &mut Child) -> io::Result { - let raw_process = child.raw_handle().ok_or_else(|| { - io::Error::new( - io::ErrorKind::NotFound, - "CLI exited before Job Object ownership was established", - ) - })?; - - // SAFETY: null attributes and name create a private, non-inheritable Job Object. - let raw_job = unsafe { CreateJobObjectW(ptr::null(), ptr::null()) }; - if raw_job.is_null() { - return Err(io::Error::last_os_error()); - } - let job = OwnedHandle(raw_job); - - let mut limits = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default(); - limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; - // SAFETY: `job` and `raw_process` are live handles, and `limits` - // has the exact layout required by JobObjectExtendedLimitInformation. - if unsafe { - SetInformationJobObject( - job.0, - JobObjectExtendedLimitInformation, - ptr::from_ref(&limits).cast(), - size_of::() as u32, - ) - } == 0 - { - return Err(io::Error::last_os_error()); - } - if unsafe { AssignProcessToJobObject(job.0, raw_process.cast()) } == 0 { - return Err(io::Error::last_os_error()); - } - - resume_primary_thread(child.id().ok_or_else(|| { - io::Error::new( - io::ErrorKind::NotFound, - "CLI exited before its primary thread could be resumed", - ) - })?)?; - - #[cfg(test)] - ACTIVE_TREES.fetch_add(1, Ordering::Relaxed); - Ok(Self { job }) - } - - pub(super) fn terminate(&self) -> io::Result<()> { - // SAFETY: `self.job` is a live Job Object handle owned by this guard. - if unsafe { TerminateJobObject(self.job.0, 1) } != 0 { - Ok(()) - } else { - Err(io::Error::last_os_error()) - } - } - - pub(super) fn is_empty(&self) -> io::Result { - let mut accounting = JOBOBJECT_BASIC_ACCOUNTING_INFORMATION::default(); - // SAFETY: `accounting` has the exact layout requested by the query. - if unsafe { - QueryInformationJobObject( - self.job.0, - JobObjectBasicAccountingInformation, - ptr::from_mut(&mut accounting).cast(), - size_of::() as u32, - ptr::null_mut(), - ) - } == 0 - { - return Err(io::Error::last_os_error()); - } - Ok(accounting.ActiveProcesses == 0) - } - - pub(super) fn reap_adopted(&self) -> io::Result<()> { - Ok(()) - } - } - - impl Drop for ProcessTree { - fn drop(&mut self) { - #[cfg(test)] - ACTIVE_TREES.fetch_sub(1, Ordering::Relaxed); - } - } - - pub(super) fn configure_command(command: &mut Command) { - use std::os::windows::process::CommandExt; - - command - .as_std_mut() - .creation_flags(CREATE_NO_WINDOW | CREATE_SUSPENDED); - } - - fn resume_primary_thread(pid: u32) -> io::Result<()> { - // The child was created suspended and therefore still has exactly one - // thread. Enumerating by owner PID recovers the primary thread handle - // that `std::process::Command` closes after CreateProcessW returns. - // SAFETY: the snapshot and thread handles are wrapped immediately. - let raw_snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) }; - if raw_snapshot == INVALID_HANDLE_VALUE { - return Err(io::Error::last_os_error()); - } - let snapshot = OwnedHandle(raw_snapshot); - let mut entry = THREADENTRY32 { - dwSize: size_of::() as u32, - ..Default::default() - }; - - // SAFETY: `entry` has the required size and remains live for iteration. - let mut has_entry = unsafe { Thread32First(snapshot.0, &mut entry) } != 0; - while has_entry { - if entry.th32OwnerProcessID == pid { - // SAFETY: the thread id came from the live system snapshot. - let raw_thread = - unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, entry.th32ThreadID) }; - if raw_thread.is_null() { - return Err(io::Error::last_os_error()); - } - let thread = OwnedHandle(raw_thread); - // SAFETY: this is the suspended primary thread of our child. - if unsafe { ResumeThread(thread.0) } == u32::MAX { - return Err(io::Error::last_os_error()); - } - return Ok(()); - } - // SAFETY: continue iterating the same valid snapshot and entry. - has_entry = unsafe { Thread32Next(snapshot.0, &mut entry) } != 0; - } - - Err(io::Error::new( - io::ErrorKind::NotFound, - "suspended CLI primary thread was not found", - )) - } - - #[cfg(test)] - pub(super) fn active_tree_count() -> usize { - ACTIVE_TREES.load(Ordering::Relaxed) - } -} - -#[cfg(test)] -mod tests { - use std::time::Duration; - - use serial_test::serial; - use tempfile::tempdir; - - use super::*; - - const TEST_TIMEOUT: Duration = Duration::from_secs(10); - - #[tokio::test] - #[serial] - async fn terminate_kills_grandchild_and_reaps_leader() { - let baseline = active_tree_count(); - let temp = tempdir().expect("create temp directory"); - let pid_file = temp.path().join("grandchild.pid"); - let mut child = - ManagedChild::spawn(test_tree_command(&pid_file)).expect("spawn managed process tree"); - let direct_pid = child.id().expect("direct child pid"); - let grandchild_pid = wait_for_test_pid(&pid_file).await; - - assert!(test_process_exists(direct_pid)); - assert!(test_process_exists(grandchild_pid)); - assert_eq!(active_tree_count(), baseline + 1); - - child.terminate().expect("terminate process tree"); - child.wait().await.expect("reap direct child"); - child - .wait_for_tree_exit(TEST_TIMEOUT) - .await - .expect("wait for process tree exit"); - drop(child); - - assert!(!test_process_exists(direct_pid)); - assert!(!test_process_exists(grandchild_pid)); - assert_eq!(active_tree_count(), baseline); - } - - #[tokio::test] - #[serial] - async fn drop_kills_grandchild_and_reaps_leader() { - let baseline = active_tree_count(); - let temp = tempdir().expect("create temp directory"); - let pid_file = temp.path().join("grandchild.pid"); - let child = - ManagedChild::spawn(test_tree_command(&pid_file)).expect("spawn managed process tree"); - let direct_pid = child.id().expect("direct child pid"); - let grandchild_pid = wait_for_test_pid(&pid_file).await; - - drop(child); - - assert!( - wait_for_test_condition(TEST_TIMEOUT, || { - !test_process_exists(direct_pid) && !test_process_exists(grandchild_pid) - }) - .await, - "process tree survived managed-child drop" - ); - assert!( - wait_for_test_condition(TEST_TIMEOUT, || active_tree_count() == baseline).await, - "process-tree guard survived managed-child drop" - ); - } - - #[cfg(windows)] - #[tokio::test] - #[serial] - async fn job_handle_close_kills_grandchild() { - let baseline = active_tree_count(); - let temp = tempdir().expect("create temp directory"); - let pid_file = temp.path().join("grandchild.pid"); - let mut child = - ManagedChild::spawn(test_tree_command(&pid_file)).expect("spawn managed process tree"); - let direct_pid = child.id().expect("direct child pid"); - let grandchild_pid = wait_for_test_pid(&pid_file).await; - - drop(child.tree.take().expect("Windows Job Object")); - child.wait().await.expect("reap direct child"); - drop(child); - - assert!( - wait_for_test_condition(TEST_TIMEOUT, || { - !test_process_exists(direct_pid) - && !test_process_exists(grandchild_pid) - && active_tree_count() == baseline - }) - .await, - "process tree survived KILL_ON_JOB_CLOSE" - ); - } -} From 143dd3f67ad2957f6e9c3e51f33f9bda5d0da4d0 Mon Sep 17 00:00:00 2001 From: Stephen Toub Date: Fri, 14 Aug 2026 12:37:34 -0400 Subject: [PATCH 4/5] test(rust): reproduce orphaned CLI on client drop Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- rust/src/lib.rs | 60 ++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 47 insertions(+), 13 deletions(-) diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 2f44c41d7..ee9d91ca6 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -3216,16 +3216,36 @@ mod tests { client.force_stop(); } + #[cfg(any(unix, windows))] #[tokio::test] - async fn lifecycle_dispatcher_does_not_keep_client_alive() { + async fn dropping_last_client_kills_spawned_cli() { + let temp = tempfile::tempdir().unwrap(); + let ready = temp.path().join("ready"); + let survived = temp.path().join("survived"); + let child = test_child_command(temp.path(), &ready, &survived) + .spawn() + .unwrap(); let (client_write, _server_read) = tokio::io::duplex(64); let (_server_write, client_read) = tokio::io::duplex(64); - let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap(); - let inner = Arc::downgrade(&client.inner); + let client = Client::from_transport( + client_read, + client_write, + Some(child), + temp.path().to_path_buf(), + None, + false, + false, + None, + None, + None, + ClientMode::default(), + ) + .unwrap(); + wait_for_test_child(&ready).await; drop(client); - assert!(inner.upgrade().is_none()); + assert_test_child_killed(&survived).await; } #[cfg(any(unix, windows))] @@ -3234,11 +3254,22 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let ready = temp.path().join("ready"); let survived = temp.path().join("survived"); + let child = test_child_command(temp.path(), &ready, &survived) + .spawn() + .unwrap(); + wait_for_test_child(&ready).await; + drop(child); + + assert_test_child_killed(&survived).await; + } + + #[cfg(any(unix, windows))] + fn test_child_command(temp: &Path, ready: &Path, survived: &Path) -> Command { #[cfg(unix)] let mut command = { let mut command = - Client::build_command(Path::new("sh"), &ClientOptions::default(), temp.path()); + Client::build_command(Path::new("sh"), &ClientOptions::default(), temp); command.args([ "-c", "printf ready > \"$READY\"; sleep 1; printf survived > \"$SURVIVED\"", @@ -3247,11 +3278,8 @@ mod tests { }; #[cfg(windows)] let mut command = { - let mut command = Client::build_command( - Path::new("powershell.exe"), - &ClientOptions::default(), - temp.path(), - ); + let mut command = + Client::build_command(Path::new("powershell.exe"), &ClientOptions::default(), temp); command.args([ "-NoLogo", "-NoProfile", @@ -3261,9 +3289,12 @@ mod tests { ]); command }; - command.env("READY", &ready).env("SURVIVED", &survived); - let child = command.spawn().unwrap(); + command.env("READY", ready).env("SURVIVED", survived); + command + } + #[cfg(any(unix, windows))] + async fn wait_for_test_child(ready: &Path) { let deadline = tokio::time::Instant::now() + Duration::from_secs(5); while !ready.exists() { assert!( @@ -3272,7 +3303,10 @@ mod tests { ); tokio::time::sleep(Duration::from_millis(10)).await; } - drop(child); + } + + #[cfg(any(unix, windows))] + async fn assert_test_child_killed(survived: &Path) { tokio::time::sleep(Duration::from_millis(1500)).await; assert!( From 4b67f3e590804ba0640abc022f8a2a9812db1c0c Mon Sep 17 00:00:00 2001 From: Stephen Toub Date: Fri, 14 Aug 2026 14:05:38 -0400 Subject: [PATCH 5/5] test(rust): tolerate slow child startup Generated by Copilot Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- rust/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust/src/lib.rs b/rust/src/lib.rs index ee9d91ca6..8921fd49a 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -3295,7 +3295,7 @@ mod tests { #[cfg(any(unix, windows))] async fn wait_for_test_child(ready: &Path) { - let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); while !ready.exists() { assert!( tokio::time::Instant::now() < deadline,