From e20eb1acf8d1fafaaecb6af5b2568688b9614898 Mon Sep 17 00:00:00 2001 From: l0ng-ai <24760907+l0ng-ai@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:38:33 +0800 Subject: [PATCH] fix(agents): stop idle agents, run tasks locally by default, keep worktrees next to the project (#1050) A task's agent process lived until the task was archived or the project closed, and stopping it killed only the direct child, so wrappers (npx, codex.cmd, claude shims) left the real agent running. Agents that support session/load are now stopped after 5 minutes idle and restarted with the session reloaded on the next message; agent connections, language servers and the native Claude CLI start in their own process group and are stopped with their whole tree. New tasks run in the local checkout unless the user picks a worktree, and worktrees go next to the project (/.worktrees/), on the project's drive and outside its tree. Fixes #1011 Fixes #1003 --- crates/ide-agent/src/lib.rs | 53 +++++++++++++++++++-- crates/ide-agent/src/native/claude/cli.rs | 5 ++ crates/ide-agent/src/tests.rs | 57 ++++++++++++++++++++++- crates/ide-core/src/process.rs | 47 +++++++++++++++++++ crates/ide-host/src/project.rs | 42 ++++++++++++++--- crates/ide-jsonrpc/src/lib.rs | 6 +++ src/features/agents/HomeEditor.tsx | 2 +- src/features/agents/draft.test.ts | 13 ++++++ src/features/agents/store.ts | 15 +++++- 9 files changed, 227 insertions(+), 13 deletions(-) create mode 100644 src/features/agents/draft.test.ts 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, []),