openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify selected connector activation

Browse Source
This commit is contained in:
jif-oai
2026-06-25 00:25:21 +01:00
parent f4c2118044
commit 2ae8762fb0
12 changed files with 111 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -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]]
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+1 -1
View File
@@ -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"
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+1 -1
View File
@@ -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"
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+1 -1
View File
@@ -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"
codex-rs/app-server-protocol/schema/json/v2/AppsListParams.json
Generated
+1 -1
View File
@@ -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"
codex-rs/app-server-protocol/schema/typescript/v2/AppsListParams.ts
Generated
+1 -1
View File
@@ -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,
/**
codex-rs/app-server-protocol/src/protocol/v2/apps.rs
+1 -1
View File
@@ -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 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<String>,
/// When true, bypass app caches and fetch the latest data from sources.
codex-rs/app-server/tests/suite/v2/app_list.rs
+1 -24
View File
@@ -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<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>,
@@ -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,
codex-rs/app-server/tests/suite/v2/executor_connectors.rs
+90 -182
View File
@@ -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::<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, 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::<AppsListResponse>(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<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_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<String>) -> AppInfo {
codex-rs/core/src/mcp.rs
+10 -7
View File
@@ -99,6 +99,10 @@ impl McpManager {
};
let selected_connector_snapshot =
thread_init.and_then(ExtensionDataInit::get::<ConnectorSnapshot>);
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(
codex-rs/core/src/thread_manager_tests.rs
-19
View File
@@ -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]
codex-rs/ext/connectors/Cargo.toml
+2
View File
@@ -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 }