fix(cli): cancel recursive work on bail (#14251)

Pacquet's scoped scheduler could stop dispatching after a failure, but its
workers still waited for every command already in flight. Watch-style scripts
therefore kept a bailing recursive run alive indefinitely, unlike the
TypeScript CLI whose exit cleanup terminates running children.

Track children launched by recursive run and exec commands. On Unix, launch
tracked external commands in process groups and terminate their descendants;
on Windows, terminate each tracked process tree with the system taskkill
utility. Thread cancellation through the shell emulator as well, so emulated
scripts and commands reached through bin shims do not survive a bail.

Represent interrupted scheduler work separately from natural failures. This
preserves the first failure as the command error and leaves interrupted and
undispatched tasks as running and queued in execution summaries, respectively.

Related to pnpm/pnpm#14211.

.changeset/tidy-ravens-bail.md

@@ -0,0 +1,5 @@
+---
+"pacquet": patch
+---
+
+Stop in-flight recursive `run` and `exec` commands when bailing after the first failure.

Cargo.lock

@@ -4897,6 +4897,7 @@ version = "0.0.1"
 dependencies = [
  "deno_task_shell",
  "derive_more",
+ "libc",
  "miette 7.6.0",
  "pnpm-package-manifest",
  "pnpm-reporter",
@@ -4905,6 +4906,7 @@ dependencies = [
  "tempfile",
  "tokio",
  "tracing",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]

pnpm/crates/cli/src/cli_args/exec.rs

@@ -5,7 +5,9 @@ use clap::Args;
 use derive_more::{Display, Error};
 use miette::Diagnostic;
 use pnpm_config::Config;
-use pnpm_executor::{ScriptOutput, StreamedScript, push_script_arg, select_shell};
+use pnpm_executor::{
+    ProcessTracker, ScriptOutput, StreamedScript, push_script_arg, select_shell, spawn_child,
+};
 use pnpm_package_manager::{
     make_node_package_map_option, make_node_require_option, package_map_path_for_execution,
     pnp_path_for_execution,
@@ -103,7 +105,8 @@ impl ExecArgs {
     pub fn run(self, dir: &Path, config: &Config, reporter: ReporterType) -> miette::Result<()> {
         let command = prepare_command(self.command)?;
         super::verify_deps::verify_deps_before_run(dir, config, reporter)?;
-        let status = spawn_in_dir(&command, dir, config, self.shell_mode, ScriptOutput::Inherit)?;
+        let status =
+            spawn_in_dir(&command, dir, config, self.shell_mode, ScriptOutput::Inherit, None)?;
         if !status.success() {
             // Propagate the child's exit code. A signal-terminated child
             // has no code; fall back to 1, matching pnpm's `exitCode ?? 1`.
@@ -159,22 +162,23 @@ pub(super) fn spawn_in_dir(
     config: &Config,
     shell_mode: bool,
     output: ScriptOutput<'_>,
+    process_tracker: Option<&ProcessTracker>,
 ) -> Result<ExitStatus, ExecError> {
     let mut cmd = command_in_dir(command, dir, config, shell_mode)?;
     let ScriptOutput::Streamed { dep_path, emit } = output else {
-        return cmd
-            .status()
+        let mut child = spawn_child(&mut cmd, process_tracker)
+            .map_err(|source| ExecError::Spawn { command: command[0].clone(), source })?;
+        return child
+            .wait()
             .map_err(|source| ExecError::Spawn { command: command[0].clone(), source });
     };
     let wd = dir.to_string_lossy();
     let streamed = StreamedScript { dep_path, stage: EXEC_STAGE, wd: &wd, emit };
-    let mut child = cmd
-        .stdout(Stdio::piped())
-        .stderr(Stdio::piped())
-        .spawn()
+    cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
+    let mut child = spawn_child(&mut cmd, process_tracker)
         .map_err(|source| ExecError::Spawn { command: command[0].clone(), source })?;
     let status = streamed
-        .pump(&mut child)
+        .pump(child.child_mut())
         .map_err(|source| ExecError::Spawn { command: command[0].clone(), source })?;
     streamed.finished(status.code().unwrap_or(-1));
     Ok(status)

pnpm/crates/cli/src/cli_args/exec/recursive.rs

@@ -21,7 +21,7 @@ use derive_more::{Display, Error};
 use indexmap::IndexMap;
 use miette::Diagnostic;
 use pnpm_config::Config;
-use pnpm_executor::ScriptOutput;
+use pnpm_executor::{ProcessTracker, ScriptOutput};
 use pnpm_reporter::LogEvent;
 use pnpm_workspace_task_scheduler::{
     ScheduleTasksOptions, SequenceTasksOptions, TaskCompletion, TaskGraph, TaskKey, TaskNode,
@@ -163,6 +163,7 @@ pub async fn exec_recursive(
             .collect(),
     );
     let first_failure: Mutex<Option<String>> = Mutex::new(None);
+    let process_tracker = bail.then(ProcessTracker::default);
 
     let run_task = |node: &TaskNode| -> TaskCompletion {
         let root = node.project.as_path();
@@ -171,10 +172,16 @@ pub async fn exec_recursive(
         let start = Instant::now();
         let dep_path = project_dep_path(root, dir, show_prefix);
         let output = project_output(dep_path.as_deref(), emit);
-        let outcome = spawn_in_dir(&command, root, config, args.shell_mode, output);
+        let outcome =
+            spawn_in_dir(&command, root, config, args.shell_mode, output, process_tracker.as_ref());
         let execution = project_execution(start, outcome);
         let mut result = result.lock().expect("summary lock is not poisoned");
         let entry = &mut result[&prefix];
+        if process_tracker.as_ref().is_some_and(ProcessTracker::is_cancelled)
+            && execution.message.is_none()
+        {
+            return TaskCompletion::Cancelled;
+        }
         entry.duration = Some(execution.duration);
         match execution.message {
             None => {
@@ -182,6 +189,9 @@ pub async fn exec_recursive(
                 TaskCompletion::Passed
             }
             Some(message) => {
+                if process_tracker.as_ref().is_some_and(|tracker| !tracker.cancel()) {
+                    return TaskCompletion::Cancelled;
+                }
                 entry.status = Status::Failure;
                 entry.message = Some(message);
                 entry.prefix = Some(prefix.clone());

pnpm/crates/cli/src/cli_args/run.rs

@@ -6,7 +6,9 @@ use clap::Args;
 use derive_more::{Display, Error};
 use miette::Diagnostic;
 use pnpm_config::Config;
-use pnpm_executor::{RunScript, ScriptExit, ScriptOutput, ScriptsPrependNodePath, run_script};
+use pnpm_executor::{
+    ProcessTracker, RunScript, ScriptExit, ScriptOutput, ScriptsPrependNodePath, run_script,
+};
 use pnpm_injected_deps_syncer::{SyncInjectedDeps, sync_injected_deps};
 use pnpm_package_manager::{
     make_node_package_map_option, make_node_require_option, package_map_path_for_execution,
@@ -260,6 +262,7 @@ impl RunArgs {
             extra_env: &extra_env,
             silent,
             output: ScriptOutput::Inherit,
+            process_tracker: None,
         };
         for name in &specified {
             // Resolve the main body (with `start` → `node server.js`
@@ -338,6 +341,7 @@ pub(super) struct RunContext<'a> {
     pub(super) extra_env: &'a HashMap<String, String>,
     pub(super) silent: bool,
     pub(super) output: ScriptOutput<'a>,
+    pub(super) process_tracker: Option<&'a ProcessTracker>,
 }
 
 /// Resolve `name` to a runnable main script body, or `Ok(None)` when
@@ -502,6 +506,7 @@ pub(super) fn run_stage(
         extra_env: ctx.extra_env,
         silent: ctx.silent,
         output: ctx.output,
+        process_tracker: ctx.process_tracker,
     })
     .map_err(miette::Report::new)?;
 

pnpm/crates/cli/src/cli_args/run/recursive.rs

@@ -25,7 +25,7 @@ use derive_more::{Display, Error};
 use indexmap::IndexMap;
 use miette::{Diagnostic, IntoDiagnostic};
 use pnpm_config::Config;
-use pnpm_executor::ScriptOutput;
+use pnpm_executor::{ProcessTracker, ScriptOutput};
 use pnpm_package_manager::{
     make_node_package_map_option, make_node_require_option, package_map_path_for_execution,
     pnp_path_for_execution,
@@ -219,6 +219,7 @@ pub fn run_recursive(
     let has_command = AtomicUsize::new(0);
     let first_failure: Mutex<Option<String>> = Mutex::new(None);
     let abort: Mutex<Option<miette::Report>> = Mutex::new(None);
+    let process_tracker = bail.then(ProcessTracker::default);
 
     // Compute the shared lifecycle env once. Each project adds loader
     // options for its own root before reusing that environment across
@@ -227,6 +228,11 @@ pub fn run_recursive(
     let extra_env: HashMap<String, String> = config.extra_env_with_node_options();
 
     let run_task = |node: &TaskNode| -> TaskCompletion {
+        let summary_key = task_summary_key(node);
+        let on_started = || {
+            result.lock().expect("summary lock is not poisoned")[&summary_key].status =
+                Status::Running;
+        };
         let execution = match run_project(RunProjectOptions {
             node,
             graph,
@@ -238,6 +244,8 @@ pub fn run_recursive(
             silent,
             inherit_output,
             emit,
+            process_tracker: process_tracker.as_ref(),
+            on_started: &on_started,
         }) {
             Ok(execution) => execution,
             Err(error) => {
@@ -245,6 +253,9 @@ pub fn run_recursive(
                 if abort.is_none() {
                     *abort = Some(error);
                 }
+                if let Some(process_tracker) = &process_tracker {
+                    process_tracker.cancel();
+                }
                 return TaskCompletion::Aborted;
             }
         };
@@ -252,8 +263,11 @@ pub fn run_recursive(
             has_command.fetch_add(execution.has_command, Ordering::Relaxed);
         }
         let failed = execution.status.status == Status::Failure;
-        result.lock().expect("summary lock is not poisoned")[&task_summary_key(node)] =
-            execution.status;
+        let cancelled = execution.cancelled;
+        result.lock().expect("summary lock is not poisoned")[&summary_key] = execution.status;
+        if cancelled {
+            return TaskCompletion::Cancelled;
+        }
         if failed {
             let mut first_failure =
                 first_failure.lock().expect("first-failure slot lock is not poisoned");
@@ -396,6 +410,7 @@ fn build_run_task_graph(
 struct ProjectExecution {
     status: ExecutionStatus,
     has_command: usize,
+    cancelled: bool,
 }
 
 #[derive(Clone, Copy)]
@@ -410,6 +425,8 @@ struct RunProjectOptions<'a, 'project> {
     silent: bool,
     inherit_output: bool,
     emit: fn(&LogEvent),
+    process_tracker: Option<&'a ProcessTracker>,
+    on_started: &'a dyn Fn(),
 }
 
 fn run_project(options: RunProjectOptions<'_, '_>) -> miette::Result<ProjectExecution> {
@@ -424,6 +441,8 @@ fn run_project(options: RunProjectOptions<'_, '_>) -> miette::Result<ProjectExec
         silent,
         inherit_output,
         emit,
+        process_tracker,
+        on_started,
     } = options;
     let root = node.project.as_path();
     let manifest = &graph[root].package.project.manifest;
@@ -445,7 +464,8 @@ fn run_project(options: RunProjectOptions<'_, '_>) -> miette::Result<ProjectExec
     // run's lifecycle events; the reporter renders `wd` and only groups
     // by this.
     let root_str = root.to_string_lossy().into_owned();
-    let mut execution = ProjectExecution { status: ExecutionStatus::queued(), has_command: 0 };
+    let mut execution =
+        ProjectExecution { status: ExecutionStatus::queued(), has_command: 0, cancelled: false };
     let mut project_failed = false;
     for selected in &node.scripts {
         let Some(script) = manifest.script(selected, true)? else {
@@ -460,6 +480,7 @@ fn run_project(options: RunProjectOptions<'_, '_>) -> miette::Result<ProjectExec
             continue;
         }
 
+        on_started();
         if !project_failed {
             execution.status.status = Status::Running;
         }
@@ -477,16 +498,26 @@ fn run_project(options: RunProjectOptions<'_, '_>) -> miette::Result<ProjectExec
             } else {
                 ScriptOutput::Streamed { dep_path: &root_str, emit }
             },
+            process_tracker,
         };
         let status = run_stages(&ctx, selected, script, args.script_args())?;
         let duration = start.elapsed().as_secs_f64() * 1e3;
 
+        if process_tracker.is_some_and(ProcessTracker::is_cancelled) {
+            execution.cancelled = true;
+            return Ok(execution);
+        }
+
         if status.success() {
             if !project_failed {
                 execution.status.status = Status::Passed;
                 execution.status.duration = Some(duration);
             }
         } else {
+            if process_tracker.is_some_and(|process_tracker| !process_tracker.cancel()) {
+                execution.cancelled = true;
+                return Ok(execution);
+            }
             project_failed = true;
             execution.status.status = Status::Failure;
             execution.status.duration = Some(duration);

pnpm/crates/cli/tests/suite/exec_recursive.rs

@@ -7,7 +7,13 @@ use assert_cmd::prelude::*;
 use command_extra::CommandExtra;
 use pnpm_testing_utils::{bin::CommandTempCwd, command_env::CommandTestExt};
 use serde_json::{Value, json};
-use std::{collections::HashMap, fs, path::Path, process::Command};
+use std::{
+    collections::HashMap,
+    fs,
+    path::Path,
+    process::Command,
+    time::{Duration, Instant},
+};
 
 /// Write a `pnpm-workspace.yaml` listing `names` as packages, plus a
 /// `package.json` per name under its own subdirectory of `workspace`.
@@ -320,29 +326,40 @@ fn recursive_exec_report_summary_records_every_package_status() {
 }
 
 #[test]
-fn recursive_exec_bail_summary_records_every_completed_batch_result() {
+fn recursive_exec_bail_cancels_in_flight_processes() {
     let CommandTempCwd { pacquet, root, workspace, .. } = CommandTempCwd::init();
-    write_workspace(&workspace, &["fails", "passes"]);
+    write_workspace(&workspace, &["a-slow-1", "b-fails", "c-slow-2", "z-queued"]);
 
-    pacquet
+    let start = Instant::now();
+    let output = pacquet
         .with_args([
-            "--workspace-concurrency=2",
+            "--workspace-concurrency=3",
             "--no-sort",
             "--report-summary",
             "-r",
             "exec",
             "-c",
-            // `passes` has to reach the summary, and bail stops dispatching the
-            // moment `fails` returns — so `fails` waits for the marker that
-            // proves `passes` was handed to a worker before it fails.
-            r#"if [ "$(basename "$PWD")" = fails ]; then i=0; while [ ! -f ../passes/ran.txt ] && [ $i -lt 1000 ]; do i=$((i + 1)); sleep 0.01; done; exit 1; fi; touch ran.txt"#,
+            r#"name=$(basename "$PWD"); if [ "$name" = a-slow-1 ] || [ "$name" = c-slow-2 ]; then touch ran.txt; sleep 5; elif [ "$name" = b-fails ]; then i=0; while [ ! -f ../a-slow-1/ran.txt ] || [ ! -f ../c-slow-2/ran.txt ]; do i=$((i + 1)); [ $i -lt 500 ] || exit 2; sleep 0.01; done; exit 1; else touch ran.txt; fi"#,
         ])
-        .assert()
-        .failure();
+        .output()
+        .expect("spawn pacquet");
+    let elapsed = start.elapsed();
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    eprintln!("STDERR:\n{stderr}\n");
+    assert!(!output.status.success(), "the failing project should fail the exec");
+    eprintln!("recursive exec elapsed: {elapsed:?}");
+    assert!(
+        elapsed < Duration::from_secs(4),
+        "bail should interrupt the five-second in-flight commands",
+    );
 
     let statuses = summary_statuses(&workspace);
-    assert_eq!(statuses.get("fails").map(String::as_str), Some("failure"));
-    assert_eq!(statuses.get("passes").map(String::as_str), Some("passed"));
+    dbg!(&statuses);
+    assert_eq!(statuses.get("a-slow-1").map(String::as_str), Some("running"));
+    assert_eq!(statuses.get("b-fails").map(String::as_str), Some("failure"));
+    assert_eq!(statuses.get("c-slow-2").map(String::as_str), Some("running"));
+    assert_eq!(statuses.get("z-queued").map(String::as_str), Some("queued"));
+    assert!(!workspace.join("z-queued").join("ran.txt").exists());
 
     drop(root);
 }

pnpm/crates/cli/tests/suite/run_recursive.rs

@@ -7,7 +7,14 @@ use assert_cmd::prelude::*;
 use command_extra::CommandExtra;
 use pnpm_testing_utils::bin::CommandTempCwd;
 use serde_json::{Value, json};
-use std::{collections::HashMap, fs, os::unix::fs::PermissionsExt, path::Path, process::Command};
+use std::{
+    collections::HashMap,
+    fs,
+    os::unix::fs::PermissionsExt,
+    path::Path,
+    process::Command,
+    time::{Duration, Instant},
+};
 
 /// Write a `pnpm-workspace.yaml` listing `names` as packages, plus a
 /// `package.json` per name under its own subdirectory of `workspace`.
@@ -1422,44 +1429,82 @@ fn recursive_run_report_summary_records_every_package_status() {
 }
 
 #[test]
-fn recursive_run_bail_summary_records_every_in_flight_result() {
+fn recursive_run_bail_cancels_in_flight_processes() {
+    assert_recursive_run_bail_cancels_in_flight(false);
+}
+
+#[test]
+fn recursive_run_bail_cancels_in_flight_shell_emulator_tasks() {
+    assert_recursive_run_bail_cancels_in_flight(true);
+}
+
+fn assert_recursive_run_bail_cancels_in_flight(shell_emulator: bool) {
     let CommandTempCwd { pacquet, root, workspace, .. } = CommandTempCwd::init();
     let manifest = |name: &str, body: &str| json!({ "name": name, "version": "1.0.0", "scripts": { "build": body } });
     write_workspace(
         &workspace,
         &[
             (
-                "fails",
+                "a-slow-1",
                 manifest(
-                    "fails",
-                    "i=0; while [ ! -f ../passes/ran.txt ] && [ $i -lt 1000 ]; do i=$((i + 1)); sleep 0.01; done; [ -f ../passes/ran.txt ] || exit 2; touch failed.txt; exit 1",
+                    "a-slow-1",
+                    r#"node -e "require('fs').writeFileSync('ran.txt', ''); setTimeout(() => {}, 5000)""#,
                 ),
             ),
             (
-                "passes",
+                "b-fails",
                 manifest(
-                    "passes",
-                    "touch ran.txt; i=0; while [ ! -f ../fails/failed.txt ] && [ $i -lt 1000 ]; do i=$((i + 1)); sleep 0.01; done; [ -f ../fails/failed.txt ]",
+                    "b-fails",
+                    r#"node -e "const fs = require('fs'); const wait = () => fs.existsSync('../a-slow-1/ran.txt') && fs.existsSync('../c-slow-2/ran.txt') ? process.exit(1) : setTimeout(wait, 10); wait()""#,
                 ),
             ),
+            (
+                "c-slow-2",
+                manifest(
+                    "c-slow-2",
+                    r#"node -e "require('fs').writeFileSync('ran.txt', ''); setTimeout(() => {}, 5000)""#,
+                ),
+            ),
+            ("z-queued", manifest("z-queued", "touch ran.txt")),
         ],
     );
+    if shell_emulator {
+        fs::write(
+            workspace.join("pnpm-workspace.yaml"),
+            "packages:\n  - a-slow-1\n  - b-fails\n  - c-slow-2\n  - z-queued\nshellEmulator: true\n",
+        )
+        .expect("enable the shell emulator");
+    }
 
-    pacquet
+    let start = Instant::now();
+    let output = pacquet
         .with_args([
-            "--workspace-concurrency=2",
+            "--workspace-concurrency=3",
             "--no-sort",
             "--report-summary",
             "-r",
             "run",
             "build",
         ])
-        .assert()
-        .failure();
+        .output()
+        .expect("spawn pacquet");
+    let elapsed = start.elapsed();
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    eprintln!("STDERR:\n{stderr}\n");
+    assert!(!output.status.success(), "the failing project should fail the run");
+    eprintln!("recursive run elapsed: {elapsed:?}");
+    assert!(
+        elapsed < Duration::from_secs(4),
+        "bail should interrupt the five-second in-flight scripts",
+    );
 
     let statuses = summary_statuses(&workspace);
-    assert_eq!(statuses.get("fails").map(String::as_str), Some("failure"));
-    assert_eq!(statuses.get("passes").map(String::as_str), Some("passed"));
+    dbg!(&statuses);
+    assert_eq!(statuses.get("a-slow-1").map(String::as_str), Some("running"));
+    assert_eq!(statuses.get("b-fails").map(String::as_str), Some("failure"));
+    assert_eq!(statuses.get("c-slow-2").map(String::as_str), Some("running"));
+    assert_eq!(statuses.get("z-queued").map(String::as_str), Some("queued"));
+    assert!(!workspace.join("z-queued").join("ran.txt").exists());
 
     drop(root);
 }

pnpm/crates/executor/Cargo.toml

@@ -21,6 +21,12 @@ serde_json      = { workspace = true }
 tokio           = { workspace = true, features = ["process"] }
 tracing         = { workspace = true }
 
+[target.'cfg(unix)'.dependencies]
+libc = { workspace = true }
+
+[target.'cfg(windows)'.dependencies]
+windows-sys = { workspace = true, features = ["Win32_System_SystemInformation"] }
+
 [dev-dependencies]
 pretty_assertions = { workspace = true }
 tempfile          = { workspace = true }

pnpm/crates/executor/src/lib.rs

@@ -2,6 +2,7 @@ mod bundled_node_gyp;
 mod extend_path;
 mod lifecycle;
 mod make_env;
+mod process_tracker;
 mod run_script;
 mod script_exit;
 mod shell;
@@ -16,6 +17,7 @@ pub use lifecycle::{
     run_project_lifecycle_scripts,
 };
 pub use make_env::{EnvBuild, EnvOptions, VERIFY_DEPS_BEFORE_RUN_ENV, build_env};
+pub use process_tracker::{ProcessTracker, SpawnedChild, spawn_child};
 pub use run_script::{RunScript, RunScriptError, ScriptOutput, run_script};
 pub use script_exit::ScriptExit;
 pub use shell::{ScriptShellError, SelectedShell, select_shell};

pnpm/crates/executor/src/lifecycle.rs

@@ -445,7 +445,7 @@ fn run_in_emulator<Reporter: self::Reporter>(
 ) -> Result<ScriptExit, LifecycleScriptError> {
     let target = StreamedScript { dep_path: opts.dep_path, stage, wd, emit: Reporter::emit };
     let emit_line = |stdio, line| target.emit_line(stdio, line);
-    execute_emulated(script, opts.pkg_root, env, EmulatedOutput::Lines(&emit_line))
+    execute_emulated(script, opts.pkg_root, env, EmulatedOutput::Lines(&emit_line), None)
         .map(ScriptExit::Emulated)
         .map_err(LifecycleScriptError::ShellEmulator)
 }

pnpm/crates/executor/src/process_tracker.rs

@@ -0,0 +1,276 @@
+use std::{
+    collections::HashMap,
+    io,
+    process::{Child, Command},
+    sync::Mutex,
+};
+use tokio::sync::watch;
+
+#[cfg(unix)]
+use std::{io::Read, os::unix::process::CommandExt, process::Stdio, time::Duration};
+
+/// Tracks the processes started by one recursive command so a bailing task
+/// can stop the other work that is still in flight.
+#[derive(Default)]
+pub struct ProcessTracker {
+    state: Mutex<TrackerState>,
+}
+
+#[derive(Default)]
+struct TrackerState {
+    cancelled: bool,
+    next_id: usize,
+    executions: HashMap<usize, RunningExecution>,
+}
+
+#[derive(Clone)]
+enum RunningExecution {
+    Process(u32),
+    Emulated(watch::Sender<bool>),
+}
+
+impl RunningExecution {
+    fn cancel(&self) {
+        match self {
+            Self::Process(pid) => terminate_process_tree(*pid),
+            Self::Emulated(sender) => {
+                let _ = sender.send(true);
+            }
+        }
+    }
+}
+
+impl ProcessTracker {
+    /// Cancel every registered execution. Returns `true` only to the caller
+    /// that initiated cancellation.
+    pub fn cancel(&self) -> bool {
+        let executions = {
+            let mut state = self.state.lock().expect("process tracker lock is not poisoned");
+            if state.cancelled {
+                return false;
+            }
+            state.cancelled = true;
+            state.executions.values().cloned().collect::<Vec<_>>()
+        };
+        #[cfg(unix)]
+        let descendants = descendant_processes(std::process::id());
+        for execution in executions {
+            execution.cancel();
+        }
+        #[cfg(unix)]
+        for pid in descendants {
+            terminate_process(pid);
+        }
+        true
+    }
+
+    #[must_use]
+    pub fn is_cancelled(&self) -> bool {
+        self.state.lock().expect("process tracker lock is not poisoned").cancelled
+    }
+
+    pub(crate) fn track_emulated(&self) -> EmulatedCancellation<'_> {
+        let (sender, receiver) = watch::channel(false);
+        let registration = self.register(RunningExecution::Emulated(sender));
+        EmulatedCancellation { receiver, _registration: registration }
+    }
+
+    fn register(&self, execution: RunningExecution) -> Registration<'_> {
+        let mut state = self.state.lock().expect("process tracker lock is not poisoned");
+        if state.cancelled {
+            drop(state);
+            execution.cancel();
+            return Registration { tracker: self, id: None };
+        }
+        let id = state.next_id;
+        state.next_id += 1;
+        state.executions.insert(id, execution);
+        Registration { tracker: self, id: Some(id) }
+    }
+}
+
+/// Spawn a child and optionally register it for recursive-command
+/// cancellation. On Unix a tracked child starts a process group so
+/// cancellation also reaches descendants.
+pub fn spawn_child<'tracker>(
+    command: &mut Command,
+    process_tracker: Option<&'tracker ProcessTracker>,
+) -> io::Result<SpawnedChild<'tracker>> {
+    if process_tracker.is_some() {
+        prepare_command(command);
+    }
+    let child = command.spawn()?;
+    let registration =
+        process_tracker.map(|tracker| tracker.register(RunningExecution::Process(child.id())));
+    Ok(SpawnedChild { child, _registration: registration })
+}
+
+pub struct SpawnedChild<'tracker> {
+    child: Child,
+    _registration: Option<Registration<'tracker>>,
+}
+
+impl SpawnedChild<'_> {
+    pub fn child_mut(&mut self) -> &mut Child {
+        &mut self.child
+    }
+
+    pub fn wait(&mut self) -> io::Result<std::process::ExitStatus> {
+        self.child.wait()
+    }
+}
+
+pub(crate) struct EmulatedCancellation<'tracker> {
+    receiver: watch::Receiver<bool>,
+    _registration: Registration<'tracker>,
+}
+
+impl EmulatedCancellation<'_> {
+    pub(crate) fn receiver(&self) -> watch::Receiver<bool> {
+        self.receiver.clone()
+    }
+}
+
+struct Registration<'tracker> {
+    tracker: &'tracker ProcessTracker,
+    id: Option<usize>,
+}
+
+impl Drop for Registration<'_> {
+    fn drop(&mut self) {
+        let Some(id) = self.id else { return };
+        self.tracker
+            .state
+            .lock()
+            .expect("process tracker lock is not poisoned")
+            .executions
+            .remove(&id);
+    }
+}
+
+#[cfg(unix)]
+fn prepare_command(command: &mut Command) {
+    command.process_group(0);
+}
+
+#[cfg(not(unix))]
+fn prepare_command(_: &mut Command) {}
+
+#[cfg(unix)]
+fn terminate_process_tree(pid: u32) {
+    let Ok(pid) = i32::try_from(pid) else { return };
+    // SAFETY: a negative PID asks `kill` to signal the process group whose
+    // ID is the tracked child's PID. The child was placed in that group by
+    // `prepare_command`; ESRCH is harmless when it exited concurrently.
+    unsafe {
+        libc::kill(-pid, libc::SIGKILL);
+    }
+}
+
+#[cfg(unix)]
+fn terminate_process(pid: i32) {
+    // SAFETY: every PID comes from the OS process list as a descendant of
+    // this process. ESRCH is harmless when it exited concurrently.
+    unsafe {
+        libc::kill(pid, libc::SIGKILL);
+    }
+}
+
+#[cfg(unix)]
+fn descendant_processes(root: u32) -> Vec<i32> {
+    let mut command = Command::new("/bin/ps");
+    command.args(["-A", "-o", "pid=", "-o", "ppid="]).stdout(Stdio::piped());
+    let Ok(mut child) = command.spawn() else {
+        return Vec::new();
+    };
+    let Some(mut stdout) = child.stdout.take() else {
+        return Vec::new();
+    };
+    let output = std::thread::spawn(move || {
+        let mut listing = String::new();
+        stdout.read_to_string(&mut listing).map(|_| listing)
+    });
+    let mut completed = false;
+    for _ in 0..50 {
+        match child.try_wait() {
+            Ok(Some(_)) => {
+                completed = true;
+                break;
+            }
+            Ok(None) => std::thread::sleep(Duration::from_millis(10)),
+            Err(_) => break,
+        }
+    }
+    if !completed {
+        let _ = child.kill();
+        let _ = child.wait();
+    }
+    let Ok(Ok(listing)) = output.join() else {
+        return Vec::new();
+    };
+    if !completed {
+        return Vec::new();
+    }
+
+    let mut children: HashMap<u32, Vec<u32>> = HashMap::new();
+    for line in listing.lines() {
+        let mut fields = line.split_whitespace();
+        let (Some(pid), Some(parent)) = (fields.next(), fields.next()) else {
+            continue;
+        };
+        let (Ok(pid), Ok(parent)) = (pid.parse(), parent.parse()) else {
+            continue;
+        };
+        children.entry(parent).or_default().push(pid);
+    }
+    let mut descendants = Vec::new();
+    let mut stack = vec![root];
+    while let Some(parent) = stack.pop() {
+        for &pid in children.get(&parent).into_iter().flatten() {
+            stack.push(pid);
+            if let Ok(pid) = i32::try_from(pid) {
+                descendants.push(pid);
+            }
+        }
+    }
+    descendants
+}
+
+#[cfg(windows)]
+fn terminate_process_tree(pid: u32) {
+    use std::{os::windows::process::CommandExt, process::Stdio};
+
+    let Some(taskkill) = taskkill_path() else { return };
+    let _ = Command::new(taskkill)
+        .args(["/pid", &pid.to_string(), "/T", "/F"])
+        .stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .creation_flags(0x0800_0000)
+        .spawn();
+}
+
+#[cfg(windows)]
+fn taskkill_path() -> Option<std::path::PathBuf> {
+    use std::{ffi::OsString, os::windows::ffi::OsStringExt, ptr};
+    use windows_sys::Win32::System::SystemInformation::GetSystemDirectoryW;
+
+    // SAFETY: the first call requests the required UTF-16 buffer length.
+    // The second call writes into a buffer of that size, and its returned
+    // length is checked before constructing the path.
+    unsafe {
+        let required = GetSystemDirectoryW(ptr::null_mut(), 0);
+        if required == 0 {
+            return None;
+        }
+        let mut buffer = vec![0_u16; required as usize];
+        let length = GetSystemDirectoryW(buffer.as_mut_ptr(), required);
+        if length == 0 || length >= required {
+            return None;
+        }
+        Some(
+            std::path::PathBuf::from(OsString::from_wide(&buffer[..length as usize]))
+                .join("taskkill.exe"),
+        )
+    }
+}

pnpm/crates/executor/src/run_script.rs

@@ -2,6 +2,7 @@ use crate::{
     extend_path::{ScriptsPrependNodePath, extend_path},
     lifecycle::{StreamedScript, push_script_arg},
     make_env::{EnvOptions, build_env, path_value},
+    process_tracker::{ProcessTracker, spawn_child},
     script_exit::ScriptExit,
     shell::{ScriptShellError, SelectedShell, select_shell},
     shell_emulator::{EmulatedOutput, ShellEmulatorError, execute_emulated},
@@ -103,6 +104,8 @@ pub struct RunScript<'a> {
     pub silent: bool,
     /// Where the script's output goes.
     pub output: ScriptOutput<'a>,
+    /// Tracks this script for cancellation when a recursive command bails.
+    pub process_tracker: Option<&'a ProcessTracker>,
 }
 
 /// Run a single user script in the foreground, sending its output where
@@ -154,11 +157,17 @@ pub fn run_script(opts: &RunScript<'_>) -> Result<ScriptExit, RunScriptError> {
         streamed.started(&command);
         let status = if opts.shell_emulator {
             let emit_line = |stdio, line| streamed.emit_line(stdio, line);
-            execute_emulated(&command, opts.pkg_root, &child_env, EmulatedOutput::Lines(&emit_line))
-                .map(ScriptExit::Emulated)
-                .map_err(RunScriptError::ShellEmulator)?
+            execute_emulated(
+                &command,
+                opts.pkg_root,
+                &child_env,
+                EmulatedOutput::Lines(&emit_line),
+                opts.process_tracker,
+            )
+            .map(ScriptExit::Emulated)
+            .map_err(RunScriptError::ShellEmulator)?
         } else {
-            run_piped(&shell, &command, opts.pkg_root, &child_env, streamed)?
+            run_piped(&shell, &command, opts.pkg_root, &child_env, streamed, opts.process_tracker)?
         };
         streamed.finished(status.code().unwrap_or(-1));
         return Ok(status);
@@ -172,9 +181,15 @@ pub fn run_script(opts: &RunScript<'_>) -> Result<ScriptExit, RunScriptError> {
     }
 
     if opts.shell_emulator {
-        return execute_emulated(&command, opts.pkg_root, &child_env, EmulatedOutput::Inherit)
-            .map(ScriptExit::Emulated)
-            .map_err(RunScriptError::ShellEmulator);
+        return execute_emulated(
+            &command,
+            opts.pkg_root,
+            &child_env,
+            EmulatedOutput::Inherit,
+            opts.process_tracker,
+        )
+        .map(ScriptExit::Emulated)
+        .map_err(RunScriptError::ShellEmulator);
     }
 
     // The script is appended through `push_script_arg` (not a chained
@@ -184,12 +199,11 @@ pub fn run_script(opts: &RunScript<'_>) -> Result<ScriptExit, RunScriptError> {
     let mut cmd = Command::new(&shell.program);
     cmd.args(&shell.args);
     push_script_arg(&mut cmd, &command, shell.windows_verbatim_args);
-    let status = cmd
-        .current_dir(opts.pkg_root)
-        .env_clear()
-        .envs(&child_env)
-        .status()
+    cmd.current_dir(opts.pkg_root).env_clear().envs(&child_env);
+    let mut child = spawn_child(&mut cmd, opts.process_tracker)
         .map_err(|source| RunScriptError::Spawn { script: command.clone(), source })?;
+    let status =
+        child.wait().map_err(|source| RunScriptError::Wait { script: command.clone(), source })?;
 
     Ok(ScriptExit::Process(status))
 }
@@ -202,21 +216,21 @@ fn run_piped(
     pkg_root: &Path,
     child_env: &HashMap<String, String>,
     streamed: StreamedScript<'_>,
+    process_tracker: Option<&ProcessTracker>,
 ) -> Result<ScriptExit, RunScriptError> {
     let mut cmd = Command::new(&shell.program);
     cmd.args(&shell.args);
     push_script_arg(&mut cmd, command, shell.windows_verbatim_args);
-    let mut child = cmd
-        .current_dir(pkg_root)
+    cmd.current_dir(pkg_root)
         .env_clear()
         .envs(child_env)
         .stdout(Stdio::piped())
-        .stderr(Stdio::piped())
-        .spawn()
+        .stderr(Stdio::piped());
+    let mut child = spawn_child(&mut cmd, process_tracker)
         .map_err(|source| RunScriptError::Spawn { script: command.to_string(), source })?;
 
     streamed
-        .pump(&mut child)
+        .pump(child.child_mut())
         .map(ScriptExit::Process)
         .map_err(|source| RunScriptError::Wait { script: command.to_string(), source })
 }

pnpm/crates/executor/src/run_script/tests.rs

@@ -56,6 +56,7 @@ fn run(pkg_root: &Path, stage: &str, script: &str, args: &[String]) -> ScriptExi
         extra_env: &extra_env,
         silent: true,
         output: ScriptOutput::Inherit,
+        process_tracker: None,
     })
     .expect("run the script")
 }

pnpm/crates/executor/src/shell_emulator.rs

@@ -1,6 +1,6 @@
 use deno_task_shell::{
-    KillSignal, ShellPipeReader, ShellPipeWriter, ShellState, execute_with_pipes, parser,
-    parser::SequentialList, pipe,
+    KillSignal, ShellPipeReader, ShellPipeWriter, ShellState, SignalKind, execute_with_pipes,
+    parser, parser::SequentialList, pipe,
 };
 use derive_more::{Display, Error};
 use miette::Diagnostic;
@@ -14,6 +14,8 @@ use std::{
 };
 use tokio::{runtime::Builder, task::LocalSet};
 
+use crate::process_tracker::{EmulatedCancellation, ProcessTracker};
+
 /// Failure to run a script under the `shellEmulator` setting. A script
 /// that runs and exits non-zero is not an error here — the exit code is
 /// returned to the caller, which decides what a failure means.
@@ -56,6 +58,7 @@ pub fn execute_emulated(
     cwd: &Path,
     env: &HashMap<String, String>,
     output: EmulatedOutput<'_>,
+    process_tracker: Option<&ProcessTracker>,
 ) -> Result<i32, ShellEmulatorError> {
     let list = parser::parse(script).map_err(|error| ShellEmulatorError::Parse {
         script: script.to_string(),
@@ -67,6 +70,8 @@ pub fn execute_emulated(
     let cwd = path::absolute(cwd)
         .map_err(|source| ShellEmulatorError::Start { script: script.to_string(), source })?;
     let env = env.iter().map(|(key, value)| (OsString::from(key), OsString::from(value))).collect();
+    let cancellation = process_tracker.map(ProcessTracker::track_emulated);
+    let receiver = cancellation.as_ref().map(EmulatedCancellation::receiver);
 
     match output {
         EmulatedOutput::Inherit => run_to_completion(
@@ -76,6 +81,7 @@ pub fn execute_emulated(
             cwd,
             ShellPipeWriter::stdout(),
             ShellPipeWriter::stderr(),
+            receiver,
         ),
         EmulatedOutput::Lines(sink) => thread::scope(|scope| {
             let (stdout_reader, stdout_writer) = pipe();
@@ -87,7 +93,8 @@ pub fn execute_emulated(
 
             // Both writers are consumed by the run, so the pumps see EOF
             // as soon as it returns and the joins below finish promptly.
-            let code = run_to_completion(script, list, env, cwd, stdout_writer, stderr_writer);
+            let code =
+                run_to_completion(script, list, env, cwd, stdout_writer, stderr_writer, receiver);
             let _ = stdout_pump.join();
             let _ = stderr_pump.join();
             code
@@ -108,13 +115,23 @@ fn run_to_completion(
     cwd: PathBuf,
     stdout: ShellPipeWriter,
     stderr: ShellPipeWriter,
+    cancellation: Option<tokio::sync::watch::Receiver<bool>>,
 ) -> Result<i32, ShellEmulatorError> {
     let run = thread::spawn(move || {
         let runtime = Builder::new_current_thread().enable_all().build()?;
-        let state = ShellState::new(env, cwd, HashMap::new(), KillSignal::default());
+        let kill_signal = KillSignal::default();
+        let local_set = LocalSet::new();
+        if let Some(mut cancellation) = cancellation {
+            let cancellation_signal = kill_signal.clone();
+            local_set.spawn_local(async move {
+                if *cancellation.borrow_and_update() || cancellation.changed().await.is_ok() {
+                    cancellation_signal.send(SignalKind::SIGKILL);
+                }
+            });
+        }
+        let state = ShellState::new(env, cwd, HashMap::new(), kill_signal);
         let stdin = ShellPipeReader::stdin();
-        Ok(LocalSet::new()
-            .block_on(&runtime, execute_with_pipes(list, state, stdin, stdout, stderr)))
+        Ok(local_set.block_on(&runtime, execute_with_pipes(list, state, stdin, stdout, stderr)))
     });
 
     match run.join() {

pnpm/crates/executor/src/shell_emulator/tests.rs

@@ -12,7 +12,7 @@ fn run(
 ) -> (i32, Vec<(LifecycleStdio, String)>) {
     let lines = Mutex::new(Vec::new());
     let sink = |stdio, line| lines.lock().expect("the sink is never poisoned").push((stdio, line));
-    let code = execute_emulated(script, cwd, env, EmulatedOutput::Lines(&sink))
+    let code = execute_emulated(script, cwd, env, EmulatedOutput::Lines(&sink), None)
         .expect("run the script under the emulator");
     (code, lines.into_inner().expect("the sink is never poisoned"))
 }
@@ -104,6 +104,7 @@ fn rejects_a_script_the_shell_cannot_parse() {
         dir.path(),
         &HashMap::new(),
         EmulatedOutput::Inherit,
+        None,
     )
     .expect_err("an unparsable script is an error, not an exit code");
     dbg!(&error);

pnpm/crates/workspace-task-scheduler/src/lib.rs

@@ -433,6 +433,8 @@ pub fn render_task_graph_dry_run(
 pub enum TaskCompletion {
     Passed,
     Failed,
+    /// The task was interrupted because another task caused the run to bail.
+    Cancelled,
     /// The task's work errored before it could run — an infrastructure
     /// failure, not a script failure. Stops dispatch like a bail.
     Aborted,
@@ -779,7 +781,7 @@ where
                                 block(&mut guard, index);
                             }
                         }
-                        TaskCompletion::Aborted => {
+                        TaskCompletion::Aborted | TaskCompletion::Cancelled => {
                             guard.settled[index] = true;
                             guard.unsettled -= 1;
                             guard.stop_dispatch = true;
@@ -860,7 +862,7 @@ pub async fn schedule_graph_async<Node, Run, Skip, Fut>(
                 }
             }
             TaskCompletion::Failed if options.bail => stop_dispatch = true,
-            TaskCompletion::Aborted => stop_dispatch = true,
+            TaskCompletion::Aborted | TaskCompletion::Cancelled => stop_dispatch = true,
             TaskCompletion::Failed if options.continue_on_failure => {
                 for &dependent in &dependents[index] {
                     pending_dependencies[dependent] -= 1;