fix(agents): stop Claude Code processes when their connection closes

Each agent task talks to an in-process shim that runs one `claude -p` per
session. Closing the connection (an idle task released after five minutes,
archived or deleted) stopped those processes from a task that upgraded a weak
reference to the shim; the connection drops the shim as it closes, so the
upgrade could fail and the sessions went without stopping their processes,
which then ran until the app quit. The task now holds the shim until the
connection is closed, and a session dropped without being stopped stops its
process.

Undoing turns and copying a conversation also start a stopped agent without a
turn; they now schedule the same idle release.

crates/ide-agent/src/lib.rs

@@ -1097,6 +1097,8 @@ impl AgentManager {
                 title_renamed: s.title_renamed,
             }
         };
+        // Copying may have started the source's agent: it stops again once idle.
+        source.release_when_idle();
         let task = Task::restore(file, self.shared.clone());
         self.tasks.write().insert(task.id.clone(), task.clone());
         task.persist();
@@ -1166,6 +1168,8 @@ impl AgentManager {
         task.persist();
         self.shared.emit(AgentEvent::Truncated { task: id.to_owned(), length: cut as u32 });
         task.emit_task();
+        // Undoing may have started the agent: it stops again once idle.
+        task.release_when_idle();
         Ok(first)
     }
 

crates/ide-agent/src/native/claude/mod.rs

@@ -583,11 +583,12 @@ impl Shim {
 impl super::Shim for Shim {
     fn attach(&self, client: Peer) {
         let _ = self.client.set(client.clone());
-        // The client went away: so do the processes.
-        let weak = self.me.clone();
+        // The client went away: so do the processes. Held until then: the connection drops
+        // its handler (this shim) as it closes, and the sessions would go without stopping.
+        let me = self.me.upgrade();
         tokio::spawn(async move {
             client.closed().await;
-            if let Some(me) = weak.upgrade() {
+            if let Some(me) = me {
                 me.closed.store(true, Ordering::Release);
                 let sessions: Vec<Arc<Session>> = me.sessions.lock().drain().map(|(_, s)| s).collect();
                 for s in sessions {
@@ -598,6 +599,15 @@ impl super::Shim for Shim {
     }
 }
 
+impl Drop for Session {
+    /// A session dropped without being stopped (the shim went away) takes its process along.
+    fn drop(&mut self) {
+        if let Some(p) = self.st.get_mut().proc.take() {
+            p.kill.cancel();
+        }
+    }
+}
+
 impl Session {
     fn shim(&self) -> Option<Arc<Shim>> {
         self.shim.upgrade()

crates/ide-agent/src/native/claude/tests.rs

@@ -311,6 +311,8 @@ struct Replay {
     sent: Arc<Mutex<Vec<Vec<Value>>>>,
     launched: Mutex<Vec<Vec<String>>>,
     errors: Arc<Mutex<Vec<String>>>,
+    /// Each process's kill token.
+    kills: Mutex<Vec<CancellationToken>>,
 }
 
 impl Cli for Replay {
@@ -331,6 +333,7 @@ impl Cli for Replay {
         let (shim_in, cli_in) = tokio::io::duplex(1 << 20);
         let (mut cli_out, shim_out) = tokio::io::duplex(1 << 20);
         let kill = CancellationToken::new();
+        self.kills.lock().push(kill.clone());
         let (sent, errors, token) = (self.sent.clone(), self.errors.clone(), kill.clone());
         sent.lock().push(Vec::new());
         tokio::spawn(async move {
@@ -428,6 +431,7 @@ async fn replay(name: &str) -> Replayed {
         sent: Arc::default(),
         launched: Mutex::default(),
         errors: Arc::default(),
+        kills: Mutex::default(),
     });
     let (config, cwd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
     let cwd_path = std::fs::canonicalize(cwd.path()).unwrap();
@@ -461,6 +465,23 @@ async fn replay_simple() {
     assert_eq!(user["message"]["content"][0]["text"], "Reply with exactly: hello there");
 }
 
+/// The app closing the connection (an idle task released, archived or deleted) stops the
+/// CLI processes of its sessions, idle ones included.
+#[tokio::test(flavor = "multi_thread")]
+async fn closing_the_connection_stops_the_processes() {
+    let r = replay("simple").await;
+    assert!(r.cli.kills.lock().iter().all(|k| !k.is_cancelled()));
+    r.h.client.kill();
+    let kills = r.cli.kills.lock().clone();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        for k in kills {
+            k.cancelled().await;
+        }
+    })
+    .await
+    .expect("every process was stopped");
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn replay_multi_turn() {
     let r = replay("multi_turn").await;
@@ -727,6 +748,7 @@ async fn a_stop_ahead_of_its_prompt_cancels_it() {
         sent: Arc::default(),
         launched: Mutex::default(),
         errors: Arc::default(),
+        kills: Mutex::default(),
     });
     let (dir, cwd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
     let h = Harness::connect(cli.clone(), dir.path().into(), dir.path().join("p"), cwd.path().into(), Vec::new()).await;
@@ -770,6 +792,7 @@ async fn a_failed_turn_reports_its_failure() {
             sent: Arc::default(),
             launched: Mutex::default(),
             errors: Arc::default(),
+            kills: Mutex::default(),
         });
         let (dir, cwd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
         let h =

crates/ide-agent/src/tests.rs

@@ -2391,6 +2391,27 @@ async fn idle_agents_stop_and_resume_their_session_on_the_next_message() {
     assert!(live(&mgr, &task.id));
     mgr.shutdown();
 
+    // Undoing turns or copying the conversation starts a stopped agent: it stops again too.
+    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;
+    mgr.prompt(&task.id, text("again"), false).await.unwrap();
+    wait_for(&mgr, &task.id, AgentStatus::Ready).await;
+    tokio::time::sleep(Duration::from_millis(600)).await;
+    assert!(!live(&mgr, &task.id));
+    mgr.rollback(&task.id, 1).await.unwrap();
+    assert!(live(&mgr, &task.id));
+    tokio::time::sleep(Duration::from_millis(600)).await;
+    assert!(!live(&mgr, &task.id), "an agent started to undo turns stops too");
+    mgr.fork(&task.id, None).await.unwrap();
+    assert!(live(&mgr, &task.id));
+    tokio::time::sleep(Duration::from_millis(600)).await;
+    assert!(!live(&mgr, &task.id), "an agent started to copy the conversation stops too");
+    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() });