fix(windows): launch canonicalized batch shims

crates/aft/src/windows_command.rs

@@ -72,6 +72,12 @@ where
             "batch path cannot be represented safely for cmd.exe",
         ));
     }
+    // `fs::canonicalize` produces extended-length (`\\?\`) paths on Windows.
+    // CreateProcess accepts those paths, but cmd.exe does not reliably execute a
+    // batch file through that namespace, and npm shims additionally derive
+    // `%~dp0` paths that fail with "The system cannot find the path specified."
+    // Convert only the namespace spelling; the path remains canonical.
+    let command_path = cmd_compatible_path(command_path);
 
     let mut command_line = format!("\"\"%{BATCH_COMMAND_ENV}%\"");
     let mut argument_env = Vec::new();
@@ -108,15 +114,74 @@ where
         // argument escaping would add another quoting layer and break paths
         // containing spaces.
         .raw_arg(command_line)
-        .env(BATCH_COMMAND_ENV, binary)
+        .env(BATCH_COMMAND_ENV, command_path)
         .envs(argument_env);
     Ok(command)
 }
 
+#[cfg(windows)]
+fn cmd_compatible_path(path: &str) -> String {
+    for prefix in [r"\\?\UNC\", r"\\??\UNC\", r"\??\UNC\"] {
+        if let Some(tail) = strip_ascii_prefix(path, prefix) {
+            let mut components = tail
+                .split(['\\', '/'])
+                .filter(|component| !component.is_empty());
+            if components.next().is_some() && components.next().is_some() {
+                return format!(r"\\{tail}");
+            }
+            return path.to_string();
+        }
+    }
+
+    for prefix in [r"\\?\", r"\\??\", r"\??\"] {
+        if let Some(tail) = strip_ascii_prefix(path, prefix) {
+            let bytes = tail.as_bytes();
+            if bytes.len() >= 3
+                && bytes[0].is_ascii_alphabetic()
+                && bytes[1] == b':'
+                && matches!(bytes[2], b'\\' | b'/')
+            {
+                return tail.to_string();
+            }
+            // Namespaces such as `\\?\Volume{GUID}\` cannot be safely
+            // converted into a DOS path by dropping their prefix.
+            return path.to_string();
+        }
+    }
+
+    path.to_string()
+}
+
+#[cfg(windows)]
+fn strip_ascii_prefix<'a>(value: &'a str, prefix: &str) -> Option<&'a str> {
+    let head = value.get(..prefix.len())?;
+    if head.eq_ignore_ascii_case(prefix) {
+        value.get(prefix.len()..)
+    } else {
+        None
+    }
+}
+
 #[cfg(all(test, windows))]
 mod tests {
     use super::*;
 
+    #[test]
+    fn cmd_compatible_path_only_converts_dos_and_unc_namespaces() {
+        assert_eq!(
+            cmd_compatible_path(r"\\?\C:\cache\server.cmd"),
+            r"C:\cache\server.cmd"
+        );
+        assert_eq!(
+            cmd_compatible_path(r"\\?\unc\host\share\server.cmd"),
+            r"\\host\share\server.cmd"
+        );
+        assert_eq!(
+            cmd_compatible_path(r"\\?\Volume{1234}\server.cmd"),
+            r"\\?\Volume{1234}\server.cmd"
+        );
+    }
+
     #[test]
     fn batch_command_invokes_a_spaced_shim_with_args() {
         let temp = tempfile::tempdir().unwrap();
@@ -129,6 +194,38 @@ mod tests {
         assert_eq!(String::from_utf8_lossy(&output.stdout).trim(), "--stdio");
     }
 
+    #[test]
+    fn batch_command_invokes_canonicalized_npm_style_shim() {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("npm cache 100%");
+        let bin = root.join("node_modules").join(".bin");
+        let package = root.join("node_modules").join("language-server");
+        std::fs::create_dir_all(&bin).unwrap();
+        std::fs::create_dir_all(&package).unwrap();
+        let shim = bin.join("language-server.cmd");
+        let target = package.join("server.cmd");
+        std::fs::write(
+            &shim,
+            "@echo off\r\nset dp0=%~dp0\r\n\"%dp0%\\..\\language-server\\server.cmd\" %*\r\n",
+        )
+        .unwrap();
+        std::fs::write(&target, "@echo off\r\necho %~1\r\n").unwrap();
+        let canonical_shim = std::fs::canonicalize(&shim).unwrap();
+        assert!(canonical_shim.to_string_lossy().starts_with(r"\\?\"));
+
+        let output = batch_command(&canonical_shim, ["--stdio"])
+            .unwrap()
+            .output()
+            .unwrap();
+
+        assert!(
+            output.status.success(),
+            "stderr: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+        assert_eq!(String::from_utf8_lossy(&output.stdout).trim(), "--stdio");
+    }
+
     #[test]
     fn batch_command_preserves_percent_in_argument() {
         let temp = tempfile::tempdir().unwrap();

crates/aft/tests/integration/bash_background_persistence_test.rs

@@ -1732,6 +1732,20 @@ fn session_isolation_on_replay() {
 #[test]
 fn restart_sweep_marks_dead_pid_fate_unknown_once() {
     let storage = tempfile::tempdir().unwrap();
+    let registry = BgTaskRegistry::new(Arc::new(Mutex::new(None)));
+    // The first replay's persisted GC runs detached. Let that storage-wide
+    // sweep finish before planting a fixture, or it can observe and remove the
+    // fixture concurrently with this test's replay.
+    registry.replay_session(storage.path(), SESSION).unwrap();
+    let deadline = Instant::now() + Duration::from_secs(10);
+    while registry.persisted_gc_thread().is_none() {
+        assert!(
+            Instant::now() < deadline,
+            "first persisted GC never finished after replay"
+        );
+        std::thread::sleep(Duration::from_millis(10));
+    }
+
     let task_id = "bash-0000000000000120";
     let mut metadata = PersistedTask::starting(
         task_id.to_string(),
@@ -1753,7 +1767,6 @@ fn restart_sweep_marks_dead_pid_fate_unknown_once() {
     )
     .unwrap();
 
-    let registry = BgTaskRegistry::new(Arc::new(Mutex::new(None)));
     registry.replay_session(storage.path(), SESSION).unwrap();
     let replayed = read_json(storage.path(), SESSION, task_id);
     assert_eq!(replayed["status"], "fate_unknown");

crates/aft/tests/integration/bash_orchestrate_test.rs

@@ -391,15 +391,22 @@ fn bash_gate_off_still_returns_spawn_response() {
 fn pending_orchestrated_bash_does_not_starve_push_frames() {
     let mut aft = spawn_with_wait("5000");
     let dir = tempfile::tempdir().unwrap();
+    let child_started = dir.path().join("pending-bash-started");
+    let child_release = dir.path().join("pending-bash-release");
+    // Declare after the TempDir so a panic releases the gated child before
+    // the directory is removed.
+    let _release_guard = ReleaseOnDrop::new(child_release.clone());
+    let command = hold_until_release_command(&child_started, &child_release, "true");
 
     aft.send_silent(&bash_request(
         "bash-drain-orchestrated",
         json!({
-            "command": "sleep 2",
+            "command": command,
             "foreground_orchestrate": true,
             "block_to_completion": true,
         }),
     ));
+    wait_for_file(&child_started, "pending foreground child start");
 
     let configure = aft.send(
         &json!({
@@ -416,32 +423,31 @@ fn pending_orchestrated_bash_does_not_starve_push_frames() {
         "configure failed: {configure:?}"
     );
 
-    // The invariant: the pending deferred bash response must not starve push
-    // frames — configure_warnings has to arrive BEFORE that deferred response.
-    // Under host load the 2s sleep can finish while configure is still running,
-    // so the task's own completion push may legitimately land first; pushes
-    // overtaking pushes is not starvation. Tolerate the task's push frames and
-    // fail only if the deferred RESPONSE (an "id" frame) beats the warnings.
+    // Keep the child pending until configure_warnings has arrived. The former
+    // fixed `sleep 2` completion raced the warning worker's OS scheduling under
+    // loaded CI, so it could report a false starvation failure before the
+    // warning work had even run.
     let push = loop {
         let frame = aft
-            .try_read_next_timeout(Duration::from_secs(12))
-            .expect("configure warning push before deferred bash response");
+            .try_read_next_timeout(HANG_CATCH)
+            .expect("configure warning push while deferred bash remains pending");
         if frame.get("type").is_some() {
             if frame["type"] == "configure_warnings" {
                 break frame;
             }
             continue;
         }
-        panic!("deferred response overtook the configure push (starvation): {frame:?}");
+        panic!("deferred response settled before its gated child was released: {frame:?}");
     };
     assert_eq!(
         push["type"], "configure_warnings",
         "unexpected frame: {push:?}"
     );
 
+    std::fs::write(&child_release, b"release").expect("release pending foreground child");
     let bash_response = loop {
         let value = aft
-            .try_read_next_timeout(Duration::from_secs(12))
+            .try_read_next_timeout(HANG_CATCH)
             .expect("deferred bash response after command completion");
         if value["id"] == "bash-drain-orchestrated" {
             break value;