feat: concurrent task execution with DAG scheduler

crates/fspy/Cargo.toml

@@ -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/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,
+        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/lib.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/unix/mod.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/windows/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/tests/cancellation.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/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 }
@@ -46,5 +48,8 @@ tempfile = { workspace = true }
 [target.'cfg(unix)'.dependencies]
 nix = { workspace = true }
 
+[target.'cfg(windows)'.dependencies]
+winapi = { workspace = true, features = ["handleapi", "jobapi2", "winnt"] }
+
 [lib]
 doctest = false