From 5ac6b1fbe59caeb55f0289ed63a5166c35da86ca Mon Sep 17 00:00:00 2001 From: Luke Hoban Date: Thu, 6 Aug 2026 22:41:36 -0700 Subject: [PATCH 1/2] 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 8de6797989..b4445c0b63 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 0f18a9b159..3fdee0f2ef 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 6e05bbfae1..ddd57d854d 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 cafa3c5968..12bf6c563c 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 0000000000..01d0a7e3e2 --- /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/2] 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 01d0a7e3e2..88c0ad8498 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};