[codex] [3/4] Activate endpoint plugin recommendations

codex-rs/app-server/tests/suite/v2/mod.rs

@@ -44,6 +44,7 @@ mod process_exec;
 mod rate_limit_reset_credits;
 mod rate_limits;
 mod realtime_conversation;
+mod recommended_plugins;
 mod remote_control;
 #[cfg(debug_assertions)]
 mod remote_thread_store;

codex-rs/app-server/tests/suite/v2/recommended_plugins.rs

@@ -0,0 +1,165 @@
+use anyhow::Result;
+use app_test_support::ChatGptIdTokenClaims;
+use app_test_support::TestAppServer;
+use app_test_support::encode_id_token;
+use app_test_support::to_response;
+use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::LoginAccountResponse;
+use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnStartParams;
+use codex_app_server_protocol::UserInput;
+use core_test_support::apps_test_server::AppsTestServer;
+use core_test_support::responses;
+use serde_json::Value;
+use serde_json::json;
+use std::time::Duration;
+use tempfile::TempDir;
+use tokio::time::timeout;
+use wiremock::Mock;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
+use wiremock::matchers::query_param;
+
+const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(20);
+const WORKSPACE_ID: &str = "123e4567-e89b-42d3-a456-426614174010";
+
+#[tokio::test]
+async fn first_turn_after_external_login_waits_for_recommended_plugins() -> Result<()> {
+    let server = responses::start_mock_server().await;
+    let apps_server = AppsTestServer::mount(&server).await?;
+    Mock::given(method("GET"))
+        .and(path("/ps/plugins/suggested"))
+        .and(query_param("scope", "GLOBAL"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_delay(Duration::from_millis(250))
+                .set_body_json(json!({
+                    "enabled": true,
+                    "plugins": [{
+                        "id": "plugin_github",
+                        "name": "github",
+                        "status": "ENABLED",
+                        "installation_policy": "AVAILABLE",
+                        "release": {"display_name": "GitHub"}
+                    }]
+                })),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+    let response = responses::sse(vec![
+        responses::ev_response_created("resp-1"),
+        responses::ev_assistant_message("msg-1", "done"),
+        responses::ev_completed("resp-1"),
+    ]);
+    let responses_mock = responses::mount_sse_once(&server, response).await;
+
+    let codex_home = TempDir::new()?;
+    write_mock_responses_config_toml_with_chatgpt_base_url(
+        codex_home.path(),
+        &server.uri(),
+        &apps_server.chatgpt_base_url,
+    )?;
+    let config_path = codex_home.path().join("config.toml");
+    let config = std::fs::read_to_string(&config_path)?;
+    std::fs::write(
+        config_path,
+        format!(
+            "{config}\n[features]\napps = true\nplugins = true\nremote_plugin = true\ntool_suggest = true\n"
+        ),
+    )?;
+
+    let sqlite_home = codex_home.path().to_string_lossy();
+    let mut app_server = TestAppServer::new_without_managed_config_with_env(
+        codex_home.path(),
+        &[("CODEX_SQLITE_HOME", Some(sqlite_home.as_ref()))],
+    )
+    .await?;
+    timeout(DEFAULT_READ_TIMEOUT, app_server.initialize()).await??;
+
+    let access_token = encode_id_token(
+        &ChatGptIdTokenClaims::new()
+            .email("embedded@example.com")
+            .plan_type("pro")
+            .chatgpt_account_id(WORKSPACE_ID),
+    )?;
+    let login_id = app_server
+        .send_chatgpt_auth_tokens_login_request(
+            access_token,
+            WORKSPACE_ID.to_string(),
+            Some("pro".to_string()),
+        )
+        .await?;
+    let login_response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        app_server.read_stream_until_response_message(RequestId::Integer(login_id)),
+    )
+    .await??;
+    assert_eq!(
+        to_response::<LoginAccountResponse>(login_response)?,
+        LoginAccountResponse::ChatgptAuthTokens {}
+    );
+
+    let thread_id = app_server
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let thread_response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        app_server.read_stream_until_response_message(RequestId::Integer(thread_id)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response(thread_response)?;
+
+    let turn_id = app_server
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id,
+            input: vec![UserInput::Text {
+                text: "suggest a plugin".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    let _: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        app_server.read_stream_until_response_message(RequestId::Integer(turn_id)),
+    )
+    .await??;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        app_server.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+
+    let requests = responses_mock.requests();
+    let request = requests
+        .iter()
+        .find(|request| {
+            request
+                .message_input_texts("user")
+                .iter()
+                .any(|text| text.contains("suggest a plugin"))
+        })
+        .expect("turn request");
+    let contextual_user_message = request.message_input_texts("user").join("\n");
+    assert!(contextual_user_message.contains("<recommended_plugins>"));
+    assert!(contextual_user_message.contains("- GitHub (github@openai-curated-remote)"));
+    let body = request.body_json();
+    let tool_names = body
+        .get("tools")
+        .and_then(Value::as_array)
+        .into_iter()
+        .flatten()
+        .filter_map(|tool| tool.get("name").and_then(Value::as_str))
+        .collect::<Vec<_>>();
+    assert!(tool_names.contains(&"request_plugin_install"));
+    assert!(!tool_names.contains(&"list_available_plugins_to_install"));
+    Ok(())
+}

codex-rs/core-plugins/Cargo.toml

@@ -27,6 +27,7 @@ codex-model-provider = { workspace = true }
 codex-otel = { workspace = true }
 codex-plugin = { workspace = true }
 codex-protocol = { workspace = true }
+codex-tools = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-path-uri = { workspace = true }
 codex-utils-plugins = { workspace = true }

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

@@ -49,6 +49,7 @@ pub use manager::PluginReadRequest;
 pub use manager::PluginUninstallError;
 pub use manager::PluginsConfigInput;
 pub use manager::PluginsManager;
+pub use manager::RecommendedPluginCandidatesInput;
 pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeError as PluginMarketplaceUpgradeError;
 pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome as PluginMarketplaceUpgradeOutcome;
 pub use provider::ExecutorPluginProvider;

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

@@ -57,6 +57,8 @@ use codex_config::ConfigLayerStack;
 use codex_config::clear_user_plugin;
 use codex_config::set_user_plugin_enabled;
 use codex_config::types::PluginConfig;
+use codex_config::types::ToolSuggestDisabledTool;
+use codex_config::types::ToolSuggestDiscoverableType;
 use codex_core_skills::SkillMetadata;
 use codex_core_skills::config_rules::SkillConfigRules;
 use codex_core_skills::config_rules::skill_config_rules_from_stack;
@@ -71,6 +73,9 @@ use codex_plugin::app_connector_ids_from_declarations;
 use codex_plugin::prompt_safe_plugin_description;
 use codex_protocol::protocol::HookEventName;
 use codex_protocol::protocol::Product;
+use codex_tools::DiscoverablePluginInfo;
+use codex_tools::DiscoverableTool;
+use codex_tools::filter_request_plugin_install_discoverable_tools_for_client;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_plugins::PluginSkillRoot;
 use std::collections::HashMap;
@@ -114,6 +119,15 @@ impl PluginsConfigInput {
     }
 }
 
+/// Inputs used to select endpoint-backed plugin install candidates.
+pub struct RecommendedPluginCandidatesInput<'a> {
+    pub plugins_config: &'a PluginsConfigInput,
+    pub loaded_plugins: &'a PluginLoadOutcome,
+    pub auth: Option<&'a CodexAuth>,
+    pub disabled_tools: &'a [ToolSuggestDisabledTool],
+    pub app_server_client_name: Option<&'a str>,
+}
+
 #[derive(Clone, PartialEq, Eq)]
 struct FeaturedPluginIdsCacheKey {
     chatgpt_base_url: String,
@@ -997,6 +1011,59 @@ impl PluginsManager {
         mode
     }
 
+    /// Returns endpoint recommendations eligible for installation in the current client.
+    /// `None` selects the legacy discovery workflow.
+    pub async fn recommended_plugin_candidates_for_config(
+        &self,
+        input: RecommendedPluginCandidatesInput<'_>,
+    ) -> Option<Vec<DiscoverableTool>> {
+        let RecommendedPluginsMode::Endpoint { plugins } = self
+            .recommended_plugins_mode_for_config(input.plugins_config, input.auth)
+            .await
+        else {
+            return None;
+        };
+        if plugins.is_empty() {
+            return Some(Vec::new());
+        }
+
+        let installed_plugin_ids = input
+            .loaded_plugins
+            .plugins()
+            .iter()
+            .map(|plugin| plugin.config_name.as_str())
+            .collect::<HashSet<_>>();
+        let disabled_plugin_ids = input
+            .disabled_tools
+            .iter()
+            .filter(|tool| tool.kind == ToolSuggestDiscoverableType::Plugin)
+            .map(|tool| tool.id.as_str())
+            .collect::<HashSet<_>>();
+
+        let candidates = plugins
+            .into_iter()
+            .filter(|plugin| {
+                !installed_plugin_ids.contains(plugin.config_id.as_str())
+                    && !disabled_plugin_ids.contains(plugin.config_id.as_str())
+            })
+            .map(|plugin| {
+                DiscoverableTool::from(DiscoverablePluginInfo {
+                    id: plugin.config_id,
+                    remote_plugin_id: Some(plugin.remote_plugin_id),
+                    name: plugin.display_name,
+                    description: None,
+                    has_skills: false,
+                    mcp_server_names: Vec::new(),
+                    app_connector_ids: plugin.app_connector_ids,
+                })
+            })
+            .collect();
+        Some(filter_request_plugin_install_discoverable_tools_for_client(
+            candidates,
+            input.app_server_client_name,
+        ))
+    }
+
     fn cached_recommended_plugins_mode(
         &self,
         cache_key: &RecommendedPluginsCacheKey,

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

codex-rs/core/src/context/contextual_user_message.rs

@@ -10,6 +10,7 @@ use super::InternalModelContextFragment;
 use super::LegacyApplyPatchExecCommandWarning;
 use super::LegacyModelMismatchWarning;
 use super::LegacyUnifiedExecProcessLimitWarning;
+use super::RecommendedPluginsInstructions;
 use super::SkillInstructions;
 use super::SubagentNotification;
 use super::TurnAborted;
@@ -33,6 +34,8 @@ static SUBAGENT_NOTIFICATION_REGISTRATION: FragmentRegistrationProxy<SubagentNot
 static INTERNAL_MODEL_CONTEXT_REGISTRATION: FragmentRegistrationProxy<
     InternalModelContextFragment,
 > = FragmentRegistrationProxy::new();
+static RECOMMENDED_PLUGINS_REGISTRATION: FragmentRegistrationProxy<RecommendedPluginsInstructions> =
+    FragmentRegistrationProxy::new();
 static LEGACY_UNIFIED_EXEC_PROCESS_LIMIT_WARNING_REGISTRATION: FragmentRegistrationProxy<
     LegacyUnifiedExecProcessLimitWarning,
 > = FragmentRegistrationProxy::new();
@@ -52,6 +55,7 @@ static CONTEXTUAL_USER_FRAGMENTS: &[&dyn FragmentRegistration] = &[
     &TURN_ABORTED_REGISTRATION,
     &SUBAGENT_NOTIFICATION_REGISTRATION,
     &INTERNAL_MODEL_CONTEXT_REGISTRATION,
+    &RECOMMENDED_PLUGINS_REGISTRATION,
     &LEGACY_UNIFIED_EXEC_PROCESS_LIMIT_WARNING_REGISTRATION,
     &LEGACY_APPLY_PATCH_EXEC_COMMAND_WARNING_REGISTRATION,
     &LEGACY_MODEL_MISMATCH_WARNING_REGISTRATION,