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)) => {