Support remote marketplaces in the plugin CLI (#42150)

## What changed

- Include remote catalog entries in `codex plugin list`, including their source,
  version, install policy, and authentication policy in JSON output.
- Support adding and removing remote plugins through the existing plugin CLI.
- Cache remote catalogs by scope and collection. Prefer fresh cached results, and
  refetch once when an add request misses a plugin in the cache.
- Preserve the local curated catalog when an unfiltered remote listing fails,
  while surfacing errors for explicitly selected remote marketplaces.

## Testing

- Cover remote listing, installation, removal, catalog fallback, cache refresh,
  collection isolation, and install failure behavior.

GitOrigin-RevId: 09796b2c393d102e00ba9289f784d78a2e166a54

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

@@ -204,6 +204,8 @@ pub enum RemoteMarketplaceSource {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum RemotePluginCatalogCacheMode {
     PreferCache,
+    /// Reuse fresh entries and synchronously refresh missing or stale entries.
+    PreferFreshCache,
     ForceRefetch,
 }
 
@@ -211,6 +213,15 @@ pub enum RemotePluginCatalogCacheMode {
 pub struct RemoteMarketplacesFetchOutcome {
     pub marketplaces: Vec<RemoteMarketplace>,
     pub catalog_cache_refresh_scopes: BTreeSet<RemotePluginScope>,
+    /// Whether any requested directory was served from disk cache, even if it was empty.
+    pub catalog_cache_used: bool,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct RemoteMarketplaceFetchOutcome {
+    pub marketplace: Option<RemoteMarketplace>,
+    /// Whether the requested directory was served from disk cache, even if it was empty.
+    pub catalog_cache_used: bool,
 }
 
 #[derive(Debug, Clone, PartialEq)]
@@ -791,6 +802,7 @@ pub async fn fetch_remote_marketplaces(
     let auth = ensure_chatgpt_auth(auth)?;
     let mut marketplaces = Vec::new();
     let mut catalog_cache_refresh_scopes = BTreeSet::new();
+    let mut catalog_cache_used = false;
     let needs_workspace_installed = sources.iter().any(|source| {
         matches!(
             source,
@@ -813,6 +825,7 @@ pub async fn fetch_remote_marketplaces(
                         config,
                         auth,
                         scope,
+                        /*collection*/ None,
                         catalog_cache_mode,
                     ),
                     fetch_installed_plugins_for_scope(config, auth, scope),
@@ -820,6 +833,7 @@ pub async fn fetch_remote_marketplaces(
                 if directory_plugins.cache_refresh_needed {
                     catalog_cache_refresh_scopes.insert(scope);
                 }
+                catalog_cache_used |= directory_plugins.catalog_cache_used;
                 if let Some(marketplace) = build_remote_marketplace(
                     scope.marketplace_name(),
                     scope.marketplace_display_name(),
@@ -838,6 +852,7 @@ pub async fn fetch_remote_marketplaces(
                         config,
                         auth,
                         scope,
+                        /*collection*/ None,
                         catalog_cache_mode,
                     ),
                     fetch_installed_plugins_for_scope(config, auth, scope),
@@ -845,6 +860,7 @@ pub async fn fetch_remote_marketplaces(
                 if directory_plugins.cache_refresh_needed {
                     catalog_cache_refresh_scopes.insert(scope);
                 }
+                catalog_cache_used |= directory_plugins.catalog_cache_used;
                 if let Some(marketplace) = build_remote_marketplace(
                     scope.marketplace_name(),
                     scope.marketplace_display_name(),
@@ -862,12 +878,14 @@ pub async fn fetch_remote_marketplaces(
                     config,
                     auth,
                     scope,
+                    /*collection*/ None,
                     catalog_cache_mode,
                 )
                 .await?;
                 if directory_plugins.cache_refresh_needed {
                     catalog_cache_refresh_scopes.insert(scope);
                 }
+                catalog_cache_used |= directory_plugins.catalog_cache_used;
                 if let Some(marketplace) = build_remote_marketplace(
                     scope.marketplace_name(),
                     scope.marketplace_display_name(),
@@ -942,6 +960,7 @@ pub async fn fetch_remote_marketplaces(
     Ok(RemoteMarketplacesFetchOutcome {
         marketplaces,
         catalog_cache_refresh_scopes,
+        catalog_cache_used,
     })
 }
 
@@ -953,7 +972,9 @@ pub(crate) async fn fetch_and_cache_remote_plugin_catalog(
 ) -> Result<(), RemotePluginCatalogError> {
     let auth = ensure_chatgpt_auth(auth)?;
     let plugins = fetch_directory_plugins_for_scope(config, auth, scope).await?;
-    catalog_cache::write_cached_directory_plugins(codex_home, config, auth, scope, &plugins);
+    catalog_cache::write_cached_directory_plugins(
+        codex_home, config, auth, scope, /*collection*/ None, &plugins,
+    );
     Ok(())
 }
 
@@ -975,7 +996,18 @@ pub fn invalidate_cached_remote_plugin_catalog_scopes(
         return;
     };
     for scope in scopes {
-        catalog_cache::remove_cached_directory_plugins(codex_home, config, auth, *scope);
+        catalog_cache::remove_cached_directory_plugins(
+            codex_home, config, auth, *scope, /*collection*/ None,
+        );
+        if *scope == RemotePluginScope::Global {
+            catalog_cache::remove_cached_directory_plugins(
+                codex_home,
+                config,
+                auth,
+                *scope,
+                Some(OPENAI_CURATED_REMOTE_COLLECTION_KEY),
+            );
+        }
     }
 }
 
@@ -1094,14 +1126,15 @@ pub(crate) fn has_fresh_cached_remote_plugin_catalog(
     let Ok(auth) = ensure_chatgpt_auth(auth) else {
         return false;
     };
-    catalog_cache::load_cached_directory_plugins(codex_home, config, auth, scope).is_some_and(
-        |cached| {
-            matches!(
-                cached.freshness,
-                catalog_cache::RemotePluginCatalogCacheFreshness::Fresh
-            )
-        },
+    catalog_cache::load_cached_directory_plugins(
+        codex_home, config, auth, scope, /*collection*/ None,
     )
+    .is_some_and(|cached| {
+        matches!(
+            cached.freshness,
+            catalog_cache::RemotePluginCatalogCacheFreshness::Fresh
+        )
+    })
 }
 
 pub(crate) fn cached_remote_plugin_catalog_scopes(
@@ -1115,7 +1148,10 @@ pub(crate) fn cached_remote_plugin_catalog_scopes(
     RemotePluginScope::CATALOG_CACHE_SCOPES
         .into_iter()
         .filter(|scope| {
-            catalog_cache::load_cached_directory_plugins(codex_home, config, auth, *scope).is_some()
+            catalog_cache::load_cached_directory_plugins(
+                codex_home, config, auth, *scope, /*collection*/ None,
+            )
+            .is_some()
         })
         .collect()
 }
@@ -1130,6 +1166,7 @@ pub fn cached_global_remote_discoverable_plugins(
         config,
         auth,
         RemotePluginScope::Global,
+        /*collection*/ None,
     )
     .map(|cached| cached.plugins)
     .unwrap_or_default()
@@ -1149,26 +1186,34 @@ pub fn cached_global_remote_discoverable_plugins(
 pub async fn fetch_openai_curated_remote_collection_marketplace(
     config: &RemotePluginServiceConfig,
     auth: Option<&CodexAuth>,
-) -> Result<Option<RemoteMarketplace>, RemotePluginCatalogError> {
+    catalog_cache_root: Option<&Path>,
+    catalog_cache_mode: RemotePluginCatalogCacheMode,
+) -> Result<RemoteMarketplaceFetchOutcome, RemotePluginCatalogError> {
     let auth = ensure_chatgpt_auth(auth)?;
     let scope = RemotePluginScope::Global;
     let (directory_plugins, installed_plugins) = tokio::try_join!(
-        fetch_directory_plugins_for_scope_with_collection(
+        fetch_directory_plugins_for_scope_with_cache(
+            catalog_cache_root,
             config,
             auth,
             scope,
-            OPENAI_CURATED_REMOTE_COLLECTION_KEY,
+            Some(OPENAI_CURATED_REMOTE_COLLECTION_KEY),
+            catalog_cache_mode,
         ),
         fetch_installed_plugins_for_scope(config, auth, scope),
     )?;
 
-    build_remote_marketplace(
+    let marketplace = build_remote_marketplace(
         REMOTE_GLOBAL_MARKETPLACE_NAME,
         REMOTE_GLOBAL_MARKETPLACE_DISPLAY_NAME,
-        directory_plugins,
+        directory_plugins.plugins,
         installed_plugins,
         /*include_installed_only*/ false,
-    )
+    )?;
+    Ok(RemoteMarketplaceFetchOutcome {
+        marketplace,
+        catalog_cache_used: directory_plugins.catalog_cache_used,
+    })
 }
 
 fn build_remote_marketplace(
@@ -1974,6 +2019,7 @@ fn normalize_remote_default_prompt(prompt: &str) -> Option<String> {
 struct DirectoryPluginsFetchOutcome {
     plugins: Vec<RemotePluginDirectoryItem>,
     cache_refresh_needed: bool,
+    catalog_cache_used: bool,
 }
 
 async fn fetch_directory_plugins_for_scope_with_cache(
@@ -1981,12 +2027,16 @@ async fn fetch_directory_plugins_for_scope_with_cache(
     config: &RemotePluginServiceConfig,
     auth: &CodexAuth,
     scope: RemotePluginScope,
+    collection: Option<&str>,
     cache_mode: RemotePluginCatalogCacheMode,
 ) -> Result<DirectoryPluginsFetchOutcome, RemotePluginCatalogError> {
-    if cache_mode == RemotePluginCatalogCacheMode::PreferCache
+    if cache_mode != RemotePluginCatalogCacheMode::ForceRefetch
         && let Some(codex_home) = codex_home
-        && let Some(cached) =
-            catalog_cache::load_cached_directory_plugins(codex_home, config, auth, scope)
+        && let Some(cached) = catalog_cache::load_cached_directory_plugins(
+            codex_home, config, auth, scope, collection,
+        )
+        && (cache_mode == RemotePluginCatalogCacheMode::PreferCache
+            || cached.freshness == catalog_cache::RemotePluginCatalogCacheFreshness::Fresh)
     {
         return Ok(DirectoryPluginsFetchOutcome {
             plugins: cached.plugins,
@@ -1994,16 +2044,22 @@ async fn fetch_directory_plugins_for_scope_with_cache(
                 cached.freshness,
                 catalog_cache::RemotePluginCatalogCacheFreshness::Stale
             ),
+            catalog_cache_used: true,
         });
     }
 
-    let plugins = fetch_directory_plugins_for_scope(config, auth, scope).await?;
+    let plugins =
+        fetch_directory_plugins_for_scope_with_optional_collection(config, auth, scope, collection)
+            .await?;
     if let Some(codex_home) = codex_home {
-        catalog_cache::write_cached_directory_plugins(codex_home, config, auth, scope, &plugins);
+        catalog_cache::write_cached_directory_plugins(
+            codex_home, config, auth, scope, collection, &plugins,
+        );
     }
     Ok(DirectoryPluginsFetchOutcome {
         plugins,
         cache_refresh_needed: false,
+        catalog_cache_used: false,
     })
 }
 
@@ -2018,21 +2074,6 @@ async fn fetch_directory_plugins_for_scope(
     .await
 }
 
-async fn fetch_directory_plugins_for_scope_with_collection(
-    config: &RemotePluginServiceConfig,
-    auth: &CodexAuth,
-    scope: RemotePluginScope,
-    collection: &str,
-) -> Result<Vec<RemotePluginDirectoryItem>, RemotePluginCatalogError> {
-    fetch_directory_plugins_for_scope_with_optional_collection(
-        config,
-        auth,
-        scope,
-        Some(collection),
-    )
-    .await
-}
-
 async fn fetch_directory_plugins_for_scope_with_optional_collection(
     config: &RemotePluginServiceConfig,
     auth: &CodexAuth,

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

@@ -26,6 +26,8 @@ struct RemotePluginCatalogCacheKey {
     // Global catalogs predate scoped cache keys and must keep their existing filenames.
     #[serde(skip_serializing_if = "Option::is_none")]
     scope: Option<RemotePluginScope>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    collection: Option<String>,
 }
 
 impl RemotePluginCatalogCacheKey {
@@ -33,6 +35,7 @@ impl RemotePluginCatalogCacheKey {
         config: &RemotePluginServiceConfig,
         auth: &CodexAuth,
         scope: RemotePluginScope,
+        collection: Option<&str>,
     ) -> Option<Self> {
         let cache_key = Self {
             chatgpt_base_url: config.chatgpt_base_url.clone(),
@@ -40,6 +43,7 @@ impl RemotePluginCatalogCacheKey {
             chatgpt_user_id: auth.get_chatgpt_user_id(),
             is_workspace_account: auth.is_workspace_account(),
             scope: (scope != RemotePluginScope::Global).then_some(scope),
+            collection: collection.map(str::to_owned),
         };
         // Preserve global catalog caching for existing header-auth clients, but never share
         // user or workspace catalogs when the auth mode cannot identify their owner.
@@ -79,8 +83,9 @@ pub(crate) fn load_cached_directory_plugins(
     config: &RemotePluginServiceConfig,
     auth: &CodexAuth,
     scope: RemotePluginScope,
+    collection: Option<&str>,
 ) -> Option<CachedDirectoryPlugins> {
-    let cache_key = RemotePluginCatalogCacheKey::new(config, auth, scope)?;
+    let cache_key = RemotePluginCatalogCacheKey::new(config, auth, scope, collection)?;
     let cache_path = cache_path(codex_home, &cache_key);
     let bytes = match std::fs::read(&cache_path) {
         Ok(bytes) => bytes,
@@ -125,9 +130,10 @@ pub(crate) fn write_cached_directory_plugins(
     config: &RemotePluginServiceConfig,
     auth: &CodexAuth,
     scope: RemotePluginScope,
+    collection: Option<&str>,
     plugins: &[RemotePluginDirectoryItem],
 ) {
-    let Some(cache_key) = RemotePluginCatalogCacheKey::new(config, auth, scope) else {
+    let Some(cache_key) = RemotePluginCatalogCacheKey::new(config, auth, scope, collection) else {
         return;
     };
     let cache_path = cache_path(codex_home, &cache_key);
@@ -146,8 +152,9 @@ pub(crate) fn remove_cached_directory_plugins(
     config: &RemotePluginServiceConfig,
     auth: &CodexAuth,
     scope: RemotePluginScope,
+    collection: Option<&str>,
 ) {
-    let Some(cache_key) = RemotePluginCatalogCacheKey::new(config, auth, scope) else {
+    let Some(cache_key) = RemotePluginCatalogCacheKey::new(config, auth, scope, collection) else {
         return;
     };
     let cache_path = cache_path(codex_home, &cache_key);

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

@@ -18,7 +18,7 @@ fn catalog_cache_freshness_honors_ttl() {
 }
 
 #[test]
-fn catalog_cache_paths_are_isolated_by_scope() {
+fn catalog_cache_paths_are_isolated_by_scope_and_collection() {
     let codex_home = Path::new("/tmp/codex-home");
     let cache_key_for_scope = |scope| RemotePluginCatalogCacheKey {
         chatgpt_base_url: "https://chatgpt.com/backend-api".to_string(),
@@ -26,6 +26,7 @@ fn catalog_cache_paths_are_isolated_by_scope() {
         chatgpt_user_id: Some("user-id".to_string()),
         is_workspace_account: true,
         scope: (scope != RemotePluginScope::Global).then_some(scope),
+        collection: None,
     };
 
     let paths = [
@@ -38,6 +39,10 @@ fn catalog_cache_paths_are_isolated_by_scope() {
     assert_ne!(paths[0], paths[1]);
     assert_ne!(paths[0], paths[2]);
     assert_ne!(paths[1], paths[2]);
+
+    let mut collection_key = cache_key_for_scope(RemotePluginScope::Global);
+    collection_key.collection = Some("vertical".to_string());
+    assert!(!paths.contains(&cache_path(codex_home, &collection_key)));
 }
 
 #[test]
@@ -59,9 +64,14 @@ fn global_catalog_cache_reuses_legacy_cache_file() {
     let contents = serde_json::to_string_pretty(&legacy_cache).expect("serialize legacy cache");
     codex_utils_path::write_atomically(&legacy_cache_path, &contents).expect("write legacy cache");
 
-    let cached =
-        load_cached_directory_plugins(codex_home.path(), &config, &auth, RemotePluginScope::Global)
-            .expect("load legacy global cache");
+    let cached = load_cached_directory_plugins(
+        codex_home.path(),
+        &config,
+        &auth,
+        RemotePluginScope::Global,
+        /*collection*/ None,
+    )
+    .expect("load legacy global cache");
     assert!(cached.plugins.is_empty());
     assert_eq!(cached.freshness, RemotePluginCatalogCacheFreshness::Stale);
 
@@ -70,6 +80,7 @@ fn global_catalog_cache_reuses_legacy_cache_file() {
         &config,
         &auth,
         RemotePluginScope::Global,
+        /*collection*/ None,
         &[],
     );
     let refreshed_cache: RemotePluginCatalogDiskCache = serde_json::from_slice(
@@ -93,11 +104,18 @@ fn header_auth_does_not_cache_private_catalogs_without_a_stable_identity() {
         &config,
         &auth,
         RemotePluginScope::Global,
+        /*collection*/ None,
         &[],
     );
     assert!(
-        load_cached_directory_plugins(codex_home.path(), &config, &auth, RemotePluginScope::Global)
-            .is_some()
+        load_cached_directory_plugins(
+            codex_home.path(),
+            &config,
+            &auth,
+            RemotePluginScope::Global,
+            /*collection*/ None,
+        )
+        .is_some()
     );
 
     for scope in [RemotePluginScope::User, RemotePluginScope::Workspace] {
@@ -107,10 +125,18 @@ fn header_auth_does_not_cache_private_catalogs_without_a_stable_identity() {
             chatgpt_user_id: None,
             is_workspace_account: false,
             scope: Some(scope),
+            collection: None,
         };
         let insecure_cache_path = cache_path(codex_home.path(), &insecure_cache_key);
 
-        write_cached_directory_plugins(codex_home.path(), &config, &auth, scope, &[]);
+        write_cached_directory_plugins(
+            codex_home.path(),
+            &config,
+            &auth,
+            scope,
+            /*collection*/ None,
+            &[],
+        );
         assert!(!insecure_cache_path.exists());
 
         let insecure_cache = RemotePluginCatalogDiskCache {
@@ -122,6 +148,15 @@ fn header_auth_does_not_cache_private_catalogs_without_a_stable_identity() {
         codex_utils_path::write_atomically(&insecure_cache_path, &contents)
             .expect("write insecure cache");
 
-        assert!(load_cached_directory_plugins(codex_home.path(), &config, &auth, scope).is_none());
+        assert!(
+            load_cached_directory_plugins(
+                codex_home.path(),
+                &config,
+                &auth,
+                scope,
+                /*collection*/ None,
+            )
+            .is_none()
+        );
     }
 }

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

@@ -162,8 +162,116 @@ async fn remote_installed_plugins_paginate_across_all_scopes_without_download_ur
     );
 }
 
+#[tokio::test]
+async fn remote_catalog_cache_modes_control_refresh_and_persist_fetched_results() {
+    let server = MockServer::start().await;
+    let plugin = directory_plugin("plugin-gmail", "gmail");
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/list"))
+        .and(query_param("scope", "GLOBAL"))
+        .and(query_param_is_missing("collection"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "plugins": [plugin],
+            "pagination": {"next_page_token": null},
+        })))
+        .expect(3)
+        .mount(&server)
+        .await;
+    let codex_home = tempfile::tempdir().expect("create codex home");
+    let (config, selected_urls) =
+        recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    for (mode, expire_cache, expected_refresh_needed, expected_cache_used) in [
+        (
+            RemotePluginCatalogCacheMode::PreferFreshCache,
+            false,
+            false,
+            false,
+        ),
+        (
+            RemotePluginCatalogCacheMode::PreferFreshCache,
+            false,
+            false,
+            true,
+        ),
+        (RemotePluginCatalogCacheMode::PreferCache, true, true, true),
+        (
+            RemotePluginCatalogCacheMode::PreferFreshCache,
+            false,
+            false,
+            false,
+        ),
+        (
+            RemotePluginCatalogCacheMode::ForceRefetch,
+            false,
+            false,
+            false,
+        ),
+    ] {
+        if expire_cache {
+            let cache_path =
+                std::fs::read_dir(codex_home.path().join("cache/remote_plugin_catalog"))
+                    .expect("read catalog cache directory")
+                    .next()
+                    .expect("catalog cache exists")
+                    .expect("read cache entry")
+                    .path();
+            let mut cached: serde_json::Value =
+                serde_json::from_slice(&std::fs::read(&cache_path).expect("read cache"))
+                    .expect("decode cache");
+            cached["fetched_at"] = serde_json::json!("2000-01-01T00:00:00Z");
+            std::fs::write(
+                cache_path,
+                serde_json::to_vec(&cached).expect("encode cache"),
+            )
+            .expect("expire catalog cache");
+        }
+
+        let outcome = fetch_directory_plugins_for_scope_with_cache(
+            Some(codex_home.path()),
+            &config,
+            &auth,
+            RemotePluginScope::Global,
+            /*collection*/ None,
+            mode,
+        )
+        .await
+        .expect("fetch directory plugins");
+        assert_eq!(
+            (
+                outcome.plugins,
+                outcome.cache_refresh_needed,
+                outcome.catalog_cache_used,
+            ),
+            (
+                vec![plugin.clone()],
+                expected_refresh_needed,
+                expected_cache_used,
+            ),
+        );
+    }
+
+    assert_eq!(
+        recorded_http_client_urls(&selected_urls),
+        vec![
+            format!(
+                "{}/backend-api/ps/plugins/list?scope=GLOBAL&limit=200",
+                server.uri()
+            );
+            3
+        ],
+    );
+    assert!(has_fresh_cached_remote_plugin_catalog(
+        codex_home.path(),
+        &config,
+        Some(&auth),
+        RemotePluginScope::Global,
+    ));
+}
+
 #[test]
-fn cached_remote_plugin_catalog_scopes_returns_existing_scopes() {
+fn catalog_cache_invalidation_clears_global_collections_and_preserves_other_scopes() {
     let codex_home = tempfile::tempdir().expect("create codex home");
     let config = RemotePluginServiceConfig::new(
         "https://chatgpt.com/backend-api".to_string(),
@@ -176,6 +284,7 @@ fn cached_remote_plugin_catalog_scopes_returns_existing_scopes() {
             &config,
             &auth,
             scope,
+            /*collection*/ None,
             &[],
         );
     }
@@ -184,6 +293,36 @@ fn cached_remote_plugin_catalog_scopes_returns_existing_scopes() {
         cached_remote_plugin_catalog_scopes(codex_home.path(), &config, Some(&auth)),
         BTreeSet::from([RemotePluginScope::Global, RemotePluginScope::Workspace])
     );
+    catalog_cache::write_cached_directory_plugins(
+        codex_home.path(),
+        &config,
+        &auth,
+        RemotePluginScope::Global,
+        Some(OPENAI_CURATED_REMOTE_COLLECTION_KEY),
+        &[],
+    );
+
+    invalidate_cached_remote_plugin_catalog_scopes(
+        codex_home.path(),
+        &config,
+        Some(&auth),
+        &[RemotePluginScope::Global],
+    );
+
+    assert_eq!(
+        cached_remote_plugin_catalog_scopes(codex_home.path(), &config, Some(&auth)),
+        BTreeSet::from([RemotePluginScope::Workspace]),
+    );
+    assert!(
+        catalog_cache::load_cached_directory_plugins(
+            codex_home.path(),
+            &config,
+            &auth,
+            RemotePluginScope::Global,
+            Some(OPENAI_CURATED_REMOTE_COLLECTION_KEY),
+        )
+        .is_none()
+    );
 }
 
 #[test]