From 3e08cf73b5775651bf69b6affb28595b48e02526 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:43:48 +0000 Subject: [PATCH 01/16] fix: own foreground command groups through cancellation and drop --- .github/workflows/foreground-cancellation.yml | 27 ++++ Cargo.lock | 1 + libs/mcp/server/Cargo.toml | 3 + libs/mcp/server/src/foreground_command.rs | 98 ++++++++++++ libs/mcp/server/src/lib.rs | 1 + libs/mcp/server/src/local_tools.rs | 142 +++++++++++++++++- 6 files changed, 264 insertions(+), 8 deletions(-) create mode 100644 .github/workflows/foreground-cancellation.yml create mode 100644 libs/mcp/server/src/foreground_command.rs diff --git a/.github/workflows/foreground-cancellation.yml b/.github/workflows/foreground-cancellation.yml new file mode 100644 index 000000000..537aecc87 --- /dev/null +++ b/.github/workflows/foreground-cancellation.yml @@ -0,0 +1,27 @@ +name: Foreground cancellation qualification +on: + push: + branches: [fix/foreground-command-cancellation] + pull_request: + paths: + - 'libs/mcp/server/**' + - '.github/workflows/foreground-cancellation.yml' +permissions: + contents: read +jobs: + linux: + runs-on: ubuntu-22.04 + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@master + with: + toolchain: '1.94.1' + components: rustfmt + - uses: Swatinem/rust-cache@v2 + - run: cargo fmt --all -- --check + - run: cargo test --locked -p stakpak-mcp-server foreground_ -- --nocapture + - run: cargo build --locked --release -p stakpak + - uses: actions/upload-artifact@v4 + with: + name: stakpak-linux-prerelease-${{ github.sha }} + path: target/release/stakpak diff --git a/Cargo.lock b/Cargo.lock index ddd572cc5..50e566d6d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6326,6 +6326,7 @@ dependencies = [ "grep-regex", "grep-searcher", "ignore", + "libc", "rand 0.9.2", "reqwest", "rmcp", diff --git a/libs/mcp/server/Cargo.toml b/libs/mcp/server/Cargo.toml index 976183c7a..c40ba6dee 100644 --- a/libs/mcp/server/Cargo.toml +++ b/libs/mcp/server/Cargo.toml @@ -35,6 +35,9 @@ grep-searcher = { workspace = true } ignore = { workspace = true } globset = { workspace = true } +[target.'cfg(unix)'.dependencies] +libc = "0.2" + [dev-dependencies] tempfile = "3.8" diff --git a/libs/mcp/server/src/foreground_command.rs b/libs/mcp/server/src/foreground_command.rs new file mode 100644 index 000000000..f7f639830 --- /dev/null +++ b/libs/mcp/server/src/foreground_command.rs @@ -0,0 +1,98 @@ +use std::io; +use std::process::ExitStatus; +use tokio::process::Child; + +/// Owns one foreground command, never the persistent MCP server's group. +/// On Unix the unreaped leader pins the PGID until the last group signal. +/// Descendants that deliberately leave this group are outside this boundary. +pub(crate) struct ForegroundCommand { + child: Option, +} + +impl ForegroundCommand { + pub(crate) fn new(child: Child) -> Self { + Self { child: Some(child) } + } + + pub(crate) fn child_mut(&mut self) -> &mut Child { + // The child is taken only by Drop, after all methods have returned. + #[allow(clippy::expect_used)] + self.child.as_mut().expect("foreground child is owned") + } + + #[cfg(unix)] + fn kill_group(&mut self) -> io::Result<()> { + if let Some(pid) = self.child_mut().id() { + // SAFETY: process_group(0) made the owned, unreaped child its group + // leader. No wait/try_wait may reap it before this signal. + if unsafe { libc::kill(-(pid as libc::pid_t), libc::SIGKILL) } == -1 { + let error = io::Error::last_os_error(); + if error.raw_os_error() != Some(libc::ESRCH) { + return Err(error); + } + } + } + Ok(()) + } + + pub(crate) async fn terminate(&mut self) -> io::Result { + #[cfg(unix)] + self.kill_group()?; + #[cfg(not(unix))] + self.child_mut().start_kill()?; + self.child_mut().wait().await + } + + #[cfg(unix)] + pub(crate) fn has_exited(&mut self) -> io::Result { + let Some(pid) = self.child_mut().id() else { + return Ok(true); + }; + let mut info = std::mem::MaybeUninit::::zeroed(); + // SAFETY: info is writable. WNOWAIT preserves the owned leader's PID + // until kill_group finishes, including when it has become a zombie. + let rc = unsafe { + libc::waitid( + libc::P_PID, + pid as libc::id_t, + info.as_mut_ptr(), + libc::WEXITED | libc::WNOHANG | libc::WNOWAIT, + ) + }; + if rc == -1 { + return Err(io::Error::last_os_error()); + } + // SAFETY: successful waitid initializes info; si_pid is zero when + // WNOHANG finds no waitable child. + Ok(unsafe { info.assume_init().si_pid() } != 0) + } + + pub(crate) async fn wait(&mut self) -> io::Result { + #[cfg(unix)] + { + while !self.has_exited()? { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + self.kill_group()?; + } + self.child_mut().wait().await + } + +} + +impl Drop for ForegroundCommand { + fn drop(&mut self) { + #[cfg(unix)] + let _ = self.kill_group(); + if let Some(mut child) = self.child.take() { + let _ = child.start_kill(); + // Reaping survives an aborted request future. Runtime shutdown is + // still responsible for driving Tokio's process reaper. + if let Ok(runtime) = tokio::runtime::Handle::try_current() { + runtime.spawn(async move { + let _ = child.wait().await; + }); + } + } + } +} diff --git a/libs/mcp/server/src/lib.rs b/libs/mcp/server/src/lib.rs index d32723262..6b330854d 100644 --- a/libs/mcp/server/src/lib.rs +++ b/libs/mcp/server/src/lib.rs @@ -17,6 +17,7 @@ use stakpak_api::AgentProvider; use stakpak_shared::cert_utils::CertificateChain; use stakpak_shared::task_manager::{TaskManager, TaskManagerHandle}; +mod foreground_command; pub mod integrations; pub mod local_tools; pub mod remote_tools; diff --git a/libs/mcp/server/src/local_tools.rs b/libs/mcp/server/src/local_tools.rs index 9e742836c..c2823603c 100644 --- a/libs/mcp/server/src/local_tools.rs +++ b/libs/mcp/server/src/local_tools.rs @@ -1,3 +1,4 @@ +use crate::foreground_command::ForegroundCommand; use crate::tool_container::ToolContainer; use rmcp::service::RequestContext; use rmcp::{ErrorData as McpError, handler::server::wrapper::Parameters, model::*, schemars, tool}; @@ -1351,7 +1352,7 @@ SAFETY NOTES: cmd.creation_flags(0x00000200); // CREATE_NEW_PROCESS_GROUP } - let mut child = cmd.spawn().map_err(|e| { + let child = cmd.spawn().map_err(|e| { error!("Failed to run command: {}", e); CallToolResult::error(vec![ Content::text("COMMAND_ERROR"), @@ -1359,10 +1360,11 @@ SAFETY NOTES: ]) })?; + let mut child = ForegroundCommand::new(child); #[allow(clippy::unwrap_used)] - let stdout = child.stdout.take().unwrap(); + let stdout = child.child_mut().stdout.take().unwrap(); #[allow(clippy::unwrap_used)] - let stderr = child.stderr.take().unwrap(); + let stderr = child.child_mut().stderr.take().unwrap(); let mut stdout_reader = BufReader::new(stdout); let mut stderr_reader = BufReader::new(stderr); let mut stdout_buf = String::new(); @@ -1461,8 +1463,8 @@ SAFETY NOTES: } } - // Check if process has exited - if let Ok(Some(_)) = child.try_wait() { + #[cfg(unix)] + if child.has_exited()? { break; } } @@ -1479,7 +1481,7 @@ SAFETY NOTES: result = tokio::time::timeout(timeout_duration, stream_and_wait) => result, _ = ctx.ct.cancelled() => { // Cancellation occurred, kill the process - let _ = child.kill().await; + let _ = child.terminate().await; return Err(CallToolResult::cancel(Some(&vec![ Content::text("COMMAND_CANCELLED"), Content::text("Command execution was cancelled"), @@ -1490,7 +1492,7 @@ SAFETY NOTES: tokio::select! { result = stream_and_wait => Ok(result), _ = ctx.ct.cancelled() => { - let _ = child.kill().await; + let _ = child.terminate().await; return Err(CallToolResult::cancel(Some(&vec![ Content::text("COMMAND_CANCELLED"), Content::text("Command execution was cancelled"), @@ -1509,7 +1511,7 @@ SAFETY NOTES: } Err(_) => { // Timeout occurred, kill the process - let _ = child.kill().await; + let _ = child.terminate().await; result.push_str(&format!( "Command timed out after {} seconds\n", timeout.unwrap_or_default() @@ -3308,6 +3310,130 @@ fn unicode_normalized_replace( mod tests { use super::*; + #[cfg(target_os = "linux")] + mod foreground_regressions { + use super::*; + use rmcp::service::PeerRequestOptions; + use rmcp::{ServiceExt, model::ClientRequest}; + + async fn wait_for_marker(path: &Path) -> u32 { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let Ok(text) = fs::read_to_string(path) + && let Ok(pid) = text.trim().parse() + { + return pid; + } + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("nested shell must author its start marker") + } + + async fn assert_gone(pid: u32) { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let stat = fs::read_to_string(format!("/proc/{pid}/stat")); + // A killed orphan may briefly remain a zombie under CI's + // init; it cannot execute or create the delayed artifact. + if stat.as_ref().map_or(true, |s| { + s.rsplit_once(") ").is_some_and(|(_, rest)| rest.starts_with('Z')) + }) { + break; + } + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("owned shell must stop executing"); + } + + async fn exercise(kind: &str, configured_timeout: Option) { + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join("started"); + let late = dir.path().join("late"); + let command = format!( + "sh -c 'echo $$ > "{}"; sleep 2; echo late > "{}"' & wait", + marker.display(), late.display() + ); + let mut control = Command::new("sleep").arg("30").spawn().unwrap(); + let (client_io, server_io) = tokio::io::duplex(65536); + let server = tokio::spawn(async move { + local_container_with_profile(None).serve(server_io).await.unwrap() + }); + let client = ().serve(client_io).await.unwrap(); + let server = server.await.unwrap(); + let request = client + .send_cancellable_request( + ClientRequest::CallToolRequest(Request::new(CallToolRequestParam { + name: "run_command".into(), + arguments: Some(serde_json::json!({ + "command": command, + "timeout": configured_timeout, + }).as_object().unwrap().clone()), + ..Default::default() + })), + PeerRequestOptions::default(), + ) + .await + .unwrap(); + let pid = wait_for_marker(&marker).await; + if kind == "disconnect" { + client.cancel().await.unwrap(); + server.cancel().await.unwrap(); + } else { + if kind == "cancel" { + client.notify_cancelled(CancelledNotificationParam { + request_id: request.id.clone(), + reason: Some("regression cancellation".into()), + }).await.unwrap(); + } + let result = tokio::time::timeout(Duration::from_secs(5), request.await_response()) + .await.expect("executor must settle boundedly"); + if kind == "timeout" { + let result = result.unwrap(); + assert!(format!("{result:?}").contains("timed out")); + } + // The same MCP session must still execute a subsequent command. + let next = client.call_tool(CallToolRequestParam { + name: "run_command".into(), + arguments: Some(serde_json::json!({"command": "printf subsequent"}) + .as_object().unwrap().clone()), + ..Default::default() + }).await.unwrap(); + assert!(format!("{next:?}").contains("subsequent")); + client.cancel().await.unwrap(); + server.cancel().await.unwrap(); + } + assert_gone(pid).await; + assert!(control.try_wait().unwrap().is_none(), "unrelated process survives"); + sleep(Duration::from_millis(2200)).await; + assert!(!late.exists(), "cancelled descendant must not write later"); + control.kill().await.unwrap(); + } + + #[tokio::test] + async fn foreground_cancel_without_timeout() { + exercise("cancel", None).await; + } + + #[tokio::test] + async fn foreground_cancel_with_timeout() { + exercise("cancel", Some(30)).await; + } + + #[tokio::test] + async fn foreground_timeout() { + exercise("timeout", Some(1)).await; + } + + #[tokio::test] + async fn foreground_disconnect() { + exercise("disconnect", None).await; + } + } + #[test] fn run_remote_command_request_whitespace_remote_deserializes_but_validated_at_runtime() { let result = serde_json::from_value::(serde_json::json!({ From fbc096afb0adf5561cd405947072e47d9e8a09fd Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:44:50 +0000 Subject: [PATCH 02/16] fix: preserve successful command semantics and bound explicit cleanup --- libs/mcp/server/src/foreground_command.rs | 84 ++++-------- libs/mcp/server/src/foreground_tests.rs | 121 +++++++++++++++++ libs/mcp/server/src/local_tools.rs | 154 ++++------------------ 3 files changed, 168 insertions(+), 191 deletions(-) create mode 100644 libs/mcp/server/src/foreground_tests.rs diff --git a/libs/mcp/server/src/foreground_command.rs b/libs/mcp/server/src/foreground_command.rs index f7f639830..fd4007e48 100644 --- a/libs/mcp/server/src/foreground_command.rs +++ b/libs/mcp/server/src/foreground_command.rs @@ -2,30 +2,33 @@ use std::io; use std::process::ExitStatus; use tokio::process::Child; -/// Owns one foreground command, never the persistent MCP server's group. -/// On Unix the unreaped leader pins the PGID until the last group signal. -/// Descendants that deliberately leave this group are outside this boundary. +/// Owns a foreground child's group until the child is reaped. Commands that +/// deliberately leave this group are outside this cancellation boundary. pub(crate) struct ForegroundCommand { - child: Option, + child: Child, } impl ForegroundCommand { pub(crate) fn new(child: Child) -> Self { - Self { child: Some(child) } + Self { child } } pub(crate) fn child_mut(&mut self) -> &mut Child { - // The child is taken only by Drop, after all methods have returned. - #[allow(clippy::expect_used)] - self.child.as_mut().expect("foreground child is owned") + &mut self.child } #[cfg(unix)] fn kill_group(&mut self) -> io::Result<()> { - if let Some(pid) = self.child_mut().id() { - // SAFETY: process_group(0) made the owned, unreaped child its group - // leader. No wait/try_wait may reap it before this signal. - if unsafe { libc::kill(-(pid as libc::pid_t), libc::SIGKILL) } == -1 { + if let Some(pid) = self.child.id() { + let pid = libc::pid_t::try_from(pid) + .map_err(|_| io::Error::other("invalid foreground process ID"))?; + if pid <= 1 { + return Err(io::Error::other("invalid foreground process group")); + } + // SAFETY: process_group(0) made this owned child the group leader. + // Child::id() becomes None after wait/try_wait successfully reaps it; + // while unreaped (even a zombie), its PID cannot be reused. + if unsafe { libc::kill(-pid, libc::SIGKILL) } == -1 { let error = io::Error::last_os_error(); if error.raw_os_error() != Some(libc::ESRCH) { return Err(error); @@ -39,60 +42,21 @@ impl ForegroundCommand { #[cfg(unix)] self.kill_group()?; #[cfg(not(unix))] - self.child_mut().start_kill()?; - self.child_mut().wait().await - } - - #[cfg(unix)] - pub(crate) fn has_exited(&mut self) -> io::Result { - let Some(pid) = self.child_mut().id() else { - return Ok(true); - }; - let mut info = std::mem::MaybeUninit::::zeroed(); - // SAFETY: info is writable. WNOWAIT preserves the owned leader's PID - // until kill_group finishes, including when it has become a zombie. - let rc = unsafe { - libc::waitid( - libc::P_PID, - pid as libc::id_t, - info.as_mut_ptr(), - libc::WEXITED | libc::WNOHANG | libc::WNOWAIT, - ) - }; - if rc == -1 { - return Err(io::Error::last_os_error()); - } - // SAFETY: successful waitid initializes info; si_pid is zero when - // WNOHANG finds no waitable child. - Ok(unsafe { info.assume_init().si_pid() } != 0) - } - - pub(crate) async fn wait(&mut self) -> io::Result { - #[cfg(unix)] - { - while !self.has_exited()? { - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - self.kill_group()?; - } - self.child_mut().wait().await + self.child.start_kill()?; + tokio::time::timeout(std::time::Duration::from_secs(5), self.child.wait()) + .await + .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "foreground cleanup timed out"))? } - } impl Drop for ForegroundCommand { fn drop(&mut self) { #[cfg(unix)] - let _ = self.kill_group(); - if let Some(mut child) = self.child.take() { - let _ = child.start_kill(); - // Reaping survives an aborted request future. Runtime shutdown is - // still responsible for driving Tokio's process reaper. - if let Ok(runtime) = tokio::runtime::Handle::try_current() { - runtime.spawn(async move { - let _ = child.wait().await; - }); - } + if let Err(error) = self.kill_group() { + tracing::error!("Failed to kill foreground command group: {error}"); } + // The child was spawned with kill_on_drop(true). Tokio handles direct + // child fallback/reaping; group signalling must happen synchronously + // first, including when a request future is aborted or dropped. } } diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs new file mode 100644 index 000000000..c51be8e45 --- /dev/null +++ b/libs/mcp/server/src/foreground_tests.rs @@ -0,0 +1,121 @@ +use super::*; +use super::tests::local_container_with_profile; +use rmcp::service::PeerRequestOptions; +use rmcp::{ServiceExt, model::ClientRequest}; + +async fn wait_for_marker(path: &Path) -> u32 { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let Ok(text) = fs::read_to_string(path) + && let Ok(pid) = text.trim().parse() + { + return pid; + } + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("nested shell must author its start marker") +} + +async fn assert_gone(pid: u32) { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let stat = fs::read_to_string(format!("/proc/{pid}/stat")); + // A killed orphan may briefly remain a zombie under CI's + // init; it cannot execute or create the delayed artifact. + if stat.as_ref().map_or(true, |s| { + s.rsplit_once(") ").is_some_and(|(_, rest)| rest.starts_with('Z')) + }) { + break; + } + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("owned shell must stop executing"); +} + +async fn exercise(kind: &str, configured_timeout: Option) { + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join("started"); + let late = dir.path().join("late"); + let command = format!( + "sh -c 'echo $$ > "{}"; sleep 2; echo late > "{}"' & wait", + marker.display(), late.display() + ); + let mut control = Command::new("sleep").arg("30").spawn().unwrap(); + let (client_io, server_io) = tokio::io::duplex(65536); + let server = tokio::spawn(async move { + local_container_with_profile(None).serve(server_io).await.unwrap() + }); + let client = ().serve(client_io).await.unwrap(); + let server = server.await.unwrap(); + let request = client + .send_cancellable_request( + ClientRequest::CallToolRequest(Request::new(CallToolRequestParam { + name: "run_command".into(), + arguments: Some(serde_json::json!({ + "command": command, + "timeout": configured_timeout, + }).as_object().unwrap().clone()), + ..Default::default() + })), + PeerRequestOptions::default(), + ) + .await + .unwrap(); + let pid = wait_for_marker(&marker).await; + if kind == "disconnect" { + client.cancel().await.unwrap(); + server.cancel().await.unwrap(); + } else { + if kind == "cancel" { + client.notify_cancelled(CancelledNotificationParam { + request_id: request.id.clone(), + reason: Some("regression cancellation".into()), + }).await.unwrap(); + } + let result = tokio::time::timeout(Duration::from_secs(5), request.await_response()) + .await.expect("executor must settle boundedly"); + if kind == "timeout" { + let result = result.unwrap(); + assert!(format!("{result:?}").contains("timed out")); + } + // The same MCP session must still execute a subsequent command. + let next = client.call_tool(CallToolRequestParam { + name: "run_command".into(), + arguments: Some(serde_json::json!({"command": "printf subsequent"}) + .as_object().unwrap().clone()), + ..Default::default() + }).await.unwrap(); + assert!(format!("{next:?}").contains("subsequent")); + client.cancel().await.unwrap(); + server.cancel().await.unwrap(); + } + assert_gone(pid).await; + assert!(control.try_wait().unwrap().is_none(), "unrelated process survives"); + sleep(Duration::from_millis(2200)).await; + assert!(!late.exists(), "cancelled descendant must not write later"); + control.kill().await.unwrap(); +} + +#[tokio::test] +async fn foreground_cancel_without_timeout() { + exercise("cancel", None).await; +} + +#[tokio::test] +async fn foreground_cancel_with_timeout() { + exercise("cancel", Some(30)).await; +} + +#[tokio::test] +async fn foreground_timeout() { + exercise("timeout", Some(1)).await; +} + +#[tokio::test] +async fn foreground_disconnect() { + exercise("disconnect", None).await; +} diff --git a/libs/mcp/server/src/local_tools.rs b/libs/mcp/server/src/local_tools.rs index c2823603c..1396070b1 100644 --- a/libs/mcp/server/src/local_tools.rs +++ b/libs/mcp/server/src/local_tools.rs @@ -1336,6 +1336,7 @@ SAFETY NOTES: let mut cmd = Command::new("sh"); cmd.arg("-c") .arg(actual_command) + .kill_on_drop(true) .stdin(std::process::Stdio::null()) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()); @@ -1463,14 +1464,13 @@ SAFETY NOTES: } } - #[cfg(unix)] - if child.has_exited()? { + if let Ok(Some(_)) = child.child_mut().try_wait() { break; } } // Wait for the process to complete - child.wait().await + child.child_mut().wait().await }; // Execute with timeout and cancellation support @@ -1481,7 +1481,11 @@ SAFETY NOTES: result = tokio::time::timeout(timeout_duration, stream_and_wait) => result, _ = ctx.ct.cancelled() => { // Cancellation occurred, kill the process - let _ = child.terminate().await; + if let Err(error) = child.terminate().await { + return Err(CallToolResult::error(vec![Content::text(format!( + "Foreground command cleanup failed: {error}" + ))])); + } return Err(CallToolResult::cancel(Some(&vec![ Content::text("COMMAND_CANCELLED"), Content::text("Command execution was cancelled"), @@ -1492,7 +1496,11 @@ SAFETY NOTES: tokio::select! { result = stream_and_wait => Ok(result), _ = ctx.ct.cancelled() => { - let _ = child.terminate().await; + if let Err(error) = child.terminate().await { + return Err(CallToolResult::error(vec![Content::text(format!( + "Foreground command cleanup failed: {error}" + ))])); + } return Err(CallToolResult::cancel(Some(&vec![ Content::text("COMMAND_CANCELLED"), Content::text("Command execution was cancelled"), @@ -1511,7 +1519,11 @@ SAFETY NOTES: } Err(_) => { // Timeout occurred, kill the process - let _ = child.terminate().await; + if let Err(error) = child.terminate().await { + return Err(CallToolResult::error(vec![Content::text(format!( + "Foreground command cleanup failed: {error}" + ))])); + } result.push_str(&format!( "Command timed out after {} seconds\n", timeout.unwrap_or_default() @@ -3310,130 +3322,6 @@ fn unicode_normalized_replace( mod tests { use super::*; - #[cfg(target_os = "linux")] - mod foreground_regressions { - use super::*; - use rmcp::service::PeerRequestOptions; - use rmcp::{ServiceExt, model::ClientRequest}; - - async fn wait_for_marker(path: &Path) -> u32 { - tokio::time::timeout(Duration::from_secs(5), async { - loop { - if let Ok(text) = fs::read_to_string(path) - && let Ok(pid) = text.trim().parse() - { - return pid; - } - sleep(Duration::from_millis(10)).await; - } - }) - .await - .expect("nested shell must author its start marker") - } - - async fn assert_gone(pid: u32) { - tokio::time::timeout(Duration::from_secs(5), async { - loop { - let stat = fs::read_to_string(format!("/proc/{pid}/stat")); - // A killed orphan may briefly remain a zombie under CI's - // init; it cannot execute or create the delayed artifact. - if stat.as_ref().map_or(true, |s| { - s.rsplit_once(") ").is_some_and(|(_, rest)| rest.starts_with('Z')) - }) { - break; - } - sleep(Duration::from_millis(10)).await; - } - }) - .await - .expect("owned shell must stop executing"); - } - - async fn exercise(kind: &str, configured_timeout: Option) { - let dir = tempfile::tempdir().unwrap(); - let marker = dir.path().join("started"); - let late = dir.path().join("late"); - let command = format!( - "sh -c 'echo $$ > "{}"; sleep 2; echo late > "{}"' & wait", - marker.display(), late.display() - ); - let mut control = Command::new("sleep").arg("30").spawn().unwrap(); - let (client_io, server_io) = tokio::io::duplex(65536); - let server = tokio::spawn(async move { - local_container_with_profile(None).serve(server_io).await.unwrap() - }); - let client = ().serve(client_io).await.unwrap(); - let server = server.await.unwrap(); - let request = client - .send_cancellable_request( - ClientRequest::CallToolRequest(Request::new(CallToolRequestParam { - name: "run_command".into(), - arguments: Some(serde_json::json!({ - "command": command, - "timeout": configured_timeout, - }).as_object().unwrap().clone()), - ..Default::default() - })), - PeerRequestOptions::default(), - ) - .await - .unwrap(); - let pid = wait_for_marker(&marker).await; - if kind == "disconnect" { - client.cancel().await.unwrap(); - server.cancel().await.unwrap(); - } else { - if kind == "cancel" { - client.notify_cancelled(CancelledNotificationParam { - request_id: request.id.clone(), - reason: Some("regression cancellation".into()), - }).await.unwrap(); - } - let result = tokio::time::timeout(Duration::from_secs(5), request.await_response()) - .await.expect("executor must settle boundedly"); - if kind == "timeout" { - let result = result.unwrap(); - assert!(format!("{result:?}").contains("timed out")); - } - // The same MCP session must still execute a subsequent command. - let next = client.call_tool(CallToolRequestParam { - name: "run_command".into(), - arguments: Some(serde_json::json!({"command": "printf subsequent"}) - .as_object().unwrap().clone()), - ..Default::default() - }).await.unwrap(); - assert!(format!("{next:?}").contains("subsequent")); - client.cancel().await.unwrap(); - server.cancel().await.unwrap(); - } - assert_gone(pid).await; - assert!(control.try_wait().unwrap().is_none(), "unrelated process survives"); - sleep(Duration::from_millis(2200)).await; - assert!(!late.exists(), "cancelled descendant must not write later"); - control.kill().await.unwrap(); - } - - #[tokio::test] - async fn foreground_cancel_without_timeout() { - exercise("cancel", None).await; - } - - #[tokio::test] - async fn foreground_cancel_with_timeout() { - exercise("cancel", Some(30)).await; - } - - #[tokio::test] - async fn foreground_timeout() { - exercise("timeout", Some(1)).await; - } - - #[tokio::test] - async fn foreground_disconnect() { - exercise("disconnect", None).await; - } - } - #[test] fn run_remote_command_request_whitespace_remote_deserializes_but_validated_at_runtime() { let result = serde_json::from_value::(serde_json::json!({ @@ -3496,7 +3384,7 @@ mod tests { ); } - fn local_container_with_profile(profile_name: Option<&str>) -> ToolContainer { + pub(super) fn local_container_with_profile(profile_name: Option<&str>) -> ToolContainer { let task_manager = stakpak_shared::task_manager::TaskManager::new(); ToolContainer::new( None, @@ -4222,3 +4110,7 @@ mod tests { assert_eq!(result, Some(("a***b***c".to_string(), 2))); } } + +#[cfg(all(test, target_os = "linux"))] +#[path = "foreground_tests.rs"] +mod foreground_tests; From e4ce078ffa1bec2a86741c2f8024e1416a76026c Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:45:09 +0000 Subject: [PATCH 03/16] ci: retain remote formatting corrections alongside qualification --- .github/workflows/foreground-cancellation.yml | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/.github/workflows/foreground-cancellation.yml b/.github/workflows/foreground-cancellation.yml index 537aecc87..1a20c5cd8 100644 --- a/.github/workflows/foreground-cancellation.yml +++ b/.github/workflows/foreground-cancellation.yml @@ -18,9 +18,17 @@ jobs: toolchain: '1.94.1' components: rustfmt - uses: Swatinem/rust-cache@v2 - - run: cargo fmt --all -- --check + - name: Generate formatting receipt + run: | + cargo fmt --all + git diff > formatting.patch + - uses: actions/upload-artifact@v4 + with: + name: formatting-${{ github.sha }} + path: formatting.patch - run: cargo test --locked -p stakpak-mcp-server foreground_ -- --nocapture - run: cargo build --locked --release -p stakpak + - run: git diff --exit-code - uses: actions/upload-artifact@v4 with: name: stakpak-linux-prerelease-${{ github.sha }} From bfe38ec8c891a904e44e2376a15698c76f02eabc Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:46:32 +0000 Subject: [PATCH 04/16] test: retain remote formatting and track nested shell and sleep --- libs/mcp/server/src/foreground_tests.rs | 74 +++++++++++++++++-------- libs/mcp/server/src/local_tools.rs | 8 +-- 2 files changed, 56 insertions(+), 26 deletions(-) diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index c51be8e45..bee305c6e 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -1,5 +1,5 @@ -use super::*; use super::tests::local_container_with_profile; +use super::*; use rmcp::service::PeerRequestOptions; use rmcp::{ServiceExt, model::ClientRequest}; @@ -25,7 +25,8 @@ async fn assert_gone(pid: u32) { // A killed orphan may briefly remain a zombie under CI's // init; it cannot execute or create the delayed artifact. if stat.as_ref().map_or(true, |s| { - s.rsplit_once(") ").is_some_and(|(_, rest)| rest.starts_with('Z')) + s.rsplit_once(") ") + .is_some_and(|(_, rest)| rest.starts_with('Z')) }) { break; } @@ -40,14 +41,19 @@ async fn exercise(kind: &str, configured_timeout: Option) { let dir = tempfile::tempdir().unwrap(); let marker = dir.path().join("started"); let late = dir.path().join("late"); + let outer = dir.path().join("outer"); + let sleeper = dir.path().join("sleeper"); let command = format!( - "sh -c 'echo $$ > "{}"; sleep 2; echo late > "{}"' & wait", - marker.display(), late.display() + r#"echo $$ > "{}"; sh -c 'echo $$ > "{}"; sleep 2 & echo $! > "{}"; wait; echo late > "{}"' & wait"#, + outer.display(), marker.display(), sleeper.display(), late.display() ); - let mut control = Command::new("sleep").arg("30").spawn().unwrap(); + let mut control = Command::new("sleep").arg("30").kill_on_drop(true).spawn().unwrap(); let (client_io, server_io) = tokio::io::duplex(65536); let server = tokio::spawn(async move { - local_container_with_profile(None).serve(server_io).await.unwrap() + local_container_with_profile(None) + .serve(server_io) + .await + .unwrap() }); let client = ().serve(client_io).await.unwrap(); let server = server.await.unwrap(); @@ -55,10 +61,15 @@ async fn exercise(kind: &str, configured_timeout: Option) { .send_cancellable_request( ClientRequest::CallToolRequest(Request::new(CallToolRequestParam { name: "run_command".into(), - arguments: Some(serde_json::json!({ - "command": command, - "timeout": configured_timeout, - }).as_object().unwrap().clone()), + arguments: Some( + serde_json::json!({ + "command": command, + "timeout": configured_timeout, + }) + .as_object() + .unwrap() + .clone(), + ), ..Default::default() })), PeerRequestOptions::default(), @@ -66,37 +77,56 @@ async fn exercise(kind: &str, configured_timeout: Option) { .await .unwrap(); let pid = wait_for_marker(&marker).await; + let outer_pid = wait_for_marker(&outer).await; + let sleeper_pid = wait_for_marker(&sleeper).await; if kind == "disconnect" { client.cancel().await.unwrap(); server.cancel().await.unwrap(); } else { if kind == "cancel" { - client.notify_cancelled(CancelledNotificationParam { - request_id: request.id.clone(), - reason: Some("regression cancellation".into()), - }).await.unwrap(); + client + .notify_cancelled(CancelledNotificationParam { + request_id: request.id.clone(), + reason: Some("regression cancellation".into()), + }) + .await + .unwrap(); } let result = tokio::time::timeout(Duration::from_secs(5), request.await_response()) - .await.expect("executor must settle boundedly"); + .await + .expect("executor must settle boundedly"); if kind == "timeout" { let result = result.unwrap(); assert!(format!("{result:?}").contains("timed out")); } // The same MCP session must still execute a subsequent command. - let next = client.call_tool(CallToolRequestParam { - name: "run_command".into(), - arguments: Some(serde_json::json!({"command": "printf subsequent"}) - .as_object().unwrap().clone()), - ..Default::default() - }).await.unwrap(); + let next = client + .call_tool(CallToolRequestParam { + name: "run_command".into(), + arguments: Some( + serde_json::json!({"command": "printf subsequent"}) + .as_object() + .unwrap() + .clone(), + ), + ..Default::default() + }) + .await + .unwrap(); assert!(format!("{next:?}").contains("subsequent")); client.cancel().await.unwrap(); server.cancel().await.unwrap(); } assert_gone(pid).await; - assert!(control.try_wait().unwrap().is_none(), "unrelated process survives"); + assert_gone(outer_pid).await; + assert_gone(sleeper_pid).await; + assert!( + control.try_wait().unwrap().is_none(), + "unrelated process survives" + ); sleep(Duration::from_millis(2200)).await; assert!(!late.exists(), "cancelled descendant must not write later"); + assert!(control.try_wait().unwrap().is_none(), "control survives deadline"); control.kill().await.unwrap(); } diff --git a/libs/mcp/server/src/local_tools.rs b/libs/mcp/server/src/local_tools.rs index 1396070b1..46afe7861 100644 --- a/libs/mcp/server/src/local_tools.rs +++ b/libs/mcp/server/src/local_tools.rs @@ -1520,10 +1520,10 @@ SAFETY NOTES: Err(_) => { // Timeout occurred, kill the process if let Err(error) = child.terminate().await { - return Err(CallToolResult::error(vec![Content::text(format!( - "Foreground command cleanup failed: {error}" - ))])); - } + return Err(CallToolResult::error(vec![Content::text(format!( + "Foreground command cleanup failed: {error}" + ))])); + } result.push_str(&format!( "Command timed out after {} seconds\n", timeout.unwrap_or_default() From 06e22afb8b92a761a50d675a1193e0b2dee49cf1 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:47:59 +0000 Subject: [PATCH 05/16] test: exercise executor abort and process identities in CI --- .github/workflows/foreground-cancellation.yml | 11 +++- libs/mcp/server/src/foreground_tests.rs | 66 ++++++++++++++++--- 2 files changed, 67 insertions(+), 10 deletions(-) diff --git a/.github/workflows/foreground-cancellation.yml b/.github/workflows/foreground-cancellation.yml index 1a20c5cd8..f3051e8ac 100644 --- a/.github/workflows/foreground-cancellation.yml +++ b/.github/workflows/foreground-cancellation.yml @@ -16,8 +16,10 @@ jobs: - uses: dtolnay/rust-toolchain@master with: toolchain: '1.94.1' - components: rustfmt + components: rustfmt, clippy - uses: Swatinem/rust-cache@v2 + with: + cache-on-failure: true - name: Generate formatting receipt run: | cargo fmt --all @@ -27,6 +29,13 @@ jobs: name: formatting-${{ github.sha }} path: formatting.patch - run: cargo test --locked -p stakpak-mcp-server foreground_ -- --nocapture + - run: cargo clippy --locked --all-targets -- -D warnings + - run: cargo test --locked --workspace + - run: | + cargo test --locked -p stakpak-shared --features sqlite + cargo test --locked -p stakpak --features libsql-test + cargo test --locked -p stakpak-gateway --features libsql-test + cargo test --locked -p stakai --features network-tests - run: cargo build --locked --release -p stakpak - run: git diff --exit-code - uses: actions/upload-artifact@v4 diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index bee305c6e..8754201db 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -3,6 +3,45 @@ use super::*; use rmcp::service::PeerRequestOptions; use rmcp::{ServiceExt, model::ClientRequest}; +struct ExecutionHarness { + container: ToolContainer, + abort_execution: bool, +} + +impl rmcp::ServerHandler for ExecutionHarness { + fn get_info(&self) -> ServerInfo { + rmcp::ServerHandler::get_info(&self.container) + } + + async fn call_tool( + &self, + request: CallToolRequestParam, + ctx: RequestContext, + ) -> Result { + let command = request.arguments.as_ref() + .and_then(|args| args.get("command")) + .and_then(serde_json::Value::as_str).unwrap_or_default().to_owned(); + if !self.abort_execution || command == "printf subsequent" { + return rmcp::ServerHandler::call_tool(&self.container, request, ctx).await; + } + let container = self.container.clone(); + let ct = ctx.ct.clone(); + let execution = tokio::spawn(async move { + container.execute_local_command(&command, None, &ctx).await + }); + ct.cancelled().await; + execution.abort(); + assert!(execution.await.unwrap_err().is_cancelled()); + Ok(CallToolResult::cancel(None)) + } +} + +fn process_identity(pid: u32) -> Option { + let stat = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + let (_, fields) = stat.rsplit_once(") ")?; + fields.split_whitespace().nth(19).map(str::to_owned) +} + async fn wait_for_marker(path: &Path) -> u32 { tokio::time::timeout(Duration::from_secs(5), async { loop { @@ -18,13 +57,13 @@ async fn wait_for_marker(path: &Path) -> u32 { .expect("nested shell must author its start marker") } -async fn assert_gone(pid: u32) { +async fn assert_gone(pid: u32, identity: &str) { tokio::time::timeout(Duration::from_secs(5), async { loop { let stat = fs::read_to_string(format!("/proc/{pid}/stat")); // A killed orphan may briefly remain a zombie under CI's // init; it cannot execute or create the delayed artifact. - if stat.as_ref().map_or(true, |s| { + if process_identity(pid).as_deref() != Some(identity) || stat.as_ref().map_or(true, |s| { s.rsplit_once(") ") .is_some_and(|(_, rest)| rest.starts_with('Z')) }) { @@ -48,9 +87,12 @@ async fn exercise(kind: &str, configured_timeout: Option) { outer.display(), marker.display(), sleeper.display(), late.display() ); let mut control = Command::new("sleep").arg("30").kill_on_drop(true).spawn().unwrap(); + let control_pid = control.id().unwrap(); + let control_identity = process_identity(control_pid).unwrap(); let (client_io, server_io) = tokio::io::duplex(65536); + let abort_execution = kind == "abort"; let server = tokio::spawn(async move { - local_container_with_profile(None) + ExecutionHarness { container: local_container_with_profile(None), abort_execution } .serve(server_io) .await .unwrap() @@ -70,7 +112,6 @@ async fn exercise(kind: &str, configured_timeout: Option) { .unwrap() .clone(), ), - ..Default::default() })), PeerRequestOptions::default(), ) @@ -79,11 +120,12 @@ async fn exercise(kind: &str, configured_timeout: Option) { let pid = wait_for_marker(&marker).await; let outer_pid = wait_for_marker(&outer).await; let sleeper_pid = wait_for_marker(&sleeper).await; + let owned = [outer_pid, pid, sleeper_pid].map(|pid| (pid, process_identity(pid).unwrap())); if kind == "disconnect" { client.cancel().await.unwrap(); server.cancel().await.unwrap(); } else { - if kind == "cancel" { + if kind == "cancel" || kind == "abort" { client .notify_cancelled(CancelledNotificationParam { request_id: request.id.clone(), @@ -109,7 +151,6 @@ async fn exercise(kind: &str, configured_timeout: Option) { .unwrap() .clone(), ), - ..Default::default() }) .await .unwrap(); @@ -117,9 +158,10 @@ async fn exercise(kind: &str, configured_timeout: Option) { client.cancel().await.unwrap(); server.cancel().await.unwrap(); } - assert_gone(pid).await; - assert_gone(outer_pid).await; - assert_gone(sleeper_pid).await; + for (pid, identity) in owned { + assert_gone(pid, &identity).await; + } + assert_eq!(process_identity(control_pid), Some(control_identity.clone())); assert!( control.try_wait().unwrap().is_none(), "unrelated process survives" @@ -127,6 +169,7 @@ async fn exercise(kind: &str, configured_timeout: Option) { sleep(Duration::from_millis(2200)).await; assert!(!late.exists(), "cancelled descendant must not write later"); assert!(control.try_wait().unwrap().is_none(), "control survives deadline"); + assert_eq!(process_identity(control_pid), Some(control_identity)); control.kill().await.unwrap(); } @@ -149,3 +192,8 @@ async fn foreground_timeout() { async fn foreground_disconnect() { exercise("disconnect", None).await; } + +#[tokio::test] +async fn foreground_execution_future_abort() { + exercise("abort", None).await; +} From f075b5f1f8a2dc35b48967f1ad94b837ae631a9d Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:49:04 +0000 Subject: [PATCH 06/16] test: require stock failure and cover reaped leader lifetime --- .github/workflows/foreground-cancellation.yml | 18 +++++++++ libs/mcp/server/src/foreground_command.rs | 39 +++++++++++++++++++ 2 files changed, 57 insertions(+) diff --git a/.github/workflows/foreground-cancellation.yml b/.github/workflows/foreground-cancellation.yml index f3051e8ac..fd6c141d2 100644 --- a/.github/workflows/foreground-cancellation.yml +++ b/.github/workflows/foreground-cancellation.yml @@ -29,6 +29,24 @@ jobs: name: formatting-${{ github.sha }} path: formatting.patch - run: cargo test --locked -p stakpak-mcp-server foreground_ -- --nocapture + - name: Stock executor must fail the cancellation regression + shell: bash + run: | + git fetch origin main --depth=1 + git show FETCH_HEAD:libs/mcp/server/src/local_tools.rs > libs/mcp/server/src/local_tools.rs + python3 - <<'PYTHON' + from pathlib import Path + p = Path('libs/mcp/server/src/local_tools.rs') + s = p.read_text().replace(' fn local_container_with_profile(', ' pub(super) fn local_container_with_profile(') + s += '\n#[cfg(all(test, target_os = "linux"))]\n#[path = "foreground_tests.rs"]\nmod foreground_tests;\n' + p.write_text(s) + PYTHON + if cargo test --locked -p stakpak-mcp-server foreground_cancel_without_timeout -- --nocapture; then + echo 'Stock executor unexpectedly passed; regression does not demonstrate the defect' + exit 1 + fi + git restore libs/mcp/server/src/local_tools.rs + cargo fmt --all - run: cargo clippy --locked --all-targets -- -D warnings - run: cargo test --locked --workspace - run: | diff --git a/libs/mcp/server/src/foreground_command.rs b/libs/mcp/server/src/foreground_command.rs index fd4007e48..199f9b383 100644 --- a/libs/mcp/server/src/foreground_command.rs +++ b/libs/mcp/server/src/foreground_command.rs @@ -60,3 +60,42 @@ impl Drop for ForegroundCommand { // first, including when a request future is aborted or dropped. } } + +#[cfg(all(test, unix))] +mod tests { + use super::*; + use tokio::process::Command; + + fn successful_child() -> ForegroundCommand { + ForegroundCommand::new(Command::new("sh") + .arg("-c").arg("exit 7").process_group(0) + .kill_on_drop(true).spawn().unwrap()) + } + + #[tokio::test] + async fn foreground_wait_disarms_group_after_reap() { + let mut owned = successful_child(); + assert_eq!(owned.child_mut().wait().await.unwrap().code(), Some(7)); + assert!(owned.child_mut().id().is_none()); + // No numeric authority remains even if Drop is delayed. + owned.kill_group().unwrap(); + drop(owned); + } + + #[tokio::test] + async fn foreground_try_wait_disarms_group_after_reap() { + let mut owned = successful_child(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if let Some(status) = owned.child_mut().try_wait().unwrap() { + assert_eq!(status.code(), Some(7)); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }).await.unwrap(); + assert!(owned.child_mut().id().is_none()); + owned.kill_group().unwrap(); + drop(owned); + } +} From bbf1845cee8779209765b18fdd39903a5a1d55d0 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:50:31 +0000 Subject: [PATCH 07/16] test: separate foreground ownership regression fixtures --- libs/mcp/server/src/foreground_command.rs | 39 +------------------ .../server/src/foreground_command_tests.rs | 35 +++++++++++++++++ 2 files changed, 37 insertions(+), 37 deletions(-) create mode 100644 libs/mcp/server/src/foreground_command_tests.rs diff --git a/libs/mcp/server/src/foreground_command.rs b/libs/mcp/server/src/foreground_command.rs index 199f9b383..e84a92333 100644 --- a/libs/mcp/server/src/foreground_command.rs +++ b/libs/mcp/server/src/foreground_command.rs @@ -62,40 +62,5 @@ impl Drop for ForegroundCommand { } #[cfg(all(test, unix))] -mod tests { - use super::*; - use tokio::process::Command; - - fn successful_child() -> ForegroundCommand { - ForegroundCommand::new(Command::new("sh") - .arg("-c").arg("exit 7").process_group(0) - .kill_on_drop(true).spawn().unwrap()) - } - - #[tokio::test] - async fn foreground_wait_disarms_group_after_reap() { - let mut owned = successful_child(); - assert_eq!(owned.child_mut().wait().await.unwrap().code(), Some(7)); - assert!(owned.child_mut().id().is_none()); - // No numeric authority remains even if Drop is delayed. - owned.kill_group().unwrap(); - drop(owned); - } - - #[tokio::test] - async fn foreground_try_wait_disarms_group_after_reap() { - let mut owned = successful_child(); - tokio::time::timeout(std::time::Duration::from_secs(5), async { - loop { - if let Some(status) = owned.child_mut().try_wait().unwrap() { - assert_eq!(status.code(), Some(7)); - break; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - }).await.unwrap(); - assert!(owned.child_mut().id().is_none()); - owned.kill_group().unwrap(); - drop(owned); - } -} +#[path = "foreground_command_tests.rs"] +mod tests; diff --git a/libs/mcp/server/src/foreground_command_tests.rs b/libs/mcp/server/src/foreground_command_tests.rs new file mode 100644 index 000000000..558557753 --- /dev/null +++ b/libs/mcp/server/src/foreground_command_tests.rs @@ -0,0 +1,35 @@ +use super::*; +use tokio::process::Command; + +fn successful_child() -> ForegroundCommand { + ForegroundCommand::new(Command::new("sh") + .arg("-c").arg("exit 7").process_group(0) + .kill_on_drop(true).spawn().unwrap()) +} + +#[tokio::test] +async fn foreground_wait_disarms_group_after_reap() { + let mut owned = successful_child(); + assert_eq!(owned.child_mut().wait().await.unwrap().code(), Some(7)); + assert!(owned.child_mut().id().is_none()); + // No numeric authority remains even if Drop is delayed. + owned.kill_group().unwrap(); + drop(owned); +} + +#[tokio::test] +async fn foreground_try_wait_disarms_group_after_reap() { + let mut owned = successful_child(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if let Some(status) = owned.child_mut().try_wait().unwrap() { + assert_eq!(status.code(), Some(7)); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }).await.unwrap(); + assert!(owned.child_mut().id().is_none()); + owned.kill_group().unwrap(); + drop(owned); +} From 10aa137edc2b0dfe9852d81efc6da9e6f4942a1f Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:51:20 +0000 Subject: [PATCH 08/16] fix: drain both foreground output pipes before reaping --- libs/mcp/server/src/foreground_tests.rs | 2 +- libs/mcp/server/src/local_tools.rs | 18 ++++++++++-------- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index 8754201db..59d0572a4 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -154,7 +154,7 @@ async fn exercise(kind: &str, configured_timeout: Option) { }) .await .unwrap(); - assert!(format!("{next:?}").contains("subsequent")); + assert!(format!("{next:?}").contains("subsequent"), "{next:?}"); client.cancel().await.unwrap(); server.cancel().await.unwrap(); } diff --git a/libs/mcp/server/src/local_tools.rs b/libs/mcp/server/src/local_tools.rs index 46afe7861..1e8e6cb98 100644 --- a/libs/mcp/server/src/local_tools.rs +++ b/libs/mcp/server/src/local_tools.rs @@ -1370,6 +1370,8 @@ SAFETY NOTES: let mut stderr_reader = BufReader::new(stderr); let mut stdout_buf = String::new(); let mut stderr_buf = String::new(); + let mut stdout_done = false; + let mut stderr_done = false; let mut result = String::new(); let progress_id = Uuid::new_v4(); @@ -1385,9 +1387,9 @@ SAFETY NOTES: stall_check_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); macro_rules! handle_output { - ($read_result:expr, $buf:expr) => { + ($read_result:expr, $buf:expr, $done:expr) => { match $read_result { - Ok(Ok(0)) => break, // EOF + Ok(Ok(0)) => $done = true, // Drain the other pipe before waiting Ok(Ok(_)) => { last_output_time = std::time::Instant::now(); stall_start_time = None; // Reset stall tracking on output @@ -1413,7 +1415,7 @@ SAFETY NOTES: }) .await; } - Ok(Err(_)) => break, // Read error + Ok(Err(error)) => return Err(error), Err(_) => {} // Timeout - continue loop } }; @@ -1455,16 +1457,16 @@ SAFETY NOTES: } } - read_result = tokio::time::timeout(Duration::from_millis(100), stderr_reader.read_line(&mut stderr_buf)) => { - handle_output!(read_result, stderr_buf); + read_result = tokio::time::timeout(Duration::from_millis(100), stderr_reader.read_line(&mut stderr_buf)), if !stderr_done => { + handle_output!(read_result, stderr_buf, stderr_done); } - read_result = tokio::time::timeout(Duration::from_millis(100), stdout_reader.read_line(&mut stdout_buf)) => { - handle_output!(read_result, stdout_buf); + read_result = tokio::time::timeout(Duration::from_millis(100), stdout_reader.read_line(&mut stdout_buf)), if !stdout_done => { + handle_output!(read_result, stdout_buf, stdout_done); } } - if let Ok(Some(_)) = child.child_mut().try_wait() { + if stdout_done && stderr_done { break; } } From 6246f3da8e21ef697bf2068dddb80b6dd51f4355 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:51:55 +0000 Subject: [PATCH 09/16] test: check cancellation side effects before follow-up request --- libs/mcp/server/src/foreground_tests.rs | 27 ++++++++++++++----------- 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index 59d0572a4..ce9bacc6c 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -76,6 +76,19 @@ async fn assert_gone(pid: u32, identity: &str) { .expect("owned shell must stop executing"); } +async fn prove_cleanup(owned: &[(u32, String)], late: &Path, control: &mut tokio::process::Child, identity: &str) { + for (pid, identity) in owned { + assert_gone(*pid, identity).await; + } + let control_pid = control.id().unwrap(); + assert!(control.try_wait().unwrap().is_none(), "unrelated process survives"); + assert_eq!(process_identity(control_pid).as_deref(), Some(identity)); + sleep(Duration::from_millis(2200)).await; + assert!(!late.exists(), "cancelled descendant must not write later"); + assert!(control.try_wait().unwrap().is_none(), "control survives deadline"); + assert_eq!(process_identity(control_pid).as_deref(), Some(identity)); +} + async fn exercise(kind: &str, configured_timeout: Option) { let dir = tempfile::tempdir().unwrap(); let marker = dir.path().join("started"); @@ -124,6 +137,7 @@ async fn exercise(kind: &str, configured_timeout: Option) { if kind == "disconnect" { client.cancel().await.unwrap(); server.cancel().await.unwrap(); + prove_cleanup(&owned, &late, &mut control, &control_identity).await; } else { if kind == "cancel" || kind == "abort" { client @@ -141,6 +155,7 @@ async fn exercise(kind: &str, configured_timeout: Option) { let result = result.unwrap(); assert!(format!("{result:?}").contains("timed out")); } + prove_cleanup(&owned, &late, &mut control, &control_identity).await; // The same MCP session must still execute a subsequent command. let next = client .call_tool(CallToolRequestParam { @@ -158,18 +173,6 @@ async fn exercise(kind: &str, configured_timeout: Option) { client.cancel().await.unwrap(); server.cancel().await.unwrap(); } - for (pid, identity) in owned { - assert_gone(pid, &identity).await; - } - assert_eq!(process_identity(control_pid), Some(control_identity.clone())); - assert!( - control.try_wait().unwrap().is_none(), - "unrelated process survives" - ); - sleep(Duration::from_millis(2200)).await; - assert!(!late.exists(), "cancelled descendant must not write later"); - assert!(control.try_wait().unwrap().is_none(), "control survives deadline"); - assert_eq!(process_identity(control_pid), Some(control_identity)); control.kill().await.unwrap(); } From 0f6d279b383e8bd46171d1b78ad7d10f28ae1b44 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:54:04 +0000 Subject: [PATCH 10/16] fix: use cancellation-safe foreground line readers --- .github/workflows/foreground-cancellation.yml | 1 + .../server/src/foreground_command_tests.rs | 16 +++-- libs/mcp/server/src/foreground_tests.rs | 64 +++++++++++++------ libs/mcp/server/src/local_tools.rs | 24 +++---- 4 files changed, 69 insertions(+), 36 deletions(-) diff --git a/.github/workflows/foreground-cancellation.yml b/.github/workflows/foreground-cancellation.yml index fd6c141d2..85d223d07 100644 --- a/.github/workflows/foreground-cancellation.yml +++ b/.github/workflows/foreground-cancellation.yml @@ -41,6 +41,7 @@ jobs: s += '\n#[cfg(all(test, target_os = "linux"))]\n#[path = "foreground_tests.rs"]\nmod foreground_tests;\n' p.write_text(s) PYTHON + cargo test --locked -p stakpak-mcp-server --no-run if cargo test --locked -p stakpak-mcp-server foreground_cancel_without_timeout -- --nocapture; then echo 'Stock executor unexpectedly passed; regression does not demonstrate the defect' exit 1 diff --git a/libs/mcp/server/src/foreground_command_tests.rs b/libs/mcp/server/src/foreground_command_tests.rs index 558557753..1a747d396 100644 --- a/libs/mcp/server/src/foreground_command_tests.rs +++ b/libs/mcp/server/src/foreground_command_tests.rs @@ -2,9 +2,15 @@ use super::*; use tokio::process::Command; fn successful_child() -> ForegroundCommand { - ForegroundCommand::new(Command::new("sh") - .arg("-c").arg("exit 7").process_group(0) - .kill_on_drop(true).spawn().unwrap()) + ForegroundCommand::new( + Command::new("sh") + .arg("-c") + .arg("exit 7") + .process_group(0) + .kill_on_drop(true) + .spawn() + .unwrap(), + ) } #[tokio::test] @@ -28,7 +34,9 @@ async fn foreground_try_wait_disarms_group_after_reap() { } tokio::time::sleep(std::time::Duration::from_millis(10)).await; } - }).await.unwrap(); + }) + .await + .unwrap(); assert!(owned.child_mut().id().is_none()); owned.kill_group().unwrap(); drop(owned); diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index ce9bacc6c..509cf6c80 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -18,17 +18,22 @@ impl rmcp::ServerHandler for ExecutionHarness { request: CallToolRequestParam, ctx: RequestContext, ) -> Result { - let command = request.arguments.as_ref() + let command = request + .arguments + .as_ref() .and_then(|args| args.get("command")) - .and_then(serde_json::Value::as_str).unwrap_or_default().to_owned(); + .and_then(serde_json::Value::as_str) + .unwrap_or_default() + .to_owned(); if !self.abort_execution || command == "printf subsequent" { return rmcp::ServerHandler::call_tool(&self.container, request, ctx).await; } let container = self.container.clone(); let ct = ctx.ct.clone(); - let execution = tokio::spawn(async move { - container.execute_local_command(&command, None, &ctx).await - }); + let execution = + tokio::spawn( + async move { container.execute_local_command(&command, None, &ctx).await }, + ); ct.cancelled().await; execution.abort(); assert!(execution.await.unwrap_err().is_cancelled()); @@ -63,10 +68,12 @@ async fn assert_gone(pid: u32, identity: &str) { let stat = fs::read_to_string(format!("/proc/{pid}/stat")); // A killed orphan may briefly remain a zombie under CI's // init; it cannot execute or create the delayed artifact. - if process_identity(pid).as_deref() != Some(identity) || stat.as_ref().map_or(true, |s| { - s.rsplit_once(") ") - .is_some_and(|(_, rest)| rest.starts_with('Z')) - }) { + if process_identity(pid).as_deref() != Some(identity) + || stat.as_ref().map_or(true, |s| { + s.rsplit_once(") ") + .is_some_and(|(_, rest)| rest.starts_with('Z')) + }) + { break; } sleep(Duration::from_millis(10)).await; @@ -76,16 +83,27 @@ async fn assert_gone(pid: u32, identity: &str) { .expect("owned shell must stop executing"); } -async fn prove_cleanup(owned: &[(u32, String)], late: &Path, control: &mut tokio::process::Child, identity: &str) { +async fn prove_cleanup( + owned: &[(u32, String)], + late: &Path, + control: &mut tokio::process::Child, + identity: &str, +) { for (pid, identity) in owned { assert_gone(*pid, identity).await; } let control_pid = control.id().unwrap(); - assert!(control.try_wait().unwrap().is_none(), "unrelated process survives"); + assert!( + control.try_wait().unwrap().is_none(), + "unrelated process survives" + ); assert_eq!(process_identity(control_pid).as_deref(), Some(identity)); sleep(Duration::from_millis(2200)).await; assert!(!late.exists(), "cancelled descendant must not write later"); - assert!(control.try_wait().unwrap().is_none(), "control survives deadline"); + assert!( + control.try_wait().unwrap().is_none(), + "control survives deadline" + ); assert_eq!(process_identity(control_pid).as_deref(), Some(identity)); } @@ -97,18 +115,28 @@ async fn exercise(kind: &str, configured_timeout: Option) { let sleeper = dir.path().join("sleeper"); let command = format!( r#"echo $$ > "{}"; sh -c 'echo $$ > "{}"; sleep 2 & echo $! > "{}"; wait; echo late > "{}"' & wait"#, - outer.display(), marker.display(), sleeper.display(), late.display() + outer.display(), + marker.display(), + sleeper.display(), + late.display() ); - let mut control = Command::new("sleep").arg("30").kill_on_drop(true).spawn().unwrap(); + let mut control = Command::new("sleep") + .arg("30") + .kill_on_drop(true) + .spawn() + .unwrap(); let control_pid = control.id().unwrap(); let control_identity = process_identity(control_pid).unwrap(); let (client_io, server_io) = tokio::io::duplex(65536); let abort_execution = kind == "abort"; let server = tokio::spawn(async move { - ExecutionHarness { container: local_container_with_profile(None), abort_execution } - .serve(server_io) - .await - .unwrap() + ExecutionHarness { + container: local_container_with_profile(None), + abort_execution, + } + .serve(server_io) + .await + .unwrap() }); let client = ().serve(client_io).await.unwrap(); let server = server.await.unwrap(); diff --git a/libs/mcp/server/src/local_tools.rs b/libs/mcp/server/src/local_tools.rs index 1e8e6cb98..ddfed6a99 100644 --- a/libs/mcp/server/src/local_tools.rs +++ b/libs/mcp/server/src/local_tools.rs @@ -1366,10 +1366,8 @@ SAFETY NOTES: let stdout = child.child_mut().stdout.take().unwrap(); #[allow(clippy::unwrap_used)] let stderr = child.child_mut().stderr.take().unwrap(); - let mut stdout_reader = BufReader::new(stdout); - let mut stderr_reader = BufReader::new(stderr); - let mut stdout_buf = String::new(); - let mut stderr_buf = String::new(); + let mut stdout_reader = BufReader::new(stdout).lines(); + let mut stderr_reader = BufReader::new(stderr).lines(); let mut stdout_done = false; let mut stderr_done = false; let mut result = String::new(); @@ -1387,14 +1385,12 @@ SAFETY NOTES: stall_check_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); macro_rules! handle_output { - ($read_result:expr, $buf:expr, $done:expr) => { + ($read_result:expr, $done:expr) => { match $read_result { - Ok(Ok(0)) => $done = true, // Drain the other pipe before waiting - Ok(Ok(_)) => { + Ok(None) => $done = true, // Drain the other pipe before waiting + Ok(Some(line)) => { last_output_time = std::time::Instant::now(); stall_start_time = None; // Reset stall tracking on output - let line = $buf.trim_end_matches('\n').to_string(); - $buf.clear(); result.push_str(&format!("{}\n", line)); let _ = ctx .peer @@ -1416,7 +1412,7 @@ SAFETY NOTES: .await; } Ok(Err(error)) => return Err(error), - Err(_) => {} // Timeout - continue loop + Err(_) => {} // Timeout - continue loop } }; } @@ -1457,12 +1453,12 @@ SAFETY NOTES: } } - read_result = tokio::time::timeout(Duration::from_millis(100), stderr_reader.read_line(&mut stderr_buf)), if !stderr_done => { - handle_output!(read_result, stderr_buf, stderr_done); + read_result = stderr_reader.next_line(), if !stderr_done => { + handle_output!(read_result, stderr_done); } - read_result = tokio::time::timeout(Duration::from_millis(100), stdout_reader.read_line(&mut stdout_buf)), if !stdout_done => { - handle_output!(read_result, stdout_buf, stdout_done); + read_result = stdout_reader.next_line(), if !stdout_done => { + handle_output!(read_result, stdout_done); } } From 43c0eac24a33500626c8a972a3d1a1eb37eae636 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:54:40 +0000 Subject: [PATCH 11/16] test: make executor abort independent of token cancellation --- libs/mcp/server/src/foreground_tests.rs | 4 ++++ libs/mcp/server/src/local_tools.rs | 3 +-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index 509cf6c80..3d2b9c93d 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -30,6 +30,10 @@ impl rmcp::ServerHandler for ExecutionHarness { } let container = self.container.clone(); let ct = ctx.ct.clone(); + // This case aborts the execution future itself, independently of the + // executor's ordinary request-token cancellation branches. + let mut ctx = ctx; + ctx.ct = Default::default(); let execution = tokio::spawn( async move { container.execute_local_command(&command, None, &ctx).await }, diff --git a/libs/mcp/server/src/local_tools.rs b/libs/mcp/server/src/local_tools.rs index ddfed6a99..3b8ca1b59 100644 --- a/libs/mcp/server/src/local_tools.rs +++ b/libs/mcp/server/src/local_tools.rs @@ -1411,8 +1411,7 @@ SAFETY NOTES: }) .await; } - Ok(Err(error)) => return Err(error), - Err(_) => {} // Timeout - continue loop + Err(error) => return Err(error), } }; } From ae25636171b031f34733f93a0bfe7af9f264cf55 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:55:09 +0000 Subject: [PATCH 12/16] test: cancel descendants after their leader exits with open pipes --- libs/mcp/server/src/foreground_tests.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index 3d2b9c93d..d92d2d8db 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -117,8 +117,9 @@ async fn exercise(kind: &str, configured_timeout: Option) { let late = dir.path().join("late"); let outer = dir.path().join("outer"); let sleeper = dir.path().join("sleeper"); + let completion = if kind == "leader-exit" { "exit 0" } else { "wait" }; let command = format!( - r#"echo $$ > "{}"; sh -c 'echo $$ > "{}"; sleep 2 & echo $! > "{}"; wait; echo late > "{}"' & wait"#, + r#"echo $$ > "{}"; sh -c 'echo $$ > "{}"; sleep 2 & echo $! > "{}"; wait; echo late > "{}"' & {completion}"#, outer.display(), marker.display(), sleeper.display(), @@ -171,7 +172,7 @@ async fn exercise(kind: &str, configured_timeout: Option) { server.cancel().await.unwrap(); prove_cleanup(&owned, &late, &mut control, &control_identity).await; } else { - if kind == "cancel" || kind == "abort" { + if kind == "cancel" || kind == "abort" || kind == "leader-exit" { client .notify_cancelled(CancelledNotificationParam { request_id: request.id.clone(), @@ -232,3 +233,8 @@ async fn foreground_disconnect() { async fn foreground_execution_future_abort() { exercise("abort", None).await; } + +#[tokio::test] +async fn foreground_cancel_after_leader_exit_before_pipe_eof() { + exercise("leader-exit", None).await; +} From 5ea84482deefe6aa71580599fd2760514f01a7d9 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 03:57:21 +0000 Subject: [PATCH 13/16] ci: retain formatted sources and exact prerelease provenance --- .github/workflows/foreground-cancellation.yml | 12 +++++++++++- libs/mcp/server/src/foreground_tests.rs | 6 +++++- libs/mcp/server/src/local_tools.rs | 2 ++ 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/.github/workflows/foreground-cancellation.yml b/.github/workflows/foreground-cancellation.yml index 85d223d07..1063470df 100644 --- a/.github/workflows/foreground-cancellation.yml +++ b/.github/workflows/foreground-cancellation.yml @@ -48,6 +48,7 @@ jobs: fi git restore libs/mcp/server/src/local_tools.rs cargo fmt --all + - run: cargo clippy --locked -p stakpak-mcp-server --all-targets -- -D warnings - run: cargo clippy --locked --all-targets -- -D warnings - run: cargo test --locked --workspace - run: | @@ -57,7 +58,16 @@ jobs: cargo test --locked -p stakai --features network-tests - run: cargo build --locked --release -p stakpak - run: git diff --exit-code + - name: Record prerelease artifact provenance + run: | + git rev-parse HEAD > target/release/source-commit.txt + rustc --version > target/release/rust-version.txt + sha256sum target/release/stakpak > target/release/stakpak.sha256 - uses: actions/upload-artifact@v4 with: name: stakpak-linux-prerelease-${{ github.sha }} - path: target/release/stakpak + path: | + target/release/stakpak + target/release/stakpak.sha256 + target/release/source-commit.txt + target/release/rust-version.txt diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index d92d2d8db..600258ea0 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -117,7 +117,11 @@ async fn exercise(kind: &str, configured_timeout: Option) { let late = dir.path().join("late"); let outer = dir.path().join("outer"); let sleeper = dir.path().join("sleeper"); - let completion = if kind == "leader-exit" { "exit 0" } else { "wait" }; + let completion = if kind == "leader-exit" { + "exit 0" + } else { + "wait" + }; let command = format!( r#"echo $$ > "{}"; sh -c 'echo $$ > "{}"; sleep 2 & echo $! > "{}"; wait; echo late > "{}"' & {completion}"#, outer.display(), diff --git a/libs/mcp/server/src/local_tools.rs b/libs/mcp/server/src/local_tools.rs index 3b8ca1b59..06b693633 100644 --- a/libs/mcp/server/src/local_tools.rs +++ b/libs/mcp/server/src/local_tools.rs @@ -1366,6 +1366,8 @@ SAFETY NOTES: let stdout = child.child_mut().stdout.take().unwrap(); #[allow(clippy::unwrap_used)] let stderr = child.child_mut().stderr.take().unwrap(); + // Lines::next_line is cancellation-safe in select!: read_line can + // discard a partial final line when the other pipe or timer wins. let mut stdout_reader = BufReader::new(stdout).lines(); let mut stderr_reader = BufReader::new(stderr).lines(); let mut stdout_done = false; From bcd8ddc86fe3ac71428f9d8e1dfc54492868f2dd Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 04:01:25 +0000 Subject: [PATCH 14/16] test: scope fail-fast fixture lint allowances to test modules --- libs/mcp/server/src/foreground_command_tests.rs | 3 +++ libs/mcp/server/src/foreground_tests.rs | 3 +++ 2 files changed, 6 insertions(+) diff --git a/libs/mcp/server/src/foreground_command_tests.rs b/libs/mcp/server/src/foreground_command_tests.rs index 1a747d396..953e26c4f 100644 --- a/libs/mcp/server/src/foreground_command_tests.rs +++ b/libs/mcp/server/src/foreground_command_tests.rs @@ -1,3 +1,6 @@ +// Fixture helpers intentionally fail fast on setup/transport errors. +#![allow(clippy::unwrap_used, clippy::expect_used)] + use super::*; use tokio::process::Command; diff --git a/libs/mcp/server/src/foreground_tests.rs b/libs/mcp/server/src/foreground_tests.rs index 600258ea0..11eb667c3 100644 --- a/libs/mcp/server/src/foreground_tests.rs +++ b/libs/mcp/server/src/foreground_tests.rs @@ -1,3 +1,6 @@ +// Fixture helpers intentionally fail fast on setup/transport errors. +#![allow(clippy::unwrap_used, clippy::expect_used)] + use super::tests::local_container_with_profile; use super::*; use rmcp::service::PeerRequestOptions; From 16e63ed35a3eb32fdadd7407c8bb55d6754195e3 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 04:30:14 +0000 Subject: [PATCH 15/16] ci: bound full suites and qualify native artifacts independently --- .github/workflows/foreground-cancellation.yml | 36 ++++++++++++++----- 1 file changed, 28 insertions(+), 8 deletions(-) diff --git a/.github/workflows/foreground-cancellation.yml b/.github/workflows/foreground-cancellation.yml index 1063470df..61c2524bb 100644 --- a/.github/workflows/foreground-cancellation.yml +++ b/.github/workflows/foreground-cancellation.yml @@ -9,7 +9,8 @@ on: permissions: contents: read jobs: - linux: + linux-qualification: + timeout-minutes: 30 runs-on: ubuntu-22.04 steps: - uses: actions/checkout@v4 @@ -49,13 +50,6 @@ jobs: git restore libs/mcp/server/src/local_tools.rs cargo fmt --all - run: cargo clippy --locked -p stakpak-mcp-server --all-targets -- -D warnings - - run: cargo clippy --locked --all-targets -- -D warnings - - run: cargo test --locked --workspace - - run: | - cargo test --locked -p stakpak-shared --features sqlite - cargo test --locked -p stakpak --features libsql-test - cargo test --locked -p stakpak-gateway --features libsql-test - cargo test --locked -p stakai --features network-tests - run: cargo build --locked --release -p stakpak - run: git diff --exit-code - name: Record prerelease artifact provenance @@ -71,3 +65,29 @@ jobs: target/release/stakpak.sha256 target/release/source-commit.txt target/release/rust-version.txt + + upstream-suites: + timeout-minutes: 45 + runs-on: ubuntu-22.04 + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@master + with: + toolchain: '1.94.1' + components: rustfmt, clippy + - uses: Swatinem/rust-cache@v2 + with: + cache-on-failure: true + - run: cargo fmt --all -- --check + - run: cargo clippy --locked --all-targets -- -D warnings + - name: Complete workspace suite + timeout-minutes: 20 + run: cargo test --locked --workspace --verbose + - name: Complete feature suites + timeout-minutes: 15 + run: | + cargo test --locked -p stakpak-shared --features sqlite --verbose + cargo test --locked -p stakpak --features libsql-test --verbose + cargo test --locked -p stakpak-gateway --features libsql-test --verbose + cargo test --locked -p stakai --features network-tests --verbose + - run: cargo build --locked --verbose From 62bd8833600b0f906ff841930433264ece36dbab Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 12 Oct 2026 05:14:44 +0000 Subject: [PATCH 16/16] fix: disarm foreground signal authority when child ownership is lost --- .github/workflows/foreground-cancellation.yml | 7 +++- libs/mcp/server/src/foreground_command.rs | 34 ++++++++++++++++--- .../server/src/foreground_command_tests.rs | 18 ++++++++-- libs/mcp/server/src/local_tools.rs | 3 +- 4 files changed, 52 insertions(+), 10 deletions(-) diff --git a/.github/workflows/foreground-cancellation.yml b/.github/workflows/foreground-cancellation.yml index 61c2524bb..f2b3bef1a 100644 --- a/.github/workflows/foreground-cancellation.yml +++ b/.github/workflows/foreground-cancellation.yml @@ -68,7 +68,12 @@ jobs: upstream-suites: timeout-minutes: 45 - runs-on: ubuntu-22.04 + runs-on: ubuntu-latest + env: + # Keep assertions/debug behavior; omit symbols to bound linker memory. + CARGO_BUILD_JOBS: '1' + CARGO_PROFILE_DEV_DEBUG: '0' + CARGO_PROFILE_TEST_DEBUG: '0' steps: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@master diff --git a/libs/mcp/server/src/foreground_command.rs b/libs/mcp/server/src/foreground_command.rs index e84a92333..b67c26436 100644 --- a/libs/mcp/server/src/foreground_command.rs +++ b/libs/mcp/server/src/foreground_command.rs @@ -6,11 +6,12 @@ use tokio::process::Child; /// deliberately leave this group are outside this cancellation boundary. pub(crate) struct ForegroundCommand { child: Child, + owned: bool, } impl ForegroundCommand { pub(crate) fn new(child: Child) -> Self { - Self { child } + Self { child, owned: true } } pub(crate) fn child_mut(&mut self) -> &mut Child { @@ -19,6 +20,9 @@ impl ForegroundCommand { #[cfg(unix)] fn kill_group(&mut self) -> io::Result<()> { + if !self.owned { + return Ok(()); + } if let Some(pid) = self.child.id() { let pid = libc::pid_t::try_from(pid) .map_err(|_| io::Error::other("invalid foreground process ID"))?; @@ -38,12 +42,26 @@ impl ForegroundCommand { Ok(()) } + pub(crate) async fn wait(&mut self) -> io::Result { + let result = self.child.wait().await; + match &result { + Ok(_) => self.owned = false, + // Child::id remains Some on a wait error. ECHILD means the + // exclusive ownership assumption was broken by another reaper; + // never signal that numeric PID/group again, including in Drop. + #[cfg(unix)] + Err(error) if error.raw_os_error() == Some(libc::ECHILD) => self.owned = false, + Err(_) => {} + } + result + } + pub(crate) async fn terminate(&mut self) -> io::Result { #[cfg(unix)] self.kill_group()?; #[cfg(not(unix))] self.child.start_kill()?; - tokio::time::timeout(std::time::Duration::from_secs(5), self.child.wait()) + tokio::time::timeout(std::time::Duration::from_secs(5), self.wait()) .await .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "foreground cleanup timed out"))? } @@ -51,13 +69,19 @@ impl ForegroundCommand { impl Drop for ForegroundCommand { fn drop(&mut self) { + if !self.owned { + return; + } #[cfg(unix)] if let Err(error) = self.kill_group() { tracing::error!("Failed to kill foreground command group: {error}"); } - // The child was spawned with kill_on_drop(true). Tokio handles direct - // child fallback/reaping; group signalling must happen synchronously - // first, including when a request future is aborted or dropped. + // Direct-child fallback is also restricted to a still-owned child. + // Command's default kill_on_drop(false) avoids a deferred stale-PID + // kill after ECHILD; Tokio still handles reaping when Child is dropped. + if self.child.id().is_some() { + let _ = self.child.start_kill(); + } } } diff --git a/libs/mcp/server/src/foreground_command_tests.rs b/libs/mcp/server/src/foreground_command_tests.rs index 953e26c4f..689f778e5 100644 --- a/libs/mcp/server/src/foreground_command_tests.rs +++ b/libs/mcp/server/src/foreground_command_tests.rs @@ -10,7 +10,6 @@ fn successful_child() -> ForegroundCommand { .arg("-c") .arg("exit 7") .process_group(0) - .kill_on_drop(true) .spawn() .unwrap(), ) @@ -19,7 +18,7 @@ fn successful_child() -> ForegroundCommand { #[tokio::test] async fn foreground_wait_disarms_group_after_reap() { let mut owned = successful_child(); - assert_eq!(owned.child_mut().wait().await.unwrap().code(), Some(7)); + assert_eq!(owned.wait().await.unwrap().code(), Some(7)); assert!(owned.child_mut().id().is_none()); // No numeric authority remains even if Drop is delayed. owned.kill_group().unwrap(); @@ -44,3 +43,18 @@ async fn foreground_try_wait_disarms_group_after_reap() { owned.kill_group().unwrap(); drop(owned); } + +#[tokio::test] +async fn foreground_external_reap_disarms_signal_authority() { + let mut owned = successful_child(); + let pid = owned.child_mut().id().unwrap() as libc::pid_t; + let mut status = 0; + // SAFETY: reap only this test's exact owned child, simulating an external + // reaper. It immediately exits; no global waitpid(-1) or PID reuse tricks. + assert_eq!(unsafe { libc::waitpid(pid, &mut status, 0) }, pid); + let error = owned.wait().await.unwrap_err(); + assert_eq!(error.raw_os_error(), Some(libc::ECHILD)); + assert!(!owned.owned); + owned.kill_group().unwrap(); + drop(owned); +} diff --git a/libs/mcp/server/src/local_tools.rs b/libs/mcp/server/src/local_tools.rs index 06b693633..5f3df1c70 100644 --- a/libs/mcp/server/src/local_tools.rs +++ b/libs/mcp/server/src/local_tools.rs @@ -1336,7 +1336,6 @@ SAFETY NOTES: let mut cmd = Command::new("sh"); cmd.arg("-c") .arg(actual_command) - .kill_on_drop(true) .stdin(std::process::Stdio::null()) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()); @@ -1469,7 +1468,7 @@ SAFETY NOTES: } // Wait for the process to complete - child.child_mut().wait().await + child.wait().await }; // Execute with timeout and cancellation support