From 2ae8762fb072730f5573a61ab12b9cd75c2c2378 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Wed, 24 Jun 2026 16:19:47 +0100 Subject: [PATCH] Simplify selected connector activation --- codex-rs/Cargo.lock | 2 + .../schema/json/ClientRequest.json | 2 +- .../codex_app_server_protocol.schemas.json | 2 +- .../codex_app_server_protocol.v2.schemas.json | 2 +- .../schema/json/v2/AppsListParams.json | 2 +- .../schema/typescript/v2/AppsListParams.ts | 2 +- .../src/protocol/v2/apps.rs | 2 +- .../app-server/tests/suite/v2/app_list.rs | 25 +- .../tests/suite/v2/executor_connectors.rs | 272 ++++++------------ codex-rs/core/src/mcp.rs | 17 +- codex-rs/core/src/thread_manager_tests.rs | 19 -- codex-rs/ext/connectors/Cargo.toml | 2 + 12 files changed, 111 insertions(+), 238 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index ae881f184a..76e42d2831 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2603,11 +2603,13 @@ version = "0.0.0" dependencies = [ "codex-connectors", "codex-core-plugins", + "codex-exec-server", "codex-plugin", "codex-protocol", "codex-utils-path-uri", "serde_json", "thiserror 2.0.18", + "tracing", ] [[package]] diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 7ec2109d6e..68c242504f 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 for feature gating and thread-selected plugin associations.", + "description": "Optional thread id used to evaluate app feature gating from that thread's config.", "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 2a128691f0..4d417ffb2f 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 for feature gating and thread-selected plugin associations.", + "description": "Optional thread id used to evaluate app feature gating from that thread's config.", "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 9e14296dab..0df923a0f4 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 for feature gating and thread-selected plugin associations.", + "description": "Optional thread id used to evaluate app feature gating from that thread's config.", "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 214bcf7a64..385e5ba298 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 for feature gating and thread-selected plugin associations.", + "description": "Optional thread id used to evaluate app feature gating from that thread's config.", "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 27d25d67b2..b96829567b 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 for feature gating and thread-selected plugin associations. + * Optional thread id used to evaluate app feature gating from that thread's config. */ 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 fb345b53d5..f14c11b0d8 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 for feature gating and thread-selected plugin associations. + /// Optional thread id used to evaluate app feature gating from that thread's config. #[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/tests/suite/v2/app_list.rs b/codex-rs/app-server/tests/suite/v2/app_list.rs index f87c642170..5fad04929f 100644 --- a/codex-rs/app-server/tests/suite/v2/app_list.rs +++ b/codex-rs/app-server/tests/suite/v2/app_list.rs @@ -1545,26 +1545,6 @@ pub(super) 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, @@ -1577,7 +1557,6 @@ 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)) @@ -1595,7 +1574,6 @@ async fn start_apps_server_with_delays_and_control( directory_delay, tools_delay, /*workspace_plugins_enabled*/ true, - "Bearer chatgpt-token".to_string(), ) .await } @@ -1606,14 +1584,13 @@ 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, + expected_bearer: "Bearer chatgpt-token".to_string(), expected_account_id: "account-123".to_string(), response: response.clone(), directory_delay, diff --git a/codex-rs/app-server/tests/suite/v2/executor_connectors.rs b/codex-rs/app-server/tests/suite/v2/executor_connectors.rs index d2839ad576..47ef16527d 100644 --- a/codex-rs/app-server/tests/suite/v2/executor_connectors.rs +++ b/codex-rs/app-server/tests/suite/v2/executor_connectors.rs @@ -2,10 +2,10 @@ use std::time::Duration; use anyhow::Context; use anyhow::Result; -use app_test_support::ChatGptIdTokenClaims; +use app_test_support::ChatGptAuthFixture; use app_test_support::TestAppServer; -use app_test_support::encode_id_token; use app_test_support::to_response; +use app_test_support::write_chatgpt_auth; 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; @@ -13,13 +13,13 @@ use codex_app_server_protocol::AppsListResponse; use codex_app_server_protocol::CapabilityRootLocation; use codex_app_server_protocol::ListMcpServerStatusParams; use codex_app_server_protocol::ListMcpServerStatusResponse; -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_config::types::AuthCredentialsStoreMode; use codex_exec_server::CreateDirectoryOptions; use core_test_support::responses; use core_test_support::test_codex::test_env; @@ -28,39 +28,29 @@ 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; +use super::app_list::start_apps_server_with_delays; const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60); +const ACCESS_TOKEN: &str = "chatgpt-token"; const APP_CONFIG: &[u8] = br#"{"apps":{"calendar_app":{"id":"calendar"}}}"#; -const APP_NAME: &str = "calendar_app"; const CONFLICTING_MCP_CONFIG: &[u8] = br#"{"mcpServers":{"calendar_app":{"command":"must-not-start","startup_timeout_sec":1}}}"#; 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<()> { +async fn selected_executor_connector_is_listed_hosted_and_suppresses_mcp_fallback() -> 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( + let (apps_url, apps_server_handle) = start_apps_server_with_delays( vec![AppInfo { is_accessible: false, install_url: None, - ..directory_app.clone() + ..expected_app(Vec::new()) }], vec![connector_tool(CONNECTOR_ID, CONNECTOR_NAME)?], Duration::ZERO, Duration::ZERO, - &access_token, ) .await?; @@ -77,6 +67,16 @@ async fn selected_executor_connector_is_frozen_thread_scoped_and_hosted() -> Res 1, ); std::fs::write(config_path, format!("{config}\n[features]\napps = true\n"))?; + write_chatgpt_auth( + codex_home.path(), + ChatGptAuthFixture::new(ACCESS_TOKEN) + .account_id("account-123") + .email("executor-connectors@example.com") + .plan_type("pro") + .chatgpt_account_id("account-123"), + AuthCredentialsStoreMode::File, + )?; + 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")?; @@ -91,96 +91,18 @@ async fn selected_executor_connector_is_frozen_thread_scoped_and_hosted() -> Res executor_file_system .write_file( &manifest_dir.join("plugin.json")?, - br#"{"name":"executor-calendar","interface":{"displayName":"Executor Calendar"}}"# + br#"{"name":"executor-calendar","apps":"./.app.json","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, APP_CONFIG.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(), + &plugin_root.join(".app.json")?, + APP_CONFIG.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, APP_CONFIG.to_vec(), /*sandbox*/ None) - .await?; executor_file_system .write_file( &plugin_root.join(".mcp.json")?, @@ -188,93 +110,75 @@ async fn selected_executor_connector_is_frozen_thread_scoped_and_hosted() -> Res /*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, - }, - }), + + 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 thread_start_id = app_server + .send_thread_start_request_with_auto_env(ThreadStartParams { + model: Some("mock-model".to_string()), + selected_capability_roots: Some(vec![SelectedCapabilityRoot { + id: "executor-calendar@1".to_string(), + location: CapabilityRootLocation::Environment { + environment_id, + path: plugin_root, + }, + }]), + ..Default::default() + }) + .await?; + let thread_start_response = timeout( + DEFAULT_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(thread_start_id)), ) - .await?; - let request_id = app_server + .await??; + let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?; + let thread_id = thread.id; + + let apps_list_id = app_server + .send_apps_list_request(AppsListParams { + cursor: None, + limit: None, + thread_id: Some(thread_id.clone()), + force_refetch: true, + }) + .await?; + let apps_list_response = timeout( + DEFAULT_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(apps_list_id)), + ) + .await??; + assert_eq!( + to_response::(apps_list_response)?, + AppsListResponse { + data: vec![expected_app(vec![PLUGIN_DISPLAY_NAME.to_string()])], + next_cursor: None, + } + ); + + let mcp_status_id = app_server .send_list_mcp_server_status_request(ListMcpServerStatusParams { cursor: None, limit: None, detail: None, - thread_id: Some(model_selected_thread.clone()), + thread_id: Some(thread_id.clone()), }) .await?; - let response = timeout( + let mcp_status_response = timeout( DEFAULT_TIMEOUT, - app_server.read_stream_until_response_message(RequestId::Integer(request_id)), + app_server.read_stream_until_response_message(RequestId::Integer(mcp_status_id)), ) .await??; - let response: ListMcpServerStatusResponse = to_response(response)?; - assert!(response.data.iter().all(|server| server.name != APP_NAME)); + let mcp_status_response: ListMcpServerStatusResponse = to_response(mcp_status_response)?; + assert!( + mcp_status_response + .data + .iter() + .all(|server| server.name != "calendar_app") + ); - 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_server, responses::sse(vec![ responses::ev_response_created("resp-1"), responses::ev_assistant_message("msg-1", "Done"), @@ -282,9 +186,9 @@ async fn model_tool_description( ]), ) .await; - let request_id = app_server + let turn_start_id = app_server .send_turn_start_request(TurnStartParams { - thread_id: thread_id.to_string(), + thread_id, input: vec![UserInput::Text { text: "Use Calendar".to_string(), text_elements: Vec::new(), @@ -294,7 +198,7 @@ async fn model_tool_description( .await?; timeout( DEFAULT_TIMEOUT, - app_server.read_stream_until_response_message(RequestId::Integer(request_id)), + app_server.read_stream_until_response_message(RequestId::Integer(turn_start_id)), ) .await??; timeout( @@ -302,15 +206,19 @@ async fn model_tool_description( app_server.read_stream_until_notification_message("turn/completed"), ) .await??; - - response_mock + let description = response_mock .single_request() - .tool_by_name(TOOL_NAMESPACE, TOOL_NAME) + .tool_by_name("mcp__codex_apps__calendar", "connector_calendar") .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") + .map(str::to_owned) + .context("Calendar connector tool should have a description")?; + assert!(description.contains("This tool is part of plugin `Executor Calendar`.")); + + apps_server_handle.abort(); + let _ = apps_server_handle.await; + Ok(()) } fn expected_app(plugin_display_names: Vec) -> AppInfo { diff --git a/codex-rs/core/src/mcp.rs b/codex-rs/core/src/mcp.rs index 910c3a4307..f5abfeae94 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -99,6 +99,10 @@ impl McpManager { }; let selected_connector_snapshot = thread_init.and_then(ExtensionDataInit::get::); + let hosted_apps_available = config.orchestrator_mcp_enabled + && config + .features + .apps_enabled_for_auth(apps_route_available(self.plugins_manager.auth_mode())); let mut selected_plugin_registrations = Vec::new(); let mut overlays = Vec::new(); // A contributor can emit multiple ordered actions, so order each action globally rather @@ -122,13 +126,12 @@ impl McpManager { selection_order, config, } => { - let routed_through_app = - apps_route_available(self.plugins_manager.auth_mode()) - && selected_connector_snapshot - .as_ref() - .is_some_and(|snapshot| { - snapshot.plugin_declares_app(&plugin_id, &name) - }); + let routed_through_app = hosted_apps_available + && selected_connector_snapshot + .as_ref() + .is_some_and(|snapshot| { + snapshot.plugin_declares_app(&plugin_id, &name) + }); if !routed_through_app { selected_plugin_registrations.push( McpServerRegistration::from_selected_plugin( diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index cfc735f425..75f0bdf08a 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -549,13 +549,6 @@ 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 { @@ -645,18 +638,6 @@ 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 3d4cd2e93d..a59cf65d7a 100644 --- a/codex-rs/ext/connectors/Cargo.toml +++ b/codex-rs/ext/connectors/Cargo.toml @@ -16,8 +16,10 @@ workspace = true [dependencies] codex-connectors = { workspace = true } codex-core-plugins = { workspace = true } +codex-exec-server = { workspace = true } codex-plugin = { workspace = true } codex-protocol = { workspace = true } codex-utils-path-uri = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } +tracing = { workspace = true }