diff --git a/Cargo.lock b/Cargo.lock index 76ff7212..6713938e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4341,6 +4341,7 @@ dependencies = [ "parking_lot", "serde", "serde_json", + "tempfile", "thiserror 2.0.21", "tokio", "tokio-util", diff --git a/crates/ide-agent/src/lib.rs b/crates/ide-agent/src/lib.rs index a65400a6..7116047f 100644 --- a/crates/ide-agent/src/lib.rs +++ b/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 { 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) } } diff --git a/crates/ide-agent/src/tests.rs b/crates/ide-agent/src/tests.rs index e33338a7..e1da00d1 100644 --- a/crates/ide-agent/src/tests.rs +++ b/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>> = 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"); +} diff --git a/crates/ide-agent/src/worktree.rs b/crates/ide-agent/src/worktree.rs index b0e2f62f..ddd45c41 100644 --- a/crates/ide-agent/src/worktree.rs +++ b/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"); diff --git a/crates/ide-jsonrpc/Cargo.toml b/crates/ide-jsonrpc/Cargo.toml index 615d1f81..4161989a 100644 --- a/crates/ide-jsonrpc/Cargo.toml +++ b/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 diff --git a/crates/ide-jsonrpc/src/lib.rs b/crates/ide-jsonrpc/src/lib.rs index c04331ad..929fcef3 100644 --- a/crates/ide-jsonrpc/src/lib.rs +++ b/crates/ide-jsonrpc/src/lib.rs @@ -243,6 +243,9 @@ struct Inner { pid: Option, /// 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 diff --git a/crates/ide-jsonrpc/src/tests.rs b/crates/ide-jsonrpc/src/tests.rs index 7db0823f..0eb57031 100644 --- a/crates/ide-jsonrpc/src/tests.rs +++ b/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.