Freeze selected connector declarations per thread

This commit is contained in:
jif-oai
2026-06-25 00:25:21 +01:00
parent ed5f212879
commit 62c0392bb6
24 changed files with 561 additions and 56 deletions
+2
View File
@@ -1977,6 +1977,7 @@ dependencies = [
"codex-cloud-config",
"codex-config",
"codex-connectors",
"codex-connectors-extension",
"codex-core",
"codex-core-plugins",
"codex-exec-server",
@@ -2603,6 +2604,7 @@ dependencies = [
"codex-connectors",
"codex-core-plugins",
"codex-plugin",
"codex-protocol",
"codex-utils-path-uri",
"serde_json",
"thiserror 2.0.18",
+1
View File
@@ -166,6 +166,7 @@ codex-code-mode-protocol = { path = "code-mode-protocol" }
codex-home = { path = "codex-home" }
codex-config = { path = "config" }
codex-connectors = { path = "connectors" }
codex-connectors-extension = { path = "ext/connectors" }
codex-context-fragments = { path = "context-fragments" }
codex-core = { path = "core" }
codex-core-api = { path = "core-api" }
+1 -1
View File
@@ -111,7 +111,7 @@
]
},
"threadId": {
"description": "Optional thread id used to evaluate app feature gating from that thread's config.",
"description": "Optional thread id used for feature gating and thread-selected plugin associations.",
"type": [
"string",
"null"
@@ -6843,7 +6843,7 @@
]
},
"threadId": {
"description": "Optional thread id used to evaluate app feature gating from that thread's config.",
"description": "Optional thread id used for feature gating and thread-selected plugin associations.",
"type": [
"string",
"null"
@@ -969,7 +969,7 @@
]
},
"threadId": {
"description": "Optional thread id used to evaluate app feature gating from that thread's config.",
"description": "Optional thread id used for feature gating and thread-selected plugin associations.",
"type": [
"string",
"null"
@@ -23,7 +23,7 @@
]
},
"threadId": {
"description": "Optional thread id used to evaluate app feature gating from that thread's config.",
"description": "Optional thread id used for feature gating and thread-selected plugin associations.",
"type": [
"string",
"null"
@@ -15,7 +15,7 @@ cursor?: string | null,
*/
limit?: number | null,
/**
* Optional thread id used to evaluate app feature gating from that thread's config.
* Optional thread id used for feature gating and thread-selected plugin associations.
*/
threadId?: string | null,
/**
@@ -16,7 +16,7 @@ pub struct AppsListParams {
/// Optional page size; defaults to a reasonable server-side value.
#[ts(optional = nullable)]
pub limit: Option<u32>,
/// Optional thread id used to evaluate app feature gating from that thread's config.
/// Optional thread id used for feature gating and thread-selected plugin associations.
#[ts(optional = nullable)]
pub thread_id: Option<String>,
/// When true, bypass app caches and fetch the latest data from sources.
+1
View File
@@ -34,6 +34,7 @@ codex-arg0 = { workspace = true }
codex-cloud-config = { workspace = true }
codex-config = { workspace = true }
codex-connectors = { workspace = true }
codex-connectors-extension = { workspace = true }
codex-core = { workspace = true }
codex-core-plugins = { workspace = true }
codex-home = { workspace = true }
+1 -1
View File
@@ -1821,7 +1821,7 @@ Use `app/list` to fetch available apps (connectors). Each entry includes metadat
} }
```
When `threadId` is provided, app feature gating (`Feature::Apps`) is evaluated using that thread's config snapshot. When omitted, the latest global config is used.
When `threadId` is provided, app feature gating (`Feature::Apps`) is evaluated using that thread's config snapshot, and plugin associations come from its frozen selected-capability snapshot. When omitted, the latest global config is used.
`app/list` returns after both accessible apps and directory apps are loaded. Set `forceRefetch: true` to bypass app caches and fetch fresh data from sources. Cache entries are only replaced when those refetches succeed.
@@ -305,7 +305,6 @@ use codex_connectors::AppInfo;
use codex_core::CodexThread;
use codex_core::CodexThreadSettingsOverrides;
use codex_core::ForkSnapshot;
use codex_core::McpManager;
use codex_core::NewThread;
#[cfg(test)]
use codex_core::SessionMeta;
@@ -335,7 +334,6 @@ use codex_core_plugins::PluginInstallError as CorePluginInstallError;
use codex_core_plugins::PluginInstallRequest;
use codex_core_plugins::PluginReadRequest;
use codex_core_plugins::PluginUninstallError as CorePluginUninstallError;
use codex_core_plugins::PluginsManager;
use codex_core_plugins::loader::load_plugin_apps;
use codex_core_plugins::loader::load_plugin_mcp_servers;
use codex_core_plugins::manifest::PluginManifestInterface;
@@ -59,7 +59,7 @@ impl AppsRequestProcessor {
};
let mut config = self.load_latest_config(fallback_cwd).await?;
if let Some(thread) = thread {
if let Some(thread) = thread.as_ref() {
let _ = config
.features
.set_enabled(Feature::Apps, thread.enabled(Feature::Apps));
@@ -90,7 +90,10 @@ impl AppsRequestProcessor {
let outgoing = Arc::clone(&self.outgoing);
let environment_manager = self.thread_manager.environment_manager();
let mcp_manager = self.thread_manager.mcp_manager();
let plugins_manager = self.thread_manager.plugins_manager();
let mcp_config = match thread.as_ref() {
Some(thread) => thread.runtime_mcp_config(&config).await,
None => mcp_manager.runtime_config(&config).await,
};
let shutdown_token = self.shutdown_token.child_token();
tokio::spawn(async move {
tokio::select! {
@@ -101,8 +104,7 @@ impl AppsRequestProcessor {
params,
config,
environment_manager,
mcp_manager,
plugins_manager,
mcp_config,
) => {}
}
});
@@ -119,23 +121,15 @@ impl AppsRequestProcessor {
params: AppsListParams,
config: Config,
environment_manager: Arc<EnvironmentManager>,
mcp_manager: Arc<McpManager>,
plugins_manager: Arc<PluginsManager>,
mcp_config: codex_mcp::McpConfig,
) {
let retry_params = params.clone();
let retry_config = config.clone();
let retry_environment_manager = Arc::clone(&environment_manager);
let retry_mcp_manager = Arc::clone(&mcp_manager);
let retry_plugins_manager = Arc::clone(&plugins_manager);
let result = Self::apps_list_response(
&outgoing,
params,
config,
environment_manager,
mcp_manager,
plugins_manager,
)
.await;
let retry_mcp_config = mcp_config.clone();
let result =
Self::apps_list_response(&outgoing, params, config, environment_manager, mcp_config)
.await;
let should_retry = result
.as_ref()
.is_ok_and(|(_, codex_apps_ready)| !codex_apps_ready);
@@ -151,8 +145,7 @@ impl AppsRequestProcessor {
retry_params,
retry_config,
retry_environment_manager,
retry_mcp_manager,
retry_plugins_manager,
retry_mcp_config,
)
.await
{
@@ -166,8 +159,7 @@ impl AppsRequestProcessor {
params: AppsListParams,
config: Config,
environment_manager: Arc<EnvironmentManager>,
mcp_manager: Arc<McpManager>,
plugins_manager: Arc<PluginsManager>,
mcp_config: codex_mcp::McpConfig,
) -> Result<(AppsListResponse, bool), JSONRPCErrorError> {
let AppsListParams {
cursor,
@@ -183,13 +175,7 @@ impl AppsRequestProcessor {
None => 0,
};
let loaded_plugins = plugins_manager
.plugins_for_config(&config.plugins_config_input())
.await;
let connector_snapshot =
codex_connectors::ConnectorSnapshot::from_plugin_capability_summaries(
loaded_plugins.capability_summaries(),
);
let connector_snapshot = mcp_config.connector_snapshot.clone();
let plugin_apps = connector_snapshot.connector_ids().to_vec();
let (mut accessible_connectors, mut all_connectors) = tokio::join!(
connectors::list_cached_accessible_connectors_from_mcp_tools(&config),
@@ -202,11 +188,11 @@ impl AppsRequestProcessor {
let accessible_config = config.clone();
let accessible_tx = tx.clone();
tokio::spawn(async move {
let result = connectors::list_accessible_connectors_from_mcp_tools_with_mcp_manager(
let result = connectors::list_accessible_connectors_from_mcp_tools_with_mcp_config(
&accessible_config,
force_refetch,
Arc::clone(&environment_manager),
mcp_manager,
mcp_config,
)
.await
.map_err(|err| format!("failed to load accessible apps: {err}"));
@@ -233,9 +219,12 @@ impl AppsRequestProcessor {
let mut last_notified_apps = None;
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()),
&config,
let merged = with_connector_plugin_sources(
connectors::with_app_enabled_state(
merge_loaded_apps(all_connectors.as_deref(), accessible_connectors.as_deref()),
&config,
),
&connector_snapshot,
);
if should_send_app_list_updated_notification(
merged.as_slice(),
@@ -292,9 +281,12 @@ impl AppsRequestProcessor {
} else {
accessible_connectors.as_deref()
};
let merged = connectors::with_app_enabled_state(
merge_loaded_apps(all_connectors_for_update, accessible_connectors_for_update),
&config,
let merged = with_connector_plugin_sources(
connectors::with_app_enabled_state(
merge_loaded_apps(all_connectors_for_update, accessible_connectors_for_update),
&config,
),
&connector_snapshot,
);
if should_send_app_list_updated_notification(
merged.as_slice(),
@@ -379,6 +371,18 @@ fn merge_loaded_apps(
connectors::merge_connectors_with_accessible(all, accessible, all_connectors_loaded)
}
fn with_connector_plugin_sources(
mut connectors: Vec<AppInfo>,
connector_snapshot: &codex_connectors::ConnectorSnapshot,
) -> Vec<AppInfo> {
for connector in &mut connectors {
connector.plugin_display_names = connector_snapshot
.plugin_display_names_for_connector_id(&connector.id)
.to_vec();
}
connectors
}
fn should_send_app_list_updated_notification(
connectors: &[AppInfo],
accessible_loaded: bool,
+26 -3
View File
@@ -1533,7 +1533,7 @@ impl ServerHandler for AppListMcpServer {
}
}
async fn start_apps_server_with_delays(
pub(super) async fn start_apps_server_with_delays(
connectors: Vec<AppInfo>,
tools: Vec<Tool>,
directory_delay: Duration,
@@ -1545,6 +1545,26 @@ async fn start_apps_server_with_delays(
Ok((server_url, server_handle))
}
pub(super) async fn start_apps_server_with_delays_and_access_token(
connectors: Vec<AppInfo>,
tools: Vec<Tool>,
directory_delay: Duration,
tools_delay: Duration,
access_token: &str,
) -> Result<(String, JoinHandle<()>)> {
let (server_url, server_handle, _server_control) =
start_apps_server_with_delays_and_control_inner(
connectors,
tools,
directory_delay,
tools_delay,
/*workspace_plugins_enabled*/ true,
format!("Bearer {access_token}"),
)
.await?;
Ok((server_url, server_handle))
}
async fn start_apps_server_with_workspace_plugins_enabled(
connectors: Vec<AppInfo>,
tools: Vec<Tool>,
@@ -1557,6 +1577,7 @@ async fn start_apps_server_with_workspace_plugins_enabled(
Duration::ZERO,
Duration::ZERO,
workspace_plugins_enabled,
"Bearer chatgpt-token".to_string(),
)
.await?;
Ok((server_url, server_handle))
@@ -1574,6 +1595,7 @@ async fn start_apps_server_with_delays_and_control(
directory_delay,
tools_delay,
/*workspace_plugins_enabled*/ true,
"Bearer chatgpt-token".to_string(),
)
.await
}
@@ -1584,13 +1606,14 @@ async fn start_apps_server_with_delays_and_control_inner(
directory_delay: Duration,
tools_delay: Duration,
workspace_plugins_enabled: bool,
expected_bearer: String,
) -> Result<(String, JoinHandle<()>, AppsServerControl)> {
let response = Arc::new(StdMutex::new(
json!({ "apps": connectors, "next_token": null }),
));
let tools = Arc::new(StdMutex::new(tools));
let state = AppsServerState {
expected_bearer: "Bearer chatgpt-token".to_string(),
expected_bearer,
expected_account_id: "account-123".to_string(),
response: response.clone(),
directory_delay,
@@ -1693,7 +1716,7 @@ async fn list_directory_connectors(
}
}
fn connector_tool(connector_id: &str, connector_name: &str) -> Result<Tool> {
pub(super) fn connector_tool(connector_id: &str, connector_name: &str) -> Result<Tool> {
let schema: JsonObject = serde_json::from_value(json!({
"type": "object",
"additionalProperties": false
@@ -0,0 +1,315 @@
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use app_test_support::ChatGptIdTokenClaims;
use app_test_support::TestAppServer;
use app_test_support::encode_id_token;
use app_test_support::to_response;
use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url;
use codex_app_server_protocol::AppInfo;
use codex_app_server_protocol::AppsListParams;
use codex_app_server_protocol::AppsListResponse;
use codex_app_server_protocol::CapabilityRootLocation;
use codex_app_server_protocol::LoginAccountResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SelectedCapabilityRoot;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::UserInput;
use codex_exec_server::CreateDirectoryOptions;
use core_test_support::responses;
use core_test_support::test_codex::test_env;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::time::timeout;
use super::app_list::connector_tool;
use super::app_list::start_apps_server_with_delays_and_access_token;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60);
const CONNECTOR_ID: &str = "calendar";
const CONNECTOR_NAME: &str = "Calendar";
const PLUGIN_DISPLAY_NAME: &str = "Executor Calendar";
const TOOL_NAMESPACE: &str = "mcp__codex_apps__calendar";
const TOOL_NAME: &str = "connector_calendar";
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn selected_executor_connector_is_frozen_thread_scoped_and_hosted() -> Result<()> {
let responses_server = responses::start_mock_server().await;
let access_token = encode_id_token(
&ChatGptIdTokenClaims::new()
.email("executor-connectors@example.com")
.plan_type("pro")
.chatgpt_account_id("account-123"),
)?;
let directory_app = expected_app(Vec::new());
let (apps_url, apps_server_handle) = start_apps_server_with_delays_and_access_token(
vec![AppInfo {
is_accessible: false,
install_url: None,
..directory_app.clone()
}],
vec![connector_tool(CONNECTOR_ID, CONNECTOR_NAME)?],
Duration::ZERO,
Duration::ZERO,
&access_token,
)
.await?;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml_with_chatgpt_base_url(
codex_home.path(),
&responses_server.uri(),
&apps_url,
)?;
let config_path = codex_home.path().join("config.toml");
let config = std::fs::read_to_string(&config_path)?.replacen(
"model_provider = \"mock_provider\"",
"mcp_oauth_credentials_store = \"file\"\nmodel_provider = \"mock_provider\"",
1,
);
std::fs::write(config_path, format!("{config}\n[features]\napps = true\n"))?;
let executor_fixture = test_env().await?;
let executor_file_system = executor_fixture.environment().get_filesystem();
let plugin_root = executor_fixture.selection().cwd.join("executor-calendar")?;
let manifest_dir = plugin_root.join(".codex-plugin")?;
executor_file_system
.create_directory(
&manifest_dir,
CreateDirectoryOptions { recursive: true },
/*sandbox*/ None,
)
.await?;
executor_file_system
.write_file(
&manifest_dir.join("plugin.json")?,
br#"{"name":"executor-calendar","interface":{"displayName":"Executor Calendar"}}"#
.to_vec(),
/*sandbox*/ None,
)
.await?;
let app_config_path = plugin_root.join(".app.json")?;
executor_file_system
.write_file(
&app_config_path,
br#"{"apps":{"calendar":{"id":"calendar"}}}"#.to_vec(),
/*sandbox*/ None,
)
.await?;
let mut app_server = TestAppServer::new_with_auto_env(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, app_server.initialize()).await??;
let environment_id = app_server.auto_env_params()?.environment_id;
let selected_thread = start_thread(
&mut app_server,
Some(SelectedCapabilityRoot {
id: "executor-calendar@1".to_string(),
location: CapabilityRootLocation::Environment {
environment_id: environment_id.clone(),
path: plugin_root.clone(),
},
}),
)
.await?;
executor_file_system
.write_file(
&app_config_path,
br#"{"apps":{}}"#.to_vec(),
/*sandbox*/ None,
)
.await?;
let login_id = app_server
.send_chatgpt_auth_tokens_login_request(
access_token,
"account-123".to_string(),
Some("pro".to_string()),
)
.await?;
let login_response = timeout(
DEFAULT_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(login_id)),
)
.await??;
assert_eq!(
to_response::<LoginAccountResponse>(login_response)?,
LoginAccountResponse::ChatgptAuthTokens {}
);
assert_eq!(
list_apps(
&mut app_server,
&selected_thread,
/*force_refetch*/ true
)
.await?,
AppsListResponse {
data: vec![expected_app(vec![PLUGIN_DISPLAY_NAME.to_string()])],
next_cursor: None,
}
);
assert_eq!(
list_apps(
&mut app_server,
&selected_thread,
/*force_refetch*/ false,
)
.await?,
AppsListResponse {
data: vec![expected_app(vec![PLUGIN_DISPLAY_NAME.to_string()])],
next_cursor: None,
}
);
let unselected_thread = start_thread(&mut app_server, /*selected_root*/ None).await?;
assert_eq!(
list_apps(
&mut app_server,
&unselected_thread,
/*force_refetch*/ false,
)
.await?,
AppsListResponse {
data: vec![expected_app(Vec::new())],
next_cursor: None,
}
);
executor_file_system
.write_file(
&app_config_path,
br#"{"apps":{"calendar":{"id":"calendar"}}}"#.to_vec(),
/*sandbox*/ None,
)
.await?;
let model_selected_thread = start_thread(
&mut app_server,
Some(SelectedCapabilityRoot {
id: "executor-calendar@1".to_string(),
location: CapabilityRootLocation::Environment {
environment_id,
path: plugin_root,
},
}),
)
.await?;
let selected_description =
model_tool_description(&mut app_server, &responses_server, &model_selected_thread).await?;
let unselected_description =
model_tool_description(&mut app_server, &responses_server, &unselected_thread).await?;
assert!(selected_description.contains("This tool is part of plugin `Executor Calendar`."));
assert!(!unselected_description.contains(PLUGIN_DISPLAY_NAME));
apps_server_handle.abort();
let _ = apps_server_handle.await;
Ok(())
}
async fn start_thread(
app_server: &mut TestAppServer,
selected_root: Option<SelectedCapabilityRoot>,
) -> Result<String> {
let request_id = app_server
.send_thread_start_request_with_auto_env(ThreadStartParams {
model: Some("mock-model".to_string()),
selected_capability_roots: selected_root.map(|root| vec![root]),
..Default::default()
})
.await?;
let response = timeout(
DEFAULT_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response(response)?;
Ok(thread.id)
}
async fn list_apps(
app_server: &mut TestAppServer,
thread_id: &str,
force_refetch: bool,
) -> Result<AppsListResponse> {
let request_id = app_server
.send_apps_list_request(AppsListParams {
cursor: None,
limit: None,
thread_id: Some(thread_id.to_string()),
force_refetch,
})
.await?;
let response = timeout(
DEFAULT_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
to_response(response)
}
async fn model_tool_description(
app_server: &mut TestAppServer,
responses_server: &wiremock::MockServer,
thread_id: &str,
) -> Result<String> {
let response_mock = responses::mount_sse_once(
responses_server,
responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_assistant_message("msg-1", "Done"),
responses::ev_completed("resp-1"),
]),
)
.await;
let request_id = app_server
.send_turn_start_request(TurnStartParams {
thread_id: thread_id.to_string(),
input: vec![UserInput::Text {
text: "Use Calendar".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
timeout(
DEFAULT_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
timeout(
DEFAULT_TIMEOUT,
app_server.read_stream_until_notification_message("turn/completed"),
)
.await??;
response_mock
.single_request()
.tool_by_name(TOOL_NAMESPACE, TOOL_NAME)
.context("Calendar connector tool should be model-visible")?
.get("description")
.and_then(serde_json::Value::as_str)
.map(str::to_string)
.context("Calendar connector tool should have a description")
}
fn expected_app(plugin_display_names: Vec<String>) -> AppInfo {
AppInfo {
id: CONNECTOR_ID.to_string(),
name: CONNECTOR_NAME.to_string(),
description: Some("Calendar connector".to_string()),
logo_url: None,
logo_url_dark: None,
distribution_channel: None,
branding: None,
app_metadata: None,
labels: None,
install_url: Some(codex_connectors::metadata::connector_install_url(
CONNECTOR_NAME,
CONNECTOR_ID,
)),
is_accessible: true,
is_enabled: true,
plugin_display_names,
}
}
@@ -15,6 +15,7 @@ mod connection_handling_websocket_unix;
mod current_time;
mod dynamic_tools;
mod environment_add;
mod executor_connectors;
#[cfg(not(target_os = "windows"))]
mod executor_mcp;
mod executor_skills;
+1
View File
@@ -13,6 +13,7 @@ use codex_connectors::merge::merge_plugin_connectors;
use codex_core::config::Config;
pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools;
pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager;
pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools_with_mcp_config;
pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools_with_mcp_manager;
pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools_with_options;
pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools_with_options_and_status;
+55 -2
View File
@@ -33,6 +33,7 @@ use codex_login::AuthManager;
use codex_login::CodexAuth;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY;
use codex_mcp::McpConfig;
use codex_mcp::McpConnectionManager;
use codex_mcp::McpRuntimeContext;
use codex_mcp::ToolInfo;
@@ -225,8 +226,53 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager(
codex_apps_ready: true,
});
}
let cache_key = accessible_connectors_cache_key(config, auth.as_ref());
let mcp_config = mcp_manager.runtime_config(config).await;
list_accessible_connectors_from_mcp_tools_with_resolved_config(
config,
force_refetch,
environment_manager,
auth,
mcp_config,
)
.await
}
pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_config(
config: &Config,
force_refetch: bool,
environment_manager: Arc<EnvironmentManager>,
mcp_config: McpConfig,
) -> anyhow::Result<AccessibleConnectorsStatus> {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await;
let auth = auth_manager.auth().await;
if !config
.features
.apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend))
{
return Ok(AccessibleConnectorsStatus {
connectors: Vec::new(),
codex_apps_ready: true,
});
}
list_accessible_connectors_from_mcp_tools_with_resolved_config(
config,
force_refetch,
environment_manager,
auth,
mcp_config,
)
.await
}
async fn list_accessible_connectors_from_mcp_tools_with_resolved_config(
config: &Config,
force_refetch: bool,
environment_manager: Arc<EnvironmentManager>,
auth: Option<CodexAuth>,
mcp_config: McpConfig,
) -> anyhow::Result<AccessibleConnectorsStatus> {
let cache_key = accessible_connectors_cache_key(config, auth.as_ref());
let tool_plugin_provenance = tool_plugin_provenance(&mcp_config);
if !force_refetch && let Some(cached_connectors) = read_cached_accessible_connectors(&cache_key)
{
@@ -395,7 +441,14 @@ fn write_cached_accessible_connectors(
*cache_guard = Some(CachedAccessibleConnectors {
key: cache_key,
expires_at: Instant::now() + codex_connectors::CONNECTORS_CACHE_TTL,
connectors: connectors.to_vec(),
connectors: connectors
.iter()
.cloned()
.map(|mut connector| {
connector.plugin_display_names.clear();
connector
})
.collect(),
});
}
+2 -2
View File
@@ -213,7 +213,7 @@ fn synthetic_links_are_exposed_to_the_agent_but_not_accessible_in_app_list() {
}
#[tokio::test]
async fn refresh_accessible_connectors_cache_from_mcp_tools_writes_latest_installed_apps() {
async fn refresh_accessible_connectors_cache_writes_apps_without_plugin_provenance() {
let codex_home = tempdir().expect("tempdir should succeed");
let mut config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
@@ -260,7 +260,7 @@ async fn refresh_accessible_connectors_cache_from_mcp_tools_writes_latest_instal
labels: None,
is_accessible: true,
is_enabled: true,
plugin_display_names: plugin_names(&["calendar-plugin"]),
plugin_display_names: Vec::new(),
},
AppInfo {
id: "connector_openai_hidden".to_string(),
+14
View File
@@ -3,6 +3,7 @@ use std::sync::Arc;
use crate::config::Config;
use codex_config::McpServerConfig;
use codex_connectors::ConnectorSnapshot;
use codex_core_plugins::PluginsManager;
use codex_extension_api::ExtensionDataInit;
use codex_extension_api::ExtensionRegistry;
@@ -34,6 +35,17 @@ enum OrderedMcpOverlay {
},
}
pub(crate) fn connector_snapshot_for_thread(
local_snapshot: ConnectorSnapshot,
thread_init: Option<&ExtensionDataInit>,
) -> ConnectorSnapshot {
let Some(selected_snapshot) = thread_init.and_then(ExtensionDataInit::get::<ConnectorSnapshot>)
else {
return local_snapshot;
};
local_snapshot.merged_with(selected_snapshot.as_ref())
}
#[derive(Clone)]
pub struct McpManager {
plugins_manager: Arc<PluginsManager>,
@@ -132,6 +144,8 @@ impl McpManager {
selected_plugin_registrations,
)
.await;
mcp_config.connector_snapshot =
connector_snapshot_for_thread(mcp_config.connector_snapshot, thread_init);
let mut catalog = mcp_config.mcp_server_catalog.to_builder();
if mcp_config.apps_enabled {
catalog.register(McpServerRegistration::from_compatibility(
+10 -4
View File
@@ -532,8 +532,11 @@ async fn build_skills_and_plugins(
// enabled plugins, then converted into turn-scoped guidance below.
let mentioned_plugins =
collect_explicit_plugin_mentions(&user_input, loaded_plugins.capability_summaries());
let connector_snapshot = codex_connectors::ConnectorSnapshot::from_plugin_capability_summaries(
loaded_plugins.capability_summaries(),
let connector_snapshot = crate::mcp::connector_snapshot_for_thread(
codex_connectors::ConnectorSnapshot::from_plugin_capability_summaries(
loaded_plugins.capability_summaries(),
),
Some(&sess.services.mcp_thread_init),
);
let mcp_tools = if turn_context.apps_enabled() || !mentioned_plugins.is_empty() {
// Plugin mentions need raw MCP/app inventory even when app tools
@@ -1180,8 +1183,11 @@ pub(crate) async fn built_tools(
.plugins_for_config(&turn_context.config.plugins_config_input())
.instrument(trace_span!("built_tools.load_plugins"))
.await;
let connector_snapshot = codex_connectors::ConnectorSnapshot::from_plugin_capability_summaries(
loaded_plugins.capability_summaries(),
let connector_snapshot = crate::mcp::connector_snapshot_for_thread(
codex_connectors::ConnectorSnapshot::from_plugin_capability_summaries(
loaded_plugins.capability_summaries(),
),
Some(&sess.services.mcp_thread_init),
);
let apps_enabled = turn_context.apps_enabled();
+19
View File
@@ -549,6 +549,13 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors()
);
let selected_root_init = |id: &str, environment_id: &str| {
let mut init = codex_extension_api::ExtensionDataInit::new();
init.insert(codex_connectors::ConnectorSnapshot::from_plugin_sources([
codex_connectors::PluginConnectorSource::from_connector_ids(
id,
id,
[codex_plugin::AppConnectorId(format!("{id}-connector"))],
),
]));
init.insert(vec![SelectedCapabilityRoot {
id: id.to_string(),
location: CapabilityRootLocation::Environment {
@@ -638,6 +645,18 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors()
selected_servers(&second_resolved),
std::collections::BTreeMap::from([("selected-b".to_string(), "env-b".to_string())])
);
assert_eq!(
first_resolved.connector_snapshot.connector_ids(),
&[codex_plugin::AppConnectorId(
"selected-a-connector".to_string()
)]
);
assert_eq!(
second_resolved.connector_snapshot.connector_ids(),
&[codex_plugin::AppConnectorId(
"selected-b-connector".to_string()
)]
);
}
#[tokio::test]
+1
View File
@@ -17,6 +17,7 @@ workspace = true
codex-connectors = { workspace = true }
codex-core-plugins = { workspace = true }
codex-plugin = { workspace = true }
codex-protocol = { workspace = true }
codex-utils-path-uri = { workspace = true }
serde_json = { workspace = true }
thiserror = { workspace = true }
+2
View File
@@ -1,6 +1,8 @@
//! Executor-backed connector declaration loading.
mod executor_plugin;
mod selected;
pub use executor_plugin::ExecutorPluginConnectorProvider;
pub use executor_plugin::ExecutorPluginConnectorProviderError;
pub use selected::SelectedExecutorConnectorProvider;
+64
View File
@@ -0,0 +1,64 @@
use codex_connectors::ConnectorSnapshot;
use codex_connectors::PluginConnectorSource;
use codex_core_plugins::ExecutorPluginProvider;
use codex_exec_server::EnvironmentManager;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use std::sync::Arc;
use crate::ExecutorPluginConnectorProvider;
/// Resolves connector declarations from thread-selected executor plugins.
#[derive(Clone, Debug)]
pub struct SelectedExecutorConnectorProvider {
plugin_provider: ExecutorPluginProvider,
connector_provider: ExecutorPluginConnectorProvider,
}
impl SelectedExecutorConnectorProvider {
/// Creates a provider backed by the active execution environments.
pub fn new(environment_manager: Arc<EnvironmentManager>) -> Self {
Self {
plugin_provider: ExecutorPluginProvider::new(environment_manager),
connector_provider: ExecutorPluginConnectorProvider,
}
}
/// Resolves one immutable connector snapshot in selected-root order.
pub async fn snapshot_for_roots(
&self,
selected_roots: &[SelectedCapabilityRoot],
) -> ConnectorSnapshot {
let mut sources = Vec::new();
for selected_root in selected_roots {
let plugin = match self.plugin_provider.resolve_bound(selected_root).await {
Ok(Some(plugin)) => plugin,
Ok(None) => continue,
Err(err) => {
tracing::warn!(
selected_root = selected_root.id,
error = %err,
"failed to resolve selected executor plugin for connector discovery"
);
continue;
}
};
match self.connector_provider.load(&plugin).await {
Ok(declarations) => sources.push(PluginConnectorSource::new(
plugin.plugin().selected_root_id(),
plugin.plugin().manifest().display_name(),
declarations,
)),
Err(err) => {
tracing::warn!(
selected_root = selected_root.id,
error = %err,
"failed to load selected executor plugin connectors"
);
}
}
}
ConnectorSnapshot::from_plugin_sources(sources)
}
}