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

@@ -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(&notification);
-        match &notification {
+            .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(&notification);
 
         store.rebase_buffer_after_session_refresh();
 

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, &notification)
             .await;
+        let is_turn_started = matches!(notification, ServerNotification::TurnStarted(_));
+        let notification_status_change = SideParentStatusChange::for_notification(&notification);
         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(&notification);
+                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(&notification);
 
-        if should_send {
+        if let Some(notification) = notification {
             match sender.try_send(ThreadBufferedEvent::Notification(notification)) {
                 Ok(()) => {}
                 Err(TrySendError::Full(event)) => {