Retry transient Codex Apps connector omissions

codex-rs/app-server/tests/suite/v2/app_list.rs

@@ -1,5 +1,6 @@
 use std::borrow::Cow;
 use std::collections::HashMap;
+use std::collections::VecDeque;
 use std::path::Path;
 use std::sync::Arc;
 use std::sync::Mutex as StdMutex;
@@ -264,6 +265,84 @@ async fn list_apps_includes_plugin_apps_for_chatgpt_auth() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn list_apps_force_refetch_retries_plugin_omission() -> Result<()> {
+    let connector_id = "asdk_app_workspace_docs";
+    let tools = vec![connector_tool(connector_id, "Workspace Docs")?];
+    let (server_url, server_handle, server_control) = start_apps_server_with_delays_and_control(
+        Vec::new(),
+        tools.clone(),
+        Duration::ZERO,
+        Duration::ZERO,
+    )
+    .await?;
+
+    let codex_home = TempDir::new()?;
+    write_connectors_and_plugins_config(codex_home.path(), &server_url)?;
+    write_plugin_app_fixture(codex_home.path(), "sample", connector_id)?;
+    write_chatgpt_auth(
+        codex_home.path(),
+        ChatGptAuthFixture::new("chatgpt-token")
+            .account_id("account-123")
+            .chatgpt_user_id("user-plugin-cache")
+            .chatgpt_account_id("account-123"),
+        AuthCredentialsStoreMode::File,
+    )?;
+
+    let mut mcp = TestAppServer::new(codex_home.path()).await?;
+    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
+
+    let initial_request = mcp
+        .send_apps_list_request(AppsListParams {
+            limit: None,
+            cursor: None,
+            thread_id: None,
+            force_refetch: true,
+        })
+        .await?;
+    let initial_response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(initial_request)),
+    )
+    .await??;
+    let AppsListResponse {
+        data: initial_data, ..
+    } = to_response(initial_response)?;
+    assert!(
+        initial_data
+            .iter()
+            .any(|app| app.id == connector_id && app.is_accessible)
+    );
+    server_control.queue_tools_responses(vec![tools.clone(), Vec::new(), tools]);
+
+    let refetch_request = mcp
+        .send_apps_list_request(AppsListParams {
+            limit: None,
+            cursor: None,
+            thread_id: None,
+            force_refetch: true,
+        })
+        .await?;
+    let refetch_response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(refetch_request)),
+    )
+    .await??;
+    let AppsListResponse {
+        data: refetched_data,
+        ..
+    } = to_response(refetch_response)?;
+    assert!(
+        refetched_data
+            .iter()
+            .any(|app| app.id == connector_id && app.is_accessible)
+    );
+
+    server_handle.abort();
+    let _ = server_handle.await;
+    Ok(())
+}
+
 #[tokio::test]
 async fn list_apps_uses_thread_feature_flag_when_thread_id_is_provided() -> Result<()> {
     let connectors = vec![AppInfo {
@@ -1470,19 +1549,29 @@ struct AppsServerState {
 #[derive(Clone)]
 struct AppListMcpServer {
     tools: Arc<StdMutex<Vec<Tool>>>,
+    queued_tools: Arc<StdMutex<VecDeque<Vec<Tool>>>>,
     tools_delay: Duration,
 }
 
 impl AppListMcpServer {
-    fn new(tools: Arc<StdMutex<Vec<Tool>>>, tools_delay: Duration) -> Self {
-        Self { tools, tools_delay }
+    fn new(
+        tools: Arc<StdMutex<Vec<Tool>>>,
+        queued_tools: Arc<StdMutex<VecDeque<Vec<Tool>>>>,
+        tools_delay: Duration,
+    ) -> Self {
+        Self {
+            tools,
+            queued_tools,
+            tools_delay,
+        }
     }
 }
 
 #[derive(Clone)]
 struct AppsServerControl {
     response: Arc<StdMutex<serde_json::Value>>,
     tools: Arc<StdMutex<Vec<Tool>>>,
+    queued_tools: Arc<StdMutex<VecDeque<Vec<Tool>>>>,
 }
 
 impl AppsServerControl {
@@ -1501,6 +1590,14 @@ impl AppsServerControl {
             .unwrap_or_else(std::sync::PoisonError::into_inner);
         *tools_guard = tools;
     }
+
+    fn queue_tools_responses(&self, responses: Vec<Vec<Tool>>) {
+        let mut queued_tools_guard = self
+            .queued_tools
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        queued_tools_guard.extend(responses);
+    }
 }
 
 impl ServerHandler for AppListMcpServer {
@@ -1515,15 +1612,22 @@ impl ServerHandler for AppListMcpServer {
     ) -> impl std::future::Future<Output = Result<ListToolsResult, rmcp::ErrorData>> + Send + '_
     {
         let tools = self.tools.clone();
+        let queued_tools = self.queued_tools.clone();
         let tools_delay = self.tools_delay;
         async move {
             if tools_delay > Duration::ZERO {
                 tokio::time::sleep(tools_delay).await;
             }
-            let tools = tools
+            let tools = queued_tools
                 .lock()
                 .unwrap_or_else(std::sync::PoisonError::into_inner)
-                .clone();
+                .pop_front()
+                .unwrap_or_else(|| {
+                    tools
+                        .lock()
+                        .unwrap_or_else(std::sync::PoisonError::into_inner)
+                        .clone()
+                });
             Ok(ListToolsResult {
                 tools,
                 next_cursor: None,
@@ -1589,6 +1693,7 @@ async fn start_apps_server_with_delays_and_control_inner(
         json!({ "apps": connectors, "next_token": null }),
     ));
     let tools = Arc::new(StdMutex::new(tools));
+    let queued_tools = Arc::new(StdMutex::new(VecDeque::new()));
     let state = AppsServerState {
         expected_bearer: "Bearer chatgpt-token".to_string(),
         expected_account_id: "account-123".to_string(),
@@ -1600,6 +1705,7 @@ async fn start_apps_server_with_delays_and_control_inner(
     let server_control = AppsServerControl {
         response,
         tools: tools.clone(),
+        queued_tools: queued_tools.clone(),
     };
 
     let listener = TcpListener::bind("127.0.0.1:0").await?;
@@ -1608,7 +1714,14 @@ async fn start_apps_server_with_delays_and_control_inner(
     let mcp_service = StreamableHttpService::new(
         {
             let tools = tools.clone();
-            move || Ok(AppListMcpServer::new(tools.clone(), tools_delay))
+            let queued_tools = queued_tools.clone();
+            move || {
+                Ok(AppListMcpServer::new(
+                    tools.clone(),
+                    queued_tools.clone(),
+                    tools_delay,
+                ))
+            }
         },
         Arc::new(LocalSessionManager::default()),
         StreamableHttpServerConfig::default(),

codex-rs/codex-mcp/src/codex_apps_cache.rs

@@ -65,6 +65,7 @@ pub struct CodexAppsToolsCache {
 #[derive(Clone)]
 pub(crate) struct CodexAppsToolsCacheContext {
     entry: Arc<CodexAppsToolsCacheEntry>,
+    expected_connector_ids: Arc<Vec<String>>,
 }
 
 impl CodexAppsToolsCacheContext {
@@ -77,7 +78,7 @@ impl CodexAppsToolsCacheContext {
     pub(crate) fn server_info_cache_path(&self) -> PathBuf {
         self.entry
             .identity
-            .cache_path_in(CODEX_APPS_SERVER_INFO_CACHE_DIR)
+            .auth_cache_path_in(CODEX_APPS_SERVER_INFO_CACHE_DIR)
     }
 
     pub(crate) fn current_tools(&self) -> Option<Vec<ToolInfo>> {
@@ -135,6 +136,10 @@ impl CodexAppsToolsCacheContext {
         tools
     }
 
+    pub(crate) fn expected_connector_ids(&self) -> &[String] {
+        self.expected_connector_ids.as_slice()
+    }
+
     #[cfg(test)]
     pub(crate) fn store_current_tools_for_test(&self, tools: Vec<ToolInfo>) {
         self.entry.current_tools.store(Some(Arc::new(tools)));
@@ -146,17 +151,26 @@ impl CodexAppsToolsCache {
         &self,
         codex_home: PathBuf,
         auth_key: CodexAppsToolsCacheKey,
+        expected_connector_ids: impl IntoIterator<Item = String>,
     ) -> CodexAppsToolsCacheContext {
+        let mut expected_connector_ids = expected_connector_ids.into_iter().collect::<Vec<_>>();
+        expected_connector_ids.sort_unstable();
+        expected_connector_ids.dedup();
         let identity = CodexAppsToolsCacheIdentity {
             codex_home,
             auth_key,
+            expected_connector_ids: expected_connector_ids.clone(),
         };
+        let expected_connector_ids = Arc::new(expected_connector_ids);
         let mut entries = lock_unpoisoned(&self.entries);
         let entry = entries
             .entry(identity.clone())
             .or_insert_with(|| Arc::new(CodexAppsToolsCacheEntry::new(identity)))
             .clone();
-        CodexAppsToolsCacheContext { entry }
+        CodexAppsToolsCacheContext {
+            entry,
+            expected_connector_ids,
+        }
     }
 }
 
@@ -167,7 +181,7 @@ pub(crate) enum CodexAppsToolsFetchSource {
 }
 
 impl CodexAppsToolsFetchSource {
-    fn as_str(self) -> &'static str {
+    pub(crate) fn as_str(self) -> &'static str {
         match self {
             Self::Startup => "startup",
             Self::HardRefresh => "hard_refresh",
@@ -201,19 +215,30 @@ impl CodexAppsToolsCacheEntry {
 
 /// Everything that decides whether two Codex Apps clients can share tools.
 ///
-/// The auth key says whose catalog we are reading. `codex_home` keeps the
+/// The auth key says whose catalog we are reading. Expected connector IDs keep
+/// plugin contexts from sharing a raw snapshot. `codex_home` keeps the
 /// persisted cache under the right home directory.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct CodexAppsToolsCacheIdentity {
     codex_home: PathBuf,
     auth_key: CodexAppsToolsCacheKey,
+    expected_connector_ids: Vec<String>,
 }
 
 impl CodexAppsToolsCacheIdentity {
     fn cache_path_in(&self, cache_dir: &str) -> PathBuf {
         // `codex_home` is already the parent directory. Keep it out of the
         // filename hash so non-UTF-8 Unix paths cannot collapse distinct auth
         // keys onto the same disk cache file.
+        let identity_json = serde_json::to_string(&(&self.auth_key, &self.expected_connector_ids))
+            .unwrap_or_default();
+        let identity_hash = sha1_hex(&identity_json);
+        self.codex_home
+            .join(cache_dir)
+            .join(format!("{identity_hash}.json"))
+    }
+
+    fn auth_cache_path_in(&self, cache_dir: &str) -> PathBuf {
         let identity_json = serde_json::to_string(&self.auth_key).unwrap_or_default();
         let identity_hash = sha1_hex(&identity_json);
         self.codex_home
@@ -355,7 +380,7 @@ struct CodexAppsServerInfoDiskCache {
 }
 
 const CODEX_APPS_TOOLS_CACHE_DIR: &str = "cache/codex_apps_tools";
-const CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION: u8 = 4;
+const CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION: u8 = 5;
 
 const CODEX_APPS_SERVER_INFO_CACHE_DIR: &str = "cache/codex_apps_server_info";
 const CODEX_APPS_SERVER_INFO_CACHE_SCHEMA_VERSION: u8 = 1;

codex-rs/codex-mcp/src/codex_apps_cache_tests.rs

@@ -56,6 +56,7 @@ fn create_codex_apps_tools_cache_context(
             chatgpt_user_id: chatgpt_user_id.map(ToOwned::to_owned),
             is_workspace_account: false,
         },
+        std::iter::empty(),
     )
 }
 
@@ -362,6 +363,7 @@ fn codex_apps_tools_cache_publishes_newest_shared_snapshot() {
             chatgpt_user_id: Some("user-one".to_string()),
             is_workspace_account: false,
         },
+        std::iter::empty(),
     );
     let cache_context_2 = cache.context(
         codex_home.path().to_path_buf(),
@@ -370,6 +372,7 @@ fn codex_apps_tools_cache_publishes_newest_shared_snapshot() {
             chatgpt_user_id: Some("user-one".to_string()),
             is_workspace_account: false,
         },
+        std::iter::empty(),
     );
     let older_ticket = cache_context_1.begin_fetch(CodexAppsToolsFetchSource::Startup);
     let newer_ticket = cache_context_2.begin_fetch(CodexAppsToolsFetchSource::HardRefresh);
@@ -402,6 +405,43 @@ fn codex_apps_tools_cache_publishes_newest_shared_snapshot() {
     );
 }
 
+#[test]
+fn codex_apps_tools_cache_scopes_plugin_connector_expectations() {
+    let codex_home = tempdir().expect("tempdir");
+    let cache = CodexAppsToolsCache::default();
+    let auth_key = CodexAppsToolsCacheKey {
+        account_id: Some("account-one".to_string()),
+        chatgpt_user_id: Some("user-one".to_string()),
+        is_workspace_account: false,
+    };
+    let connector_a = cache.context(
+        codex_home.path().to_path_buf(),
+        auth_key.clone(),
+        ["connector_a".to_string()],
+    );
+    let connector_b = cache.context(
+        codex_home.path().to_path_buf(),
+        auth_key,
+        ["connector_b".to_string()],
+    );
+
+    connector_a.store_current_tools_for_test(vec![create_test_tool(
+        CODEX_APPS_MCP_SERVER_NAME,
+        "connector_a_search",
+    )]);
+
+    assert!(connector_a.current_tools().is_some());
+    assert!(connector_b.current_tools().is_none());
+    assert_ne!(
+        connector_a.tools_cache_path(),
+        connector_b.tools_cache_path()
+    );
+    assert_eq!(
+        connector_a.server_info_cache_path(),
+        connector_b.server_info_cache_path()
+    );
+}
+
 #[test]
 fn codex_apps_tools_cache_keeps_live_publish_when_disk_persistence_fails() {
     let codex_home = tempdir().expect("tempdir");
@@ -414,6 +454,7 @@ fn codex_apps_tools_cache_keeps_live_publish_when_disk_persistence_fails() {
             chatgpt_user_id: Some("user-one".to_string()),
             is_workspace_account: false,
         },
+        std::iter::empty(),
     );
     let tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "live")];
     let published_tools = cache_context.publish_if_newest_accepted(
@@ -443,6 +484,7 @@ fn codex_apps_tools_cache_scopes_non_utf8_home_disk_paths() {
             chatgpt_user_id: Some("user-one".to_string()),
             is_workspace_account: false,
         },
+        std::iter::empty(),
     );
     let user_two_context = cache.context(
         codex_home,
@@ -451,6 +493,7 @@ fn codex_apps_tools_cache_scopes_non_utf8_home_disk_paths() {
             chatgpt_user_id: Some("user-two".to_string()),
             is_workspace_account: false,
         },
+        std::iter::empty(),
     );
     let cache_paths = [
         user_one_context.tools_cache_path(),