Retry failed Codex Apps MCP startup (#29920)

## Problem

The built-in Codex Apps MCP client shares a future for the full startup
operation: connect, complete `initialize`, fetch the initial tools, and
return a usable client. Sharing deduplicates startup work, but it also
memoizes terminal errors.

After a transient connection, handshake, or initial `tools/list`
failure, later tool builds observe the same failed future. The thread
cannot reconnect after the backend recovers and continues serving its
startup-time cached tool snapshot, which may be empty or stale.

## Fix

When Apps MCP startup ends in an error, Codex starts bounded recovery
without putting startup latency on tool-router construction:

1. The current tool build immediately continues with the cached startup
snapshot.
2. After the initial failure is reported, Codex starts one fresh full
startup attempt in the background.
3. Concurrent tool builds share that in-flight attempt and also continue
with cached tools.
4. On success, the recovered client becomes active, refreshes the Apps
tools cache, emits a `Ready` startup status, and is reused by later
operations.
5. On failure, the cache remains unchanged and later tool builds may
start another background attempt after exponential cooldown: 1s, 2s, 4s,
8s, 16s, then 30s maximum.

Each recreated startup performs a fresh MCP `initialize` and uncached
`tools/list`. The MCP client retains its existing bounded retries for
retryable `initialize` and `tools/list` failures.

This avoids adding the Apps startup timeout to every request during a
sustained outage.

## Scope

This is limited to the built-in Codex Apps MCP client:

- no reconnects for user-configured MCP servers;
- no cache deletion; and
- no proactive refresh for a healthy client with stale tools.

## Tests

Coverage verifies:

- tool builds return cached tools without waiting for a blocked
reconnect;
- concurrent tool builds start only one background reconnect;
- failed reconnects preserve cached tools and respect exponential
cooldown;
- a recovered client is retained and reused; and
- a long-lived thread exposes recovered app tools on a later follow-up.

Validation:

- `just test -p codex-mcp` — 95 passed
- `just test -p codex-core
later_follow_up_uses_background_recovered_apps_after_mid_thread_startup_failures
--no-capture` — passed
- `just fix -p codex-mcp`
- `just fmt`

codex-rs/codex-mcp/src/rmcp_client.rs

@@ -12,6 +12,7 @@ use std::collections::HashMap;
 use std::env;
 use std::ffi::OsString;
 use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use std::time::Duration;
@@ -48,6 +49,9 @@ use codex_exec_server::HttpClient;
 use codex_exec_server::ReqwestHttpClient;
 use codex_protocol::mcp::McpServerInfo;
 use codex_protocol::protocol::Event;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::McpStartupStatus;
+use codex_protocol::protocol::McpStartupUpdateEvent;
 use codex_rmcp_client::ExecutorStdioServerLauncher;
 use codex_rmcp_client::LocalStdioServerLauncher;
 use codex_rmcp_client::RmcpClient;
@@ -64,6 +68,7 @@ use rmcp::model::JsonObject;
 use rmcp::model::ProtocolVersion;
 use rmcp::model::Tool as RmcpTool;
 use rmcp::transport::auth::AuthError;
+use tokio::time::Instant as TokioInstant;
 use tokio_util::sync::CancellationToken;
 use tracing::Instrument;
 use tracing::instrument;
@@ -80,6 +85,9 @@ pub(crate) const MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC: &str =
 pub(crate) const DEFAULT_STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
 pub(crate) const DEFAULT_TOOL_TIMEOUT: Duration = Duration::from_secs(300);
 
+pub(crate) const CODEX_APPS_RECONNECT_INITIAL_BACKOFF: Duration = Duration::from_secs(1);
+const CODEX_APPS_RECONNECT_MAX_BACKOFF: Duration = Duration::from_secs(30);
+
 const UNTRUSTED_CONNECTOR_META_KEYS: &[&str] = &[
     "connector_id",
     "connector_name",
@@ -128,56 +136,174 @@ impl ManagedClient {
     }
 }
 
-#[derive(Clone)]
-pub(crate) struct AsyncManagedClient {
-    pub(crate) client: Shared<BoxFuture<'static, Result<ManagedClient, StartupOutcomeError>>>,
-    pub(crate) is_codex_apps_mcp_server: bool,
-    pub(crate) cached_server_info: Option<McpServerInfo>,
-    pub(crate) codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
-    pub(crate) tool_filter: ToolFilter,
-    pub(crate) startup_complete: Arc<AtomicBool>,
-    pub(crate) tool_plugin_provenance: Arc<ToolPluginProvenance>,
-    pub(crate) cancel_token: CancellationToken,
+pub(crate) type ManagedClientFuture =
+    Shared<BoxFuture<'static, Result<ManagedClient, StartupOutcomeError>>>;
+
+#[derive(Default)]
+struct CodexAppsStartupReconnectState {
+    current_client: Option<ManagedClient>,
+    reconnect_in_flight: bool,
+    consecutive_failures: u32,
+    retry_not_before: Option<TokioInstant>,
 }
 
-impl AsyncManagedClient {
-    // Keep this constructor flat so the startup inputs remain readable at the
-    // single call site instead of introducing a one-off params wrapper.
-    #[instrument(level = "trace", skip_all, fields(server_name = %server_name))]
-    #[allow(clippy::too_many_arguments)]
-    pub(crate) fn new(
+#[derive(Clone)]
+struct CodexAppsStartupStatusContext {
+    submit_id: String,
+    server_name: String,
+    tx_event: Sender<Event>,
+}
+
+pub(crate) struct CodexAppsStartupReconnect {
+    factory: Arc<dyn Fn() -> ManagedClientFuture + Send + Sync>,
+    state: StdMutex<CodexAppsStartupReconnectState>,
+    startup_status_context: Option<CodexAppsStartupStatusContext>,
+}
+
+impl CodexAppsStartupReconnect {
+    pub(crate) fn new(factory: Arc<dyn Fn() -> ManagedClientFuture + Send + Sync>) -> Self {
+        Self {
+            factory,
+            state: StdMutex::new(CodexAppsStartupReconnectState::default()),
+            startup_status_context: None,
+        }
+    }
+
+    fn with_startup_status_context(
+        mut self,
+        submit_id: String,
         server_name: String,
-        server: EffectiveMcpServer,
-        store_mode: OAuthCredentialsStoreMode,
-        keyring_backend_kind: AuthKeyringBackendKind,
-        cancel_token: CancellationToken,
         tx_event: Sender<Event>,
-        elicitation_requests: ElicitationRequestManager,
-        codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
-        tool_plugin_provenance: Arc<ToolPluginProvenance>,
-        runtime_context: McpRuntimeContext,
-        runtime_auth_provider: Option<SharedAuthProvider>,
-        client_elicitation_capability: ElicitationCapability,
-        supports_openai_form_elicitation: bool,
     ) -> Self {
+        self.startup_status_context = Some(CodexAppsStartupStatusContext {
+            submit_id,
+            server_name,
+            tx_event,
+        });
+        self
+    }
+
+    fn current_client(&self) -> Option<ManagedClient> {
+        self.state
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .current_client
+            .clone()
+    }
+
+    fn reconnect_in_background(self: &Arc<Self>) {
+        {
+            let mut state = self
+                .state
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            if state.current_client.is_some() || state.reconnect_in_flight {
+                return;
+            }
+            if state
+                .retry_not_before
+                .is_some_and(|retry_not_before| TokioInstant::now() < retry_not_before)
+            {
+                return;
+            }
+            state.reconnect_in_flight = true;
+        }
+
+        let reconnect = Arc::clone(self);
+        tokio::spawn(async move {
+            let result = (reconnect.factory)().await;
+            let startup_status_context = reconnect.startup_status_context.clone();
+            let recovered = {
+                let mut state = reconnect
+                    .state
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner);
+                state.reconnect_in_flight = false;
+                match result {
+                    Ok(client) => {
+                        state.current_client = Some(client);
+                        state.consecutive_failures = 0;
+                        state.retry_not_before = None;
+                        true
+                    }
+                    Err(error) => {
+                        state.consecutive_failures = state.consecutive_failures.saturating_add(1);
+                        let retry_after = codex_apps_reconnect_backoff(state.consecutive_failures);
+                        state.retry_not_before = Some(TokioInstant::now() + retry_after);
+                        warn!(
+                            error = %error,
+                            retry_after_ms = retry_after.as_millis(),
+                            "Apps MCP startup reconnect failed; continuing with cached tools"
+                        );
+                        false
+                    }
+                }
+            };
+
+            if recovered && let Some(context) = startup_status_context {
+                let _ = context
+                    .tx_event
+                    .send(Event {
+                        id: context.submit_id,
+                        msg: EventMsg::McpStartupUpdate(McpStartupUpdateEvent {
+                            server: context.server_name,
+                            status: McpStartupStatus::Ready,
+                        }),
+                    })
+                    .await;
+            }
+        });
+    }
+}
+
+fn codex_apps_reconnect_backoff(consecutive_failures: u32) -> Duration {
+    let exponent = consecutive_failures.saturating_sub(1).min(5);
+    CODEX_APPS_RECONNECT_INITIAL_BACKOFF
+        .saturating_mul(1 << exponent)
+        .min(CODEX_APPS_RECONNECT_MAX_BACKOFF)
+}
+
+#[derive(Clone)]
+struct ManagedClientStartup {
+    server_name: String,
+    server: EffectiveMcpServer,
+    store_mode: OAuthCredentialsStoreMode,
+    keyring_backend_kind: AuthKeyringBackendKind,
+    tx_event: Sender<Event>,
+    elicitation_requests: ElicitationRequestManager,
+    codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
+    runtime_context: McpRuntimeContext,
+    runtime_auth_provider: Option<SharedAuthProvider>,
+    client_elicitation_capability: ElicitationCapability,
+    supports_openai_form_elicitation: bool,
+    cancel_token: CancellationToken,
+    startup_complete: Arc<AtomicBool>,
+}
+
+impl ManagedClientStartup {
+    fn start(&self) -> ManagedClientFuture {
+        let Self {
+            server_name,
+            server,
+            store_mode,
+            keyring_backend_kind,
+            tx_event,
+            elicitation_requests,
+            codex_apps_tools_cache_context,
+            runtime_context,
+            runtime_auth_provider,
+            client_elicitation_capability,
+            supports_openai_form_elicitation,
+            cancel_token,
+            startup_complete,
+        } = self.clone();
         let is_codex_apps_mcp_server = server_name == CODEX_APPS_MCP_SERVER_NAME;
         let tool_filter = server
             .configured_config()
             .map(ToolFilter::from_config)
             .unwrap_or_default();
-        let cached_server_info = if is_codex_apps_mcp_server {
-            codex_apps_tools_cache_context
-                .as_ref()
-                .and_then(load_startup_cached_codex_apps_server_info)
-        } else {
-            None
-        };
-        let startup_tool_filter = tool_filter.clone();
-        let codex_apps_tools_cache_context_for_fut = codex_apps_tools_cache_context.clone();
-        let startup_complete = Arc::new(AtomicBool::new(false));
-        let startup_complete_for_fut = Arc::clone(&startup_complete);
-        let cancel_token_for_fut = cancel_token.clone();
-        let fut = async move {
+        let cancel_token_for_fut = cancel_token;
+        async move {
             let outcome = match async {
                 if let Err(error) = validate_mcp_server_name(&server_name) {
                     return Err(error.into());
@@ -207,10 +333,10 @@ impl AsyncManagedClient {
                             .configured_config()
                             .and_then(|config| config.tool_timeout_sec)
                             .unwrap_or(DEFAULT_TOOL_TIMEOUT),
-                        tool_filter: startup_tool_filter,
+                        tool_filter,
                         tx_event,
                         elicitation_requests,
-                        codex_apps_tools_cache_context: codex_apps_tools_cache_context_for_fut,
+                        codex_apps_tools_cache_context,
                         client_elicitation_capability,
                         supports_openai_form_elicitation,
                     },
@@ -224,10 +350,91 @@ impl AsyncManagedClient {
                 Err(CancelErr::Cancelled) => Err(StartupOutcomeError::Cancelled),
             };
 
-            startup_complete_for_fut.store(true, Ordering::Release);
+            startup_complete.store(true, Ordering::Release);
             outcome
+        }
+        .in_current_span()
+        .boxed()
+        .shared()
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct AsyncManagedClient {
+    pub(crate) client: ManagedClientFuture,
+    pub(crate) is_codex_apps_mcp_server: bool,
+    pub(crate) cached_server_info: Option<McpServerInfo>,
+    pub(crate) codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
+    pub(crate) tool_filter: ToolFilter,
+    pub(crate) startup_complete: Arc<AtomicBool>,
+    pub(crate) startup_reconnect: Option<Arc<CodexAppsStartupReconnect>>,
+    pub(crate) tool_plugin_provenance: Arc<ToolPluginProvenance>,
+    pub(crate) cancel_token: CancellationToken,
+}
+
+impl AsyncManagedClient {
+    // Keep this constructor flat so the startup inputs remain readable at the
+    // single call site instead of introducing a one-off params wrapper.
+    #[instrument(level = "trace", skip_all, fields(server_name = %server_name))]
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn new(
+        server_name: String,
+        startup_submit_id: String,
+        server: EffectiveMcpServer,
+        store_mode: OAuthCredentialsStoreMode,
+        keyring_backend_kind: AuthKeyringBackendKind,
+        cancel_token: CancellationToken,
+        tx_event: Sender<Event>,
+        elicitation_requests: ElicitationRequestManager,
+        codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
+        tool_plugin_provenance: Arc<ToolPluginProvenance>,
+        runtime_context: McpRuntimeContext,
+        runtime_auth_provider: Option<SharedAuthProvider>,
+        client_elicitation_capability: ElicitationCapability,
+        supports_openai_form_elicitation: bool,
+    ) -> Self {
+        let is_codex_apps_mcp_server = server_name == CODEX_APPS_MCP_SERVER_NAME;
+        let reconnect_server_name = server_name.clone();
+        let reconnect_tx_event = tx_event.clone();
+        let tool_filter = server
+            .configured_config()
+            .map(ToolFilter::from_config)
+            .unwrap_or_default();
+        let cached_server_info = if is_codex_apps_mcp_server {
+            codex_apps_tools_cache_context
+                .as_ref()
+                .and_then(load_startup_cached_codex_apps_server_info)
+        } else {
+            None
         };
-        let client = fut.in_current_span().boxed().shared();
+        let startup_complete = Arc::new(AtomicBool::new(false));
+        let startup = Arc::new(ManagedClientStartup {
+            server_name,
+            server,
+            store_mode,
+            keyring_backend_kind,
+            tx_event,
+            elicitation_requests,
+            codex_apps_tools_cache_context: codex_apps_tools_cache_context.clone(),
+            runtime_context,
+            runtime_auth_provider,
+            client_elicitation_capability,
+            supports_openai_form_elicitation,
+            cancel_token: cancel_token.clone(),
+            startup_complete: Arc::clone(&startup_complete),
+        });
+        let client = startup.start();
+        let startup_reconnect = is_codex_apps_mcp_server.then(|| {
+            let startup = Arc::clone(&startup);
+            Arc::new(
+                CodexAppsStartupReconnect::new(Arc::new(move || startup.start()))
+                    .with_startup_status_context(
+                        startup_submit_id,
+                        reconnect_server_name,
+                        reconnect_tx_event,
+                    ),
+            )
+        });
         if codex_apps_tools_cache_context
             .as_ref()
             .is_some_and(CodexAppsToolsCacheContext::has_current_tools)
@@ -245,15 +452,35 @@ impl AsyncManagedClient {
             codex_apps_tools_cache_context,
             tool_filter,
             startup_complete,
+            startup_reconnect,
             tool_plugin_provenance,
             cancel_token,
         }
     }
 
     pub(crate) async fn client(&self) -> Result<ManagedClient, StartupOutcomeError> {
+        if let Some(client) = self
+            .startup_reconnect
+            .as_ref()
+            .and_then(|reconnect| reconnect.current_client())
+        {
+            return Ok(client);
+        }
         self.client.clone().await
     }
 
+    pub(crate) async fn reconnect_failed_startup(&self) {
+        let Some(startup_reconnect) = self.startup_reconnect.as_ref() else {
+            return;
+        };
+        if !self.startup_complete.load(Ordering::Acquire) {
+            return;
+        }
+        if matches!(self.client().await, Err(StartupOutcomeError::Failed { .. })) {
+            startup_reconnect.reconnect_in_background();
+        }
+    }
+
     pub(crate) async fn shutdown(&self) {
         self.cancel_token.cancel();
         match self.client().await {