Report affected capabilities from remote plugin syncs (#41193)

## Why

Remote plugin sync results need to identify which runtime categories may require
invalidation when a cached bundle changes.

## What changed

- Add `RemotePluginCapabilities` hints for MCP servers, Apps, hooks, and skills.
- Attach capability hints to materialized and removed plugins. Updates union the
  declarations from the old and new bundles, while removals inspect the active
  bundle before deleting its cache.
- Discover declarations without applying runtime policy or creating runtime
  data directories.

## Testing

- Cover Agent Plugin capability discovery with an unavailable runtime data
  directory.
- Cover capability accumulation across bundle versions and removal reporting.

GitOrigin-RevId: 338075d22d86f30f52be5785a2d87c77b33609c6
This commit is contained in:
Matthew Zeng
2026-08-27 19:54:40 +00:00
committed by copyberry
parent c29bc99949
commit 56239d7a61
8 changed files with 318 additions and 55 deletions
@@ -14,6 +14,7 @@ fn only_workspace_listed_materializations_are_eligible() {
scope,
discoverability,
authenticated_account_id: Some("account-123".to_string()),
capabilities: Default::default(),
}
};
+1 -1
View File
@@ -1067,7 +1067,7 @@ pub(crate) async fn load_plugin_skill_inventory(
}
}
fn plugin_skill_roots(
pub(crate) fn plugin_skill_roots(
plugin_root: &AbsolutePathBuf,
manifest_paths: &PluginManifestPaths,
manifest_format: PluginManifestFormat,
+1 -1
View File
@@ -1471,7 +1471,7 @@ impl PluginsManager {
Ok((outcome, effective_plugins_changed)) => {
tracing::info!(
materialized_remote_plugins = ?outcome.materialized_remote_plugins,
removed_cache_plugin_ids = ?outcome.removed_cache_plugin_ids,
removed_plugins = ?outcome.removed_plugins,
failed_remote_plugin_ids = ?outcome.failed_remote_plugin_ids,
failed_materialization_remote_plugin_ids = ?outcome.failed_materialization_remote_plugin_ids,
"completed remote installed plugin bundle sync"
@@ -6962,6 +6962,7 @@ fn remote_installed_plugins_cache_refresh_coalesces_materializations() {
scope: crate::remote::RemotePluginScope::Workspace,
discoverability: Some(crate::remote::RemotePluginShareDiscoverability::Listed),
authenticated_account_id: Some("account-123".to_string()),
capabilities: Default::default(),
};
let change = |name: &str| EffectivePluginsChange {
materialized_remote_plugins: vec![materialization(name)],
+3
View File
@@ -47,6 +47,7 @@ use tracing::instrument;
use url::Url;
mod catalog_cache;
mod plugin_capabilities;
mod remote_installed_plugin_sync;
mod search;
mod share;
@@ -55,10 +56,12 @@ mod share;
#[path = "remote_tests.rs"]
mod tests;
pub use plugin_capabilities::RemotePluginCapabilities;
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::RemotePluginMaterialization;
pub use remote_installed_plugin_sync::RemotePluginRemoval;
pub use remote_installed_plugin_sync::mark_remote_plugin_cache_mutation_in_flight;
pub(crate) use remote_installed_plugin_sync::remote_installed_plugin_bundle_sync_gate;
pub use remote_installed_plugin_sync::sync_remote_installed_plugin_bundles_once;
@@ -0,0 +1,65 @@
//! Discovers best-effort runtime invalidation hints from locally cached plugin bundles.
//!
//! Discovery ignores runtime policy and does not prepare runtime directories. Repeated inclusion
//! unions old and new declarations for updates; removals capture declarations before deletion.
use crate::loader::load_plugin_apps_from_manifest;
use crate::loader::load_plugin_hooks;
use crate::loader::load_plugin_mcp_servers_from_manifest_with_format;
use crate::loader::plugin_skill_roots;
use crate::manifest::PluginManifestFormat;
use crate::manifest::load_plugin_manifest_with_format;
use crate::store::PluginStore;
use codex_hooks::plugin_hook_declarations;
use codex_plugin::PluginId;
/// Runtime categories affected by a bundle change, before runtime policy filtering.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RemotePluginCapabilities {
pub has_mcps: bool,
pub has_apps: bool,
pub has_hooks: bool,
pub has_skills: bool,
}
impl RemotePluginCapabilities {
pub(super) async fn include_active_bundle(
&mut self,
store: &PluginStore,
plugin_id: &PluginId,
) {
let Some(root) = store.active_plugin_root(plugin_id) else {
return;
};
let Some(loaded) = load_plugin_manifest_with_format(root.as_path()) else {
return;
};
let paths = &loaded.manifest.paths;
self.has_skills |= !plugin_skill_roots(&root, paths, loaded.format).is_empty();
self.has_mcps |= !load_plugin_mcp_servers_from_manifest_with_format(
root.as_path(),
paths,
/*plugin_policy*/ None,
// Declaration discovery must not create runtime directories or depend on their readiness.
/*plugin_data_root*/
None,
loaded.format,
)
.await
.is_empty();
// Agent Plugins can declare Apps through an OpenAI extension even when the local
// runtime does not activate them. Refresh hints describe declarations, not activation.
self.has_apps |= !load_plugin_apps_from_manifest(root.as_path(), paths)
.await
.is_empty();
if loaded.format == PluginManifestFormat::Legacy {
let (sources, _) =
load_plugin_hooks(&root, plugin_id, &store.plugin_data_root(plugin_id), paths);
self.has_hooks |= !plugin_hook_declarations(&sources).is_empty();
}
}
}
#[cfg(test)]
#[path = "plugin_capabilities_tests.rs"]
mod tests;
@@ -0,0 +1,157 @@
//! Covers local capability accumulation and removal hints from the shared bundle sync.
use super::*;
use crate::remote::RemoteInstalledPluginBundleSyncOutcome;
use crate::remote::RemotePluginRemoval;
use crate::remote::RemotePluginServiceConfig;
use crate::remote::sync_remote_installed_plugin_bundles_once;
use crate::test_support::write_file;
use codex_login::CodexAuth;
use codex_utils_plugins::AGENT_PLUGIN_SCHEMA_URI;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::matchers::query_param;
#[tokio::test]
async fn agent_plugin_capabilities_do_not_require_runtime_data_directory() {
let codex_home = TempDir::new().expect("Codex home");
let store = PluginStore::new(codex_home.path().to_path_buf());
let plugin_id = PluginId::parse("example@test").expect("plugin id");
let root = store.plugin_root(&plugin_id, "1.0.0");
write_file(
root.join("plugin.json").as_path(),
&json!({
"$schema": AGENT_PLUGIN_SCHEMA_URI,
"name": "example",
"extensions": {"com.openai": {"apps": "./connectors.json"}},
})
.to_string(),
);
write_file(
root.join("mcp.json").as_path(),
&json!({
"$schema": "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json",
"mcpServers": {"example": {"type": "stdio", "command": "echo"}},
})
.to_string(),
);
write_file(
root.join("connectors.json").as_path(),
&json!({"apps": {"example": {"id": "connector_example"}}}).to_string(),
);
let data_root = store.mcp_data_root(&plugin_id, PluginManifestFormat::AgentPlugin);
for block_data_root in [false, true] {
if block_data_root {
write_file(data_root.as_path(), "not a directory");
}
let mut capabilities = RemotePluginCapabilities::default();
capabilities.include_active_bundle(&store, &plugin_id).await;
assert_eq!(
capabilities,
RemotePluginCapabilities {
has_mcps: true,
has_apps: true,
// Agent Plugins declare a conventional skills root even if it is absent.
has_skills: true,
..Default::default()
}
);
assert_eq!(data_root.as_path().exists(), block_data_root);
assert!(!data_root.as_path().is_dir());
}
}
#[tokio::test]
async fn capabilities_union_cached_versions_and_sync_reports_removal() -> anyhow::Result<()> {
let codex_home = TempDir::new()?;
let store = PluginStore::new(codex_home.path().to_path_buf());
let plugin_id = PluginId::parse("example@openai-curated-remote")?;
let old_root = store.plugin_root(&plugin_id, "1.0.0");
for (path, contents) in [
(".codex-plugin/plugin.json", r#"{"name":"example"}"#),
(
".mcp.json",
r#"{"mcpServers":{"example":{"command":"unused"}}}"#,
),
(
"hooks/hooks.json",
r#"{"hooks":{"SessionStart":[{"hooks":[{"type":"command","command":"echo example"}]}]}}"#,
),
(
"skills/example/SKILL.md",
"---\nname: example\ndescription: Example skill\n---\nExample skill",
),
] {
write_file(old_root.join(path).as_path(), contents);
}
let mut capabilities = RemotePluginCapabilities::default();
capabilities.include_active_bundle(&store, &plugin_id).await;
// Accumulate the old and new declarations without involving bundle transport.
let new_root = store.plugin_root(&plugin_id, "2.0.0");
write_file(
new_root.join(".codex-plugin/plugin.json").as_path(),
r#"{"name":"example"}"#,
);
write_file(
new_root.join(".app.json").as_path(),
r#"{"apps":{"example":{"id":"connector_example"}}}"#,
);
capabilities.include_active_bundle(&store, &plugin_id).await;
assert_eq!(
capabilities,
RemotePluginCapabilities {
has_mcps: true,
has_apps: true,
has_hooks: true,
has_skills: true,
}
);
let server = MockServer::start().await;
let config = RemotePluginServiceConfig::new(
format!("{}/backend-api", server.uri()),
crate::test_support::test_http_client_factory(),
);
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
Mock::given(method("GET"))
.and(path("/backend-api/ps/plugins/installed"))
.and(query_param("includeDownloadUrls", "true"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"plugins": [],
"pagination": {"next_page_token": null},
})))
.expect(1)
.mount(&server)
.await;
// Removal describes only the active version, not the union accumulated above.
assert_eq!(
sync_remote_installed_plugin_bundles_once(
codex_home.path().to_path_buf(),
&config,
Some(&auth),
)
.await?,
RemoteInstalledPluginBundleSyncOutcome {
removed_plugins: vec![RemotePluginRemoval {
plugin_id: plugin_id.as_key(),
capabilities: RemotePluginCapabilities {
has_apps: true,
..Default::default()
},
}],
..Default::default()
}
);
assert!(!store.plugin_base_root(&plugin_id).as_path().exists());
server.verify().await;
Ok(())
}
@@ -6,6 +6,7 @@ use super::REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME;
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME;
use super::RemoteInstalledPlugin;
use super::RemoteInstalledPluginScope;
use super::RemotePluginCapabilities;
use super::RemotePluginCatalogError;
use super::RemotePluginScope;
use super::RemotePluginServiceConfig;
@@ -20,13 +21,13 @@ use codex_plugin::PluginId;
use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::collections::HashMap;
use std::fs;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use std::sync::Weak;
use tokio::fs;
use tokio::sync::Semaphore;
use tracing::warn;
@@ -44,12 +45,21 @@ pub struct RemotePluginMaterialization {
pub scope: RemotePluginScope,
pub discoverability: Option<RemotePluginShareDiscoverability>,
pub authenticated_account_id: Option<String>,
/// Runtime categories declared by either the old or new bundle.
pub capabilities: RemotePluginCapabilities,
}
/// A removed cache entry and the runtime categories declared by its previous bundle.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemotePluginRemoval {
pub plugin_id: String,
pub capabilities: RemotePluginCapabilities,
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RemoteInstalledPluginBundleSyncOutcome {
pub materialized_remote_plugins: Vec<RemotePluginMaterialization>,
pub removed_cache_plugin_ids: Vec<String>,
pub removed_plugins: Vec<RemotePluginRemoval>,
pub failed_remote_plugin_ids: Vec<String>,
/// Failures that leave an otherwise valid installed plugin unavailable locally.
pub failed_materialization_remote_plugin_ids: Vec<String>,
@@ -62,7 +72,7 @@ pub(crate) struct RemoteInstalledPluginBundleSyncResult {
impl RemoteInstalledPluginBundleSyncOutcome {
pub fn changed_local_cache(&self) -> bool {
!self.materialized_remote_plugins.is_empty() || !self.removed_cache_plugin_ids.is_empty()
!self.materialized_remote_plugins.is_empty() || !self.removed_plugins.is_empty()
}
}
@@ -237,6 +247,10 @@ pub(crate) async fn sync_remote_installed_plugin_bundles_once_with_snapshot(
}
};
// Read the old bundle before installation replaces it, including capabilities removed
// by the new version. Unchanged bundles never reach this metadata-loading path.
let mut capabilities = RemotePluginCapabilities::default();
capabilities.include_active_bundle(&store, &plugin_id).await;
match crate::remote_bundle::download_and_install_remote_plugin_bundle(
config,
codex_home.clone(),
@@ -246,6 +260,7 @@ pub(crate) async fn sync_remote_installed_plugin_bundles_once_with_snapshot(
{
Ok(result) => {
let plugin_id = result.plugin_id;
capabilities.include_active_bundle(&store, &plugin_id).await;
materialized_remote_plugins.insert(
plugin_id.as_key(),
RemotePluginMaterialization {
@@ -253,6 +268,7 @@ pub(crate) async fn sync_remote_installed_plugin_bundles_once_with_snapshot(
scope,
discoverability,
authenticated_account_id: authenticated_account_id.clone(),
capabilities,
},
);
}
@@ -276,33 +292,22 @@ pub(crate) async fn sync_remote_installed_plugin_bundles_once_with_snapshot(
.cmp(&right.marketplace_name)
.then_with(|| left.id.cmp(&right.id))
});
let stale_cache_cleanup = tokio::task::spawn_blocking(move || {
let mut removed_cache_plugin_ids = Vec::new();
let cleanup_error = remove_stale_remote_plugin_caches(
codex_home.as_path(),
&installed_plugin_names_by_marketplace,
&mut removed_cache_plugin_ids,
)
.err();
removed_cache_plugin_ids.sort();
(removed_cache_plugin_ids, cleanup_error)
})
.await;
let (removed_cache_plugin_ids, cleanup_error) = match stale_cache_cleanup {
Ok(result) => result,
Err(err) => {
warn!(error = %err, "failed to join stale remote plugin cache cleanup task");
(Vec::new(), Some(err.to_string()))
}
};
if let Some(err) = cleanup_error {
let mut removed_plugins = Vec::new();
if let Err(err) = remove_stale_remote_plugin_caches(
&store,
&installed_plugin_names_by_marketplace,
&mut removed_plugins,
)
.await
{
warn!(error = %err, "failed to remove stale remote plugin cache entries");
}
removed_plugins.sort_unstable_by(|left, right| left.plugin_id.cmp(&right.plugin_id));
Ok(RemoteInstalledPluginBundleSyncResult {
outcome: RemoteInstalledPluginBundleSyncOutcome {
materialized_remote_plugins: materialized_remote_plugins.into_values().collect(),
removed_cache_plugin_ids,
removed_plugins,
failed_remote_plugin_ids: failed_remote_plugin_ids.into_iter().collect(),
failed_materialization_remote_plugin_ids: failed_materialization_remote_plugin_ids
.into_iter()
@@ -350,11 +355,12 @@ impl Drop for RemotePluginCacheMutationGuard {
}
}
fn remove_stale_remote_plugin_caches(
codex_home: &Path,
async fn remove_stale_remote_plugin_caches(
store: &PluginStore,
installed_plugin_names_by_marketplace: &BTreeMap<String, BTreeSet<String>>,
removed_cache_plugin_ids: &mut Vec<String>,
removed_plugins: &mut Vec<RemotePluginRemoval>,
) -> Result<(), String> {
let codex_home = store.codex_home().as_path();
for marketplace_name in [
REMOTE_GLOBAL_MARKETPLACE_NAME,
REMOTE_CREATED_BY_ME_MARKETPLACE_NAME,
@@ -371,18 +377,18 @@ fn remove_stale_remote_plugin_caches(
.get(marketplace_name)
.cloned()
.unwrap_or_default();
for entry in fs::read_dir(&marketplace_root).map_err(|err| {
let mut entries = fs::read_dir(&marketplace_root).await.map_err(|err| {
format!(
"failed to read remote plugin cache directory {}: {err}",
marketplace_root.display()
)
})?;
while let Some(entry) = entries.next_entry().await.map_err(|err| {
format!(
"failed to enumerate remote plugin cache directory {}: {err}",
marketplace_root.display()
)
})? {
let entry = entry.map_err(|err| {
format!(
"failed to enumerate remote plugin cache directory {}: {err}",
marketplace_root.display()
)
})?;
let plugin_name = entry.file_name().into_string().map_err(|file_name| {
format!(
"remote plugin cache entry under {} is not valid UTF-8: {:?}",
@@ -399,25 +405,33 @@ fn remove_stale_remote_plugin_caches(
}
let cache_path = entry.path();
let plugin_id = PluginId::new(plugin_name.clone(), marketplace_name.to_string());
let mut capabilities = RemotePluginCapabilities::default();
if let Ok(plugin_id) = &plugin_id {
capabilities.include_active_bundle(store, plugin_id).await;
}
if cache_path.is_dir() {
fs::remove_dir_all(&cache_path).map_err(|err| {
fs::remove_dir_all(&cache_path).await.map_err(|err| {
format!(
"failed to remove stale remote plugin cache entry {}: {err}",
cache_path.display()
)
})?;
} else {
fs::remove_file(&cache_path).map_err(|err| {
fs::remove_file(&cache_path).await.map_err(|err| {
format!(
"failed to remove stale remote plugin cache entry {}: {err}",
cache_path.display()
)
})?;
}
let plugin_key = PluginId::new(plugin_name.clone(), marketplace_name.to_string())
let plugin_key = plugin_id
.map(|plugin_id| plugin_id.as_key())
.unwrap_or_else(|_| format!("{plugin_name}@{marketplace_name}"));
removed_cache_plugin_ids.push(plugin_key);
removed_plugins.push(RemotePluginRemoval {
plugin_id: plugin_key,
capabilities,
});
}
}
@@ -693,7 +707,13 @@ mod tests {
outcome,
RemoteInstalledPluginBundleSyncOutcome {
materialized_remote_plugins: Vec::new(),
removed_cache_plugin_ids,
removed_plugins: removed_cache_plugin_ids
.into_iter()
.map(|plugin_id| RemotePluginRemoval {
plugin_id,
capabilities: RemotePluginCapabilities::default(),
})
.collect(),
failed_remote_plugin_ids: Vec::new(),
failed_materialization_remote_plugin_ids: Vec::new(),
}
@@ -744,8 +764,8 @@ mod tests {
}
}
#[test]
fn stale_remote_plugin_cleanup_skips_cache_mutations_in_progress() {
#[tokio::test]
async fn stale_remote_plugin_cleanup_skips_cache_mutations_in_progress() {
let codex_home = tempfile::tempdir().expect("create codex home");
let cached_manifest = codex_home
.path()
@@ -788,39 +808,49 @@ mod tests {
);
let mut removed = Vec::new();
remove_stale_remote_plugin_caches(
codex_home.path(),
&PluginStore::new(codex_home.path().to_path_buf()),
&installed_plugin_names_by_marketplace,
&mut removed,
)
.await
.expect("cleanup while install is guarded");
assert_eq!(removed, Vec::<String>::new());
assert_eq!(removed, Vec::<RemotePluginRemoval>::new());
assert!(cached_manifest.is_file());
drop(guard);
let mut removed = Vec::new();
remove_stale_remote_plugin_caches(
codex_home.path(),
&PluginStore::new(codex_home.path().to_path_buf()),
&installed_plugin_names_by_marketplace,
&mut removed,
)
.await
.expect("cleanup while second install guard is still active");
assert_eq!(removed, Vec::<String>::new());
assert_eq!(removed, Vec::<RemotePluginRemoval>::new());
assert!(cached_manifest.is_file());
drop(second_guard);
let mut removed = Vec::new();
remove_stale_remote_plugin_caches(
codex_home.path(),
&PluginStore::new(codex_home.path().to_path_buf()),
&installed_plugin_names_by_marketplace,
&mut removed,
)
.await
.expect("cleanup after install guard is dropped");
assert_eq!(removed, vec!["linear@openai-curated-remote".to_string()]);
assert_eq!(
removed,
vec![RemotePluginRemoval {
plugin_id: "linear@openai-curated-remote".to_string(),
capabilities: RemotePluginCapabilities::default(),
}]
);
assert!(!cached_manifest.exists());
}
#[test]
fn stale_remote_plugin_cleanup_removes_stale_marketplace_caches_and_keeps_canonical_cache() {
#[tokio::test]
async fn stale_remote_plugin_cleanup_removes_stale_marketplace_caches_and_keeps_canonical_cache()
{
let codex_home = tempfile::tempdir().expect("create codex home");
let created_by_me_cached_manifest = codex_home
.path()
@@ -892,18 +922,24 @@ mod tests {
let mut removed = Vec::new();
remove_stale_remote_plugin_caches(
codex_home.path(),
&PluginStore::new(codex_home.path().to_path_buf()),
&installed_plugin_names_by_marketplace,
&mut removed,
)
.await
.expect("cleanup private shared-with-me cache");
assert_eq!(
removed,
vec![
"created-by-me-plugin@created-by-me-remote".to_string(),
"private-plugin@workspace-shared-with-me-private".to_string(),
[
"created-by-me-plugin@created-by-me-remote",
"private-plugin@workspace-shared-with-me-private",
]
.map(|plugin_id| RemotePluginRemoval {
plugin_id: plugin_id.to_string(),
capabilities: RemotePluginCapabilities::default(),
})
.to_vec()
);
assert!(!created_by_me_cached_manifest.exists());
assert!(!cached_manifest.exists());