Use cached remote plugin catalog for plugin list (#26932)

## Summary

This changes the default remote plugin marketplace listing to use the
cached global remote catalog when it is already present on disk. The
foreground `plugin/list` response can then return from the local catalog
cache instead of waiting on `/ps/plugins/list`.

When a cached global catalog was present at the start of the request,
`plugin/list` still schedules a background refresh through the existing
plugin-list background task path so the disk cache is updated for future
requests. Cache misses keep the existing synchronous remote fetch path
and write the cache, and they do not schedule an extra duplicate
background `/ps/plugins/list` refresh.

Installed/enabled state continues to come from the existing remote
installed overlay path. This change only affects the global remote
catalog directory data used by `plugin/list`.

## Testing

- `just fmt`
- `just test -p codex-app-server
plugin_list_uses_cached_global_remote_catalog_and_refreshes_it`
- `just test -p codex-core-plugins`
- `git diff --check`

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

@@ -34,6 +34,7 @@ pub use manager::PluginDetailsUnavailableReason;
 pub use manager::PluginInstallError;
 pub use manager::PluginInstallOutcome;
 pub use manager::PluginInstallRequest;
+pub use manager::PluginListBackgroundTaskOptions;
 pub use manager::PluginReadOutcome;
 pub use manager::PluginReadRequest;
 pub use manager::PluginUninstallError;

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

@@ -146,6 +146,22 @@ struct RemoteInstalledPluginsCacheRefreshState {
     in_flight: bool,
 }
 
+struct GlobalRemoteCatalogCacheRefreshRequest {
+    service_config: RemotePluginServiceConfig,
+    auth: Option<CodexAuth>,
+}
+
+#[derive(Default)]
+struct GlobalRemoteCatalogCacheRefreshState {
+    requested: Option<GlobalRemoteCatalogCacheRefreshRequest>,
+    in_flight: bool,
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct PluginListBackgroundTaskOptions {
+    pub refresh_global_remote_catalog_cache: bool,
+}
+
 #[derive(Clone, PartialEq, Eq)]
 struct NonCuratedCacheRefreshRequest {
     roots: Vec<AbsolutePathBuf>,
@@ -302,6 +318,7 @@ pub struct PluginsManager {
     enabled_outcome_load_semaphore: Semaphore,
     remote_installed_plugins_cache: RwLock<Option<Vec<RemoteInstalledPlugin>>>,
     remote_installed_plugins_cache_refresh_state: RwLock<RemoteInstalledPluginsCacheRefreshState>,
+    global_remote_catalog_cache_refresh_state: RwLock<GlobalRemoteCatalogCacheRefreshState>,
     restriction_product: Option<Product>,
     analytics_events_client: RwLock<Option<AnalyticsEventsClient>>,
 }
@@ -355,6 +372,9 @@ impl PluginsManager {
             remote_installed_plugins_cache_refresh_state: RwLock::new(
                 RemoteInstalledPluginsCacheRefreshState::default(),
             ),
+            global_remote_catalog_cache_refresh_state: RwLock::new(
+                GlobalRemoteCatalogCacheRefreshState::default(),
+            ),
             restriction_product,
             analytics_events_client: RwLock::new(None),
         }
@@ -702,14 +722,33 @@ impl PluginsManager {
         );
     }
 
+    fn maybe_start_global_remote_catalog_cache_refresh(
+        self: &Arc<Self>,
+        config: &PluginsConfigInput,
+        auth: Option<CodexAuth>,
+    ) {
+        if !config.plugins_enabled || !config.remote_plugin_enabled {
+            return;
+        }
+
+        self.schedule_global_remote_catalog_cache_refresh(GlobalRemoteCatalogCacheRefreshRequest {
+            service_config: remote_plugin_service_config(config),
+            auth,
+        });
+    }
+
     pub fn maybe_start_plugin_list_background_tasks_for_config(
         self: &Arc<Self>,
         config: &PluginsConfigInput,
         auth: Option<CodexAuth>,
         roots: &[AbsolutePathBuf],
+        options: PluginListBackgroundTaskOptions,
         on_effective_plugins_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
     ) {
         self.maybe_start_non_curated_plugin_cache_refresh(roots);
+        if options.refresh_global_remote_catalog_cache {
+            self.maybe_start_global_remote_catalog_cache_refresh(config, auth.clone());
+        }
         self.maybe_start_remote_installed_plugins_cache_refresh(
             config,
             auth.clone(),
@@ -1447,6 +1486,33 @@ impl PluginsManager {
         });
     }
 
+    fn schedule_global_remote_catalog_cache_refresh(
+        self: &Arc<Self>,
+        request: GlobalRemoteCatalogCacheRefreshRequest,
+    ) {
+        let should_spawn = {
+            let mut state = match self.global_remote_catalog_cache_refresh_state.write() {
+                Ok(state) => state,
+                Err(err) => err.into_inner(),
+            };
+            state.requested = Some(request);
+            if state.in_flight {
+                false
+            } else {
+                state.in_flight = true;
+                true
+            }
+        };
+        if !should_spawn {
+            return;
+        }
+
+        let manager = Arc::clone(self);
+        tokio::spawn(async move {
+            manager.run_global_remote_catalog_cache_refresh_loop().await;
+        });
+    }
+
     fn schedule_non_curated_plugin_cache_refresh(
         self: &Arc<Self>,
         roots: &[AbsolutePathBuf],
@@ -1613,6 +1679,44 @@ impl PluginsManager {
         }
     }
 
+    async fn run_global_remote_catalog_cache_refresh_loop(self: Arc<Self>) {
+        loop {
+            let request = {
+                let mut state = match self.global_remote_catalog_cache_refresh_state.write() {
+                    Ok(state) => state,
+                    Err(err) => err.into_inner(),
+                };
+                match state.requested.take() {
+                    Some(request) => request,
+                    None => {
+                        state.in_flight = false;
+                        return;
+                    }
+                }
+            };
+
+            match crate::remote::fetch_and_cache_global_remote_plugin_catalog(
+                self.codex_home.as_path(),
+                &request.service_config,
+                request.auth.as_ref(),
+            )
+            .await
+            {
+                Ok(()) => {}
+                Err(
+                    RemotePluginCatalogError::AuthRequired
+                    | RemotePluginCatalogError::UnsupportedAuthMode,
+                ) => {}
+                Err(err) => {
+                    warn!(
+                        error = %err,
+                        "failed to refresh cached global remote plugin catalog"
+                    );
+                }
+            }
+        }
+    }
+
     fn run_non_curated_plugin_cache_refresh_loop(self: Arc<Self>) {
         loop {
             let request = {

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

@@ -576,6 +576,25 @@ pub async fn fetch_remote_marketplaces(
         match source {
             RemoteMarketplaceSource::Global => {
                 let scope = RemotePluginScope::Global;
+                if let Some(codex_home) = global_catalog_cache_path
+                    && let Some(directory_plugins) =
+                        catalog_cache::load_cached_global_directory_plugins(
+                            codex_home, config, auth,
+                        )
+                {
+                    let installed_plugins =
+                        fetch_installed_plugins_for_scope(config, auth, scope).await?;
+                    if let Some(marketplace) = build_remote_marketplace(
+                        scope.marketplace_name(),
+                        scope.marketplace_display_name(),
+                        directory_plugins,
+                        installed_plugins,
+                        /*include_installed_only*/ true,
+                    )? {
+                        marketplaces.push(marketplace);
+                    }
+                    continue;
+                }
                 let (directory_plugins, installed_plugins) = tokio::try_join!(
                     fetch_directory_plugins_for_scope(config, auth, scope),
                     fetch_installed_plugins_for_scope(config, auth, scope),
@@ -692,6 +711,17 @@ pub async fn fetch_and_cache_global_remote_plugin_catalog(
     Ok(())
 }
 
+pub fn has_cached_global_remote_plugin_catalog(
+    codex_home: &Path,
+    config: &RemotePluginServiceConfig,
+    auth: Option<&CodexAuth>,
+) -> bool {
+    let Ok(auth) = ensure_chatgpt_auth(auth) else {
+        return false;
+    };
+    catalog_cache::load_cached_global_directory_plugins(codex_home, config, auth).is_some()
+}
+
 pub fn cached_global_remote_discoverable_plugins(
     codex_home: &Path,
     config: &RemotePluginServiceConfig,