From 4e858f89db0dbd8a822d31022f9c29fed53c4ad3 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Tue, 19 May 2026 15:12:43 -0700 Subject: [PATCH] Use request originator for connector filtering --- codex-rs/app-server/src/message_processor.rs | 6 +- codex-rs/app-server/src/outgoing_message.rs | 4 + .../src/request_processors/apps_processor.rs | 68 ++++++++++++---- .../request_processors/config_processor.rs | 25 ++++-- .../src/request_processors/plugins.rs | 62 ++++++++++----- codex-rs/chatgpt/src/chatgpt_client.rs | 13 +++- codex-rs/chatgpt/src/connectors.rs | 78 ++++++++++++++++--- codex-rs/core/src/connectors.rs | 36 +++++---- codex-rs/core/src/connectors_tests.rs | 9 ++- codex-rs/core/src/mcp_tool_call.rs | 13 ++-- codex-rs/core/src/session/turn.rs | 1 + .../tools/handlers/request_plugin_install.rs | 2 + 12 files changed, 243 insertions(+), 74 deletions(-) diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index f14eee2fe2..68cabdd9bb 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -899,7 +899,7 @@ impl MessageProcessor { .map(Some), ClientRequest::ExperimentalFeatureEnablementSet { params, .. } => { self.config_processor - .experimental_feature_enablement_set(request_id.clone(), params) + .experimental_feature_enablement_set(&request_context, params) .await } ClientRequest::RemoteControlEnable { .. } => self @@ -1125,7 +1125,9 @@ impl MessageProcessor { self.plugin_processor.plugin_share_delete(params).await } ClientRequest::AppsList { params, .. } => { - self.apps_processor.apps_list(&request_id, params).await + self.apps_processor + .apps_list(&request_context, params) + .await } ClientRequest::SkillsConfigWrite { params, .. } => { self.catalog_processor.skills_config_write(params).await diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 4fcae1439f..c5d782daca 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -75,6 +75,10 @@ impl RequestContext { self } + pub(crate) fn request_id(&self) -> &ConnectionRequestId { + &self.request_id + } + pub(crate) fn originator(&self) -> &Originator { &self.originator } diff --git a/codex-rs/app-server/src/request_processors/apps_processor.rs b/codex-rs/app-server/src/request_processors/apps_processor.rs index caf60a91a0..72f9aeeabc 100644 --- a/codex-rs/app-server/src/request_processors/apps_processor.rs +++ b/codex-rs/app-server/src/request_processors/apps_processor.rs @@ -28,17 +28,17 @@ impl AppsRequestProcessor { pub(crate) async fn apps_list( &self, - request_id: &ConnectionRequestId, + request_context: &RequestContext, params: AppsListParams, ) -> Result, JSONRPCErrorError> { - self.apps_list_inner(request_id, params) + self.apps_list_inner(request_context, params) .await .map(|response| response.map(Into::into)) } async fn apps_list_inner( &self, - request_id: &ConnectionRequestId, + request_context: &RequestContext, params: AppsListParams, ) -> Result, JSONRPCErrorError> { let thread = if let Some(thread_id) = params.thread_id.as_deref() { @@ -80,11 +80,20 @@ impl AppsRequestProcessor { })); } - let request = request_id.clone(); + let request = request_context.request_id().clone(); let outgoing = Arc::clone(&self.outgoing); let environment_manager = self.thread_manager.environment_manager(); + let originator = request_context.originator().clone(); tokio::spawn(async move { - Self::apps_list_task(outgoing, request, params, config, environment_manager).await; + Self::apps_list_task( + outgoing, + request, + params, + config, + environment_manager, + originator, + ) + .await; }); Ok(None) } @@ -95,11 +104,15 @@ impl AppsRequestProcessor { params: AppsListParams, config: Config, environment_manager: Arc, + originator: Originator, ) { let retry_params = params.clone(); let retry_config = config.clone(); let retry_environment_manager = Arc::clone(&environment_manager); - let result = Self::apps_list_response(&outgoing, params, config, environment_manager).await; + let retry_originator = originator.clone(); + let result = + Self::apps_list_response(&outgoing, params, config, environment_manager, originator) + .await; let should_retry = result .as_ref() .is_ok_and(|(_, codex_apps_ready)| !codex_apps_ready); @@ -115,6 +128,7 @@ impl AppsRequestProcessor { retry_params, retry_config, retry_environment_manager, + retry_originator, ) .await { @@ -128,6 +142,7 @@ impl AppsRequestProcessor { params: AppsListParams, config: Config, environment_manager: Arc, + originator: Originator, ) -> Result<(AppsListResponse, bool), JSONRPCErrorError> { let AppsListParams { cursor, @@ -143,9 +158,13 @@ impl AppsRequestProcessor { None => 0, }; + let originator_value = originator.value().to_string(); let (mut accessible_connectors, mut all_connectors) = tokio::join!( - connectors::list_cached_accessible_connectors_from_mcp_tools(&config), - connectors::list_cached_all_connectors(&config) + connectors::list_cached_accessible_connectors_from_mcp_tools( + &config, + &originator_value + ), + connectors::list_cached_all_connectors_with_originator(&config, &originator) ); let cached_all_connectors = all_connectors.clone(); @@ -153,12 +172,14 @@ impl AppsRequestProcessor { let accessible_config = config.clone(); let accessible_tx = tx.clone(); + let accessible_originator_value = originator_value.clone(); tokio::spawn(async move { let result = connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager( &accessible_config, force_refetch, &environment_manager, + &accessible_originator_value, ) .await .map_err(|err| format!("failed to load accessible apps: {err}")); @@ -166,10 +187,15 @@ impl AppsRequestProcessor { }); let all_config = config.clone(); + let all_originator = originator.clone(); tokio::spawn(async move { - let result = connectors::list_all_connectors_with_options(&all_config, force_refetch) - .await - .map_err(|err| format!("failed to list apps: {err}")); + let result = connectors::list_all_connectors_with_options_and_originator( + &all_config, + force_refetch, + &all_originator, + ) + .await + .map_err(|err| format!("failed to list apps: {err}")); let _ = tx.send(AppListLoadResult::Directory(result)); }); @@ -181,7 +207,11 @@ impl AppsRequestProcessor { if accessible_connectors.is_some() || all_connectors.is_some() { let merged = connectors::with_app_enabled_state( - merge_loaded_apps(all_connectors.as_deref(), accessible_connectors.as_deref()), + merge_loaded_apps( + all_connectors.as_deref(), + accessible_connectors.as_deref(), + &originator_value, + ), &config, ); if should_send_app_list_updated_notification( @@ -240,7 +270,11 @@ impl AppsRequestProcessor { accessible_connectors.as_deref() }; let merged = connectors::with_app_enabled_state( - merge_loaded_apps(all_connectors_for_update, accessible_connectors_for_update), + merge_loaded_apps( + all_connectors_for_update, + accessible_connectors_for_update, + &originator_value, + ), &config, ); if should_send_app_list_updated_notification( @@ -319,11 +353,17 @@ enum AppListLoadResult { fn merge_loaded_apps( all_connectors: Option<&[AppInfo]>, accessible_connectors: Option<&[AppInfo]>, + originator_value: &str, ) -> Vec { let all_connectors_loaded = all_connectors.is_some(); let all = all_connectors.map_or_else(Vec::new, <[AppInfo]>::to_vec); let accessible = accessible_connectors.map_or_else(Vec::new, <[AppInfo]>::to_vec); - connectors::merge_connectors_with_accessible(all, accessible, all_connectors_loaded) + connectors::merge_connectors_with_accessible_for_originator( + all, + accessible, + all_connectors_loaded, + originator_value, + ) } fn should_send_app_list_updated_notification( diff --git a/codex-rs/app-server/src/request_processors/config_processor.rs b/codex-rs/app-server/src/request_processors/config_processor.rs index 9c13242960..05bd020b07 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/codex-rs/app-server/src/request_processors/config_processor.rs @@ -4,8 +4,8 @@ use crate::config_manager::ConfigManager; use crate::config_manager_service::ConfigManagerError; use crate::error_code::internal_error; use crate::error_code::invalid_request; -use crate::outgoing_message::ConnectionRequestId; use crate::outgoing_message::OutgoingMessageSender; +use crate::outgoing_message::RequestContext; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::AppListUpdatedNotification; use codex_app_server_protocol::ClientResponsePayload; @@ -148,21 +148,22 @@ impl ConfigRequestProcessor { pub(crate) async fn experimental_feature_enablement_set( &self, - request_id: ConnectionRequestId, + request_context: &RequestContext, params: ExperimentalFeatureEnablementSetParams, ) -> Result, JSONRPCErrorError> { + let originator = request_context.originator().clone(); let should_refresh_apps_list = params.enablement.get("apps").copied() == Some(true); let response = self .handle_config_mutation_result(self.set_experimental_feature_enablement(params).await) .await?; self.outgoing .send_response_as( - request_id, + request_context.request_id().clone(), ClientResponsePayload::ExperimentalFeatureEnablementSet(response), ) .await; if should_refresh_apps_list { - self.refresh_apps_list_after_experimental_feature_enablement_set() + self.refresh_apps_list_after_experimental_feature_enablement_set(originator) .await; } Ok(None) @@ -195,7 +196,10 @@ impl ConfigRequestProcessor { Ok(response) } - async fn refresh_apps_list_after_experimental_feature_enablement_set(&self) { + async fn refresh_apps_list_after_experimental_feature_enablement_set( + &self, + originator: Originator, + ) { let config = match self.load_latest_config(/*fallback_cwd*/ None).await { Ok(config) => config, Err(error) => { @@ -217,12 +221,18 @@ impl ConfigRequestProcessor { let outgoing = Arc::clone(&self.outgoing); let environment_manager = self.thread_manager.environment_manager(); tokio::spawn(async move { + let originator_value = originator.value().to_string(); let (all_connectors_result, accessible_connectors_result) = tokio::join!( - connectors::list_all_connectors_with_options(&config, /*force_refetch*/ true), + connectors::list_all_connectors_with_options_and_originator( + &config, + /*force_refetch*/ true, + &originator, + ), connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager( &config, /*force_refetch*/ true, &environment_manager, + &originator_value, ), ); let all_connectors = match all_connectors_result { @@ -245,10 +255,11 @@ impl ConfigRequestProcessor { }; let data = connectors::with_app_enabled_state( - connectors::merge_connectors_with_accessible( + connectors::merge_connectors_with_accessible_for_originator( all_connectors, accessible_connectors, /*all_connectors_loaded*/ true, + &originator_value, ), &config, ); diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 64e3697408..830d014e69 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -1486,12 +1486,19 @@ impl PluginRequestProcessor { } let environment_manager = self.thread_manager.environment_manager(); + let originator = Originator::process_default(); + let originator_value = originator.value().to_string(); let (all_connectors_result, accessible_connectors_result) = tokio::join!( - connectors::list_all_connectors_with_options(config, /*force_refetch*/ true), + connectors::list_all_connectors_with_options_and_originator( + config, + /*force_refetch*/ true, + &originator, + ), connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager( config, /*force_refetch*/ true, - &environment_manager + &environment_manager, + &originator_value, ), ); @@ -1502,12 +1509,16 @@ impl PluginRequestProcessor { plugin = plugin_id, "failed to load app metadata after plugin install: {err:#}" ); - connectors::list_cached_all_connectors(config) + connectors::list_cached_all_connectors_with_originator(config, &originator) .await .unwrap_or_default() } }; - let all_connectors = connectors::connectors_for_plugin_apps(all_connectors, plugin_apps); + let all_connectors = connectors::connectors_for_plugin_apps_for_originator( + all_connectors, + plugin_apps, + &originator_value, + ); let (accessible_connectors, codex_apps_ready) = match accessible_connectors_result { Ok(status) => (status.connectors, status.codex_apps_ready), Err(err) => { @@ -1516,9 +1527,12 @@ impl PluginRequestProcessor { "failed to load accessible apps after plugin install: {err:#}" ); ( - connectors::list_cached_accessible_connectors_from_mcp_tools(config) - .await - .unwrap_or_default(), + connectors::list_cached_accessible_connectors_from_mcp_tools( + config, + &originator_value, + ) + .await + .unwrap_or_default(), false, ) } @@ -1764,24 +1778,36 @@ async fn load_plugin_app_summaries( return Vec::new(); } - let connectors = - match connectors::list_all_connectors_with_options(config, /*force_refetch*/ false).await { - Ok(connectors) => connectors, - Err(err) => { - warn!("failed to load app metadata for plugin/read: {err:#}"); - connectors::list_cached_all_connectors(config) - .await - .unwrap_or_default() - } - }; + let originator = Originator::process_default(); + let originator_value = originator.value().to_string(); + let connectors = match connectors::list_all_connectors_with_options_and_originator( + config, + /*force_refetch*/ false, + &originator, + ) + .await + { + Ok(connectors) => connectors, + Err(err) => { + warn!("failed to load app metadata for plugin/read: {err:#}"); + connectors::list_cached_all_connectors_with_originator(config, &originator) + .await + .unwrap_or_default() + } + }; - let plugin_connectors = connectors::connectors_for_plugin_apps(connectors, plugin_apps); + let plugin_connectors = connectors::connectors_for_plugin_apps_for_originator( + connectors, + plugin_apps, + &originator_value, + ); let accessible_connectors = match connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager( config, /*force_refetch*/ false, environment_manager, + &originator_value, ) .await { diff --git a/codex-rs/chatgpt/src/chatgpt_client.rs b/codex-rs/chatgpt/src/chatgpt_client.rs index 099f33e183..15178fd31b 100644 --- a/codex-rs/chatgpt/src/chatgpt_client.rs +++ b/codex-rs/chatgpt/src/chatgpt_client.rs @@ -22,6 +22,16 @@ pub(crate) async fn chatgpt_get_request_with_timeout( config: &Config, path: String, timeout: Option, +) -> anyhow::Result { + let originator = Originator::process_default(); + chatgpt_get_request_with_timeout_and_originator(config, path, timeout, &originator).await +} + +pub(crate) async fn chatgpt_get_request_with_timeout_and_originator( + config: &Config, + path: String, + timeout: Option, + originator: &Originator, ) -> anyhow::Result { let chatgpt_base_url = &config.chatgpt_base_url; let auth_manager = @@ -40,8 +50,7 @@ pub(crate) async fn chatgpt_get_request_with_timeout( ); // Make direct HTTP request to ChatGPT backend API with the token - let originator = Originator::process_default(); - let client = create_client(&originator); + let client = create_client(originator); let url = format!( "{}/{}", chatgpt_base_url.trim_end_matches('/'), diff --git a/codex-rs/chatgpt/src/connectors.rs b/codex-rs/chatgpt/src/connectors.rs index 26102bc968..449735da78 100644 --- a/codex-rs/chatgpt/src/connectors.rs +++ b/codex-rs/chatgpt/src/connectors.rs @@ -1,7 +1,7 @@ use std::collections::HashSet; use std::time::Duration; -use crate::chatgpt_client::chatgpt_get_request_with_timeout; +use crate::chatgpt_client::chatgpt_get_request_with_timeout_and_originator; use codex_app_server_protocol::AppInfo; use codex_connectors::ConnectorDirectoryCacheContext; @@ -52,9 +52,11 @@ pub async fn list_connectors(config: &Config) -> anyhow::Result> { if !apps_enabled(config).await { return Ok(Vec::new()); } + let originator = Originator::process_default(); + let originator_value = originator.value().to_string(); let (connectors_result, accessible_result) = tokio::join!( - list_all_connectors(config), - list_accessible_connectors_from_mcp_tools(config), + list_all_connectors_with_originator(config, &originator), + list_accessible_connectors_from_mcp_tools(config, &originator_value), ); let connectors = connectors_result?; let accessible = accessible_result?; @@ -67,10 +69,24 @@ pub async fn list_connectors(config: &Config) -> anyhow::Result> { } pub async fn list_all_connectors(config: &Config) -> anyhow::Result> { - list_all_connectors_with_options(config, /*force_refetch*/ false).await + let originator = Originator::process_default(); + list_all_connectors_with_options_and_originator( + config, + /*force_refetch*/ false, + &originator, + ) + .await } pub async fn list_cached_all_connectors(config: &Config) -> Option> { + let originator = Originator::process_default(); + list_cached_all_connectors_with_originator(config, &originator).await +} + +pub async fn list_cached_all_connectors_with_originator( + config: &Config, + originator: &Originator, +) -> Option> { if !apps_enabled(config).await { return Some(Vec::new()); } @@ -85,13 +101,31 @@ pub async fn list_cached_all_connectors(config: &Config) -> Option> .into_iter() .map(|connector_id| connector_id.0), ); - let originator = Originator::process_default(); Some(filter_disallowed_connectors(connectors, originator.value())) } pub async fn list_all_connectors_with_options( config: &Config, force_refetch: bool, +) -> anyhow::Result> { + let originator = Originator::process_default(); + list_all_connectors_with_options_and_originator(config, force_refetch, &originator).await +} + +pub async fn list_all_connectors_with_originator( + config: &Config, + originator: &Originator, +) -> anyhow::Result> { + list_all_connectors_with_options_and_originator( + config, /*force_refetch*/ false, originator, + ) + .await +} + +pub async fn list_all_connectors_with_options_and_originator( + config: &Config, + force_refetch: bool, + originator: &Originator, ) -> anyhow::Result> { if !apps_enabled(config).await { return Ok(Vec::new()); @@ -103,10 +137,11 @@ pub async fn list_all_connectors_with_options( auth.is_workspace_account(), force_refetch, |path| async move { - chatgpt_get_request_with_timeout::( + chatgpt_get_request_with_timeout_and_originator::( config, path, Some(DIRECTORY_CONNECTORS_TIMEOUT), + originator, ) .await }, @@ -119,7 +154,6 @@ pub async fn list_all_connectors_with_options( .into_iter() .map(|connector_id| connector_id.0), ); - let originator = Originator::process_default(); Ok(filter_disallowed_connectors(connectors, originator.value())) } @@ -149,6 +183,15 @@ async fn plugin_apps_for_config(config: &Config) -> Vec { pub fn connectors_for_plugin_apps( connectors: Vec, plugin_apps: &[AppConnectorId], +) -> Vec { + let originator = Originator::process_default(); + connectors_for_plugin_apps_for_originator(connectors, plugin_apps, originator.value()) +} + +pub fn connectors_for_plugin_apps_for_originator( + connectors: Vec, + plugin_apps: &[AppConnectorId], + originator_value: &str, ) -> Vec { let plugin_app_ids = plugin_apps .iter() @@ -161,8 +204,7 @@ pub fn connectors_for_plugin_apps( .iter() .map(|connector_id| connector_id.0.clone()), ); - let originator = Originator::process_default(); - filter_disallowed_connectors(connectors, originator.value()) + filter_disallowed_connectors(connectors, originator_value) .into_iter() .filter(|connector| plugin_app_ids.contains(connector.id.as_str())) .collect() @@ -172,6 +214,21 @@ pub fn merge_connectors_with_accessible( connectors: Vec, accessible_connectors: Vec, all_connectors_loaded: bool, +) -> Vec { + let originator = Originator::process_default(); + merge_connectors_with_accessible_for_originator( + connectors, + accessible_connectors, + all_connectors_loaded, + originator.value(), + ) +} + +pub fn merge_connectors_with_accessible_for_originator( + connectors: Vec, + accessible_connectors: Vec, + all_connectors_loaded: bool, + originator_value: &str, ) -> Vec { let accessible_connectors = if all_connectors_loaded { let connector_ids: HashSet<&str> = connectors @@ -186,8 +243,7 @@ pub fn merge_connectors_with_accessible( accessible_connectors }; let merged = merge_connectors(connectors, accessible_connectors); - let originator = Originator::process_default(); - filter_disallowed_connectors(merged, originator.value()) + filter_disallowed_connectors(merged, originator_value) } #[cfg(test)] diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 3a132d0e97..30949a0b36 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -32,7 +32,6 @@ use codex_core_plugins::PluginsManager; use codex_features::Feature; use codex_login::AuthManager; use codex_login::CodexAuth; -use codex_login::default_client::Originator; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::McpConnectionManager; use codex_mcp::McpRuntimeEnvironment; @@ -86,10 +85,13 @@ pub struct AccessibleConnectorsStatus { pub async fn list_accessible_connectors_from_mcp_tools( config: &Config, + originator_value: &str, ) -> anyhow::Result> { Ok( list_accessible_connectors_from_mcp_tools_with_options_and_status( - config, /*force_refetch*/ false, + config, + /*force_refetch*/ false, + originator_value, ) .await? .connectors, @@ -113,6 +115,7 @@ pub(crate) async fn list_tool_suggest_discoverable_tools_with_auth( config: &Config, auth: Option<&CodexAuth>, accessible_connectors: &[AppInfo], + originator_value: &str, ) -> anyhow::Result> { let connector_ids = tool_suggest_connector_ids(config).await; let directory_connectors = codex_connectors::merge::merge_plugin_connectors( @@ -124,7 +127,7 @@ pub(crate) async fn list_tool_suggest_discoverable_tools_with_auth( directory_connectors, accessible_connectors, &connector_ids, - Originator::process_default().value(), + originator_value, ) .into_iter() .map(DiscoverableTool::from); @@ -139,6 +142,7 @@ pub(crate) async fn list_tool_suggest_discoverable_tools_with_auth( pub async fn list_cached_accessible_connectors_from_mcp_tools( config: &Config, + originator_value: &str, ) -> Option> { let auth_manager = AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await; @@ -151,10 +155,7 @@ pub async fn list_cached_accessible_connectors_from_mcp_tools( } let cache_key = accessible_connectors_cache_key(config, auth.as_ref()); read_cached_accessible_connectors(&cache_key).map(|connectors| { - codex_connectors::filter::filter_disallowed_connectors( - connectors, - Originator::process_default().value(), - ) + codex_connectors::filter::filter_disallowed_connectors(connectors, originator_value) }) } @@ -162,6 +163,7 @@ pub(crate) fn refresh_accessible_connectors_cache_from_mcp_tools( config: &Config, auth: Option<&CodexAuth>, mcp_tools: &[ToolInfo], + originator_value: &str, ) { if !config.features.enabled(Feature::Apps) { return; @@ -170,7 +172,7 @@ pub(crate) fn refresh_accessible_connectors_cache_from_mcp_tools( let cache_key = accessible_connectors_cache_key(config, auth); let accessible_connectors = codex_connectors::filter::filter_disallowed_connectors( accessible_connectors_from_mcp_tools(mcp_tools), - Originator::process_default().value(), + originator_value, ); write_cached_accessible_connectors(cache_key, &accessible_connectors); } @@ -178,17 +180,23 @@ pub(crate) fn refresh_accessible_connectors_cache_from_mcp_tools( pub async fn list_accessible_connectors_from_mcp_tools_with_options( config: &Config, force_refetch: bool, + originator_value: &str, ) -> anyhow::Result> { Ok( - list_accessible_connectors_from_mcp_tools_with_options_and_status(config, force_refetch) - .await? - .connectors, + list_accessible_connectors_from_mcp_tools_with_options_and_status( + config, + force_refetch, + originator_value, + ) + .await? + .connectors, ) } pub async fn list_accessible_connectors_from_mcp_tools_with_options_and_status( config: &Config, force_refetch: bool, + originator_value: &str, ) -> anyhow::Result { // TODO: Wire callers that already own an EnvironmentManager into // list_accessible_connectors_from_mcp_tools_with_environment_manager instead @@ -203,6 +211,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_options_and_status( config, force_refetch, &environment_manager, + originator_value, ) .await } @@ -211,6 +220,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_environment_manager( config: &Config, force_refetch: bool, environment_manager: &EnvironmentManager, + originator_value: &str, ) -> anyhow::Result { let auth_manager = AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await; @@ -232,7 +242,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_environment_manager( { let cached_connectors = codex_connectors::filter::filter_disallowed_connectors( cached_connectors, - Originator::process_default().value(), + originator_value, ); let cached_connectors = with_app_plugin_sources(cached_connectors, &tool_plugin_provenance); return Ok(AccessibleConnectorsStatus { @@ -341,7 +351,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_environment_manager( let accessible_connectors = codex_connectors::filter::filter_disallowed_connectors( accessible_connectors_from_mcp_tools(&tools), - Originator::process_default().value(), + originator_value, ); if codex_apps_ready || !accessible_connectors.is_empty() { write_cached_accessible_connectors(cache_key, &accessible_connectors); diff --git a/codex-rs/core/src/connectors_tests.rs b/codex-rs/core/src/connectors_tests.rs index cce2491679..708512ae4d 100644 --- a/codex-rs/core/src/connectors_tests.rs +++ b/codex-rs/core/src/connectors_tests.rs @@ -193,7 +193,12 @@ async fn refresh_accessible_connectors_cache_from_mcp_tools_writes_latest_instal ]; let cached = with_accessible_connectors_cache_cleared(|| { - refresh_accessible_connectors_cache_from_mcp_tools(&config, /*auth*/ None, &tools); + refresh_accessible_connectors_cache_from_mcp_tools( + &config, + /*auth*/ None, + &tools, + "codex_vscode", + ); read_cached_accessible_connectors(&cache_key).expect("cache should be populated") }); @@ -1254,7 +1259,7 @@ discoverables = [ let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); let discoverable_tools = - list_tool_suggest_discoverable_tools_with_auth(&config, Some(&auth), &[]) + list_tool_suggest_discoverable_tools_with_auth(&config, Some(&auth), &[], "codex_vscode") .await .expect("discoverable tools should load"); diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index 06d874d8f7..9e51c51952 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -690,6 +690,7 @@ async fn refresh_codex_apps_after_connector_auth(sess: &Session, turn_context: & &turn_context.config, auth.as_ref(), &mcp_tools, + turn_context.originator.value(), ); } Err(err) => { @@ -1500,15 +1501,17 @@ pub(crate) async fn lookup_mcp_tool_metadata( let connector_description = if server == CODEX_APPS_MCP_SERVER_NAME { let connectors = match connectors::list_cached_accessible_connectors_from_mcp_tools( turn_context.config.as_ref(), + turn_context.originator.value(), ) .await { Some(connectors) => Some(connectors), - None => { - connectors::list_accessible_connectors_from_mcp_tools(turn_context.config.as_ref()) - .await - .ok() - } + None => connectors::list_accessible_connectors_from_mcp_tools( + turn_context.config.as_ref(), + turn_context.originator.value(), + ) + .await + .ok(), }; connectors.and_then(|connectors| { let connector_id = tool_info.connector_id.as_deref()?; diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 2da3ade2d4..07323293b9 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1097,6 +1097,7 @@ pub(crate) async fn built_tools( &turn_context.config, auth.as_ref(), accessible_connectors.as_slice(), + turn_context.originator.value(), ) .await .map(|discoverable_tools| { diff --git a/codex-rs/core/src/tools/handlers/request_plugin_install.rs b/codex-rs/core/src/tools/handlers/request_plugin_install.rs index 3b53fd007a..4419d77ea5 100644 --- a/codex-rs/core/src/tools/handlers/request_plugin_install.rs +++ b/codex-rs/core/src/tools/handlers/request_plugin_install.rs @@ -123,6 +123,7 @@ impl ToolExecutor for RequestPluginInstallHandler { &turn.config, auth.as_ref(), &accessible_connectors, + turn.originator.value(), ) .await .map(|discoverable_tools| { @@ -334,6 +335,7 @@ async fn refresh_missing_requested_connectors( &turn.config, auth, &mcp_tools, + turn.originator.value(), ); Some(accessible_connectors) }