diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 2c50f268f0..ae881f184a 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index ef17cb9ce5..0824f7db7b 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -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" } diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 68c242504f..7ec2109d6e 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -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" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 4d417ffb2f..2a128691f0 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -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" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 0df923a0f4..9e14296dab 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -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" diff --git a/codex-rs/app-server-protocol/schema/json/v2/AppsListParams.json b/codex-rs/app-server-protocol/schema/json/v2/AppsListParams.json index 385e5ba298..214bcf7a64 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/AppsListParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/AppsListParams.json @@ -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" diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/AppsListParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/AppsListParams.ts index b96829567b..27d25d67b2 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/AppsListParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/AppsListParams.ts @@ -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, /** diff --git a/codex-rs/app-server-protocol/src/protocol/v2/apps.rs b/codex-rs/app-server-protocol/src/protocol/v2/apps.rs index f14c11b0d8..fb345b53d5 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/apps.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/apps.rs @@ -16,7 +16,7 @@ pub struct AppsListParams { /// Optional page size; defaults to a reasonable server-side value. #[ts(optional = nullable)] pub limit: Option, - /// 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, /// When true, bypass app caches and fetch the latest data from sources. diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index bde8048e76..694c1f154c 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -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 } diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 71557bb730..f033200ad8 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -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. diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index d73a41706c..3b4e25c02b 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -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; 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 67d768fa57..047a97929a 100644 --- a/codex-rs/app-server/src/request_processors/apps_processor.rs +++ b/codex-rs/app-server/src/request_processors/apps_processor.rs @@ -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, - mcp_manager: Arc, - plugins_manager: Arc, + 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, - mcp_manager: Arc, - plugins_manager: Arc, + 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, + connector_snapshot: &codex_connectors::ConnectorSnapshot, +) -> Vec { + 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, 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 1ddf355269..f87c642170 100644 --- a/codex-rs/app-server/tests/suite/v2/app_list.rs +++ b/codex-rs/app-server/tests/suite/v2/app_list.rs @@ -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, tools: Vec, 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, + tools: Vec, + 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, tools: Vec, @@ -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 { +pub(super) fn connector_tool(connector_id: &str, connector_name: &str) -> Result { let schema: JsonObject = serde_json::from_value(json!({ "type": "object", "additionalProperties": false diff --git a/codex-rs/app-server/tests/suite/v2/executor_connectors.rs b/codex-rs/app-server/tests/suite/v2/executor_connectors.rs new file mode 100644 index 0000000000..b284cb3166 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/executor_connectors.rs @@ -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::(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, +) -> Result { + 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 { + 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 { + 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) -> 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, + } +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index bbd9952754..dd0d4450f4 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -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; diff --git a/codex-rs/chatgpt/src/connectors.rs b/codex-rs/chatgpt/src/connectors.rs index bc84e5126d..4eb5ced97a 100644 --- a/codex-rs/chatgpt/src/connectors.rs +++ b/codex-rs/chatgpt/src/connectors.rs @@ -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; diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 79978da998..1704b7e6a5 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -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, + mcp_config: McpConfig, +) -> anyhow::Result { + 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, + auth: Option, + mcp_config: McpConfig, +) -> anyhow::Result { + 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(), }); } diff --git a/codex-rs/core/src/connectors_tests.rs b/codex-rs/core/src/connectors_tests.rs index fb28539048..280e0e20ce 100644 --- a/codex-rs/core/src/connectors_tests.rs +++ b/codex-rs/core/src/connectors_tests.rs @@ -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(), diff --git a/codex-rs/core/src/mcp.rs b/codex-rs/core/src/mcp.rs index 90e6b01e75..c372966e1c 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -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::) + else { + return local_snapshot; + }; + local_snapshot.merged_with(selected_snapshot.as_ref()) +} + #[derive(Clone)] pub struct McpManager { plugins_manager: Arc, @@ -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( diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index a8fe50b787..24ed4c3db8 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -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(); diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 75f0bdf08a..cfc735f425 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -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] diff --git a/codex-rs/ext/connectors/Cargo.toml b/codex-rs/ext/connectors/Cargo.toml index 1103ae4e2b..3d4cd2e93d 100644 --- a/codex-rs/ext/connectors/Cargo.toml +++ b/codex-rs/ext/connectors/Cargo.toml @@ -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 } diff --git a/codex-rs/ext/connectors/src/lib.rs b/codex-rs/ext/connectors/src/lib.rs index f60e5f916a..0504ef6885 100644 --- a/codex-rs/ext/connectors/src/lib.rs +++ b/codex-rs/ext/connectors/src/lib.rs @@ -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; diff --git a/codex-rs/ext/connectors/src/selected.rs b/codex-rs/ext/connectors/src/selected.rs new file mode 100644 index 0000000000..3eab067140 --- /dev/null +++ b/codex-rs/ext/connectors/src/selected.rs @@ -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) -> 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) + } +}