[codex] [3/4] Activate endpoint plugin recommendations (#27704)

Summary\n- Await endpoint recommendation selection while constructing
each authenticated turn, removing the first-turn cache race.\n- Snapshot
and filter endpoint candidates once per turn, then use that same set for
the bounded contextual user fragment, tool exposure, and exact install
validation.\n- Keep recommendation selection ephemeral: do not persist
recommendation state in or gate resumed threads on prior context.\n-
Hide the legacy list tool in endpoint mode and preserve legacy discovery
unchanged when the endpoint is disabled or unavailable.\n- Keep remote
plugin and connector app identities out of model-visible context and
attach them only to Codex-owned elicitation metadata.\n\nStack\n- 3/4,
based on #28400.\n- Endpoint client and cache: #28399.\n- Generalized
suggestion presentation: #28400.\n- Install-schema follow-up:
#28403.\n\nValidation\n- \n- \n- \n- \n- Full : 2,649 passed and 88
environment-dependent tests failed because this sandbox cannot write ,
nest Seatbelt, or locate auxiliary test binaries.

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

@@ -3844,6 +3844,79 @@ remote_plugin = true
     );
 }
 
+#[tokio::test]
+async fn recommended_plugin_candidates_filter_installed_and_disabled_plugins() {
+    let tmp = tempfile::tempdir().unwrap();
+    write_file(
+        &tmp.path().join(CONFIG_TOML_FILE),
+        r#"[features]
+plugins = true
+remote_plugin = true
+"#,
+    );
+    write_cached_plugin(tmp.path(), REMOTE_GLOBAL_MARKETPLACE_NAME, "linear");
+
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/ps/plugins/suggested"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "enabled": true,
+            "plugins": [
+                {
+                    "id": "plugin_linear",
+                    "name": "linear",
+                    "release": {"display_name": "Linear"}
+                },
+                {
+                    "id": "plugin_github",
+                    "name": "github",
+                    "release": {"display_name": "GitHub"}
+                },
+                {
+                    "id": "plugin_slack",
+                    "name": "slack",
+                    "release": {"display_name": "Slack"}
+                }
+            ]
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+
+    let mut config = load_config(tmp.path(), tmp.path()).await;
+    config.chatgpt_base_url = server.uri();
+    let manager = PluginsManager::new(tmp.path().to_path_buf());
+    manager.write_remote_installed_plugins_cache(vec![remote_installed_plugin("linear")]);
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+    let disabled_tools = [ToolSuggestDisabledTool::plugin(
+        "github@openai-curated-remote",
+    )];
+    let loaded_plugins = manager.plugins_for_config(&config).await;
+
+    let candidates = manager
+        .recommended_plugin_candidates_for_config(RecommendedPluginCandidatesInput {
+            plugins_config: &config,
+            loaded_plugins: &loaded_plugins,
+            auth: Some(&auth),
+            disabled_tools: &disabled_tools,
+            app_server_client_name: None,
+        })
+        .await;
+
+    assert_eq!(
+        candidates,
+        Some(vec![DiscoverableTool::from(DiscoverablePluginInfo {
+            id: "slack@openai-curated-remote".to_string(),
+            remote_plugin_id: Some("plugin_slack".to_string()),
+            name: "Slack".to_string(),
+            description: None,
+            has_skills: false,
+            mcp_server_names: Vec::new(),
+            app_connector_ids: Vec::new(),
+        })])
+    );
+}
+
 #[tokio::test]
 async fn recommended_plugins_mode_caches_explicit_false() {
     let tmp = tempfile::tempdir().unwrap();