diff --git a/crates/ide-agent/src/lib.rs b/crates/ide-agent/src/lib.rs index d85bd0b77..66786c1d6 100644 --- a/crates/ide-agent/src/lib.rs +++ b/crates/ide-agent/src/lib.rs @@ -29,7 +29,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque}; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Weak}; use std::time::Duration; @@ -44,6 +44,9 @@ const REPLAY_GRACE: Duration = Duration::from_millis(300); /// can trail its `session/prompt` response by this much. const TURN_GRACE: Duration = Duration::from_millis(100); const STDERR_LINES: usize = 20; +/// An agent idle this long after its turn is stopped when it can reload its session +/// (`session/load`): the next message starts it again where it left off. +const IDLE_RELEASE: Duration = Duration::from_secs(5 * 60); /// Updates of a session's state rather than its conversation. const SESSION_STATE: [&str; 5] = ["session_info_update", "config_option_update", "current_mode_update", "available_commands_update", "usage_update"]; @@ -92,6 +95,8 @@ struct Shared { providers: RwLock>, /// Last probe of each provider. probes: Mutex>, + /// [`IDLE_RELEASE`], in milliseconds (shorter in tests). + idle_release_ms: AtomicU64, } /// What an agent offered in its last session, for the New Task page. @@ -207,6 +212,8 @@ struct Live { capabilities: AgentCapabilities, /// `promptCapabilities.embeddedContext`: referenced files are sent with their text. embedded: bool, + /// `session/load`: the process can be stopped between turns and the session reloaded. + resumable: bool, } struct Task { @@ -228,6 +235,8 @@ struct Task { cancel_pending: AtomicBool, /// The agent reported something for the current turn. turn_started: AtomicBool, + /// Turns sent so far: an idle release scheduled before the latest one is void. + turns: AtomicU64, /// Settings already sent this session (`(option, value)`), so a refusal is not retried. config_sent: Mutex>, dirty: Mutex>, @@ -238,6 +247,11 @@ struct Task { } impl AgentManager { + #[cfg(test)] + fn set_idle_release(&self, delay: Duration) { + self.shared.idle_release_ms.store(delay.as_millis() as u64, Ordering::Relaxed); + } + /// Loads the project's saved tasks. Spawned work runs on `runtime`. pub fn new(paths: AgentPaths, runtime: tokio::runtime::Handle) -> Self { let connector: Connector = Arc::new(|spec, cwd, client, on_stderr| match spec.driver { @@ -262,6 +276,7 @@ impl AgentManager { known: Mutex::new(known), providers, probes: Mutex::new(HashMap::new()), + idle_release_ms: AtomicU64::new(IDLE_RELEASE.as_millis() as u64), }); let mut tasks = HashMap::new(); for file in std::fs::read_dir(&shared.paths.store_dir).into_iter().flatten().flatten() { @@ -888,6 +903,7 @@ impl Task { send_now: AtomicBool::new(false), cancel_pending: AtomicBool::new(false), turn_started: AtomicBool::new(false), + turns: AtomicU64::new(0), config_sent: Mutex::new(HashSet::new()), dirty: Mutex::new(BTreeSet::new()), flush_scheduled: AtomicBool::new(false), @@ -1030,12 +1046,42 @@ impl Task { } match (next, result) { (Some(queued), _) => prompt = queued.prompt, - (None, Ok(_)) => return self.set_status(AgentStatus::Ready, None), + (None, Ok(_)) => { + self.set_status(AgentStatus::Ready, None); + return self.release_when_idle(); + } (None, Err(failure)) => return self.fail(failure), } } } + /// Stops the agent's process once the task has stayed idle for [`IDLE_RELEASE`], so + /// finished tasks don't keep an agent (and what it started) running. Only an agent that + /// reloads its session: the next message starts it again with the conversation intact. + fn release_when_idle(self: &Arc) { + let turn = self.turns.load(Ordering::Acquire); + let delay = Duration::from_millis(self.shared.idle_release_ms.load(Ordering::Relaxed)); + let task = Arc::downgrade(self); + self.shared.runtime.spawn(async move { + tokio::time::sleep(delay).await; + let Some(task) = task.upgrade() else { return }; + if task.turns.load(Ordering::Acquire) != turn + || task.state.lock().busy + || !task.permissions.lock().is_empty() + { + return; + } + let live = { + let mut live = task.live.lock(); + if live.as_ref().is_some_and(|l| l.resumable && !l.peer.is_closed()) { live.take() } else { None } + }; + if let Some(live) = live { + tracing::debug!("agent task {}: idle, stopping the agent until the next message", task.id); + live.peer.kill(); + } + }); + } + /// Ends a turn that failed: nothing it started stays open. fn fail(self: &Arc, failure: Failure) { self.cancel_permissions(); @@ -1096,6 +1142,7 @@ impl Task { /// Sends one prompt, starting the agent first if needed; returns the stop reason. async fn turn(self: &Arc, prompt: &AgentPrompt) -> Result { + self.turns.fetch_add(1, Ordering::AcqRel); self.turn_started.store(false, Ordering::Release); self.state.lock().task.resets_ms = None; let attachments = prompt @@ -1292,7 +1339,7 @@ impl Task { } self.persist(); self.emit_task(); - let live = Live { peer, session: session_id, capabilities, embedded }; + let live = Live { peer, session: session_id, capabilities, embedded, resumable: can_load }; *self.live.lock() = Some(live.clone()); Ok(live) } diff --git a/crates/ide-agent/src/native/claude/cli.rs b/crates/ide-agent/src/native/claude/cli.rs index fca21d858..26e3204e1 100644 --- a/crates/ide-agent/src/native/claude/cli.rs +++ b/crates/ide-agent/src/native/claude/cli.rs @@ -98,6 +98,8 @@ impl Cli for Process { use std::process::Stdio; let mut cmd = tokio::process::Command::new(&l.program); ide_core::process::prepare(cmd.as_std_mut()); + // `claude` may be a wrapper (`claude.cmd` → node): stopping it stops what it started. + ide_core::process::own_group(cmd.as_std_mut()); cmd.args(&l.args) .current_dir(&l.cwd) .envs(ide_core::process::shell_env()) @@ -120,6 +122,9 @@ impl Cli for Process { tokio::select! { _ = child.wait() => {} _ = token.cancelled() => { + if let Some(pid) = child.id() { + let _ = tokio::task::spawn_blocking(move || ide_core::process::kill_tree_of(pid)).await; + } let _ = child.start_kill(); let _ = child.wait().await; } diff --git a/crates/ide-agent/src/tests.rs b/crates/ide-agent/src/tests.rs index eb64fd67c..e33338a77 100644 --- a/crates/ide-agent/src/tests.rs +++ b/crates/ide-agent/src/tests.rs @@ -9,6 +9,8 @@ struct FakeAgent { /// Every connection's agent side, kept open. servers: Mutex>, calls: Mutex>, + /// Without `loadSession` (a session lives only as long as its process). + no_load: bool, } impl FakeAgent { @@ -89,7 +91,7 @@ impl Handler for FakeAgent { if params.pointer("/clientCapabilities/subagents").is_some_and(Value::is_object) { self.calls.lock().push("subagents".into()); } - Ok(json!({"protocolVersion": 1, "agentCapabilities": {"loadSession": true, + Ok(json!({"protocolVersion": 1, "agentCapabilities": {"loadSession": !self.no_load, "sessionCapabilities": {"fork": {}}, "_meta": {"openidea": {"rollback": true, "steer": true}}}})) } "session/new" => Ok(json!({"sessionId": "s1", @@ -1054,3 +1056,56 @@ async fn reload_providers_updates_registry_install_clears_stale_probes_and_notif let probe = mgr.probe("copilot", false).await.unwrap(); assert!(probe.error.is_none()); } + +#[tokio::test(flavor = "multi_thread")] +async fn idle_agents_stop_and_resume_their_session_on_the_next_message() { + let new = || NewAgentTask { + project_root: None, + prompt: text("again"), + provider: "claude".into(), + config: Vec::new(), + isolation: Isolation::Local, + base: None, + }; + let live = + |mgr: &AgentManager, id: &str| mgr.task(id).unwrap().live.lock().as_ref().is_some_and(|l| !l.peer.is_closed()); + + let tmp = tempfile::tempdir().unwrap(); + let agent = Arc::new(FakeAgent::default()); + let mgr = manager(tmp.path(), agent.clone()); + mgr.set_idle_release(Duration::from_millis(200)); + let task = mgr.create(new()).await.unwrap(); + wait_for(&mgr, &task.id, AgentStatus::Ready).await; + assert!(live(&mgr, &task.id)); + // The process is stopped once the task stays idle… + tokio::time::sleep(Duration::from_millis(600)).await; + assert!(!live(&mgr, &task.id)); + assert!(agent.servers.lock()[0].is_closed(), "its connection is closed"); + assert_eq!(mgr.task(&task.id).unwrap().state.lock().task.status, AgentStatus::Ready, "not an error"); + // …and the next message starts it again on the same session. + mgr.prompt(&task.id, text("again"), false).await.unwrap(); + wait_for(&mgr, &task.id, AgentStatus::Ready).await; + assert!(agent.calls.lock().contains(&"session/load".to_owned())); + assert_eq!( + mgr.entries(&task.id, None).unwrap().iter().filter(|e| matches!(e, AgentEntry::Agent { .. })).count(), + 2 + ); + + // A new turn before the delay ran out keeps the agent. + mgr.prompt(&task.id, text("again"), false).await.unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + wait_for(&mgr, &task.id, AgentStatus::Ready).await; + assert!(live(&mgr, &task.id)); + mgr.shutdown(); + + // An agent that can't reload its session keeps running: stopping it would lose the conversation. + let tmp = tempfile::tempdir().unwrap(); + let agent = Arc::new(FakeAgent { no_load: true, ..FakeAgent::default() }); + let mgr = manager(tmp.path(), agent.clone()); + mgr.set_idle_release(Duration::from_millis(200)); + let task = mgr.create(new()).await.unwrap(); + wait_for(&mgr, &task.id, AgentStatus::Ready).await; + tokio::time::sleep(Duration::from_millis(600)).await; + assert!(live(&mgr, &task.id)); + mgr.shutdown(); +} diff --git a/crates/ide-core/src/process.rs b/crates/ide-core/src/process.rs index b7a74e44f..49565b3da 100644 --- a/crates/ide-core/src/process.rs +++ b/crates/ide-core/src/process.rs @@ -53,6 +53,31 @@ pub fn kill_tree(child: &mut Child) { let _ = child.wait(); } +/// [`kill_tree`] for a child known by its process id (a `tokio` child): `SIGKILL` to its +/// process group (spawn it with [`own_group`]) or `taskkill /T /F`. Call it before the child +/// is reaped, so the id still names it. The child itself is left to its owner to kill and reap. +pub fn kill_tree_of(pid: u32) { + #[cfg(unix)] + if let Ok(pid) = libc::pid_t::try_from(pid) { + // SAFETY: plain syscall; see `signal_group`. + unsafe { + libc::kill(-pid, libc::SIGKILL); + } + } + #[cfg(windows)] + { + let mut taskkill = Command::new("taskkill"); + prepare(&mut taskkill) + .args(["/T", "/F", "/PID", &pid.to_string()]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let _ = taskkill.status(); + } + #[cfg(not(any(unix, windows)))] + let _ = pid; +} + /// [`kill_tree`] that first asks the group to exit: `SIGTERM`, then `grace`, then `SIGKILL`. /// Git removes its `index.lock` on `SIGTERM`, where `SIGKILL` leaves it behind and the next /// git command fails. Windows has no such signal to send from outside the child's console, @@ -395,6 +420,28 @@ fn append_path_entries(path: &OsStr, extra: &OsStr) -> OsString { mod tests { use super::*; + #[cfg(unix)] + #[test] + fn kill_tree_of_reaches_what_the_child_started() { + use std::io::BufRead; + let mut cmd = Command::new("sh"); + own_group(&mut cmd).args(["-c", "sleep 30 & echo $!; wait"]).stdout(std::process::Stdio::piped()); + let mut child = cmd.spawn().unwrap(); + let mut line = String::new(); + std::io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut line).unwrap(); + let grandchild: libc::pid_t = line.trim().parse().unwrap(); + let alive = |pid: libc::pid_t| unsafe { libc::kill(pid, 0) } == 0; + assert!(alive(grandchild)); + kill_tree_of(child.id()); + let _ = child.wait(); + // Reparented to init once its parent is gone, which reaps it. + let gone = (0..100).any(|_| { + std::thread::sleep(Duration::from_millis(20)); + !alive(grandchild) + }); + assert!(gone, "the grandchild {grandchild} survived"); + } + #[cfg(windows)] #[test] fn user_path_reads_the_current_users_path() { diff --git a/crates/ide-host/src/project.rs b/crates/ide-host/src/project.rs index 63d69ae8c..82e4b0579 100644 --- a/crates/ide-host/src/project.rs +++ b/crates/ide-host/src/project.rs @@ -494,12 +494,14 @@ impl Host { ide_agent::AgentPaths { project_root: root.to_path_buf(), store_dir: self.data_dir.join("agents").join(project_key), - // Outside the app data dir: its path has spaces on macOS, which trips up build scripts. - worktrees_dir: dirs::home_dir() - .unwrap_or_else(|| self.data_dir.clone()) - .join(".openidea") - .join("worktrees") - .join(project_key), + worktrees_dir: agent_worktrees_dir(root, || { + // Outside the app data dir: its path has spaces on macOS, which trips up build scripts. + dirs::home_dir() + .unwrap_or_else(|| self.data_dir.clone()) + .join(".openidea") + .join("worktrees") + .join(project_key) + }), models_file: self.data_dir.join("agents").join("models.json"), custom_agents_dir: Some(self.config_dir.join("agents")), home_dir: dirs::home_dir(), @@ -1004,6 +1006,20 @@ fn content_root_watcher( .ok() } +/// Where agent tasks create their worktrees: next to the project, `/.worktrees`, +/// so they sit on the project's drive and outside its tree (not indexed, watched or shown in +/// its git status); `fallback` for a project at a filesystem root. +fn agent_worktrees_dir(root: &Path, fallback: impl FnOnce() -> PathBuf) -> PathBuf { + match (root.parent(), root.file_name()) { + (Some(parent), Some(name)) => { + let mut dir = name.to_os_string(); + dir.push(".worktrees"); + parent.join(dir) + } + _ => fallback(), + } +} + pub(crate) fn unlinked_maven_poms(root: &Path, settings: &ide_core::settings::Settings) -> Vec { ide_build::maven::UnlinkedProjects::load(root, &settings.languages.java.maven).poms().map(PathBuf::from).collect() } @@ -1279,3 +1295,17 @@ mod workspace_edit_tests { assert_eq!(documents.snapshot(doc.id).unwrap().text, "sum_values();\n"); } } + +#[cfg(test)] +mod agent_worktree_tests { + use super::agent_worktrees_dir; + use std::path::{Path, PathBuf}; + + #[test] + fn agent_worktrees_sit_next_to_the_project() { + let fallback = || PathBuf::from("/home/me/.openidea/worktrees/key"); + let project = Path::new("/work").join("acme-api"); + assert_eq!(agent_worktrees_dir(&project, fallback), Path::new("/work").join("acme-api.worktrees")); + assert_eq!(agent_worktrees_dir(Path::new("/"), fallback), fallback()); + } +} diff --git a/crates/ide-jsonrpc/src/lib.rs b/crates/ide-jsonrpc/src/lib.rs index b7af9fcfc..c04331ad9 100644 --- a/crates/ide-jsonrpc/src/lib.rs +++ b/crates/ide-jsonrpc/src/lib.rs @@ -384,6 +384,9 @@ impl Peer { let resolved = resolve_spawn_program(command, env); let mut cmd = tokio::process::Command::new(&resolved); ide_core::process::prepare(cmd.as_std_mut()); + // Wrappers (`npx`, `codex.cmd`, launcher scripts) start the real server as a child: + // stopping the connection stops the whole tree. + ide_core::process::own_group(cmd.as_std_mut()); cmd.args(args) .current_dir(cwd) .envs(ide_core::process::shell_env()) @@ -422,6 +425,9 @@ impl Peer { let (status, killed) = tokio::select! { status = child.wait() => (status.ok(), false), _ = inner.kill.cancelled() => { + if let Some(pid) = child.id() { + let _ = tokio::task::spawn_blocking(move || ide_core::process::kill_tree_of(pid)).await; + } let _ = child.start_kill(); (child.wait().await.ok(), true) } diff --git a/src/features/agents/HomeEditor.tsx b/src/features/agents/HomeEditor.tsx index e12ef1e29..e909aa5e4 100644 --- a/src/features/agents/HomeEditor.tsx +++ b/src/features/agents/HomeEditor.tsx @@ -165,7 +165,7 @@ export function HomeEditor({ docked = false }: { docked?: boolean; input?: Edito detail: tx("edits your files directly"), }, ]} - onChange={(isolation) => updateDraft({ isolation })} + onChange={(isolation) => updateDraft({ isolation, isolationChosen: true })} /> )} {isolation === "local" && current && ( diff --git a/src/features/agents/draft.test.ts b/src/features/agents/draft.test.ts new file mode 100644 index 000000000..1553504c2 --- /dev/null +++ b/src/features/agents/draft.test.ts @@ -0,0 +1,13 @@ +import { expect, it } from "vitest"; +import { savedDraft } from "./store"; + +it("runs tasks in the local checkout unless the user picked a worktree", () => { + expect(savedDraft({}).isolation).toBe("local"); + // Saved before the choice was recorded: `worktree` was only the old default. + expect(savedDraft({ provider: "codex", isolation: "worktree" })).toMatchObject({ + provider: "codex", + isolation: "local", + }); + expect(savedDraft({ isolation: "worktree", isolationChosen: true }).isolation).toBe("worktree"); + expect(savedDraft({ isolation: "local", isolationChosen: true }).isolation).toBe("local"); +}); diff --git a/src/features/agents/store.ts b/src/features/agents/store.ts index f22a72237..4bb32f11f 100644 --- a/src/features/agents/store.ts +++ b/src/features/agents/store.ts @@ -38,6 +38,8 @@ export interface Draft { /** Settings chosen for the next task, by option id (reset when the agent changes). */ config: Record; isolation: Isolation; + /** The user picked `isolation`: otherwise a task runs in the local checkout. */ + isolationChosen?: boolean; /** `null`: the current branch. */ base: string | null; } @@ -99,7 +101,16 @@ function save(key: string, value: unknown) { } } -const fallbackDraft: Draft = { provider: "claude", config: {}, isolation: "worktree", base: null }; +const fallbackDraft: Draft = { provider: "claude", config: {}, isolation: "local", base: null }; + +/** + * The saved New Task settings. A worktree is used only when the user picked it: drafts + * saved before that choice was recorded carried the old default (`worktree`) instead. + */ +export function savedDraft(saved: Partial): Draft { + const draft = { ...fallbackDraft, ...saved }; + return draft.isolationChosen ? draft : { ...draft, isolation: "local" }; +} export const useAgents = create(() => ({ loaded: false, @@ -110,7 +121,7 @@ export const useAgents = create(() => ({ changes: new Map(), providers: [], limits: new Map(), - draft: { ...fallbackDraft, ...load>(DRAFT_KEY, {}) }, + draft: savedDraft(load>(DRAFT_KEY, {})), composers: new Map(), lastComposer: HOME, history: load(HISTORY_KEY, []),