feat: concurrent execution with cancellation reporting

crates/fspy/Cargo.toml

@@ -19,7 +19,8 @@ rand = { workspace = true }
 rustc-hash = { workspace = true }
 tempfile = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true, features = ["net", "process", "io-util", "sync", "rt"] }
+tokio = { workspace = true, features = ["net", "process", "io-util", "sync", "rt", "macros"] }
+tokio-util = { workspace = true }
 which = { workspace = true, features = ["tracing"] }
 xxhash-rust = { workspace = true }
 

crates/fspy/examples/cli.rs

@@ -18,7 +18,7 @@ async fn main() -> anyhow::Result<()> {
     let mut command = fspy::Command::new(program);
     command.envs(std::env::vars_os()).args(args);
 
-    let child = command.spawn().await?;
+    let child = command.spawn(tokio_util::sync::CancellationToken::new()).await?;
     let termination = child.wait_handle.await?;
 
     let mut path_count = 0usize;

crates/fspy/src/command.rs

@@ -8,6 +8,7 @@ use std::{
 use fspy_shared_unix::exec::Exec;
 use rustc_hash::FxHashMap;
 use tokio::process::Command as TokioCommand;
+use tokio_util::sync::CancellationToken;
 
 use crate::{SPY_IMPL, TrackedChild, error::SpawnError};
 
@@ -167,9 +168,12 @@ impl Command {
     /// # Errors
     ///
     /// Returns [`SpawnError`] if program resolution fails or the process cannot be spawned.
-    pub async fn spawn(mut self) -> Result<TrackedChild, SpawnError> {
+    pub async fn spawn(
+        mut self,
+        cancel_token: CancellationToken,
+    ) -> Result<TrackedChild, SpawnError> {
         self.resolve_program()?;
-        SPY_IMPL.spawn(self).await
+        SPY_IMPL.spawn(self, cancel_token).await
     }
 
     /// Resolve program name to full path using `PATH` and cwd.

crates/fspy/src/unix/mod.rs

@@ -67,7 +67,11 @@ impl SpyImpl {
         })
     }
 
-    pub(crate) async fn spawn(&self, mut command: Command) -> Result<TrackedChild, SpawnError> {
+    pub(crate) async fn spawn(
+        &self,
+        mut command: Command,
+        cancel_token: tokio_util::sync::CancellationToken,
+    ) -> Result<TrackedChild, SpawnError> {
         #[cfg(target_os = "linux")]
         let supervisor = supervise::<SyscallHandler>().map_err(SpawnError::Supervisor)?;
 
@@ -129,7 +133,13 @@ impl SpyImpl {
             // Keep polling for the child to exit in the background even if `wait_handle` is not awaited,
             // because we need to stop the supervisor and lock the channel as soon as the child exits.
             wait_handle: tokio::spawn(async move {
-                let status = child.wait().await?;
+                let status = tokio::select! {
+                    status = child.wait() => status?,
+                    () = cancel_token.cancelled() => {
+                        let _ = child.start_kill();
+                        child.wait().await?
+                    }
+                };
 
                 let arenas = std::iter::once(exec_resolve_accesses);
                 // Stop the supervisor and collect path accesses from it.

crates/fspy/src/windows/mod.rs

@@ -73,7 +73,11 @@ impl SpyImpl {
     }
 
     #[expect(clippy::unused_async, reason = "async signature required by SpyImpl trait")]
-    pub(crate) async fn spawn(&self, mut command: Command) -> Result<TrackedChild, SpawnError> {
+    pub(crate) async fn spawn(
+        &self,
+        mut command: Command,
+        cancel_token: tokio_util::sync::CancellationToken,
+    ) -> Result<TrackedChild, SpawnError> {
         let ansi_dll_path_with_nul = Arc::clone(&self.ansi_dll_path_with_nul);
         command.env("FSPY", "1");
         let mut command = command.into_tokio_command();
@@ -142,7 +146,13 @@ impl SpyImpl {
             // Keep polling for the child to exit in the background even if `wait_handle` is not awaited,
             // because we need to stop the supervisor and lock the channel as soon as the child exits.
             wait_handle: tokio::spawn(async move {
-                let status = child.wait().await?;
+                let status = tokio::select! {
+                    status = child.wait() => status?,
+                    () = cancel_token.cancelled() => {
+                        let _ = child.start_kill();
+                        child.wait().await?
+                    }
+                };
                 // Lock the ipc channel after the child has exited.
                 // We are not interested in path accesses from descendants after the main child has exited.
                 let ipc_receiver_lock_guard = OwnedReceiverLockGuard::lock_async(receiver).await?;

crates/fspy/tests/node_fs.rs

@@ -16,7 +16,7 @@ async fn track_node_script(script: &str, args: &[&OsStr]) -> anyhow::Result<Path
         .envs(vars_os()) // https://github.com/jdx/mise/discussions/5968
         .arg(script)
         .args(args);
-    let child = command.spawn().await?;
+    let child = command.spawn(tokio_util::sync::CancellationToken::new()).await?;
     let termination = child.wait_handle.await?;
     assert!(termination.status.success());
     Ok(termination.path_accesses)

crates/fspy/tests/oxlint.rs

@@ -45,7 +45,7 @@ async fn track_oxlint(dir: &std::path::Path, args: &[&str]) -> anyhow::Result<Pa
         .env("PATH", new_path)
         .current_dir(dir);
 
-    let child = command.spawn().await?;
+    let child = command.spawn(tokio_util::sync::CancellationToken::new()).await?;
     let termination = child.wait_handle.await?;
     // oxlint may return non-zero if it finds lint errors, that's OK
     Ok(termination.path_accesses)

crates/fspy/tests/static_executable.rs

@@ -41,7 +41,7 @@ async fn track_test_bin(args: &[&str], cwd: Option<&str>) -> PathAccessIterable
         cmd.current_dir(cwd);
     }
     cmd.args(args);
-    let tracked_child = cmd.spawn().await.unwrap();
+    let tracked_child = cmd.spawn(tokio_util::sync::CancellationToken::new()).await.unwrap();
 
     let termination = tracked_child.wait_handle.await.unwrap();
     assert!(termination.status.success());

crates/fspy/tests/test_utils/mod.rs

@@ -78,7 +78,11 @@ macro_rules! track_fn {
 )]
 #[allow(dead_code, reason = "used by track_fn! macro; not all test files use this macro")]
 pub async fn spawn_command(cmd: subprocess_test::Command) -> anyhow::Result<PathAccessIterable> {
-    let termination = fspy::Command::from(cmd).spawn().await?.wait_handle.await?;
+    let termination = fspy::Command::from(cmd)
+        .spawn(tokio_util::sync::CancellationToken::new())
+        .await?
+        .wait_handle
+        .await?;
     assert!(termination.status.success());
     Ok(termination.path_accesses)
 }

crates/fspy_e2e/Cargo.toml

@@ -9,6 +9,7 @@ fspy = { workspace = true }
 rustc-hash = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 tokio = { workspace = true, features = ["full"] }
+tokio-util = { workspace = true }
 toml = { workspace = true }
 
 [lints]

crates/fspy_e2e/src/main.rs

@@ -84,7 +84,8 @@ async fn main() {
             .stderr(Stdio::piped())
             .current_dir(&dir);
 
-        let mut tracked_child = cmd.spawn().await.unwrap();
+        let mut tracked_child =
+            cmd.spawn(tokio_util::sync::CancellationToken::new()).await.unwrap();
 
         let mut stdout_bytes = Vec::<u8>::new();
         tracked_child.stdout.take().unwrap().read_to_end(&mut stdout_bytes).await.unwrap();

crates/vite_task/Cargo.toml

@@ -22,6 +22,7 @@ fspy = { workspace = true }
 futures-util = { workspace = true }
 once_cell = { workspace = true }
 owo-colors = { workspace = true }
+petgraph = { workspace = true }
 pty_terminal_test_client = { workspace = true }
 rayon = { workspace = true }
 rusqlite = { workspace = true, features = ["bundled"] }
@@ -30,6 +31,7 @@ serde = { workspace = true, features = ["derive", "rc"] }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "io-std", "io-util", "macros", "sync"] }
+tokio-util = { workspace = true }
 tracing = { workspace = true }
 twox-hash = { workspace = true }
 vite_path = { workspace = true }

crates/vite_task/src/cli/mod.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{num::NonZeroUsize, str::FromStr, sync::Arc};
 
 use clap::Parser;
 use vite_path::AbsolutePath;
@@ -7,6 +7,35 @@ use vite_task_graph::{TaskSpecifier, query::TaskQuery};
 use vite_task_plan::plan_request::{CacheOverride, PlanOptions, QueryPlanRequest};
 use vite_workspace::package_filter::{PackageQueryArgs, PackageQueryError};
 
+/// Parsed `--concurrency` value: a positive integer or a percentage of logical CPUs.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Concurrency(pub usize);
+
+impl FromStr for Concurrency {
+    type Err = ConcurrencyParseError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        if let Some(pct) = s.strip_suffix('%') {
+            let pct_val: u32 = pct.parse().map_err(|_| ConcurrencyParseError(s.into()))?;
+            let cpus = std::thread::available_parallelism().map(NonZeroUsize::get).unwrap_or(1);
+            let result = (cpus as u64 * u64::from(pct_val) / 100).max(1) as usize;
+            Ok(Self(result))
+        } else {
+            let val: usize = s.parse().map_err(|_| ConcurrencyParseError(s.into()))?;
+            if val == 0 {
+                return Err(ConcurrencyParseError(s.into()));
+            }
+            Ok(Self(val))
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error(
+    "invalid concurrency value '{0}': expected a positive integer or a percentage (e.g. '4' or '50%')"
+)]
+pub struct ConcurrencyParseError(Str);
+
 #[derive(Debug, Clone, clap::Subcommand)]
 pub enum CacheSubcommand {
     /// Clean up all the cache
@@ -35,6 +64,11 @@ pub struct RunFlags {
     /// Force caching off for all tasks and scripts.
     #[clap(long, conflicts_with = "cache")]
     pub no_cache: bool,
+
+    /// Maximum number of tasks to run concurrently (default: 10).
+    /// Accepts a number (e.g. 4) or a percentage of logical CPUs (e.g. 50%).
+    #[clap(long)]
+    pub concurrency: Option<Concurrency>,
 }
 
 impl RunFlags {
@@ -193,6 +227,7 @@ impl ResolvedRunCommand {
                 plan_options: PlanOptions {
                     extra_args: self.additional_args.into(),
                     cache_override,
+                    concurrency: self.flags.concurrency.map(|c| c.0),
                 },
             },
             is_cwd_only,