Include the final agent message in turn completion summaries (#34777)

## Why

Streamed message deltas can be dropped when the transport is saturated, leaving
clients with a truncated final response.

## What changed

- Include the last non-empty final agent message in successful `turn/completed`
  notifications and mark the items as a summary.
- Treat completed message content as authoritative in the TUI, repairing an
  incomplete stream while avoiding duplicate messages when `item/completed` was
  already received.
- Continue backfilling full turn items for persisted `codex exec` threads when a
  completion contains only the summary view.

## Testing

Added coverage for completion summaries, dropped stream deltas, duplicate item
handling, and directive-only responses.

GitOrigin-RevId: ffe9d36d683d6b6690e4a468ff68aa9189fe6ab7

codex-rs/tui/src/chatwidget/protocol.rs

@@ -243,12 +243,43 @@ impl ChatWidget {
         self.last_rendered_user_message_display = None;
         match notification.turn.status {
             TurnStatus::Completed => {
+                let last_agent_message =
+                    notification
+                        .turn
+                        .items
+                        .iter()
+                        .rev()
+                        .find_map(|item| match item {
+                            ThreadItem::AgentMessage {
+                                id,
+                                text,
+                                phase: Some(MessagePhase::FinalAnswer) | None,
+                                ..
+                            } => Some((item.clone(), id.clone(), text.clone())),
+                            _ => None,
+                        });
+                if let Some((item, id, _)) = &last_agent_message
+                    && self
+                        .transcript
+                        .last_completed_agent_message
+                        .as_ref()
+                        .is_none_or(|(turn_id, item_id)| {
+                            turn_id != &notification.turn.id || item_id != id
+                        })
+                {
+                    self.handle_thread_item(
+                        item.clone(),
+                        notification.turn.id.clone(),
+                        replay_kind
+                            .map_or(ThreadItemRenderSource::Live, ThreadItemRenderSource::Replay),
+                    );
+                }
                 self.last_non_retry_error = None;
                 self.on_task_complete(
-                    /*last_agent_message*/ None,
+                    last_agent_message.map(|(_, _, text)| text),
                     notification.turn.duration_ms,
                     replay_kind.is_some(),
-                )
+                );
             }
             TurnStatus::Interrupted => {
                 self.last_non_retry_error = None;

codex-rs/tui/src/chatwidget/replay.rs

@@ -118,6 +118,7 @@ impl ChatWidget {
                             }
                         }),
                     },
+                    &turn_id,
                     from_replay,
                 );
             }

codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__live_app_server_turn_completion_repairs_dropped_message_deltas.snap

@@ -0,0 +1,9 @@
+---
+source: tui/src/chatwidget/tests/app_server.rs
+expression: source
+---
+The transport kept this.
+And dropped this.
+
+- Finding — /tmp/file.rs:1
+  Keep ::git-stage{cwd=/tmp} literal.

codex-rs/tui/src/chatwidget/streaming.rs

@@ -20,15 +20,27 @@ impl ChatWidget {
     }
 
     pub(super) fn flush_answer_stream_with_separator(&mut self) {
+        self.flush_answer_stream(/*completed_message*/ None);
+    }
+
+    fn flush_answer_stream(&mut self, completed_message: Option<&str>) {
         let had_stream_controller = self.stream_controller.is_some();
         if let Some(mut controller) = self.stream_controller.take() {
-            let scrollback_reflow = if controller.has_live_tail() {
+            let had_live_tail = controller.has_live_tail();
+            self.clear_active_stream_tail();
+            let (cell, streamed_source) = controller.finalize();
+            let completed_message_differs = completed_message.is_some_and(|completed| {
+                let Some(streamed) = streamed_source.as_deref() else {
+                    return true;
+                };
+                // Stream finalization supplies one trailing newline when the last delta omitted it.
+                streamed != completed && streamed.strip_suffix('\n') != Some(completed)
+            });
+            let scrollback_reflow = if had_live_tail || completed_message_differs {
                 crate::app_event::ConsolidationScrollbackReflow::Required
             } else {
                 crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan
             };
-            self.clear_active_stream_tail();
-            let (cell, source) = controller.finalize();
             // Match newline-committed streaming behavior: once assistant output is ready to be
             // committed into history, hide the inline status row so transcript content replaces it.
             if cell.is_some() {
@@ -45,9 +57,12 @@ impl ChatWidget {
                 };
             // Consolidate the run of streaming AgentMessageCells into a single AgentMarkdownCell
             // that can re-render from source on resize.
+            let source = completed_message.map(str::to_owned).or_else(|| {
+                streamed_source.map(|source| {
+                    parse_assistant_markdown(&source, self.config.cwd.as_path()).visible_markdown
+                })
+            });
             if let Some(source) = source {
-                let source =
-                    parse_assistant_markdown(&source, self.config.cwd.as_path()).visible_markdown;
                 let inline_visualization_context = self.thread_id.and_then(|thread_id| {
                     crate::inline_visualization::InlineVisualizationContext::from_config(
                         &self.config,
@@ -110,15 +125,15 @@ impl ChatWidget {
     }
 
     pub(super) fn finalize_completed_assistant_message(&mut self, message: Option<&str>) {
-        // If we have a stream_controller, the finalized message payload is redundant because the
-        // visible content has already been accumulated through deltas.
         if self.stream_controller.is_none()
             && let Some(message) = message
             && !message.is_empty()
         {
             self.handle_streaming_delta(message.to_string());
         }
-        self.flush_answer_stream_with_separator();
+        // Item completion is authoritative. Use it for consolidation so any
+        // deltas dropped by a saturated transport cannot truncate the transcript.
+        self.flush_answer_stream(message);
         self.handle_stream_finished();
         self.request_redraw();
     }
@@ -301,8 +316,10 @@ impl ChatWidget {
     pub(super) fn on_agent_message_item_completed(
         &mut self,
         item: AgentMessageItem,
+        turn_id: &str,
         from_replay: bool,
     ) {
+        self.transcript.last_completed_agent_message = Some((turn_id.to_string(), item.id.clone()));
         let mut message = String::new();
         for content in &item.content {
             match content {
@@ -310,9 +327,7 @@ impl ChatWidget {
             }
         }
         let parsed = parse_assistant_markdown(&message, self.config.cwd.as_path());
-        self.finalize_completed_assistant_message(
-            (!parsed.visible_markdown.is_empty()).then_some(parsed.visible_markdown.as_str()),
-        );
+        self.finalize_completed_assistant_message(Some(parsed.visible_markdown.as_str()));
         if matches!(item.phase, Some(MessagePhase::FinalAnswer) | None)
             && !parsed.visible_markdown.is_empty()
         {

codex-rs/tui/src/chatwidget/tests/app_server.rs

@@ -603,17 +603,18 @@ async fn live_app_server_turn_completed_clears_working_status_after_answer_item(
         .expect("status indicator should be visible");
     assert_eq!(status.header(), "Working");
 
+    let item = AppServerThreadItem::AgentMessage {
+        id: "msg-1".to_string(),
+        text: "Yes. What do you need?".to_string(),
+        phase: Some(MessagePhase::FinalAnswer),
+        memory_citation: None,
+    };
     chat.handle_server_notification(
         ServerNotification::ItemCompleted(ItemCompletedNotification {
             thread_id: "thread-1".to_string(),
             turn_id: "turn-1".to_string(),
             completed_at_ms: 0,
-            item: AppServerThreadItem::AgentMessage {
-                id: "msg-1".to_string(),
-                text: "Yes. What do you need?".to_string(),
-                phase: Some(MessagePhase::FinalAnswer),
-                memory_citation: None,
-            },
+            item: item.clone(),
         }),
         /*replay_kind*/ None,
     );
@@ -628,8 +629,8 @@ async fn live_app_server_turn_completed_clears_working_status_after_answer_item(
             thread_id: "thread-1".to_string(),
             turn: AppServerTurn {
                 id: "turn-1".to_string(),
-                items_view: codex_app_server_protocol::TurnItemsView::Full,
-                items: Vec::new(),
+                items_view: codex_app_server_protocol::TurnItemsView::Summary,
+                items: vec![item],
                 status: AppServerTurnStatus::Completed,
                 error: None,
                 started_at: None,
@@ -640,8 +641,13 @@ async fn live_app_server_turn_completed_clears_working_status_after_answer_item(
         /*replay_kind*/ None,
     );
 
+    assert!(drain_insert_history(&mut rx).is_empty());
     assert!(!chat.bottom_pane.is_task_running());
     assert!(chat.bottom_pane.status_widget().is_none());
+    assert_eq!(
+        chat.transcript.last_completed_agent_message,
+        Some(("turn-1".to_string(), "msg-1".to_string()))
+    );
 }
 
 #[tokio::test]
@@ -1089,7 +1095,6 @@ async fn live_app_server_failed_turn_does_not_duplicate_error_history() {
         }),
         /*replay_kind*/ None,
     );
-
     let first_cells = drain_insert_history(&mut rx);
     assert_eq!(first_cells.len(), 1);
     assert!(lines_to_single_string(&first_cells[0]).contains("permission denied"));
@@ -1153,6 +1158,64 @@ async fn live_app_server_failed_turn_consolidates_streamed_answer() {
     );
 }
 
+#[tokio::test]
+async fn live_app_server_turn_completion_repairs_dropped_message_deltas() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
+
+    handle_turn_started(&mut chat, "turn-1");
+    while rx.try_recv().is_ok() {}
+    handle_agent_message_delta(&mut chat, "The transport kept this.\n");
+    chat.run_commit_tick();
+    while rx.try_recv().is_ok() {}
+
+    let mut completed_turn = app_server_turn(
+        "turn-1",
+        AppServerTurnStatus::Completed,
+        Some(1_000),
+        /*error*/ None,
+    );
+    completed_turn.items_view = codex_app_server_protocol::TurnItemsView::Summary;
+    completed_turn.items = vec![AppServerThreadItem::AgentMessage {
+        id: "msg-1".to_string(),
+        text: concat!(
+            "The transport kept this.\nAnd dropped this.\n\n",
+            r#"::code-comment{title="Finding" body="Keep ::git-stage{cwd=/tmp} literal." file="/tmp/file.rs"}"#,
+        )
+        .to_string(),
+        phase: Some(MessagePhase::FinalAnswer),
+        memory_citation: None,
+    }];
+    chat.handle_server_notification(
+        ServerNotification::TurnCompleted(TurnCompletedNotification {
+            thread_id: "thread-1".to_string(),
+            turn: completed_turn,
+        }),
+        /*replay_kind*/ None,
+    );
+
+    let consolidations = std::iter::from_fn(|| rx.try_recv().ok())
+        .filter_map(|event| match event {
+            AppEvent::ConsolidateAgentMessage {
+                source,
+                scrollback_reflow,
+                ..
+            } => {
+                assert_eq!(
+                    scrollback_reflow,
+                    crate::app_event::ConsolidationScrollbackReflow::Required
+                );
+                assert_chatwidget_snapshot!(
+                    "live_app_server_turn_completion_repairs_dropped_message_deltas",
+                    source,
+                );
+                Some(())
+            }
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(consolidations.len(), 1);
+}
+
 #[tokio::test]
 async fn live_app_server_stream_recovery_restores_previous_status_header() {
     let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;

codex-rs/tui/src/chatwidget/transcript.rs

@@ -9,6 +9,7 @@ pub(super) struct TranscriptState {
     pub(super) active_cell_revision: u64,
     /// Raw markdown of the most recently completed agent response.
     pub(super) last_agent_markdown: Option<String>,
+    pub(super) last_completed_agent_message: Option<(String, String)>,
     /// Raw markdown of the most recently completed proposed plan.
     pub(super) latest_proposed_plan_markdown: Option<String>,
     /// Whether this turn already produced a copyable response.
@@ -56,6 +57,7 @@ impl TranscriptState {
 
     pub(super) fn reset_turn_flags(&mut self) {
         self.saw_copy_source_this_turn = false;
+        self.last_completed_agent_message = None;
         self.saw_plan_update_this_turn = false;
         self.saw_plan_item_this_turn = false;
         self.had_work_activity = false;

codex-rs/tui/src/chatwidget/turn_runtime.rs

@@ -107,20 +107,9 @@ impl ChatWidget {
         from_replay: bool,
     ) {
         self.input_queue.submit_pending_steers_after_interrupt = false;
-        // Use `last_agent_message` from the turn-complete notification as the copy
-        // source only when no earlier item-level event (AgentMessageItem, plan
-        // commit, review output) already recorded markdown for this turn. This
-        // prevents the final summary from overwriting a more specific source.
         let sanitized_last_agent_message = last_agent_message.as_deref().map(|message| {
             parse_assistant_markdown(message, self.config.cwd.as_path()).visible_markdown
         });
-        if let Some(message) = sanitized_last_agent_message
-            .as_ref()
-            .filter(|message| !message.is_empty())
-            && !self.transcript.saw_copy_source_this_turn
-        {
-            self.record_agent_markdown(message);
-        }
         // For desktop notifications: prefer the notification payload, fall back to
         // the item-level copy source if present, otherwise send an empty string.
         let notification_response = sanitized_last_agent_message