diff --git a/crates/acp_thread/src/terminal.rs b/crates/acp_thread/src/terminal.rs index f70e044fbc1b38..76582aa2ae053e 100644 --- a/crates/acp_thread/src/terminal.rs +++ b/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, output_byte_limit: Option, _output_task: Shared>, + /// 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, } 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); diff --git a/crates/agent/src/agent.rs b/crates/agent/src/agent.rs index 612360fe887e11..cbab898871b3e7 100644 --- a/crates/agent/src/agent.rs +++ b/crates/agent/src/agent.rs @@ -1492,6 +1492,11 @@ impl TerminalHandle for AcpTerminalHandle { })?; Ok(()) } + + fn was_stopped_by_user(&self, cx: &AsyncApp) -> Result { + self.terminal + .read_with(cx, |term, _cx| term.was_stopped_by_user()) + } } #[cfg(test)] diff --git a/crates/agent/src/tests/mod.rs b/crates/agent/src/tests/mod.rs index 45028902e467fe..4a12b47a0e1df0 100644 --- a/crates/agent/src/tests/mod.rs +++ b/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 { + Ok(false) + } } struct FakeThreadEnvironment { diff --git a/crates/agent/src/thread.rs b/crates/agent/src/thread.rs index 95b0c9bcb5d0f7..9c70cad9665652 100644 --- a/crates/agent/src/thread.rs +++ b/crates/agent/src/thread.rs @@ -538,6 +538,7 @@ pub trait TerminalHandle { fn current_output(&self, cx: &AsyncApp) -> Result; fn wait_for_exit(&self, cx: &AsyncApp) -> Result>>; fn kill(&self, cx: &AsyncApp) -> Result<()>; + fn was_stopped_by_user(&self, cx: &AsyncApp) -> Result; } 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, model: Arc, event_stream: &ThreadEventStream, + cancellation_rx: watch::Receiver, 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, cx: &mut Context, ) -> Result>> { 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, cx: &mut Context, ) -> Option> { 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>, + /// 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, } 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>, + cancellation_rx: watch::Receiver, } impl ToolCallEventStream { #[cfg(any(test, feature = "test-support"))] pub fn test() -> (Self, ToolCallEventStreamReceiver) { let (events_tx, events_rx) = mpsc::unbounded::>(); + 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>, + cancellation_rx: watch::Receiver, ) -> 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 + '_ { + 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); diff --git a/crates/agent/src/tools/terminal_tool.rs b/crates/agent/src/tools/terminal_tool.rs index f3302fb1894612..d153c047f171c2 100644 --- a/crates/agent/src/tools/terminal_tool.rs +++ b/crates/agent/src/tools/terminal_tool.rs @@ -124,27 +124,44 @@ impl AgentTool for TerminalTool { let timeout = input.timeout_ms.map(Duration::from_millis); - let exit_status = match timeout { + let mut timed_out = false; + let wait_for_exit = terminal.wait_for_exit(cx)?; + + match timeout { Some(timeout) => { - let wait_for_exit = terminal.wait_for_exit(cx)?; let timeout_task = cx.background_spawn(async move { smol::Timer::after(timeout).await; }); futures::select! { - status = wait_for_exit.clone().fuse() => status, + _ = wait_for_exit.clone().fuse() => {}, _ = timeout_task.fuse() => { + timed_out = true; terminal.kill(cx)?; - wait_for_exit.await + wait_for_exit.await; } } } - None => terminal.wait_for_exit(cx)?.await, + None => { + wait_for_exit.await; + } }; + // Check if user stopped - we check both: + // 1. The cancellation signal from RunningTurn::cancel (e.g. user pressed main Stop button) + // 2. The terminal's user_stopped flag (e.g. user clicked Stop on the terminal card) + let user_stopped_via_signal = event_stream.was_cancelled_by_user(); + let user_stopped_via_terminal = terminal.was_stopped_by_user(cx).unwrap_or(false); + let user_stopped = user_stopped_via_signal || user_stopped_via_terminal; + let output = terminal.current_output(cx)?; - Ok(process_content(output, &input.command, exit_status)) + Ok(process_content( + output, + &input.command, + timed_out, + user_stopped, + )) }) } } @@ -152,7 +169,8 @@ impl AgentTool for TerminalTool { fn process_content( output: acp::TerminalOutputResponse, command: &str, - exit_status: acp::TerminalExitStatus, + timed_out: bool, + user_stopped: bool, ) -> String { let content = output.output.trim(); let is_empty = content.is_empty(); @@ -167,30 +185,59 @@ fn process_content( content }; - let content = match exit_status.exit_code { - Some(0) => { - if is_empty { - "Command executed successfully.".to_string() - } else { + let content = if user_stopped { + if is_empty { + "The user stopped this command. No output was captured before stopping.\n\n\ + Since the user intentionally interrupted this command, ask them what they would like to do next \ + rather than automatically retrying or assuming something went wrong.".to_string() + } else { + format!( + "The user stopped this command. Output captured before stopping:\n\n{}\n\n\ + Since the user intentionally interrupted this command, ask them what they would like to do next \ + rather than automatically retrying or assuming something went wrong.", content - } - } - Some(exit_code) => { - if is_empty { - format!("Command \"{command}\" failed with exit code {}.", exit_code) - } else { - format!( - "Command \"{command}\" failed with exit code {}.\n\n{content}", - exit_code - ) - } + ) } - None => { + } else if timed_out { + if is_empty { + format!("Command \"{command}\" timed out. No output was captured.") + } else { format!( - "Command failed or was interrupted.\nPartial output captured:\n\n{}", - content, + "Command \"{command}\" timed out. Output captured before timeout:\n\n{}", + content ) } + } else { + let exit_code = output.exit_status.as_ref().and_then(|s| s.exit_code); + match exit_code { + Some(0) => { + if is_empty { + "Command executed successfully.".to_string() + } else { + content + } + } + Some(exit_code) => { + if is_empty { + format!("Command \"{command}\" failed with exit code {}.", exit_code) + } else { + format!( + "Command \"{command}\" failed with exit code {}.\n\n{content}", + exit_code + ) + } + } + None => { + if is_empty { + "Command terminated unexpectedly. No output was captured.".to_string() + } else { + format!( + "Command terminated unexpectedly. Output captured:\n\n{}", + content + ) + } + } + } }; content } @@ -235,3 +282,187 @@ fn working_dir( anyhow::bail!("`cd` directory {cd:?} was not in any of the project's worktrees."); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_process_content_user_stopped() { + let output = acp::TerminalOutputResponse::new("partial output".to_string(), false); + + let result = process_content(output, "cargo build", false, true); + + assert!( + result.contains("user stopped"), + "Expected 'user stopped' message, got: {}", + result + ); + assert!( + result.contains("partial output"), + "Expected output to be included, got: {}", + result + ); + assert!( + result.contains("ask them what they would like to do"), + "Should instruct agent to ask user, got: {}", + result + ); + } + + #[test] + fn test_process_content_user_stopped_empty_output() { + let output = acp::TerminalOutputResponse::new("".to_string(), false); + + let result = process_content(output, "cargo build", false, true); + + assert!( + result.contains("user stopped"), + "Expected 'user stopped' message, got: {}", + result + ); + assert!( + result.contains("No output was captured"), + "Expected 'No output was captured', got: {}", + result + ); + } + + #[test] + fn test_process_content_timed_out() { + let output = acp::TerminalOutputResponse::new("build output here".to_string(), false); + + let result = process_content(output, "cargo build", true, false); + + assert!( + result.contains("timed out"), + "Expected 'timed out' message for timeout, got: {}", + result + ); + assert!( + result.contains("build output here"), + "Expected output to be included, got: {}", + result + ); + } + + #[test] + fn test_process_content_timed_out_with_empty_output() { + let output = acp::TerminalOutputResponse::new("".to_string(), false); + + let result = process_content(output, "sleep 1000", true, false); + + assert!( + result.contains("timed out"), + "Expected 'timed out' for timeout, got: {}", + result + ); + assert!( + result.contains("No output was captured"), + "Expected 'No output was captured' for empty output, got: {}", + result + ); + } + + #[test] + fn test_process_content_with_success() { + let output = acp::TerminalOutputResponse::new("success output".to_string(), false) + .exit_status(acp::TerminalExitStatus::new().exit_code(0)); + + let result = process_content(output, "echo hello", false, false); + + assert!( + result.contains("success output"), + "Expected output to be included, got: {}", + result + ); + assert!( + !result.contains("failed"), + "Success should not say 'failed', got: {}", + result + ); + } + + #[test] + fn test_process_content_with_success_empty_output() { + let output = acp::TerminalOutputResponse::new("".to_string(), false) + .exit_status(acp::TerminalExitStatus::new().exit_code(0)); + + let result = process_content(output, "true", false, false); + + assert!( + result.contains("executed successfully"), + "Expected success message for empty output, got: {}", + result + ); + } + + #[test] + fn test_process_content_with_error_exit() { + let output = acp::TerminalOutputResponse::new("error output".to_string(), false) + .exit_status(acp::TerminalExitStatus::new().exit_code(1)); + + let result = process_content(output, "false", false, false); + + assert!( + result.contains("failed with exit code 1"), + "Expected failure message, got: {}", + result + ); + assert!( + result.contains("error output"), + "Expected output to be included, got: {}", + result + ); + } + + #[test] + fn test_process_content_with_error_exit_empty_output() { + let output = acp::TerminalOutputResponse::new("".to_string(), false) + .exit_status(acp::TerminalExitStatus::new().exit_code(1)); + + let result = process_content(output, "false", false, false); + + assert!( + result.contains("failed with exit code 1"), + "Expected failure message, got: {}", + result + ); + } + + #[test] + fn test_process_content_unexpected_termination() { + let output = acp::TerminalOutputResponse::new("some output".to_string(), false); + + let result = process_content(output, "some_command", false, false); + + assert!( + result.contains("terminated unexpectedly"), + "Expected 'terminated unexpectedly' message, got: {}", + result + ); + assert!( + result.contains("some output"), + "Expected output to be included, got: {}", + result + ); + } + + #[test] + fn test_process_content_unexpected_termination_empty_output() { + let output = acp::TerminalOutputResponse::new("".to_string(), false); + + let result = process_content(output, "some_command", false, false); + + assert!( + result.contains("terminated unexpectedly"), + "Expected 'terminated unexpectedly' message, got: {}", + result + ); + assert!( + result.contains("No output was captured"), + "Expected 'No output was captured' for empty output, got: {}", + result + ); + } +} diff --git a/crates/agent_ui/src/acp/thread_view.rs b/crates/agent_ui/src/acp/thread_view.rs index b6f560b0de515d..94ef65503c7258 100644 --- a/crates/agent_ui/src/acp/thread_view.rs +++ b/crates/agent_ui/src/acp/thread_view.rs @@ -3707,9 +3707,8 @@ impl AcpThreadView { .on_click({ let terminal = terminal.clone(); cx.listener(move |_this, _event, _window, cx| { - let inner_terminal = terminal.read(cx).inner().clone(); - inner_terminal.update(cx, |inner_terminal, _cx| { - inner_terminal.kill_active_task(); + terminal.update(cx, |terminal, cx| { + terminal.stop_by_user(cx); }); }) }), diff --git a/crates/eval/src/instance.rs b/crates/eval/src/instance.rs index 8c9da3eefab61e..c6affb557d0f16 100644 --- a/crates/eval/src/instance.rs +++ b/crates/eval/src/instance.rs @@ -635,6 +635,11 @@ impl agent::TerminalHandle for EvalTerminalHandle { })?; Ok(()) } + + fn was_stopped_by_user(&self, cx: &AsyncApp) -> Result { + self.terminal + .read_with(cx, |term, _cx| term.was_stopped_by_user()) + } } impl agent::ThreadEnvironment for EvalThreadEnvironment { diff --git a/crates/terminal/src/terminal.rs b/crates/terminal/src/terminal.rs index ca556cf78ee831..279626f197c114 100644 --- a/crates/terminal/src/terminal.rs +++ b/crates/terminal/src/terminal.rs @@ -2146,7 +2146,11 @@ impl Terminal { && task.status == TaskStatus::Running { if let TerminalType::Pty { info, .. } = &mut self.terminal_type { + // First kill the foreground process group (the command running in the shell) info.kill_current_process(); + // Then kill the shell itself so that the terminal exits properly + // and wait_for_completed_task can complete + info.kill_child_process(); } } } @@ -2487,7 +2491,48 @@ mod tests { Point, TestAppContext, bounds, point, size, smol_timeout, }; use rand::{Rng, distr, rngs::ThreadRng}; - use task::ShellBuilder; + use smol::channel::Receiver; + use task::{Shell, ShellBuilder}; + + /// Helper to build a test terminal running a shell command. + /// Returns the terminal entity and a receiver for the completion signal. + async fn build_test_terminal( + cx: &mut TestAppContext, + command: &str, + args: &[&str], + ) -> (Entity, Receiver>) { + let (completion_tx, completion_rx) = smol::channel::unbounded(); + let args: Vec = args.iter().map(|s| s.to_string()).collect(); + let (program, args) = + ShellBuilder::new(&Shell::System, false).build(Some(command.to_owned()), &args); + let builder = cx + .update(|cx| { + TerminalBuilder::new( + None, + None, + task::Shell::WithArguments { + program, + args, + title_override: None, + }, + HashMap::default(), + CursorShape::default(), + AlternateScroll::On, + None, + vec![], + 0, + false, + 0, + Some(completion_tx), + cx, + vec![], + ) + }) + .await + .unwrap(); + let terminal = cx.new(|cx| builder.subscribe(cx)); + (terminal, completion_rx) + } fn init_ctrl_click_hyperlink_test(cx: &mut TestAppContext, output: &[u8]) -> Entity { cx.update(|cx| { @@ -2571,35 +2616,7 @@ mod tests { async fn test_basic_terminal(cx: &mut TestAppContext) { cx.executor().allow_parking(); - let (completion_tx, completion_rx) = smol::channel::unbounded(); - let (program, args) = ShellBuilder::new(&Shell::System, false) - .build(Some("echo".to_owned()), &["hello".to_owned()]); - let builder = cx - .update(|cx| { - TerminalBuilder::new( - None, - None, - task::Shell::WithArguments { - program, - args, - title_override: None, - }, - HashMap::default(), - CursorShape::default(), - AlternateScroll::On, - None, - vec![], - 0, - false, - 0, - Some(completion_tx), - cx, - vec![], - ) - }) - .await - .unwrap(); - let terminal = cx.new(|cx| builder.subscribe(cx)); + let (terminal, completion_rx) = build_test_terminal(cx, "echo", &["hello"]).await; assert_eq!( completion_rx.recv().await.unwrap(), Some(ExitStatus::default()) @@ -3083,6 +3100,76 @@ mod tests { }); } + /// Test that kill_active_task properly terminates both the foreground process + /// and the shell, allowing wait_for_completed_task to complete and output to be captured. + #[cfg(unix)] + #[gpui::test] + async fn test_kill_active_task_completes_and_captures_output(cx: &mut TestAppContext) { + cx.executor().allow_parking(); + + // Run a command that prints output then sleeps for a long time + // The echo ensures we have output to capture before killing + let (terminal, completion_rx) = + build_test_terminal(cx, "echo", &["test_output_before_kill; sleep 60"]).await; + + // Wait a bit for the echo to execute and produce output + smol::Timer::after(Duration::from_millis(200)).await; + + // Kill the active task + terminal.update(cx, |term, _cx| { + term.kill_active_task(); + }); + + // wait_for_completed_task should complete within a reasonable time (not hang) + let completion_result = smol_timeout(Duration::from_secs(5), completion_rx.recv()).await; + assert!( + completion_result.is_ok(), + "wait_for_completed_task should complete after kill_active_task, but it timed out" + ); + + // The exit status should indicate the process was killed (not a clean exit) + let exit_status = completion_result.unwrap().unwrap(); + assert!( + exit_status.is_some(), + "Should have received an exit status after killing" + ); + + // Verify that output captured before killing is still available + let content = terminal.update(cx, |term, _| term.get_content()); + assert!( + content.contains("test_output_before_kill"), + "Output from before kill should be captured, got: {content}" + ); + } + + /// Test that kill_active_task on a task that's not running is a no-op + #[gpui::test] + async fn test_kill_active_task_on_completed_task_is_noop(cx: &mut TestAppContext) { + cx.executor().allow_parking(); + + // Run a command that exits immediately + let (terminal, completion_rx) = build_test_terminal(cx, "echo", &["done"]).await; + + // Wait for the command to complete naturally + let exit_status = smol_timeout(Duration::from_secs(5), completion_rx.recv()) + .await + .expect("Command should complete") + .expect("Should receive exit status"); + assert_eq!(exit_status, Some(ExitStatus::default())); + + // Now try to kill - should be a no-op since task already completed + terminal.update(cx, |term, _cx| { + term.kill_active_task(); + }); + + // Content should still be there + let content = terminal.update(cx, |term, _| term.get_content()); + assert!( + content.contains("done"), + "Output should still be present after no-op kill, got: {content}" + ); + } + mod perf { use super::super::*; use gpui::{