diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs
index 724acb265c..a0aa76b889 100644
--- a/codex-rs/core-plugins/src/manager.rs
+++ b/codex-rs/core-plugins/src/manager.rs
@@ -1764,11 +1764,22 @@ impl PluginsManager {
.await
{
Ok(mode) => {
- let mut cache = match self.recommended_plugins_cache.write() {
- Ok(cache) => cache,
+ let refreshes = match self.recommended_plugins_refreshes.read() {
+ Ok(refreshes) => refreshes,
Err(err) => err.into_inner(),
};
- cache.insert(cache_key.clone(), mode.clone());
+ // Hold the refresh lock through publication, matching cache-clear lock
+ // order. An invalidated initializer must not repopulate the cache.
+ if refreshes
+ .get(&cache_key)
+ .is_some_and(|current| Arc::ptr_eq(current, &refresh))
+ {
+ let mut cache = match self.recommended_plugins_cache.write() {
+ Ok(cache) => cache,
+ Err(err) => err.into_inner(),
+ };
+ cache.insert(cache_key.clone(), mode.clone());
+ }
mode
}
Err(err) => {
diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs
index 123dfb9f4d..8648b0ff24 100644
--- a/codex-rs/core-plugins/src/manager_tests.rs
+++ b/codex-rs/core-plugins/src/manager_tests.rs
@@ -5896,6 +5896,94 @@ plugins = true
);
}
+#[tokio::test]
+async fn recommended_plugins_cache_clear_rejects_stale_fetch_completion() {
+ let tmp = tempfile::tempdir().unwrap();
+ write_file(
+ &tmp.path().join(CONFIG_TOML_FILE),
+ "[features]\nplugins = true\n",
+ );
+ let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+ let server = MockServer::start().await;
+ let request_started = Arc::new(tokio::sync::Notify::new());
+ let started = Arc::clone(&request_started);
+ let (release_response, released) = std::sync::mpsc::channel();
+ let released = std::sync::Mutex::new(released);
+ let old_response = ResponseTemplate::new(200).set_body_json(serde_json::json!({
+ "enabled": true,
+ "plugins": [{"id": "plugin_old", "name": "old", "display_name": "Old"}]
+ }));
+ Mock::given(method("GET"))
+ .and(path("/ps/plugins/suggested/codex"))
+ .respond_with(move |_: &wiremock::Request| {
+ started.notify_one();
+ // Wiremock runs its own server thread. Dropping the sender also releases it.
+ let _ = released.lock().unwrap().recv();
+ old_response.clone()
+ })
+ .up_to_n_times(1)
+ .expect(1)
+ .mount(&server)
+ .await;
+ Mock::given(method("GET"))
+ .and(path("/ps/plugins/suggested/codex"))
+ .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+ "enabled": true,
+ "plugins": []
+ })))
+ .expect(1)
+ .mount(&server)
+ .await;
+
+ let mut config = load_config(tmp.path(), tmp.path()).await;
+ config.chatgpt_base_url = server.uri();
+ let manager = test_plugins_manager(tmp.path().to_path_buf());
+ let old_lookup = manager.recommended_plugins_mode_for_config(&config, Some(&auth));
+ tokio::pin!(old_lookup);
+ tokio::select! {
+ biased;
+ _ = request_started.notified() => {}
+ mode = old_lookup.as_mut() => panic!("old fetch finished before release: {mode:?}"),
+ }
+
+ manager.clear_recommended_plugins_cache();
+ release_response
+ .send(())
+ .expect("release old HTTP response");
+
+ // Keep the old initializer unpolled until the new fetch publishes. An invalidated
+ // fetch must not overwrite the fresh result.
+ let new_mode = tokio::time::timeout(
+ Duration::from_secs(10),
+ manager.recommended_plugins_mode_for_config(&config, Some(&auth)),
+ )
+ .await
+ .expect("new lookup should finish independently of the invalidated fetch");
+ assert_eq!(
+ new_mode,
+ RecommendedPluginsMode::Endpoint {
+ plugins: Vec::new()
+ }
+ );
+ assert_eq!(
+ old_lookup.await,
+ RecommendedPluginsMode::Endpoint {
+ plugins: vec![RecommendedPlugin {
+ config_id: "old@openai-curated-remote".to_string(),
+ remote_plugin_id: "plugin_old".to_string(),
+ display_name: "Old".to_string(),
+ }]
+ }
+ );
+ assert_eq!(
+ manager
+ .recommended_plugins_mode_for_config(&config, Some(&auth))
+ .await,
+ new_mode
+ );
+ server.verify().await;
+}
+
#[tokio::test]
async fn recommended_plugin_candidates_filter_installed_and_disabled_plugins() {
let tmp = tempfile::tempdir().unwrap();
diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs
index 70b04c3f65..790afd253f 100644
--- a/codex-rs/core/src/session/mod.rs
+++ b/codex-rs/core/src/session/mod.rs
@@ -3731,11 +3731,10 @@ impl Session {
.plugins_manager
.plugins_for_config(&turn_context.config.plugins_config_input())
.await;
- let features = turn_context.config.features.get();
- let recommended_plugin_candidates = if features.enabled(Feature::Apps)
- && features.enabled(Feature::Plugins)
- && (features.enabled(Feature::ToolSuggest)
- || features.enabled(Feature::RecommendedPlugins))
+ let recommended_plugin_candidates = if turn_context
+ .config
+ .features
+ .plugin_recommendations_enabled()
{
let auth = self.services.auth_manager.auth().await;
let plugins_config = turn_context.config.plugins_config_input();
diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs
index 0e3daef637..c944907274 100644
--- a/codex-rs/core/src/session/session.rs
+++ b/codex-rs/core/src/session/session.rs
@@ -928,6 +928,7 @@ impl Session {
let mut mcp_auth_changes = auth_manager.auth_change_receiver();
let auth_manager_clone = Arc::clone(&auth_manager);
+ let plugins_manager_for_prewarm = Arc::clone(&plugins_manager);
let config_for_mcp = Arc::clone(&config);
let mcp_manager_for_mcp = Arc::clone(&mcp_manager);
let mcp_thread_init_for_startup = &mcp_thread_init;
@@ -941,6 +942,20 @@ impl Session {
.unwrap_or_else(|| session_configuration.cwd().to_path_buf());
let auth_and_mcp_fut = async move {
let auth = auth_manager_clone.auth().await;
+ if config_for_mcp.features.plugin_recommendations_enabled() {
+ let plugins_config = config_for_mcp.plugins_config_input();
+ let auth_for_prewarm = auth.clone();
+ // Fetch the catalog while MCP and plugin/skill initialization continue.
+ // Context construction still handles filtering and prompt insertion.
+ tokio::spawn(async move {
+ plugins_manager_for_prewarm
+ .recommended_plugins_mode_for_config(
+ &plugins_config,
+ auth_for_prewarm.as_ref(),
+ )
+ .await;
+ });
+ }
let mcp_projection = mcp_manager_for_mcp
.runtime_config_for_step(
&config_for_mcp,
diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs
index b334571814..a8ff8d77b3 100644
--- a/codex-rs/core/src/session/tests.rs
+++ b/codex-rs/core/src/session/tests.rs
@@ -9726,6 +9726,86 @@ async fn record_context_updates_emits_realtime_end_when_session_stops_being_live
);
}
+#[tokio::test]
+async fn build_initial_context_reuses_in_flight_recommendation_prewarm() {
+ use wiremock::Mock;
+ use wiremock::ResponseTemplate;
+ use wiremock::matchers::method;
+ use wiremock::matchers::path;
+ use wiremock::matchers::query_param;
+
+ core_test_support::skip_if_no_network!();
+
+ let server = start_mock_server().await;
+ Mock::given(method("GET"))
+ .and(path("/ps/plugins/suggested/codex"))
+ .and(query_param("scope", "GLOBAL"))
+ .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+ "enabled": true,
+ "plugins": [{
+ "id": "plugin_github",
+ "name": "github",
+ "display_name": "GitHub"
+ }]
+ })))
+ .expect(1)
+ .mount(&server)
+ .await;
+ let (session, turn_context, _rx_event) = make_session_and_context_with_auth_and_config_and_rx(
+ CodexAuth::create_dummy_chatgpt_auth_for_testing(),
+ /*dynamic_tools*/ Vec::new(),
+ |config| {
+ config.chatgpt_base_url = server.uri();
+ config
+ .features
+ .disable(Feature::ToolSuggest)
+ .expect("test config should allow feature update");
+ for enabled_feature in [
+ Feature::Apps,
+ Feature::Plugins,
+ Feature::RemotePlugin,
+ Feature::RecommendedPlugins,
+ ] {
+ config
+ .features
+ .enable(enabled_feature)
+ .expect("test config should allow feature update");
+ }
+ },
+ )
+ .await;
+ let plugins_manager = &session.services.plugins_manager;
+ let plugins_config = turn_context.config.plugins_config_input();
+ let auth = session.services.auth_manager.auth().await;
+ // Cached plugin loading and auth let initial context reach the shared recommendation lookup
+ // without awaiting unrelated I/O.
+ plugins_manager.plugins_for_config(&plugins_config).await;
+ let prewarm =
+ plugins_manager.recommended_plugins_mode_for_config(&plugins_config, auth.as_ref());
+ tokio::pin!(prewarm);
+ assert!(futures::poll!(prewarm.as_mut()).is_pending());
+
+ // Keep the OnceCell initializer unpolled while first-thread context joins its in-flight fetch.
+ // This does not depend on how quickly the HTTP server returns its response.
+ let world_state = WorldState::default();
+ let initial_context =
+ session.build_initial_context_with_world_state(&turn_context, &world_state);
+ tokio::pin!(initial_context);
+ assert!(futures::poll!(initial_context.as_mut()).is_pending());
+
+ let (_, initial_context) = tokio::join!(prewarm, initial_context);
+ assert_eq!(
+ user_input_texts(&initial_context),
+ vec![concat!(
+ "\n",
+ "Here is a list of plugins that are available but not installed.\n\n",
+ "- GitHub (github@openai-curated-remote)\n",
+ "",
+ )]
+ );
+ server.verify().await;
+}
+
#[tokio::test]
async fn build_initial_context_describes_active_realtime_state() {
let (session, mut turn_context) = make_session_and_context().await;
diff --git a/codex-rs/core/tests/suite/request_plugin_install.rs b/codex-rs/core/tests/suite/request_plugin_install.rs
index 5d3eda19bb..950dfb7bec 100644
--- a/codex-rs/core/tests/suite/request_plugin_install.rs
+++ b/codex-rs/core/tests/suite/request_plugin_install.rs
@@ -8,6 +8,11 @@ use codex_config::types::ToolSuggestDiscoverableType;
use codex_core::TurnInputRequest;
use codex_core::config::Config;
use codex_core_plugins::startup_sync::curated_plugins_repo_path;
+use codex_extension_api::ExtensionFuture;
+use codex_extension_api::ExtensionRegistryBuilder;
+use codex_extension_api::McpServerContribution;
+use codex_extension_api::McpServerContributionContext;
+use codex_extension_api::McpServerContributor;
use codex_features::Feature;
use codex_login::CodexAuth;
use codex_models_manager::bundled_models_response;
@@ -40,6 +45,7 @@ use core_test_support::test_codex::test_codex;
use core_test_support::test_codex::turn_permission_fields;
use core_test_support::wait_for_event;
use core_test_support::wait_for_event_match;
+use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::sync::Arc;
@@ -49,6 +55,7 @@ use std::time::Duration;
use std::time::Instant;
use tempfile::TempDir;
use test_case::test_case;
+use tokio::sync::Notify;
use wiremock::Mock;
use wiremock::MockGuard;
use wiremock::MockServer;
@@ -71,6 +78,31 @@ const CALENDAR_NAMESPACE: &str = "mcp__codex_apps__calendar";
const CALENDAR_CREATE_EVENT_TOOL: &str = "_create_event";
const STEP_PREPARATION_MCP_SERVER: &str = "step_preparation";
+struct StartupMcpBarrier {
+ block_next: AtomicBool,
+ entered: Notify,
+ release: Notify,
+}
+
+impl McpServerContributor for StartupMcpBarrier {
+ fn id(&self) -> &'static str {
+ "startup_recommendation_barrier"
+ }
+
+ fn contribute<'a>(
+ &'a self,
+ _context: McpServerContributionContext<'a, Config>,
+ ) -> ExtensionFuture<'a, Vec> {
+ Box::pin(async move {
+ if self.block_next.swap(false, Ordering::SeqCst) {
+ self.entered.notify_one();
+ self.release.notified().await;
+ }
+ Vec::new()
+ })
+ }
+}
+
fn tool_names(body: &Value) -> Vec {
body.get("tools")
.and_then(Value::as_array)
@@ -126,6 +158,25 @@ async fn mount_recommendations(server: &wiremock::MockServer, response: Response
.await;
}
+async fn wait_for_startup_recommendations(server: &MockServer) -> Result<()> {
+ tokio::time::timeout(Duration::from_secs(5), async {
+ loop {
+ if server
+ .received_requests()
+ .await
+ .unwrap_or_default()
+ .iter()
+ .any(|request| request.url.path() == "/ps/plugins/suggested/codex")
+ {
+ return;
+ }
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ }
+ })
+ .await
+ .context("session startup should request recommendations before the first turn")
+}
+
fn assert_legacy_tools(body: &Value) {
let tools = tool_names(body);
assert!(!tools.iter().any(|name| name == TOOL_SEARCH_TOOL_NAME));
@@ -204,6 +255,15 @@ async fn build_gated_step_preparation_test(
}
async fn start_gated_step_preparation(test: &TestCodex, server: &MockServer) -> Result {
+ // Settle startup work before clearing its cache so these tests still exercise
+ // a fresh recommendation fetch alongside first-turn MCP discovery.
+ wait_for_startup_recommendations(server).await?;
+ let plugins_manager = test.thread_manager.plugins_manager();
+ let auth = test.thread_manager.auth_manager().auth().await;
+ plugins_manager
+ .recommended_plugins_mode_for_config(&test.config.plugins_config_input(), auth.as_ref())
+ .await;
+ plugins_manager.clear_recommended_plugins_cache();
let prior_recommendation_count = server
.received_requests()
.await
@@ -385,6 +445,124 @@ async fn mount_remote_calendar_installed_plugins(server: &wiremock::MockServer)
.await;
}
+#[test_case(Feature::ToolSuggest, None; "tool suggest")]
+#[test_case(Feature::RecommendedPlugins, None; "recommended plugins")]
+#[test_case(Feature::RecommendedPlugins, Some(Feature::Apps); "apps disabled")]
+#[test_case(Feature::RecommendedPlugins, Some(Feature::Plugins); "plugins disabled")]
+#[test_case(Feature::RecommendedPlugins, Some(Feature::RemotePlugin); "remote plugins disabled")]
+#[test_case(Feature::RecommendedPlugins, Some(Feature::RecommendedPlugins); "recommendations disabled")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn startup_recommendations(
+ feature: Feature,
+ disabled_feature: Option,
+) -> Result<()> {
+ skip_if_no_network!(Ok(()));
+
+ let expect_recommendations = disabled_feature.is_none();
+ let server = start_mock_server().await;
+ let apps_server = AppsTestServer::mount(&server).await?;
+ let recommendation_started = Arc::new(Notify::new());
+ let started = Arc::clone(&recommendation_started);
+ let response = ResponseTemplate::new(200).set_body_json(json!({
+ "enabled": true,
+ "plugins": [{
+ "id": "plugin_github",
+ "name": "github",
+ "display_name": "GitHub"
+ }]
+ }));
+ Mock::given(method("GET"))
+ .and(path("/ps/plugins/suggested/codex"))
+ .respond_with(move |_: &wiremock::Request| {
+ started.notify_one();
+ response.clone()
+ })
+ .expect(u64::from(expect_recommendations))
+ .mount(&server)
+ .await;
+ let model_response = mount_sse_once(
+ &server,
+ sse(vec![
+ ev_response_created("startup-recommendations"),
+ ev_assistant_message("startup-recommendations-message", "done"),
+ ev_completed("startup-recommendations"),
+ ]),
+ )
+ .await;
+ let barrier = Arc::new(StartupMcpBarrier {
+ block_next: AtomicBool::new(true),
+ entered: Notify::new(),
+ release: Notify::new(),
+ });
+ let mut extensions = ExtensionRegistryBuilder::::new();
+ extensions.mcp_server_contributor(barrier.clone());
+ let mut builder = test_codex()
+ .with_extensions(Arc::new(extensions.build()))
+ .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
+ .with_config(move |config| {
+ configure_apps_without_search_tool(config, &apps_server.chatgpt_base_url);
+ config.model_provider.supports_websockets = false;
+ for suggestion_feature in [Feature::ToolSuggest, Feature::RecommendedPlugins] {
+ config
+ .features
+ .disable(suggestion_feature)
+ .expect("test config should allow feature update");
+ }
+ config
+ .features
+ .enable(feature)
+ .expect("test config should allow feature update");
+ if let Some(disabled_feature) = disabled_feature {
+ config
+ .features
+ .disable(disabled_feature)
+ .expect("test config should allow feature update");
+ }
+ });
+ let startup = builder.build_with_auto_env(&server);
+ tokio::pin!(startup);
+
+ tokio::select! {
+ result = &mut startup => {
+ result?;
+ anyhow::bail!("session startup should remain blocked in MCP configuration");
+ }
+ _ = barrier.entered.notified() => {}
+ }
+ if expect_recommendations {
+ tokio::time::timeout(Duration::from_secs(5), recommendation_started.notified())
+ .await
+ .context("recommendations must start before MCP configuration is released")?;
+ }
+ assert!(futures::poll!(&mut startup).is_pending());
+ assert!(model_response.requests().is_empty());
+ barrier.release.notify_one();
+ let test = startup.await?;
+
+ test.submit_turn("suggest a plugin").await?;
+
+ let request = model_response.single_request();
+ let user_context = request.message_input_texts("user").join("\n");
+ assert_eq!(
+ user_context.contains(""),
+ expect_recommendations,
+ );
+ assert_eq!(
+ user_context.contains("- GitHub (github@openai-curated-remote)"),
+ expect_recommendations,
+ );
+ let tools = tool_names(&request.body_json());
+ assert_eq!(
+ tools
+ .iter()
+ .any(|name| name == REQUEST_PLUGIN_INSTALL_TOOL_NAME),
+ expect_recommendations && feature == Feature::ToolSuggest,
+ );
+ test.codex.shutdown_and_wait().await?;
+ server.verify().await;
+ Ok(())
+}
+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mcp_discovery_overlaps_endpoint_plugin_recommendations() -> Result<()> {
skip_if_wine_exec!(
@@ -509,17 +687,17 @@ async fn interrupting_concurrent_step_preparation_prevents_sampling() -> Result<
Ok(())
}
+#[test_case(ResponseTemplate::new(200).set_body_json(json!({"enabled": false, "plugins": []})); "endpoint disabled")]
+#[test_case(ResponseTemplate::new(503); "fetch failure")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn explicit_false_preserves_legacy_workflow() -> Result<()> {
+async fn unavailable_recommendations_preserve_legacy_workflow(
+ response: ResponseTemplate,
+) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let apps_server = AppsTestServer::mount(&server).await?;
- mount_recommendations(
- &server,
- ResponseTemplate::new(200).set_body_json(json!({"enabled": false, "plugins": []})),
- )
- .await;
+ mount_recommendations(&server, response).await;
let call_id = "list-installable-tools";
let mock = mount_sse_sequence(
&server,
diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs
index 3724a4f2d8..6edf4e627c 100644
--- a/codex-rs/features/src/lib.rs
+++ b/codex-rs/features/src/lib.rs
@@ -474,6 +474,12 @@ impl Features {
self.enabled(Feature::Apps) && has_chatgpt_auth
}
+ pub fn plugin_recommendations_enabled(&self) -> bool {
+ self.enabled(Feature::Apps)
+ && self.enabled(Feature::Plugins)
+ && (self.enabled(Feature::ToolSuggest) || self.enabled(Feature::RecommendedPlugins))
+ }
+
pub fn use_legacy_landlock(&self) -> bool {
self.enabled(Feature::UseLegacyLandlock)
}