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

agent: Use receipts and reported state for native turns (#65420)

Browse source 
Activate native acceptance and reported state atomically with its
completion consumers.

- Return acceptance with the retained message ID. Session-owned drivers
report running/requires-action/idle independently of receipt and
completion observers.
- Keep completion observers native-only for subagents and eval. Native
no longer implements the legacy completion-based prompt path; external
v1 behavior and SDK/wire rollout pins remain unchanged.
- Preserve checkpoints, rewind, command expansion order, retained source
fields, usage and provider-shaped persistence. Compact markers retain
identity without user bubbles.
- Use `NativeRetryAction` for retained command intent. Failed
preparation retries the resolved action; prepared history resumes
without duplicate expansion.

Prerequisites #65502 and #65504 are merged. This PR targets `main`;
conflicts are resolved and only the six-file native slice remains.

Validation on the merged base: **1,815 tests passed, 11 ignored**;
100-seed native/subagent/E2E-helper/skill-body checks; eight-crate
Clippy; test-target and E2E-feature compilation; both Zed binary checks;
formatting and diff checks. Scoped conflict-resolution review passed.
Credentialed live E2E/provider, live SIGTERM and non-macOS workflows
remain unrun.

Review thread:
https://delta.dev/trunk/join/thread_01m4e0prq2aaq6k0avcqe3688r#s=Deoszm1s

Release Notes:

- N/A
This commit is contained in:
Ben Brandt 2026-10-14 19:28:02 +00:00 • committed by GitHub
parent 0943bcb76c
commit d67bbfda3c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 2513 additions and 501 deletions
Download patch file Download diff file Expand all files Collapse all files

14
crates/acp_thread/src/connection.rs
View file

@ -210,13 +210,17 @@ pub trait AgentConnection {
Ok(())
}
/// This completion-based path retains a legacy response; receipt transports
/// acknowledge acceptance separately through `receipt_submissions`.
/// Receipt transports acknowledge acceptance through `receipt_submissions`
/// instead of implementing this completion-based path.
fn prompt(
&self,
params: acp_v2::PromptRequest,
cx: &mut App,
) -> Task<Result<acp_v1::PromptResponse>>;
_params: acp_v2::PromptRequest,
_cx: &mut App,
) -> Task<Result<acp_v1::PromptResponse>> {
Task::ready(Err(anyhow::anyhow!(
"This agent does not support completion-based prompts"
)))
}
fn retry(
&self,

2398
crates/agent/src/agent.rs
View file

File diff suppressed because it is too large Load diff

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

@ -54,7 +54,7 @@ use std::{
};
use util::path;
mod test_tools;
pub(crate) mod test_tools;
use test_tools::*;
/// Returns the fake provider that serves the registry's default model.
@ -4504,12 +4504,10 @@ async fn test_agent_connection(cx: &mut TestAppContext) {
release_dropped_entities(cx);
let result = cx
.update(|cx| {
acp_thread::AgentSessionClientUserMessageIds::prompt(
&connection,
acp_thread::ClientUserMessageId::new(),
acp_v2::PromptRequest::new(session_id.0.clone(), vec!["ghi".into()]),
cx,
)
connection
.receipt_submissions(&session_id, cx)
.expect("native receipts")
.prompt(vec!["ghi".into()], cx)
})
.await;
assert_eq!(
@ -5849,7 +5847,9 @@ async fn test_ask_user_elicitation_references_scoped_tool_call_id(cx: &mut TestA
});
cx.run_until_parked();
let send = acp_thread.update(cx, |thread, cx| thread.send_raw("Prompt", cx));
let send = cx.update(|cx| {
connection.send_and_wait_for_completion(acp_thread.clone(), vec!["Prompt".into()], cx)
});
cx.run_until_parked();
let ask_user_input = AskUserToolInput {
question: "Which directory should we explore?".to_string(),
@ -6053,7 +6053,9 @@ async fn test_subagent_tool_call_end_to_end(cx: &mut TestAppContext) {
});
cx.run_until_parked();
let send = acp_thread.update(cx, |thread, cx| thread.send_raw("Prompt", cx));
let send = cx.update(|cx| {
connection.send_and_wait_for_completion(acp_thread.clone(), vec!["Prompt".into()], cx)
});
cx.run_until_parked();
let request = fake.pending_completions_for(&model).pop().unwrap();
fake.send_text(&model, &request, "spawning subagent");
@ -6193,7 +6195,9 @@ async fn test_subagent_tool_output_does_not_include_thinking(cx: &mut TestAppCon
});
cx.run_until_parked();
let send = acp_thread.update(cx, |thread, cx| thread.send_raw("Prompt", cx));
let send = cx.update(|cx| {
connection.send_and_wait_for_completion(acp_thread.clone(), vec!["Prompt".into()], cx)
});
cx.run_until_parked();
let request = fake.pending_completions_for(&model).pop().unwrap();
fake.send_text(&model, &request, "spawning subagent");
@ -6350,7 +6354,9 @@ async fn test_subagent_tool_call_cancellation_during_task_prompt(cx: &mut TestAp
});
cx.run_until_parked();
let send = acp_thread.update(cx, |thread, cx| thread.send_raw("Prompt", cx));
let send = cx.update(|cx| {
connection.send_and_wait_for_completion(acp_thread.clone(), vec!["Prompt".into()], cx)
});
cx.run_until_parked();
let request = fake.pending_completions_for(&model).pop().unwrap();
fake.send_text(&model, &request, "spawning subagent");
@ -6483,7 +6489,9 @@ async fn test_subagent_tool_resume_session(cx: &mut TestAppContext) {
cx.run_until_parked();
// === First turn: create subagent ===
let send = acp_thread.update(cx, |thread, cx| thread.send_raw("First prompt", cx));
let send = cx.update(|cx| {
connection.send_and_wait_for_completion(acp_thread.clone(), vec!["First prompt".into()], cx)
});
cx.run_until_parked();
let request = fake.pending_completions_for(&model).pop().unwrap();
fake.send_text(&model, &request, "spawning subagent");
@ -6553,7 +6561,9 @@ async fn test_subagent_tool_resume_session(cx: &mut TestAppContext) {
});
// === Second turn: resume subagent with session_id ===
let send2 = acp_thread.update(cx, |thread, cx| thread.send_raw("Follow up", cx));
let send2 = cx.update(|cx| {
connection.send_and_wait_for_completion(acp_thread.clone(), vec!["Follow up".into()], cx)
});
cx.run_until_parked();
let request = fake.pending_completions_for(&model).pop().unwrap();
fake.send_text(&model, &request, "resuming subagent");
@ -7389,6 +7399,33 @@ async fn test_subagent_compaction_exhausts_transient_retries(cx: &mut TestAppCon
assert_eq!(test.fake.pending_completions(), Vec::new());
}
#[gpui::test]
async fn test_subagent_parent_cancellation_before_acceptance(cx: &mut TestAppContext) {
let test = SubagentCompactionTest::new(cx).await;
let send = cx.update(|cx| {
test.handle
.send("subagent task prompt".into(), &cx.to_async())
});
while test
.parent
.read_with(cx, |thread, cx| thread.running_subagent_ids(cx).is_empty())
{
assert!(cx.executor().tick(), "subagent registers with its parent");
}
assert!(
test.thread
.read_with(cx, |thread, _| thread.last_user_message_id().is_none())
);
test.parent.update(cx, |thread, cx| thread.cancel(cx)).await;
cx.run_until_parked();
assert!(
test.fake.pending_completions().is_empty(),
"cancelled child must not start a provider request"
);
assert_eq!(send.await.unwrap_err().to_string(), "User canceled");
test.assert_stopped(cx);
}
#[gpui::test]
async fn test_subagent_compaction_parent_cancellation(cx: &mut TestAppContext) {
let test = SubagentCompactionTest::new(cx).await;
@ -7486,7 +7523,9 @@ async fn test_subagent_error_propagation(cx: &mut TestAppContext) {
cx.run_until_parked();
// Start the parent turn
let send = acp_thread.update(cx, |thread, cx| thread.send_raw("Prompt", cx));
let send = cx.update(|cx| {
connection.send_and_wait_for_completion(acp_thread.clone(), vec!["Prompt".into()], cx)
});
cx.run_until_parked();
let request = fake.pending_completions_for(&model).pop().unwrap();
fake.send_text(&model, &request, "spawning subagent");

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

@ -1274,6 +1274,7 @@ impl From<&ThreadModel> for Option<DbLanguageModel> {
#[derive(PartialEq)]
pub(crate) struct StreamingSaveKey {
message_count: usize,
retained_history_revision: usize,
title: Option<SharedString>,
summary: Option<SharedString>,
model: Option<DbLanguageModel>,
@ -1295,6 +1296,8 @@ pub struct Thread {
pending_summary_generation: Option<Shared<Task<Option<SharedString>>>>,
summary: Option<SharedString>,
messages: Vec<Arc<Message>>,
// Message count cannot detect replacements in frozen history while a successor streams.
retained_history_revision: usize,
user_store: Entity<UserStore>,
/// Holds the task that handles agent interaction until the end of the turn.
/// Survives across multiple requests as the model performs tool calls and
@ -1452,6 +1455,7 @@ impl Thread {
pending_summary_generation: None,
summary: None,
messages: Vec::new(),
retained_history_revision: 0,
user_store: project.read(cx).user_store(),
running_turn: None,
end_turn_at_next_boundary: false,
@ -1827,6 +1831,7 @@ impl Thread {
pending_summary_generation: None,
summary: db_thread.detailed_summary,
messages: db_thread.messages,
retained_history_revision: 0,
user_store: project.read(cx).user_store(),
running_turn: None,
end_turn_at_next_boundary: false,
@ -1979,6 +1984,7 @@ impl Thread {
pub(crate) fn streaming_save_key(&self) -> StreamingSaveKey {
StreamingSaveKey {
message_count: self.messages.len(),
retained_history_revision: self.retained_history_revision,
title: self.title.clone(),
summary: self.summary.clone(),
model: (&self.model).into(),
@ -2356,7 +2362,9 @@ impl Thread {
pub fn cancel(&mut self, cx: &mut Context<Self>) -> Task<()> {
for subagent in self.running_subagents.drain(..) {
if let Some(subagent) = subagent.upgrade() {
subagent.update(cx, |thread, cx| thread.cancel(cx)).detach();
subagent
.update(cx, |thread, cx| thread.request_cancel(cx))
.detach();
}
}
@ -2365,17 +2373,26 @@ impl Thread {
return Task::ready(());
};
let prompt_id = running_turn.prompt_id.clone();
let turn_task = running_turn.cancel();
cx.spawn(async move |this, cx| {
turn_task.await;
this.update(cx, |this, cx| {
this.flush_pending_message(cx);
if this.prompt_id == prompt_id {
this.flush_pending_message(cx);
}
})
.ok();
})
}
pub(crate) fn request_cancel(&mut self, cx: &mut Context<Self>) -> Task<()> {
// A native session can own work before there is a model turn to cancel.
cx.emit(CancellationRequested);
self.cancel(cx)
}
pub fn set_end_turn_at_next_boundary(&mut self, end_at_boundary: bool) {
self.end_turn_at_next_boundary = end_at_boundary;
}
@ -2462,6 +2479,7 @@ impl Thread {
cx: &mut Context<Self>,
) -> Result<()> {
self.cancel(cx).detach();
self.advance_prompt_id();
// Clear pending message since cancel will try to flush it asynchronously,
// and we don't want that content to be added after we truncate
self.pending_message.take();
@ -2584,6 +2602,7 @@ impl Thread {
&mut self,
cx: &mut Context<Self>,
) -> Result<mpsc::UnboundedReceiver<Result<ThreadEvent>>> {
self.advance_prompt_id();
self.messages.push(Arc::new(Message::Resume));
cx.notify();
@ -2606,13 +2625,44 @@ impl Thread {
let content = content.into_iter().map(Into::into).collect::<Arc<_>>();
log::debug!("Thread::send content: {:?}", content);
self.messages
.push(Arc::new(Message::User(UserMessage { id, content })));
cx.notify();
self.upsert_user_message(id, content, cx);
self.send_existing(cx)
}
pub(crate) fn upsert_user_message(
&mut self,
id: ClientUserMessageId,
content: Arc<[UserMessageContent]>,
cx: &mut Context<Self>,
) {
let message = Arc::new(Message::User(UserMessage {
id: id.clone(),
content,
}));
if let Some(existing) = self
.messages
.iter_mut()
.find(|message| matches!(message.as_ref(), Message::User(user) if user.id == id))
{
*existing = message;
self.retained_history_revision += 1;
} else {
self.messages.push(message);
}
cx.notify();
}
pub(crate) fn last_user_message_id(&self) -> Option<ClientUserMessageId> {
self.last_user_message().map(|message| message.id.clone())
}
pub(crate) fn has_user_message(&self, id: &ClientUserMessageId) -> bool {
self.messages
.iter()
.any(|message| matches!(message.as_ref(), Message::User(user) if &user.id == id))
}
pub fn send_existing(
&mut self,
cx: &mut Context<Self>,
@ -2645,7 +2695,23 @@ impl Thread {
self.flush_pending_message(cx);
self.cancel(cx).detach();
let compaction = self.forced_compaction_target_ix().map(|request_end_ix| {
let accepted_marker_index = self.messages.iter().position(
|message| matches!(message.as_ref(), Message::User(user) if user.id == id && user.content.is_empty()),
);
let request_end_index = accepted_marker_index.unwrap_or(self.messages.len());
let target = if accepted_marker_index.is_some_and(|index| {
matches!(
self.messages.get(index + 1).map(|message| message.as_ref()),
Some(Message::Compaction(_))
)
}) {
None
} else if accepted_marker_index.is_some() {
self.forced_compaction_target_ix_before(request_end_index)
} else {
self.forced_compaction_target_ix()
};
let compaction = target.map(|request_end_ix| {
self.advance_prompt_id();
let request = self.build_compaction_request(request_end_ix, &model, cx);
self.current_request_token_usage = TokenUsage::default();
@ -2664,6 +2730,7 @@ impl Thread {
let event_stream = ThreadEventStream::new(events_tx);
let (cancellation_tx, mut cancellation_rx) = watch::channel(false);
let task = cx.spawn({
let prompt_id = self.prompt_id.clone();
let event_stream = event_stream.clone();
async move |this, cx| {
let result = if let Some((model, request)) = compaction {
@ -2685,7 +2752,9 @@ impl Thread {
// (possibly for a new turn), so leave it alone.
if *cancellation_rx.borrow() {
this.update(cx, |this, _| {
this.emit_compaction_telemetry_outcome("canceled", None)
if this.prompt_id == prompt_id {
this.emit_compaction_telemetry_outcome("canceled", None);
}
})
.log_err();
return;
@ -2713,6 +2782,7 @@ impl Thread {
}
});
self.running_turn = Some(RunningTurn::new(
self.prompt_id.clone(),
event_stream,
BTreeMap::default(),
cancellation_tx,
@ -2722,6 +2792,13 @@ impl Thread {
Ok(events_rx)
}
pub(crate) fn prepare_local_turn(&mut self, cx: &mut Context<Self>) {
// Freeze the old partial before inserting a successor's user message,
// and fence detached cleanup while its checkpoint or command is prepared.
self.flush_pending_message(cx);
self.advance_prompt_id();
}
pub fn push_acp_user_block(
&mut self,
id: ClientUserMessageId,
@ -2733,9 +2810,7 @@ impl Thread {
.into_iter()
.map(|block| UserMessageContent::from_content_block(block, path_style))
.collect::<Result<Arc<_>>>()?;
self.messages
.push(Arc::new(Message::User(UserMessage { id, content })));
cx.notify();
self.upsert_user_message(id, content, cx);
Ok(())
}
@ -2782,13 +2857,21 @@ impl Thread {
let tools = self.enabled_tools(cx);
let (cancellation_tx, mut cancellation_rx) = watch::channel(false);
let task = cx.spawn({
let prompt_id = self.prompt_id.clone();
let user_message_id = self.last_user_message_id();
let event_stream = event_stream.clone();
async move |this, cx| {
log::debug!("Starting agent turn execution");
let turn_result =
Self::run_turn_internal(&this, &event_stream, cancellation_rx.clone(), cx)
.await;
let turn_result = Self::run_turn_internal(
&this,
&event_stream,
&prompt_id,
user_message_id.as_ref(),
cancellation_rx.clone(),
cx,
)
.await;
// Check if we were cancelled - if so, cancel() already took running_turn
// and we shouldn't touch it (it might be a NEW turn now)
@ -2825,13 +2908,21 @@ impl Thread {
_ = this.update(cx, |this, _| this.running_turn.take());
}
});
self.running_turn = Some(RunningTurn::new(event_stream, tools, cancellation_tx, task));
self.running_turn = Some(RunningTurn::new(
self.prompt_id.clone(),
event_stream,
tools,
cancellation_tx,
task,
));
Ok(events_rx)
}
async fn run_turn_internal(
this: &WeakEntity<Self>,
event_stream: &ThreadEventStream,
prompt_id: &PromptId,
user_message_id: Option<&ClientUserMessageId>,
mut cancellation_rx: watch::Receiver<bool>,
cx: &mut AsyncApp,
) -> Result<()> {
@ -2840,6 +2931,9 @@ impl Thread {
// Set when a refusal fallback occurs so subsequent iterations use the fallback model.
let mut refusal_fallback_model: Option<LanguageModel> = None;
loop {
if *cancellation_rx.borrow() {
return Ok(());
}
match Self::perform_compaction_if_needed(
this,
event_stream,
@ -2855,7 +2949,9 @@ impl Thread {
Ok(ControlFlow::Continue(())) => {}
Ok(ControlFlow::Break(())) => {
this.update(cx, |this, _| {
this.emit_compaction_telemetry_outcome("canceled", None)
if &this.prompt_id == prompt_id {
this.emit_compaction_telemetry_outcome("canceled", None);
}
})?;
return Ok(());
}
@ -2877,7 +2973,11 @@ impl Thread {
{
Ok(ControlFlow::Break(())) => {
this.update(cx, |this, _| {
this.emit_compaction_telemetry_outcome("canceled", None)
if &this.prompt_id == prompt_id {
this.emit_compaction_telemetry_outcome(
"canceled", None,
);
}
})?;
return Ok(());
}
@ -2917,6 +3017,9 @@ impl Thread {
}
}
if *cancellation_rx.borrow() {
return Ok(());
}
// Re-read the model and refresh tools on each iteration so that
// mid-turn changes (e.g. the user switches model, toggles tools,
// or changes profile) take effect between tool-call rounds.
@ -2962,7 +3065,14 @@ impl Thread {
let mut had_refusal = false;
loop {
// Race between getting the first event, tool completion, and cancellation.
let first_event = futures::select! {
let first_event = futures::select_biased! {
_ = cancellation_rx.changed().fuse() => {
if *cancellation_rx.borrow() {
cancelled = true;
break;
}
continue;
}
event = events.next().fuse() => event,
tool_result = futures::StreamExt::select_next_some(&mut tool_results) => {
let (owning_message_ix, tool_result) = tool_result;
@ -2988,13 +3098,6 @@ impl Thread {
}
continue;
}
_ = cancellation_rx.changed().fuse() => {
if *cancellation_rx.borrow() {
cancelled = true;
break;
}
continue;
}
};
let Some(first_event) = first_event else {
break;
@ -3066,6 +3169,9 @@ impl Thread {
// This prevents deadlock when the LLM stream ends (e.g. because of an error)
// before sending a tool use with `is_input_complete: true`.
this.update(cx, |this, _cx| {
if &this.prompt_id != prompt_id {
return;
}
if let Some(running_turn) = this.running_turn.as_mut() {
if running_turn.streaming_tool_inputs.is_empty() {
return;
@ -3133,20 +3239,38 @@ impl Thread {
let end_turn = tool_results.is_empty() && early_tool_results.is_empty();
for (owning_message_ix, tool_result) in early_tool_results {
Self::process_tool_result(this, event_stream, cx, owning_message_ix, tool_result)?;
Self::process_tool_result(
this,
event_stream,
prompt_id,
user_message_id,
cx,
owning_message_ix,
tool_result,
)?;
}
while let Some((owning_message_ix, tool_result)) = tool_results.next().await {
Self::process_tool_result(this, event_stream, cx, owning_message_ix, tool_result)?;
Self::process_tool_result(
this,
event_stream,
prompt_id,
user_message_id,
cx,
owning_message_ix,
tool_result,
)?;
}
this.update(cx, |this, cx| {
this.flush_pending_message(cx);
if this.title.is_none() {
this.generate_title(cx);
if &this.prompt_id == prompt_id {
this.flush_pending_message(cx);
if this.title.is_none() {
this.generate_title(cx);
}
}
})?;
if cancelled {
if cancelled || *cancellation_rx.borrow() {
log::debug!("Turn cancelled by user, exiting");
return Ok(());
}
@ -3166,6 +3290,9 @@ impl Thread {
ControlFlow::Break(_) => return Ok(()),
ControlFlow::Continue(_) => {}
}
if *cancellation_rx.borrow() {
return Ok(());
}
this.update(cx, |this, _cx| {
if let Some(Message::Agent(message)) = this.last_message() {
if message.tool_results.is_empty() {
@ -3276,8 +3403,7 @@ impl Thread {
let result: Result<ControlFlow<()>> = async {
let provider =
cx.update(|cx| LanguageModelRegistry::read_global(cx).provider_for_model(&model))?;
let stream = futures::select! {
result = provider.stream_completion(&model, request, cx).fuse() => result,
let stream = futures::select_biased! {
_ = cancellation_rx.changed().fuse() => {
if *cancellation_rx.borrow() {
log::debug!("Compaction cancelled before request started");
@ -3285,13 +3411,13 @@ impl Thread {
}
return Ok(ControlFlow::Continue(()));
}
result = provider.stream_completion(&model, request, cx).fuse() => result,
};
let mut stream = stream?;
let mut summary = String::new();
loop {
let event = futures::select! {
event = stream.next().fuse() => event,
let event = futures::select_biased! {
_ = cancellation_rx.changed().fuse() => {
if *cancellation_rx.borrow() {
log::debug!("Compaction cancelled while summarizing");
@ -3299,8 +3425,12 @@ impl Thread {
}
continue;
}
event = stream.next().fuse() => event,
};
if *cancellation_rx.borrow() {
return Ok(ControlFlow::Break(()));
}
let Some(event) = event else {
break;
};
@ -3357,10 +3487,7 @@ impl Thread {
}
}
CompactionInsertion::Manual { marker_id } => {
this.messages.push(Arc::new(Message::User(UserMessage {
id: marker_id,
content: Arc::from([]),
})));
this.upsert_user_message(marker_id, Arc::from([]), cx);
this.messages.push(compaction);
}
}
@ -3383,6 +3510,8 @@ impl Thread {
fn process_tool_result(
this: &WeakEntity<Thread>,
event_stream: &ThreadEventStream,
prompt_id: &PromptId,
user_message_id: Option<&ClientUserMessageId>,
cx: &mut AsyncApp,
owning_message_ix: usize,
tool_result: LanguageModelToolResult,
@ -3404,10 +3533,48 @@ impl Thread {
}
},
);
this.update(cx, |this, _cx| {
this.pending_message()
this.update(cx, |this, cx| {
// A cancelled turn can finish its tools after a successor has started.
if &this.prompt_id == prompt_id {
this.pending_message()
.tool_results
.insert(tool_result.tool_use_id.clone(), tool_result);
return;
}
let Some(user_message_id) = user_message_id else {
return;
};
let owning_user_message = this.messages.iter().take(owning_message_ix).rev().find_map(
|message| match message.as_ref() {
Message::User(user) => Some(&user.id),
_ => None,
},
);
if owning_user_message != Some(user_message_id) {
return;
}
let Some(message) = this.messages.get_mut(owning_message_ix) else {
return;
};
let Some(agent_message) = message.as_agent_message() else {
return;
};
if !agent_message.content.iter().any(|content| {
matches!(content, AgentMessageContent::ToolUse(tool_use)
if tool_use.id == tool_result.tool_use_id)
}) {
return;
}
// Keep late terminal output on its frozen owner, never on a reused
// message index after rewind or on the successor's pending message.
let mut agent_message = agent_message.clone();
agent_message
.tool_results
.insert(tool_result.tool_use_id.clone(), tool_result)
.insert(tool_result.tool_use_id.clone(), tool_result);
*message = Arc::new(Message::Agent(agent_message));
this.retained_history_revision += 1;
this.clear_summary();
cx.notify();
})?;
Ok(())
}
@ -4585,13 +4752,28 @@ impl Thread {
/// Insertion point for a manually-triggered compaction.
/// Returns `None` only when there is nothing to summarize (no messages, or the thread already ends in a compaction).
fn forced_compaction_target_ix(&self) -> Option<usize> {
self.forced_compaction_target_ix_before(self.messages.len())
}
fn forced_compaction_target_ix_before(&self, mut end_index: usize) -> Option<usize> {
while end_index > 0
&& matches!(
self.messages.get(end_index - 1).map(|message| message.as_ref()),
Some(Message::User(user)) if user.content.is_empty()
)
{
end_index -= 1;
}
if matches!(
self.messages.last().map(|message| &**message),
end_index
.checked_sub(1)
.and_then(|index| self.messages.get(index))
.map(|message| &**message),
None | Some(Message::Compaction(_))
) {
return None;
}
Some(self.messages.len())
Some(end_index)
}
fn compaction_model(&self, cx: &App) -> Option<LanguageModel> {
@ -4906,6 +5088,7 @@ enum CompactionInsertion {
}
struct RunningTurn {
prompt_id: PromptId,
/// Holds the task that handles agent interaction until the end of the turn.
/// Survives across multiple requests as the model performs tool calls and
/// we run tools, report their results.
@ -4926,12 +5109,14 @@ struct RunningTurn {
impl RunningTurn {
fn new(
prompt_id: PromptId,
event_stream: ThreadEventStream,
tools: BTreeMap<SharedString, Arc<dyn AnyAgentTool>>,
cancellation_tx: watch::Sender<bool>,
task: Task<()>,
) -> Self {
Self {
prompt_id,
_task: task,
event_stream,
tools,
@ -5080,6 +5265,10 @@ pub struct TokenUsageUpdated(pub Option<acp_thread::TokenUsage>);
impl EventEmitter<TokenUsageUpdated> for Thread {}
pub(crate) struct CancellationRequested;
impl EventEmitter<CancellationRequested> for Thread {}
pub struct TitleUpdated;
impl EventEmitter<TitleUpdated> for Thread {}

235
crates/agent_servers/src/e2e_tests.rs
View file

@ -1,10 +1,14 @@
use crate::{AgentServer, AgentServerDelegate};
use acp_thread::{AcpThread, AgentThreadEntry, ToolCallStatus};
use acp_thread::{
AcpThread, AcpThreadEvent, AgentThreadEntry, ForegroundActivity, SubmissionResponse,
ToolCallStatus,
};
use agent_client_protocol::schema::v2 as acp_v2;
use anyhow::{Context as _, Result, anyhow};
use client::RefreshLlmTokenListener;
use futures::{FutureExt, StreamExt, channel::mpsc, select};
use futures::{FutureExt, StreamExt, channel::mpsc, channel::oneshot, select};
use gpui::AppContext;
use gpui::{Entity, TestAppContext};
use gpui::{Entity, Task, TestAppContext};
use indoc::indoc;
use project::{FakeFs, Project};
#[cfg(test)]
@ -26,8 +30,7 @@ where
let project = Project::test(fs.clone(), [], cx).await;
let thread = new_test_thread(server(&fs, cx).await, project.clone(), "/private/tmp", cx).await;
thread
.update(cx, |thread, cx| thread.send_raw("Hello from Zed!", cx))
send_and_wait_for_turn(&thread, vec!["Hello from Zed!".into()], cx)
.await
.unwrap();
@ -67,21 +70,17 @@ where
.expect("failed to write file");
let project = Project::example([tempdir.path()], &mut cx.to_async()).await;
let thread = new_test_thread(server(&fs, cx).await, project.clone(), tempdir.path(), cx).await;
thread
.update(cx, |thread, cx| {
thread.send(
vec![
"Read the file ".into(),
acp_v2::ContentBlock::ResourceLink(acp_v2::ResourceLink::new(
"foo.rs", "foo.rs",
)),
" and tell me what the content of the println! is".into(),
],
cx,
)
})
.await
.unwrap();
send_and_wait_for_turn(
&thread,
vec![
"Read the file ".into(),
acp_v2::ContentBlock::ResourceLink(acp_v2::ResourceLink::new("foo.rs", "foo.rs")),
" and tell me what the content of the println! is".into(),
],
cx,
)
.await
.unwrap();
thread.read_with(cx, |thread, cx| {
assert!(matches!(
@ -122,15 +121,13 @@ where
let project = Project::example([tempdir.path()], &mut cx.to_async()).await;
let thread = new_test_thread(server(&fs, cx).await, project.clone(), "/private/tmp", cx).await;
thread
.update(cx, |thread, cx| {
thread.send_raw(
&format!("Read {} and tell me what you see.", foo_path.display()),
cx,
)
})
.await
.unwrap();
send_and_wait_for_turn(
&thread,
vec![format!("Read {} and tell me what you see.", foo_path.display()).into()],
cx,
)
.await
.unwrap();
thread.read_with(cx, |thread, _cx| {
assert!(thread.entries().iter().any(|entry| {
matches!(
@ -162,12 +159,11 @@ pub async fn test_tool_call_with_permission<T, F>(
let fs = init_test(cx).await as Arc<dyn fs::Fs>;
let project = Project::test(fs.clone(), [path!("/private/tmp").as_ref()], cx).await;
let thread = new_test_thread(server(&fs, cx).await, project.clone(), "/private/tmp", cx).await;
let full_turn = thread.update(cx, |thread, cx| {
thread.send_raw(
r#"Run exactly `touch hello.txt && echo "Hello, world!" | tee hello.txt` in the terminal."#,
cx,
)
});
let full_turn = send_and_wait_for_turn(
&thread,
vec![r#"Run exactly `touch hello.txt && echo "Hello, world!" | tee hello.txt` in the terminal."#.into()],
cx,
);
run_until_first_tool_call(
&thread,
@ -251,12 +247,11 @@ where
let project = Project::test(fs.clone(), [path!("/private/tmp").as_ref()], cx).await;
let thread = new_test_thread(server(&fs, cx).await, project.clone(), "/private/tmp", cx).await;
let _ = thread.update(cx, |thread, cx| {
thread.send_raw(
r#"Run exactly `touch hello.txt && echo "Hello, world!" | tee hello.txt` in the terminal."#,
cx,
)
});
let full_turn = send_and_wait_for_turn(
&thread,
vec![r#"Run exactly `touch hello.txt && echo "Hello, world!" | tee hello.txt` in the terminal."#.into()],
cx,
);
let first_tool_call_ix = run_until_first_tool_call(
&thread,
@ -283,6 +278,7 @@ where
});
thread.update(cx, |thread, cx| thread.cancel(cx)).await;
full_turn.await.unwrap();
thread.read_with(cx, |thread, _cx| {
let AgentThreadEntry::ToolCall(tool_call) = &thread.entries()[first_tool_call_ix] else {
panic!();
@ -290,12 +286,13 @@ where
assert_eq!(tool_call.status(), ToolCallStatus::Canceled);
});
thread
.update(cx, |thread, cx| {
thread.send_raw(r#"Stop running and say goodbye to me."#, cx)
})
.await
.unwrap();
send_and_wait_for_turn(
&thread,
vec!["Stop running and say goodbye to me.".into()],
cx,
)
.await
.unwrap();
thread.read_with(cx, |thread, _| {
assert!(matches!(
&thread.entries().last().unwrap(),
@ -313,8 +310,7 @@ where
let project = Project::test(fs.clone(), [], cx).await;
let thread = new_test_thread(server(&fs, cx).await, project.clone(), "/private/tmp", cx).await;
thread
.update(cx, |thread, cx| thread.send_raw("Hello from test!", cx))
send_and_wait_for_turn(&thread, vec!["Hello from test!".into()], cx)
.await
.unwrap();
@ -382,6 +378,59 @@ pub use common_e2e_tests;
// Helpers
pub fn send_and_wait_for_turn(
thread: &Entity<AcpThread>,
content: Vec<acp_v2::ContentBlock>,
cx: &mut TestAppContext,
) -> Task<Result<()>> {
cx.update(|cx| {
// These tests send sequential turns; otherwise a predecessor's stop could win this wait.
if thread.read(cx).foreground_activity() != ForegroundActivity::Idle
|| thread.read(cx).has_unsettled_submissions()
{
return Task::ready(Err(anyhow!(
"send_and_wait_for_turn requires an idle thread with no unsettled submissions"
)));
}
let generation = thread.read(cx).activity_generation();
let (sender, receiver) = oneshot::channel();
let mut sender = Some(sender);
// Native acceptance can precede completion, and completion can precede polling the receipt.
let subscription = cx.subscribe(thread, move |_, event, _| {
if let AcpThreadEvent::Stopped {
activity_generation,
stop_reason,
..
} = event
&& *activity_generation > generation
&& let Some(sender) = sender.take()
{
let result = match stop_reason {
Some(acp_v2::StopReason::Error(reason)) => Err(match &reason.error {
Some(error) => anyhow::Error::new((**error).clone()),
None => anyhow!("Agent stopped with an error"),
}),
_ => Ok(()),
};
if sender.send(result).is_err() {
log::debug!("Turn completion waiter was dropped");
}
}
});
let submission = thread.update(cx, |thread, cx| thread.send(content, cx));
cx.spawn(async move |_| {
let result = match submission.await? {
Some(SubmissionResponse::Accepted(_)) => receiver
.await
.context("Thread dropped before the turn stopped")?,
Some(SubmissionResponse::LegacyCompleted(_)) | None => Ok(()),
};
drop(subscription);
result
})
})
}
pub async fn init_test(cx: &mut TestAppContext) -> Arc<FakeFs> {
env_logger::try_init().ok();
@ -477,3 +526,91 @@ pub fn get_zed_path() -> PathBuf {
zed_path
}
#[cfg(test)]
mod tests {
use super::*;
use acp_thread::{AgentConnection, StubAgentConnection};
use agent_client_protocol::schema::v1 as acp_v1;
use std::rc::Rc;
async fn stub_thread(
receipt_submissions: bool,
cx: &mut TestAppContext,
) -> (Rc<StubAgentConnection>, Entity<AcpThread>) {
cx.update(|cx| {
let settings_store = settings::SettingsStore::test(cx);
cx.set_global(settings_store);
});
let project = Project::test(FakeFs::new(cx.executor()), [], cx).await;
let connection =
Rc::new(StubAgentConnection::new().with_receipt_submissions(receipt_submissions));
let thread = cx
.update(|cx| {
connection.clone().new_session(
project,
PathList::new(&[Path::new("/private/tmp")]),
cx,
)
})
.await
.expect("stub session");
(connection, thread)
}
#[gpui::test]
async fn test_e2e_helper_preserves_v1_completion(cx: &mut TestAppContext) {
let (connection, thread) = stub_thread(false, cx).await;
let response = connection.defer_next_prompt_response();
let mut completion = send_and_wait_for_turn(&thread, vec!["hello".into()], cx);
cx.run_until_parked();
assert!((&mut completion).now_or_never().is_none());
response
.send(Ok(acp_v1::PromptResponse::new(acp_v1::StopReason::EndTurn)))
.expect("prompt response");
completion.await.expect("v1 completion");
}
#[gpui::test]
async fn test_e2e_helper_preserves_stop_before_receipt(cx: &mut TestAppContext) {
let (connection, thread) = stub_thread(true, cx).await;
let receipt = connection.defer_next_receipt_response();
let completion = send_and_wait_for_turn(&thread, vec!["fail".into()], cx);
send_and_wait_for_turn(&thread, vec!["overlapping".into()], cx)
.await
.expect_err("an unsettled receipt must reject overlapping prompts");
thread.update(cx, |thread, cx| {
cx.emit(AcpThreadEvent::Stopped {
activity_generation: thread.activity_generation(),
activity_duration: None,
stop_reason: Some(acp_v2::StopReason::EndTurn),
});
thread
.update_session_state(
acp_v2::StateUpdate::Running(acp_v2::RunningStateUpdate::new()),
cx,
)
.expect("running");
thread
.update_session_state(
acp_v2::StateUpdate::Idle(
acp_v2::IdleStateUpdate::new()
.stop_reason(acp_v2::StopReason::Error(acp_v2::ErrorStopReason::new())),
),
cx,
)
.expect("stopped");
});
cx.run_until_parked();
receipt
.send(Ok(acp_v2::PromptResponse::new("accepted")))
.expect("receipt");
assert_eq!(
completion
.await
.expect_err("error without details")
.to_string(),
"Agent stopped with an error"
);
}
}

25
crates/eval_cli/src/main.rs
View file

@ -41,7 +41,7 @@ use std::time::{Duration, Instant};
use acp_thread::AgentConnection as _;
use agent::{NativeAgent, NativeAgentConnection, Templates, ThreadStore};
use agent_client_protocol::schema::{v1 as acp, v2 as acp_v2};
use agent_client_protocol::schema::v2 as acp_v2;
use anyhow::{Context, Result};
use clap::Parser;
use feature_flags::FeatureFlagAppExt as _;
@ -819,9 +819,8 @@ async fn run_agent(
instruction.to_string(),
))];
let send_future = acp_thread.update(cx, |acp_thread: &mut acp_thread::AcpThread, cx| {
acp_thread.send(message, cx)
});
let send_future =
cx.update(|cx| connection.send_and_wait_for_completion(acp_thread.clone(), message, cx));
let timeout_future = if let Some(timeout_secs) = timeout {
futures::future::Either::Left(
@ -843,21 +842,19 @@ async fn run_agent(
let outcome = select_biased! {
result = send_future.fuse() => match result {
Ok(Some(acp_thread::SubmissionResponse::LegacyCompleted(response))) => {
eprintln!("[eval-cli] stopped: {:?}", response.stop_reason);
if response.stop_reason == acp::StopReason::MaxTokens {
Ok(stop_reason) => {
eprintln!("[eval-cli] stopped: {:?}", stop_reason);
if let acp_v2::StopReason::Error(details) = stop_reason {
Err(details
.error
.map(|error| anyhow::Error::new(*error))
.unwrap_or_else(|| anyhow::anyhow!("Agent stopped because of an error")))
} else if stop_reason == acp_v2::StopReason::MaxTokens {
Err(anyhow::anyhow!("Model hit maximum token limit"))
} else {
Ok(AgentOutcome::Completed)
}
}
Ok(Some(acp_thread::SubmissionResponse::Accepted(_))) => {
Err(anyhow::anyhow!("Native agent returned acceptance instead of turn completion"))
}
Ok(None) => {
eprintln!("[eval-cli] completed (no response)");
Ok(AgentOutcome::Completed)
}
Err(e) => Err(e).context("agent run failed"),
},
_ = sigterm_future.fuse() => {