agent_ui: Guard follow-up sends from stale send results (#64917)

Keep an older agent send's response from mutating the UI state of a
newer started turn.

After cancel/send, the old RPC response may still be outstanding. The
core already guards its own turn completion, but the view previously
cleared `in_flight_prompt` on every successful response and displayed
every returned error, even when a successor was running.

- Apply prompt cleanup, following-state refresh, and response errors
inside one view update guarded by the existing turn generation.
- Clear the previous displayed error again at actual turn start. This
covers an old error arriving while the successor's contents are still
resolving, without starting turn timing early.
- Preserve current-send error visibility, prompt recovery,
generation-protected timer cleanup, and per-send completion telemetry.
Superseded failures are debug-logged rather than shown on the newer
turn.
- Add a test-only controlled prompt-response gate and one regression
covering late success/error, deferred preparation, current success/error
cleanup, and timer ownership.

The guarantee starts when the successor reaches `start_turn`.
Cancellation alone does not invalidate the view generation; an old error
can appear while successor content is preparing, but cannot carry into
that new turn.

Release Notes:

- agent: Fixed stale agent responses clearing a newer prompt or showing
errors from an earlier send.

crates/acp_thread/src/connection.rs

@@ -767,6 +767,7 @@ mod test_support {
         sessions: Arc<Mutex<HashMap<acp_v1::SessionId, Session>>>,
         permission_requests: HashMap<acp_v1::ToolCallId, PermissionOptions>,
         next_prompt_updates: Arc<Mutex<Vec<acp_v1::SessionUpdate>>>,
+        next_prompt_response: Arc<Mutex<Option<oneshot::Receiver<Result<acp_v1::PromptResponse>>>>>,
         next_truncate: Arc<Mutex<Option<oneshot::Receiver<()>>>>,
         supports_load_session: bool,
         supports_session_additional_directories: bool,
@@ -789,6 +790,7 @@ mod test_support {
         pub fn new() -> Self {
             Self {
                 next_prompt_updates: Default::default(),
+                next_prompt_response: Default::default(),
                 next_truncate: Default::default(),
                 permission_requests: HashMap::default(),
                 sessions: Arc::default(),
@@ -803,6 +805,14 @@ mod test_support {
             *self.next_prompt_updates.lock() = updates;
         }
 
+        pub fn defer_next_prompt_response(
+            &self,
+        ) -> oneshot::Sender<Result<acp_v1::PromptResponse>> {
+            let (sender, receiver) = oneshot::channel();
+            assert!(self.next_prompt_response.lock().replace(receiver).is_none());
+            sender
+        }
+
         pub fn defer_next_truncate(&self) -> oneshot::Sender<()> {
             let (sender, receiver) = oneshot::channel();
             assert!(self.next_truncate.lock().replace(receiver).is_none());
@@ -986,7 +996,9 @@ mod test_support {
                 response_tx,
             } = sessions.get_mut(&params.session_id).unwrap();
             let mut tasks = vec![];
-            if self.next_prompt_updates.lock().is_empty() {
+            if let Some(receiver) = self.next_prompt_response.lock().take() {
+                cx.spawn(async move |_| receiver.await?)
+            } else if self.next_prompt_updates.lock().is_empty() {
                 let (tx, rx) = oneshot::channel();
                 response_tx.replace(tx);
                 cx.spawn(async move |_| {

crates/agent_ui/src/conversation_view.rs

@@ -4315,6 +4315,117 @@ pub(crate) mod tests {
         });
     }
 
+    #[gpui::test]
+    async fn test_stale_send_result_preserves_new_prompt(cx: &mut TestAppContext) {
+        init_test(cx);
+
+        for (responses_are_errors, defer_contents) in [(false, false), (true, false), (true, true)]
+        {
+            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 thread_view = active_thread(&conversation_view, cx);
+            let message_editor = message_editor(&conversation_view, cx);
+
+            let finish_first = connection.defer_next_prompt_response();
+            message_editor.update_in(cx, |editor, window, cx| {
+                editor.set_text("first", window, cx);
+            });
+            thread_view.update_in(cx, |view, window, cx| view.send(window, cx));
+            cx.run_until_parked();
+
+            thread_view.update_in(cx, |view, _window, cx| view.cancel_generation(cx));
+            cx.run_until_parked();
+            let finish_second = connection.defer_next_prompt_response();
+            message_editor.update_in(cx, |editor, window, cx| {
+                editor.set_text("second", window, cx);
+            });
+            let finish_contents = if defer_contents {
+                let (sender, receiver) = futures::channel::oneshot::channel();
+                thread_view.update_in(cx, |view, window, cx| {
+                    let contents = cx.spawn(async move |_, _| {
+                        receiver.await?;
+                        anyhow::Ok(Some((vec!["second".into()], Vec::new())))
+                    });
+                    view.send_content(contents, false, window, cx);
+                });
+                Some(sender)
+            } else {
+                thread_view.update_in(cx, |view, window, cx| view.send(window, cx));
+                None
+            };
+            cx.run_until_parked();
+
+            thread_view.read_with(cx, |view, cx| {
+                if defer_contents {
+                    assert!(view.is_loading_contents);
+                    assert_eq!(view.thread.read(cx).status(), ThreadStatus::Idle);
+                } else {
+                    assert_eq!(view.thread.read(cx).status(), ThreadStatus::Generating);
+                    assert!(matches!(
+                        view.in_flight_prompt.as_deref(),
+                        Some([acp_v1::ContentBlock::Text(text)]) if text.text == "second"
+                    ));
+                }
+                assert!(view.thread_error.is_none());
+            });
+
+            let first_response = if responses_are_errors {
+                Err(anyhow!("stale prompt failed"))
+            } else {
+                Ok(acp_v1::PromptResponse::new(acp_v1::StopReason::EndTurn))
+            };
+            finish_first
+                .send(first_response)
+                .expect("first prompt response should still be pending");
+            cx.run_until_parked();
+            if let Some(finish_contents) = finish_contents {
+                finish_contents
+                    .send(())
+                    .expect("second contents are pending");
+                cx.run_until_parked();
+            }
+
+            thread_view.read_with(cx, |view, cx| {
+                assert_eq!(view.thread.read(cx).status(), ThreadStatus::Generating);
+                assert!(view.turn_fields.turn_started_at.is_some());
+                assert!(matches!(
+                    view.in_flight_prompt.as_deref(),
+                    Some([acp_v1::ContentBlock::Text(text)]) if text.text == "second"
+                ));
+                assert!(view.thread_error.is_none());
+            });
+
+            let second_response = if responses_are_errors {
+                Err(anyhow!("current prompt failed"))
+            } else {
+                Ok(acp_v1::PromptResponse::new(acp_v1::StopReason::EndTurn))
+            };
+            finish_second
+                .send(second_response)
+                .expect("second prompt response should be pending");
+            cx.run_until_parked();
+            thread_view.read_with(cx, |view, cx| {
+                assert_eq!(view.thread.read(cx).status(), ThreadStatus::Idle);
+                assert!(view.turn_fields.turn_started_at.is_none());
+                if responses_are_errors {
+                    assert!(matches!(
+                        &view.thread_error,
+                        Some(ThreadError::Other { message, .. }) if message.as_ref() == "current prompt failed"
+                    ));
+                    assert!(matches!(
+                        view.in_flight_prompt.as_deref(),
+                        Some([acp_v1::ContentBlock::Text(text)]) if text.text == "second"
+                    ));
+                } else {
+                    assert!(view.thread_error.is_none());
+                    assert!(view.in_flight_prompt.is_none());
+                }
+            });
+        }
+    }
+
     #[gpui::test]
     async fn test_queue_resumes_after_stop_and_new_message(cx: &mut TestAppContext) {
         init_test(cx);

crates/agent_ui/src/conversation_view/thread_view.rs

@@ -1402,6 +1402,8 @@ impl ThreadView {
     // turns
 
     pub fn start_turn(&mut self, cx: &mut Context<Self>) -> usize {
+        // A previous response may have settled while the new prompt's contents were loading.
+        self.thread_error.take();
         self.turn_fields.turn_generation += 1;
         let generation = self.turn_fields.turn_generation;
         self.turn_fields.turn_started_at = Some(Instant::now());
@@ -1783,12 +1785,7 @@ impl ThreadView {
             let res = send.await;
             let turn_time_ms = turn_start_time.elapsed().as_millis();
             drop(_stop_turn);
-            let status = if res.is_ok() {
-                let _ = this.update(cx, |this, _| this.in_flight_prompt.take());
-                "success"
-            } else {
-                "failure"
-            };
+            let status = if res.is_ok() { "success" } else { "failure" };
             telemetry::event!(
                 "Agent Turn Completed",
                 agent = agent_telemetry_id,
@@ -1800,7 +1797,28 @@ impl ThreadView {
                 turn_time_ms,
                 side = side
             );
-            res.map(|_| ())
+            this.update(cx, |this, cx| {
+                // Cancellation can return control to the UI before this response settles.
+                if this.turn_fields.turn_generation != generation {
+                    if let Err(error) = res {
+                        log::debug!("Ignoring error from a superseded agent send: {error:#}");
+                    }
+                    return;
+                }
+                match res {
+                    Ok(_) => {
+                        this.in_flight_prompt.take();
+                        this.should_be_following = this
+                            .workspace
+                            .update(cx, |workspace, _| {
+                                workspace.is_being_followed(CollaboratorId::Agent)
+                            })
+                            .unwrap_or_default();
+                    }
+                    Err(error) => this.handle_thread_error(error, cx),
+                }
+            })?;
+            anyhow::Ok(())
         });
 
         cx.spawn(async move |this, cx| {
@@ -1809,17 +1827,6 @@ impl ThreadView {
                     this.handle_thread_error(err, cx);
                 })
                 .ok();
-            } else {
-                this.update(cx, |this, cx| {
-                    let should_be_following = this
-                        .workspace
-                        .update(cx, |workspace, _| {
-                            workspace.is_being_followed(CollaboratorId::Agent)
-                        })
-                        .unwrap_or_default();
-                    this.should_be_following = should_be_following;
-                })
-                .ok();
             }
         })
         .detach();