diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 4314f590dd..52232dfab4 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -216,7 +216,7 @@ Example with notification opt-out: - `marketplace/add` — add a remote plugin marketplace from an HTTP(S) Git URL, SSH Git URL, or GitHub `owner/repo` shorthand, then persist it into the user marketplace config. Returns the installed root path plus whether the marketplace was already present. - `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 `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. 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/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 `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/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`; local rows return `null`. 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. 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. diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index c63a5531ca..a694a24ee4 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -571,6 +571,14 @@ impl PluginRequestProcessor { 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(); + if include_local + && force_refetch + && plugins_manager + .refresh_non_curated_plugin_cache_for_config(&plugins_input, &roots) + .await + { + self.on_effective_plugins_changed(); + } let include_shared_with_me = marketplace_kinds.contains(&PluginListMarketplaceKind::SharedWithMe); let include_created_by_me_remote = marketplace_kinds diff --git a/codex-rs/app-server/tests/suite/v2/plugin_list.rs b/codex-rs/app-server/tests/suite/v2/plugin_list.rs index 29916e189d..a08361a411 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_list.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_list.rs @@ -12,6 +12,7 @@ use codex_app_server_protocol::HookMetadata; use codex_app_server_protocol::HookTrustStatus; use codex_app_server_protocol::HooksListParams; use codex_app_server_protocol::HooksListResponse; +use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::PluginAuthPolicy; use codex_app_server_protocol::PluginInstallPolicy; use codex_app_server_protocol::PluginInstallPolicySource; @@ -944,6 +945,127 @@ async fn plugin_list_returns_share_context_for_shared_local_plugin() -> Result<( Ok(()) } +#[tokio::test] +async fn plugin_list_force_refetch_waits_for_same_path_local_plugin_upgrade() -> Result<()> { + let codex_home = TempDir::new()?; + let marketplace_root = TempDir::new()?; + std::fs::create_dir_all(marketplace_root.path().join(".git"))?; + std::fs::create_dir_all(marketplace_root.path().join(".agents/plugins"))?; + let source_manifest = marketplace_root + .path() + .join("sample-plugin/.codex-plugin/plugin.json"); + std::fs::create_dir_all(source_manifest.parent().expect("source manifest parent"))?; + std::fs::write( + &source_manifest, + r#"{"name":"sample-plugin","version":"1.0.0"}"#, + )?; + std::fs::write( + marketplace_root + .path() + .join(".agents/plugins/marketplace.json"), + r#"{ + "name": "sample-marketplace", + "plugins": [ + { + "name": "sample-plugin", + "source": { + "source": "local", + "path": "./sample-plugin" + } + } + ] +}"#, + )?; + std::fs::write( + codex_home.path().join("config.toml"), + r#"[features] +plugins = true +remote_plugin = false + +[plugins."sample-plugin@sample-marketplace"] +enabled = true +"#, + )?; + write_installed_plugin_with_version( + &codex_home, + "sample-marketplace", + "sample-plugin", + "1.0.0", + )?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: Some(vec![AbsolutePathBuf::try_from(marketplace_root.path())?]), + marketplace_kinds: Some(vec![PluginListMarketplaceKind::Local]), + force_refetch: true, + }) + .await?; + let initial_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let _: PluginListResponse = to_response(initial_response)?; + + std::fs::write( + &source_manifest, + r#"{"name":"sample-plugin","version":"1.1.0"}"#, + )?; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: Some(vec![AbsolutePathBuf::try_from(marketplace_root.path())?]), + marketplace_kinds: Some(vec![PluginListMarketplaceKind::Local]), + force_refetch: true, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: PluginListResponse = to_response(response)?; + let plugin = response + .marketplaces + .iter() + .find(|marketplace| marketplace.name == "sample-marketplace") + .and_then(|marketplace| { + marketplace + .plugins + .iter() + .find(|plugin| plugin.name == "sample-plugin") + }) + .expect("upgraded local plugin should appear in its marketplace response"); + assert!(plugin.installed); + assert!(plugin.enabled); + assert_eq!(plugin.local_version.as_deref(), Some("1.1.0")); + + let plugin_cache = codex_home + .path() + .join("plugins/cache/sample-marketplace/sample-plugin"); + let installed_manifest = plugin_cache.join("1.1.0/.codex-plugin/plugin.json"); + assert!( + installed_manifest.is_file(), + "force-refetched plugin/list must finish installing the newer local plugin before responding" + ); + assert!( + !plugin_cache.join("1.0.0").exists(), + "force-refetched plugin/list must remove the superseded local plugin before responding" + ); + let installed_manifest: serde_json::Value = + serde_json::from_slice(&std::fs::read(installed_manifest)?)?; + assert_eq!(installed_manifest["version"], serde_json::json!("1.1.0")); + + Ok(()) +} + #[tokio::test] async fn plugin_list_includes_install_and_enabled_state_from_config() -> Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index df8341284b..3c9b47ec78 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -104,6 +104,7 @@ use std::sync::atomic::Ordering; use std::time::Instant; use tokio::sync::OnceCell; use tokio::sync::Semaphore; +use tokio::sync::watch; use tracing::instrument; use tracing::warn; @@ -249,9 +250,18 @@ pub struct PluginListBackgroundTaskOptions { struct NonCuratedCacheRefreshRequest { roots: Vec, configured_plugin_keys: Vec, + configured_plugin_sources: Vec, mode: NonCuratedCacheRefreshMode, } +#[derive(Clone, PartialEq, Eq)] +struct NonCuratedPluginSource { + marketplace_path: AbsolutePathBuf, + plugin_key: String, + source: MarketplacePluginSource, + local_version: Option, +} + #[derive(Clone, Copy, PartialEq, Eq)] enum NonCuratedCacheRefreshMode { IfVersionChanged, @@ -265,6 +275,12 @@ struct NonCuratedCacheRefreshState { in_flight: bool, } +#[derive(Clone, Copy, Default)] +struct NonCuratedCacheRefreshCompletion { + sequence: u64, + changed_sequence: u64, +} + #[derive(Default)] struct ConfiguredMarketplaceUpgradeState { in_flight: bool, @@ -406,7 +422,9 @@ pub struct PluginsManager { recommended_plugins_refreshes: RwLock>>>, configured_marketplace_upgrade_state: RwLock, + non_curated_cache_refresh_lock: Semaphore, non_curated_cache_refresh_state: RwLock, + non_curated_cache_refresh_completion: watch::Sender, // Keep the cache auth-independent so auth changes only need to resolve capabilities again. loaded_plugins_cache: RwLock, loaded_plugins_load_semaphore: Semaphore, @@ -484,7 +502,12 @@ impl PluginsManager { configured_marketplace_upgrade_state: RwLock::new( ConfiguredMarketplaceUpgradeState::default(), ), + non_curated_cache_refresh_lock: Semaphore::new(/*permits*/ 1), non_curated_cache_refresh_state: RwLock::new(NonCuratedCacheRefreshState::default()), + non_curated_cache_refresh_completion: watch::channel( + NonCuratedCacheRefreshCompletion::default(), + ) + .0, loaded_plugins_cache: RwLock::new(LoadedPluginsCache::default()), loaded_plugins_load_semaphore: Semaphore::new(/*permits*/ 1), skill_root_scan_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), @@ -2190,6 +2213,32 @@ impl PluginsManager { ); } + pub async fn refresh_non_curated_plugin_cache_for_config( + self: &Arc, + config: &PluginsConfigInput, + roots: &[AbsolutePathBuf], + ) -> bool { + let Ok(_refresh_permit) = self.non_curated_cache_refresh_lock.acquire().await else { + return false; + }; + let mut completion = self.non_curated_cache_refresh_completion.subscribe(); + let changed_sequence = completion.borrow_and_update().changed_sequence; + self.maybe_start_non_curated_plugin_cache_refresh(config, roots); + + loop { + let in_flight = match self.non_curated_cache_refresh_state.read() { + Ok(state) => state.in_flight, + Err(err) => err.into_inner().in_flight, + }; + if !in_flight { + return completion.borrow().changed_sequence != changed_sequence; + } + if completion.changed().await.is_err() { + return false; + } + } + } + fn schedule_remote_installed_plugins_cache_refresh( self: &Arc, mut request: RemoteInstalledPluginsCacheRefreshRequest, @@ -2319,6 +2368,12 @@ impl PluginsManager { } }; let policy = MarketplacePolicy::from_requirements(config.config_layer_stack.requirements()); + let mut configured_plugin_keys = + configured_plugins_from_stack(&config.config_layer_stack, self.codex_home.as_path()) + .into_keys() + .collect::>(); + configured_plugin_keys.sort_unstable(); + let mut configured_plugin_sources = Vec::new(); let mut roots = outcome .marketplaces .into_iter() @@ -2330,7 +2385,20 @@ impl PluginsManager { &marketplace.path, &marketplace.name, ) { - Ok(()) => Some(marketplace.path), + Ok(()) => { + for plugin in marketplace.plugins { + let plugin_key = format!("{}@{}", plugin.name, marketplace.name); + if configured_plugin_keys.binary_search(&plugin_key).is_ok() { + configured_plugin_sources.push(NonCuratedPluginSource { + marketplace_path: marketplace.path.clone(), + plugin_key, + source: plugin.source, + local_version: plugin.local_version, + }); + } + } + Some(marketplace.path) + } Err(err) => { warn!( marketplace = marketplace.name, @@ -2345,17 +2413,18 @@ impl PluginsManager { .collect::>(); roots.sort_unstable(); roots.dedup(); - let mut configured_plugin_keys = - configured_plugins_from_stack(&config.config_layer_stack, self.codex_home.as_path()) - .into_keys() - .collect::>(); - configured_plugin_keys.sort_unstable(); if roots.is_empty() || configured_plugin_keys.is_empty() { return; } + configured_plugin_sources.sort_by(|left, right| { + left.marketplace_path + .cmp(&right.marketplace_path) + .then_with(|| left.plugin_key.cmp(&right.plugin_key)) + }); let mut request = NonCuratedCacheRefreshRequest { roots, configured_plugin_keys, + configured_plugin_sources, mode, }; @@ -2364,6 +2433,23 @@ impl PluginsManager { Ok(state) => state, Err(err) => err.into_inner(), }; + if request.mode == NonCuratedCacheRefreshMode::IfVersionChanged + && state.last_refreshed.as_ref().is_some_and(|last_refreshed| { + request.configured_plugin_sources.iter().any(|source| { + last_refreshed + .configured_plugin_sources + .iter() + .any(|previous_source| { + previous_source.plugin_key == source.plugin_key + && previous_source.local_version == source.local_version + && (previous_source.marketplace_path != source.marketplace_path + || previous_source.source != source.source) + }) + }) + }) + { + request.mode = NonCuratedCacheRefreshMode::ForceReinstall; + } if request.mode == NonCuratedCacheRefreshMode::IfVersionChanged && state.requested.as_ref().is_some_and(|requested| { requested.mode == NonCuratedCacheRefreshMode::ForceReinstall @@ -2372,14 +2458,17 @@ impl PluginsManager { { request.mode = NonCuratedCacheRefreshMode::ForceReinstall; } - // Collapse repeated plugin/list requests onto one worker and only queue another pass - // when the roots or configured plugin set changes. Forced reinstall requests are not - // deduped against the last completed pass because the same marketplace root path can - // point at newly activated files after an auto-upgrade. + // Reconcile each canonical plugin generation once before publishing its resource. if state.requested.as_ref() == Some(&request) || (request.mode == NonCuratedCacheRefreshMode::IfVersionChanged && !state.in_flight - && state.last_refreshed.as_ref() == Some(&request)) + && state.last_refreshed.as_ref().is_some_and(|last_refreshed| { + last_refreshed.roots == request.roots + && last_refreshed.configured_plugin_keys + == request.configured_plugin_keys + && last_refreshed.configured_plugin_sources + == request.configured_plugin_sources + })) { return; } @@ -2406,6 +2495,10 @@ impl PluginsManager { }; state.in_flight = false; state.requested = None; + self.non_curated_cache_refresh_completion + .send_modify(|completion| { + completion.sequence = completion.sequence.wrapping_add(1); + }); warn!("failed to start non-curated plugin cache refresh task: {err}"); } } @@ -2585,6 +2678,10 @@ impl PluginsManager { Err(err) => err.into_inner(), }; state.in_flight = false; + self.non_curated_cache_refresh_completion + .send_modify(|completion| { + completion.sequence = completion.sequence.wrapping_add(1); + }); return; }; @@ -2604,7 +2701,7 @@ impl PluginsManager { ) } }; - let refreshed = match refresh_result { + let (refreshed, cache_changed) = match refresh_result { Ok(refresh_outcome) => { if refresh_outcome.cache_refreshed { self.clear_cache(); @@ -2616,12 +2713,15 @@ impl PluginsManager { "failed to refresh configured plugin cache" ); } - refresh_outcome.errors.is_empty() + ( + refresh_outcome.errors.is_empty(), + refresh_outcome.cache_refreshed, + ) } Err(err) => { self.clear_cache(); warn!("failed to refresh non-curated plugin cache: {err}"); - false + (false, false) } }; @@ -2632,9 +2732,19 @@ impl PluginsManager { if refreshed { state.last_refreshed = Some(request.clone()); } - if state.requested.as_ref() == Some(&request) { + let complete = state.requested.as_ref() == Some(&request); + if complete { state.requested = None; state.in_flight = false; + } + self.non_curated_cache_refresh_completion + .send_modify(|completion| { + completion.sequence = completion.sequence.wrapping_add(1); + if cache_changed { + completion.changed_sequence = completion.changed_sequence.wrapping_add(1); + } + }); + if complete { return; } }