fix(acp): prevent Windows terminal hangs from pipe orphans and .cmd shims
This commit is contained in:
@@ -1,6 +1,7 @@
|
|||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::process::Stdio;
|
use std::process::Stdio;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use sacp::schema::{
|
use sacp::schema::{
|
||||||
CreateTerminalRequest, CreateTerminalResponse, KillTerminalRequest, KillTerminalResponse,
|
CreateTerminalRequest, CreateTerminalResponse, KillTerminalRequest, KillTerminalResponse,
|
||||||
@@ -9,9 +10,16 @@ use sacp::schema::{
|
|||||||
};
|
};
|
||||||
use tokio::io::{AsyncRead, AsyncReadExt};
|
use tokio::io::{AsyncRead, AsyncReadExt};
|
||||||
use tokio::sync::Mutex;
|
use tokio::sync::Mutex;
|
||||||
|
use tokio::task::JoinHandle;
|
||||||
|
|
||||||
type TerminalMap = HashMap<String, Arc<TerminalInstance>>;
|
type TerminalMap = HashMap<String, Arc<TerminalInstance>>;
|
||||||
const DEFAULT_OUTPUT_BYTE_LIMIT: u64 = 1_000_000;
|
const DEFAULT_OUTPUT_BYTE_LIMIT: u64 = 1_000_000;
|
||||||
|
/// After the child process exits, wait up to this long for the stdout/stderr
|
||||||
|
/// reader tasks to drain naturally before aborting them. Needed because a
|
||||||
|
/// grandchild process (e.g. Node spawned from a `.cmd` shim on Windows) can
|
||||||
|
/// inherit the pipe handle and keep it open long after the direct child
|
||||||
|
/// exits, turning `wait_for_exit` into a silent hang.
|
||||||
|
const READER_DRAIN_GRACE: Duration = Duration::from_millis(200);
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub enum TerminalRuntimeError {
|
pub enum TerminalRuntimeError {
|
||||||
@@ -41,6 +49,7 @@ struct TerminalInstance {
|
|||||||
output_limit: Option<usize>,
|
output_limit: Option<usize>,
|
||||||
child: Mutex<Option<tokio::process::Child>>,
|
child: Mutex<Option<tokio::process::Child>>,
|
||||||
snapshot: Mutex<TerminalSnapshot>,
|
snapshot: Mutex<TerminalSnapshot>,
|
||||||
|
reader_handles: Mutex<Vec<JoinHandle<()>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl TerminalInstance {
|
impl TerminalInstance {
|
||||||
@@ -50,6 +59,24 @@ impl TerminalInstance {
|
|||||||
output_limit: output_limit.and_then(|v| usize::try_from(v).ok()),
|
output_limit: output_limit.and_then(|v| usize::try_from(v).ok()),
|
||||||
child: Mutex::new(Some(child)),
|
child: Mutex::new(Some(child)),
|
||||||
snapshot: Mutex::new(TerminalSnapshot::default()),
|
snapshot: Mutex::new(TerminalSnapshot::default()),
|
||||||
|
reader_handles: Mutex::new(Vec::new()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Wait briefly for stdout/stderr reader tasks to finish; abort any that
|
||||||
|
/// remain. Must be called after the direct child has already exited —
|
||||||
|
/// otherwise we would abort readers that are still making progress.
|
||||||
|
async fn drain_readers(&self) {
|
||||||
|
let handles: Vec<JoinHandle<()>> =
|
||||||
|
std::mem::take(&mut *self.reader_handles.lock().await);
|
||||||
|
for handle in handles {
|
||||||
|
let abort = handle.abort_handle();
|
||||||
|
if tokio::time::timeout(READER_DRAIN_GRACE, handle)
|
||||||
|
.await
|
||||||
|
.is_err()
|
||||||
|
{
|
||||||
|
abort.abort();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -105,11 +132,10 @@ impl TerminalInstance {
|
|||||||
|
|
||||||
async fn wait_for_exit(&self) -> Result<TerminalExitStatus, TerminalRuntimeError> {
|
async fn wait_for_exit(&self) -> Result<TerminalExitStatus, TerminalRuntimeError> {
|
||||||
self.refresh_exit_status().await?;
|
self.refresh_exit_status().await?;
|
||||||
{
|
let cached_exit = self.snapshot.lock().await.exit_status.clone();
|
||||||
let snapshot = self.snapshot.lock().await;
|
if let Some(exit_status) = cached_exit {
|
||||||
if let Some(exit_status) = snapshot.exit_status.clone() {
|
self.drain_readers().await;
|
||||||
return Ok(exit_status);
|
return Ok(exit_status);
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
let exit_status = {
|
let exit_status = {
|
||||||
@@ -128,6 +154,8 @@ impl TerminalInstance {
|
|||||||
map_exit_status(status)
|
map_exit_status(status)
|
||||||
};
|
};
|
||||||
|
|
||||||
|
self.drain_readers().await;
|
||||||
|
|
||||||
let mut snapshot = self.snapshot.lock().await;
|
let mut snapshot = self.snapshot.lock().await;
|
||||||
snapshot.exit_status = Some(exit_status.clone());
|
snapshot.exit_status = Some(exit_status.clone());
|
||||||
Ok(exit_status)
|
Ok(exit_status)
|
||||||
@@ -135,11 +163,10 @@ impl TerminalInstance {
|
|||||||
|
|
||||||
async fn kill_command(&self) -> Result<(), TerminalRuntimeError> {
|
async fn kill_command(&self) -> Result<(), TerminalRuntimeError> {
|
||||||
self.refresh_exit_status().await?;
|
self.refresh_exit_status().await?;
|
||||||
{
|
let already_exited = self.snapshot.lock().await.exit_status.is_some();
|
||||||
let snapshot = self.snapshot.lock().await;
|
if already_exited {
|
||||||
if snapshot.exit_status.is_some() {
|
self.drain_readers().await;
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
let exit_status = {
|
let exit_status = {
|
||||||
@@ -163,6 +190,8 @@ impl TerminalInstance {
|
|||||||
map_exit_status(status)
|
map_exit_status(status)
|
||||||
};
|
};
|
||||||
|
|
||||||
|
self.drain_readers().await;
|
||||||
|
|
||||||
let mut snapshot = self.snapshot.lock().await;
|
let mut snapshot = self.snapshot.lock().await;
|
||||||
snapshot.exit_status = Some(exit_status);
|
snapshot.exit_status = Some(exit_status);
|
||||||
Ok(())
|
Ok(())
|
||||||
@@ -246,18 +275,23 @@ impl TerminalRuntime {
|
|||||||
child,
|
child,
|
||||||
));
|
));
|
||||||
|
|
||||||
|
let mut handles: Vec<JoinHandle<()>> = Vec::new();
|
||||||
if let Some(reader) = stdout {
|
if let Some(reader) = stdout {
|
||||||
let terminal_ref = terminal.clone();
|
let terminal_ref = terminal.clone();
|
||||||
tokio::spawn(async move {
|
handles.push(tokio::spawn(async move {
|
||||||
read_stream(reader, terminal_ref).await;
|
read_stream(reader, terminal_ref).await;
|
||||||
});
|
}));
|
||||||
}
|
}
|
||||||
|
|
||||||
if let Some(reader) = stderr {
|
if let Some(reader) = stderr {
|
||||||
let terminal_ref = terminal.clone();
|
let terminal_ref = terminal.clone();
|
||||||
tokio::spawn(async move {
|
handles.push(tokio::spawn(async move {
|
||||||
read_stream(reader, terminal_ref).await;
|
read_stream(reader, terminal_ref).await;
|
||||||
});
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
if !handles.is_empty() {
|
||||||
|
terminal.reader_handles.lock().await.extend(handles);
|
||||||
}
|
}
|
||||||
|
|
||||||
self.terminals
|
self.terminals
|
||||||
|
|||||||
@@ -71,25 +71,32 @@ impl SetEnv for tokio::process::Command {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// On Windows, resolve a bare program name to its concrete file on PATH
|
||||||
|
/// by trying `.exe → .cmd → .bat` in order.
|
||||||
|
///
|
||||||
|
/// Rust's `Command::new("foo")` on Windows relies on `CreateProcessW`'s
|
||||||
|
/// implicit extension lookup, which does not locate `.cmd` / `.bat` shims
|
||||||
|
/// reliably for many npm-installed tools (`tsc`, `vite`, `eslint`, ...).
|
||||||
|
/// Without this helper those agents hang or ENOENT when ACP agents send
|
||||||
|
/// bare names. Extension fallback is **purely additive**: if the caller
|
||||||
|
/// already supplied a path, extension, or the `.exe` is found, the result
|
||||||
|
/// is identical to the previous behavior.
|
||||||
#[cfg(windows)]
|
#[cfg(windows)]
|
||||||
fn maybe_windows_cmd_shim(program: &OsStr) -> Option<OsString> {
|
fn resolve_windows_program(program: &OsStr) -> Option<OsString> {
|
||||||
let path = Path::new(program);
|
let path = Path::new(program);
|
||||||
|
// Only apply fallback for bare names (no path components, no extension).
|
||||||
if path.components().count() != 1 || path.extension().is_some() {
|
if path.components().count() != 1 || path.extension().is_some() {
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
|
|
||||||
let raw = program.to_string_lossy();
|
let raw = program.to_string_lossy();
|
||||||
let normalized = raw.to_ascii_lowercase();
|
for ext in ["exe", "cmd", "bat"] {
|
||||||
let needs_cmd_shim = matches!(
|
let candidate = format!("{raw}.{ext}");
|
||||||
normalized.as_str(),
|
if which::which(&candidate).is_ok() {
|
||||||
"npm" | "npx" | "pnpm" | "pnpx" | "yarn" | "yarnpkg" | "corepack"
|
return Some(OsString::from(candidate));
|
||||||
);
|
}
|
||||||
|
|
||||||
if needs_cmd_shim {
|
|
||||||
Some(OsString::from(format!("{raw}.cmd")))
|
|
||||||
} else {
|
|
||||||
None
|
|
||||||
}
|
}
|
||||||
|
None
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn normalized_program<S>(program: S) -> OsString
|
pub fn normalized_program<S>(program: S) -> OsString
|
||||||
@@ -98,8 +105,8 @@ where
|
|||||||
{
|
{
|
||||||
#[cfg(windows)]
|
#[cfg(windows)]
|
||||||
{
|
{
|
||||||
if let Some(shimmed) = maybe_windows_cmd_shim(program.as_ref()) {
|
if let Some(resolved) = resolve_windows_program(program.as_ref()) {
|
||||||
return shimmed;
|
return resolved;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user