Let agent see output of killed terminal tools

crates/acp_thread/src/terminal.rs

@@ -5,7 +5,15 @@ use gpui::{App, AppContext, AsyncApp, Context, Entity, Task};
 use language::LanguageRegistry;
 use markdown::Markdown;
 use project::Project;
-use std::{path::PathBuf, process::ExitStatus, sync::Arc, time::Instant};
+use std::{
+    path::PathBuf,
+    process::ExitStatus,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::Instant,
+};
 use task::Shell;
 use util::get_default_system_shell_preferring_bash;
 
@@ -18,6 +26,10 @@ pub struct Terminal {
     output: Option<TerminalOutput>,
     output_byte_limit: Option<usize>,
     _output_task: Shared<Task<acp::TerminalExitStatus>>,
+    /// Flag indicating whether this terminal was stopped by explicit user action
+    /// (e.g., clicking the Stop button). This is set before kill() is called
+    /// so that code awaiting wait_for_exit() can check it deterministically.
+    user_stopped: Arc<AtomicBool>,
 }
 
 pub struct TerminalOutput {
@@ -54,6 +66,7 @@ impl Terminal {
             started_at: Instant::now(),
             output: None,
             output_byte_limit,
+            user_stopped: Arc::new(AtomicBool::new(false)),
             _output_task: cx
                 .spawn(async move |this, cx| {
                     let exit_status = command_task.await;
@@ -97,6 +110,18 @@ impl Terminal {
         });
     }
 
+    /// Marks this terminal as stopped by user action and then kills it.
+    /// This should be called when the user explicitly clicks a Stop button.
+    pub fn stop_by_user(&mut self, cx: &mut App) {
+        self.user_stopped.store(true, Ordering::SeqCst);
+        self.kill(cx);
+    }
+
+    /// Returns whether this terminal was stopped by explicit user action.
+    pub fn was_stopped_by_user(&self) -> bool {
+        self.user_stopped.load(Ordering::SeqCst)
+    }
+
     pub fn current_output(&self, cx: &App) -> acp::TerminalOutputResponse {
         if let Some(output) = self.output.as_ref() {
             let exit_status = output.exit_status.map(portable_pty::ExitStatus::from);

crates/agent/src/agent.rs

@@ -1492,6 +1492,11 @@ impl TerminalHandle for AcpTerminalHandle {
         })?;
         Ok(())
     }
+
+    fn was_stopped_by_user(&self, cx: &AsyncApp) -> Result<bool> {
+        self.terminal
+            .read_with(cx, |term, _cx| term.was_stopped_by_user())
+    }
 }
 
 #[cfg(test)]

crates/agent/src/tests/mod.rs

@@ -114,6 +114,10 @@ impl crate::TerminalHandle for FakeTerminalHandle {
         self.killed.store(true, Ordering::SeqCst);
         Ok(())
     }
+
+    fn was_stopped_by_user(&self, _cx: &AsyncApp) -> Result<bool> {
+        Ok(false)
+    }
 }
 
 struct FakeThreadEnvironment {

crates/agent/src/thread.rs

@@ -538,6 +538,7 @@ pub trait TerminalHandle {
     fn current_output(&self, cx: &AsyncApp) -> Result<acp::TerminalOutputResponse>;
     fn wait_for_exit(&self, cx: &AsyncApp) -> Result<Shared<Task<acp::TerminalExitStatus>>>;
     fn kill(&self, cx: &AsyncApp) -> Result<()>;
+    fn was_stopped_by_user(&self, cx: &AsyncApp) -> Result<bool>;
 }
 
 pub trait ThreadEnvironment {
@@ -773,10 +774,13 @@ impl Thread {
             .as_ref()
             .and_then(|result| result.output.clone());
         if let Some(output) = output.clone() {
+            // For replay, we use a dummy cancellation receiver since the tool already completed
+            let (_cancellation_tx, cancellation_rx) = watch::channel(false);
             let tool_event_stream = ToolCallEventStream::new(
                 tool_use.id.clone(),
                 stream.clone(),
                 Some(self.project.read(cx).fs().clone()),
+                cancellation_rx,
             );
             tool.replay(tool_use.input.clone(), output, tool_event_stream, cx)
                 .log_err();
@@ -1263,13 +1267,16 @@ impl Thread {
         let message_ix = self.messages.len().saturating_sub(1);
         self.tool_use_limit_reached = false;
         self.clear_summary();
+        let (cancellation_tx, cancellation_rx) = watch::channel(false);
         self.running_turn = Some(RunningTurn {
             event_stream: event_stream.clone(),
             tools: self.enabled_tools(profile, &model, cx),
+            cancellation_tx,
             _task: cx.spawn(async move |this, cx| {
                 log::debug!("Starting agent turn execution");
 
-                let turn_result = Self::run_turn_internal(&this, model, &event_stream, cx).await;
+                let turn_result =
+                    Self::run_turn_internal(&this, model, &event_stream, cancellation_rx, cx).await;
                 _ = this.update(cx, |this, cx| this.flush_pending_message(cx));
 
                 match turn_result {
@@ -1304,6 +1311,7 @@ impl Thread {
         this: &WeakEntity<Self>,
         model: Arc<dyn LanguageModel>,
         event_stream: &ThreadEventStream,
+        cancellation_rx: watch::Receiver<bool>,
         cx: &mut AsyncApp,
     ) -> Result<()> {
         let mut attempt = 0;
@@ -1333,7 +1341,12 @@ impl Thread {
                 match event {
                     Ok(event) => {
                         tool_results.extend(this.update(cx, |this, cx| {
-                            this.handle_completion_event(event, event_stream, cx)
+                            this.handle_completion_event(
+                                event,
+                                event_stream,
+                                cancellation_rx.clone(),
+                                cx,
+                            )
                         })??);
                     }
                     Err(err) => {
@@ -1461,6 +1474,7 @@ impl Thread {
         &mut self,
         event: LanguageModelCompletionEvent,
         event_stream: &ThreadEventStream,
+        cancellation_rx: watch::Receiver<bool>,
         cx: &mut Context<Self>,
     ) -> Result<Option<Task<LanguageModelToolResult>>> {
         log::trace!("Handling streamed completion event: {:?}", event);
@@ -1489,7 +1503,7 @@ impl Thread {
                 }
             }
             ToolUse(tool_use) => {
-                return Ok(self.handle_tool_use_event(tool_use, event_stream, cx));
+                return Ok(self.handle_tool_use_event(tool_use, event_stream, cancellation_rx, cx));
             }
             ToolUseJsonParseError {
                 id,
@@ -1592,6 +1606,7 @@ impl Thread {
         &mut self,
         tool_use: LanguageModelToolUse,
         event_stream: &ThreadEventStream,
+        cancellation_rx: watch::Receiver<bool>,
         cx: &mut Context<Self>,
     ) -> Option<Task<LanguageModelToolResult>> {
         cx.notify();
@@ -1656,8 +1671,12 @@ impl Thread {
         };
 
         let fs = self.project.read(cx).fs().clone();
-        let tool_event_stream =
-            ToolCallEventStream::new(tool_use.id.clone(), event_stream.clone(), Some(fs));
+        let tool_event_stream = ToolCallEventStream::new(
+            tool_use.id.clone(),
+            event_stream.clone(),
+            Some(fs),
+            cancellation_rx,
+        );
         tool_event_stream.update_fields(
             acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::InProgress),
         );
@@ -2239,11 +2258,15 @@ struct RunningTurn {
     event_stream: ThreadEventStream,
     /// The tools that were enabled for this turn.
     tools: BTreeMap<SharedString, Arc<dyn AnyAgentTool>>,
+    /// Sender to signal tool cancellation. When cancel is called, this is
+    /// set to true so all tools can detect user-initiated cancellation.
+    cancellation_tx: watch::Sender<bool>,
 }
 
 impl RunningTurn {
-    fn cancel(self) {
+    fn cancel(mut self) {
         log::debug!("Cancelling in progress turn");
+        self.cancellation_tx.send(true).ok();
         self.event_stream.send_canceled();
     }
 }
@@ -2506,14 +2529,21 @@ pub struct ToolCallEventStream {
     tool_use_id: LanguageModelToolUseId,
     stream: ThreadEventStream,
     fs: Option<Arc<dyn Fs>>,
+    cancellation_rx: watch::Receiver<bool>,
 }
 
 impl ToolCallEventStream {
     #[cfg(any(test, feature = "test-support"))]
     pub fn test() -> (Self, ToolCallEventStreamReceiver) {
         let (events_tx, events_rx) = mpsc::unbounded::<Result<ThreadEvent>>();
+        let (_cancellation_tx, cancellation_rx) = watch::channel(false);
 
-        let stream = ToolCallEventStream::new("test_id".into(), ThreadEventStream(events_tx), None);
+        let stream = ToolCallEventStream::new(
+            "test_id".into(),
+            ThreadEventStream(events_tx),
+            None,
+            cancellation_rx,
+        );
 
         (stream, ToolCallEventStreamReceiver(events_rx))
     }
@@ -2522,14 +2552,40 @@ impl ToolCallEventStream {
         tool_use_id: LanguageModelToolUseId,
         stream: ThreadEventStream,
         fs: Option<Arc<dyn Fs>>,
+        cancellation_rx: watch::Receiver<bool>,
     ) -> Self {
         Self {
             tool_use_id,
             stream,
             fs,
+            cancellation_rx,
         }
     }
 
+    /// Returns a future that resolves when the user cancels the tool call.
+    /// Tools should select on this alongside their main work to detect user cancellation.
+    pub fn cancelled_by_user(&self) -> impl std::future::Future<Output = ()> + '_ {
+        let mut rx = self.cancellation_rx.clone();
+        async move {
+            loop {
+                if *rx.borrow() {
+                    return;
+                }
+                if rx.changed().await.is_err() {
+                    // Sender dropped, will never be cancelled
+                    std::future::pending::<()>().await;
+                }
+            }
+        }
+    }
+
+    /// Returns true if the user has cancelled this tool call.
+    /// This is useful for checking cancellation state after an operation completes,
+    /// to determine if the completion was due to user cancellation.
+    pub fn was_cancelled_by_user(&self) -> bool {
+        *self.cancellation_rx.clone().borrow()
+    }
+
     pub fn update_fields(&self, fields: acp::ToolCallUpdateFields) {
         self.stream
             .update_tool_call_fields(&self.tool_use_id, fields);