feat: concurrent task execution with DAG scheduler (#288)

## Summary

- Replace sequential task execution with a concurrent DAG scheduler using `FuturesUnordered` + `tokio::sync::Semaphore` (limit 10 per graph level)
- Each nested `Expanded` graph gets its own semaphore for independent concurrency limits
- Failure propagation via `CancellationToken` — any task failure immediately kills all in-flight processes
- Add `barrier` test tool (cross-platform, `fs.watch`-based) for concurrency testing
- Add e2e tests proving concurrent execution and kill-on-failure behavior

## Test plan

- [x] All existing e2e snapshot tests pass (10 consecutive runs)
- [x] New `concurrent-execution` fixture proves independent tasks run concurrently (barrier with 2 participants — would timeout if sequential)
- [x] Failure test proves cancellation kills concurrent tasks (task b hangs on stdin after barrier, killed when task a fails)
- [x] `cargo clippy` clean
- [x] `cargo test -p vite_task` unit tests pass

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: branchseer

Cargo.lock

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

crates/fspy/Cargo.toml

@@ -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/examples/cli.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,
+        cancellation_token: CancellationToken,
+    ) -> Result<TrackedChild, SpawnError> {
         self.resolve_program()?;
-        SPY_IMPL.spawn(self).await
+        SPY_IMPL.spawn(self, cancellation_token).await
     }
 
     /// Resolve program name to full path using `PATH` and cwd.

crates/fspy/src/command.rs

@@ -54,6 +54,13 @@ pub struct TrackedChild {
 
     /// The future that resolves to exit status and path accesses when the process exits.
     pub wait_handle: BoxFuture<'static, io::Result<ChildTermination>>,
+
+    /// A duplicated process handle of the child, captured before the tokio `Child`
+    /// is moved into the background wait task. This is an independently owned handle
+    /// (via `DuplicateHandle`) so it remains valid even after tokio closes its copy.
+    /// Callers can use this to assign the process to a Win32 Job Object.
+    #[cfg(windows)]
+    pub process_handle: std::os::windows::io::OwnedHandle,
 }
 
 pub(crate) static SPY_IMPL: LazyLock<SpyImpl> = LazyLock::new(|| {

crates/fspy/src/lib.rs

@@ -22,6 +22,7 @@ use futures_util::FutureExt;
 #[cfg(target_os = "linux")]
 use syscall_handler::SyscallHandler;
 use tokio::task::spawn_blocking;
+use tokio_util::sync::CancellationToken;
 
 use crate::{
     ChildTermination, Command, TrackedChild,
@@ -80,7 +81,11 @@ impl SpyImpl {
         })
     }
 
-    pub(crate) async fn spawn(&self, mut command: Command) -> Result<TrackedChild, SpawnError> {
+    pub(crate) async fn spawn(
+        &self,
+        mut command: Command,
+        cancellation_token: CancellationToken,
+    ) -> Result<TrackedChild, SpawnError> {
         #[cfg(target_os = "linux")]
         let supervisor = supervise::<SyscallHandler>().map_err(SpawnError::Supervisor)?;
 
@@ -143,7 +148,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?,
+                    () = cancellation_token.cancelled() => {
+                        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/unix/mod.rs

@@ -13,6 +13,7 @@ use fspy_shared::{
     windows::{PAYLOAD_ID, Payload},
 };
 use futures_util::FutureExt;
+use tokio_util::sync::CancellationToken;
 use winapi::{
     shared::minwindef::TRUE,
     um::{processthreadsapi::ResumeThread, winbase::CREATE_SUSPENDED},
@@ -73,7 +74,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,
+        cancellation_token: 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();
@@ -135,14 +140,32 @@ impl SpyImpl {
                 if *spawn_success { SpawnError::OsSpawn(err) } else { SpawnError::Injection(err) }
             })?;
 
+        // Duplicate the process handle before the child is moved into the background
+        // task. The duplicate is independently owned (its own ref count), so it stays
+        // valid even after tokio closes its copy when the process exits.
+        let process_handle = {
+            use std::os::windows::io::BorrowedHandle;
+            // SAFETY: The child was just spawned and hasn't been moved yet, so its
+            // raw handle is valid. `borrow_raw` creates a temporary borrow.
+            let borrowed = unsafe { BorrowedHandle::borrow_raw(child.raw_handle().unwrap()) };
+            borrowed.try_clone_to_owned().map_err(SpawnError::OsSpawn)?
+        };
+
         Ok(TrackedChild {
             stdin: child.stdin.take(),
             stdout: child.stdout.take(),
             stderr: child.stderr.take(),
+            process_handle,
             // 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?,
+                    () = cancellation_token.cancelled() => {
+                        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/src/windows/mod.rs

@@ -0,0 +1,32 @@
+use std::process::Stdio;
+
+use tokio::io::AsyncReadExt as _;
+use tokio_util::sync::CancellationToken;
+
+#[test_log::test(tokio::test)]
+async fn cancellation_kills_tracked_child() -> anyhow::Result<()> {
+    let cmd = subprocess_test::command_for_fn!((), |()| {
+        use std::io::Write as _;
+        // Signal readiness via stdout
+        std::io::stdout().write_all(b"ready\n").unwrap();
+        std::io::stdout().flush().unwrap();
+        // Block on stdin — will be killed by cancellation
+        let _ = std::io::stdin().read_line(&mut String::new());
+    });
+    let token = CancellationToken::new();
+    let mut fspy_cmd = fspy::Command::from(cmd);
+    fspy_cmd.stdout(Stdio::piped()).stdin(Stdio::piped());
+    let mut child = fspy_cmd.spawn(token.clone()).await?;
+
+    // Wait for child to signal readiness
+    let mut stdout = child.stdout.take().unwrap();
+    let mut buf = vec![0u8; 64];
+    let n = stdout.read(&mut buf).await?;
+    assert!(std::str::from_utf8(&buf[..n])?.contains("ready"));
+
+    // Cancel — fspy background task calls start_kill
+    token.cancel();
+    let termination = child.wait_handle.await?;
+    assert!(!termination.status.success());
+    Ok(())
+}

crates/fspy/tests/cancellation.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/node_fs.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)