use super::*;use crate::error_code::internal_error;use crate::error_code::invalid_request;use codex_app_server_protocol::PluginAvailability;use codex_app_server_protocol::PluginInstallPolicy;use codex_app_server_protocol::PluginSharePrincipalRole;use codex_app_server_protocol::PluginShareTargetRole;use codex_config::types::McpServerConfig;use codex_core_plugins::OPENAI_CURATED_MARKETPLACE_NAME;use codex_core_plugins::PluginListBackgroundTaskOptions;use codex_core_plugins::is_openai_curated_marketplace_name;use codex_core_plugins::remote::REMOTE_CREATED_BY_ME_MARKETPLACE_NAME;use codex_core_plugins::remote::REMOTE_GLOBAL_MARKETPLACE_NAME;use codex_core_plugins::remote::REMOTE_WORKSPACE_MARKETPLACE_NAME;use codex_core_plugins::remote::REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME;use codex_core_plugins::remote::REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME;use codex_core_plugins::remote::REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME;use codex_core_plugins::remote::RemoteAppTemplateUnavailableReason;use codex_core_plugins::remote::is_valid_remote_plugin_id;use codex_core_plugins::remote::validate_remote_plugin_id;use codex_mcp::McpOAuthLoginSupport;use codex_mcp::oauth_login_support;use codex_mcp::should_retry_without_scopes;use codex_rmcp_client::perform_oauth_login_silent;#[derive(Clone)]pub(crate) struct PluginRequestProcessor { auth_manager: Arc<AuthManager>, thread_manager: Arc<ThreadManager>, outgoing: Arc<OutgoingMessageSender>, analytics_events_client: AnalyticsEventsClient, config_manager: ConfigManager, workspace_settings_cache: Arc<workspace_settings::WorkspaceSettingsCache>,}fn plugin_skills_to_info( skills: &[codex_core::skills::SkillMetadata], disabled_skill_paths: &HashSet<AbsolutePathBuf>,) -> Vec<SkillSummary> { skills .iter() .map(|skill| SkillSummary { name: skill.name.clone(), description: skill.description.clone(), short_description: skill.short_description.clone(), interface: skill.interface.clone().map(|interface| { codex_app_server_protocol::SkillInterface { display_name: interface.display_name, short_description: interface.short_description, icon_small: interface.icon_small, icon_large: interface.icon_large, brand_color: interface.brand_color, default_prompt: interface.default_prompt, } }), path: Some(skill.path_to_skills_md.clone()), enabled: !disabled_skill_paths.contains(&skill.path_to_skills_md), }) .collect()}fn local_plugin_interface_to_info(interface: PluginManifestInterface) -> PluginInterface { PluginInterface { display_name: interface.display_name, short_description: interface.short_description, long_description: interface.long_description, developer_name: interface.developer_name, category: interface.category, capabilities: interface.capabilities, website_url: interface.website_url, privacy_policy_url: interface.privacy_policy_url, terms_of_service_url: interface.terms_of_service_url, default_prompt: interface.default_prompt, brand_color: interface.brand_color, composer_icon: interface.composer_icon, composer_icon_url: None, logo: interface.logo, logo_url: None, screenshots: interface.screenshots, screenshot_urls: Vec::new(), }}fn marketplace_plugin_source_to_info(source: MarketplacePluginSource) -> PluginSource { match source { MarketplacePluginSource::Local { path } => PluginSource::Local { path }, MarketplacePluginSource::Git { url, path, ref_name, sha, } => PluginSource::Git { url, path, ref_name, sha, }, }}fn load_shared_plugin_ids_by_local_path( config: &Config,) -> Result<std::collections::BTreeMap<AbsolutePathBuf, String>, JSONRPCErrorError> { codex_core_plugins::remote::load_plugin_share_remote_ids_by_local_path( config.codex_home.as_path(), ) .map_err(|err| { internal_error(format!( "failed to load plugin share local path mapping: {err}" )) })}fn share_context_for_source( source: &MarketplacePluginSource, shared_plugin_ids_by_local_path: &std::collections::BTreeMap<AbsolutePathBuf, String>,) -> Option<PluginShareContext> { match source { MarketplacePluginSource::Local { path } => shared_plugin_ids_by_local_path .get(path) .cloned() .map(|remote_plugin_id| PluginShareContext { remote_plugin_id, remote_version: None, discoverability: None, share_url: None, creator_account_user_id: None, creator_name: None, share_principals: None, }), MarketplacePluginSource::Git { .. } => None, }}fn convert_configured_marketplace_plugin_to_plugin_summary( plugin: codex_core_plugins::ConfiguredMarketplacePlugin, shared_plugin_ids_by_local_path: &std::collections::BTreeMap<AbsolutePathBuf, String>,) -> PluginSummary { let share_context = share_context_for_source(&plugin.source, shared_plugin_ids_by_local_path); PluginSummary { id: plugin.id, remote_plugin_id: None, local_version: plugin.local_version, installed: plugin.installed, enabled: plugin.enabled, name: plugin.name, share_context, source: marketplace_plugin_source_to_info(plugin.source), install_policy: plugin.policy.installation.into(), auth_policy: plugin.policy.authentication.into(), availability: PluginAvailability::Available, interface: plugin.interface.map(local_plugin_interface_to_info), keywords: plugin.keywords, }}fn remote_installed_plugin_visible_marketplaces(config: &Config) -> Vec<&'static str> { let mut marketplaces = Vec::new(); if config.features.enabled(Feature::RemotePlugin) { marketplaces.push(REMOTE_GLOBAL_MARKETPLACE_NAME); marketplaces.push(REMOTE_CREATED_BY_ME_MARKETPLACE_NAME); } marketplaces.push(REMOTE_WORKSPACE_MARKETPLACE_NAME); if config.features.enabled(Feature::PluginSharing) { marketplaces.push(REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME); marketplaces.push(REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME); marketplaces.push(REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME); } marketplaces}fn filter_openai_curated_installed_conflicts( marketplaces: &mut Vec<PluginMarketplaceEntry>, prefer_remote_curated_conflicts: bool,) { let local_installed_plugin_names = marketplaces .iter() .filter(|marketplace| is_openai_curated_marketplace_name(&marketplace.name)) .flat_map(|marketplace| installed_plugin_names(&marketplace.plugins)) .collect::<HashSet<_>>(); let remote_installed_plugin_names = marketplaces .iter() .find(|marketplace| marketplace.name == REMOTE_GLOBAL_MARKETPLACE_NAME) .map(|marketplace| installed_plugin_names(&marketplace.plugins)) .unwrap_or_default(); let conflicting_plugin_names = local_installed_plugin_names .intersection(&remote_installed_plugin_names) .cloned() .collect::<HashSet<_>>(); if conflicting_plugin_names.is_empty() { return; } for marketplace in marketplaces.iter_mut() { if prefer_remote_curated_conflicts { if !is_openai_curated_marketplace_name(&marketplace.name) { continue; } } else if marketplace.name != REMOTE_GLOBAL_MARKETPLACE_NAME { continue; } marketplace .plugins .retain(|plugin| !plugin.installed || !conflicting_plugin_names.contains(&plugin.name)); } marketplaces.retain(|marketplace| !marketplace.plugins.is_empty());}fn installed_plugin_names(plugins: &[PluginSummary]) -> HashSet<String> { plugins .iter() .filter(|plugin| plugin.installed) .map(|plugin| plugin.name.clone()) .collect()}fn remote_plugin_share_discoverability( discoverability: PluginShareDiscoverability,) -> codex_core_plugins::remote::RemotePluginShareDiscoverability { match discoverability { PluginShareDiscoverability::Listed => { codex_core_plugins::remote::RemotePluginShareDiscoverability::Listed } PluginShareDiscoverability::Unlisted => { codex_core_plugins::remote::RemotePluginShareDiscoverability::Unlisted } PluginShareDiscoverability::Private => { codex_core_plugins::remote::RemotePluginShareDiscoverability::Private } }}fn remote_plugin_share_update_discoverability( discoverability: PluginShareUpdateDiscoverability,) -> codex_core_plugins::remote::RemotePluginShareUpdateDiscoverability { match discoverability { PluginShareUpdateDiscoverability::Unlisted => { codex_core_plugins::remote::RemotePluginShareUpdateDiscoverability::Unlisted } PluginShareUpdateDiscoverability::Private => { codex_core_plugins::remote::RemotePluginShareUpdateDiscoverability::Private } }}fn validate_client_plugin_share_targets( targets: &[PluginShareTarget],) -> Result<(), JSONRPCErrorError> { if targets .iter() .any(|target| target.principal_type == PluginSharePrincipalType::Workspace) { return Err(invalid_request( "shareTargets cannot include workspace principals; use discoverability UNLISTED for workspace link access", )); } Ok(())}fn remote_plugin_share_target_role( role: PluginShareTargetRole,) -> codex_core_plugins::remote::RemotePluginShareTargetRole { match role { PluginShareTargetRole::Reader => { codex_core_plugins::remote::RemotePluginShareTargetRole::Reader } PluginShareTargetRole::Editor => { codex_core_plugins::remote::RemotePluginShareTargetRole::Editor } }}fn plugin_share_principal_role_from_remote( role: codex_core_plugins::remote::RemotePluginSharePrincipalRole,) -> PluginSharePrincipalRole { match role { codex_core_plugins::remote::RemotePluginSharePrincipalRole::Reader => { PluginSharePrincipalRole::Reader } codex_core_plugins::remote::RemotePluginSharePrincipalRole::Editor => { PluginSharePrincipalRole::Editor } codex_core_plugins::remote::RemotePluginSharePrincipalRole::Owner => { PluginSharePrincipalRole::Owner } }}fn remote_plugin_share_targets( targets: Vec<PluginShareTarget>,) -> Vec<codex_core_plugins::remote::RemotePluginShareTarget> { targets .into_iter() .map( |target| codex_core_plugins::remote::RemotePluginShareTarget { principal_type: match target.principal_type { PluginSharePrincipalType::User => { codex_core_plugins::remote::RemotePluginSharePrincipalType::User } PluginSharePrincipalType::Group => { codex_core_plugins::remote::RemotePluginSharePrincipalType::Group } PluginSharePrincipalType::Workspace => { codex_core_plugins::remote::RemotePluginSharePrincipalType::Workspace } }, principal_id: target.principal_id, role: remote_plugin_share_target_role(target.role), }, ) .collect()}fn plugin_share_principal_from_remote( principal: codex_core_plugins::remote::RemotePluginSharePrincipal,) -> PluginSharePrincipal { PluginSharePrincipal { principal_type: match principal.principal_type { codex_core_plugins::remote::RemotePluginSharePrincipalType::User => { PluginSharePrincipalType::User } codex_core_plugins::remote::RemotePluginSharePrincipalType::Group => { PluginSharePrincipalType::Group } codex_core_plugins::remote::RemotePluginSharePrincipalType::Workspace => { PluginSharePrincipalType::Workspace } }, principal_id: principal.principal_id, role: plugin_share_principal_role_from_remote(principal.role), name: principal.name, }}impl PluginRequestProcessor { pub(crate) fn new( auth_manager: Arc<AuthManager>, thread_manager: Arc<ThreadManager>, outgoing: Arc<OutgoingMessageSender>, analytics_events_client: AnalyticsEventsClient, config_manager: ConfigManager, workspace_settings_cache: Arc<workspace_settings::WorkspaceSettingsCache>, ) -> Self { Self { auth_manager, thread_manager, outgoing, analytics_events_client, config_manager, workspace_settings_cache, } } pub(crate) async fn plugin_list( &self, params: PluginListParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_list_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_installed( &self, params: PluginInstalledParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_installed_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_read( &self, params: PluginReadParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_read_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_skill_read( &self, params: PluginSkillReadParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_skill_read_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_share_save( &self, params: PluginShareSaveParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_share_save_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_share_update_targets( &self, params: PluginShareUpdateTargetsParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_share_update_targets_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_share_list( &self, params: PluginShareListParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_share_list_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_share_checkout( &self, params: PluginShareCheckoutParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_share_checkout_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_share_delete( &self, params: PluginShareDeleteParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_share_delete_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_install( &self, params: PluginInstallParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_install_response(params) .await .map(|response| Some(response.into())) } pub(crate) async fn plugin_uninstall( &self, params: PluginUninstallParams, ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { self.plugin_uninstall_response(params) .await .map(|response| Some(response.into())) } pub(crate) fn effective_plugins_changed_callback(&self) -> Arc<dyn Fn() + Send + Sync> { let thread_manager = Arc::clone(&self.thread_manager); let config_manager = self.config_manager.clone(); Arc::new(move || { Self::spawn_effective_plugins_changed_task( Arc::clone(&thread_manager), config_manager.clone(), ); }) } fn on_effective_plugins_changed(&self) { Self::spawn_effective_plugins_changed_task( Arc::clone(&self.thread_manager), self.config_manager.clone(), ); } fn spawn_effective_plugins_changed_task( thread_manager: Arc<ThreadManager>, config_manager: ConfigManager, ) { tokio::spawn(async move { thread_manager.plugins_manager().clear_cache(); thread_manager.skills_manager().clear_cache(); if thread_manager.list_thread_ids().await.is_empty() { return; } crate::mcp_refresh::queue_best_effort_refresh(&thread_manager, &config_manager).await; }); } fn clear_plugin_related_caches(&self) { self.thread_manager.plugins_manager().clear_cache(); self.thread_manager.skills_manager().clear_cache(); } async fn load_latest_config( &self, fallback_cwd: Option<PathBuf>, ) -> Result<Config, JSONRPCErrorError> { self.config_manager .load_latest_config(fallback_cwd) .await .map_err(|err| internal_error(format!("failed to reload config: {err}"))) } async fn workspace_codex_plugins_enabled( &self, config: &Config, auth: Option<&CodexAuth>, ) -> bool { match workspace_settings::codex_plugins_enabled_for_workspace( config, auth, Some(&self.workspace_settings_cache), ) .await { Ok(enabled) => enabled, Err(err) => { warn!( "failed to fetch workspace Codex plugins setting; allowing Codex plugins: {err:#}" ); true } } } async fn plugin_list_response( &self, params: PluginListParams, ) -> Result<PluginListResponse, JSONRPCErrorError> { let plugins_manager = self.thread_manager.plugins_manager(); let PluginListParams { cwds, marketplace_kinds, } = params; let roots = cwds.unwrap_or_default(); let explicit_marketplace_kinds = marketplace_kinds.is_some(); let marketplace_kinds = marketplace_kinds.unwrap_or_else(|| vec![PluginListMarketplaceKind::Local]); let include_local = marketplace_kinds.contains(&PluginListMarketplaceKind::Local); let include_vertical = marketplace_kinds.contains(&PluginListMarketplaceKind::Vertical); let config = self.load_latest_config(/*fallback_cwd*/ None).await?; let empty_response = || PluginListResponse { marketplaces: Vec::new(), marketplace_load_errors: Vec::new(), featured_plugin_ids: Vec::new(), }; if !config.features.enabled(Feature::Plugins) { return Ok(empty_response()); } let auth = self.auth_manager.auth().await; if !self .workspace_codex_plugins_enabled(&config, auth.as_ref()) .await { return Ok(empty_response()); } let auth_mode = auth.as_ref().map(CodexAuth::api_auth_mode); plugins_manager.set_auth_mode(auth_mode); let plugins_input = config.plugins_config_input(); let include_shared_with_me = marketplace_kinds.contains(&PluginListMarketplaceKind::SharedWithMe); let include_created_by_me_remote = marketplace_kinds .contains(&PluginListMarketplaceKind::CreatedByMeRemote) && config.features.enabled(Feature::RemotePlugin); let include_global_remote = !explicit_marketplace_kinds && config.features.enabled(Feature::RemotePlugin); let use_remote_global_catalog = include_global_remote && auth_mode.is_some_and(AuthMode::uses_codex_backend); let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; let refresh_global_remote_catalog_cache = use_remote_global_catalog && codex_core_plugins::remote::has_cached_global_remote_plugin_catalog( config.codex_home.as_path(), &remote_plugin_service_config, auth.as_ref(), ); let (mut data, marketplace_load_errors) = if include_local { let config_for_marketplace_listing = plugins_input.clone(); let plugins_manager_for_marketplace_listing = plugins_manager.clone(); let roots_for_marketplace_listing = roots.clone(); let shared_plugin_ids_by_local_path = load_shared_plugin_ids_by_local_path(&config)?; match tokio::task::spawn_blocking(move || { let outcome = plugins_manager_for_marketplace_listing .list_marketplaces_for_config( &config_for_marketplace_listing, &roots_for_marketplace_listing, /*include_openai_curated*/ !use_remote_global_catalog, )?; Ok::< ( Vec<PluginMarketplaceEntry>, Vec<codex_app_server_protocol::MarketplaceLoadErrorInfo>, ), MarketplaceError, >(( outcome .marketplaces .into_iter() .map(|marketplace| PluginMarketplaceEntry { name: marketplace.name, path: Some(marketplace.path), interface: marketplace.interface.map(|interface| { MarketplaceInterface { display_name: interface.display_name, } }), plugins: marketplace .plugins .into_iter() .map(|plugin| { convert_configured_marketplace_plugin_to_plugin_summary( plugin, &shared_plugin_ids_by_local_path, ) }) .collect(), }) .collect(), outcome .errors .into_iter() .map(|err| codex_app_server_protocol::MarketplaceLoadErrorInfo { marketplace_path: err.path, message: err.message, }) .collect(), )) }) .await { Ok(Ok(outcome)) => outcome, Ok(Err(err)) => { return Err(Self::marketplace_error(err, "list marketplace plugins")); } Err(err) => { return Err(internal_error(format!( "failed to list marketplace plugins: {err}" ))); } } } else { (Vec::new(), Vec::new()) }; // TODO(remote plugins): Remove this once remote plugins are ready and vertical plugins are // served directly from the normal remote catalog. if include_vertical && !config.features.enabled(Feature::RemotePlugin) { match codex_core_plugins::remote::fetch_openai_curated_remote_collection_marketplace( &remote_plugin_service_config, auth.as_ref(), ) .await { Ok(Some(remote_marketplace)) => { data.push(remote_marketplace_to_info(remote_marketplace)); } Ok(None) => {} Err(RemotePluginCatalogError::UnsupportedAuthMode) => {} Err(err) if explicit_marketplace_kinds => { return Err(remote_plugin_catalog_error_to_jsonrpc( err, "list OpenAI Curated remote plugin catalog", )); } Err(RemotePluginCatalogError::AuthRequired) => {} Err(err) => { warn!( error = %err, "plugin/list openai-curated-remote collection fetch failed; returning local marketplaces only" ); } } } let mut remote_sources = Vec::new(); if use_remote_global_catalog { remote_sources.push(RemoteMarketplaceSource::Global); } if include_created_by_me_remote { remote_sources.push(RemoteMarketplaceSource::CreatedByMeRemote); } if marketplace_kinds.contains(&PluginListMarketplaceKind::WorkspaceDirectory) { remote_sources.push(RemoteMarketplaceSource::WorkspaceDirectory); } if include_shared_with_me && config.features.enabled(Feature::PluginSharing) { remote_sources.push(RemoteMarketplaceSource::SharedWithMe); } if !remote_sources.is_empty() { match codex_core_plugins::remote::fetch_remote_marketplaces( &remote_plugin_service_config, auth.as_ref(), &remote_sources, /*global_catalog_cache_path*/ Some(config.codex_home.as_path()), ) .await { Ok(remote_marketplaces) => { for remote_marketplace in remote_marketplaces .into_iter() .map(remote_marketplace_to_info) { data.push(remote_marketplace); } } Err( err @ (RemotePluginCatalogError::AuthRequired | RemotePluginCatalogError::UnsupportedAuthMode), ) if explicit_marketplace_kinds => { return Err(remote_plugin_catalog_error_to_jsonrpc( err, "list remote plugin catalog", )); } Err( RemotePluginCatalogError::AuthRequired | RemotePluginCatalogError::UnsupportedAuthMode, ) => {} Err(err) if explicit_marketplace_kinds => { return Err(remote_plugin_catalog_error_to_jsonrpc( err, "list remote plugin catalog", )); } Err(err) => { warn!( error = %err, "plugin/list remote plugin catalog fetch failed; returning local marketplaces only" ); } } } if include_local || include_created_by_me_remote || include_shared_with_me || include_global_remote { plugins_manager.maybe_start_plugin_list_background_tasks_for_config( &plugins_input, auth.clone(), &roots, PluginListBackgroundTaskOptions { refresh_global_remote_catalog_cache, }, Some(self.effective_plugins_changed_callback()), ); } let featured_plugin_ids = if !plugins_input.remote_plugin_enabled && data .iter() .any(|marketplace| marketplace.name == OPENAI_CURATED_MARKETPLACE_NAME) { match plugins_manager .featured_plugin_ids_for_config(&plugins_input, auth.as_ref()) .await { Ok(featured_plugin_ids) => featured_plugin_ids, Err(err) => { warn!( error = %err, "plugin/list featured plugin fetch failed; returning empty featured ids" ); Vec::new() } } } else { Vec::new() }; Ok(PluginListResponse { marketplaces: data, marketplace_load_errors, featured_plugin_ids, }) } async fn plugin_installed_response( &self, params: PluginInstalledParams, ) -> Result<PluginInstalledResponse, JSONRPCErrorError> { let plugins_manager = self.thread_manager.plugins_manager(); let PluginInstalledParams { cwds, install_suggestion_plugin_names, } = params; let roots = cwds.unwrap_or_default(); let install_suggestion_plugin_names = install_suggestion_plugin_names .unwrap_or_default() .into_iter() .collect::<HashSet<_>>(); let empty_response = || PluginInstalledResponse { marketplaces: Vec::new(), marketplace_load_errors: Vec::new(), }; let config = self.load_latest_config(/*fallback_cwd*/ None).await?; if !config.features.enabled(Feature::Plugins) { return Ok(empty_response()); } let auth = self.auth_manager.auth().await; if !self .workspace_codex_plugins_enabled(&config, auth.as_ref()) .await { return Ok(empty_response()); } plugins_manager.set_auth_mode(auth.as_ref().map(CodexAuth::api_auth_mode)); let plugins_input = config.plugins_config_input(); let remote_installed_plugin_visible_marketplaces = remote_installed_plugin_visible_marketplaces(&config); plugins_manager.maybe_start_remote_installed_plugin_bundle_sync( &plugins_input, auth.clone(), Some(self.effective_plugins_changed_callback()), ); let (mut data, marketplace_load_errors) = self .load_local_installed_and_suggested_plugins( plugins_manager.clone(), &config, &plugins_input, roots, install_suggestion_plugin_names, ) .await?; data.extend( self.load_remote_installed_plugins( plugins_manager, &plugins_input, &remote_installed_plugin_visible_marketplaces, auth.as_ref(), ) .await, ); filter_openai_curated_installed_conflicts( &mut data, config.features.enabled(Feature::RemotePlugin), ); Ok(PluginInstalledResponse { marketplaces: data, marketplace_load_errors, }) } async fn load_local_installed_and_suggested_plugins( &self, plugins_manager: Arc<codex_core_plugins::PluginsManager>, config: &Config, plugins_input: &codex_core_plugins::PluginsConfigInput, roots: Vec<AbsolutePathBuf>, install_suggestion_plugin_names: HashSet<String>, ) -> Result< ( Vec<PluginMarketplaceEntry>, Vec<codex_app_server_protocol::MarketplaceLoadErrorInfo>, ), JSONRPCErrorError, > { let config_for_marketplace_listing = plugins_input.clone(); let shared_plugin_ids_by_local_path = load_shared_plugin_ids_by_local_path(config)?; match tokio::task::spawn_blocking(move || { let outcome = plugins_manager.list_marketplaces_for_config( &config_for_marketplace_listing, &roots, /*include_openai_curated*/ true, )?; Ok::< ( Vec<PluginMarketplaceEntry>, Vec<codex_app_server_protocol::MarketplaceLoadErrorInfo>, ), MarketplaceError, >(( outcome .marketplaces .into_iter() .filter_map(|marketplace| { let plugins = marketplace .plugins .into_iter() .filter(|plugin| { plugin.installed || install_suggestion_plugin_names.contains(&plugin.name) }) .map(|plugin| { convert_configured_marketplace_plugin_to_plugin_summary( plugin, &shared_plugin_ids_by_local_path, ) }) .collect::<Vec<_>>(); (!plugins.is_empty()).then_some(PluginMarketplaceEntry { name: marketplace.name, path: Some(marketplace.path), interface: marketplace.interface.map(|interface| { MarketplaceInterface { display_name: interface.display_name, } }), plugins, }) }) .collect(), outcome .errors .into_iter() .map(|err| codex_app_server_protocol::MarketplaceLoadErrorInfo { marketplace_path: err.path, message: err.message, }) .collect(), )) }) .await { Ok(Ok(outcome)) => Ok(outcome), Ok(Err(err)) => Err(Self::marketplace_error( err, "list installed and suggested marketplace plugins", )), Err(err) => Err(internal_error(format!( "failed to list installed and suggested plugins: {err}" ))), } } async fn load_remote_installed_plugins( &self, plugins_manager: Arc<codex_core_plugins::PluginsManager>, plugins_input: &codex_core_plugins::PluginsConfigInput, visible_marketplaces: &[&str], auth: Option<&CodexAuth>, ) -> Vec<PluginMarketplaceEntry> { let remote_marketplaces = if let Some(remote_marketplaces) = plugins_manager .build_remote_installed_plugin_marketplaces_from_cache(visible_marketplaces) { Ok(remote_marketplaces) } else { plugins_manager .build_and_cache_remote_installed_plugin_marketplaces( plugins_input, auth, visible_marketplaces, Some(self.effective_plugins_changed_callback()), ) .await }; match remote_marketplaces { Ok(remote_marketplaces) => remote_marketplaces .into_iter() .map(remote_marketplace_to_info) .collect(), Err( RemotePluginCatalogError::AuthRequired | RemotePluginCatalogError::UnsupportedAuthMode, ) => Vec::new(), Err(err) => { warn!( error = %err, "plugin/installed remote installed plugin fetch failed; returning local marketplaces only" ); Vec::new() } } } async fn plugin_read_response( &self, params: PluginReadParams, ) -> Result<PluginReadResponse, JSONRPCErrorError> { let plugins_manager = self.thread_manager.plugins_manager(); let PluginReadParams { marketplace_path, remote_marketplace_name, plugin_name, } = params; let read_source = match (marketplace_path, remote_marketplace_name) { (Some(marketplace_path), None) => Ok(marketplace_path), (None, Some(remote_marketplace_name)) => Err(remote_marketplace_name), (Some(_), Some(_)) | (None, None) => { return Err(invalid_request( "plugin/read requires exactly one of marketplacePath or remoteMarketplaceName", )); } }; let config_cwd = read_source.as_ref().ok().and_then(|marketplace_path| { marketplace_path.as_path().parent().map(Path::to_path_buf) }); let config = self.load_latest_config(config_cwd).await?; let plugins_input = config.plugins_config_input(); let auth = self.auth_manager.auth().await; plugins_manager.set_auth_mode(auth.as_ref().map(CodexAuth::api_auth_mode)); let plugin = match read_source { Ok(marketplace_path) => { let request = PluginReadRequest { plugin_name, marketplace_path, }; let outcome = plugins_manager .read_plugin_for_config(&plugins_input, &request) .await .map_err(|err| Self::marketplace_error(err, "read plugin details"))?; let shared_plugin_ids_by_local_path = load_shared_plugin_ids_by_local_path(&config)?; let share_context = share_context_for_source( &outcome.plugin.source, &shared_plugin_ids_by_local_path, ); let share_context = match share_context { Some(context) => { let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; match codex_core_plugins::remote::fetch_remote_plugin_share_context( &remote_plugin_service_config, auth.as_ref(), &context.remote_plugin_id, ) .await { Ok(Some(remote_share_context)) => { if remote_share_context.share_principals.is_some() { Some(remote_plugin_share_context_to_info(remote_share_context)) } else { let remote_version = remote_share_context.remote_version; let remote_plugin_id = context.remote_plugin_id.clone(); warn!( remote_plugin_id = %remote_plugin_id, "remote shared plugin detail did not include share principals; returning local share mapping context with remote version" ); Some(PluginShareContext { remote_version, ..context }) } } Ok(None) => { warn!( remote_plugin_id = %context.remote_plugin_id, "remote shared plugin detail did not include share context; returning local share mapping context" ); Some(context) } Err(err) => { warn!( remote_plugin_id = %context.remote_plugin_id, error = %err, "failed to hydrate local plugin share context; returning local share mapping context" ); Some(context) } } } None => None, }; let app_summaries = load_plugin_app_summaries( &config, &outcome.plugin.apps, &outcome.plugin.app_category_by_id, ) .await; let visible_skills = outcome .plugin .skills .iter() .filter(|skill| { skill.matches_product_restriction_for_product( self.thread_manager.session_source().restriction_product(), ) }) .cloned() .collect::<Vec<_>>(); PluginDetail { marketplace_name: outcome.marketplace_name, marketplace_path: outcome.marketplace_path, summary: PluginSummary { id: outcome.plugin.id, remote_plugin_id: None, local_version: outcome.plugin.local_version, name: outcome.plugin.name, share_context, source: marketplace_plugin_source_to_info(outcome.plugin.source), installed: outcome.plugin.installed, enabled: outcome.plugin.enabled, install_policy: outcome.plugin.policy.installation.into(), auth_policy: outcome.plugin.policy.authentication.into(), availability: PluginAvailability::Available, interface: outcome.plugin.interface.map(local_plugin_interface_to_info), keywords: outcome.plugin.keywords, }, share_url: None, description: outcome.plugin.description, skills: plugin_skills_to_info( &visible_skills, &outcome.plugin.disabled_skill_paths, ), hooks: outcome .plugin .hooks .into_iter() .map(|hook| codex_app_server_protocol::PluginHookSummary { key: hook.key, event_name: hook.event_name.into(), }) .collect(), apps: app_summaries, app_templates: Vec::new(), mcp_servers: outcome.plugin.mcp_server_names, } } Err(remote_marketplace_name) => { if !config.features.enabled(Feature::Plugins) { return Err(invalid_request(format!( "remote plugin read is not enabled for marketplace {remote_marketplace_name}" ))); } let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; validate_remote_plugin_id(&plugin_name)?; let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail( &remote_plugin_service_config, auth.as_ref(), &remote_marketplace_name, &plugin_name, ) .await .map_err(|err| { remote_plugin_catalog_error_to_jsonrpc(err, "read remote plugin details") })?; let plugin_apps = remote_detail .app_ids .iter() .cloned() .map(codex_plugin::AppConnectorId) .collect::<Vec<_>>(); let app_category_by_id = remote_detail .app_manifest .as_ref() .map(plugin_app_category_by_id_from_value) .unwrap_or_default(); let app_summaries = load_plugin_app_summaries(&config, &plugin_apps, &app_category_by_id).await; remote_plugin_detail_to_info(remote_detail, app_summaries) } }; Ok(PluginReadResponse { plugin }) } async fn plugin_skill_read_response( &self, params: PluginSkillReadParams, ) -> Result<PluginSkillReadResponse, JSONRPCErrorError> { let PluginSkillReadParams { remote_marketplace_name, remote_plugin_id, skill_name, } = params; let config = self.load_latest_config(/*fallback_cwd*/ None).await?; if !config.features.enabled(Feature::Plugins) { return Err(invalid_request(format!( "remote plugin skill read is not enabled for marketplace {remote_marketplace_name}" ))); } validate_remote_plugin_id(&remote_plugin_id)?; if skill_name.is_empty() { return Err(invalid_request( "invalid remote plugin skill name: cannot be empty", )); } let auth = self.auth_manager.auth().await; let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; let remote_skill_detail = codex_core_plugins::remote::fetch_remote_plugin_skill_detail( &remote_plugin_service_config, auth.as_ref(), &remote_marketplace_name, &remote_plugin_id, &skill_name, ) .await .map_err(|err| { remote_plugin_catalog_error_to_jsonrpc(err, "read remote plugin skill details") })?; Ok(PluginSkillReadResponse { contents: remote_skill_detail.contents, }) } async fn plugin_share_save_response( &self, params: PluginShareSaveParams, ) -> Result<PluginShareSaveResponse, JSONRPCErrorError> { let (config, auth) = self.load_plugin_share_config_and_auth().await?; if !config.features.enabled(Feature::PluginSharing) { return Err(invalid_request("plugin sharing is disabled")); } let PluginShareSaveParams { plugin_path, remote_plugin_id, discoverability, share_targets, } = params; if let Some(remote_plugin_id) = remote_plugin_id.as_ref() && (remote_plugin_id.is_empty() || !is_valid_remote_plugin_id(remote_plugin_id)) { return Err(invalid_request("invalid remote plugin id")); } if remote_plugin_id.is_some() && (discoverability.is_some() || share_targets.is_some()) { return Err(invalid_request( "discoverability and shareTargets are only supported when creating a plugin share; use plugin/share/updateTargets to update share settings", )); } if discoverability == Some(PluginShareDiscoverability::Listed) { return Err(invalid_request( "discoverability LISTED is not supported for plugin/share/save; use UNLISTED or PRIVATE", )); } if let Some(share_targets) = share_targets.as_ref() { validate_client_plugin_share_targets(share_targets)?; } let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; let access_policy = codex_core_plugins::remote::RemotePluginShareAccessPolicy { discoverability: discoverability.map(remote_plugin_share_discoverability), share_targets: share_targets.map(remote_plugin_share_targets), }; let result = codex_core_plugins::remote::save_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), config.codex_home.as_path(), &plugin_path, remote_plugin_id.as_deref(), access_policy, ) .await .map_err(|err| remote_plugin_catalog_error_to_jsonrpc(err, "save remote plugin share"))?; let remote_plugin_id = result.remote_plugin_id; self.clear_plugin_related_caches(); Ok(PluginShareSaveResponse { remote_plugin_id, share_url: result.share_url.unwrap_or_default(), }) } async fn plugin_share_update_targets_response( &self, params: PluginShareUpdateTargetsParams, ) -> Result<PluginShareUpdateTargetsResponse, JSONRPCErrorError> { let (config, auth) = self.load_plugin_share_config_and_auth().await?; if !config.features.enabled(Feature::PluginSharing) { return Err(invalid_request("plugin sharing is disabled")); } let PluginShareUpdateTargetsParams { remote_plugin_id, discoverability, share_targets, } = params; if remote_plugin_id.is_empty() || !is_valid_remote_plugin_id(&remote_plugin_id) { return Err(invalid_request("invalid remote plugin id")); } validate_client_plugin_share_targets(&share_targets)?; let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; let result = codex_core_plugins::remote::update_remote_plugin_share_targets( &remote_plugin_service_config, auth.as_ref(), &remote_plugin_id, remote_plugin_share_targets(share_targets), remote_plugin_share_update_discoverability(discoverability), ) .await .map_err(|err| { remote_plugin_catalog_error_to_jsonrpc(err, "update remote plugin share targets") })?; self.clear_plugin_related_caches(); Ok(PluginShareUpdateTargetsResponse { principals: result .principals .into_iter() .map(plugin_share_principal_from_remote) .collect(), discoverability: remote_plugin_share_discoverability_to_info(result.discoverability), }) } async fn plugin_share_list_response( &self, _params: PluginShareListParams, ) -> Result<PluginShareListResponse, JSONRPCErrorError> { let (config, auth) = self.load_plugin_share_config_and_auth().await?; let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; let data = codex_core_plugins::remote::list_remote_plugin_shares( &remote_plugin_service_config, auth.as_ref(), config.codex_home.as_path(), ) .await .map_err(|err| remote_plugin_catalog_error_to_jsonrpc(err, "list remote plugin shares"))? .into_iter() .map(|summary| { let RemoteCatalogPluginShareSummary { summary, local_plugin_path, } = summary; let plugin = remote_plugin_summary_to_info(summary); PluginShareListItem { plugin, local_plugin_path, } }) .collect(); Ok(PluginShareListResponse { data }) } async fn plugin_share_checkout_response( &self, params: PluginShareCheckoutParams, ) -> Result<PluginShareCheckoutResponse, JSONRPCErrorError> { let (config, auth) = self.load_plugin_share_config_and_auth().await?; if !config.features.enabled(Feature::PluginSharing) { return Err(invalid_request("plugin sharing is disabled")); } let PluginShareCheckoutParams { remote_plugin_id } = params; if remote_plugin_id.is_empty() || !is_valid_remote_plugin_id(&remote_plugin_id) { return Err(invalid_request("invalid remote plugin id")); } let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; let result = codex_core_plugins::remote::checkout_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), config.codex_home.as_path(), &remote_plugin_id, ) .await .map_err(|err| remote_plugin_catalog_error_to_jsonrpc(err, "checkout plugin share"))?; self.clear_plugin_related_caches(); Ok(PluginShareCheckoutResponse { remote_plugin_id: result.remote_plugin_id, plugin_id: result.plugin_id, plugin_name: result.plugin_name, plugin_path: result.plugin_path, marketplace_name: result.marketplace_name, marketplace_path: result.marketplace_path, remote_version: result.remote_version, }) } async fn plugin_share_delete_response( &self, params: PluginShareDeleteParams, ) -> Result<PluginShareDeleteResponse, JSONRPCErrorError> { let (config, auth) = self.load_plugin_share_config_and_auth().await?; let PluginShareDeleteParams { remote_plugin_id } = params; if remote_plugin_id.is_empty() || !is_valid_remote_plugin_id(&remote_plugin_id) { return Err(invalid_request("invalid remote plugin id")); } let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; codex_core_plugins::remote::delete_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), config.codex_home.as_path(), &remote_plugin_id, ) .await .map_err(|err| remote_plugin_catalog_error_to_jsonrpc(err, "delete remote plugin share"))?; self.clear_plugin_related_caches(); Ok(PluginShareDeleteResponse {}) } async fn load_plugin_share_config_and_auth( &self, ) -> Result<(Config, Option<CodexAuth>), JSONRPCErrorError> { let config = self.load_latest_config(/*fallback_cwd*/ None).await?; if !config.features.enabled(Feature::Plugins) { return Err(invalid_request("plugin sharing is not enabled")); } let auth = self.auth_manager.auth().await; Ok((config, auth)) } async fn plugin_install_response( &self, params: PluginInstallParams, ) -> Result<PluginInstallResponse, JSONRPCErrorError> { let PluginInstallParams { marketplace_path, remote_marketplace_name, plugin_name, } = params; let marketplace_path = match (marketplace_path, remote_marketplace_name) { (Some(marketplace_path), None) => marketplace_path, (None, Some(remote_marketplace_name)) => { return self .remote_plugin_install_response(remote_marketplace_name, plugin_name) .await; } (Some(_), Some(_)) | (None, None) => { return Err(invalid_request( "plugin/install requires exactly one of marketplacePath or remoteMarketplaceName", )); } }; let config_cwd = marketplace_path.as_path().parent().map(Path::to_path_buf); let config = self.load_latest_config(config_cwd.clone()).await?; let auth = self.auth_manager.auth().await; if !self .workspace_codex_plugins_enabled(&config, auth.as_ref()) .await { return Err(invalid_request( "Codex plugins are disabled for this workspace", )); } let plugins_manager = self.thread_manager.plugins_manager(); let request = PluginInstallRequest { plugin_name, marketplace_path, }; let result = plugins_manager .install_plugin(request) .await .map_err(Self::plugin_install_error)?; let config = match self.load_latest_config(config_cwd).await { Ok(config) => config, Err(err) => { warn!( "failed to reload config after plugin install, using current config: {err:?}" ); config } }; self.on_effective_plugins_changed(); let plugin_mcp_servers = load_plugin_mcp_servers( result.installed_path.as_path(), auth.as_ref().map(CodexAuth::auth_mode), ) .await; if !plugin_mcp_servers.is_empty() { self.start_plugin_mcp_oauth_logins(&config, plugin_mcp_servers) .await; } let plugin_app_declarations = load_plugin_apps(result.installed_path.as_path()).await; let plugin_apps = codex_plugin::app_connector_ids_from_declarations(&plugin_app_declarations); let apps_needing_auth = self .plugin_apps_needing_auth_for_install( &config, auth.as_ref().is_some_and(CodexAuth::is_chatgpt_auth), &result.plugin_id.as_key(), &plugin_apps, ) .await; Ok(PluginInstallResponse { auth_policy: result.auth_policy.into(), apps_needing_auth, }) } async fn remote_plugin_install_response( &self, remote_marketplace_name: String, remote_plugin_id: String, ) -> Result<PluginInstallResponse, JSONRPCErrorError> { let config = self.load_latest_config(/*fallback_cwd*/ None).await?; if !config.features.enabled(Feature::Plugins) { return Err(invalid_request(format!( "remote plugin install is not enabled for marketplace {remote_marketplace_name}" ))); } validate_remote_plugin_id(&remote_plugin_id)?; let auth = self.auth_manager.auth().await; let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail_with_download_urls( &remote_plugin_service_config, auth.as_ref(), &remote_marketplace_name, &remote_plugin_id, ) .await .map_err(|err| { remote_plugin_catalog_error_to_jsonrpc( err, "read remote plugin details before install", ) })?; if remote_detail.summary.availability == PluginAvailability::DisabledByAdmin { return Err(invalid_request(format!( "remote plugin {remote_plugin_id} is disabled by admin" ))); } if remote_detail.summary.install_policy == PluginInstallPolicy::NotAvailable { return Err(invalid_request(format!( "remote plugin {remote_plugin_id} is not available for install" ))); } let actual_remote_marketplace_name = remote_detail.marketplace_name.clone(); // Direct install writes the same cache tree that installed-plugin sync // prunes before the backend installed snapshot can include this plugin. let _remote_plugin_cache_mutation = codex_core_plugins::remote::mark_remote_plugin_cache_mutation_in_flight( config.codex_home.as_path(), &actual_remote_marketplace_name, &remote_detail.summary.name, ); let validated_bundle = codex_core_plugins::remote_bundle::validate_remote_plugin_bundle( &remote_plugin_id, &actual_remote_marketplace_name, &remote_detail.summary.name, remote_detail.release_version.as_deref(), remote_detail.bundle_download_url.as_deref(), remote_detail.app_manifest.clone(), ) .map_err(remote_plugin_bundle_install_error_to_jsonrpc)?; let result = codex_core_plugins::remote_bundle::download_and_install_remote_plugin_bundle( config.codex_home.to_path_buf(), validated_bundle, ) .await .map_err(remote_plugin_bundle_install_error_to_jsonrpc)?; // Cache first so a backend install cannot succeed when local materialization fails. // If this backend call fails, the cache entry is harmless because remote installed state // is still backend-gated. let install_result = codex_core_plugins::remote::install_remote_plugin( &remote_plugin_service_config, auth.as_ref(), &actual_remote_marketplace_name, &remote_plugin_id, ) .await .map_err(|err| remote_plugin_catalog_error_to_jsonrpc(err, "install remote plugin"))?; self.thread_manager .plugins_manager() .maybe_start_remote_installed_plugins_cache_refresh_after_mutation( &config.plugins_config_input(), auth.clone(), Some(self.effective_plugins_changed_callback()), ); let mut plugin_metadata = plugin_telemetry_metadata_from_root(&result.plugin_id, &result.installed_path).await; plugin_metadata.remote_plugin_id = Some(remote_plugin_id.clone()); self.analytics_events_client .track_plugin_installed(plugin_metadata); let plugin_mcp_servers = load_plugin_mcp_servers( result.installed_path.as_path(), auth.as_ref().map(CodexAuth::auth_mode), ) .await; if !plugin_mcp_servers.is_empty() { self.start_plugin_mcp_oauth_logins(&config, plugin_mcp_servers) .await; } let is_chatgpt_auth = auth.as_ref().is_some_and(CodexAuth::is_chatgpt_auth); let apps_needing_auth = if let Some(app_ids_needing_auth) = install_result.app_ids_needing_auth { if app_ids_needing_auth.is_empty() || !config.features.apps_enabled_for_auth(is_chatgpt_auth) { Vec::new() } else { let plugin_apps = app_ids_needing_auth .into_iter() .map(codex_plugin::AppConnectorId) .collect::<Vec<_>>(); let app_category_by_id = remote_detail .app_manifest .as_ref() .map(plugin_app_category_by_id_from_value) .unwrap_or_default(); let all_connectors = connectors::list_cached_all_connectors(&config, &[]) .await .unwrap_or_default(); connectors::connectors_for_plugin_apps(all_connectors, &plugin_apps) .into_iter() .map(|connector| { let category = app_category_by_id .get(&connector.id) .cloned() .or_else(|| connector.category()); AppSummary { category, id: connector.id, name: connector.name, description: connector.description, install_url: connector.install_url, } }) .collect() } } else { let plugin_app_declarations = load_plugin_apps(result.installed_path.as_path()).await; let plugin_apps = codex_plugin::app_connector_ids_from_declarations(&plugin_app_declarations); self.plugin_apps_needing_auth_for_install( &config, is_chatgpt_auth, &result.plugin_id.as_key(), &plugin_apps, ) .await }; Ok(PluginInstallResponse { auth_policy: remote_detail.summary.auth_policy, apps_needing_auth, }) } async fn plugin_apps_needing_auth_for_install( &self, config: &Config, is_chatgpt_auth: bool, plugin_id: &str, plugin_apps: &[codex_plugin::AppConnectorId], ) -> Vec<AppSummary> { if plugin_apps.is_empty() || !config.features.apps_enabled_for_auth(is_chatgpt_auth) { return Vec::new(); } let environment_manager = self.thread_manager.environment_manager(); let (all_connectors_result, accessible_connectors_result) = tokio::join!( connectors::list_all_connectors_with_options(config, /*force_refetch*/ false, &[]), connectors::list_accessible_connectors_from_mcp_tools_with_mcp_manager( config, /*force_refetch*/ true, Arc::clone(&environment_manager), self.thread_manager.mcp_manager(), ), ); let all_connectors = match all_connectors_result { Ok(connectors) => connectors, Err(err) => { warn!( plugin = plugin_id, "failed to load app metadata after plugin install: {err:#}" ); connectors::list_cached_all_connectors(config, &[]) .await .unwrap_or_default() } }; let all_connectors = connectors::connectors_for_plugin_apps(all_connectors, plugin_apps); let (accessible_connectors, codex_apps_ready) = match accessible_connectors_result { Ok(status) => (status.connectors, status.codex_apps_ready), Err(err) => { warn!( plugin = plugin_id, "failed to load accessible apps after plugin install: {err:#}" ); ( connectors::list_cached_accessible_connectors_from_mcp_tools(config) .await .unwrap_or_default(), false, ) } }; if !codex_apps_ready { warn!( plugin = plugin_id, "codex_apps MCP not ready after plugin install; skipping appsNeedingAuth check" ); } plugin_apps_needing_auth( &all_connectors, &accessible_connectors, plugin_apps, codex_apps_ready, ) } async fn start_plugin_mcp_oauth_logins( &self, config: &Config, plugin_mcp_servers: HashMap<String, McpServerConfig>, ) { for (name, server) in plugin_mcp_servers { let oauth_config = match oauth_login_support(&server.transport).await { McpOAuthLoginSupport::Supported(config) => config, McpOAuthLoginSupport::Unsupported => continue, McpOAuthLoginSupport::Unknown(err) => { warn!( "MCP server may or may not require login for plugin install {name}: {err}" ); continue; } }; let resolved_scopes = resolve_oauth_scopes( /*explicit_scopes*/ None, server.scopes.clone(), oauth_config.discovered_scopes.clone(), ); let store_mode = config.mcp_oauth_credentials_store_mode; let keyring_backend_kind = config.auth_keyring_backend_kind(); let callback_port = config.mcp_oauth_callback_port; let callback_url = config.mcp_oauth_callback_url.clone(); let outgoing = Arc::clone(&self.outgoing); let notification_name = name.clone(); tokio::spawn(async move { let oauth_client_id = server.oauth_client_id(); let first_attempt = perform_oauth_login_silent( &name, &oauth_config.url, store_mode, keyring_backend_kind, oauth_config.http_headers.clone(), oauth_config.env_http_headers.clone(), &resolved_scopes.scopes, oauth_client_id, server.oauth_resource.as_deref(), callback_port, callback_url.as_deref(), ) .await; let final_result = match first_attempt { Err(err) if should_retry_without_scopes(&resolved_scopes, &err) => { perform_oauth_login_silent( &name, &oauth_config.url, store_mode, keyring_backend_kind, oauth_config.http_headers, oauth_config.env_http_headers, &[], oauth_client_id, server.oauth_resource.as_deref(), callback_port, callback_url.as_deref(), ) .await } result => result, }; let (success, error) = match final_result { Ok(()) => (true, None), Err(err) => (false, Some(err.to_string())), }; let notification = ServerNotification::McpServerOauthLoginCompleted( McpServerOauthLoginCompletedNotification { name: notification_name, success, error, }, ); outgoing.send_server_notification(notification).await; }); } } async fn plugin_uninstall_response( &self, params: PluginUninstallParams, ) -> Result<PluginUninstallResponse, JSONRPCErrorError> { let PluginUninstallParams { plugin_id } = params; if codex_plugin::PluginId::parse(&plugin_id).is_err() && !is_valid_remote_plugin_id(&plugin_id) { return Err(invalid_request("invalid remote plugin id")); } if is_valid_remote_plugin_id(&plugin_id) { return self.remote_plugin_uninstall_response(plugin_id).await; } let plugins_manager = self.thread_manager.plugins_manager(); plugins_manager .uninstall_plugin(plugin_id) .await .map_err(Self::plugin_uninstall_error)?; match self.load_latest_config(/*fallback_cwd*/ None).await { Ok(_) => self.on_effective_plugins_changed(), Err(err) => { warn!( "failed to reload config after plugin uninstall, clearing plugin-related caches only: {err:?}" ); self.clear_plugin_related_caches(); } } Ok(PluginUninstallResponse {}) } fn plugin_install_error(err: CorePluginInstallError) -> JSONRPCErrorError { if err.is_invalid_request() { return invalid_request(err.to_string()); } match err { CorePluginInstallError::Marketplace(err) => { Self::marketplace_error(err, "install plugin") } CorePluginInstallError::Config(err) => { internal_error(format!("failed to persist installed plugin config: {err}")) } CorePluginInstallError::Remote(err) => { internal_error(format!("failed to enable remote plugin: {err}")) } CorePluginInstallError::Join(err) => { internal_error(format!("failed to install plugin: {err}")) } CorePluginInstallError::Store(err) => { internal_error(format!("failed to install plugin: {err}")) } } } fn plugin_uninstall_error(err: CorePluginUninstallError) -> JSONRPCErrorError { if err.is_invalid_request() { return invalid_request(err.to_string()); } match err { CorePluginUninstallError::Config(err) => { internal_error(format!("failed to clear plugin config: {err}")) } CorePluginUninstallError::Remote(err) => { internal_error(format!("failed to uninstall remote plugin: {err}")) } CorePluginUninstallError::Join(err) => { internal_error(format!("failed to uninstall plugin: {err}")) } CorePluginUninstallError::Store(err) => { internal_error(format!("failed to uninstall plugin: {err}")) } CorePluginUninstallError::InvalidPluginId(_) => { unreachable!("invalid plugin ids are handled above"); } } } fn marketplace_error(err: MarketplaceError, action: &str) -> JSONRPCErrorError { match err { MarketplaceError::MarketplaceNotFound { .. } | MarketplaceError::InvalidMarketplaceFile { .. } | MarketplaceError::PluginNotFound { .. } | MarketplaceError::PluginNotAvailable { .. } | MarketplaceError::PluginsDisabled | MarketplaceError::InvalidPlugin(_) => invalid_request(err.to_string()), MarketplaceError::Io { .. } => internal_error(format!("failed to {action}: {err}")), } } async fn remote_plugin_uninstall_response( &self, plugin_id: String, ) -> Result<PluginUninstallResponse, JSONRPCErrorError> { let config = self.load_latest_config(/*fallback_cwd*/ None).await?; if !config.features.enabled(Feature::Plugins) { return Err(invalid_request("remote plugin uninstall is not enabled")); } validate_remote_plugin_id(&plugin_id)?; let auth = self.auth_manager.auth().await; let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; let uninstall_result = codex_core_plugins::remote::uninstall_remote_plugin( &remote_plugin_service_config, auth.as_ref(), config.codex_home.to_path_buf(), &plugin_id, ) .await; if matches!( &uninstall_result, Ok(()) | Err(RemotePluginCatalogError::CacheRemove(_)) ) { let plugins_manager = self.thread_manager.plugins_manager(); if plugins_manager.clear_remote_installed_plugins_cache() { self.on_effective_plugins_changed(); } plugins_manager.maybe_start_remote_installed_plugins_cache_refresh_after_mutation( &config.plugins_config_input(), auth.clone(), Some(self.effective_plugins_changed_callback()), ); } uninstall_result.map_err(|err| { remote_plugin_catalog_error_to_jsonrpc(err, "uninstall remote plugin") })?; Ok(PluginUninstallResponse {}) }}async fn load_plugin_app_summaries( config: &Config, plugin_apps: &[codex_plugin::AppConnectorId], app_category_by_id: &HashMap<String, String>,) -> Vec<AppSummary> { if plugin_apps.is_empty() { return Vec::new(); } let connectors = match connectors::list_all_connectors_with_options( config, /*force_refetch*/ false, &[], ) .await { Ok(connectors) => connectors, Err(err) => { warn!("failed to load app metadata for plugin/read: {err:#}"); connectors::list_cached_all_connectors(config, &[]) .await .unwrap_or_default() } }; let plugin_connectors = connectors::connectors_for_plugin_apps(connectors, plugin_apps); plugin_connectors .into_iter() .map(|connector| { let category = app_category_by_id .get(&connector.id) .cloned() .or_else(|| connector.category()); AppSummary { id: connector.id, name: connector.name, description: connector.description, install_url: connector.install_url, category, } }) .collect()}fn plugin_app_category_by_id_from_value(value: &serde_json::Value) -> HashMap<String, String> { codex_core_plugins::loader::plugin_app_declarations_from_value(value) .into_iter() .filter_map(|app| app.category.map(|category| (app.connector_id.0, category))) .collect()}fn plugin_apps_needing_auth( all_connectors: &[AppInfo], accessible_connectors: &[AppInfo], plugin_apps: &[codex_plugin::AppConnectorId], codex_apps_ready: bool,) -> Vec<AppSummary> { if !codex_apps_ready { return Vec::new(); } let accessible_ids = accessible_connectors .iter() .map(|connector| connector.id.as_str()) .collect::<HashSet<_>>(); let plugin_app_ids = plugin_apps .iter() .map(|connector_id| connector_id.0.as_str()) .collect::<HashSet<_>>(); all_connectors .iter() .filter(|connector| { plugin_app_ids.contains(connector.id.as_str()) && !accessible_ids.contains(connector.id.as_str()) }) .cloned() .map(|connector| { let category = connector.category(); AppSummary { category, id: connector.id, name: connector.name, description: connector.description, install_url: connector.install_url, } }) .collect()}fn remote_marketplace_to_info(marketplace: RemoteMarketplace) -> PluginMarketplaceEntry { PluginMarketplaceEntry { name: marketplace.name, path: None, interface: Some(MarketplaceInterface { display_name: Some(marketplace.display_name), }), plugins: marketplace .plugins .into_iter() .map(remote_plugin_summary_to_info) .collect(), }}fn remote_plugin_summary_to_info(summary: RemoteCatalogPluginSummary) -> PluginSummary { PluginSummary { id: summary.id, remote_plugin_id: Some(summary.remote_plugin_id), local_version: None, name: summary.name, share_context: summary .share_context .map(remote_plugin_share_context_to_info), source: PluginSource::Remote, installed: summary.installed, enabled: summary.enabled, install_policy: summary.install_policy, auth_policy: summary.auth_policy, availability: summary.availability, interface: summary.interface, keywords: summary.keywords, }}fn remote_plugin_share_context_to_info( context: RemoteCatalogPluginShareContext,) -> PluginShareContext { PluginShareContext { remote_plugin_id: context.remote_plugin_id, remote_version: context.remote_version, discoverability: Some(remote_plugin_share_discoverability_to_info( context.discoverability, )), share_url: context.share_url, creator_account_user_id: context.creator_account_user_id, creator_name: context.creator_name, share_principals: context.share_principals.map(|principals| { principals .into_iter() .map(plugin_share_principal_from_remote) .collect() }), }}fn remote_plugin_share_discoverability_to_info( discoverability: codex_core_plugins::remote::RemotePluginShareDiscoverability,) -> PluginShareDiscoverability { match discoverability { codex_core_plugins::remote::RemotePluginShareDiscoverability::Listed => { PluginShareDiscoverability::Listed } codex_core_plugins::remote::RemotePluginShareDiscoverability::Unlisted => { PluginShareDiscoverability::Unlisted } codex_core_plugins::remote::RemotePluginShareDiscoverability::Private => { PluginShareDiscoverability::Private } }}fn remote_plugin_detail_to_info( detail: RemoteCatalogPluginDetail, apps: Vec<AppSummary>,) -> PluginDetail { let app_templates = detail .app_templates .into_iter() .map(|template| AppTemplateSummary { template_id: template.template_id, name: template.name, description: template.description, category: template.category, canonical_connector_id: template.canonical_connector_id, logo_url: template.logo_url, logo_url_dark: template.logo_url_dark, materialized_app_ids: template.materialized_app_ids, reason: template.reason.map(|reason| match reason { RemoteAppTemplateUnavailableReason::NotConfiguredForWorkspace => { AppTemplateUnavailableReason::NotConfiguredForWorkspace } RemoteAppTemplateUnavailableReason::NoActiveWorkspace => { AppTemplateUnavailableReason::NoActiveWorkspace } }), }) .collect(); PluginDetail { marketplace_name: detail.marketplace_name, marketplace_path: None, summary: remote_plugin_summary_to_info(detail.summary), share_url: detail.share_url, description: detail.description, skills: detail .skills .into_iter() .map(|skill| SkillSummary { name: skill.name, description: skill.description, short_description: skill.short_description, interface: skill.interface, path: None, enabled: skill.enabled, }) .collect(), hooks: Vec::new(), apps, app_templates, mcp_servers: detail.mcp_servers, }}fn remote_plugin_catalog_error_to_jsonrpc( err: RemotePluginCatalogError, context: &str,) -> JSONRPCErrorError { let message = format!("{context}: {err}"); match &err { RemotePluginCatalogError::AuthRequired | RemotePluginCatalogError::UnsupportedAuthMode => { invalid_request(message) } RemotePluginCatalogError::UnexpectedStatus { status, .. } if status.as_u16() == 404 => { invalid_request(message) } RemotePluginCatalogError::InvalidPluginPath { .. } | RemotePluginCatalogError::PluginShareCheckoutNotAvailable { .. } | RemotePluginCatalogError::ArchiveTooLarge { .. } | RemotePluginCatalogError::UnknownMarketplace { .. } => invalid_request(message), RemotePluginCatalogError::AuthToken(_) | RemotePluginCatalogError::Request { .. } | RemotePluginCatalogError::UnexpectedStatus { .. } | RemotePluginCatalogError::Decode { .. } | RemotePluginCatalogError::InvalidBaseUrl(_) | RemotePluginCatalogError::InvalidBaseUrlPath | RemotePluginCatalogError::UnexpectedPluginId { .. } | RemotePluginCatalogError::UnexpectedSkillName { .. } | RemotePluginCatalogError::UnexpectedEnabledState { .. } | RemotePluginCatalogError::Archive { .. } | RemotePluginCatalogError::ArchiveJoin(_) | RemotePluginCatalogError::MissingUploadEtag | RemotePluginCatalogError::UnexpectedResponse(_) | RemotePluginCatalogError::CacheRemove(_) => internal_error(message), }}fn remote_plugin_bundle_install_error_to_jsonrpc( err: codex_core_plugins::remote_bundle::RemotePluginBundleInstallError,) -> JSONRPCErrorError { internal_error(format!("install remote plugin bundle: {err}"))}