From 6107b9886c114702e5c1cfd1c1b54f7c48bfaf49 Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Mon, 9 Nov 2026 02:50:42 +0000 Subject: [PATCH] Avoid duplicate assistant messages when resuming Cloud threads (#60335) ## Why After Cloud thread history is replayed in the TUI, notifications for the same completed assistant items can render their text again. ## What changed Track completed assistant items rendered from Cloud history by turn and item ID, and ignore subsequent starts, text deltas, and completions for those items. Apply this tracking when resuming threads and replaying thread snapshots, while allowing unfinished items and new turns to continue streaming. ## Testing Extend the Cloud resume picker test to cover repeated notifications for completed messages, continued streaming in an active turn, and responses in a later turn. Retain coverage for queued user message reconciliation without resubmission. GitOrigin-RevId: 1a829a2964586d876339c7c759022e4d366c8b2e --- .../app/tests/session_lifecycle_requests.rs | 125 +++++++++++++++--- codex-rs/tui/src/app/thread_routing.rs | 14 +- codex-rs/tui/src/chatwidget/protocol.rs | 8 +- codex-rs/tui/src/chatwidget/replay.rs | 17 ++- codex-rs/tui/src/chatwidget/streaming.rs | 10 ++ codex-rs/tui/src/chatwidget/transcript.rs | 11 ++ 6 files changed, 163 insertions(+), 22 deletions(-) diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index d8c96a4942..3040e2d921 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -206,12 +206,11 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { use crate::app_server_session::user_message_submission::UserMessageAdmission; use codex_app_server_client::CloudOperation; use codex_app_server_protocol::ItemCompletedNotification; + use codex_protocol::models::MessagePhase; let (mut app, mut events, mut ops) = make_test_app_with_channels().await; let [original, other] = [ThreadId::new(), ThreadId::new()]; - app.app_server_target = AppServerTarget::Remote { - endpoint: crate::resolve_remote_addr("ws://127.0.0.1:1")?, - }; + app.app_server_target = AppServerTarget::Cloud; app.enqueue_primary_thread_session( test_thread_session(original, app.config.cwd.to_path_buf()), Vec::new(), @@ -248,6 +247,31 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { text_elements: Vec::new(), }], }; + let agent_item = |id: &str, text: &str, phase| ThreadItemPayload::AgentMessage { + moderation: None, + id: id.into(), + text: text.into(), + phase, + memory_citation: None, + delivery: None, + questions: None, + }; + let stored = agent_item("stored-item", "Canonical response", None); + let commentary = agent_item( + "commentary", + "Commentary response", + Some(MessagePhase::Commentary), + ); + let ongoing = "ongoing-turn"; + let mut ongoing_turn = test_turn( + ongoing, + TurnStatus::InProgress, + vec![ + commentary.clone(), + agent_item("live-item", "Partial response", None), + ], + ); + ongoing_turn.items[0].completed_at_ms = Some(1); let mut original_thread = overview_thread( original, /*parent_thread_id*/ None, @@ -258,15 +282,18 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { original_thread.turns = vec![test_turn( turn_id, TurnStatus::Completed, - vec![user_item(confirmed_id)], + vec![user_item(confirmed_id), stored.clone()], )]; let mut other_thread = overview_thread( other, /*parent_thread_id*/ None, "Other", - ThreadStatus::Idle, + ThreadStatus::Active { + active_flags: Vec::new(), + }, ); other_thread.cwd = app.config.cwd.clone(); + other_thread.turns = vec![ongoing_turn]; let (client, mut commands, _server_events) = codex_app_server_client::CloudClient::channel(/*capacity*/ 1); @@ -340,6 +367,24 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { resumed_threads }); + let completed = |thread_id: ThreadId, turn_id: &str, item| { + ServerNotification::ItemCompleted(ItemCompletedNotification::new( + thread_id.to_string(), + turn_id.into(), + item, + /*started_at_ms*/ None, + /*completed_at_ms*/ 1, + )) + }; + let rendered_responses = |app: &App| { + app.transcript_cells + .iter() + .flat_map(|cell| cell.display_lines(/*width*/ 80)) + .map(|line| line.to_string()) + .filter(|line| line.contains("response")) + .collect::>() + .join("\n") + }; let mut tui = crate::tui::test_support::make_test_tui()?; while events.try_recv().is_ok() {} for thread_id in [other, original] { @@ -355,7 +400,36 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { }), ) .await?; + drain_app_events(&mut app, &mut tui, &mut server, &mut events).await?; assert_eq!(app.chat_widget.thread_id(), Some(thread_id)); + if thread_id == other { + insta::assert_snapshot!(rendered_responses(&app), @" + • Commentary response + • Partial response + "); + for notification in [ + agent_message_delta_notification(other, ongoing, "live-item", "Live response "), + agent_message_delta_notification( + other, + ongoing, + "commentary", + "Commentary response", + ), + completed(other, ongoing, commentary.clone()), + agent_message_delta_notification(other, ongoing, "live-item", "continues"), + turn_completed_notification(other, ongoing, TurnStatus::Completed), + ] { + app.enqueue_thread_notification(other, notification).await?; + let event = app.active_thread_rx.as_mut().unwrap().recv().await.unwrap(); + app.handle_thread_event_now(event); + } + drain_app_events(&mut app, &mut tui, &mut server, &mut events).await?; + insta::assert_snapshot!(rendered_responses(&app), @" + • Commentary response + • Partial response + • Live response continues + "); + } } assert_eq!(app.chat_widget.queued_user_message_texts(), ["same text"]); @@ -364,21 +438,40 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { .capture_thread_input_state() .is_some_and(|input| input.has_unconfirmed_messages()) ); - assert!( - !std::iter::from_fn(|| events.try_recv().ok()) - .any(|event| { matches!(event, AppEvent::CodexOp(Op::UserTurn { .. })) }) - ); app.chat_widget.handle_server_notification( - ServerNotification::ItemCompleted(ItemCompletedNotification::new( - original.to_string(), - turn_id.into(), - user_item(unmatched_id), - /*started_at_ms*/ None, - /*completed_at_ms*/ 1, - )), + completed(original, turn_id, user_item(unmatched_id)), /*replay_kind*/ None, ); assert!(app.chat_widget.queued_user_message_texts().is_empty()); + insta::assert_snapshot!(rendered_responses(&app), @"• Canonical response"); + + for notification in [ + ServerNotification::ItemStarted(ItemStartedNotification::new( + original.to_string(), + turn_id.into(), + stored.clone(), + /*started_at_ms*/ 1, + )), + agent_message_delta_notification(original, turn_id, "stored-item", "Canonical response"), + completed(original, turn_id, stored), + turn_completed_notification(original, turn_id, TurnStatus::Completed), + turn_started_notification(original, "later-turn"), + agent_message_delta_notification(original, "later-turn", "later-item", "New response"), + turn_completed_notification(original, "later-turn", TurnStatus::Completed), + ] { + app.enqueue_thread_notification(original, notification) + .await?; + let event = app.active_thread_rx.as_mut().unwrap().recv().await.unwrap(); + app.handle_thread_event_now(event); + } + drain_app_events(&mut app, &mut tui, &mut server, &mut events).await?; + insta::assert_snapshot!(rendered_responses(&app), @" + • Canonical response + • New response + "); + assert!( + !std::iter::from_fn(|| ops.try_recv().ok()).any(|op| matches!(op, Op::UserTurn { .. })) + ); server.shutdown().await?; assert_eq!(serve.await?, [other.to_string(), original.to_string()]); Ok(()) diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index ccd5d3f1b8..4eb16ba6df 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -1573,8 +1573,11 @@ impl App { self.recap.seed_from_turns(&turns, now); self.schedule_recap_check(thread_id, now); - self.chat_widget - .replay_thread_turns(turns, ReplayKind::ResumeInitialMessages); + self.chat_widget.replay_backend_thread_turns( + turns, + ReplayKind::ResumeInitialMessages, + &self.app_server_target, + ); if should_buffer_initial_replay { self.app_event_tx .send(AppEvent::EndInitialHistoryReplayBuffer); @@ -1873,8 +1876,11 @@ impl App { }, ); if !snapshot.turns.is_empty() { - self.chat_widget - .replay_thread_turns(snapshot.turns, ReplayKind::ThreadSnapshot); + self.chat_widget.replay_backend_thread_turns( + snapshot.turns, + ReplayKind::ThreadSnapshot, + &self.app_server_target, + ); } for (event, changes) in snapshot.events.into_iter().zip(request_changes) { reasoning_replay.before_event(&event, &mut self.chat_widget); diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index a085cee8ac..3f1ea5a7c3 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -134,7 +134,10 @@ impl ChatWidget { } ServerNotification::AgentMessageDelta(notification) => { self.restore_realtime_transcripts_before_turn(¬ification.turn_id); - if !self.is_realtime_delegated_reasoning_turn(¬ification.turn_id) + if !self + .transcript + .has_replayed_cloud_text(¬ification.turn_id, ¬ification.item_id) + && !self.is_realtime_delegated_reasoning_turn(¬ification.turn_id) && (from_replay || !self.is_realtime_delegated_agent_item( ¬ification.turn_id, @@ -603,6 +606,9 @@ impl ChatWidget { // render with their choices on completion. if delegated || self.is_realtime_delegated_reasoning_turn(¬ification.turn_id) + || self + .transcript + .has_replayed_cloud_text(¬ification.turn_id, &id) || questions.is_some() || text.is_empty() { diff --git a/codex-rs/tui/src/chatwidget/replay.rs b/codex-rs/tui/src/chatwidget/replay.rs index 62b4645ba2..a2f7bb677b 100644 --- a/codex-rs/tui/src/chatwidget/replay.rs +++ b/codex-rs/tui/src/chatwidget/replay.rs @@ -92,6 +92,17 @@ impl ChatWidget { self.flush_active_cell(); } + pub(crate) fn replay_backend_thread_turns( + &mut self, + turns: Vec, + replay_kind: ReplayKind, + target: &crate::AppServerTarget, + ) { + self.transcript.replayed_cloud_text = + matches!(target, crate::AppServerTarget::Cloud).then(std::collections::HashMap::new); + self.replay_thread_turns(turns, replay_kind); + } + /// Replay a subset of initial events into the UI to seed the transcript when /// resuming an existing session. This approximates the live event flow and /// is intentionally conservative: only safe-to-replay items are rendered to @@ -153,7 +164,8 @@ impl ChatWidget { _ => None, }); let mut replaying_delegation = false; - for item in items.into_iter().map(|entry| entry.payload) { + for entry in items { + let item = entry.payload; if matches!(&item, ThreadItemPayload::UserMessage { content, .. } if realtime::realtime_delegation_input(content).is_some()) { @@ -190,7 +202,10 @@ impl ChatWidget { Some((summary, content)), ); } else { + self.transcript.recording_finished_cloud_text = + status != TurnStatus::InProgress || entry.completed_at_ms.is_some(); self.replay_thread_item(item, turn_id.clone(), replay_kind); + self.transcript.recording_finished_cloud_text = false; } } let status = if hidden_nested_review_turn { diff --git a/codex-rs/tui/src/chatwidget/streaming.rs b/codex-rs/tui/src/chatwidget/streaming.rs index bb1d1be95b..fdf5fda1db 100644 --- a/codex-rs/tui/src/chatwidget/streaming.rs +++ b/codex-rs/tui/src/chatwidget/streaming.rs @@ -534,6 +534,9 @@ impl ChatWidget { if !from_replay && let Some(questions) = &item.questions { self.add_async_questions(&item.id, questions); } + if self.transcript.has_replayed_cloud_text(turn_id, &item.id) { + return; + } self.transcript.last_completed_agent_message = Some((turn_id.to_string(), item.id.clone())); let mut message = String::new(); for content in &item.content { @@ -579,6 +582,13 @@ impl ChatWidget { } else { self.add_to_history(cell); } + if self.transcript.recording_finished_cloud_text + && let Some(seen) = &mut self.transcript.replayed_cloud_text + { + seen.entry(turn_id.to_owned()) + .or_default() + .insert(item.id.clone()); + } self.handle_stream_finished(); self.request_redraw(); } else { diff --git a/codex-rs/tui/src/chatwidget/transcript.rs b/codex-rs/tui/src/chatwidget/transcript.rs index 724a0225b9..e001eb7404 100644 --- a/codex-rs/tui/src/chatwidget/transcript.rs +++ b/codex-rs/tui/src/chatwidget/transcript.rs @@ -48,6 +48,10 @@ pub(super) struct TranscriptState { /// Latest `/status` card and fields, retained across copying until a later turn or command. pub(super) last_status_copy_targets: Option, pub(super) last_completed_agent_message: Option<(String, String)>, + /// Finished assistant items projected into this widget from Cloud history, keyed by turn. + pub(super) replayed_cloud_text: + Option>>, + pub(super) recording_finished_cloud_text: bool, /// Turn and item IDs for the agent message currently supplying streamed deltas. pub(super) active_streamed_agent_message: Option<(String, String)>, /// Turn/item identity and raw source of the latest interleaved stream flush. @@ -77,6 +81,13 @@ impl TranscriptState { } } + pub(super) fn has_replayed_cloud_text(&self, turn_id: &str, item_id: &str) -> bool { + self.replayed_cloud_text + .as_ref() + .and_then(|turns| turns.get(turn_id)) + .is_some_and(|items| items.contains(item_id)) + } + pub(super) fn bump_active_cell_revision(&mut self) { // Wrapping avoids overflow; wraparound would require 2^64 bumps and at // worst causes a one-time cache-key collision.