tui: extract app-server event routing

codex-rs/tui/src/app.rs

@@ -183,7 +183,8 @@ use tokio::task::JoinHandle;
 use toml::Value as TomlValue;
 use uuid::Uuid;
 mod agent_navigation;
-mod app_server_adapter;
+mod app_server_event_targets;
+mod app_server_events;
 pub(crate) mod app_server_requests;
 mod background_requests;
 mod config_persistence;

codex-rs/tui/src/app/app_server_event_targets.rs

@@ -0,0 +1,218 @@
+//! Thread targeting helpers for app-server requests and notifications.
+
+use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::ServerRequest;
+use codex_protocol::ThreadId;
+
+pub(super) fn server_request_thread_id(request: &ServerRequest) -> Option<ThreadId> {
+    match request {
+        ServerRequest::CommandExecutionRequestApproval { params, .. } => {
+            ThreadId::from_string(&params.thread_id).ok()
+        }
+        ServerRequest::FileChangeRequestApproval { params, .. } => {
+            ThreadId::from_string(&params.thread_id).ok()
+        }
+        ServerRequest::ToolRequestUserInput { params, .. } => {
+            ThreadId::from_string(&params.thread_id).ok()
+        }
+        ServerRequest::McpServerElicitationRequest { params, .. } => {
+            ThreadId::from_string(&params.thread_id).ok()
+        }
+        ServerRequest::PermissionsRequestApproval { params, .. } => {
+            ThreadId::from_string(&params.thread_id).ok()
+        }
+        ServerRequest::DynamicToolCall { params, .. } => {
+            ThreadId::from_string(&params.thread_id).ok()
+        }
+        ServerRequest::ChatgptAuthTokensRefresh { .. }
+        | ServerRequest::ApplyPatchApproval { .. }
+        | ServerRequest::ExecCommandApproval { .. } => None,
+    }
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub(super) enum ServerNotificationThreadTarget {
+    Thread(ThreadId),
+    InvalidThreadId(String),
+    Global,
+}
+
+pub(super) fn server_notification_thread_target(
+    notification: &ServerNotification,
+) -> ServerNotificationThreadTarget {
+    let thread_id = match notification {
+        ServerNotification::Error(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::ThreadStarted(notification) => Some(notification.thread.id.as_str()),
+        ServerNotification::ThreadStatusChanged(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadArchived(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::ThreadUnarchived(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::ThreadClosed(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::ThreadNameUpdated(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadTokenUsageUpdated(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadGoalUpdated(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadGoalCleared(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::TurnStarted(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::HookStarted(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::TurnCompleted(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::HookCompleted(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::TurnDiffUpdated(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::TurnPlanUpdated(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::ItemStarted(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::ItemGuardianApprovalReviewStarted(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ItemGuardianApprovalReviewCompleted(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ItemCompleted(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::RawResponseItemCompleted(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::AgentMessageDelta(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::PlanDelta(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::CommandExecutionOutputDelta(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::TerminalInteraction(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::FileChangeOutputDelta(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::FileChangePatchUpdated(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ServerRequestResolved(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::McpToolCallProgress(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ReasoningSummaryTextDelta(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ReasoningSummaryPartAdded(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ReasoningTextDelta(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ContextCompacted(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::ModelRerouted(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::ModelVerification(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadRealtimeStarted(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadRealtimeItemAdded(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadRealtimeTranscriptDelta(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadRealtimeTranscriptDone(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadRealtimeOutputAudioDelta(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadRealtimeSdp(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadRealtimeError(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::ThreadRealtimeClosed(notification) => {
+            Some(notification.thread_id.as_str())
+        }
+        ServerNotification::Warning(notification) => notification.thread_id.as_deref(),
+        ServerNotification::GuardianWarning(notification) => Some(notification.thread_id.as_str()),
+        ServerNotification::SkillsChanged(_)
+        | ServerNotification::McpServerStatusUpdated(_)
+        | ServerNotification::McpServerOauthLoginCompleted(_)
+        | ServerNotification::AccountUpdated(_)
+        | ServerNotification::AccountRateLimitsUpdated(_)
+        | ServerNotification::AppListUpdated(_)
+        | ServerNotification::RemoteControlStatusChanged(_)
+        | ServerNotification::ExternalAgentConfigImportCompleted(_)
+        | ServerNotification::DeprecationNotice(_)
+        | ServerNotification::ConfigWarning(_)
+        | ServerNotification::FuzzyFileSearchSessionUpdated(_)
+        | ServerNotification::FuzzyFileSearchSessionCompleted(_)
+        | ServerNotification::CommandExecOutputDelta(_)
+        | ServerNotification::FsChanged(_)
+        | ServerNotification::WindowsWorldWritableWarning(_)
+        | ServerNotification::WindowsSandboxSetupCompleted(_)
+        | ServerNotification::AccountLoginCompleted(_) => None,
+    };
+
+    match thread_id {
+        Some(thread_id) => match ThreadId::from_string(thread_id) {
+            Ok(thread_id) => ServerNotificationThreadTarget::Thread(thread_id),
+            Err(_) => ServerNotificationThreadTarget::InvalidThreadId(thread_id.to_string()),
+        },
+        None => ServerNotificationThreadTarget::Global,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::ServerNotificationThreadTarget;
+    use super::server_notification_thread_target;
+    use codex_app_server_protocol::GuardianWarningNotification;
+    use codex_app_server_protocol::ServerNotification;
+    use codex_app_server_protocol::WarningNotification;
+    use codex_protocol::ThreadId;
+    use pretty_assertions::assert_eq;
+
+    #[test]
+    fn warning_notifications_without_threads_are_global() {
+        let notification = ServerNotification::Warning(WarningNotification {
+            thread_id: None,
+            message: "warning".to_string(),
+        });
+
+        let target = server_notification_thread_target(&notification);
+
+        assert_eq!(target, ServerNotificationThreadTarget::Global);
+    }
+
+    #[test]
+    fn warning_notifications_route_to_threads_when_thread_id_is_present() {
+        let thread_id = ThreadId::new();
+        let notification = ServerNotification::Warning(WarningNotification {
+            thread_id: Some(thread_id.to_string()),
+            message: "warning".to_string(),
+        });
+
+        let target = server_notification_thread_target(&notification);
+
+        assert_eq!(target, ServerNotificationThreadTarget::Thread(thread_id));
+    }
+
+    #[test]
+    fn guardian_warning_notifications_route_to_threads() {
+        let thread_id = ThreadId::new();
+        let notification = ServerNotification::GuardianWarning(GuardianWarningNotification {
+            thread_id: thread_id.to_string(),
+            message: "warning".to_string(),
+        });
+
+        let target = server_notification_thread_target(&notification);
+
+        assert_eq!(target, ServerNotificationThreadTarget::Thread(thread_id));
+    }
+}

codex-rs/tui/src/app/app_server_events.rs

@@ -0,0 +1,208 @@
+//! App-server event stream handling for the TUI app.
+
+use super::App;
+use super::app_server_event_targets::ServerNotificationThreadTarget;
+use super::app_server_event_targets::server_notification_thread_target;
+use super::app_server_event_targets::server_request_thread_id;
+use crate::app_command::AppCommand;
+use crate::app_event::AppEvent;
+use crate::app_server_session::AppServerSession;
+use crate::app_server_session::app_server_rate_limit_snapshot_to_core;
+use crate::app_server_session::status_account_display_from_auth_mode;
+use codex_app_server_client::AppServerEvent;
+use codex_app_server_protocol::AuthMode;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::ServerRequest;
+
+impl App {
+    fn refresh_mcp_startup_expected_servers_from_config(&mut self) {
+        let enabled_config_mcp_servers: Vec<String> = self
+            .chat_widget
+            .config_ref()
+            .mcp_servers
+            .get()
+            .iter()
+            .filter_map(|(name, server)| server.enabled.then_some(name.clone()))
+            .collect();
+        self.chat_widget
+            .set_mcp_startup_expected_servers(enabled_config_mcp_servers);
+    }
+
+    pub(super) async fn handle_app_server_event(
+        &mut self,
+        app_server_client: &AppServerSession,
+        event: AppServerEvent,
+    ) {
+        match event {
+            AppServerEvent::Lagged { skipped } => {
+                tracing::warn!(
+                    skipped,
+                    "app-server event consumer lagged; dropping ignored events"
+                );
+                self.refresh_mcp_startup_expected_servers_from_config();
+                self.chat_widget.finish_mcp_startup_after_lag();
+            }
+            AppServerEvent::ServerNotification(notification) => {
+                self.handle_server_notification_event(app_server_client, notification)
+                    .await;
+            }
+            AppServerEvent::ServerRequest(request) => {
+                self.handle_server_request_event(app_server_client, request)
+                    .await;
+            }
+            AppServerEvent::Disconnected { message } => {
+                tracing::warn!("app-server event stream disconnected: {message}");
+                self.chat_widget.add_error_message(message.clone());
+                self.app_event_tx.send(AppEvent::FatalExitRequest(message));
+            }
+        }
+    }
+
+    async fn handle_server_notification_event(
+        &mut self,
+        app_server_client: &AppServerSession,
+        notification: ServerNotification,
+    ) {
+        match &notification {
+            ServerNotification::ServerRequestResolved(notification) => {
+                if let Some(request) = self
+                    .pending_app_server_requests
+                    .resolve_notification(&notification.request_id)
+                {
+                    self.chat_widget.dismiss_app_server_request(&request);
+                }
+            }
+            ServerNotification::McpServerStatusUpdated(_) => {
+                self.refresh_mcp_startup_expected_servers_from_config();
+            }
+            ServerNotification::AccountRateLimitsUpdated(notification) => {
+                self.chat_widget.on_rate_limit_snapshot(Some(
+                    app_server_rate_limit_snapshot_to_core(notification.rate_limits.clone()),
+                ));
+                return;
+            }
+            ServerNotification::AccountUpdated(notification) => {
+                self.chat_widget.update_account_state(
+                    status_account_display_from_auth_mode(
+                        notification.auth_mode,
+                        notification.plan_type,
+                    ),
+                    notification.plan_type,
+                    matches!(
+                        notification.auth_mode,
+                        Some(AuthMode::Chatgpt) | Some(AuthMode::ChatgptAuthTokens)
+                    ),
+                );
+                return;
+            }
+            ServerNotification::ExternalAgentConfigImportCompleted(_) => {
+                let cwd = self.chat_widget.config_ref().cwd.to_path_buf();
+                if let Err(err) = self.refresh_in_memory_config_from_disk().await {
+                    tracing::warn!(
+                        error = %err,
+                        "failed to refresh config after external agent config import"
+                    );
+                }
+                self.chat_widget.refresh_plugin_mentions();
+                self.chat_widget.submit_op(AppCommand::reload_user_config());
+                self.fetch_plugins_list(app_server_client, cwd);
+                return;
+            }
+            _ => {}
+        }
+
+        match server_notification_thread_target(&notification) {
+            ServerNotificationThreadTarget::Thread(thread_id) => {
+                let result = if self.primary_thread_id == Some(thread_id)
+                    || self.primary_thread_id.is_none()
+                {
+                    self.enqueue_primary_thread_notification(notification).await
+                } else {
+                    self.enqueue_thread_notification(thread_id, notification)
+                        .await
+                };
+
+                if let Err(err) = result {
+                    tracing::warn!("failed to enqueue app-server notification: {err}");
+                }
+                return;
+            }
+            ServerNotificationThreadTarget::InvalidThreadId(thread_id) => {
+                tracing::warn!(
+                    thread_id,
+                    "ignoring app-server notification with invalid thread_id"
+                );
+                return;
+            }
+            ServerNotificationThreadTarget::Global => {}
+        }
+
+        self.chat_widget
+            .handle_server_notification(notification, /*replay_kind*/ None);
+    }
+
+    async fn handle_server_request_event(
+        &mut self,
+        app_server_client: &AppServerSession,
+        request: ServerRequest,
+    ) {
+        if let Some(unsupported) = self
+            .pending_app_server_requests
+            .note_server_request(&request)
+        {
+            tracing::warn!(
+                request_id = ?unsupported.request_id,
+                message = unsupported.message,
+                "rejecting unsupported app-server request"
+            );
+            self.chat_widget
+                .add_error_message(unsupported.message.clone());
+            if let Err(err) = self
+                .reject_app_server_request(
+                    app_server_client,
+                    unsupported.request_id,
+                    unsupported.message,
+                )
+                .await
+            {
+                tracing::warn!("{err}");
+            }
+            return;
+        }
+
+        let Some(thread_id) = server_request_thread_id(&request) else {
+            tracing::warn!("ignoring threadless app-server request");
+            return;
+        };
+
+        let result =
+            if self.primary_thread_id == Some(thread_id) || self.primary_thread_id.is_none() {
+                self.enqueue_primary_thread_request(request).await
+            } else {
+                self.enqueue_thread_request(thread_id, request).await
+            };
+        if let Err(err) = result {
+            tracing::warn!("failed to enqueue app-server request: {err}");
+        }
+    }
+
+    async fn reject_app_server_request(
+        &self,
+        app_server_client: &AppServerSession,
+        request_id: codex_app_server_protocol::RequestId,
+        reason: String,
+    ) -> std::result::Result<(), String> {
+        app_server_client
+            .reject_server_request(
+                request_id,
+                JSONRPCErrorError {
+                    code: -32000,
+                    message: reason,
+                    data: None,
+                },
+            )
+            .await
+            .map_err(|err| format!("failed to reject app-server request: {err}"))
+    }
+}