Implement remote plugin search (#36409)

## What changed

- Implement `plugin/search` by querying the remote plugin service without using the catalog cache.
- Support global, workspace, and personal scopes with bounded page sizes and passthrough cursors.
- Respect plugin feature gates and omit shared workspace results when plugin sharing is disabled.
- Keep search terms and pagination tokens out of transport errors and telemetry, and return search results as uninstalled plugin summaries.

## Testing

- Add remote search coverage for request parameters, result conversion, authentication, pagination, scope mapping, and error redaction.
- Add app-server coverage for remote-plugin and plugin-sharing feature gates.

GitOrigin-RevId: ac29c5480ed8089d998b6275bc5f11c8d9a43fd1

codex-rs/app-server/README.md

@@ -224,7 +224,7 @@ Example with notification opt-out:
 - `marketplace/remove` — remove a configured marketplace by name from the user marketplace config, and delete its installed marketplace root when one exists.
 - `marketplace/upgrade` — upgrade all configured Git plugin marketplaces, or one named marketplace when `marketplaceName` is provided. Returns selected marketplace names, upgraded roots, and per-marketplace errors.
 - `plugin/list` — list discovered plugin marketplaces and plugin state, including effective marketplace install/auth policy metadata, nullable remote install-policy provenance in `installPolicySource` (`WORKSPACE_SETTING` or `IMPLICIT_CANONICAL_APP`), the remote marketplace `version` and locally materialized `localVersion` when available, plugin `availability` (`AVAILABLE` by default or `DISABLED_BY_ADMIN` for remote plugins blocked upstream), fail-open `marketplaceLoadErrors` entries for marketplace files that could not be parsed or loaded, and best-effort `featuredPluginIds` for the official curated marketplace. Every `PluginSummary` returned by plugin list, installed, read, and share-list methods includes nullable `disabledReason` and `eligiblePlanTypes`, preserving plugin-service availability metadata and raw plan identifiers for remote plugins while returning `null` for local plugins or older remote responses. The same summaries include `mustShowInstallationInterstitial`: remote service values preserve `true` or `false`, while local plugins and remote responses that omit the policy return `null`. Clients should fail closed when the value is `null`. Clients can explicitly request the remote `workspace-directory`, `shared-with-me`, or `created-by-me-remote` marketplace kinds. Set `forceRefetch: true` to bypass TTL-backed remote catalog caches for the requested marketplaces and wait for fresh data; cache entries are replaced only after a successful fetch. When local marketplaces are included, the request also waits for configured plugin caches to reconcile before marketplace summaries are returned. At app-server startup, existing cached catalogs remain available to `plugin/list` while they refresh in the background. `interface.category` uses the marketplace category when present; otherwise it falls back to the plugin manifest category (**under development; do not call from production clients yet**).
-- `plugin/search` — declares the paginated plugin-search request and response shapes. The RPC currently returns JSON-RPC method-not-found (`-32601`) with `plugin/search is not implemented` (**under development; do not call from production clients yet**).
+- `plugin/search` — search the remote plugin service directly without using cached plugin catalogs. Accepts a `searchTerm`, optional `global`, `workspace`, or `personal` scope, and optional `cursor` and `limit`; `personal` searches user-owned plugins. When `remote_plugin` is disabled, an omitted scope is treated as `workspace`, explicit workspace search remains available, and global or personal search returns an empty page without querying the remote service. Returns marketplace-qualified plugin summaries in `data` and passes the remote pagination token through unchanged as `nextCursor`. When `plugin_sharing` is disabled, shared/private workspace results are omitted after the remote page is fetched, so a page can contain fewer than `limit` entries while retaining its upstream `nextCursor`. Because this endpoint does not join results with the installed-plugin snapshot, every returned summary has `installed: false` (**under development; do not call from production clients yet**).
 - `plugin/installed` — list installed plugin rows plus any explicitly requested local install-suggestion plugin names, without fetching the broader remote catalog. Remote rows include nullable `installPolicySource` and `installedAt`, the backend installation timestamp in Unix seconds. `installedAt` is also returned by `plugin/list`, `plugin/read`, and `plugin/share/list`; it is `null` for local plugins, uninstalled plugins, plugins installed by default, and older backend responses that do not include an installation timestamp. Mention surfaces can use this narrower view when they need plugin mention payloads rather than plugin-page discovery data (**under development; do not call from production clients yet**).
 - `plugin/read` — read one plugin by `marketplacePath` plus `pluginName`, returning marketplace info, a list-style `summary`, manifest descriptions/interface metadata, and bundled skills/hooks/apps/MCP server names. Remote plugin details can include scheduled task summaries from the catalog; `scheduledTasks: null` means the metadata is unavailable, while an empty array means the catalog found no scheduled tasks. Remote plugin details expose the canonical `shareUrl` supplied by the remote catalog when available; it is `null` for local plugins or when the catalog omits it. This field is separate from `summary.shareContext`, which continues to describe user and workspace sharing state. For owned workspace plugins, `summary.shareContext.canPublishToWorkspace` reports whether the current user may add the plugin to the workspace directory; `plugin/share/save` returns the same capability after creating or updating a share, and clients should fail closed when either value is `null`. Remote skill interfaces expose `iconSmallUrl` and `iconLargeUrl` when the catalog supplies icon URLs. Returned plugin skills include their current `enabled` state after local config filtering; bundled hooks are returned as lightweight declaration summaries keyed for correlation with `hooks/list`. Use `plugin/install`'s `appsNeedingAuth` to drive post-install authentication and `app/list`'s `isAccessible` to determine current connector accessibility (**under development; do not call from production clients yet**).
 - `plugin/skill/read` — read remote plugin skill markdown on demand by `remoteMarketplaceName`, `remotePluginId`, and `skillName`. This lets clients preview uninstalled remote plugin skills without downloading the plugin bundle.

codex-rs/app-server/src/request_processors/plugins/search.rs

@@ -1,12 +1,123 @@
 use super::*;
-use crate::error_code::method_not_found;
 use codex_app_server_protocol::PluginSearchParams;
+use codex_app_server_protocol::PluginSearchResponse;
+use codex_app_server_protocol::PluginSearchResult;
+use codex_app_server_protocol::PluginSearchScope;
+use codex_core_plugins::remote::RemotePluginSearchRequest;
+use codex_core_plugins::remote::search_remote_plugins;
+
+const DEFAULT_PLUGIN_SEARCH_LIMIT: u32 = 16;
+const MAX_PLUGIN_SEARCH_LIMIT: u32 = 1_000;
 
 impl PluginRequestProcessor {
     pub(crate) async fn plugin_search(
         &self,
-        _params: PluginSearchParams,
+        params: PluginSearchParams,
     ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
-        Err(method_not_found("plugin/search is not implemented"))
+        self.plugin_search_response(params)
+            .await
+            .map(|response| Some(response.into()))
+    }
+
+    async fn plugin_search_response(
+        &self,
+        params: PluginSearchParams,
+    ) -> Result<PluginSearchResponse, JSONRPCErrorError> {
+        let PluginSearchParams {
+            search_term,
+            scope,
+            cwds: _,
+            cursor,
+            limit,
+        } = params;
+        let search_term = search_term.trim();
+        let empty_response = || PluginSearchResponse {
+            data: Vec::new(),
+            next_cursor: None,
+        };
+        if search_term.is_empty() {
+            return Ok(empty_response());
+        }
+
+        let config = self.load_latest_config(/*fallback_cwd*/ None).await?;
+        if !config.features.enabled(Feature::Plugins) {
+            return Ok(empty_response());
+        }
+        let scope = if config.features.enabled(Feature::RemotePlugin) {
+            scope
+        } else {
+            match scope {
+                None | Some(PluginSearchScope::Workspace) => Some(PluginSearchScope::Workspace),
+                Some(PluginSearchScope::Global | PluginSearchScope::Personal) => {
+                    return Ok(empty_response());
+                }
+            }
+        };
+        let plugin_sharing_enabled = config.features.enabled(Feature::PluginSharing);
+
+        let auth = self.auth_manager.auth().await;
+        if !self
+            .workspace_codex_plugins_enabled(&config, auth.as_ref())
+            .await
+            || !auth
+                .as_ref()
+                .map(CodexAuth::api_auth_mode)
+                .is_some_and(DomainAuthMode::uses_codex_backend)
+        {
+            return Ok(empty_response());
+        }
+
+        let scope = scope.map(|scope| match scope {
+            PluginSearchScope::Global => RemotePluginScope::Global,
+            PluginSearchScope::Workspace => RemotePluginScope::Workspace,
+            PluginSearchScope::Personal => RemotePluginScope::User,
+        });
+        let limit = limit
+            .unwrap_or(DEFAULT_PLUGIN_SEARCH_LIMIT)
+            .clamp(1, MAX_PLUGIN_SEARCH_LIMIT);
+        let page = search_remote_plugins(
+            &remote_plugin_service_config(&config),
+            auth.as_ref(),
+            RemotePluginSearchRequest {
+                query: search_term,
+                scope,
+                limit,
+                page_token: cursor.as_deref(),
+            },
+        )
+        .await
+        .map_err(|err| {
+            remote_plugin_catalog_error_to_jsonrpc(err, "search remote plugin catalog")
+        })?;
+
+        let next_cursor = page.next_page_token;
+        let mut data = Vec::with_capacity(page.plugins.len());
+        for plugin in page.plugins {
+            let plugin_id = PluginId::parse(&plugin.id).map_err(|err| {
+                internal_error(format!("invalid remote plugin search result id: {err}"))
+            })?;
+
+            // NOTE: (brisebois) filter out plugins from the results that belong to "shared"
+            // marketplaces if plugin sharing is disabled. There is a chance that this filters
+            // out all results and returns an empty list to the client. Ideally this filtering
+            // would be done server-side to avoid this problem.
+            if !plugin_sharing_enabled
+                && matches!(
+                    plugin_id.marketplace_name.as_str(),
+                    REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME
+                        | REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME
+                        | REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME
+                )
+            {
+                continue;
+            }
+            data.push(PluginSearchResult {
+                plugin: remote_plugin_summary_to_info(plugin),
+                marketplace_name: plugin_id.marketplace_name,
+                marketplace_path: None,
+            });
+        }
+
+        Ok(PluginSearchResponse { data, next_cursor })
     }
 }

codex-rs/app-server/tests/suite/v2/plugin_search.rs

@@ -1,19 +1,98 @@
 use std::time::Duration;
 
 use anyhow::Result;
+use app_test_support::ChatGptAuthFixture;
 use app_test_support::TestAppServer;
-use codex_app_server_protocol::JSONRPCError;
+use app_test_support::write_chatgpt_auth;
 use codex_app_server_protocol::PluginSearchParams;
-use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::PluginSearchResponse;
+use codex_app_server_protocol::PluginSearchScope;
+use codex_config::types::AuthCredentialsStoreMode;
 use pretty_assertions::assert_eq;
+use serde_json::json;
 use tempfile::TempDir;
 use tokio::time::timeout;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
+use wiremock::matchers::query_param;
+use wiremock::matchers::query_param_is_missing;
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
 
 #[tokio::test]
-async fn plugin_search_returns_not_implemented() -> Result<()> {
+async fn plugin_search_omits_shared_workspace_results_when_plugin_sharing_disabled() -> Result<()> {
     let codex_home = TempDir::new()?;
+    let server = MockServer::start().await;
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            r#"chatgpt_base_url = "{}/backend-api/"
+
+[features]
+plugins = true
+remote_plugin = true
+plugin_sharing = false
+"#,
+            server.uri()
+        ),
+    )?;
+    write_chatgpt_auth(
+        codex_home.path(),
+        ChatGptAuthFixture::new("chatgpt-token")
+            .account_id("account-123")
+            .chatgpt_user_id("user-123")
+            .chatgpt_account_id("account-123"),
+        AuthCredentialsStoreMode::File,
+    )?;
+
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/search"))
+        .and(query_param("q", "linear"))
+        .and(query_param("limit", "16"))
+        .and(query_param("pageToken", "incoming-token"))
+        .and(query_param_is_missing("scope"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "plugins": [
+                remote_plugin_json(
+                    "plugin-global",
+                    "global-linear",
+                    "GLOBAL",
+                    /*discoverability*/ None,
+                ),
+                remote_plugin_json(
+                    "plugin-user",
+                    "personal-linear",
+                    "USER",
+                    /*discoverability*/ None,
+                ),
+                remote_plugin_json(
+                    "plugin-listed",
+                    "listed-linear",
+                    "WORKSPACE",
+                    /*discoverability*/ Some("LISTED"),
+                ),
+                remote_plugin_json(
+                    "plugin-private",
+                    "private-linear",
+                    "WORKSPACE",
+                    /*discoverability*/ Some("PRIVATE"),
+                ),
+                remote_plugin_json(
+                    "plugin-unlisted",
+                    "unlisted-linear",
+                    "WORKSPACE",
+                    /*discoverability*/ Some("UNLISTED"),
+                ),
+            ],
+            "pagination": {"next_page_token": "outgoing-token"},
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+
     let mut mcp = TestAppServer::builder()
         .with_codex_home(codex_home.path())
         .without_auto_env()
@@ -25,17 +104,174 @@ async fn plugin_search_returns_not_implemented() -> Result<()> {
             search_term: "linear".to_string(),
             scope: None,
             cwds: None,
-            cursor: None,
+            cursor: Some("incoming-token".to_string()),
             limit: None,
         })
         .await?;
-    let error: JSONRPCError = timeout(
-        DEFAULT_TIMEOUT,
-        mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
-    )
-    .await??;
+    let response: PluginSearchResponse =
+        timeout(DEFAULT_TIMEOUT, mcp.read_response(request_id)).await??;
 
-    assert_eq!(error.error.code, -32601);
-    assert_eq!(error.error.message, "plugin/search is not implemented");
+    assert_eq!(response.next_cursor.as_deref(), Some("outgoing-token"));
+    assert_eq!(
+        response
+            .data
+            .iter()
+            .map(|result| { (result.marketplace_name.as_str(), result.plugin.id.as_str(),) })
+            .collect::<Vec<_>>(),
+        vec![
+            (
+                "openai-curated-remote",
+                "global-linear@openai-curated-remote"
+            ),
+            (
+                "created-by-me-remote",
+                "personal-linear@created-by-me-remote"
+            ),
+            ("workspace-directory", "listed-linear@workspace-directory"),
+        ]
+    );
     Ok(())
 }
+
+#[tokio::test]
+async fn plugin_search_only_searches_workspace_when_remote_plugin_disabled() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let server = MockServer::start().await;
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            r#"chatgpt_base_url = "{}/backend-api/"
+
+[features]
+plugins = true
+remote_plugin = false
+"#,
+            server.uri()
+        ),
+    )?;
+    write_chatgpt_auth(
+        codex_home.path(),
+        ChatGptAuthFixture::new("chatgpt-token")
+            .account_id("account-123")
+            .chatgpt_user_id("user-123")
+            .chatgpt_account_id("account-123"),
+        AuthCredentialsStoreMode::File,
+    )?;
+
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/search"))
+        .and(query_param("q", "linear"))
+        .and(query_param("scope", "WORKSPACE"))
+        .and(query_param("limit", "16"))
+        .and(query_param_is_missing("pageToken"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "plugins": [remote_plugin_json(
+                "plugin-workspace",
+                "workspace-linear",
+                "WORKSPACE",
+                /*discoverability*/ Some("LISTED"),
+            )],
+            "pagination": {"next_page_token": null},
+        })))
+        .expect(2)
+        .mount(&server)
+        .await;
+
+    let mut mcp = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_auto_env()
+        .build_initialized_with_timeout(DEFAULT_TIMEOUT)
+        .await?;
+
+    for scope in [None, Some(PluginSearchScope::Workspace)] {
+        let request_id = mcp
+            .send_plugin_search_request(PluginSearchParams {
+                search_term: "linear".to_string(),
+                scope,
+                cwds: None,
+                cursor: None,
+                limit: None,
+            })
+            .await?;
+        let response: PluginSearchResponse =
+            timeout(DEFAULT_TIMEOUT, mcp.read_response(request_id)).await??;
+
+        assert_eq!(
+            response
+                .data
+                .iter()
+                .map(|result| (result.marketplace_name.as_str(), result.plugin.id.as_str(),))
+                .collect::<Vec<_>>(),
+            vec![(
+                "workspace-directory",
+                "workspace-linear@workspace-directory",
+            )]
+        );
+    }
+
+    for scope in [PluginSearchScope::Global, PluginSearchScope::Personal] {
+        let request_id = mcp
+            .send_plugin_search_request(PluginSearchParams {
+                search_term: "linear".to_string(),
+                scope: Some(scope),
+                cwds: None,
+                cursor: None,
+                limit: None,
+            })
+            .await?;
+        let response: PluginSearchResponse =
+            timeout(DEFAULT_TIMEOUT, mcp.read_response(request_id)).await??;
+
+        assert_eq!(
+            response,
+            PluginSearchResponse {
+                data: Vec::new(),
+                next_cursor: None,
+            }
+        );
+    }
+
+    let search_requests = server
+        .received_requests()
+        .await
+        .expect("wiremock should record requests")
+        .into_iter()
+        .filter(|request| request.url.path() == "/backend-api/ps/plugins/search")
+        .collect::<Vec<_>>();
+    assert_eq!(search_requests.len(), 2);
+    assert_eq!(
+        search_requests
+            .iter()
+            .filter_map(|request| {
+                request
+                    .url
+                    .query_pairs()
+                    .find(|(name, _value)| name == "scope")
+                    .map(|(_name, value)| value.into_owned())
+            })
+            .collect::<Vec<_>>(),
+        vec!["WORKSPACE", "WORKSPACE"]
+    );
+    Ok(())
+}
+
+fn remote_plugin_json(
+    remote_plugin_id: &str,
+    plugin_name: &str,
+    scope: &str,
+    discoverability: Option<&str>,
+) -> serde_json::Value {
+    json!({
+        "id": remote_plugin_id,
+        "name": plugin_name,
+        "scope": scope,
+        "discoverability": discoverability,
+        "installation_policy": "AVAILABLE",
+        "authentication_policy": "ON_USE",
+        "release": {
+            "display_name": plugin_name,
+            "description": format!("{plugin_name} description"),
+            "interface": {},
+        },
+    })
+}

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

@@ -47,6 +47,7 @@ use url::Url;
 
 mod catalog_cache;
 mod remote_installed_plugin_sync;
+mod search;
 mod share;
 
 #[cfg(test)]
@@ -60,6 +61,9 @@ pub use remote_installed_plugin_sync::RemotePluginMaterialization;
 pub use remote_installed_plugin_sync::mark_remote_plugin_cache_mutation_in_flight;
 pub(crate) use remote_installed_plugin_sync::maybe_start_remote_installed_plugin_bundle_sync;
 pub use remote_installed_plugin_sync::sync_remote_installed_plugin_bundles_once;
+pub use search::RemotePluginSearchPage;
+pub use search::RemotePluginSearchRequest;
+pub use search::search_remote_plugins;
 pub use share::RemotePluginShareAccessPolicy;
 pub use share::RemotePluginShareDiscoverability;
 pub use share::RemotePluginSharePrincipal;

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

@@ -0,0 +1,91 @@
+use super::RemotePluginCatalogError;
+use super::RemotePluginListResponse;
+use super::RemotePluginScope;
+use super::RemotePluginServiceConfig;
+use super::RemotePluginSummary;
+use super::authenticated_request;
+use super::build_remote_plugin_summary;
+use super::ensure_chatgpt_auth;
+use super::send_and_decode;
+use codex_login::CodexAuth;
+use http::Method;
+use tracing::instrument;
+use url::Url;
+
+/// Search parameters forwarded directly to plugin-service.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct RemotePluginSearchRequest<'a> {
+    pub query: &'a str,
+    pub scope: Option<RemotePluginScope>,
+    pub limit: u32,
+    pub page_token: Option<&'a str>,
+}
+
+/// One uncached page of remote plugin search results.
+#[derive(Debug, Clone, PartialEq)]
+pub struct RemotePluginSearchPage {
+    pub plugins: Vec<RemotePluginSummary>,
+    pub next_page_token: Option<String>,
+}
+
+/// Searches plugin-service without reading or populating the remote catalog cache.
+#[instrument(
+    level = "debug",
+    skip_all,
+    fields(plugin.scope = ?search.scope, plugin.limit = search.limit)
+)]
+pub async fn search_remote_plugins(
+    config: &RemotePluginServiceConfig,
+    auth: Option<&CodexAuth>,
+    search: RemotePluginSearchRequest<'_>,
+) -> Result<RemotePluginSearchPage, RemotePluginCatalogError> {
+    let auth = ensure_chatgpt_auth(auth)?;
+    let base_url = config.chatgpt_base_url.trim_end_matches('/');
+    let mut url = Url::parse(&format!("{base_url}/ps/plugins/search"))
+        .map_err(RemotePluginCatalogError::InvalidBaseUrl)?;
+    // Search terms and page tokens can contain user data. Keep the queryless endpoint for
+    // diagnostics so neither value is exposed through errors or telemetry.
+    let url_for_error = url.to_string();
+
+    {
+        let mut query_pairs = url.query_pairs_mut();
+        query_pairs.append_pair("q", search.query);
+        if let Some(scope) = search.scope {
+            query_pairs.append_pair("scope", scope.api_value());
+        }
+        query_pairs.append_pair("limit", &search.limit.to_string());
+        if let Some(page_token) = search.page_token {
+            query_pairs.append_pair("pageToken", page_token);
+        }
+    }
+
+    let url = url.to_string();
+    let request = authenticated_request(config.http_request(Method::GET, &url), auth);
+    let response: RemotePluginListResponse = send_and_decode(request, &url_for_error)
+        .await
+        .map_err(|error| match error {
+            RemotePluginCatalogError::Request { url, source } => {
+                RemotePluginCatalogError::Request {
+                    url,
+                    source: source.without_url(),
+                }
+            }
+            other => other,
+        })?;
+    let plugins = response
+        .plugins
+        .iter()
+        // Search intentionally does not join against `/ps/plugins/installed`, so these
+        // summaries always report `installed: false`.
+        .map(|plugin| build_remote_plugin_summary(plugin, /*installed_plugin*/ None))
+        .collect::<Result<Vec<_>, _>>()?;
+
+    Ok(RemotePluginSearchPage {
+        plugins,
+        next_page_token: response.pagination.next_page_token,
+    })
+}
+
+#[cfg(test)]
+#[path = "search_tests.rs"]
+mod tests;

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

@@ -0,0 +1,490 @@
+use super::*;
+use crate::remote::REMOTE_CREATED_BY_ME_MARKETPLACE_NAME;
+use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME;
+use crate::remote::REMOTE_WORKSPACE_MARKETPLACE_NAME;
+use crate::remote::REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME;
+use crate::remote::RemotePluginShareDiscoverability;
+use crate::test_support::recorded_http_client_urls;
+use crate::test_support::recording_remote_plugin_service_config;
+use codex_app_server_protocol::PluginAuthPolicy;
+use codex_app_server_protocol::PluginAvailability;
+use codex_app_server_protocol::PluginDisabledReason;
+use codex_app_server_protocol::PluginInstallPolicy;
+use codex_app_server_protocol::PluginInstallPolicySource;
+use codex_app_server_protocol::PluginInterface;
+use http::StatusCode;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::header;
+use wiremock::matchers::header_exists;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
+use wiremock::matchers::query_param;
+use wiremock::matchers::query_param_is_missing;
+
+fn remote_plugin_json(remote_plugin_id: &str, plugin_name: &str, scope: &str) -> serde_json::Value {
+    let discoverability = (scope == "WORKSPACE").then_some("LISTED");
+    json!({
+        "id": remote_plugin_id,
+        "name": plugin_name,
+        "scope": scope,
+        "discoverability": discoverability,
+        "installation_policy": "AVAILABLE",
+        "authentication_policy": "ON_USE",
+        "release": {
+            "display_name": plugin_name,
+            "description": format!("{plugin_name} description"),
+            "interface": {},
+        },
+    })
+}
+
+#[tokio::test]
+async fn search_remote_plugins_forwards_parameters_and_converts_results() {
+    let server = MockServer::start().await;
+    let remote_plugin = json!({
+        "id": "plugins~Plugin_linear",
+        "name": "linear",
+        "scope": "GLOBAL",
+        "installation_policy": "NOT_AVAILABLE",
+        "installation_policy_source": "WORKSPACE_SETTING",
+        "must_show_installation_interstitial": true,
+        "authentication_policy": "ON_INSTALL",
+        "status": "DISABLED_BY_ADMIN",
+        "disabled_reason": "plan_not_eligible",
+        "eligible_plan_types": ["pro"],
+        "release": {
+            "version": "1.2.3",
+            "display_name": "Linear",
+            "description": "Track issues",
+            "keywords": ["issues"],
+            "interface": {
+                "short_description": "Issue tracking",
+                "category": "Productivity",
+                "capabilities": ["Create issues"],
+                "logo_url": "https://example.com/linear.png",
+            },
+        },
+    });
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/search"))
+        .and(query_param("q", "linear & docs/+"))
+        .and(query_param("scope", "GLOBAL"))
+        .and(query_param("limit", "16"))
+        .and(query_param("pageToken", "next page/+"))
+        .and(header("authorization", "Bearer Access Token"))
+        .and(header("chatgpt-account-id", "account_id"))
+        .and(header("oai-product-sku", "codex"))
+        .and(header_exists("user-agent"))
+        .and(header_exists("originator"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "plugins": [remote_plugin],
+            "pagination": {"next_page_token": "later page/+"},
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let (config, selected_urls) =
+        recording_remote_plugin_service_config(format!("{}/backend-api/", server.uri()));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    let result = search_remote_plugins(
+        &config,
+        Some(&auth),
+        RemotePluginSearchRequest {
+            query: "linear & docs/+",
+            scope: Some(RemotePluginScope::Global),
+            limit: 16,
+            page_token: Some("next page/+"),
+        },
+    )
+    .await
+    .expect("plugin search should succeed");
+
+    assert_eq!(
+        result,
+        RemotePluginSearchPage {
+            plugins: vec![RemotePluginSummary {
+                id: format!("linear@{REMOTE_GLOBAL_MARKETPLACE_NAME}"),
+                remote_plugin_id: "plugins~Plugin_linear".to_string(),
+                version: Some("1.2.3".to_string()),
+                local_version: None,
+                name: "linear".to_string(),
+                share_context: None,
+                installed: false,
+                installed_at: None,
+                enabled: false,
+                install_policy: PluginInstallPolicy::NotAvailable,
+                install_policy_source: Some(PluginInstallPolicySource::WorkspaceSetting),
+                must_show_installation_interstitial: Some(true),
+                auth_policy: PluginAuthPolicy::OnInstall,
+                availability: PluginAvailability::DisabledByAdmin,
+                disabled_reason: Some(PluginDisabledReason::PlanNotEligible),
+                eligible_plan_types: Some(vec!["pro".to_string()]),
+                interface: Some(PluginInterface {
+                    display_name: Some("Linear".to_string()),
+                    short_description: Some("Issue tracking".to_string()),
+                    long_description: None,
+                    developer_name: None,
+                    category: Some("Productivity".to_string()),
+                    capabilities: vec!["Create issues".to_string()],
+                    website_url: None,
+                    privacy_policy_url: None,
+                    terms_of_service_url: None,
+                    default_prompt: None,
+                    brand_color: None,
+                    composer_icon: None,
+                    composer_icon_url: None,
+                    logo: None,
+                    logo_dark: None,
+                    logo_url: Some("https://example.com/linear.png".to_string()),
+                    logo_url_dark: None,
+                    screenshots: Vec::new(),
+                    screenshot_urls: Vec::new(),
+                }),
+                keywords: vec!["issues".to_string()],
+            }],
+            next_page_token: Some("later page/+".to_string()),
+        }
+    );
+    assert_eq!(
+        recorded_http_client_urls(&selected_urls),
+        vec![format!(
+            "{}/backend-api/ps/plugins/search?q=linear+%26+docs%2F%2B&scope=GLOBAL&limit=16&pageToken=next+page%2F%2B",
+            server.uri()
+        )]
+    );
+}
+
+#[tokio::test]
+async fn search_remote_plugins_omits_optional_scope_and_page_token() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/search"))
+        .and(query_param("q", "calendar"))
+        .and(query_param("limit", "25"))
+        .and(query_param_is_missing("scope"))
+        .and(query_param_is_missing("pageToken"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "plugins": [],
+            "pagination": {"next_page_token": null},
+        })))
+        .expect(2)
+        .mount(&server)
+        .await;
+    let (config, _) =
+        recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    for _ in 0..2 {
+        let result = search_remote_plugins(
+            &config,
+            Some(&auth),
+            RemotePluginSearchRequest {
+                query: "calendar",
+                scope: None,
+                limit: 25,
+                page_token: None,
+            },
+        )
+        .await
+        .expect("unscoped plugin search should succeed");
+
+        assert_eq!(
+            result,
+            RemotePluginSearchPage {
+                plugins: Vec::new(),
+                next_page_token: None,
+            }
+        );
+    }
+}
+
+#[tokio::test]
+async fn search_remote_plugins_forwards_each_supported_scope() {
+    let server = MockServer::start().await;
+    let (config, _) =
+        recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    for (scope, expected_scope) in [
+        (RemotePluginScope::Global, "GLOBAL"),
+        (RemotePluginScope::User, "USER"),
+        (RemotePluginScope::Workspace, "WORKSPACE"),
+    ] {
+        Mock::given(method("GET"))
+            .and(path("/backend-api/ps/plugins/search"))
+            .and(query_param("scope", expected_scope))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+                "plugins": [],
+                "pagination": {"next_page_token": null},
+            })))
+            .expect(1)
+            .mount(&server)
+            .await;
+
+        search_remote_plugins(
+            &config,
+            Some(&auth),
+            RemotePluginSearchRequest {
+                query: "calendar",
+                scope: Some(scope),
+                limit: 16,
+                page_token: None,
+            },
+        )
+        .await
+        .expect("scoped plugin search should succeed");
+    }
+}
+
+#[tokio::test]
+async fn search_remote_plugins_preserves_order_and_canonical_marketplaces() {
+    let server = MockServer::start().await;
+    let mut shared_plugin = remote_plugin_json("plugin-shared", "shared", "WORKSPACE");
+    shared_plugin["discoverability"] = json!("PRIVATE");
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/search"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "plugins": [
+                remote_plugin_json("plugin-user", "personal", "USER"),
+                remote_plugin_json("plugin-global", "global", "GLOBAL"),
+                remote_plugin_json("plugin-workspace", "workspace", "WORKSPACE"),
+                shared_plugin,
+            ],
+            "pagination": {"next_page_token": null},
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let (config, _) =
+        recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    let result = search_remote_plugins(
+        &config,
+        Some(&auth),
+        RemotePluginSearchRequest {
+            query: "plugin",
+            scope: None,
+            limit: 16,
+            page_token: None,
+        },
+    )
+    .await
+    .expect("mixed-scope plugin search should succeed");
+
+    let identities = result
+        .plugins
+        .into_iter()
+        .map(|plugin| {
+            (
+                plugin.id,
+                plugin.remote_plugin_id,
+                plugin.share_context.map(|context| context.discoverability),
+            )
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(
+        identities,
+        vec![
+            (
+                format!("personal@{REMOTE_CREATED_BY_ME_MARKETPLACE_NAME}"),
+                "plugin-user".to_string(),
+                None,
+            ),
+            (
+                format!("global@{REMOTE_GLOBAL_MARKETPLACE_NAME}"),
+                "plugin-global".to_string(),
+                None,
+            ),
+            (
+                format!("workspace@{REMOTE_WORKSPACE_MARKETPLACE_NAME}"),
+                "plugin-workspace".to_string(),
+                Some(RemotePluginShareDiscoverability::Listed),
+            ),
+            (
+                format!("shared@{REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME}"),
+                "plugin-shared".to_string(),
+                Some(RemotePluginShareDiscoverability::Private),
+            ),
+        ]
+    );
+}
+
+#[tokio::test]
+async fn search_remote_plugins_requires_chatgpt_authentication() {
+    let (config, selected_urls) =
+        recording_remote_plugin_service_config("https://chatgpt.example/backend-api".to_string());
+
+    let result = search_remote_plugins(
+        &config,
+        /*auth*/ None,
+        RemotePluginSearchRequest {
+            query: "calendar",
+            scope: None,
+            limit: 16,
+            page_token: None,
+        },
+    )
+    .await;
+
+    assert!(matches!(
+        result,
+        Err(RemotePluginCatalogError::AuthRequired)
+    ));
+    assert_eq!(
+        recorded_http_client_urls(&selected_urls),
+        Vec::<String>::new()
+    );
+}
+
+#[tokio::test]
+async fn search_remote_plugins_rejects_api_key_authentication() {
+    let (config, selected_urls) =
+        recording_remote_plugin_service_config("https://chatgpt.example/backend-api".to_string());
+    let auth = CodexAuth::from_api_key("test-api-key");
+
+    let result = search_remote_plugins(
+        &config,
+        Some(&auth),
+        RemotePluginSearchRequest {
+            query: "calendar",
+            scope: Some(RemotePluginScope::Global),
+            limit: 16,
+            page_token: None,
+        },
+    )
+    .await;
+
+    assert!(matches!(
+        result,
+        Err(RemotePluginCatalogError::UnsupportedAuthMode)
+    ));
+    assert_eq!(
+        recorded_http_client_urls(&selected_urls),
+        Vec::<String>::new()
+    );
+}
+
+#[tokio::test]
+async fn search_remote_plugins_redacts_sensitive_parameters_from_transport_errors() {
+    let listener = std::net::TcpListener::bind("127.0.0.1:0")
+        .expect("test listener should bind to a local port");
+    let address = listener
+        .local_addr()
+        .expect("test listener should have a local address");
+    let connection = std::thread::spawn(move || {
+        let (stream, _) = listener
+            .accept()
+            .expect("test listener should accept the plugin search request");
+        drop(stream);
+    });
+    let (config, _) =
+        recording_remote_plugin_service_config(format!("http://{address}/backend-api"));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    let error = search_remote_plugins(
+        &config,
+        Some(&auth),
+        RemotePluginSearchRequest {
+            query: "sensitive search term",
+            scope: Some(RemotePluginScope::Global),
+            limit: 16,
+            page_token: Some("sensitive pagination token"),
+        },
+    )
+    .await
+    .expect_err("closed connection should fail the plugin search request");
+    connection
+        .join()
+        .expect("test listener should close the accepted connection");
+
+    let error_message = error.to_string();
+    assert!(!error_message.contains("sensitive search term"));
+    assert!(!error_message.contains("sensitive pagination token"));
+    let RemotePluginCatalogError::Request { url, source } = error else {
+        panic!("expected transport request error");
+    };
+    assert_eq!(
+        url,
+        format!("http://{address}/backend-api/ps/plugins/search")
+    );
+    assert!(!source.to_string().contains("sensitive"));
+}
+
+#[tokio::test]
+async fn search_remote_plugins_preserves_upstream_http_errors() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/search"))
+        .respond_with(ResponseTemplate::new(503).set_body_string("plugin search unavailable"))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let (config, _) =
+        recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    let result = search_remote_plugins(
+        &config,
+        Some(&auth),
+        RemotePluginSearchRequest {
+            query: "sensitive search term",
+            scope: Some(RemotePluginScope::Global),
+            limit: 16,
+            page_token: Some("sensitive pagination token"),
+        },
+    )
+    .await;
+
+    let error = result.expect_err("upstream HTTP status should fail");
+    let error_message = error.to_string();
+    assert!(!error_message.contains("sensitive search term"));
+    assert!(!error_message.contains("sensitive pagination token"));
+    let RemotePluginCatalogError::UnexpectedStatus { url, status, body } = error else {
+        panic!("expected upstream HTTP status error");
+    };
+    assert_eq!(
+        (url, status, body),
+        (
+            format!("{}/backend-api/ps/plugins/search", server.uri()),
+            StatusCode::SERVICE_UNAVAILABLE,
+            "plugin search unavailable".to_string(),
+        )
+    );
+}
+
+#[tokio::test]
+async fn search_remote_plugins_preserves_response_decode_errors() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/search"))
+        .respond_with(ResponseTemplate::new(200).set_body_string("not-json"))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let (config, _) =
+        recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    let result = search_remote_plugins(
+        &config,
+        Some(&auth),
+        RemotePluginSearchRequest {
+            query: "calendar",
+            scope: None,
+            limit: 16,
+            page_token: None,
+        },
+    )
+    .await;
+
+    assert!(matches!(
+        result,
+        Err(RemotePluginCatalogError::Decode { .. })
+    ));
+}