openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-09 20:50:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Avoid duplicate assistant messages when resuming Cloud threads (#60335)

Browse Source 
## 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
This commit is contained in:
Benjamin Carlsson
2026-11-09 03:01:00 +00:00
committed by copyberry
parent ffdf2daddb
commit 6107b9886c
6 changed files with 163 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app/tests/session_lifecycle_requests.rs
+109 -16
View File
@@ -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::<Vec<_>>()
.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(())
codex-rs/tui/src/app/thread_routing.rs
+10 -4
View File
@@ -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);
codex-rs/tui/src/chatwidget/protocol.rs
+7 -1
View File
@@ -134,7 +134,10 @@ impl ChatWidget {
}
ServerNotification::AgentMessageDelta(notification) => {
self.restore_realtime_transcripts_before_turn(&notification.turn_id);
if !self.is_realtime_delegated_reasoning_turn(&notification.turn_id)
if !self
.transcript
.has_replayed_cloud_text(&notification.turn_id, &notification.item_id)
&& !self.is_realtime_delegated_reasoning_turn(&notification.turn_id)
&& (from_replay
|| !self.is_realtime_delegated_agent_item(
&notification.turn_id,
@@ -603,6 +606,9 @@ impl ChatWidget {
// render with their choices on completion.
if delegated
|| self.is_realtime_delegated_reasoning_turn(&notification.turn_id)
|| self
.transcript
.has_replayed_cloud_text(&notification.turn_id, &id)
|| questions.is_some()
|| text.is_empty()
{
codex-rs/tui/src/chatwidget/replay.rs
+16 -1
View File
@@ -92,6 +92,17 @@ impl ChatWidget {
self.flush_active_cell();
}
pub(crate) fn replay_backend_thread_turns(
&mut self,
turns: Vec<Turn>,
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 {
codex-rs/tui/src/chatwidget/streaming.rs
+10
View File
@@ -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 {
codex-rs/tui/src/chatwidget/transcript.rs
+11
View File
@@ -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<StatusCopySource>,
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<std::collections::HashMap<String, std::collections::HashSet<String>>>,
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.