clone/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-05-21 19:45:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Gate remote installed plugin scopes by features

Browse Source
This commit is contained in:
xli-oai
2026-05-15 20:52:30 -07:00
parent 8b239f81e5
commit 09d90aae9e
8 changed files with 361 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
codex-rs/app-server/src/request_processors/plugins.rs
View File

@@ -700,7 +700,7 @@ impl PluginRequestProcessor {
.await?;
data.extend(
self.load_remote_installed_plugins(plugins_manager, &config, auth.as_ref())
self.load_remote_installed_plugins(plugins_manager, &plugins_input, auth.as_ref())
.await,
);
@@ -781,20 +781,17 @@ impl PluginRequestProcessor {
async fn load_remote_installed_plugins(
&self,
plugins_manager: Arc<codex_core_plugins::PluginsManager>,
config: &Config,
plugins_input: &codex_core_plugins::PluginsConfigInput,
auth: Option<&CodexAuth>,
) -> Vec<PluginMarketplaceEntry> {
let remote_marketplaces = if let Some(remote_marketplaces) =
plugins_manager.build_remote_installed_plugin_marketplaces_from_cache()
plugins_manager.build_remote_installed_plugin_marketplaces_from_cache(plugins_input)
{
Ok(remote_marketplaces)
} else {
let remote_plugin_service_config = RemotePluginServiceConfig {
chatgpt_base_url: config.chatgpt_base_url.clone(),
};
plugins_manager
.build_and_cache_remote_installed_plugin_marketplaces(
&remote_plugin_service_config,
plugins_input,
auth,
Some(self.effective_plugins_changed_callback()),
)

135
codex-rs/app-server/tests/suite/v2/plugin_list.rs
View File

@@ -1887,6 +1887,67 @@ async fn plugin_installed_includes_remote_shared_with_me_plugins() -> Result<()>
)
]
);
wait_for_remote_installed_scope_request(&server, "WORKSPACE").await?;
assert_no_remote_installed_scope_request(&server, "GLOBAL").await?;
Ok(())
}
#[tokio::test]
async fn plugin_installed_skips_remote_when_remote_flags_are_disabled() -> Result<()> {
let codex_home = TempDir::new()?;
let server = MockServer::start().await;
std::fs::write(
codex_home.path().join("config.toml"),
format!(
r#"chatgpt_base_url = "{}/backend-api/"
[features]
plugins = true
plugin_sharing = false
"#,
server.uri()
),
)?;
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 = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
let request_id = mcp
.send_plugin_installed_request(PluginInstalledParams {
cwds: None,
install_suggestion_plugin_names: None,
})
.await?;
let response: JSONRPCResponse = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let response: PluginInstalledResponse = to_response(response)?;
assert_eq!(
response,
PluginInstalledResponse {
marketplaces: Vec::new(),
marketplace_load_errors: Vec::new(),
}
);
tokio::time::sleep(Duration::from_millis(100)).await;
wait_for_remote_plugin_request_count(
&server,
"/ps/plugins/installed",
/*expected_count*/ 0,
)
.await?;
Ok(())
}
@@ -1894,9 +1955,18 @@ async fn plugin_installed_includes_remote_shared_with_me_plugins() -> Result<()>
async fn plugin_installed_starts_remote_installed_bundle_sync() -> Result<()> {
let codex_home = TempDir::new()?;
let server = MockServer::start().await;
write_remote_plugin_catalog_config(
codex_home.path(),
&format!("{}/backend-api/", server.uri()),
std::fs::write(
codex_home.path().join("config.toml"),
format!(
r#"chatgpt_base_url = "{}/backend-api/"
[features]
plugins = true
remote_plugin = true
plugin_sharing = false
"#,
server.uri()
),
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -1916,8 +1986,6 @@ async fn plugin_installed_starts_remote_installed_bundle_sync() -> Result<()> {
let global_installed_body =
remote_installed_plugin_body(&bundle_url, "1.2.3", /*enabled*/ true);
mount_remote_installed_plugins(&server, "GLOBAL", &global_installed_body).await;
mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body())
.await;
let mut mcp = McpProcess::new_with_env(
codex_home.path(),
@@ -1954,12 +2022,8 @@ async fn plugin_installed_starts_remote_installed_bundle_sync() -> Result<()> {
.path()
.join("plugins/cache/chatgpt-global/linear/1.2.3/.codex-plugin/plugin.json");
wait_for_path_exists(&installed_path).await?;
wait_for_remote_plugin_request_count(
&server,
"/ps/plugins/installed",
/*expected_count*/ 6,
)
.await?;
wait_for_remote_installed_scope_request(&server, "GLOBAL").await?;
assert_no_remote_installed_scope_request(&server, "WORKSPACE").await?;
Ok(())
}
@@ -2240,12 +2304,8 @@ async fn plugin_list_fetches_shared_with_me_kind() -> Result<()> {
share_context.discoverability,
Some(PluginShareDiscoverability::Unlisted)
);
wait_for_remote_plugin_request_count(
&server,
"/ps/plugins/installed",
/*expected_count*/ 5,
)
.await?;
wait_for_remote_installed_scope_request(&server, "WORKSPACE").await?;
assert_no_remote_installed_scope_request(&server, "GLOBAL").await?;
wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 0).await?;
Ok(())
}
@@ -2743,6 +2803,47 @@ async fn wait_for_remote_plugin_request_count(
Ok(())
}
async fn wait_for_remote_installed_scope_request(server: &MockServer, scope: &str) -> Result<()> {
timeout(DEFAULT_TIMEOUT, async {
loop {
let Some(requests) = server.received_requests().await else {
bail!("wiremock did not record requests");
};
if requests.iter().any(|request| {
request.method == "GET"
&& request.url.path().ends_with("/ps/plugins/installed")
&& request
.url
.query_pairs()
.any(|(name, value)| name == "scope" && value == scope)
}) {
return Ok::<(), anyhow::Error>(());
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await??;
Ok(())
}
async fn assert_no_remote_installed_scope_request(server: &MockServer, scope: &str) -> Result<()> {
let Some(requests) = server.received_requests().await else {
bail!("wiremock did not record requests");
};
assert!(
!requests.iter().any(|request| {
request.method == "GET"
&& request.url.path().ends_with("/ps/plugins/installed")
&& request
.url
.query_pairs()
.any(|(name, value)| name == "scope" && value == scope)
}),
"expected no /ps/plugins/installed requests for scope {scope}"
);
Ok(())
}
async fn wait_for_path_exists(path: &std::path::Path) -> Result<()> {
timeout(DEFAULT_TIMEOUT, async {
loop {

79
codex-rs/core-plugins/src/manager.rs
View File

@@ -36,6 +36,7 @@ use crate::marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome;
use crate::marketplace_upgrade::configured_git_marketplace_names;
use crate::marketplace_upgrade::upgrade_configured_git_marketplaces;
use crate::remote::RemoteInstalledPlugin;
use crate::remote::RemoteInstalledPluginScopes;
use crate::remote::RemotePluginCatalogError;
use crate::remote::RemotePluginServiceConfig;
use crate::remote_legacy::RemotePluginFetchError;
@@ -88,6 +89,7 @@ pub struct PluginsConfigInput {
pub config_layer_stack: ConfigLayerStack,
pub plugins_enabled: bool,
pub remote_plugin_enabled: bool,
pub plugin_sharing_enabled: bool,
pub plugin_hooks_enabled: bool,
pub chatgpt_base_url: String,
}
@@ -97,6 +99,7 @@ impl PluginsConfigInput {
config_layer_stack: ConfigLayerStack,
plugins_enabled: bool,
remote_plugin_enabled: bool,
plugin_sharing_enabled: bool,
plugin_hooks_enabled: bool,
chatgpt_base_url: String,
) -> Self {
@@ -104,10 +107,18 @@ impl PluginsConfigInput {
config_layer_stack,
plugins_enabled,
remote_plugin_enabled,
plugin_sharing_enabled,
plugin_hooks_enabled,
chatgpt_base_url,
}
}
pub(crate) fn remote_installed_plugin_scopes(&self) -> RemoteInstalledPluginScopes {
RemoteInstalledPluginScopes::from_features(
self.remote_plugin_enabled,
self.plugin_sharing_enabled,
)
}
}
#[derive(Clone, PartialEq, Eq)]
@@ -128,6 +139,7 @@ struct CachedFeaturedPluginIds {
struct RemoteInstalledPluginsCacheRefreshRequest {
service_config: RemotePluginServiceConfig,
auth: Option<CodexAuth>,
scopes: RemoteInstalledPluginScopes,
notify: RemoteInstalledPluginsCacheRefreshNotify,
// App-server attaches side effects such as skills metadata invalidation and MCP refreshes when
// remote installed state changes.
@@ -413,6 +425,7 @@ pub struct PluginsManager {
struct CachedPluginLoadOutcome {
config_version: String,
plugin_hooks_enabled: bool,
remote_installed_plugin_scopes: RemoteInstalledPluginScopes,
outcome: PluginLoadOutcome,
}
@@ -484,17 +497,21 @@ impl PluginsManager {
}
let plugin_hooks_enabled = config.plugin_hooks_enabled;
let remote_installed_plugin_scopes = config.remote_installed_plugin_scopes();
let config_version = version_for_toml(&config.config_layer_stack.effective_config());
if !force_reload
&& let Some(outcome) =
self.cached_enabled_outcome(&config_version, plugin_hooks_enabled)
&& let Some(outcome) = self.cached_enabled_outcome(
&config_version,
plugin_hooks_enabled,
remote_installed_plugin_scopes,
)
{
return outcome;
}
let outcome = load_plugins_from_layer_stack(
&config.config_layer_stack,
self.remote_installed_plugin_configs(),
self.remote_installed_plugin_configs(remote_installed_plugin_scopes),
&self.store,
self.restriction_product,
plugin_hooks_enabled,
@@ -508,6 +525,7 @@ impl PluginsManager {
*cache = Some(CachedPluginLoadOutcome {
config_version,
plugin_hooks_enabled,
remote_installed_plugin_scopes,
outcome: outcome.clone(),
});
outcome
@@ -542,7 +560,7 @@ impl PluginsManager {
}
load_plugins_from_layer_stack(
config_layer_stack,
self.remote_installed_plugin_configs(),
self.remote_installed_plugin_configs(config.remote_installed_plugin_scopes()),
&self.store,
self.restriction_product,
plugin_hooks_feature_enabled,
@@ -565,6 +583,7 @@ impl PluginsManager {
&self,
config_version: &str,
plugin_hooks_enabled: bool,
remote_installed_plugin_scopes: RemoteInstalledPluginScopes,
) -> Option<PluginLoadOutcome> {
match self.cached_enabled_outcome.read() {
Ok(cache) => cache
@@ -572,6 +591,7 @@ impl PluginsManager {
.filter(|cached| {
cached.config_version == config_version
&& cached.plugin_hooks_enabled == plugin_hooks_enabled
&& cached.remote_installed_plugin_scopes == remote_installed_plugin_scopes
})
.map(|cached| cached.outcome.clone()),
Err(err) => err
@@ -580,12 +600,16 @@ impl PluginsManager {
.filter(|cached| {
cached.config_version == config_version
&& cached.plugin_hooks_enabled == plugin_hooks_enabled
&& cached.remote_installed_plugin_scopes == remote_installed_plugin_scopes
})
.map(|cached| cached.outcome.clone()),
}
}
fn remote_installed_plugin_configs(&self) -> HashMap<String, PluginConfig> {
fn remote_installed_plugin_configs(
&self,
scopes: RemoteInstalledPluginScopes,
) -> HashMap<String, PluginConfig> {
let cache = match self.remote_installed_plugins_cache.read() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
@@ -594,27 +618,52 @@ impl PluginsManager {
return HashMap::new();
};
remote_installed_plugins_to_config(plugins, &self.store)
let plugins = plugins
.iter()
.filter(|plugin| scopes.allows_marketplace_name(&plugin.marketplace_name))
.cloned()
.collect::<Vec<_>>();
remote_installed_plugins_to_config(&plugins, &self.store)
}
pub fn build_remote_installed_plugin_marketplaces_from_cache(
&self,
config: &PluginsConfigInput,
) -> Option<Vec<crate::remote::RemoteMarketplace>> {
let scopes = config.remote_installed_plugin_scopes();
if scopes.is_empty() {
return None;
}
let cache = match self.remote_installed_plugins_cache.read() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
let plugins = cache.as_ref()?;
Some(crate::remote::group_remote_installed_plugins_by_marketplaces(plugins))
let plugins = plugins
.iter()
.filter(|plugin| scopes.allows_marketplace_name(&plugin.marketplace_name))
.cloned()
.collect::<Vec<_>>();
Some(crate::remote::group_remote_installed_plugins_by_marketplaces(&plugins))
}
pub async fn build_and_cache_remote_installed_plugin_marketplaces(
&self,
config: &RemotePluginServiceConfig,
config: &PluginsConfigInput,
auth: Option<&CodexAuth>,
on_effective_plugins_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
) -> Result<Vec<crate::remote::RemoteMarketplace>, RemotePluginCatalogError> {
let plugins = crate::remote::fetch_remote_installed_plugins(config, auth).await?;
let scopes = config.remote_installed_plugin_scopes();
if scopes.is_empty() {
return Ok(Vec::new());
}
let plugins = crate::remote::fetch_remote_installed_plugins_for_scopes(
&remote_plugin_service_config(config),
auth,
scopes,
)
.await?;
let marketplaces = crate::remote::group_remote_installed_plugins_by_marketplaces(&plugins);
let changed = self.write_remote_installed_plugins_cache(plugins);
if changed && let Some(on_effective_plugins_changed) = on_effective_plugins_changed {
@@ -689,11 +738,16 @@ impl PluginsManager {
if !config.plugins_enabled {
return;
}
let scopes = config.remote_installed_plugin_scopes();
if scopes.is_empty() {
return;
}
self.schedule_remote_installed_plugins_cache_refresh(
RemoteInstalledPluginsCacheRefreshRequest {
service_config: remote_plugin_service_config(config),
auth,
scopes,
notify,
on_effective_plugins_changed,
},
@@ -709,6 +763,10 @@ impl PluginsManager {
if !config.plugins_enabled {
return;
}
let scopes = config.remote_installed_plugin_scopes();
if scopes.is_empty() {
return;
}
let manager = Arc::clone(self);
let config_for_refresh = config.clone();
@@ -725,6 +783,7 @@ impl PluginsManager {
self.codex_home.clone(),
remote_plugin_service_config(config),
auth,
scopes,
Some(on_local_cache_changed),
);
}
@@ -1624,6 +1683,7 @@ impl PluginsManager {
Err(err) => err.into_inner(),
};
if let Some(existing_request) = state.requested.as_ref() {
request.scopes = request.scopes.union(existing_request.scopes);
if matches!(
existing_request.notify,
RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh
@@ -1781,6 +1841,7 @@ impl PluginsManager {
let installed_plugins = crate::remote::fetch_remote_installed_plugins(
&request.service_config,
request.auth.as_ref(),
request.scopes,
)
.await;
match installed_plugins {

14
codex-rs/core-plugins/src/manager_tests.rs
View File

@@ -400,9 +400,17 @@ remote_plugin = true
assert_eq!(outcome, PluginLoadOutcome::default());
}
#[test]
fn build_remote_installed_plugin_marketplaces_from_cache_uses_remote_metadata() {
#[tokio::test]
async fn build_remote_installed_plugin_marketplaces_from_cache_uses_remote_metadata() {
let codex_home = TempDir::new().unwrap();
write_file(
&codex_home.path().join(CONFIG_TOML_FILE),
r#"[features]
plugins = true
remote_plugin = true
"#,
);
let config = load_plugins_config_input(codex_home.path(), codex_home.path()).await;
let manager = PluginsManager::new(codex_home.path().to_path_buf());
manager.write_remote_installed_plugins_cache(vec![RemoteInstalledPlugin {
@@ -436,7 +444,7 @@ fn build_remote_installed_plugin_marketplaces_from_cache_uses_remote_metadata()
}]);
let marketplaces = manager
.build_remote_installed_plugin_marketplaces_from_cache()
.build_remote_installed_plugin_marketplaces_from_cache(&config)
.expect("remote installed cache should be present");
assert_eq!(marketplaces.len(), 1);
assert_eq!(marketplaces[0].name, "chatgpt-global");

108
codex-rs/core-plugins/src/remote.rs
View File

@@ -27,7 +27,7 @@ pub use remote_installed_plugin_sync::RemoteInstalledPluginBundleSyncError;
pub use remote_installed_plugin_sync::RemoteInstalledPluginBundleSyncOutcome;
pub use remote_installed_plugin_sync::RemotePluginCacheMutationGuard;
pub use remote_installed_plugin_sync::mark_remote_plugin_cache_mutation_in_flight;
pub use remote_installed_plugin_sync::maybe_start_remote_installed_plugin_bundle_sync;
pub(crate) use remote_installed_plugin_sync::maybe_start_remote_installed_plugin_bundle_sync;
pub use remote_installed_plugin_sync::sync_remote_installed_plugin_bundles_once;
pub use share::RemotePluginShareAccessPolicy;
pub use share::RemotePluginShareDiscoverability;
@@ -291,7 +291,7 @@ pub enum RemotePluginCatalogError {
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Deserialize)]
enum RemotePluginScope {
pub(crate) enum RemotePluginScope {
#[serde(rename = "GLOBAL")]
Global,
#[serde(rename = "WORKSPACE")]
@@ -332,6 +332,77 @@ impl RemotePluginScope {
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) struct RemoteInstalledPluginScopes {
global: bool,
workspace: bool,
}
impl RemoteInstalledPluginScopes {
pub(crate) fn all() -> Self {
Self {
global: true,
workspace: true,
}
}
pub(crate) fn from_features(remote_plugin_enabled: bool, plugin_sharing_enabled: bool) -> Self {
Self {
global: remote_plugin_enabled,
workspace: plugin_sharing_enabled,
}
}
pub(crate) fn is_empty(self) -> bool {
!self.global && !self.workspace
}
pub(crate) fn union(self, other: Self) -> Self {
Self {
global: self.global || other.global,
workspace: self.workspace || other.workspace,
}
}
pub(crate) fn iter(self) -> impl Iterator<Item = RemotePluginScope> {
[
(self.global, RemotePluginScope::Global),
(self.workspace, RemotePluginScope::Workspace),
]
.into_iter()
.filter_map(|(enabled, scope)| enabled.then_some(scope))
}
pub(crate) fn allows_marketplace_name(self, marketplace_name: &str) -> bool {
match RemotePluginScope::from_marketplace_name(marketplace_name) {
Some(RemotePluginScope::Global) => self.global,
Some(RemotePluginScope::Workspace) => self.workspace,
None => false,
}
}
pub(crate) fn marketplace_names(self) -> impl Iterator<Item = &'static str> {
[
(self.global, REMOTE_GLOBAL_MARKETPLACE_NAME),
(self.workspace, REMOTE_WORKSPACE_MARKETPLACE_NAME),
(
self.workspace,
REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME,
),
(
self.workspace,
REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME,
),
(
self.workspace,
REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME,
),
]
.into_iter()
.filter_map(|(enabled, marketplace_name)| enabled.then_some(marketplace_name))
}
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
struct RemotePluginPagination {
next_page_token: Option<String>,
@@ -643,26 +714,29 @@ fn build_remote_marketplace(
}))
}
pub async fn fetch_remote_installed_plugins(
pub(crate) async fn fetch_remote_installed_plugins(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
scopes: RemoteInstalledPluginScopes,
) -> Result<Vec<RemoteInstalledPlugin>, RemotePluginCatalogError> {
let auth = ensure_chatgpt_auth(auth)?;
let global = async {
let scope = RemotePluginScope::Global;
let installed_plugins = fetch_installed_plugins_for_scope(config, auth, scope).await?;
Ok::<_, RemotePluginCatalogError>((scope, installed_plugins))
};
let workspace = async {
let scope = RemotePluginScope::Workspace;
let installed_plugins = fetch_installed_plugins_for_scope(config, auth, scope).await?;
Ok::<_, RemotePluginCatalogError>((scope, installed_plugins))
};
fetch_remote_installed_plugins_for_scopes(config, auth, scopes).await
}
let (global, workspace) = tokio::try_join!(global, workspace)?;
let mut installed_plugins = [global, workspace]
pub(crate) async fn fetch_remote_installed_plugins_for_scopes(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
scopes: RemoteInstalledPluginScopes,
) -> Result<Vec<RemoteInstalledPlugin>, RemotePluginCatalogError> {
if scopes.is_empty() {
return Ok(Vec::new());
}
let auth = ensure_chatgpt_auth(auth)?;
let mut installed_plugins = Vec::new();
for scope in scopes.iter() {
installed_plugins.extend(fetch_installed_plugins_for_scope(config, auth, scope).await?);
}
let mut installed_plugins = installed_plugins
.into_iter()
.flat_map(|(_scope, plugins)| plugins)
.map(|plugin| remote_installed_plugin_to_cache_entry(&plugin))
.collect::<Result<Vec<_>, _>>()?;
installed_plugins.sort_by(|left, right| {

120
codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs
View File

@@ -1,10 +1,5 @@
use super::REMOTE_GLOBAL_MARKETPLACE_NAME;
use super::REMOTE_WORKSPACE_MARKETPLACE_NAME;
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME;
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME;
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME;
use super::RemoteInstalledPluginScopes;
use super::RemotePluginCatalogError;
use super::RemotePluginScope;
use super::RemotePluginServiceConfig;
use super::ensure_chatgpt_auth;
use super::fetch_installed_plugins_for_scope_with_download_url;
@@ -27,6 +22,17 @@ use std::sync::OnceLock;
use tracing::info;
use tracing::warn;
#[cfg(test)]
use super::REMOTE_GLOBAL_MARKETPLACE_NAME;
#[cfg(test)]
use super::REMOTE_WORKSPACE_MARKETPLACE_NAME;
#[cfg(test)]
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME;
#[cfg(test)]
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME;
#[cfg(test)]
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME;
static REMOTE_INSTALLED_PLUGIN_BUNDLE_SYNC_IN_FLIGHT: OnceLock<
Mutex<HashSet<RemoteInstalledPluginBundleSyncKey>>,
> = OnceLock::new();
@@ -65,6 +71,7 @@ pub enum RemoteInstalledPluginBundleSyncError {
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct RemoteInstalledPluginBundleSyncKey {
plugin_cache_root: PathBuf,
scopes: RemoteInstalledPluginScopes,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -78,25 +85,35 @@ pub struct RemotePluginCacheMutationGuard {
key: RemotePluginCacheMutationKey,
}
pub fn maybe_start_remote_installed_plugin_bundle_sync(
pub(crate) fn maybe_start_remote_installed_plugin_bundle_sync(
codex_home: PathBuf,
config: RemotePluginServiceConfig,
auth: Option<CodexAuth>,
scopes: RemoteInstalledPluginScopes,
on_local_cache_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
) {
if scopes.is_empty() {
return;
}
let Some(auth) = auth else {
return;
};
let key = RemoteInstalledPluginBundleSyncKey {
plugin_cache_root: remote_plugin_cache_root(&codex_home),
scopes,
};
if !mark_remote_installed_plugin_bundle_sync_in_flight(key.clone()) {
return;
}
tokio::spawn(async move {
let result =
sync_remote_installed_plugin_bundles_once(codex_home, &config, Some(&auth)).await;
let result = sync_remote_installed_plugin_bundles_once_for_scopes(
codex_home,
&config,
Some(&auth),
scopes,
)
.await;
match result {
Ok(outcome) => {
if outcome.changed_local_cache()
@@ -127,50 +144,44 @@ pub async fn sync_remote_installed_plugin_bundles_once(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
) -> Result<RemoteInstalledPluginBundleSyncOutcome, RemoteInstalledPluginBundleSyncError> {
let auth = ensure_chatgpt_auth(auth)?;
let global = async {
let scope = RemotePluginScope::Global;
let installed_plugins = fetch_installed_plugins_for_scope_with_download_url(
config, auth, scope, /*include_download_urls*/ true,
)
.await?;
Ok::<_, RemotePluginCatalogError>((scope, installed_plugins))
};
let workspace = async {
let scope = RemotePluginScope::Workspace;
let installed_plugins = fetch_installed_plugins_for_scope_with_download_url(
config, auth, scope, /*include_download_urls*/ true,
)
.await?;
Ok::<_, RemotePluginCatalogError>((scope, installed_plugins))
};
sync_remote_installed_plugin_bundles_once_for_scopes(
codex_home,
config,
auth,
RemoteInstalledPluginScopes::all(),
)
.await
}
async fn sync_remote_installed_plugin_bundles_once_for_scopes(
codex_home: PathBuf,
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
scopes: RemoteInstalledPluginScopes,
) -> Result<RemoteInstalledPluginBundleSyncOutcome, RemoteInstalledPluginBundleSyncError> {
if scopes.is_empty() {
return Ok(RemoteInstalledPluginBundleSyncOutcome::default());
}
let auth = ensure_chatgpt_auth(auth)?;
let mut installed_plugins_by_scope = Vec::new();
for scope in scopes.iter() {
let installed_plugins = fetch_installed_plugins_for_scope_with_download_url(
config, auth, scope, /*include_download_urls*/ true,
)
.await?;
installed_plugins_by_scope.push((scope, installed_plugins));
}
let (global, workspace) = tokio::try_join!(global, workspace)?;
let store = PluginStore::try_new(codex_home.clone())?;
let mut installed_plugin_names_by_marketplace =
BTreeMap::<String, BTreeSet<String>>::from_iter([
(REMOTE_GLOBAL_MARKETPLACE_NAME.to_string(), BTreeSet::new()),
(
REMOTE_WORKSPACE_MARKETPLACE_NAME.to_string(),
BTreeSet::new(),
),
(
REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME.to_string(),
BTreeSet::new(),
),
(
REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME.to_string(),
BTreeSet::new(),
),
(
REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME.to_string(),
BTreeSet::new(),
),
]);
let mut installed_plugin_names_by_marketplace = BTreeMap::<String, BTreeSet<String>>::from_iter(
scopes
.marketplace_names()
.map(|marketplace_name| (marketplace_name.to_string(), BTreeSet::new())),
);
let mut installed_plugin_ids = BTreeSet::new();
let mut failed_remote_plugin_ids = BTreeSet::new();
for (_scope, installed_plugins) in [global, workspace] {
for (_scope, installed_plugins) in installed_plugins_by_scope {
for installed_plugin in installed_plugins {
let plugin = installed_plugin.plugin;
let marketplace_name = remote_plugin_canonical_marketplace_name(&plugin)?.to_string();
@@ -305,21 +316,11 @@ fn remove_stale_remote_plugin_caches(
installed_plugin_names_by_marketplace: &BTreeMap<String, BTreeSet<String>>,
) -> Result<Vec<String>, String> {
let mut removed_cache_plugin_ids = Vec::new();
for marketplace_name in [
REMOTE_GLOBAL_MARKETPLACE_NAME,
REMOTE_WORKSPACE_MARKETPLACE_NAME,
REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME,
REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME,
REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME,
] {
for (marketplace_name, installed_plugin_names) in installed_plugin_names_by_marketplace {
let marketplace_root = codex_home.join(PLUGINS_CACHE_DIR).join(marketplace_name);
if !marketplace_root.exists() {
continue;
}
let installed_plugin_names = installed_plugin_names_by_marketplace
.get(marketplace_name)
.cloned()
.unwrap_or_default();
for entry in fs::read_dir(&marketplace_root).map_err(|err| {
format!(
"failed to read remote plugin cache directory {}: {err}",
@@ -430,6 +431,7 @@ mod tests {
let codex_home = tempfile::tempdir().expect("create codex home");
let key = RemoteInstalledPluginBundleSyncKey {
plugin_cache_root: remote_plugin_cache_root(codex_home.path()),
scopes: RemoteInstalledPluginScopes::all(),
};
assert!(mark_remote_installed_plugin_bundle_sync_in_flight(

5
codex-rs/core-plugins/src/test_support.rs
View File

@@ -120,6 +120,11 @@ pub(crate) async fn load_plugins_config(codex_home: &Path, cwd: &Path) -> Plugin
"remote_plugin",
/*default_enabled*/ false,
),
feature_enabled(
&effective_config,
"plugin_sharing",
/*default_enabled*/ true,
),
feature_enabled(
&effective_config,
"plugin_hooks",

1
codex-rs/core/src/config/mod.rs
View File

@@ -1249,6 +1249,7 @@ impl Config {
self.config_layer_stack.clone(),
self.features.enabled(Feature::Plugins),
self.features.enabled(Feature::RemotePlugin),
self.features.enabled(Feature::PluginSharing),
self.features.enabled(Feature::PluginHooks),
self.chatgpt_base_url.clone(),
)