plugins: reconcile same-name installations

codex-rs/core-plugins/src/manager.rs

@@ -56,7 +56,7 @@ use crate::tool_suggest_metadata::ToolSuggestMetadataCache;
 use codex_analytics::AnalyticsEventsClient;
 use codex_config::ConfigLayerStack;
 use codex_config::clear_user_plugin;
-use codex_config::set_user_plugin_enabled;
+use codex_config::reconcile_user_plugin_registrations;
 use codex_config::types::PluginConfig;
 use codex_config::types::ToolSuggestDisabledTool;
 use codex_config::types::ToolSuggestDiscoverableType;
@@ -365,6 +365,7 @@ pub struct PluginsManager {
     restriction_product: Option<Product>,
     auth_mode: RwLock<Option<AuthMode>>,
     analytics_events_client: RwLock<Option<AnalyticsEventsClient>>,
+    plugin_mutation_lock: Semaphore,
 }
 
 #[derive(Clone)]
@@ -437,6 +438,7 @@ impl PluginsManager {
             restriction_product,
             auth_mode: RwLock::new(auth_mode),
             analytics_events_client: RwLock::new(None),
+            plugin_mutation_lock: Semaphore::new(/*permits*/ 1),
         }
     }
 
@@ -1258,7 +1260,10 @@ impl PluginsManager {
     ) -> Result<PluginInstallOutcome, PluginInstallError> {
         let resolved = self.resolve_installable_plugin(config_layer_stack, &request)?;
         let plugin_id = resolved.plugin_id.clone();
-        match self.install_resolved_plugin(resolved).await {
+        match self
+            .install_resolved_plugin(resolved, Some(config_layer_stack))
+            .await
+        {
             Ok(outcome) => Ok(outcome),
             Err(err) => {
                 self.track_plugin_install_failed(
@@ -1331,7 +1336,10 @@ impl PluginsManager {
             return Err(err);
         }
         let plugin_id = resolved.plugin_id.clone();
-        match self.install_resolved_plugin(resolved).await {
+        match self
+            .install_resolved_plugin(resolved, Some(&config.config_layer_stack))
+            .await
+        {
             Ok(outcome) => Ok(outcome),
             Err(err) => {
                 self.track_plugin_install_failed(
@@ -1402,7 +1410,13 @@ impl PluginsManager {
     async fn install_resolved_plugin(
         &self,
         resolved: ResolvedMarketplacePlugin,
+        config_layer_stack: Option<&ConfigLayerStack>,
     ) -> Result<PluginInstallOutcome, PluginInstallError> {
+        let _mutation_guard = self
+            .plugin_mutation_lock
+            .acquire()
+            .await
+            .unwrap_or_else(|_| unreachable!("plugin mutation lock should stay open"));
         let auth_policy = resolved.policy.authentication;
         let plugin_version =
             if is_openai_curated_marketplace_name(&resolved.plugin_id.marketplace_name) {
@@ -1449,13 +1463,12 @@ impl PluginsManager {
         .await
         .map_err(PluginInstallError::join)??;
 
-        set_user_plugin_enabled(
-            &self.codex_home,
-            result.plugin_id.as_key(),
-            /*enabled*/ true,
+        self.reconcile_plugin_install_locked(
+            &result.plugin_id,
+            config_layer_stack,
+            /*enable_canonical*/ true,
         )
-        .await
-        .map_err(anyhow::Error::from)?;
+        .await?;
 
         let analytics_events_client = match self.analytics_events_client.read() {
             Ok(client) => client.clone(),
@@ -1476,6 +1489,60 @@ impl PluginsManager {
         })
     }
 
+    pub async fn reconcile_remote_plugin_install(
+        &self,
+        config: &PluginsConfigInput,
+        plugin_id: &PluginId,
+    ) -> Result<(), PluginInstallError> {
+        let _mutation_guard = self
+            .plugin_mutation_lock
+            .acquire()
+            .await
+            .unwrap_or_else(|_| unreachable!("plugin mutation lock should stay open"));
+        self.reconcile_plugin_install_locked(
+            plugin_id,
+            Some(&config.config_layer_stack),
+            /*enable_canonical*/ false,
+        )
+        .await
+    }
+
+    async fn reconcile_plugin_install_locked(
+        &self,
+        plugin_id: &PluginId,
+        config_layer_stack: Option<&ConfigLayerStack>,
+        enable_canonical: bool,
+    ) -> Result<(), PluginInstallError> {
+        reconcile_user_plugin_registrations(
+            &self.codex_home,
+            config_layer_stack,
+            plugin_id.as_key(),
+            plugin_id.plugin_name.clone(),
+            enable_canonical,
+        )
+        .await
+        .map_err(anyhow::Error::from)?;
+        let store = self.store.clone();
+        let canonical_plugin_id = plugin_id.clone();
+        let removed_plugin_ids = tokio::task::spawn_blocking(move || {
+            let removed = store.other_sources(&canonical_plugin_id, /*active_only*/ false)?;
+            removed
+                .iter()
+                .try_for_each(|plugin_id| store.uninstall(plugin_id))?;
+            Ok::<_, PluginStoreError>(removed)
+        })
+        .await
+        .map_err(PluginInstallError::join)??;
+        for removed_plugin_id in removed_plugin_ids {
+            tracing::info!(
+                canonical_plugin = %plugin_id.as_key(),
+                removed_plugin = %removed_plugin_id.as_key(),
+                "removed stale duplicate plugin installation"
+            );
+        }
+        Ok(())
+    }
+
     pub async fn uninstall_plugin(&self, plugin_id: String) -> Result<(), PluginUninstallError> {
         let plugin_id = PluginId::parse(&plugin_id)?;
         self.uninstall_plugin_id(plugin_id).await
@@ -1503,6 +1570,11 @@ impl PluginsManager {
     }
 
     async fn uninstall_plugin_id(&self, plugin_id: PluginId) -> Result<(), PluginUninstallError> {
+        let _mutation_guard = self
+            .plugin_mutation_lock
+            .acquire()
+            .await
+            .unwrap_or_else(|_| unreachable!("plugin mutation lock should stay open"));
         let plugin_telemetry = if self.store.active_plugin_root(&plugin_id).is_some() {
             Some(
                 self.telemetry_metadata_for_installed_plugin(&plugin_id)

codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs

@@ -187,10 +187,6 @@ pub async fn sync_remote_installed_plugin_bundles_once(
         for installed_plugin in installed_plugins {
             let plugin = installed_plugin.plugin;
             let marketplace_name = remote_plugin_canonical_marketplace_name(&plugin)?.to_string();
-            installed_plugin_names_by_marketplace
-                .entry(marketplace_name.clone())
-                .or_default()
-                .insert(plugin.name.clone());
             let plugin_id = match PluginId::new(plugin.name.clone(), marketplace_name.clone()) {
                 Ok(plugin_id) => plugin_id,
                 Err(err) => {
@@ -205,6 +201,13 @@ pub async fn sync_remote_installed_plugin_bundles_once(
                     continue;
                 }
             };
+            if has_active_local_competitor(&store, &plugin_id)? {
+                continue;
+            }
+            installed_plugin_names_by_marketplace
+                .entry(marketplace_name.clone())
+                .or_default()
+                .insert(plugin.name.clone());
             let release_version = plugin
                 .release
                 .version
@@ -254,7 +257,11 @@ pub async fn sync_remote_installed_plugin_bundles_once(
             .await
             {
                 Ok(result) => {
-                    installed_plugin_ids.insert(result.plugin_id.as_key());
+                    if !has_active_local_competitor(&store, &result.plugin_id)? {
+                        installed_plugin_ids.insert(result.plugin_id.as_key());
+                    } else {
+                        store.uninstall(&result.plugin_id)?;
+                    }
                 }
                 Err(err) => {
                     warn!(
@@ -286,6 +293,15 @@ pub async fn sync_remote_installed_plugin_bundles_once(
     })
 }
 
+fn has_active_local_competitor(
+    store: &PluginStore,
+    plugin_id: &PluginId,
+) -> Result<bool, PluginStoreError> {
+    Ok(store
+        .other_sources(plugin_id, /*active_only*/ true)?
+        .iter()
+        .any(|other| RemotePluginScope::from_marketplace_name(&other.marketplace_name).is_none()))
+}
 pub fn mark_remote_plugin_cache_mutation_in_flight(
     codex_home: &Path,
     marketplace_name: &str,
@@ -479,7 +495,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn sync_backfills_remote_plugin_install_metadata_for_current_bundle() {
+    async fn sync_backfills_metadata_and_keeps_local_same_name_install_canonical() {
         let server = MockServer::start().await;
         let codex_home = tempfile::tempdir().expect("create codex home");
         let cached_manifest = codex_home
@@ -494,6 +510,11 @@ mod tests {
             .expect("create cached plugin manifest parent");
         std::fs::write(&cached_manifest, r#"{"name":"linear","version":"1.2.3"}"#)
             .expect("write cached plugin manifest");
+        assert!(fs::create_dir_all(codex_home.path().join("plugins/cache/debug/linear")).is_ok());
+        let stale_remote = codex_home
+            .path()
+            .join("plugins/cache/workspace-shared-with-me-private/linear/local");
+        assert!(fs::create_dir_all(&stale_remote).is_ok());
         let remote_plugin_id = "plugins~Plugin_linear";
         Mock::given(method("GET"))
             .and(path("/backend-api/ps/plugins/installed"))
@@ -517,7 +538,7 @@ mod tests {
                 }],
                 "pagination": {"next_page_token": null},
             })))
-            .expect(1)
+            .expect(2)
             .mount(&server)
             .await;
         Mock::given(method("GET"))
@@ -528,7 +549,7 @@ mod tests {
                 "plugins": [],
                 "pagination": {"next_page_token": null},
             })))
-            .expect(1)
+            .expect(2)
             .mount(&server)
             .await;
         Mock::given(method("GET"))
@@ -539,7 +560,7 @@ mod tests {
                 "plugins": [],
                 "pagination": {"next_page_token": null},
             })))
-            .expect(1)
+            .expect(2)
             .mount(&server)
             .await;
         let config = RemotePluginServiceConfig {
@@ -555,18 +576,23 @@ mod tests {
         .await
         .expect("sync current remote plugin bundle");
 
-        assert_eq!(outcome, RemoteInstalledPluginBundleSyncOutcome::default());
+        assert_eq!(
+            outcome.removed_cache_plugin_ids,
+            vec!["linear@workspace-shared-with-me-private".to_string()]
+        );
+        assert!(!stale_remote.exists());
         let plugin_id = PluginId::new(
             "linear".to_string(),
             REMOTE_GLOBAL_MARKETPLACE_NAME.to_string(),
         )
         .expect("valid plugin id");
-        let metadata_path = PluginStore::new(codex_home.path().to_path_buf())
+        let store = PluginStore::new(codex_home.path().to_path_buf());
+        let metadata = store
             .plugin_base_root(&plugin_id)
             .join(".codex-remote-plugin-install.json");
         assert_eq!(
             serde_json::from_str::<serde_json::Value>(
-                &std::fs::read_to_string(metadata_path.as_path())
+                &std::fs::read_to_string(metadata.as_path())
                     .expect("read remote plugin install metadata")
             )
             .expect("parse remote plugin install metadata"),
@@ -575,6 +601,23 @@ mod tests {
                 "remote_plugin_id": remote_plugin_id,
             })
         );
+        let local_manifest = codex_home
+            .path()
+            .join("plugins/cache/debug/linear/local/.codex-plugin/plugin.json");
+        std::fs::create_dir_all(local_manifest.parent().expect("manifest parent"))
+            .expect("create local plugin manifest parent");
+        std::fs::write(&local_manifest, r#"{"name":"linear"}"#)
+            .expect("write local plugin manifest");
+        store.uninstall(&plugin_id).expect("remove remote cache");
+        let outcome = sync_remote_installed_plugin_bundles_once(
+            codex_home.path().to_path_buf(),
+            &config,
+            Some(&auth),
+        )
+        .await
+        .expect("sync after local plugin becomes canonical");
+        assert_eq!(outcome, RemoteInstalledPluginBundleSyncOutcome::default());
+        assert!(local_manifest.is_file() && !cached_manifest.exists());
     }
 
     #[test]