vrr/zed
2
0
Fork 0
mirror of https://github.com/zed-industries/zed.git synced 2026-07-10 00:13:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 18

agent_ui: Refactor the queue feature and add steering ability (#59310)

Browse source 
Closes AI-367

This PR significantly refactors the queue feature so that we don't rely
on indexes being counted and synced across various state points of the
thread view. Instead, I'm introducing a dedicated message queue module
where we control through a stable queue entry ID all the states of a
given queued message.

Additionally, it's relevant to note that I changed the default behavior
of queued messages on this PR so that they get sent only at the end of
the generation. This makes the out-of-the-box, default behavior of the
feature consistent across the native and external agents, given we never
could pull off the feature with the latter. I personally never liked
that behavior anyway, but I do appreciate how some people do. Because of
that, though, for the Zed agent, a "Steer" toggle button is available
for whenever you'd like to signal that a queued message should be sent
at the turn boundary. This is not a setting, though; decided to go
simpler at this moment. It's something you can choose on a per-message
basis.

Ultimately, I think this implementation makes the overall shape of the
feature much more stable and easier to maintain overtime.

Release Notes:

- Agent: Changed the default behavior of queued messages for the Zed
agent so that they get send at the end of the generation. The ability to
steer them (i.e., send the message at the end of a turn boundary) is
still possible to be toggled.
This commit is contained in:
Danilo Leal 2026-06-26 08:45:58 -03:00 • committed by GitHub
parent f0cb9fc8af
commit 5c58d5c49a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 793 additions and 501 deletions
Download patch file Download diff file Expand all files Collapse all files

1
assets/keymaps/default-linux.json
View file

@ -337,6 +337,7 @@
"ctrl-shift-alt-enter": "agent::SendNextQueuedMessage",
"ctrl-shift-backspace": "agent::RemoveFirstQueuedMessage",
"ctrl-alt-e": "agent::EditFirstQueuedMessage",
"ctrl-alt-s": "agent::ToggleSteerFirstQueuedMessage",
"ctrl-alt-backspace": "agent::ClearMessageQueue",
"ctrl-shift-v": "agent::PasteRaw",
"ctrl-i": "agent::ToggleProfileSelector",

1
assets/keymaps/default-macos.json
View file

@ -381,6 +381,7 @@
"cmd-shift-alt-enter": "agent::SendNextQueuedMessage",
"cmd-shift-backspace": "agent::RemoveFirstQueuedMessage",
"cmd-ctrl-e": "agent::EditFirstQueuedMessage",
"cmd-ctrl-s": "agent::ToggleSteerFirstQueuedMessage",
"cmd-alt-backspace": "agent::ClearMessageQueue",
"cmd-shift-v": "agent::PasteRaw",
"cmd-i": "agent::ToggleProfileSelector",

1
assets/keymaps/default-windows.json
View file

@ -338,6 +338,7 @@
"ctrl-shift-alt-enter": "agent::SendNextQueuedMessage",
"ctrl-shift-backspace": "agent::RemoveFirstQueuedMessage",
"ctrl-alt-e": "agent::EditFirstQueuedMessage",
"ctrl-alt-s": "agent::ToggleSteerFirstQueuedMessage",
"ctrl-alt-backspace": "agent::ClearMessageQueue",
"ctrl-shift-v": "agent::PasteRaw",
"ctrl-i": "agent::ToggleProfileSelector",

72
crates/agent/src/tests/mod.rs
View file

@ -7487,9 +7487,9 @@ async fn test_queued_message_ends_turn_at_boundary(cx: &mut TestAppContext) {
fake_model
.send_last_completion_stream_event(LanguageModelCompletionEvent::Stop(StopReason::ToolUse));
// Signal that a message is queued before ending the stream
// Request that the turn end at the next boundary (a "steering" queued message)
thread.update(cx, |thread, _cx| {
thread.set_has_queued_message(true);
thread.set_end_turn_at_next_boundary(true);
});
// Now end the stream - tool will run, and the boundary check should see the queue
@ -7520,11 +7520,11 @@ async fn test_queued_message_ends_turn_at_boundary(cx: &mut TestAppContext) {
"Turn should have ended after tool completion due to queued message"
);
// Verify the queued message flag is still set
// Verify the boundary flag is still set
thread.update(cx, |thread, _cx| {
assert!(
thread.has_queued_message(),
"Should still have queued message flag set"
thread.end_turn_at_next_boundary(),
"Should still have the end-turn-at-boundary flag set"
);
});
@ -7537,6 +7537,68 @@ async fn test_queued_message_ends_turn_at_boundary(cx: &mut TestAppContext) {
});
}
#[gpui::test]
async fn test_queued_message_does_not_end_turn_without_boundary_flag(cx: &mut TestAppContext) {
init_test(cx);
always_allow_tools(cx);
let ThreadTest { model, thread, .. } = setup(cx, TestModel::Fake).await;
let fake_model = model.as_fake();
thread.update(cx, |thread, _cx| {
thread.add_tool(EchoTool);
});
let mut events = thread
.update(cx, |thread, cx| {
thread.send(UserMessageId::new(), ["Use the echo tool"], cx)
})
.unwrap();
cx.run_until_parked();
fake_model.send_last_completion_stream_event(LanguageModelCompletionEvent::ToolUse(
LanguageModelToolUse {
id: "tool_1".into(),
name: "echo".into(),
raw_input: r#"{"text": "hello"}"#.into(),
input: json!({"text": "hello"}),
is_input_complete: true,
thought_signature: None,
},
));
fake_model
.send_last_completion_stream_event(LanguageModelCompletionEvent::Stop(StopReason::ToolUse));
// Default behavior: even though a message is conceptually queued, we do NOT
// set the boundary flag, so the agent must keep going past the tool boundary
// (running to completion) rather than ending the turn early.
fake_model.end_last_completion_stream();
cx.run_until_parked();
// The agent should have issued a fresh completion request with the tool
// results instead of stopping — proof it continued past the boundary.
let continuation = fake_model.pending_completions();
assert_eq!(
continuation.len(),
1,
"Without the boundary flag, the turn should continue with another completion request"
);
// Let the continuation finish the turn naturally.
fake_model.send_last_completion_stream_text_chunk("All done");
fake_model
.send_last_completion_stream_event(LanguageModelCompletionEvent::Stop(StopReason::EndTurn));
fake_model.end_last_completion_stream();
let all_events = collect_events_until_stop(&mut events, cx).await;
let stop_reasons = stop_events(all_events);
assert_eq!(
stop_reasons,
vec![acp::StopReason::EndTurn],
"Turn should end only after the agent finishes, not at the tool boundary"
);
}
#[gpui::test]
async fn test_streaming_tool_error_breaks_stream_loop_immediately(cx: &mut TestAppContext) {
init_test(cx);

26
crates/agent/src/thread.rs
View file

@ -1209,9 +1209,10 @@ pub struct Thread {
/// Survives across multiple requests as the model performs tool calls and
/// we run tools, report their results.
running_turn: Option<RunningTurn>,
/// Flag indicating the UI has a queued message waiting to be sent.
/// Used to signal that the turn should end at the next message boundary.
has_queued_message: bool,
/// When set, the current turn ends at the next message boundary instead of
/// running to completion. The UI sets this to deliver a "steering" queued
/// message mid-task; by default queued messages wait for the turn to finish.
end_turn_at_next_boundary: bool,
pending_message: Option<AgentMessage>,
pub(crate) tools: BTreeMap<SharedString, Arc<dyn AnyAgentTool>>,
request_token_usage: HashMap<UserMessageId, language_model::TokenUsage>,
@ -1349,7 +1350,7 @@ impl Thread {
messages: Vec::new(),
user_store: project.read(cx).user_store(),
running_turn: None,
has_queued_message: false,
end_turn_at_next_boundary: false,
pending_message: None,
tools: BTreeMap::default(),
request_token_usage: HashMap::default(),
@ -1735,7 +1736,7 @@ impl Thread {
messages: db_thread.messages,
user_store: project.read(cx).user_store(),
running_turn: None,
has_queued_message: false,
end_turn_at_next_boundary: false,
pending_message: None,
tools: BTreeMap::default(),
request_token_usage: db_thread.request_token_usage.clone(),
@ -2246,12 +2247,12 @@ impl Thread {
})
}
pub fn set_has_queued_message(&mut self, has_queued: bool) {
self.has_queued_message = has_queued;
pub fn set_end_turn_at_next_boundary(&mut self, end_at_boundary: bool) {
self.end_turn_at_next_boundary = end_at_boundary;
}
pub fn has_queued_message(&self) -> bool {
self.has_queued_message
pub fn end_turn_at_next_boundary(&self) -> bool {
self.end_turn_at_next_boundary
}
fn accumulate_token_usage(&mut self, update: language_model::TokenUsage) {
@ -2981,9 +2982,10 @@ impl Thread {
} else if end_turn {
return Ok(());
} else {
let has_queued = this.update(cx, |this, _| this.has_queued_message())?;
if has_queued {
log::debug!("Queued message found, ending turn at message boundary");
let end_at_boundary =
this.update(cx, |this, _| this.end_turn_at_next_boundary())?;
if end_at_boundary {
log::debug!("Steering message queued, ending turn at message boundary");
return Ok(());
}
intent = CompletionIntent::ToolResults;

3
crates/agent_ui/src/agent_ui.rs
View file

@ -260,6 +260,9 @@ actions!(
RemoveFirstQueuedMessage,
/// Edits the first message in the queue (the next one to be sent).
EditFirstQueuedMessage,
/// Toggles steering for the first queued message: when on, it interrupts
/// the agent at its next step instead of waiting for it to finish.
ToggleSteerFirstQueuedMessage,
/// Clears all messages from the queue.
ClearMessageQueue,
/// Opens the permission granularity dropdown for the current tool call.

411
crates/agent_ui/src/conversation_view.rs
View file

@ -32,8 +32,8 @@ use gpui::{
Action, Animation, AnimationExt, AnyView, App, ClickEvent, ClipboardItem, CursorStyle,
ElementId, Empty, Entity, EventEmitter, FocusHandle, Focusable, Hsla, ListOffset, ListState,
ObjectFit, PlatformDisplay, ScrollHandle, SharedString, StyledText, Subscription, Task,
TaskExt, TextRun, TextStyle, WeakEntity, Window, WindowHandle, div, ease_in_out, img,
linear_color_stop, linear_gradient, list, pulsating_between,
TextRun, TextStyle, WeakEntity, Window, WindowHandle, div, ease_in_out, img, linear_color_stop,
linear_gradient, list, pulsating_between,
};
use language::{Buffer, Language, Rope};
use language_model::{LanguageModelCompletionError, LanguageModelRegistry};
@ -98,7 +98,8 @@ use crate::{
ScrollOutputLineDown, ScrollOutputLineUp, ScrollOutputPageDown, ScrollOutputPageUp,
ScrollOutputToBottom, ScrollOutputToNextMessage, ScrollOutputToPreviousMessage,
ScrollOutputToTop, SendImmediately, SendNextQueuedMessage, ToggleFastMode,
ToggleProfileSelector, ToggleThinkingEffortMenu, ToggleThinkingMode, UndoLastReject,
ToggleProfileSelector, ToggleSteerFirstQueuedMessage, ToggleThinkingEffortMenu,
ToggleThinkingMode, UndoLastReject,
};
const STOPWATCH_THRESHOLD: Duration = Duration::from_secs(30);
@ -106,15 +107,12 @@ const TOKEN_THRESHOLD: u64 = 250;
pub(crate) const DRAFT_PROMPT_PERSIST_DEBOUNCE: Duration = Duration::from_millis(250);
mod message_queue;
mod thread_search_bar;
mod thread_view;
pub use message_queue::*;
pub use thread_view::*;
pub struct QueuedMessage {
pub content: Vec<acp::ContentBlock>,
pub tracked_buffers: Vec<Entity<Buffer>>,
}
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
enum ThreadFeedback {
Positive,
@ -1473,41 +1471,6 @@ impl ConversationView {
matches!(self.server_state, ServerState::Loading { .. })
}
fn send_queued_message_at_index(
&mut self,
index: usize,
is_send_now: bool,
window: &mut Window,
cx: &mut Context<Self>,
) {
if let Some(active) = self.root_thread_view() {
active.update(cx, |active, cx| {
active.send_queued_message_at_index(index, is_send_now, window, cx);
});
}
}
fn move_queued_message_to_main_editor(
&mut self,
index: usize,
attempt: Option<InputAttempt>,
cursor_offset: Option<usize>,
window: &mut Window,
cx: &mut Context<Self>,
) {
if let Some(active) = self.root_thread_view() {
active.update(cx, |active, cx| {
active.move_queued_message_to_main_editor(
index,
attempt,
cursor_offset,
window,
cx,
);
});
}
}
fn handle_thread_event(
&mut self,
thread: &Entity<AcpThread>,
@ -1611,34 +1574,31 @@ impl ConversationView {
return;
}
let should_send_queued = if let Some(active) = self.root_thread_view() {
let sent_queued_message = if let Some(active) = self.root_thread_view() {
active.update(cx, |active, cx| {
if active.skip_queue_processing_count > 0 {
active.skip_queue_processing_count -= 1;
false
} else if active.user_interrupted_generation {
// Manual interruption: don't auto-process queue.
// Reset the flag so future completions can process normally.
active.user_interrupted_generation = false;
false
// Don't auto-send while the user is editing the next message.
let is_first_editor_focused = active
.message_queue
.first()
.is_some_and(|entry| entry.editor.focus_handle(cx).is_focused(window));
if let Some(entry) = active
.message_queue
.on_generation_stopped(is_first_editor_focused)
{
active.dispatch_queued_entry(entry, window, cx);
true
} else {
let has_queued = !active.local_queued_messages.is_empty();
// Don't auto-send if the first message editor is currently focused
let is_first_editor_focused = active
.queued_message_editors
.first()
.is_some_and(|editor| editor.focus_handle(cx).is_focused(window));
has_queued && !is_first_editor_focused
false
}
})
} else {
false
};
// Skip notifying when a queued message is about to be auto-sent: the agent
// Skip notifying when a queued message was just auto-sent: the agent
// is not actually idle and a notification here would fire just before the
// next turn starts.
if !should_send_queued {
if !sent_queued_message {
let used_tools = thread.read(cx).used_tools_since_last_user_message();
self.notify_with_sound(
if used_tools {
@ -1650,8 +1610,6 @@ impl ConversationView {
window,
cx,
);
} else {
self.send_queued_message_at_index(0, false, window, cx);
}
}
AcpThreadEvent::Refusal => {
@ -2407,185 +2365,6 @@ impl ConversationView {
.thread(self.root_session_id.as_ref()?, cx)
}
fn queued_messages_len(&self, cx: &App) -> usize {
self.root_thread_view()
.map(|thread| thread.read(cx).local_queued_messages.len())
.unwrap_or_default()
}
fn update_queued_message(
&mut self,
index: usize,
content: Vec<acp::ContentBlock>,
tracked_buffers: Vec<Entity<Buffer>>,
cx: &mut Context<Self>,
) -> bool {
match self.root_thread_view() {
Some(thread) => thread.update(cx, |thread, _cx| {
if index < thread.local_queued_messages.len() {
thread.local_queued_messages[index] = QueuedMessage {
content,
tracked_buffers,
};
true
} else {
false
}
}),
None => false,
}
}
fn queued_message_contents(&self, cx: &App) -> Vec<Vec<acp::ContentBlock>> {
match self.root_thread_view() {
None => Vec::new(),
Some(thread) => thread
.read(cx)
.local_queued_messages
.iter()
.map(|q| q.content.clone())
.collect(),
}
}
fn save_queued_message_at_index(&mut self, index: usize, cx: &mut Context<Self>) {
let editor = match self.root_thread_view() {
Some(thread) => thread.read(cx).queued_message_editors.get(index).cloned(),
None => None,
};
let Some(editor) = editor else {
return;
};
let contents_task = editor.update(cx, |editor, cx| editor.contents(false, cx));
cx.spawn(async move |this, cx| {
let Ok((content, tracked_buffers)) = contents_task.await else {
return Ok::<(), anyhow::Error>(());
};
this.update(cx, |this, cx| {
this.update_queued_message(index, content, tracked_buffers, cx);
cx.notify();
})?;
Ok(())
})
.detach_and_log_err(cx);
}
fn sync_queued_message_editors(&mut self, window: &mut Window, cx: &mut Context<Self>) {
let needed_count = self.queued_messages_len(cx);
let queued_messages = self.queued_message_contents(cx);
let agent_name = self.agent.agent_id();
let workspace = self.workspace.clone();
let project = self.project.downgrade();
let Some(connected) = self.as_connected() else {
return;
};
let Some(thread) = connected.active_view() else {
return;
};
let session_capabilities = thread.read(cx).session_capabilities.clone();
let current_count = thread.read(cx).queued_message_editors.len();
let last_synced = thread.read(cx).last_synced_queue_length;
if current_count == needed_count && needed_count == last_synced {
return;
}
if current_count > needed_count {
thread.update(cx, |thread, _cx| {
thread.queued_message_editors.truncate(needed_count);
thread
.queued_message_editor_subscriptions
.truncate(needed_count);
});
let editors = thread.read(cx).queued_message_editors.clone();
for (index, editor) in editors.into_iter().enumerate() {
if let Some(content) = queued_messages.get(index) {
editor.update(cx, |editor, cx| {
editor.set_read_only(true, cx);
editor.set_message(content.clone(), window, cx);
});
}
}
}
while thread.read(cx).queued_message_editors.len() < needed_count {
let index = thread.read(cx).queued_message_editors.len();
let content = queued_messages.get(index).cloned().unwrap_or_default();
let editor = cx.new(|cx| {
let mut editor = MessageEditor::new(
workspace.clone(),
project.clone(),
None,
session_capabilities.clone(),
agent_name.clone(),
"",
EditorMode::AutoHeight {
min_lines: 1,
max_lines: Some(10),
},
window,
cx,
);
editor.set_read_only(true, cx);
editor.set_message(content, window, cx);
editor
});
let subscription = cx.subscribe_in(
&editor,
window,
move |this, _editor, event, window, cx| match event {
MessageEditorEvent::InputAttempted {
attempt,
cursor_offset,
} => {
this.move_queued_message_to_main_editor(
index,
Some(attempt.clone()),
Some(*cursor_offset),
window,
cx,
);
}
MessageEditorEvent::LostFocus => {
this.save_queued_message_at_index(index, cx);
}
MessageEditorEvent::Cancel => {
window.focus(&this.focus_handle(cx), cx);
}
MessageEditorEvent::Send => {
window.focus(&this.focus_handle(cx), cx);
}
MessageEditorEvent::SendImmediately => {
this.send_queued_message_at_index(index, true, window, cx);
}
_ => {}
},
);
thread.update(cx, |thread, _cx| {
thread.queued_message_editors.push(editor);
thread
.queued_message_editor_subscriptions
.push(subscription);
});
}
if let Some(active) = self.root_thread_view() {
active.update(cx, |active, _cx| {
active.last_synced_queue_length = needed_count;
});
}
}
fn render_markdown(
&self,
markdown: Entity<Markdown>,
@ -3140,8 +2919,6 @@ impl ConversationView {
impl Render for ConversationView {
fn render(&mut self, window: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
self.sync_queued_message_editors(window, cx);
v_flex()
.track_focus(&self.focus_handle)
.size_full()
@ -3659,12 +3436,13 @@ pub(crate) mod tests {
.clone()
});
active_thread(&conversation_view, cx).update_in(cx, |thread, _window, cx| {
active_thread(&conversation_view, cx).update_in(cx, |thread, window, cx| {
thread.add_to_queue(
vec![acp::ContentBlock::Text(acp::TextContent::new(
"queued".to_string(),
))],
vec![],
window,
cx,
);
});
@ -3695,6 +3473,116 @@ pub(crate) mod tests {
);
}
#[gpui::test]
async fn test_queued_message_steer_defaults_off_and_toggles(cx: &mut TestAppContext) {
init_test(cx);
let (conversation_view, cx) =
setup_conversation_view(StubAgentServer::default_response(), cx).await;
add_to_workspace(conversation_view.clone(), cx);
let id = active_thread(&conversation_view, cx).update_in(cx, |thread, window, cx| {
thread.add_to_queue(
vec![acp::ContentBlock::Text(acp::TextContent::new(
"queued".to_string(),
))],
vec![],
window,
cx,
);
thread.message_queue.first_id().unwrap()
});
cx.run_until_parked();
// Default: steering is off, so the message waits for end-of-generation
// rather than interrupting the agent at the next boundary.
active_thread(&conversation_view, cx).read_with(cx, |thread, _cx| {
assert!(
!thread.message_queue.front_wants_steer(),
"steering should default off"
);
});
active_thread(&conversation_view, cx).update(cx, |thread, _cx| {
thread.message_queue.toggle_steer(id);
});
active_thread(&conversation_view, cx).read_with(cx, |thread, _cx| {
assert!(
thread.message_queue.front_wants_steer(),
"steering should be on after toggling"
);
});
}
#[gpui::test]
async fn test_queue_resumes_after_stop_and_new_message(cx: &mut TestAppContext) {
init_test(cx);
let connection = StubAgentConnection::new();
let (conversation_view, cx) =
setup_conversation_view(StubAgentServer::new(connection.clone()), cx).await;
add_to_workspace(conversation_view.clone(), cx);
let message_editor = message_editor(&conversation_view, cx);
message_editor.update_in(cx, |editor, window, cx| {
editor.set_text("first", window, cx);
});
active_thread(&conversation_view, cx)
.update_in(cx, |view, window, cx| view.send(window, cx));
cx.run_until_parked();
// Queue a follow-up while the agent is generating.
active_thread(&conversation_view, cx).update_in(cx, |thread, window, cx| {
thread.add_to_queue(
vec![acp::ContentBlock::Text(acp::TextContent::new(
"queued".to_string(),
))],
vec![],
window,
cx,
);
});
// User stops generation: the queued message must NOT be sent.
active_thread(&conversation_view, cx)
.update_in(cx, |thread, _window, cx| thread.cancel_generation(cx));
cx.run_until_parked();
let queue_len = active_thread(&conversation_view, cx)
.read_with(cx, |thread, _cx| thread.message_queue.len());
assert_eq!(queue_len, 1, "stopping must not send the queued message");
// User sends a new message, which should resume queue auto-processing.
message_editor.update_in(cx, |editor, window, cx| {
editor.set_text("second", window, cx);
});
active_thread(&conversation_view, cx)
.update_in(cx, |view, window, cx| view.send(window, cx));
cx.run_until_parked();
let session_id = conversation_view.read_with(cx, |view, cx| {
view.active_thread()
.unwrap()
.read(cx)
.thread
.read(cx)
.session_id()
.clone()
});
// When this generation completes, the queued message should be picked
// up automatically (regression test for the "frozen queue" bug).
connection.end_turn(session_id, acp::StopReason::EndTurn);
cx.run_until_parked();
let queue_len = active_thread(&conversation_view, cx)
.read_with(cx, |thread, _cx| thread.message_queue.len());
assert_eq!(
queue_len, 0,
"queued message should be auto-sent after the user re-engages"
);
}
#[gpui::test]
async fn test_notification_for_error(cx: &mut TestAppContext) {
init_test(cx);
@ -9744,19 +9632,21 @@ pub(crate) mod tests {
"queued message".to_string(),
))],
vec![],
window,
cx,
);
// Main editor must be empty for this path — it is by default, but
// assert to make the precondition explicit.
assert!(thread.message_editor.read(cx).is_empty(cx));
thread.move_queued_message_to_main_editor(0, None, None, window, cx);
let id = thread.message_queue.first_id().unwrap();
thread.move_queued_message_to_main_editor(id, None, None, window, cx);
});
cx.run_until_parked();
// Queue should now be empty.
let queue_len = active_thread(&conversation_view, cx)
.read_with(cx, |thread, _cx| thread.local_queued_messages.len());
.read_with(cx, |thread, _cx| thread.message_queue.len());
assert_eq!(queue_len, 0, "Queue should be empty after move");
// Main editor should contain the queued message text.
@ -9792,16 +9682,18 @@ pub(crate) mod tests {
"queued message".to_string(),
))],
vec![],
window,
cx,
);
thread.move_queued_message_to_main_editor(0, None, None, window, cx);
let id = thread.message_queue.first_id().unwrap();
thread.move_queued_message_to_main_editor(id, None, None, window, cx);
});
cx.run_until_parked();
// Queue should now be empty.
let queue_len = active_thread(&conversation_view, cx)
.read_with(cx, |thread, _cx| thread.local_queued_messages.len());
.read_with(cx, |thread, _cx| thread.message_queue.len());
assert_eq!(queue_len, 0, "Queue should be empty after move");
// Main editor should contain existing content + separator + queued content.
@ -9820,12 +9712,13 @@ pub(crate) mod tests {
setup_conversation_view(StubAgentServer::default_response(), cx).await;
add_to_workspace(conversation_view.clone(), cx);
active_thread(&conversation_view, cx).update(cx, |thread, cx| {
active_thread(&conversation_view, cx).update_in(cx, |thread, window, cx| {
thread.add_to_queue(
vec![acp::ContentBlock::Text(acp::TextContent::new(
"first queued".to_string(),
))],
vec![],
window,
cx,
);
thread.add_to_queue(
@ -9833,6 +9726,7 @@ pub(crate) mod tests {
"second queued".to_string(),
))],
vec![],
window,
cx,
);
});
@ -9847,7 +9741,7 @@ pub(crate) mod tests {
cx.run_until_parked();
let queue_len = active_thread(&conversation_view, cx)
.read_with(cx, |thread, _cx| thread.local_queued_messages.len());
.read_with(cx, |thread, _cx| thread.message_queue.len());
assert_eq!(
queue_len, 1,
"Up arrow should pull the last queued message out of the queue"
@ -9865,7 +9759,7 @@ pub(crate) mod tests {
cx.run_until_parked();
let queue_len = active_thread(&conversation_view, cx)
.read_with(cx, |thread, _cx| thread.local_queued_messages.len());
.read_with(cx, |thread, _cx| thread.message_queue.len());
assert_eq!(queue_len, 1, "Queue should be untouched");
let text = editor.update(cx, |editor, cx| editor.text(cx));
assert_eq!(text, "second queued");
@ -9879,7 +9773,7 @@ pub(crate) mod tests {
paste_into_queued_message(cx, ClipboardItem::new_string("PASTED".to_string())).await;
let queue_len = active_thread(&conversation_view, cx)
.read_with(cx, |thread, _cx| thread.local_queued_messages.len());
.read_with(cx, |thread, _cx| thread.message_queue.len());
assert_eq!(queue_len, 0);
let text = message_editor(&conversation_view, cx).update(cx, |editor, cx| editor.text(cx));
@ -9909,7 +9803,7 @@ pub(crate) mod tests {
.await;
let queue_len = active_thread(&conversation_view, cx)
.read_with(cx, |thread, _cx| thread.local_queued_messages.len());
.read_with(cx, |thread, _cx| thread.message_queue.len());
assert_eq!(queue_len, 0);
let text = message_editor(&conversation_view, cx).update(cx, |editor, cx| editor.text(cx));
@ -9933,7 +9827,7 @@ pub(crate) mod tests {
setup_conversation_view(StubAgentServer::default_response(), cx).await;
add_to_workspace(conversation_view.clone(), cx);
active_thread(&conversation_view, cx).update_in(cx, |thread, _window, cx| {
active_thread(&conversation_view, cx).update_in(cx, |thread, window, cx| {
thread
.session_capabilities
.write()
@ -9943,6 +9837,7 @@ pub(crate) mod tests {
"queued message".to_string(),
))],
vec![],
window,
cx,
);
});
@ -9951,10 +9846,10 @@ pub(crate) mod tests {
let queued_editor = active_thread(&conversation_view, cx).read_with(cx, |thread, _cx| {
thread
.queued_message_editors
.message_queue
.first()
.cloned()
.expect("queued message editor not synced")
.map(|entry| entry.editor.clone())
.expect("queued message editor not created")
});
cx.write_to_clipboard(clipboard);

191
crates/agent_ui/src/conversation_view/message_queue.rs Normal file
View file

@ -0,0 +1,191 @@
use std::collections::VecDeque;
use super::*;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct QueueEntryId(usize);
pub struct QueueEntry {
pub id: QueueEntryId,
pub content: Vec<acp::ContentBlock>,
pub tracked_buffers: Vec<Entity<Buffer>>,
/// When true, this message interrupts the agent at the next turn boundary
/// instead of waiting for generation to fully complete. Only the front
/// entry's value matters, since messages are delivered in FIFO order.
pub steer: bool,
pub editor: Entity<MessageEditor>,
pub _subscription: Subscription,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProcessingState {
AutoProcess,
Paused,
// Sending a message out of turn cancelled the current generation; we must
// absorb the Stopped event from that cancellation before resuming
// auto-processing, otherwise the queue would double-send.
AbsorbingCancel,
}
/// Holds follow-up messages typed while the agent is generating, along with
/// the state machine that decides when they're auto-sent.
pub struct MessageQueue {
entries: VecDeque<QueueEntry>,
processing_state: ProcessingState,
can_fast_track: bool,
next_id: usize,
}
impl Default for MessageQueue {
fn default() -> Self {
Self {
entries: VecDeque::new(),
processing_state: ProcessingState::AutoProcess,
can_fast_track: false,
next_id: 0,
}
}
}
impl MessageQueue {
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn first(&self) -> Option<&QueueEntry> {
self.entries.front()
}
pub fn first_id(&self) -> Option<QueueEntryId> {
self.entries.front().map(|entry| entry.id)
}
pub fn last_id(&self) -> Option<QueueEntryId> {
self.entries.back().map(|entry| entry.id)
}
/// Whether the next message should interrupt the agent at the next turn
/// boundary. Drives the native thread's boundary flag.
pub fn front_wants_steer(&self) -> bool {
self.entries.front().is_some_and(|entry| entry.steer)
}
pub fn toggle_steer(&mut self, id: QueueEntryId) {
if let Some(entry) = self.entries.iter_mut().find(|entry| entry.id == id) {
entry.steer = !entry.steer;
}
}
pub fn iter(&self) -> impl Iterator<Item = &QueueEntry> {
self.entries.iter()
}
pub fn can_fast_track(&self) -> bool {
self.can_fast_track && !self.entries.is_empty()
}
pub fn entry_by_id(&self, id: QueueEntryId) -> Option<&QueueEntry> {
self.entries.iter().find(|entry| entry.id == id)
}
pub fn entry_by_id_mut(&mut self, id: QueueEntryId) -> Option<&mut QueueEntry> {
self.entries.iter_mut().find(|entry| entry.id == id)
}
/// Allocates a stable ID for a new entry. This is separate from `enqueue`
/// because the editor event subscription must capture the ID before the
/// `QueueEntry` (which owns that subscription) can be constructed.
pub fn next_id(&mut self) -> QueueEntryId {
let id = QueueEntryId(self.next_id);
self.next_id += 1;
id
}
/// Queuing a message is active engagement, so it also resumes
/// auto-processing if the queue was paused.
pub fn enqueue(&mut self, entry: QueueEntry) {
self.entries.push_back(entry);
self.processing_state = ProcessingState::AutoProcess;
self.can_fast_track = true;
}
pub fn remove(&mut self, id: QueueEntryId) -> Option<QueueEntry> {
let index = self.entries.iter().position(|entry| entry.id == id)?;
self.entries.remove(index)
}
pub fn clear(&mut self) {
self.entries.clear();
self.can_fast_track = false;
}
/// Pops the front entry if a fast-track send is allowed (the user just
/// queued a message and pressed Enter on an empty main editor).
///
/// This works even when paused — pressing Enter is an explicit user
/// action, distinct from auto-processing. If a generation is in flight,
/// the dispatch will cancel it, so we must absorb that cancellation's
/// Stopped event to avoid double-sending the next entry.
pub fn try_fast_track(&mut self, is_generating: bool) -> Option<QueueEntry> {
if !self.can_fast_track {
return None;
}
self.can_fast_track = false;
let entry = self.entries.pop_front()?;
self.processing_state = if is_generating {
ProcessingState::AbsorbingCancel
} else {
ProcessingState::AutoProcess
};
Some(entry)
}
/// Handles a generation Stopped event, returning the entry to auto-send,
/// if any.
pub fn on_generation_stopped(&mut self, is_first_editor_focused: bool) -> Option<QueueEntry> {
match self.processing_state {
ProcessingState::AbsorbingCancel => {
// This Stopped event came from a cancellation we initiated
// ourselves (e.g. "Send Now"); swallow it and resume.
self.processing_state = ProcessingState::AutoProcess;
None
}
ProcessingState::Paused => None,
ProcessingState::AutoProcess => {
// Don't auto-send while the user is editing the next message.
if is_first_editor_focused {
None
} else {
self.entries.pop_front()
}
}
}
}
/// Removes an entry for an explicit "Send Now". If a generation is in
/// flight, the dispatch will cancel it, so we must absorb that
/// cancellation's Stopped event.
pub fn send_now(&mut self, id: QueueEntryId, is_generating: bool) -> Option<QueueEntry> {
let entry = self.remove(id)?;
if is_generating {
self.processing_state = ProcessingState::AbsorbingCancel;
}
Some(entry)
}
/// Called when the user stops generation; queued messages stay put until
/// the user re-engages.
pub fn pause(&mut self) {
self.processing_state = ProcessingState::Paused;
}
/// Called when the user sends a new message, re-enabling auto-processing.
/// This is what un-freezes the queue after a manual stop.
pub fn resume(&mut self) {
self.processing_state = ProcessingState::AutoProcess;
}
}

586
crates/agent_ui/src/conversation_view/thread_view.rs
View file

@ -591,10 +591,7 @@ pub struct ThreadView {
pub editor_expanded: bool,
pub should_be_following: bool,
pub editing_message: Option<usize>,
pub local_queued_messages: Vec<QueuedMessage>,
pub queued_message_editors: Vec<Entity<MessageEditor>>,
pub queued_message_editor_subscriptions: Vec<Subscription>,
pub last_synced_queue_length: usize,
pub message_queue: MessageQueue,
pub turn_fields: TurnFields,
pub discarded_partial_edits: HashSet<acp::ToolCallId>,
pub is_loading_contents: bool,
@ -605,9 +602,6 @@ pub struct ThreadView {
_save_task: Option<Task<()>>,
_draft_resolve_task: Option<Task<()>>,
_sandbox_status_refresh_task: Option<Task<()>>,
pub skip_queue_processing_count: usize,
pub user_interrupted_generation: bool,
pub can_fast_track_queue: bool,
pub hovered_edited_file_buttons: Option<usize>,
pub in_flight_prompt: Option<Vec<acp::ContentBlock>>,
pub _subscriptions: Vec<Subscription>,
@ -989,10 +983,7 @@ impl ThreadView {
editor_expanded: false,
should_be_following: false,
editing_message: None,
local_queued_messages: Vec::new(),
queued_message_editors: Vec::new(),
queued_message_editor_subscriptions: Vec::new(),
last_synced_queue_length: 0,
message_queue: MessageQueue::default(),
turn_fields: TurnFields::default(),
discarded_partial_edits: HashSet::default(),
is_loading_contents: false,
@ -1002,9 +993,6 @@ impl ThreadView {
_save_task: None,
_draft_resolve_task: None,
_sandbox_status_refresh_task: None,
skip_queue_processing_count: 0,
user_interrupted_generation: false,
can_fast_track_queue: false,
hovered_edited_file_buttons: None,
in_flight_prompt: None,
message_editor,
@ -1188,7 +1176,7 @@ impl ThreadView {
}
pub fn has_queued_messages(&self) -> bool {
!self.local_queued_messages.is_empty()
!self.message_queue.is_empty()
}
pub fn is_imported_thread(&self, cx: &App) -> bool {
@ -1431,14 +1419,10 @@ impl ThreadView {
let is_editor_empty = message_editor.read(cx).is_empty(cx);
let is_generating = thread.read(cx).status() != ThreadStatus::Idle;
let has_queued = self.has_queued_messages();
if is_editor_empty && self.can_fast_track_queue && has_queued {
self.can_fast_track_queue = false;
self.send_queued_message_at_index(0, true, window, cx);
return;
}
if is_editor_empty {
if let Some(entry) = self.message_queue.try_fast_track(is_generating) {
self.dispatch_queued_entry(entry, window, cx);
}
return;
}
@ -1547,7 +1531,7 @@ impl ThreadView {
// Queue the remainder first, then start the command turn; the
// queue auto-processes when the command turn stops.
if !content.is_empty() {
this.add_to_queue(content, tracked_buffers, cx);
this.add_to_queue(content, tracked_buffers, window, cx);
}
this.send_content(
Task::ready(Ok(Some((vec![command_block], Vec::new())))),
@ -1572,6 +1556,9 @@ impl ThreadView {
self.thread_error.take();
self.thread_feedback.clear();
self.editing_message.take();
// Sending a message is active engagement: un-freeze the queue if it
// was paused by a manual stop.
self.message_queue.resume();
if self.should_be_following {
self.workspace
@ -1780,8 +1767,7 @@ impl ThreadView {
cx: &mut Context<Self>,
) {
let thread = self.thread.clone();
self.skip_queue_processing_count = 0;
self.user_interrupted_generation = true;
self.message_queue.pause();
let cancelled = thread.update(cx, |thread, cx| thread.cancel(cx));
@ -1920,7 +1906,7 @@ impl ThreadView {
pub fn cancel_generation(&mut self, cx: &mut Context<Self>) {
self.thread_retry_status.take();
self.thread_error.take();
self.user_interrupted_generation = true;
self.message_queue.pause();
self._cancel_task = Some(self.thread.update(cx, |thread, cx| thread.cancel(cx)));
self.sync_generating_indicator(cx);
cx.notify();
@ -2036,8 +2022,7 @@ impl ThreadView {
}
this.update_in(cx, |this, window, cx| {
this.add_to_queue(content, tracked_buffers, cx);
this.can_fast_track_queue = true;
this.add_to_queue(content, tracked_buffers, window, cx);
message_editor.update(cx, |message_editor, cx| {
message_editor.clear(window, cx);
});
@ -2052,55 +2037,165 @@ impl ThreadView {
&mut self,
content: Vec<acp::ContentBlock>,
tracked_buffers: Vec<Entity<Buffer>>,
window: &mut Window,
cx: &mut Context<Self>,
) {
self.local_queued_messages.push(QueuedMessage {
// The ID must be allocated up front so the editor event subscription
// can capture it before the entry (which owns the subscription) exists.
let id = self.message_queue.next_id();
let editor = cx.new(|cx| {
let mut editor = MessageEditor::new(
self.workspace.clone(),
self.project.clone(),
None,
self.session_capabilities.clone(),
self.agent_id.clone(),
"",
EditorMode::AutoHeight {
min_lines: 1,
max_lines: Some(10),
},
window,
cx,
);
editor.set_read_only(true, cx);
editor.set_message(content.clone(), window, cx);
editor
});
let subscription =
cx.subscribe_in(&editor, window, move |this, _editor, event, window, cx| {
this.handle_queue_editor_event(id, event, window, cx);
});
self.message_queue.enqueue(QueueEntry {
id,
content,
tracked_buffers,
steer: false,
editor,
_subscription: subscription,
});
self.sync_queue_flag_to_native_thread(cx);
cx.notify();
}
fn handle_queue_editor_event(
&mut self,
id: QueueEntryId,
event: &MessageEditorEvent,
window: &mut Window,
cx: &mut Context<Self>,
) {
match event {
MessageEditorEvent::InputAttempted {
attempt,
cursor_offset,
} => {
self.move_queued_message_to_main_editor(
id,
Some(attempt.clone()),
Some(*cursor_offset),
window,
cx,
);
}
MessageEditorEvent::LostFocus => {
self.save_queued_message(id, cx);
}
MessageEditorEvent::Cancel | MessageEditorEvent::Send => {
window.focus(&self.message_editor.focus_handle(cx), cx);
}
MessageEditorEvent::SendImmediately => {
self.send_queued_message_now(id, window, cx);
}
_ => {}
}
}
fn save_queued_message(&mut self, id: QueueEntryId, cx: &mut Context<Self>) {
let Some(entry) = self.message_queue.entry_by_id(id) else {
return;
};
let contents_task = entry
.editor
.update(cx, |editor, cx| editor.contents(false, cx));
cx.spawn(async move |this, cx| {
let (content, tracked_buffers) = contents_task.await?;
this.update(cx, |this, cx| {
if let Some(entry) = this.message_queue.entry_by_id_mut(id) {
entry.content = content;
entry.tracked_buffers = tracked_buffers;
}
cx.notify();
})?;
Ok::<(), anyhow::Error>(())
})
.detach_and_log_err(cx);
}
pub fn remove_from_queue(
&mut self,
index: usize,
id: QueueEntryId,
cx: &mut Context<Self>,
) -> Option<QueuedMessage> {
if index < self.local_queued_messages.len() {
let removed = self.local_queued_messages.remove(index);
) -> Option<QueueEntry> {
let removed = self.message_queue.remove(id);
if removed.is_some() {
self.sync_queue_flag_to_native_thread(cx);
Some(removed)
} else {
None
}
removed
}
fn toggle_queue_entry_steer(&mut self, id: QueueEntryId, cx: &mut Context<Self>) {
self.message_queue.toggle_steer(id);
self.sync_queue_flag_to_native_thread(cx);
cx.notify();
}
pub fn sync_queue_flag_to_native_thread(&self, cx: &mut Context<Self>) {
if let Some(native_thread) = self.as_native_thread(cx) {
let has_queued = self.has_queued_messages();
// By default queued messages wait for the turn to fully complete.
// Only a "steering" front message ends the turn at the next boundary.
let end_at_boundary = self.message_queue.front_wants_steer();
native_thread.update(cx, |thread, _| {
thread.set_has_queued_message(has_queued);
thread.set_end_turn_at_next_boundary(end_at_boundary);
});
}
}
pub fn send_queued_message_at_index(
pub fn send_queued_message_now(
&mut self,
index: usize,
is_send_now: bool,
id: QueueEntryId,
window: &mut Window,
cx: &mut Context<Self>,
) {
let Some(queued) = self.remove_from_queue(index, cx) else {
return;
};
let is_generating = self.thread.read(cx).status() == acp_thread::ThreadStatus::Generating;
if let Some(entry) = self.message_queue.send_now(id, is_generating) {
self.dispatch_queued_entry(entry, window, cx);
}
}
/// The shared "actually send this entry" path, used by fast-track,
/// auto-processing on Stopped, and "Send Now". The entry must already have
/// been removed from the queue.
pub fn dispatch_queued_entry(
&mut self,
entry: QueueEntry,
window: &mut Window,
cx: &mut Context<Self>,
) {
self.sync_queue_flag_to_native_thread(cx);
cx.emit(AcpThreadViewEvent::Interacted);
self.message_editor.focus_handle(cx).focus(window, cx);
let content = queued.content;
let tracked_buffers = queued.tracked_buffers;
let content = entry.content;
let tracked_buffers = entry.tracked_buffers;
// A queued message can itself be a built-in command (e.g. the user typed
// `/compact` while a turn was generating). Detect that so we run it as a
@ -2117,16 +2212,6 @@ impl ThreadView {
})
.is_some();
// Only increment skip count for "Send Now" operations (out-of-order sends)
// Normal auto-processing from the Stopped handler doesn't need to skip.
// We only skip the Stopped event from the cancelled generation, NOT the
// Stopped event from the newly sent message (which should trigger queue processing).
if is_send_now {
let is_generating =
self.thread.read(cx).status() == acp_thread::ThreadStatus::Generating;
self.skip_queue_processing_count += if is_generating { 1 } else { 0 };
}
let cancelled = self.thread.update(cx, |thread, cx| thread.cancel(cx));
let workspace = self.workspace.clone();
@ -2150,13 +2235,13 @@ impl ThreadView {
pub fn move_queued_message_to_main_editor(
&mut self,
index: usize,
id: QueueEntryId,
attempt: Option<InputAttempt>,
cursor_offset: Option<usize>,
window: &mut Window,
cx: &mut Context<Self>,
) -> bool {
let Some(queued_message) = self.remove_from_queue(index, cx) else {
let Some(queued_message) = self.remove_from_queue(id, cx) else {
return false;
};
let queued_content = queued_message.content;
@ -2203,12 +2288,15 @@ impl ThreadView {
window: &mut Window,
cx: &mut Context<Self>,
) {
if !self.message_editor.read(cx).is_empty(cx) || self.local_queued_messages.is_empty() {
if !self.message_editor.read(cx).is_empty(cx) {
cx.propagate();
return;
}
let last_index = self.local_queued_messages.len() - 1;
self.move_queued_message_to_main_editor(last_index, None, None, window, cx);
let Some(last_id) = self.message_queue.last_id() else {
cx.propagate();
return;
};
self.move_queued_message_to_main_editor(last_id, None, None, window, cx);
}
// editor methods
@ -3398,7 +3486,7 @@ impl ThreadView {
_window: &mut Window,
cx: &Context<Self>,
) -> impl IntoElement {
let queue_count = self.local_queued_messages.len();
let queue_count = self.message_queue.len();
let title: SharedString = if queue_count == 1 {
"1 Queued Message".into()
} else {
@ -3433,16 +3521,15 @@ impl ThreadView {
)
.on_click(cx.listener(|this, _, _, cx| {
this.clear_queue(cx);
this.can_fast_track_queue = false;
cx.notify();
})),
)
.into_any_element()
}
fn clear_queue(&mut self, cx: &mut Context<Self>) {
self.local_queued_messages.clear();
self.message_queue.clear();
self.sync_queue_flag_to_native_thread(cx);
cx.notify();
}
fn render_plan_summary(
@ -4198,6 +4285,40 @@ impl ThreadView {
.into_any()
}
fn render_queue_steer_button(
&self,
entry_id: QueueEntryId,
index: usize,
is_next: bool,
steer_on: bool,
cx: &Context<Self>,
) -> impl IntoElement {
let focus_handle = self.message_editor.focus_handle(cx);
Button::new(("steer", index), "Steer")
.label_size(LabelSize::Small)
.toggle_state(steer_on)
.selected_style(ButtonStyle::Tinted(TintColor::Accent))
.when(is_next, |this| {
this.key_binding(
KeyBinding::for_action_in(&ToggleSteerFirstQueuedMessage, &focus_handle, cx)
.map(|kb| kb.size(rems_from_px(12.))),
)
})
.tooltip(move |_window, cx| {
Tooltip::with_meta(
"Steer",
None,
"Interrupt the agent at its next step to send this message. \
When off, queued messages wait for the agent to finish.",
cx,
)
})
.on_click(cx.listener(move |this, _, _, cx| {
this.toggle_queue_entry_steer(entry_id, cx);
}))
}
fn render_message_queue_entries(
&self,
_window: &mut Window,
@ -4206,176 +4327,178 @@ impl ThreadView {
let message_editor = self.message_editor.read(cx);
let focus_handle = message_editor.focus_handle(cx);
let queued_message_editors = &self.queued_message_editors;
let queue_len = queued_message_editors.len();
let can_fast_track = self.can_fast_track_queue && queue_len > 0;
let queue_len = self.message_queue.len();
let can_fast_track = self.message_queue.can_fast_track();
let is_native = self.as_native_thread(cx).is_some();
v_flex()
.id("message_queue_list")
.max_h_40()
.overflow_y_scroll()
.children(
queued_message_editors
.iter()
.enumerate()
.map(|(index, editor)| {
let is_next = index == 0;
let (icon_color, tooltip_text) = if is_next {
(Color::Accent, "Next in Queue")
} else {
(Color::Muted, "In Queue")
};
.children(self.message_queue.iter().enumerate().map(|(index, entry)| {
let entry_id = entry.id;
let editor = &entry.editor;
let is_next = index == 0;
let (icon_color, tooltip_text) = if is_next {
(Color::Accent, "Next in Queue")
} else {
(Color::Muted, "In Queue")
};
let editor_focused = editor.focus_handle(cx).is_focused(_window);
let keybinding_size = rems_from_px(12.);
let editor_focused = editor.focus_handle(cx).is_focused(_window);
let keybinding_size = rems_from_px(12.);
let steer_on = entry.steer;
let min_width = rems_from_px(160.);
h_flex()
.group("queue_entry")
.w_full()
.p_1p5()
.gap_1()
.bg(cx.theme().colors().editor_background)
.when(index < queue_len - 1, |this| {
this.border_b_1()
.border_color(cx.theme().colors().border_variant)
})
.child(
div()
.id("next_in_queue")
.child(
Icon::new(IconName::Circle)
.size(IconSize::Small)
.color(icon_color),
)
.tooltip(Tooltip::text(tooltip_text)),
)
.child(editor.clone())
.child(if editor_focused {
h_flex()
.group("queue_entry")
.w_full()
.p_1p5()
.gap_1()
.bg(cx.theme().colors().editor_background)
.when(index < queue_len - 1, |this| {
this.border_b_1()
.border_color(cx.theme().colors().border_variant)
.min_w(min_width)
.justify_end()
.child(
IconButton::new(("edit", index), IconName::Pencil)
.icon_size(IconSize::Small)
.tooltip(|_window, cx| {
Tooltip::with_meta(
"Edit Queued Message",
None,
"Type anything to edit",
cx,
)
})
.on_click(cx.listener(move |this, _, window, cx| {
this.move_queued_message_to_main_editor(
entry_id, None, None, window, cx,
);
})),
)
.when(is_native, |row| {
row.child(self.render_queue_steer_button(
entry_id, index, is_next, steer_on, cx,
))
})
.child(
div()
.id("next_in_queue")
.child(
Icon::new(IconName::Circle)
.size(IconSize::Small)
.color(icon_color),
Button::new(("send_now_focused", index), "Send Now")
.label_size(LabelSize::Small)
.style(ButtonStyle::Outlined)
.key_binding(
KeyBinding::for_action_in(
&SendImmediately,
&editor.focus_handle(cx),
cx,
)
.map(|kb| kb.size(keybinding_size)),
)
.tooltip(Tooltip::text(tooltip_text)),
.on_click(cx.listener(move |this, _, window, cx| {
this.send_queued_message_now(entry_id, window, cx);
})),
)
.child(editor.clone())
.child(if editor_focused {
h_flex()
.gap_1()
.min_w(rems_from_px(150.))
.justify_end()
.child(
IconButton::new(("edit", index), IconName::Pencil)
.icon_size(IconSize::Small)
.tooltip(|_window, cx| {
Tooltip::with_meta(
"Edit Queued Message",
None,
"Type anything to edit",
} else {
h_flex()
.when(!is_next, |this| this.visible_on_hover("queue_entry"))
.gap_1()
.min_w(min_width)
.justify_end()
.child(
IconButton::new(("delete", index), IconName::Trash)
.icon_size(IconSize::Small)
.tooltip({
let focus_handle = focus_handle.clone();
move |_window, cx| {
if is_next {
Tooltip::for_action_in(
"Remove Message from Queue",
&RemoveFirstQueuedMessage,
&focus_handle,
cx,
)
})
.on_click(cx.listener(move |this, _, window, cx| {
this.move_queued_message_to_main_editor(
index, None, None, window, cx,
);
})),
)
.child(
Button::new(("send_now_focused", index), "Send Now")
.label_size(LabelSize::Small)
.style(ButtonStyle::Outlined)
.key_binding(
KeyBinding::for_action_in(
&SendImmediately,
&editor.focus_handle(cx),
} else {
Tooltip::simple("Remove Message from Queue", cx)
}
}
})
.on_click(cx.listener(move |this, _, _, cx| {
this.remove_from_queue(entry_id, cx);
cx.notify();
})),
)
.child(
IconButton::new(("edit", index), IconName::Pencil)
.icon_size(IconSize::Small)
.tooltip({
let focus_handle = focus_handle.clone();
move |_window, cx| {
if is_next {
Tooltip::for_action_in(
"Edit",
&EditFirstQueuedMessage,
&focus_handle,
cx,
)
.map(|kb| kb.size(keybinding_size)),
)
.on_click(cx.listener(move |this, _, window, cx| {
this.send_queued_message_at_index(
index, true, window, cx,
);
})),
)
} else {
h_flex()
.when(!is_next, |this| this.visible_on_hover("queue_entry"))
.gap_1()
.min_w(rems_from_px(150.))
.justify_end()
.child(
IconButton::new(("delete", index), IconName::Trash)
.icon_size(IconSize::Small)
.tooltip({
let focus_handle = focus_handle.clone();
move |_window, cx| {
if is_next {
Tooltip::for_action_in(
"Remove Message from Queue",
&RemoveFirstQueuedMessage,
&focus_handle,
cx,
)
} else {
Tooltip::simple(
"Remove Message from Queue",
cx,
)
}
}
})
.on_click(cx.listener(move |this, _, _, cx| {
this.remove_from_queue(index, cx);
cx.notify();
})),
)
.child(
IconButton::new(("edit", index), IconName::Pencil)
.icon_size(IconSize::Small)
.tooltip({
let focus_handle = focus_handle.clone();
move |_window, cx| {
if is_next {
Tooltip::for_action_in(
"Edit",
&EditFirstQueuedMessage,
&focus_handle,
cx,
)
} else {
Tooltip::simple("Edit", cx)
}
}
})
.on_click(cx.listener(move |this, _, window, cx| {
this.move_queued_message_to_main_editor(
index, None, None, window, cx,
);
})),
)
.child(
Button::new(("send_now", index), "Send Now")
.label_size(LabelSize::Small)
.when(is_next, |this| this.style(ButtonStyle::Outlined))
.when(is_next && message_editor.is_empty(cx), |this| {
let action: Box<dyn gpui::Action> =
if can_fast_track {
Box::new(Chat)
} else {
Box::new(SendNextQueuedMessage)
};
this.key_binding(
KeyBinding::for_action_in(
action.as_ref(),
&focus_handle.clone(),
cx,
)
.map(|kb| kb.size(keybinding_size)),
)
})
.on_click(cx.listener(move |this, _, window, cx| {
this.send_queued_message_at_index(
index, true, window, cx,
);
})),
)
} else {
Tooltip::simple("Edit", cx)
}
}
})
.on_click(cx.listener(move |this, _, window, cx| {
this.move_queued_message_to_main_editor(
entry_id, None, None, window, cx,
);
})),
)
.when(is_native, |row| {
row.child(self.render_queue_steer_button(
entry_id, index, is_next, steer_on, cx,
))
})
}),
)
.child(
Button::new(("send_now", index), "Send Now")
.label_size(LabelSize::Small)
.when(is_next, |this| this.style(ButtonStyle::Outlined))
.when(is_next && message_editor.is_empty(cx), |this| {
let action: Box<dyn gpui::Action> = if can_fast_track {
Box::new(Chat)
} else {
Box::new(SendNextQueuedMessage)
};
this.key_binding(
KeyBinding::for_action_in(
action.as_ref(),
&focus_handle.clone(),
cx,
)
.map(|kb| kb.size(keybinding_size)),
)
})
.on_click(cx.listener(move |this, _, window, cx| {
this.send_queued_message_now(entry_id, window, cx);
})),
)
})
}))
.into_any_element()
}
@ -11556,20 +11679,33 @@ impl Render for ThreadView {
}),
)
.on_action(cx.listener(|this, _: &SendNextQueuedMessage, window, cx| {
this.send_queued_message_at_index(0, true, window, cx);
if let Some(id) = this.message_queue.first_id() {
this.send_queued_message_now(id, window, cx);
}
}))
.on_action(cx.listener(|this, _: &RemoveFirstQueuedMessage, _, cx| {
this.remove_from_queue(0, cx);
cx.notify();
if let Some(id) = this.message_queue.first_id() {
this.remove_from_queue(id, cx);
cx.notify();
}
}))
.on_action(cx.listener(|this, _: &EditFirstQueuedMessage, window, cx| {
this.move_queued_message_to_main_editor(0, None, None, window, cx);
if let Some(id) = this.message_queue.first_id() {
this.move_queued_message_to_main_editor(id, None, None, window, cx);
}
}))
.on_action(
cx.listener(|this, _: &ToggleSteerFirstQueuedMessage, _, cx| {
if this.as_native_thread(cx).is_none() {
return;
}
if let Some(id) = this.message_queue.first_id() {
this.toggle_queue_entry_steer(id, cx);
}
}),
)
.on_action(cx.listener(|this, _: &ClearMessageQueue, _, cx| {
this.local_queued_messages.clear();
this.sync_queue_flag_to_native_thread(cx);
this.can_fast_track_queue = false;
cx.notify();
this.clear_queue(cx);
}))
.on_action(cx.listener(|this, _: &ToggleProfileSelector, window, cx| {
if let Some(config_options_view) = this.config_options_view.clone() {

2
docs/src/ai/agent-panel.md
View file

@ -64,7 +64,7 @@ You can click on the card that contains your message and re-submit it with an ad
Messages sent while the agent is in the generating state get, by default, queued.
For the Zed Agent, queued messages get sent at the next turn boundary, which is usually between a tool call and a response, whereas for External Agents, the message gets sent at the end of the generation.
By default, queued messages get sent once the agent finishes generating. If you want a queued message to reach the Zed Agent sooner—interrupting it at its next step (usually between a tool call and a response) rather than waiting for it to finish—toggle "Steer" on that message. Steering is only available for the Zed Agent, since Zed can't detect turn boundaries for external agents.
You can edit or remove (an individual or all) queued messages.
You can also still interrupt the agent immediately if you want by either clicking on the stop button or by clicking the "Send Now" (double-enter) on a queued message.