Route curated plugins by authentication mode (#35671)

## Why

Curated plugin capabilities need to follow the active authentication mode, including after an account switch and when the configured model provider differs from the authentication source.

## What changed

- Select the ChatGPT, remote, or API curated marketplace from the current authentication mode, with an API marketplace fallback for ambient Amazon Bedrock credentials.
- Apply that selection consistently to plugin loading, hooks, skills, installed-plugin conflict filtering, marketplace listing, and `codex mcp` discovery.
- Start the local curated repository sync when an account change makes the remote catalog unavailable, and refresh existing thread MCP runtimes when the effective plugin cache changes.

## Testing

Added coverage for account switches, ChatGPT-authenticated Bedrock sessions, API-key MCP discovery, curated marketplace filtering, hook and skill routing, and existing-thread MCP refreshes.

GitOrigin-RevId: dbefdba3a3ea7281e7b6013e057a418770ccfc95
This commit is contained in:
felixxia-oai
2026-07-27 21:17:42 +00:00
committed by copyberry
parent fd41e813cb
commit 294d813263
12 changed files with 1258 additions and 88 deletions
@@ -229,19 +229,29 @@ impl AccountRequestProcessor {
Arc::clone(thread_manager),
config_manager.clone(),
);
let plugins_config = config.plugins_config_input();
let refresh_thread_manager = Arc::clone(thread_manager);
let refresh_config_manager = config_manager.clone();
let on_effective_plugins_changed: Arc<
dyn Fn(codex_core_plugins::EffectivePluginsChange) + Send + Sync,
> = Arc::new(move |_change| {
Self::spawn_effective_plugins_changed_task(
Arc::clone(&refresh_thread_manager),
refresh_config_manager.clone(),
);
});
thread_manager
.plugins_manager()
.maybe_start_curated_repo_sync_for_config(
&plugins_config,
Some(Arc::clone(&on_effective_plugins_changed)),
);
thread_manager
.plugins_manager()
.maybe_start_remote_plugin_caches_refresh(
&config.plugins_config_input(),
&plugins_config,
auth,
Some(Arc::new(move |_change| {
Self::spawn_effective_plugins_changed_task(
Arc::clone(&refresh_thread_manager),
refresh_config_manager.clone(),
);
})),
Some(on_effective_plugins_changed),
);
}
Err(err) => {
@@ -188,10 +188,15 @@ fn convert_configured_marketplace_plugin_to_plugin_summary(
}
}
fn remote_installed_plugin_visible_marketplaces(config: &Config) -> Vec<&'static str> {
fn remote_installed_plugin_visible_marketplaces(
config: &Config,
use_remote_global_catalog: bool,
) -> Vec<&'static str> {
let mut marketplaces = Vec::new();
if config.features.enabled(Feature::RemotePlugin) {
if use_remote_global_catalog {
marketplaces.push(REMOTE_GLOBAL_MARKETPLACE_NAME);
}
if config.features.enabled(Feature::RemotePlugin) {
marketplaces.push(REMOTE_CREATED_BY_ME_MARKETPLACE_NAME);
}
marketplaces.push(REMOTE_WORKSPACE_MARKETPLACE_NAME);
@@ -831,11 +836,14 @@ impl PluginRequestProcessor {
{
return Ok(empty_response());
}
plugins_manager.set_auth_mode(auth.as_ref().map(CodexAuth::api_auth_mode));
let auth_mode = auth.as_ref().map(CodexAuth::api_auth_mode);
plugins_manager.set_auth_mode(auth_mode);
let plugins_input = config.plugins_config_input();
let use_remote_global_catalog = config.features.enabled(Feature::RemotePlugin)
&& auth_mode.is_some_and(DomainAuthMode::uses_codex_backend);
let remote_installed_plugin_visible_marketplaces =
remote_installed_plugin_visible_marketplaces(&config);
remote_installed_plugin_visible_marketplaces(&config, use_remote_global_catalog);
plugins_manager.maybe_start_remote_installed_plugin_bundle_sync(
&plugins_input,
auth.clone(),
@@ -861,10 +869,7 @@ impl PluginRequestProcessor {
)
.await,
);
filter_openai_curated_installed_conflicts(
&mut data,
config.features.enabled(Feature::RemotePlugin),
);
filter_openai_curated_installed_conflicts(&mut data, use_remote_global_catalog);
Ok(PluginInstalledResponse {
marketplaces: data,
@@ -0,0 +1,335 @@
use std::os::unix::fs::PermissionsExt;
use std::path::Path;
use std::path::PathBuf;
use std::process::Command;
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use anyhow::ensure;
use app_test_support::ChatGptAuthFixture;
use app_test_support::MockResponsesConfig;
use app_test_support::TestAppServer;
use app_test_support::write_chatgpt_auth;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::LoginAccountResponse;
use codex_app_server_protocol::McpServerToolCallParams;
use codex_app_server_protocol::McpServerToolCallResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_config::types::AuthCredentialsStoreMode;
use codex_features::Feature;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
use tokio::task::JoinHandle;
use tokio::time::timeout;
use wiremock::MockServer;
use super::mcp_tool::TEST_SERVER_NAME;
use super::mcp_tool::TEST_TOOL_NAME;
use super::mcp_tool::start_mcp_server;
const API_CURATED_PLUGIN_NAME: &str = "api-plugin";
const REFRESH_PROBE_SERVER_NAME: &str = "refresh-probe";
const GITHUB_PLUGINS_GIT_URL: &str = "https://github.com/openai/plugins.git";
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
struct CuratedMcpSyncFixture {
mcp: TestAppServer,
sync_barrier: PathBuf,
mcp_server_handle: JoinHandle<()>,
_fixture_root: TempDir,
_responses_server: MockServer,
}
impl CuratedMcpSyncFixture {
async fn set_up() -> Result<Self> {
let responses_server = responses::start_mock_server().await;
let (mcp_server_url, mcp_server_handle) = start_mcp_server().await?;
let fixture_root = TempDir::new()?;
let codex_home = fixture_root.path().join("codex-home");
let curated_repo = fixture_root.path().join("curated-repo");
let git_wrapper_dir = fixture_root.path().join("bin");
let git_config = fixture_root.path().join("gitconfig");
let sync_barrier = fixture_root.path().join("allow-curated-sync");
std::fs::create_dir_all(&codex_home)?;
std::fs::create_dir_all(&curated_repo)?;
std::fs::create_dir_all(&git_wrapper_dir)?;
write_curated_marketplace(&curated_repo, "marketplace.json", "openai-curated", &[])?;
write_curated_marketplace(
&curated_repo,
"api_marketplace.json",
"openai-api-curated",
&[API_CURATED_PLUGIN_NAME],
)?;
write_plugin(
&curated_repo.join("plugins").join(API_CURATED_PLUGIN_NAME),
API_CURATED_PLUGIN_NAME,
TEST_SERVER_NAME,
&mcp_server_url,
)?;
let real_git =
find_executable_on_path("git").context("find git for curated sync fixture")?;
run_git(&real_git, &curated_repo, &["init", "-b", "main"])?;
run_git(
&real_git,
&curated_repo,
&["config", "user.email", "codex-tests@openai.com"],
)?;
run_git(
&real_git,
&curated_repo,
&["config", "user.name", "Codex Tests"],
)?;
run_git(&real_git, &curated_repo, &["add", "."])?;
run_git(
&real_git,
&curated_repo,
&["commit", "-m", "test curated plugins"],
)?;
let curated_repo_url = format!("file://{}", curated_repo.display());
let rewrite_key = format!("url.{curated_repo_url}.insteadOf");
run_git(
&real_git,
&curated_repo,
&[
"config",
"--file",
git_config
.to_str()
.context("git config path should be UTF-8")?,
&rewrite_key,
GITHUB_PLUGINS_GIT_URL,
],
)?;
let git_wrapper = git_wrapper_dir.join("git");
std::fs::write(
&git_wrapper,
r#"#!/bin/sh
if [ "$1" = "ls-remote" ]; then
while [ ! -f "$CURATED_SYNC_BARRIER" ]; do
sleep 0.01
done
fi
exec "$REAL_GIT" "$@"
"#,
)?;
let mut wrapper_permissions = std::fs::metadata(&git_wrapper)?.permissions();
wrapper_permissions.set_mode(0o755);
std::fs::set_permissions(&git_wrapper, wrapper_permissions)?;
MockResponsesConfig::new(&responses_server.uri())
.enable_feature(Feature::Plugins)
.with_root_config(&format!(
r#"chatgpt_base_url = "{}/backend-api/""#,
responses_server.uri()
))
.with_extra_config(&format!(
r#"[plugins."{API_CURATED_PLUGIN_NAME}@openai-api-curated"]
enabled = true
[mcp_servers.{REFRESH_PROBE_SERVER_NAME}]
url = "{mcp_server_url}/mcp""#
))
.write(&codex_home)?;
write_chatgpt_auth(
&codex_home,
ChatGptAuthFixture::new("chatgpt-token")
.account_id("account-123")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
let inherited_path = std::env::var_os("PATH").context("PATH should be set")?;
let child_path = std::env::join_paths(
std::iter::once(git_wrapper_dir).chain(std::env::split_paths(&inherited_path)),
)?;
let child_path = child_path.to_string_lossy();
let real_git = real_git.to_string_lossy();
let git_config = git_config.to_string_lossy();
let sync_barrier_env = sync_barrier.to_string_lossy();
let mcp = TestAppServer::builder()
.with_codex_home(&codex_home)
.with_env_overrides(&[
("PATH", Some(child_path.as_ref())),
("REAL_GIT", Some(real_git.as_ref())),
("GIT_CONFIG_GLOBAL", Some(git_config.as_ref())),
("CURATED_SYNC_BARRIER", Some(sync_barrier_env.as_ref())),
])
.build_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
Ok(Self {
mcp,
sync_barrier,
mcp_server_handle,
_fixture_root: fixture_root,
_responses_server: responses_server,
})
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn existing_thread_loads_api_curated_mcp_after_auth_switch_sync() -> Result<()> {
let mut fixture = CuratedMcpSyncFixture::set_up().await?;
let thread_id = fixture
.mcp
.start_thread(ThreadStartParams::default())
.await?
.thread
.id;
// Establish the existing thread's MCP runtime while curated Git sync is still blocked.
wait_for_mcp_starting(&mut fixture.mcp, REFRESH_PROBE_SERVER_NAME).await?;
let request_id = fixture
.mcp
.send_login_account_api_key_request("sk-test-key")
.await?;
let response: LoginAccountResponse =
timeout(DEFAULT_TIMEOUT, fixture.mcp.read_response(request_id)).await??;
assert_eq!(response, LoginAccountResponse::ApiKey {});
timeout(
DEFAULT_TIMEOUT,
fixture
.mcp
.read_stream_until_notification_message("account/updated"),
)
.await??;
// The account-change refresh has completed without the API-curated bundle on disk.
wait_for_mcp_starting(&mut fixture.mcp, REFRESH_PROBE_SERVER_NAME).await?;
// Let sync materialize the bundle; its completion callback must refresh this same thread.
std::fs::write(&fixture.sync_barrier, "continue")?;
wait_for_mcp_starting(&mut fixture.mcp, TEST_SERVER_NAME).await?;
let response: McpServerToolCallResponse = fixture
.mcp
.request(|request_id| ClientRequest::McpServerToolCall {
request_id,
params: McpServerToolCallParams {
thread_id: thread_id.clone(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
arguments: Some(json!({"message": "available after sync"})),
meta: None,
},
})
.await?;
assert_eq!(
response.structured_content,
Some(json!({
"echoed": "available after sync",
"threadId": thread_id,
}))
);
fixture.mcp_server_handle.abort();
let _ = fixture.mcp_server_handle.await;
Ok(())
}
async fn wait_for_mcp_starting(mcp: &mut TestAppServer, server_name: &str) -> Result<()> {
timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_matching_notification(
"mcpServer/startupStatus/updated starting",
|notification| {
notification.method == "mcpServer/startupStatus/updated"
&& notification
.params
.as_ref()
.and_then(|params| params.get("name"))
.and_then(serde_json::Value::as_str)
== Some(server_name)
&& notification
.params
.as_ref()
.and_then(|params| params.get("status"))
.and_then(serde_json::Value::as_str)
== Some("starting")
},
),
)
.await??;
Ok(())
}
fn write_curated_marketplace(
repo_root: &Path,
manifest_name: &str,
marketplace_name: &str,
plugin_names: &[&str],
) -> Result<()> {
let manifest_root = repo_root.join(".agents/plugins");
std::fs::create_dir_all(&manifest_root)?;
let plugins = plugin_names
.iter()
.map(|plugin_name| {
json!({
"name": plugin_name,
"source": {
"source": "local",
"path": format!("./plugins/{plugin_name}"),
},
})
})
.collect::<Vec<_>>();
std::fs::write(
manifest_root.join(manifest_name),
serde_json::to_vec_pretty(&json!({
"name": marketplace_name,
"plugins": plugins,
}))?,
)?;
Ok(())
}
fn write_plugin(
root: &Path,
plugin_name: &str,
server_name: &str,
mcp_server_url: &str,
) -> Result<()> {
std::fs::create_dir_all(root.join(".codex-plugin"))?;
std::fs::write(
root.join(".codex-plugin/plugin.json"),
serde_json::to_vec_pretty(&json!({"name": plugin_name}))?,
)?;
std::fs::write(
root.join(".mcp.json"),
serde_json::to_vec_pretty(&json!({
"mcpServers": {
server_name: {
"type": "http",
"url": format!("{mcp_server_url}/mcp"),
},
},
}))?,
)?;
Ok(())
}
fn find_executable_on_path(name: &str) -> Option<PathBuf> {
let path = std::env::var_os("PATH")?;
std::env::split_paths(&path)
.map(|directory| directory.join(name))
.find(|candidate| candidate.is_file())
}
fn run_git(git: &Path, cwd: &Path, args: &[&str]) -> Result<()> {
let output = Command::new(git).current_dir(cwd).args(args).output()?;
ensure!(
output.status.success(),
"git {} failed: {}",
args.join(" "),
String::from_utf8_lossy(&output.stderr)
);
Ok(())
}
@@ -72,8 +72,8 @@ use super::exec_server_test_support::read_exec_server_json;
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
const AUTO_COMPACT_LIMIT: i64 = 1024;
const LARGE_OUTPUT_AUTO_COMPACT_LIMIT: i64 = 1_000_000;
const TEST_SERVER_NAME: &str = "tool_server";
const TEST_TOOL_NAME: &str = "echo_tool";
pub(super) const TEST_SERVER_NAME: &str = "tool_server";
pub(super) const TEST_TOOL_NAME: &str = "echo_tool";
const LARGE_RESPONSE_MESSAGE: &str = "large";
const ELICITATION_TRIGGER_MESSAGE: &str = "confirm";
const ELICITATION_MESSAGE: &str = "Allow this request?";
@@ -708,7 +708,7 @@ impl ServerHandler for ToolAppsMcpServer {
}
}
async fn start_mcp_server() -> Result<(String, JoinHandle<()>)> {
pub(super) async fn start_mcp_server() -> Result<(String, JoinHandle<()>)> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let addr = listener.local_addr()?;
let mcp_service = StreamableHttpService::new(
@@ -15,6 +15,8 @@ mod config_rpc;
mod connection_handling_websocket;
#[cfg(unix)]
mod connection_handling_websocket_unix;
#[cfg(unix)]
mod curated_mcp_sync;
mod current_time;
mod dynamic_tools;
mod environment_add;
@@ -13,6 +13,7 @@ use codex_app_server_protocol::HookTrustStatus;
use codex_app_server_protocol::HooksListParams;
use codex_app_server_protocol::HooksListResponse;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::LoginAccountResponse;
use codex_app_server_protocol::PluginAuthPolicy;
use codex_app_server_protocol::PluginInstallPolicy;
use codex_app_server_protocol::PluginInstallPolicySource;
@@ -325,6 +326,106 @@ enabled = true
Ok(())
}
#[tokio::test]
async fn plugin_installed_prefers_api_curated_conflicts_after_switching_to_api_auth() -> Result<()>
{
let codex_home = TempDir::new()?;
let server = MockServer::start().await;
write_openai_api_curated_marketplace(codex_home.path(), &["linear"])?;
write_installed_plugin(&codex_home, "openai-api-curated", "linear")?;
let config = format!(
r#"chatgpt_base_url = "{}/backend-api/"
[features]
plugins = true
plugin_sharing = false
[plugins."linear@openai-api-curated"]
enabled = true
"#,
server.uri()
);
std::fs::write(codex_home.path().join("config.toml"), &config)?;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("chatgpt-token")
.account_id("account-123")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
mount_remote_installed_plugins(
&server,
"GLOBAL",
&remote_installed_plugin_body("", "1.2.3", /*enabled*/ true),
)
.await;
mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body())
.await;
mount_empty_user_installed_plugins(&server).await;
let mut app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
let request_id = app_server
.send_plugin_installed_request(PluginInstalledParams {
cwds: None,
install_suggestion_plugin_names: None,
})
.await?;
let response: PluginInstalledResponse =
timeout(DEFAULT_TIMEOUT, app_server.read_response(request_id)).await??;
assert_eq!(
response
.marketplaces
.iter()
.flat_map(|marketplace| &marketplace.plugins)
.map(|plugin| plugin.id.as_str())
.collect::<Vec<_>>(),
vec!["linear@openai-curated-remote"]
);
// Keep the ChatGPT remote snapshot cached while changing auth to exercise endpoint-level
// filtering even when the account-change cache refresh cannot run.
std::fs::write(codex_home.path().join("config.toml"), "invalid config")?;
let request_id = app_server
.send_login_account_api_key_request("sk-test-key")
.await?;
let response: LoginAccountResponse =
timeout(DEFAULT_TIMEOUT, app_server.read_response(request_id)).await??;
assert_eq!(response, LoginAccountResponse::ApiKey {});
timeout(
DEFAULT_TIMEOUT,
app_server.read_stream_until_notification_message("account/updated"),
)
.await??;
std::fs::write(codex_home.path().join("config.toml"), config)?;
let request_id = app_server
.send_plugin_installed_request(PluginInstalledParams {
cwds: None,
install_suggestion_plugin_names: None,
})
.await?;
let response: PluginInstalledResponse =
timeout(DEFAULT_TIMEOUT, app_server.read_response(request_id)).await??;
assert_eq!(
response
.marketplaces
.iter()
.flat_map(|marketplace| &marketplace.plugins)
.map(|plugin| plugin.id.as_str())
.collect::<Vec<_>>(),
vec!["linear@openai-api-curated"]
);
assert_eq!(response.marketplace_load_errors, Vec::new());
Ok(())
}
#[tokio::test]
async fn plugin_installed_ignores_local_cache_without_catalog() -> Result<()> {
let codex_home = TempDir::new()?;
@@ -2942,6 +3043,70 @@ plugins = true
Ok(())
}
#[tokio::test]
async fn plugin_list_includes_chatgpt_curated_marketplace_for_bedrock_with_chatgpt_auth()
-> Result<()> {
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join("config.toml"),
r#"model_provider = "amazon-bedrock"
[model_providers.amazon-bedrock.aws]
region = "us-east-2"
profile = "default"
[features]
plugins = true
remote_plugin = false
"#,
)?;
write_openai_curated_marketplace(codex_home.path(), &["chatgpt-plugin"])?;
write_openai_api_curated_marketplace(codex_home.path(), &["api-plugin"])?;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("chatgpt-token")
.account_id("account-123")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
let request_id = mcp
.send_plugin_list_request(PluginListParams {
cwds: None,
marketplace_kinds: None,
force_refetch: false,
})
.await?;
let response: PluginListResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(request_id)).await??;
let chatgpt_curated_marketplace = response
.marketplaces
.iter()
.find(|marketplace| marketplace.name == "openai-curated")
.expect("expected ChatGPT curated marketplace");
assert_eq!(chatgpt_curated_marketplace.plugins.len(), 1);
assert_eq!(
chatgpt_curated_marketplace.plugins[0].id,
"chatgpt-plugin@openai-curated"
);
assert!(
response
.marketplaces
.iter()
.all(|marketplace| marketplace.name != "openai-api-curated")
);
assert!(response.marketplace_load_errors.is_empty());
Ok(())
}
#[tokio::test]
async fn plugin_list_does_not_query_openai_curated_remote_collection_by_default() -> Result<()> {
let codex_home = TempDir::new()?;