fix: Handle Windows package manager shutdown

crates/turborepo-lib/src/run/builder.rs

@@ -34,7 +34,7 @@ use turborepo_telemetry::events::{
     repo::{RepoEventBuilder, RepoType},
     EventBuilder, TrackedErrors,
 };
-use turborepo_types::FilterMode;
+use turborepo_types::{FilterMode, UIMode};
 use turborepo_ui::ColorConfig;
 use turborepo_vercel_api::CachingStatusResponse;
 use url::Url;
@@ -84,14 +84,16 @@ impl RunBuilder {
     #[tracing::instrument(skip_all)]
     pub fn new(base: CommandBase, http_client: Option<SharedHttpClient>) -> Result<Self, Error> {
         let http_client = http_client.unwrap_or_default();
+        let opts = base.opts();
         let api_auth = base.api_auth()?;
 
         let version = base.version();
         let processes = ProcessManager::new(
-            // A terminal-backed PTY lets Turbo own interactive task input. On
-            // Windows this is also how we deliver a targeted Ctrl+C to tasks
-            // instead of relying on console-wide Ctrl+C broadcasts.
-            std::io::stdout().is_terminal(),
+            // We currently only use a pty if the following are met:
+            // - we're attached to a tty
+            std::io::stdout().is_terminal() &&
+            // - if we're on windows, we're using the UI
+            (!cfg!(windows) || matches!(opts.run_opts.ui_mode, UIMode::Tui)),
         );
 
         let CommandBase {

crates/turborepo-lib/src/run/mod.rs

@@ -542,13 +542,16 @@ impl Run {
         let color_config = self.color_config;
         let scrollback_len = self.opts.tui_opts.scrollback_length;
         let repo_root = self.repo_root.clone();
+        let signal_handler = self.signal_handler.clone();
+        let interrupt = Arc::new(move || signal_handler.notify_signal());
         let handle = tokio::task::spawn(async move {
             Ok(tui::run_app(
                 task_names,
                 receiver,
                 color_config,
                 &repo_root,
                 scrollback_len,
+                Some(interrupt),
             )
             .await?)
         });

crates/turborepo-process/Cargo.toml

@@ -30,4 +30,6 @@ windows-sys = { version = "0.59", features = [
   "Win32_System_Diagnostics_ToolHelp",
   "Win32_System_JobObjects",
   "Win32_System_Threading",
+  "Win32_UI",
+  "Win32_UI_WindowsAndMessaging",
 ] }

crates/turborepo-process/src/child/handle.rs

@@ -145,6 +145,28 @@ pub(super) fn signal_process_group(process_group_id: libc::pid_t, signal: libc::
     let _ = unsafe { libc::kill(-process_group_id, signal) };
 }
 
+#[cfg(windows)]
+fn run_child_console_helper(pid: u32, command: &str) -> bool {
+    let Ok(exe) = std::env::current_exe() else {
+        return false;
+    };
+
+    std::process::Command::new(exe)
+        .arg("__internal_windows_ctrl_c")
+        .arg(command)
+        .arg(pid.to_string())
+        .stdin(std::process::Stdio::null())
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null())
+        .status()
+        .is_ok_and(|status| status.success())
+}
+
+#[cfg(windows)]
+fn send_ctrl_c_to_child_console(pid: u32) -> bool {
+    run_child_console_helper(pid, "ctrl_c")
+}
+
 #[cfg(unix)]
 fn capture_target_identity(pid: Option<u32>) -> Option<TargetIdentity> {
     pid.and_then(|pid| match target_identity(pid as libc::pid_t) {
@@ -162,11 +184,13 @@ impl ChildHandle {
         #[cfg(windows)]
         let command_for_fallback = command.clone();
 
-        let mut command = TokioCommand::from(command);
+        let mut command = std::process::Command::from(command);
 
         // Create a new process group so we can send signals (e.g. SIGINT) to
         // the child and all of its descendants via kill(-pgid, sig).
         #[cfg(unix)]
+        use std::os::unix::process::CommandExt as _;
+        #[cfg(unix)]
         command.process_group(0);
 
         #[cfg(windows)]
@@ -178,14 +202,35 @@ impl ChildHandle {
             }
         };
 
+        #[cfg(windows)]
+        let wrapper_ctrl_c = std::env::var_os("__TURBO_WINDOWS_CTRL_C_FD").is_some();
+
+        #[cfg(windows)]
+        use std::os::windows::process::CommandExt as _;
+
+        #[cfg(windows)]
+        if wrapper_ctrl_c {
+            command.show_window(windows_sys::Win32::UI::WindowsAndMessaging::SW_HIDE as u16);
+        }
+
         #[cfg(windows)]
         if job.is_some() {
+            let mut creation_flags = windows_sys::Win32::System::Threading::CREATE_SUSPENDED
+                | windows_sys::Win32::System::Threading::CREATE_BREAKAWAY_FROM_JOB;
+            if wrapper_ctrl_c {
+                creation_flags |= windows_sys::Win32::System::Threading::CREATE_NEW_CONSOLE
+                    | windows_sys::Win32::System::Threading::CREATE_NO_WINDOW;
+            }
+            command.creation_flags(creation_flags);
+        } else if wrapper_ctrl_c {
             command.creation_flags(
-                windows_sys::Win32::System::Threading::CREATE_SUSPENDED
-                    | windows_sys::Win32::System::Threading::CREATE_BREAKAWAY_FROM_JOB,
+                windows_sys::Win32::System::Threading::CREATE_NEW_CONSOLE
+                    | windows_sys::Win32::System::Threading::CREATE_NO_WINDOW,
             );
         }
 
+        let mut command = TokioCommand::from(command);
+
         #[cfg(not(windows))]
         let mut child = command.spawn()?;
 
@@ -195,8 +240,17 @@ impl ChildHandle {
             Err(err) if job.is_some() => {
                 debug!("failed to spawn child with job breakaway: {err}");
                 let mut fallback_command = TokioCommand::from(command_for_fallback);
-                fallback_command
-                    .creation_flags(windows_sys::Win32::System::Threading::CREATE_SUSPENDED);
+                let mut creation_flags = windows_sys::Win32::System::Threading::CREATE_SUSPENDED;
+                if wrapper_ctrl_c {
+                    creation_flags |= windows_sys::Win32::System::Threading::CREATE_NEW_CONSOLE
+                        | windows_sys::Win32::System::Threading::CREATE_NO_WINDOW;
+                }
+                if wrapper_ctrl_c {
+                    fallback_command
+                        .as_std_mut()
+                        .show_window(windows_sys::Win32::UI::WindowsAndMessaging::SW_HIDE as u16);
+                }
+                fallback_command.creation_flags(creation_flags);
                 fallback_command.spawn()?
             }
             Err(err) => return Err(err),
@@ -527,11 +581,23 @@ impl ChildHandle {
     /// when a user types Ctrl-C in a real console. Returns whether the
     /// keystroke was written.
     ///
-    /// Children not attached to a ConPTY share turbo's console and receive
-    /// console Ctrl-C events directly, so there is nothing to send here.
+    /// When the npm package wrapper captures Ctrl-C in raw mode, Windows does
+    /// not generate the console event. In that case, synthesize it here so
+    /// non-ConPTY children still receive the same event as direct `turbo` use.
     #[cfg(windows)]
     pub(super) fn send_graceful_interrupt(&self) -> bool {
         let Some(pty_input) = &self.pty_input else {
+            if std::env::var_os("__TURBO_WINDOWS_CTRL_C_FD").is_some()
+                && let Some(pid) = self.pid
+            {
+                let sent = send_ctrl_c_to_child_console(pid);
+                if sent {
+                    debug!("generated console Ctrl-C for child console {pid}");
+                } else {
+                    debug!("failed to generate console Ctrl-C for child console {pid}");
+                }
+                return sent;
+            }
             return false;
         };
 

crates/turborepo-process/src/child/io.rs

@@ -12,8 +12,6 @@ use tracing::{debug, trace};
 use super::{Child, ChildExit};
 
 const POST_EXIT_OUTPUT_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(100);
-#[cfg(any(windows, test))]
-const CONPTY_CURSOR_POSITION_REQUEST: &[u8] = b"\x1b[6n";
 
 pub(super) struct ChildIO {
     pub(super) stdin: Option<ChildInput>,
@@ -153,20 +151,9 @@ impl Child {
         tokio::task::spawn_blocking(move || {
             let mut buffer = [0; 1024];
             let mut last_byte = None;
-            #[cfg(windows)]
-            let mut conpty_cursor_request_match = 0;
             loop {
                 match stdout_lines.read(&mut buffer) {
                     Ok(0) => {
-                        #[cfg(windows)]
-                        if conpty_cursor_request_match > 0 {
-                            byte_tx
-                                .blocking_send(
-                                    CONPTY_CURSOR_POSITION_REQUEST[..conpty_cursor_request_match]
-                                        .to_vec(),
-                                )
-                                .ok();
-                        }
                         if !matches!(last_byte, Some(b'\n')) {
                             // Ignore if this fails as we already are shutting down
                             byte_tx.blocking_send(vec![b'\n']).ok();
@@ -176,16 +163,6 @@ impl Child {
                     Ok(n) => {
                         let mut bytes = Vec::with_capacity(n);
                         bytes.extend_from_slice(&buffer[..n]);
-                        #[cfg(windows)]
-                        {
-                            bytes = strip_conpty_cursor_position_requests(
-                                &bytes,
-                                &mut conpty_cursor_request_match,
-                            );
-                        }
-                        if bytes.is_empty() {
-                            continue;
-                        }
                         last_byte = bytes.last().copied();
                         if byte_tx.blocking_send(bytes).is_err() {
                             // A dropped receiver indicates that there was an issue writing to the
@@ -336,56 +313,3 @@ fn add_trailing_newline(buffer: &mut Vec<u8>) {
         buffer.push(b'\n');
     }
 }
-
-#[cfg(any(windows, test))]
-fn strip_conpty_cursor_position_requests(bytes: &[u8], matched: &mut usize) -> Vec<u8> {
-    let mut output = Vec::with_capacity(bytes.len());
-
-    for byte in bytes {
-        if *byte == CONPTY_CURSOR_POSITION_REQUEST[*matched] {
-            *matched += 1;
-            if *matched == CONPTY_CURSOR_POSITION_REQUEST.len() {
-                *matched = 0;
-            }
-            continue;
-        }
-
-        if *matched > 0 {
-            output.extend_from_slice(&CONPTY_CURSOR_POSITION_REQUEST[..*matched]);
-            *matched = 0;
-        }
-
-        if *byte == CONPTY_CURSOR_POSITION_REQUEST[0] {
-            *matched = 1;
-        } else {
-            output.push(*byte);
-        }
-    }
-
-    output
-}
-
-#[cfg(test)]
-mod tests {
-    use super::strip_conpty_cursor_position_requests;
-
-    #[test]
-    fn strips_complete_conpty_cursor_position_request() {
-        let mut matched = 0;
-        let output = strip_conpty_cursor_position_requests(b"before\x1b[6nafter", &mut matched);
-
-        assert_eq!(output, b"beforeafter");
-        assert_eq!(matched, 0);
-    }
-
-    #[test]
-    fn strips_conpty_cursor_position_request_across_chunks() {
-        let mut matched = 0;
-        let first = strip_conpty_cursor_position_requests(b"before\x1b[", &mut matched);
-        let second = strip_conpty_cursor_position_requests(b"6nafter", &mut matched);
-
-        assert_eq!(first, b"before");
-        assert_eq!(second, b"after");
-        assert_eq!(matched, 0);
-    }
-}

crates/turborepo-process/src/child/test.rs

@@ -1,5 +1,5 @@
 use std::{
     assert_matches, fs, io,
     sync::{Arc, Mutex},
     time::Duration,
 };
@@ -345,7 +345,7 @@
     child.stop().await;
     let exit = child.wait().await;
 
-    if cfg!(windows) && !use_pty {
+    if cfg!(windows) {
         assert_matches!(exit, Some(ChildExit::Killed));
     } else {
         assert_matches!(exit, Some(ChildExit::Interrupted));
@@ -387,7 +387,7 @@
 
     assert!(output.contains("ready"), "missing startup output: {output}");
 
-    if cfg!(windows) && !use_pty {
+    if cfg!(windows) {
         assert_matches!(exit, Some(ChildExit::Killed));
     } else {
         assert!(

crates/turborepo-process/src/command.rs

@@ -108,7 +108,7 @@ impl Command {
     }
 }
 
-impl From<Command> for tokio::process::Command {
+impl From<Command> for std::process::Command {
     fn from(value: Command) -> Self {
         let Command {
             program,
@@ -119,7 +119,7 @@ impl From<Command> for tokio::process::Command {
             env_clear,
         } = value;
 
-        let mut cmd = tokio::process::Command::new(program);
+        let mut cmd = std::process::Command::new(program);
         if env_clear {
             cmd.env_clear();
         }
@@ -141,6 +141,12 @@ impl From<Command> for tokio::process::Command {
     }
 }
 
+impl From<Command> for tokio::process::Command {
+    fn from(value: Command) -> Self {
+        tokio::process::Command::from(std::process::Command::from(value))
+    }
+}
+
 impl From<Command> for portable_pty::CommandBuilder {
     fn from(value: Command) -> Self {
         let Command {

crates/turborepo-process/src/lib.rs

@@ -9,6 +9,7 @@
 //! As of now, the manager will execute futures in a random order, and
 //! must be either `wait`ed on or `stop`ped to drive state.
 
+#![cfg_attr(windows, feature(windows_process_extensions_show_window))]
 #![deny(clippy::all)]
 
 mod child;
@@ -311,7 +312,7 @@
 
 #[cfg(test)]
 mod test {
     use std::{fs, time::Instant};
 
     use futures::{StreamExt, stream::FuturesUnordered};
     use test_case::test_case;

crates/turborepo-signals/Cargo.toml

@@ -6,6 +6,7 @@ license = "MIT"
 
 [dependencies]
 futures = { workspace = true }
+libc = "0.2"
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full", "time"] }