Preload plugin recommendations during session startup (#41375)

## Why

Fetching plugin recommendations while building the first-turn context delays that
context when the recommendation catalog is not already cached.

## What changed

- Start fetching recommendations during session initialization, in parallel with
  MCP and plugin setup, and reuse the in-flight request when building context.
- Centralize the feature checks that enable plugin recommendations.
- Prevent an invalidated recommendation request from repopulating the cache after
  a newer request has completed.

## Testing

Added coverage for startup feature gating, reuse of the in-flight fetch, legacy
behavior when recommendations are unavailable, and cache invalidation races.

GitOrigin-RevId: 708774b9bfc472c9847973ae8b2774b66e7321e4

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) => {

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();

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();

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,

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!(
+            "<recommended_plugins>\n",
+            "Here is a list of plugins that are available but not installed.\n\n",
+            "- GitHub (github@openai-curated-remote)\n",
+            "</recommended_plugins>",
+        )]
+    );
+    server.verify().await;
+}
+
 #[tokio::test]
 async fn build_initial_context_describes_active_realtime_state() {
     let (session, mut turn_context) = make_session_and_context().await;

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<Config> for StartupMcpBarrier {
+    fn id(&self) -> &'static str {
+        "startup_recommendation_barrier"
+    }
+
+    fn contribute<'a>(
+        &'a self,
+        _context: McpServerContributionContext<'a, Config>,
+    ) -> ExtensionFuture<'a, Vec<McpServerContribution>> {
+        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<String> {
     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<PathUri> {
+    // 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<Feature>,
+) -> 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::<Config>::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("<recommended_plugins>"),
+        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,

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)
     }