Merge remote-tracking branch 'origin/main' into review/pr1240

On Windows, deleting an agent task together with its worktree failed with "os error 32" when the task's agent was still running. Agents that can't reload their session (`session/load`) stay running until the task is deleted. Deleting sent the kill and then removed the worktree right away, while the agent's processes still had the folder as their working directory and files in it open.

`Peer::exited` now resolves once the child process has been killed along with what it started and has been reaped. Deleting a task waits for it (up to 5 s) before it removes the worktree. Removing the folder also retries for a few seconds while a file in it is still in use (a helper process that outlived the agent, an antivirus scan, the search indexer).

Fixes #969.

Cargo.lock

@@ -4341,6 +4341,7 @@ dependencies = [
  "parking_lot",
  "serde",
  "serde_json",
+ "tempfile",
  "thiserror 2.0.21",
  "tokio",
  "tokio-util",

crates/ide-agent/src/lib.rs

@@ -58,6 +58,8 @@ pub(crate) const LIMITS_UPDATE: &str = "openidea_limits";
 const MAX_EMBED: u64 = 256 * 1024;
 /// How long a probe waits for the agent to answer `initialize`.
 const PROBE_TIMEOUT: Duration = Duration::from_secs(30);
+/// How long deleting a task waits for its agent to exit before removing the worktree.
+const EXIT_WAIT: Duration = Duration::from_secs(5);
 
 /// The agent tasks of one project.
 pub struct AgentManager {
@@ -839,7 +841,7 @@ impl AgentManager {
     /// Forgets the task; with `remove_worktree` also deletes its worktree (the branch stays).
     pub async fn delete(&self, id: &str, remove_worktree: bool) -> IdeResult<()> {
         let task = self.task(id)?;
-        task.kill();
+        let live = task.kill();
         let (isolation, cwd, root) = {
             let s = task.state.lock();
             (
@@ -851,6 +853,11 @@ impl AgentManager {
         // A copy of the task (a fork) may work in the same worktree.
         let shared = self.tasks.read().values().any(|t| t.id != id && t.state.lock().task.cwd.0 == cwd);
         if remove_worktree && !shared && isolation == Isolation::Worktree && cwd.exists() {
+            // On Windows the agent's processes lock the folder (their working directory,
+            // open files) until they have exited.
+            if let Some(peer) = live {
+                let _ = tokio::time::timeout(EXIT_WAIT, peer.exited()).await;
+            }
             let worktrees = self.shared.paths.worktrees_dir.clone();
             blocking(move || worktree::remove(&root, &worktrees, &cwd)).await?;
         }
@@ -1605,11 +1612,12 @@ impl Task {
         }
     }
 
-    fn kill(&self) {
+    /// Stops the agent; returns its connection, to wait for its process to exit.
+    fn kill(&self) -> Option<acp::Conn> {
         self.cancel_permissions();
-        if let Some(live) = self.live.lock().take() {
-            live.peer.kill();
-        }
+        let live = self.live.lock().take()?;
+        live.peer.kill();
+        Some(live.peer)
     }
 }
 

crates/ide-agent/src/tests.rs

@@ -1109,3 +1109,114 @@ async fn idle_agents_stop_and_resume_their_session_on_the_next_message() {
     assert!(live(&mgr, &task.id));
     mgr.shutdown();
 }
+
+/// Deleting a task whose agent still runs in the worktree removes both and leaves no
+/// process behind, whether it is waiting for permission or idle without session/load.
+#[tokio::test(flavor = "multi_thread")]
+async fn deleting_a_task_waits_for_its_agent_to_exit_before_removing_the_worktree() {
+    for no_load in [false, true] {
+        delete_running_agent(no_load).await;
+    }
+}
+
+async fn delete_running_agent(no_load: bool) {
+    let tmp = tempfile::tempdir().unwrap();
+    let root = dunce::canonicalize(tmp.path()).unwrap();
+    let git = |args: &[&str]| {
+        let out = std::process::Command::new("git")
+            .args(["-c", "user.name=t", "-c", "user.email=t@t", "-c", "commit.gpgsign=false"])
+            .args(args)
+            .current_dir(&root)
+            .output()
+            .unwrap();
+        assert!(out.status.success(), "{}", String::from_utf8_lossy(&out.stderr));
+    };
+    git(&["init", "-q", "-b", "main"]);
+    std::fs::write(root.join("file.txt"), "original\n").unwrap();
+    git(&["add", "."]);
+    git(&["commit", "-qm", "initial"]);
+
+    let agent = Arc::new(FakeAgent { no_load, ..FakeAgent::default() });
+    let pids: Arc<Mutex<Vec<u32>>> = Arc::default();
+    let (fake, started) = (agent.clone(), pids.clone());
+    // The fake agent, served on behalf of a real process that exits only when killed.
+    let connector: Connector = Arc::new(move |_spec, cwd, client, _stderr| {
+        #[cfg(unix)]
+        let mut cmd = {
+            let mut cmd = tokio::process::Command::new("sleep");
+            cmd.arg("30");
+            cmd
+        };
+        #[cfg(windows)]
+        let mut cmd = {
+            let mut cmd = tokio::process::Command::new("cmd");
+            cmd.args(["/D", "/C", "echo ready>agent-ready & ping -n 30 127.0.0.1 >nul"]);
+            cmd
+        };
+        ide_core::process::prepare(cmd.as_std_mut());
+        ide_core::process::own_group(cmd.as_std_mut());
+        cmd.current_dir(cwd).kill_on_drop(true);
+        let child = cmd.spawn()?;
+        started.lock().push(child.id().unwrap());
+        let (a, b) = duplex(1 << 20);
+        let (ar, aw) = tokio::io::split(a);
+        let (br, bw) = tokio::io::split(b);
+        // `connect_with_child` frames with Content-Length.
+        fake.servers.lock().push(Peer::connect(br, bw, fake.clone()));
+        Ok(Peer::connect_with_child(ar, aw, client, child))
+    });
+    let mgr = AgentManager::with_connector(
+        AgentPaths {
+            project_root: root.clone(),
+            store_dir: root.join("store"),
+            worktrees_dir: root.join("worktrees"),
+            models_file: root.join("models.json"),
+            custom_agents_dir: None,
+            home_dir: None,
+        },
+        tokio::runtime::Handle::current(),
+        connector,
+    );
+    let task = mgr
+        .create(NewAgentTask {
+            project_root: None,
+            prompt: text(if no_load { "again" } else { "Fix the lexer" }),
+            provider: "claude".into(),
+            config: vec![],
+            isolation: Isolation::Worktree,
+            base: None,
+        })
+        .await
+        .unwrap();
+    wait_for(&mgr, &task.id, if no_load { AgentStatus::Ready } else { AgentStatus::NeedsInput }).await;
+    let pid = *pids.lock().last().expect("the agent was started");
+
+    #[cfg(windows)]
+    {
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while !task.cwd.0.join("agent-ready").exists() {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .unwrap();
+        assert_eq!(
+            std::fs::rename(&task.cwd.0, task.cwd.0.with_extension("moved")).unwrap_err().raw_os_error(),
+            Some(32),
+            "the running agent really locks its worktree"
+        );
+    }
+
+    mgr.delete(&task.id, true).await.unwrap();
+    assert!(!task.cwd.0.exists());
+    // Killed and reaped: not even a zombie is left.
+    #[cfg(unix)]
+    let alive = std::process::Command::new("kill").args(["-0", &pid.to_string()]).status().unwrap().success();
+    #[cfg(windows)]
+    let alive = {
+        let mut cmd = std::process::Command::new("tasklist");
+        ide_core::process::prepare(&mut cmd).args(["/FI", &format!("PID eq {pid}"), "/FO", "CSV", "/NH"]);
+        String::from_utf8_lossy(&cmd.output().unwrap().stdout).contains(&format!("\"{pid}\""))
+    };
+    assert!(!alive, "the agent (pid {pid}) is still running");
+}

crates/ide-agent/src/worktree.rs

@@ -5,7 +5,7 @@ use ide_core::{IdeError, IdeResult, IpcPath};
 use ide_vcs::{ChangeKind, Repo};
 use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
-use std::time::UNIX_EPOCH;
+use std::time::{Duration, UNIX_EPOCH};
 
 pub(crate) struct Created {
     pub path: PathBuf,
@@ -65,7 +65,27 @@ pub(crate) fn remove(project_root: &Path, worktrees_dir: &Path, path: &Path) ->
     if !path.starts_with(worktrees_dir) || path == worktrees_dir {
         return Err(err);
     }
-    std::fs::remove_dir_all(path).map_err(|e| IdeError::internal(e.to_string()))
+    retrying(REMOVE_ATTEMPTS, REMOVE_PAUSE, || std::fs::remove_dir_all(path))
+        .map_err(|e| IdeError::internal(format!("{}: {e}", path.display())))
+}
+
+/// On Windows a folder can't be deleted while a process still has a file in it open
+/// (os error 32): an agent's helper that outlived it, an antivirus scan, the search indexer.
+/// Those let go within moments, so removing is retried for a few seconds.
+const REMOVE_ATTEMPTS: u32 = 10;
+const REMOVE_PAUSE: Duration = Duration::from_millis(300);
+
+fn retrying(attempts: u32, pause: Duration, mut op: impl FnMut() -> std::io::Result<()>) -> std::io::Result<()> {
+    for _ in 1..attempts {
+        match op() {
+            Err(e) if e.kind() != std::io::ErrorKind::NotFound => std::thread::sleep(pause),
+            _ => return Ok(()),
+        }
+    }
+    match op() {
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
+        r => r,
+    }
 }
 
 /// Files under `cwd` that differ from `base` (untracked files included), with line counts.
@@ -149,6 +169,54 @@ fn unique_slug(base: &str, taken: impl Fn(&str) -> bool) -> String {
 mod tests {
     use super::*;
 
+    #[test]
+    fn removing_is_retried_while_a_file_is_still_in_use() {
+        use std::io::{Error, ErrorKind};
+        // A sharing violation as Windows reports it (os error 32), a vanished folder, other errors.
+        let in_use = || Error::from_raw_os_error(32);
+        for (fails, outcome) in [(0, true), (3, true), (REMOVE_ATTEMPTS - 1, true), (REMOVE_ATTEMPTS, false)] {
+            let mut calls = 0;
+            let r = retrying(REMOVE_ATTEMPTS, Duration::ZERO, || {
+                calls += 1;
+                if calls <= fails { Err(in_use()) } else { Ok(()) }
+            });
+            assert_eq!(r.is_ok(), outcome, "{fails} failures");
+            assert_eq!(calls, (fails + 1).min(REMOVE_ATTEMPTS));
+        }
+        let mut calls = 0;
+        let gone = retrying(REMOVE_ATTEMPTS, Duration::ZERO, || {
+            calls += 1;
+            if calls == 1 { Err(in_use()) } else { Err(Error::from(ErrorKind::NotFound)) }
+        });
+        assert!(gone.is_ok(), "deleted meanwhile: done");
+        assert_eq!(calls, 2);
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn removing_waits_for_a_windows_file_lock_to_be_released() {
+        use std::os::windows::fs::OpenOptionsExt;
+        let tmp = tempfile::tempdir().unwrap();
+        let repo = tmp.path().join("repo");
+        std::fs::create_dir(&repo).unwrap();
+        git(&repo, &["init", "-q", "-b", "main"]);
+        std::fs::write(repo.join("a.txt"), "one\n").unwrap();
+        git(&repo, &["add", "."]);
+        git(&repo, &["commit", "-q", "-m", "init"]);
+        let worktrees = tmp.path().join("wt");
+        let created = create(&repo, &worktrees, "Locked file", None).unwrap();
+        let file = std::fs::OpenOptions::new().read(true).share_mode(0).open(created.path.join("a.txt")).unwrap();
+        assert_eq!(std::fs::remove_file(created.path.join("a.txt")).unwrap_err().raw_os_error(), Some(32));
+        let release = std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(800));
+            drop(file);
+        });
+        let result = remove(&repo, &worktrees, &created.path);
+        release.join().unwrap();
+        result.unwrap();
+        assert!(!created.path.exists());
+    }
+
     #[test]
     fn slugs_are_branch_safe() {
         assert_eq!(slugify("Fix lexer UTF-16 columns!"), "fix-lexer-utf-16-columns");

crates/ide-jsonrpc/Cargo.toml

@@ -16,3 +16,6 @@ parking_lot.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
 encoding_rs.workspace = true
+
+[dev-dependencies]
+tempfile.workspace = true

crates/ide-jsonrpc/src/lib.rs

@@ -243,6 +243,9 @@ struct Inner {
     pid: Option<u32>,
     /// Fired when the reader task has stopped (EOF, error or kill).
     reader_done: CancellationToken,
+    /// Fired once the child process has exited and been reaped (after [`Peer::kill`],
+    /// also what it started); at once when there is no child.
+    exited: CancellationToken,
 }
 
 impl Inner {
@@ -432,6 +435,7 @@ impl Peer {
                     (child.wait().await.ok(), true)
                 }
             };
+            inner.exited.cancel();
             if !killed {
                 // Bounded: a grandchild may keep the pipes open.
                 let drained = async {
@@ -510,7 +514,11 @@ impl Peer {
             kill: CancellationToken::new(),
             pid,
             reader_done: CancellationToken::new(),
+            exited: CancellationToken::new(),
         });
+        if pid.is_none() {
+            inner.exited.cancel();
+        }
 
         // Writer: a single task drains the queue so messages never interleave.
         {
@@ -704,6 +712,13 @@ impl Peer {
         self.inner.close_transport();
         self.inner.kill.cancel();
     }
+
+    /// Resolves once the child process is gone: exited on its own, or killed with what it
+    /// started and reaped. [`closed`](Self::closed) can come earlier, at EOF. Until then the
+    /// child may still hold files open, which on Windows can't be deleted.
+    pub async fn exited(&self) {
+        self.inner.exited.cancelled().await;
+    }
 }
 
 /// Removes the pending entry and tells the remote side to stop when a request

crates/ide-jsonrpc/src/tests.rs

@@ -413,6 +413,65 @@ async fn kill_terminates_process() {
     tokio::time::timeout(Duration::from_secs(5), peer.closed()).await.unwrap();
 }
 
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread")]
+async fn exited_resolves_once_the_killed_process_is_reaped() {
+    let peer =
+        Peer::spawn("sleep", &["30".into()], &std::env::temp_dir(), &[], Arc::new(NoopHandler), Box::new(|_| {}))
+            .unwrap();
+    let pid = peer.pid().unwrap().to_string();
+    let exited = tokio::time::timeout(Duration::from_millis(200), peer.exited()).await;
+    assert!(exited.is_err(), "running until killed");
+    peer.kill();
+    tokio::time::timeout(Duration::from_secs(5), peer.exited()).await.unwrap();
+    let alive = std::process::Command::new("kill").args(["-0", &pid]).status().unwrap().success();
+    assert!(!alive, "reaped, not even a zombie");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn exited_resolves_at_once_without_a_process() {
+    let (client, _server) = pair(Arc::new(NoopHandler), Arc::new(Recorder::default()));
+    tokio::time::timeout(Duration::from_secs(1), client.exited()).await.unwrap();
+}
+
+#[cfg(windows)]
+#[tokio::test(flavor = "multi_thread")]
+async fn exited_waits_for_a_windows_descendant_to_release_its_file() {
+    use std::os::windows::fs::OpenOptionsExt;
+    let tmp = tempfile::tempdir().unwrap();
+    std::fs::write(
+        tmp.path().join("hold.ps1"),
+        "$f=[IO.File]::Open('locked.txt','Create','ReadWrite','None'); [IO.File]::WriteAllText('ready.txt','ready'); Start-Sleep 30",
+    )
+    .unwrap();
+    let peer = Peer::spawn(
+        "cmd",
+        &[
+            "/D".into(),
+            "/C".into(),
+            "powershell -NoProfile -NonInteractive -ExecutionPolicy Bypass -File hold.ps1".into(),
+        ],
+        tmp.path(),
+        &[],
+        Arc::new(NoopHandler),
+        Box::new(|_| {}),
+    )
+    .unwrap();
+    tokio::time::timeout(Duration::from_secs(10), async {
+        while !tmp.path().join("ready.txt").exists() {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    let open = || std::fs::OpenOptions::new().read(true).share_mode(0).open(tmp.path().join("locked.txt"));
+    assert_eq!(open().unwrap_err().raw_os_error(), Some(32));
+    assert!(tokio::time::timeout(Duration::from_millis(200), peer.exited()).await.is_err());
+    peer.kill();
+    tokio::time::timeout(Duration::from_secs(5), peer.exited()).await.unwrap();
+    drop(open().expect("exited must wait for the descendant's file handle to close"));
+}
+
 /// A launcher that prints in the system code page and exits before speaking the
 /// protocol: `jdtls.bat` on a Chinese Windows when its Python step fails, then
 /// `pause` prints `请按任意键继续. . .` in GBK on stdout.