mirror of
https://github.com/openai/codex.git
synced 2026-10-21 17:48:44 +00:00
Centralize remote plugin mutations in PluginsManager (#42114)
## What changed - Move remote plugin install and uninstall orchestration from the app server into shared `PluginsManager` APIs. - Keep cache and backend mutations coordinated by the installed-plugin sync gate, and retain install outcomes long enough to protect newly materialized bundles during downstream setup. - Return structured operation errors and outcomes so callers can preserve JSON-RPC error mapping, telemetry, cache refreshes, and OAuth setup. ## Testing - Add regression coverage that verifies uninstall holds the mutation gate, preserves the local cache when the backend operation fails, and refreshes installed state after a successful uninstall. GitOrigin-RevId: de39f19a4e61c6e9c76ddc2c65d2ac130a4b7f88
This commit is contained in:
willwang-openai
committed by
copyberry
copyberry
parent
1f4c47343a
commit
ef76e6ac30
@@ -83,7 +83,11 @@ pub use manager::PluginUninstallError;
|
||||
pub use manager::PluginsConfigInput;
|
||||
pub use manager::PluginsManager;
|
||||
pub use manager::RecommendedPluginCandidatesInput;
|
||||
pub use manager::RemoteInstalledPluginSyncGuard;
|
||||
pub use manager::RemotePluginInstallOutcome;
|
||||
pub use manager::RemotePluginInstallRequest;
|
||||
pub use manager::RemotePluginOperationError;
|
||||
pub use manager::RemotePluginOperationErrorKind;
|
||||
pub use manager::RemotePluginUninstallOutcome;
|
||||
pub use marketplace_policy::allowed_configured_marketplace_names;
|
||||
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeError as PluginMarketplaceUpgradeError;
|
||||
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome as PluginMarketplaceUpgradeOutcome;
|
||||
|
||||
@@ -1,5 +1,12 @@
|
||||
#[path = "bundled_plugin_exclusions.rs"]
|
||||
mod bundled_plugin_exclusions;
|
||||
#[path = "remote_mutations.rs"]
|
||||
mod remote_mutations;
|
||||
pub use remote_mutations::RemotePluginInstallOutcome;
|
||||
pub use remote_mutations::RemotePluginInstallRequest;
|
||||
pub use remote_mutations::RemotePluginOperationError;
|
||||
pub use remote_mutations::RemotePluginOperationErrorKind;
|
||||
pub use remote_mutations::RemotePluginUninstallOutcome;
|
||||
#[path = "marketplace_context.rs"]
|
||||
mod marketplace_context;
|
||||
pub use marketplace_context::PluginMarketplaceContext;
|
||||
@@ -255,7 +262,7 @@ enum RemoteInstalledPluginsCachePublication {
|
||||
///
|
||||
/// Callers should keep this guard only while mutating the remote plugin cache and backend
|
||||
/// installed state, then release it before unrelated analytics, OAuth, or runtime refresh work.
|
||||
pub struct RemoteInstalledPluginSyncGuard {
|
||||
struct RemoteInstalledPluginSyncGuard {
|
||||
_permit: OwnedSemaphorePermit,
|
||||
}
|
||||
|
||||
@@ -1458,7 +1465,7 @@ impl PluginsManager {
|
||||
});
|
||||
}
|
||||
|
||||
pub fn maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
|
||||
fn maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
|
||||
self: &Arc<Self>,
|
||||
config: &PluginsConfigInput,
|
||||
auth: Option<CodexAuth>,
|
||||
@@ -1563,7 +1570,7 @@ impl PluginsManager {
|
||||
|
||||
/// Acquires the cache-root gate shared by full installed-bundle sync, reconciliation, and
|
||||
/// direct remote plugin mutations.
|
||||
pub async fn acquire_remote_installed_plugin_sync_guard(
|
||||
async fn acquire_remote_installed_plugin_sync_guard(
|
||||
&self,
|
||||
) -> Result<RemoteInstalledPluginSyncGuard, RemoteInstalledPluginBundleSyncError> {
|
||||
let permit = tokio::time::timeout(
|
||||
|
||||
@@ -0,0 +1,298 @@
|
||||
//! Shared remote plugin mutations for the CLI and app-server.
|
||||
//! The sync gate covers cache/backend changes and refresh scheduling. An install outcome keeps
|
||||
//! its bundle protected from pruning until the caller finishes downstream setup and drops it.
|
||||
|
||||
use super::EffectivePluginsChangedCallback;
|
||||
use super::PluginInstallOutcome;
|
||||
use super::PluginsConfigInput;
|
||||
use super::PluginsManager;
|
||||
use crate::marketplace::MarketplacePluginAuthPolicy;
|
||||
use crate::remote;
|
||||
use crate::remote::RemoteInstalledPluginBundleSyncError;
|
||||
use crate::remote::RemotePluginCacheMutationGuard;
|
||||
use crate::remote::RemotePluginCatalogError;
|
||||
use crate::remote::RemotePluginDetail;
|
||||
use crate::remote_bundle;
|
||||
use crate::remote_bundle::RemotePluginBundleInstallError;
|
||||
use codex_app_server_protocol::PluginAuthPolicy;
|
||||
use codex_app_server_protocol::PluginAvailability;
|
||||
use codex_app_server_protocol::PluginInstallPolicy;
|
||||
use codex_login::CodexAuth;
|
||||
use codex_plugin::PluginId;
|
||||
use codex_plugin::PluginTelemetryMetadata;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "remote_mutations_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
pub struct RemotePluginInstallRequest {
|
||||
pub marketplace_name: String,
|
||||
pub remote_plugin_id: String,
|
||||
pub install_attempt_id: Option<String>,
|
||||
}
|
||||
|
||||
/// Keep this outcome alive throughout post-install setup to protect the installed bundle.
|
||||
#[must_use]
|
||||
pub struct RemotePluginInstallOutcome {
|
||||
pub installed: PluginInstallOutcome,
|
||||
pub detail: RemotePluginDetail,
|
||||
pub app_ids_needing_auth: Option<Vec<String>>,
|
||||
_cache_mutation: RemotePluginCacheMutationGuard,
|
||||
}
|
||||
|
||||
pub struct RemotePluginUninstallOutcome {
|
||||
pub telemetry: PluginTelemetryMetadata,
|
||||
pub effective_plugins_changed: bool,
|
||||
/// The backend uninstall succeeded even when removing the local cache failed.
|
||||
pub cache_removal_error: Option<RemotePluginCatalogError>,
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
#[error("{kind}")]
|
||||
pub struct RemotePluginOperationError {
|
||||
pub plugin_id: Option<PluginId>,
|
||||
#[source]
|
||||
pub kind: Box<RemotePluginOperationErrorKind>,
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum RemotePluginOperationErrorKind {
|
||||
#[error("{context}: {source}")]
|
||||
Catalog {
|
||||
context: &'static str,
|
||||
source: RemotePluginCatalogError,
|
||||
},
|
||||
#[error("install remote plugin bundle: {0}")]
|
||||
Bundle(#[source] RemotePluginBundleInstallError),
|
||||
#[error("{context}: {source}")]
|
||||
Sync {
|
||||
context: &'static str,
|
||||
source: RemoteInstalledPluginBundleSyncError,
|
||||
},
|
||||
#[error("remote plugin {0} is disabled by admin")]
|
||||
DisabledByAdmin(String),
|
||||
#[error("remote plugin {0} is not available for install")]
|
||||
NotAvailable(String),
|
||||
#[error("{0}")]
|
||||
InvalidRequest(String),
|
||||
#[error("{0}")]
|
||||
Internal(String),
|
||||
}
|
||||
|
||||
impl PluginsManager {
|
||||
/// Resolve, validate, materialize, and install a remote plugin as one coordinated mutation.
|
||||
pub async fn install_remote_plugin(
|
||||
self: &Arc<Self>,
|
||||
config: &PluginsConfigInput,
|
||||
auth: Option<&CodexAuth>,
|
||||
request: RemotePluginInstallRequest,
|
||||
on_effective_plugins_changed: Option<EffectivePluginsChangedCallback>,
|
||||
) -> Result<RemotePluginInstallOutcome, RemotePluginOperationError> {
|
||||
use RemotePluginOperationErrorKind as Error;
|
||||
let unresolved = |kind| RemotePluginOperationError {
|
||||
plugin_id: None,
|
||||
kind: Box::new(kind),
|
||||
};
|
||||
let RemotePluginInstallRequest {
|
||||
marketplace_name,
|
||||
remote_plugin_id,
|
||||
install_attempt_id,
|
||||
} = request;
|
||||
if !config.plugins_enabled {
|
||||
return Err(unresolved(Error::InvalidRequest(format!(
|
||||
"remote plugin install is not enabled for marketplace {marketplace_name}"
|
||||
))));
|
||||
}
|
||||
remote::validate_remote_plugin_id(&remote_plugin_id)
|
||||
.map_err(|err| unresolved(Error::InvalidRequest(err.message)))?;
|
||||
let service = config.remote_plugin_service_config();
|
||||
let detail = remote::fetch_remote_plugin_detail_with_download_urls(
|
||||
&service,
|
||||
auth,
|
||||
&marketplace_name,
|
||||
&remote_plugin_id,
|
||||
)
|
||||
.await
|
||||
.map_err(|source| {
|
||||
unresolved(Error::Catalog {
|
||||
context: "read remote plugin details before install",
|
||||
source,
|
||||
})
|
||||
})?;
|
||||
let plugin_id = PluginId::parse(&detail.summary.id).map_err(|err| {
|
||||
unresolved(Error::Internal(format!(
|
||||
"invalid resolved plugin id `{}`: {err}",
|
||||
detail.summary.id
|
||||
)))
|
||||
})?;
|
||||
let resolved = |kind| RemotePluginOperationError {
|
||||
plugin_id: Some(plugin_id.clone()),
|
||||
kind: Box::new(kind),
|
||||
};
|
||||
if detail.summary.availability == PluginAvailability::DisabledByAdmin {
|
||||
return Err(resolved(Error::DisabledByAdmin(remote_plugin_id)));
|
||||
}
|
||||
if detail.summary.install_policy == PluginInstallPolicy::NotAvailable {
|
||||
return Err(resolved(Error::NotAvailable(remote_plugin_id)));
|
||||
}
|
||||
let bundle = remote_bundle::validate_remote_plugin_bundle(
|
||||
&remote_plugin_id,
|
||||
&detail.marketplace_name,
|
||||
&detail.summary.name,
|
||||
detail.release_version.as_deref(),
|
||||
detail.bundle_download_url.as_deref(),
|
||||
detail.app_manifest.clone(),
|
||||
)
|
||||
.map_err(|err| resolved(Error::Bundle(err)))?;
|
||||
let _sync_guard = self
|
||||
.acquire_remote_installed_plugin_sync_guard()
|
||||
.await
|
||||
.map_err(|source| {
|
||||
resolved(Error::Sync {
|
||||
context: "failed to coordinate remote plugin install",
|
||||
source,
|
||||
})
|
||||
})?;
|
||||
let cache_mutation = remote::mark_remote_plugin_cache_mutation_in_flight(
|
||||
&self.codex_home,
|
||||
&detail.marketplace_name,
|
||||
&detail.summary.name,
|
||||
);
|
||||
// Materialize first: a failed download must never leave a backend installation behind.
|
||||
let installed = remote_bundle::download_and_install_remote_plugin_bundle(
|
||||
&service,
|
||||
self.codex_home.clone(),
|
||||
bundle,
|
||||
)
|
||||
.await
|
||||
.map_err(|err| resolved(Error::Bundle(err)))?;
|
||||
let install_result = if let Some(install_attempt_id) = install_attempt_id.as_deref() {
|
||||
remote::install_remote_plugin_with_install_attempt_id(
|
||||
&service,
|
||||
auth,
|
||||
&detail.marketplace_name,
|
||||
&remote_plugin_id,
|
||||
install_attempt_id,
|
||||
)
|
||||
.await
|
||||
} else {
|
||||
remote::install_remote_plugin(
|
||||
&service,
|
||||
auth,
|
||||
&detail.marketplace_name,
|
||||
&remote_plugin_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
.map_err(|source| {
|
||||
resolved(Error::Catalog {
|
||||
context: "install remote plugin",
|
||||
source,
|
||||
})
|
||||
})?;
|
||||
self.maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
|
||||
config,
|
||||
auth.cloned(),
|
||||
on_effective_plugins_changed,
|
||||
);
|
||||
Ok(RemotePluginInstallOutcome {
|
||||
installed: PluginInstallOutcome {
|
||||
plugin_id: installed.plugin_id,
|
||||
plugin_version: installed.plugin_version,
|
||||
installed_path: installed.installed_path,
|
||||
auth_policy: match detail.summary.auth_policy {
|
||||
PluginAuthPolicy::OnInstall => MarketplacePluginAuthPolicy::OnInstall,
|
||||
PluginAuthPolicy::OnUse => MarketplacePluginAuthPolicy::OnUse,
|
||||
},
|
||||
},
|
||||
detail,
|
||||
app_ids_needing_auth: install_result.app_ids_needing_auth,
|
||||
_cache_mutation: cache_mutation,
|
||||
})
|
||||
}
|
||||
|
||||
/// Remove the backend installation and local cache, then schedule an installed-state refresh.
|
||||
pub async fn uninstall_remote_plugin(
|
||||
self: &Arc<Self>,
|
||||
config: &PluginsConfigInput,
|
||||
auth: Option<&CodexAuth>,
|
||||
remote_plugin_id: &str,
|
||||
on_effective_plugins_changed: Option<EffectivePluginsChangedCallback>,
|
||||
) -> Result<RemotePluginUninstallOutcome, RemotePluginOperationError> {
|
||||
use RemotePluginOperationErrorKind as Error;
|
||||
let unresolved = |kind| RemotePluginOperationError {
|
||||
plugin_id: None,
|
||||
kind: Box::new(kind),
|
||||
};
|
||||
if !config.plugins_enabled {
|
||||
return Err(unresolved(Error::InvalidRequest(
|
||||
"remote plugin uninstall is not enabled".to_string(),
|
||||
)));
|
||||
}
|
||||
remote::validate_remote_plugin_id(remote_plugin_id)
|
||||
.map_err(|err| unresolved(Error::InvalidRequest(err.message)))?;
|
||||
let service = config.remote_plugin_service_config();
|
||||
let target =
|
||||
remote::resolve_remote_plugin_uninstall_target(&service, auth, remote_plugin_id)
|
||||
.await
|
||||
.map_err(|source| {
|
||||
unresolved(Error::Catalog {
|
||||
context: "resolve remote plugin before uninstall",
|
||||
source,
|
||||
})
|
||||
})?;
|
||||
let mut telemetry = self
|
||||
.telemetry_metadata_for_installed_plugin_with_remote_id(
|
||||
&target.plugin_id,
|
||||
remote_plugin_id,
|
||||
)
|
||||
.await;
|
||||
if telemetry.capability_summary.is_none() {
|
||||
telemetry.capability_summary = Some(target.fallback_capability_summary.clone());
|
||||
}
|
||||
let plugin_id = target.plugin_id.clone();
|
||||
let resolved = |kind| RemotePluginOperationError {
|
||||
plugin_id: Some(plugin_id.clone()),
|
||||
kind: Box::new(kind),
|
||||
};
|
||||
let _sync_guard = self
|
||||
.acquire_remote_installed_plugin_sync_guard()
|
||||
.await
|
||||
.map_err(|source| {
|
||||
resolved(Error::Sync {
|
||||
context: "failed to coordinate remote plugin uninstall",
|
||||
source,
|
||||
})
|
||||
})?;
|
||||
let _cache_mutation = remote::mark_remote_plugin_cache_mutation_in_flight(
|
||||
&self.codex_home,
|
||||
&plugin_id.marketplace_name,
|
||||
&plugin_id.plugin_name,
|
||||
);
|
||||
let cache_removal_error =
|
||||
match remote::uninstall_remote_plugin(&service, auth, self.codex_home.clone(), target)
|
||||
.await
|
||||
{
|
||||
Ok(()) => None,
|
||||
Err(err @ RemotePluginCatalogError::CacheRemove(_)) => Some(err),
|
||||
Err(source) => {
|
||||
return Err(resolved(Error::Catalog {
|
||||
context: "uninstall remote plugin",
|
||||
source,
|
||||
}));
|
||||
}
|
||||
};
|
||||
let effective_plugins_changed = self.clear_remote_installed_plugins_cache();
|
||||
self.maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
|
||||
config,
|
||||
auth.cloned(),
|
||||
on_effective_plugins_changed,
|
||||
);
|
||||
Ok(RemotePluginUninstallOutcome {
|
||||
telemetry,
|
||||
effective_plugins_changed,
|
||||
cache_removal_error,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
//! Regression coverage for the shared mutation gate, refresh, and backend failure handling.
|
||||
|
||||
use super::*;
|
||||
use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME;
|
||||
use crate::test_support::load_plugins_config;
|
||||
use crate::test_support::test_auth_manager;
|
||||
use crate::test_support::test_plugins_manager_with_auth_manager;
|
||||
use crate::test_support::write_file;
|
||||
use codex_protocol::auth::AuthMode;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::Ordering;
|
||||
use tempfile::TempDir;
|
||||
use tokio::sync::Notify;
|
||||
use wiremock::Mock;
|
||||
use wiremock::MockServer;
|
||||
use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
#[tokio::test]
|
||||
async fn uninstall_serializes_backend_mutations_and_preserves_cache_on_failure() {
|
||||
for uninstall_fails in [false, true] {
|
||||
let home = TempDir::new().unwrap();
|
||||
let server = MockServer::start().await;
|
||||
let mut config = load_plugins_config(home.path(), home.path()).await;
|
||||
config.chatgpt_base_url = format!("{}/backend-api", server.uri());
|
||||
let auth_manager = test_auth_manager(Some(AuthMode::Chatgpt));
|
||||
let auth = auth_manager.auth().await;
|
||||
let manager = Arc::new(test_plugins_manager_with_auth_manager(
|
||||
home.path().to_path_buf(),
|
||||
/*restriction_product*/ None,
|
||||
auth_manager,
|
||||
));
|
||||
let remote_id = "b1234567-89ab-4cde-8f01-234567890abc";
|
||||
let plugin_id = PluginId::new(
|
||||
"sample".to_string(),
|
||||
REMOTE_GLOBAL_MARKETPLACE_NAME.to_string(),
|
||||
)
|
||||
.unwrap();
|
||||
let cache = manager.store.plugin_base_root(&plugin_id);
|
||||
write_file(
|
||||
cache.join("1.0.0/.codex-plugin/plugin.json").as_path(),
|
||||
r#"{"name":"sample","version":"1.0.0"}"#,
|
||||
);
|
||||
Mock::given(method("GET"))
|
||||
.and(path(format!("/backend-api/ps/plugins/{remote_id}")))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"id": remote_id, "name": "sample", "scope": "GLOBAL",
|
||||
"installation_policy": "AVAILABLE", "authentication_policy": "ON_USE",
|
||||
"release": {"display_name": "Sample", "description": "Sample plugin", "interface": {}}
|
||||
}))).mount(&server).await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/backend-api/ps/plugins/installed"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200)
|
||||
.set_body_json(json!({"plugins": [], "pagination": {"next_page_token": null}})),
|
||||
)
|
||||
.expect(if uninstall_fails { 0 } else { 1 })
|
||||
.mount(&server)
|
||||
.await;
|
||||
let gate_held = Arc::new(AtomicBool::new(false));
|
||||
let gate_held_at_post = Arc::clone(&gate_held);
|
||||
let manager_at_post = Arc::clone(&manager);
|
||||
Mock::given(method("POST"))
|
||||
.and(path(format!(
|
||||
"/backend-api/ps/plugins/{remote_id}/uninstall"
|
||||
)))
|
||||
.respond_with(move |_: &wiremock::Request| {
|
||||
gate_held_at_post.store(
|
||||
manager_at_post
|
||||
.remote_installed_plugin_bundle_sync_gate
|
||||
.available_permits()
|
||||
== 0,
|
||||
Ordering::SeqCst,
|
||||
);
|
||||
ResponseTemplate::new(if uninstall_fails { 400 } else { 200 })
|
||||
.set_body_json(json!({"id": remote_id, "enabled": false}))
|
||||
})
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
let refreshed = Arc::new(Notify::new());
|
||||
let refreshed_callback = Arc::clone(&refreshed);
|
||||
let outcome = manager
|
||||
.uninstall_remote_plugin(
|
||||
&config,
|
||||
auth.as_ref(),
|
||||
remote_id,
|
||||
Some(Arc::new(move |_| refreshed_callback.notify_one())),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
(
|
||||
gate_held.load(Ordering::SeqCst),
|
||||
manager
|
||||
.remote_installed_plugin_bundle_sync_gate
|
||||
.available_permits(),
|
||||
cache.as_path().exists(),
|
||||
outcome.is_err()
|
||||
),
|
||||
(true, 1, uninstall_fails, uninstall_fails),
|
||||
);
|
||||
if !uninstall_fails {
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), refreshed.notified())
|
||||
.await
|
||||
.expect("successful uninstall should refresh installed state");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user