diff --git a/codex-rs/app-server/tests/suite/v2/app_list.rs b/codex-rs/app-server/tests/suite/v2/app_list.rs index 5fad04929f7f..5a7af24a69ce 100644 --- a/codex-rs/app-server/tests/suite/v2/app_list.rs +++ b/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,12 +1549,21 @@ struct AppsServerState { #[derive(Clone)] struct AppListMcpServer { tools: Arc>>, + queued_tools: Arc>>>, tools_delay: Duration, } impl AppListMcpServer { - fn new(tools: Arc>>, tools_delay: Duration) -> Self { - Self { tools, tools_delay } + fn new( + tools: Arc>>, + queued_tools: Arc>>>, + tools_delay: Duration, + ) -> Self { + Self { + tools, + queued_tools, + tools_delay, + } } } @@ -1483,6 +1571,7 @@ impl AppListMcpServer { struct AppsServerControl { response: Arc>, tools: Arc>>, + queued_tools: Arc>>>, } 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>) { + 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> + 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(), diff --git a/codex-rs/codex-mcp/src/codex_apps_cache.rs b/codex-rs/codex-mcp/src/codex_apps_cache.rs index ea3944462883..fc6f0d51b9d6 100644 --- a/codex-rs/codex-mcp/src/codex_apps_cache.rs +++ b/codex-rs/codex-mcp/src/codex_apps_cache.rs @@ -65,6 +65,7 @@ pub struct CodexAppsToolsCache { #[derive(Clone)] pub(crate) struct CodexAppsToolsCacheContext { entry: Arc, + expected_connector_ids: Arc>, } 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> { @@ -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) { 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, ) -> CodexAppsToolsCacheContext { + let mut expected_connector_ids = expected_connector_ids.into_iter().collect::>(); + 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,12 +215,14 @@ 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, } impl CodexAppsToolsCacheIdentity { @@ -214,6 +230,15 @@ impl CodexAppsToolsCacheIdentity { // `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; diff --git a/codex-rs/codex-mcp/src/codex_apps_cache_tests.rs b/codex-rs/codex-mcp/src/codex_apps_cache_tests.rs index 46ad311a1b56..298be6361529 100644 --- a/codex-rs/codex-mcp/src/codex_apps_cache_tests.rs +++ b/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(), diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index 223d4296d51e..06e5708e6aac 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -24,11 +24,12 @@ use crate::mcp::CODEX_APPS_MCP_SERVER_NAME; use crate::mcp::ToolPluginProvenance; use crate::rmcp_client::AsyncManagedClient; use crate::rmcp_client::CODEX_APPS_REFRESH_DURATION_METRIC; +use crate::rmcp_client::CodexAppsToolsCacheRetryContext; use crate::rmcp_client::DEFAULT_STARTUP_TIMEOUT; use crate::rmcp_client::MCP_TOOLS_LIST_DURATION_METRIC; use crate::rmcp_client::ManagedClient; use crate::rmcp_client::StartupOutcomeError; -use crate::rmcp_client::list_tools_for_client_uncached; +use crate::rmcp_client::list_tools_for_client_uncached_with_cache_retry; use crate::runtime::McpRuntimeContext; use crate::runtime::emit_duration; use crate::server::EffectiveMcpServer; @@ -197,8 +198,11 @@ impl McpConnectionManager { let shares_codex_apps_tools_cache = should_share_codex_apps_tools_cache(&server_name, uses_env_bearer_token); let codex_apps_tools_cache_context = shares_codex_apps_tools_cache.then(|| { - codex_apps_tools_cache - .context(codex_home.clone(), codex_apps_tools_cache_key.clone()) + codex_apps_tools_cache.context( + codex_home.clone(), + codex_apps_tools_cache_key.clone(), + tool_plugin_provenance.connector_ids().map(str::to_owned), + ) }); // If Codex Apps has an env bearer token, that is its auth path. Do // not also attach the ambient CodexAuth provider. @@ -544,19 +548,22 @@ impl McpConnectionManager { .codex_apps_tools_cache_context .as_ref() .map(|cache_context| cache_context.begin_fetch(CodexAppsToolsFetchSource::HardRefresh)); - let tools = list_tools_for_client_uncached( + let tools = list_tools_for_client_uncached_with_cache_retry( CODEX_APPS_MCP_SERVER_NAME, /*is_codex_apps_mcp_server*/ true, - /*codex_apps_refresh_trigger*/ "explicit", &managed_client.client, managed_client.tool_timeout, managed_client.server_instructions.as_deref(), + CodexAppsToolsCacheRetryContext { + cache_context: managed_client.codex_apps_tools_cache_context.as_ref(), + source: CodexAppsToolsFetchSource::HardRefresh, + refresh_trigger: "explicit", + }, ) .await .with_context(|| { format!("failed to refresh tools for MCP server '{CODEX_APPS_MCP_SERVER_NAME}'") })?; - let tools = match ( managed_client.codex_apps_tools_cache_context.as_ref(), diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index 41fc19a594f5..21ba2466ff8a 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -81,6 +81,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(), ) } diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index 87ea25dd9f16..41ec3524cd0e 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -161,6 +161,12 @@ pub struct ToolPluginProvenance { } impl ToolPluginProvenance { + pub(crate) fn connector_ids(&self) -> impl Iterator { + self.plugin_display_names_by_connector_id + .keys() + .map(String::as_str) + } + pub fn plugin_display_names_for_connector_id(&self, connector_id: &str) -> &[String] { self.plugin_display_names_by_connector_id .get(connector_id) diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 6fe820b4fe3f..9fce11311d37 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -9,6 +9,7 @@ use std::borrow::Cow; use std::collections::BTreeMap; use std::collections::HashMap; +use std::collections::HashSet; use std::env; use std::ffi::OsString; use std::sync::Arc; @@ -18,6 +19,7 @@ use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; +use crate::auth_elicitation::MCP_TOOL_CODEX_APPS_META_KEY; use crate::codex_apps::normalize_codex_apps_callable_name; use crate::codex_apps::normalize_codex_apps_callable_namespace; use crate::codex_apps::normalize_codex_apps_tool_title; @@ -612,6 +614,138 @@ pub(crate) async fn list_tools_for_client_uncached( Ok(tools) } +/// Context needed to validate a Codex Apps tools/list response. +pub(crate) struct CodexAppsToolsCacheRetryContext<'a> { + pub(crate) cache_context: Option<&'a CodexAppsToolsCacheContext>, + pub(crate) source: CodexAppsToolsFetchSource, + pub(crate) refresh_trigger: &'static str, +} + +/// Fetches Codex Apps tools and retries one ambiguous connector-group omission +/// before allowing it to become authoritative. +/// +/// Plugin provenance identifies declared connector IDs, but one tools/list +/// response cannot distinguish a transient omission from a real disconnect. +/// A first omission is retried before callers publish anything to the shared +/// cache. Connector groups that remain absent in the second response are +/// treated as confirmed removals. +pub(crate) async fn list_tools_for_client_uncached_with_cache_retry( + server_name: &str, + is_codex_apps_mcp_server: bool, + client: &Arc, + timeout: Option, + server_instructions: Option<&str>, + cache_retry: CodexAppsToolsCacheRetryContext<'_>, +) -> Result> { + let CodexAppsToolsCacheRetryContext { + cache_context, + source, + refresh_trigger, + } = cache_retry; + let Some(cache_context) = cache_context else { + return list_tools_for_client_uncached( + server_name, + is_codex_apps_mcp_server, + refresh_trigger, + client, + timeout, + server_instructions, + ) + .await; + }; + let baseline_tools = cache_context.current_tools(); + let mut first_missing_connector_ids = None; + let deadline = timeout.map(|timeout| Instant::now() + timeout); + loop { + let attempt_timeout = + deadline.map(|deadline| deadline.saturating_duration_since(Instant::now())); + if attempt_timeout.is_some_and(|timeout| timeout.is_zero()) { + return Err(anyhow!( + "Codex Apps tools snapshot retry exceeded the tools/list timeout" + )); + } + let tools = list_tools_for_client_uncached( + server_name, + is_codex_apps_mcp_server, + refresh_trigger, + client, + attempt_timeout, + server_instructions, + ) + .await?; + let missing_connector_ids = missing_expected_connector_ids( + cache_context.expected_connector_ids(), + baseline_tools.as_deref(), + &tools, + ); + if missing_connector_ids.is_empty() { + return Ok(tools); + } + let Some(first_missing_connector_ids) = first_missing_connector_ids.as_ref() else { + tracing::warn!( + source = source.as_str(), + ?missing_connector_ids, + "retrying Codex Apps tools snapshot with connector group drop" + ); + first_missing_connector_ids = Some(missing_connector_ids); + continue; + }; + if retry_confirms_connector_drop(first_missing_connector_ids, &missing_connector_ids) { + return Ok(tools); + } + return Err(anyhow!( + "Codex Apps tools snapshot changed connector omissions after retry" + )); + } +} + +fn missing_expected_connector_ids( + expected_connector_ids: &[String], + baseline_tools: Option<&[ToolInfo]>, + tools: &[ToolInfo], +) -> Vec { + let present_connector_ids = connector_ids_with_non_synthetic_tools(tools); + let baseline_connector_ids = baseline_tools.map(connector_ids_with_non_synthetic_tools); + expected_connector_ids + .iter() + .filter(|connector_id| { + baseline_connector_ids + .as_ref() + .is_none_or(|baseline| baseline.contains(connector_id.as_str())) + && !present_connector_ids.contains(connector_id.as_str()) + }) + .cloned() + .collect() +} + +fn retry_confirms_connector_drop( + first_missing_connector_ids: &[String], + retry_missing_connector_ids: &[String], +) -> bool { + retry_missing_connector_ids + .iter() + .all(|connector_id| first_missing_connector_ids.contains(connector_id)) +} + +fn connector_ids_with_non_synthetic_tools(tools: &[ToolInfo]) -> HashSet<&str> { + tools + .iter() + .filter(|tool| !is_synthetic_link_tool(tool)) + .filter_map(|tool| tool.connector_id.as_deref()) + .collect() +} + +fn is_synthetic_link_tool(tool: &ToolInfo) -> bool { + tool.tool + .meta + .as_deref() + .and_then(|meta| meta.get(MCP_TOOL_CODEX_APPS_META_KEY)) + .and_then(serde_json::Value::as_object) + .and_then(|meta| meta.get("synthetic_link")) + .and_then(serde_json::Value::as_bool) + == Some(true) +} + /// Presents declared Codex Apps file parameters to the model as local-path inputs and adds plugin /// names to each tool. Plugin membership is resolved by connector ID, falling back to the MCP /// server when absent. @@ -849,13 +983,17 @@ async fn start_server_task( let fetch_ticket = codex_apps_tools_cache_context .as_ref() .map(|cache_context| cache_context.begin_fetch(CodexAppsToolsFetchSource::Startup)); - let tools = list_tools_for_client_uncached( + let tools = list_tools_for_client_uncached_with_cache_retry( &server_name, is_codex_apps_mcp_server, - /*codex_apps_refresh_trigger*/ "initial", &client, startup_timeout, initialize_result.instructions.as_deref(), + CodexAppsToolsCacheRetryContext { + cache_context: codex_apps_tools_cache_context.as_ref(), + source: CodexAppsToolsFetchSource::Startup, + refresh_trigger: "initial", + }, ) .await .map_err(StartupOutcomeError::from)?; @@ -1088,6 +1226,74 @@ mod tests { )) } + fn connector_tool_info(connector_id: &str, synthetic_link: bool) -> ToolInfo { + let mut tool = RmcpTool::new("search", "test tool", Arc::new(JsonObject::default())); + if synthetic_link { + let mut meta = Meta::new(); + meta.0.insert( + MCP_TOOL_CODEX_APPS_META_KEY.to_string(), + serde_json::json!({ "synthetic_link": true }), + ); + tool.meta = Some(meta); + } + ToolInfo { + server_name: CODEX_APPS_MCP_SERVER_NAME.to_string(), + supports_parallel_tool_calls: false, + server_origin: None, + callable_name: "search".to_string(), + callable_namespace: "codex_apps__test".to_string(), + namespace_description: None, + tool, + connector_id: Some(connector_id.to_string()), + connector_name: Some("Test".to_string()), + plugin_display_names: Vec::new(), + } + } + + #[test] + fn missing_expected_connector_ids_retries_cold_start_and_ignores_synthetic_links() { + let expected_connector_ids = vec!["connector_docs".to_string()]; + let synthetic_tool = connector_tool_info("connector_docs", /*synthetic_link*/ true); + + assert_eq!( + missing_expected_connector_ids( + &expected_connector_ids, + /*baseline_tools*/ None, + &[synthetic_tool], + ), + expected_connector_ids + ); + } + + #[test] + fn missing_expected_connector_ids_checks_only_baseline_connectors() { + let expected_connector_ids = + vec!["connector_docs".to_string(), "connector_mail".to_string()]; + let baseline_tools = vec![connector_tool_info( + "connector_docs", + /*synthetic_link*/ false, + )]; + + assert_eq!( + missing_expected_connector_ids(&expected_connector_ids, Some(&baseline_tools), &[]), + vec!["connector_docs".to_string()] + ); + } + + #[test] + fn retry_confirms_only_existing_connector_drops() { + let first_missing = vec!["connector_docs".to_string(), "connector_mail".to_string()]; + + assert!(retry_confirms_connector_drop( + &first_missing, + &["connector_docs".to_string()] + )); + assert!(!retry_confirms_connector_drop( + &["connector_docs".to_string()], + &first_missing + )); + } + #[test] fn custom_mcp_connector_metadata_is_stripped() { let mut tool = tool_with_connector_meta(); diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 6e643f34d5ca..74ebbe7b8149 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -267,7 +267,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( mcp_config.prefix_mcp_tool_names, mcp_config.client_elicitation_capability, /*supports_openai_form_elicitation*/ false, - ToolPluginProvenance::default(), + tool_plugin_provenance.clone(), auth.as_ref(), /*elicitation_reviewer*/ None, /*elicitation_lifecycle*/ None, @@ -275,21 +275,21 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( ) .await; - let refreshed_tools = if force_refetch { + let (refreshed_tools, force_refetch_failed) = if force_refetch { match mcp_connection_manager .hard_refresh_codex_apps_tools_cache() .await { - Ok(tools) => Some(tools), + Ok(tools) => (Some(tools), false), Err(err) => { warn!( "failed to force-refresh tools for MCP server '{CODEX_APPS_MCP_SERVER_NAME}', using cached/startup tools: {err:#}" ); - None + (None, true) } } } else { - None + (None, false) }; let refreshed_tools_succeeded = refreshed_tools.is_some(); @@ -299,7 +299,9 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( mcp_connection_manager.list_all_tools().await }; let mut should_reload_tools = false; - let codex_apps_ready = if refreshed_tools_succeeded { + let codex_apps_ready = if force_refetch_failed { + false + } else if refreshed_tools_succeeded { true } else if let Some(cfg) = mcp_servers.get(CODEX_APPS_MCP_SERVER_NAME) { let immediate_ready = mcp_connection_manager @@ -331,7 +333,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( } let accessible_connectors = accessible_connectors_for_app_list_from_mcp_tools(&tools); - if codex_apps_ready || !accessible_connectors.is_empty() { + if !force_refetch_failed && codex_apps_ready { write_cached_accessible_connectors(cache_key, &accessible_connectors); } let accessible_connectors =