From d0516cfe4ba03a87df6f88224a3d26e99907e61e Mon Sep 17 00:00:00 2001 From: Charlie Marsh Date: Sun, 19 Jul 2026 20:04:54 +0000 Subject: [PATCH] Avoid buffering replay-irrelevant thread notifications (#34222) ## Why Raw response items, realtime audio, and other notifications that the TUI does not consume during replay can carry large payloads. Retaining them in each thread's replay buffer wastes memory and can evict useful buffered events. ## What changed - Skip replay buffering for notification types that are handled before routing or ignored during replay, while still updating turn and approval state. - Borrow notifications when recording events for an active thread so routing does not require an extra clone. ## Testing Add coverage that repeatedly sends large MCP progress and realtime audio notifications and verifies that the replay buffer, pending approval, and active turn state remain intact. GitOrigin-RevId: 5b3414f1db7f3ae1feea4fe6c0275e4a7a1a4174 --- codex-rs/tui/src/app/thread_events.rs | 83 ++++++++++++++++++++++++-- codex-rs/tui/src/app/thread_routing.rs | 19 ++++-- 2 files changed, 92 insertions(+), 10 deletions(-) diff --git a/codex-rs/tui/src/app/thread_events.rs b/codex-rs/tui/src/app/thread_events.rs index 45f1c0321d..f2e0682f57 100644 --- a/codex-rs/tui/src/app/thread_events.rs +++ b/codex-rs/tui/src/app/thread_events.rs @@ -6,6 +6,7 @@ //! together with the replay behavior that consumes them. use super::*; +use std::borrow::Cow; #[derive(Debug, Clone)] pub(super) struct ThreadEventSnapshot { @@ -105,9 +106,17 @@ impl ThreadEventStore { } pub(super) fn push_notification(&mut self, notification: ServerNotification) { + self.push_notification_inner(Cow::Owned(notification)); + } + + pub(super) fn push_notification_ref(&mut self, notification: &ServerNotification) { + self.push_notification_inner(Cow::Borrowed(notification)); + } + + fn push_notification_inner(&mut self, notification: Cow<'_, ServerNotification>) { self.pending_interactive_replay - .note_server_notification(¬ification); - match ¬ification { + .note_server_notification(notification.as_ref()); + match notification.as_ref() { ServerNotification::TurnStarted(turn) => { self.active_turn_id = Some(turn.turn.id.clone()); } @@ -121,8 +130,30 @@ impl ThreadEventStore { } _ => {} } + + // These notifications are either handled before routing or ignored by ChatWidget on + // replay. In particular, raw response items and realtime audio can carry large payloads, + // so cloning them into every thread's replay buffer only retains data the TUI cannot use. + if matches!( + notification.as_ref(), + ServerNotification::RawResponseItemCompleted(_) + | ServerNotification::FileChangePatchUpdated(_) + | ServerNotification::ServerRequestResolved(_) + | ServerNotification::McpToolCallProgress(_) + | ServerNotification::ThreadRealtimeItemAdded(_) + | ServerNotification::ThreadRealtimeOutputAudioDelta(_) + | ServerNotification::ThreadRealtimeSdp(_) + | ServerNotification::ThreadRealtimeTranscriptDelta(_) + | ServerNotification::ThreadRealtimeTranscriptDone(_) + | ServerNotification::CommandExecOutputDelta(_) + | ServerNotification::ProcessOutputDelta(_) + | ServerNotification::ProcessExited(_) + ) { + return; + } + self.buffer - .push_back(ThreadBufferedEvent::Notification(notification)); + .push_back(ThreadBufferedEvent::Notification(notification.into_owned())); if self.buffer.len() > self.capacity && let Some(removed) = self.buffer.pop_front() && let ThreadBufferedEvent::Request(request) = &removed @@ -341,7 +372,10 @@ mod tests { use codex_app_server_protocol::HookRunSummary as AppServerHookRunSummary; use codex_app_server_protocol::HookScope as AppServerHookScope; use codex_app_server_protocol::HookStartedNotification; + use codex_app_server_protocol::McpToolCallProgressNotification; use codex_app_server_protocol::RequestId as AppServerRequestId; + use codex_app_server_protocol::ThreadRealtimeAudioChunk; + use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification; use codex_app_server_protocol::TurnCompletedNotification; use codex_app_server_protocol::TurnStartedNotification; use codex_config::types::ApprovalsReviewer; @@ -548,6 +582,47 @@ mod tests { assert_eq!(store.active_turn_id(), None); } + #[test] + fn thread_event_store_skips_large_replay_irrelevant_notifications() { + let thread_id = ThreadId::new(); + let mut store = ThreadEventStore::new(/*capacity*/ 2); + store.push_notification(turn_started_notification(thread_id, "turn-1")); + store.push_request(exec_approval_request( + thread_id, + "turn-1", + "command-approval", + /*approval_id*/ None, + )); + let large_payload = "x".repeat(1024 * 1024); + + for _ in 0..32 { + store.push_notification_ref(&ServerNotification::McpToolCallProgress( + McpToolCallProgressNotification { + thread_id: thread_id.to_string(), + turn_id: "turn-1".to_string(), + item_id: "mcp-1".to_string(), + message: large_payload.clone(), + }, + )); + store.push_notification_ref(&ServerNotification::ThreadRealtimeOutputAudioDelta( + ThreadRealtimeOutputAudioDeltaNotification { + thread_id: thread_id.to_string(), + audio: ThreadRealtimeAudioChunk { + data: large_payload.clone(), + sample_rate: 24_000, + num_channels: 1, + samples_per_channel: None, + item_id: None, + }, + }, + )); + } + + assert_eq!(store.buffer.len(), 2); + assert!(store.has_pending_thread_approvals()); + assert_eq!(store.active_turn_id(), Some("turn-1")); + } + #[test] fn thread_event_store_rebase_preserves_resolved_request_state() { let thread_id = ThreadId::new(); @@ -616,7 +691,7 @@ mod tests { }, ); let mut store = ThreadEventStore::new(/*capacity*/ 8); - store.push_notification(notification.clone()); + store.push_notification_ref(¬ification); store.rebase_buffer_after_session_refresh(); diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 9a53f71986..0284f2cd71 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -926,11 +926,13 @@ impl App { let inferred_session = self .infer_session_for_thread_notification(thread_id, ¬ification) .await; + let is_turn_started = matches!(notification, ServerNotification::TurnStarted(_)); + let notification_status_change = SideParentStatusChange::for_notification(¬ification); let (sender, store) = { let channel = self.ensure_thread_channel(thread_id); (channel.sender.clone(), Arc::clone(&channel.store)) }; - let (should_send, pending_status, turn_stopped) = { + let (notification, pending_status, turn_stopped) = { let mut guard = store.lock().await; if guard.session.is_none() && let Some(session) = inferred_session @@ -944,21 +946,26 @@ impl App { ServerNotification::ThreadClosed(_) => true, _ => false, }; - guard.push_notification(notification.clone()); + let notification = if guard.active { + guard.push_notification_ref(¬ification); + Some(notification) + } else { + guard.push_notification(notification); + None + }; ( - guard.active, + notification, guard.side_parent_pending_status(), turn_stopped, ) }; - if matches!(notification, ServerNotification::TurnStarted(_)) { + if is_turn_started { self.agent_navigation.mark_running(thread_id); } else if turn_stopped { self.agent_navigation.mark_stopped(thread_id); } - let notification_status_change = SideParentStatusChange::for_notification(¬ification); - if should_send { + if let Some(notification) = notification { match sender.try_send(ThreadBufferedEvent::Notification(notification)) { Ok(()) => {} Err(TrySendError::Full(event)) => {