Sync local plugin imports, async remote imports, refresh caches after… (#18246)

… import

## Why

`externalAgentConfig/import` used to spawn plugin imports in the
background and return immediately. That meant local marketplace imports
could still be in flight when the caller refreshed plugin state, so
newly imported plugins would not show up right away.

This change makes local marketplace imports complete before the RPC
returns, while keeping remote marketplace imports asynchronous so we do
not block on remote fetches.

## What changed

- split plugin migration details into local and remote marketplace
imports based on the external config source
- import local marketplaces synchronously during
`externalAgentConfig/import`
- return pending remote plugin imports to the app-server so it can
finish them in the background
- clear the plugin and skills caches before responding to plugin
imports, and again after background remote imports complete, so the next
`plugin/list` reloads fresh state
- keep marketplace source parsing encapsulated behind
`is_local_marketplace_source(...)` instead of re-exporting the internal
enum
- add core and app-server coverage for the synchronous local import path
and the pending remote import path

## Verification

- `cargo test -p codex-app-server-protocol`
- `cargo test -p codex-core` (currently fails an existing unrelated
test:
`config_loader::tests::cli_override_can_update_project_local_mcp_server_when_project_is_trusted`)
- `cargo test` (currently fails existing `codex-app-server` integration
tests in MCP/skills/thread-start areas, plus the unrelated `codex-core`
failure above)

codex-rs/app-server/src/message_processor.rs

@@ -39,7 +39,10 @@ use codex_app_server_protocol::ConfigWarningNotification;
 use codex_app_server_protocol::ExperimentalApi;
 use codex_app_server_protocol::ExperimentalFeatureEnablementSetParams;
 use codex_app_server_protocol::ExternalAgentConfigDetectParams;
+use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification;
 use codex_app_server_protocol::ExternalAgentConfigImportParams;
+use codex_app_server_protocol::ExternalAgentConfigImportResponse;
+use codex_app_server_protocol::ExternalAgentConfigMigrationItemType;
 use codex_app_server_protocol::FsCopyParams;
 use codex_app_server_protocol::FsCreateDirectoryParams;
 use codex_app_server_protocol::FsGetMetadataParams;
@@ -163,6 +166,7 @@ impl ExternalAuth for ExternalAuthRefreshBridge {
 pub(crate) struct MessageProcessor {
     outgoing: Arc<OutgoingMessageSender>,
     codex_message_processor: CodexMessageProcessor,
+    thread_manager: Arc<ThreadManager>,
     config_api: ConfigApi,
     external_agent_config_api: ExternalAgentConfigApi,
     fs_api: FsApi,
@@ -311,7 +315,7 @@ impl MessageProcessor {
             runtime_feature_enablement,
             loader_overrides,
             cloud_requirements,
-            thread_manager,
+            thread_manager.clone(),
             analytics_events_client.clone(),
         );
         let external_agent_config_api =
@@ -322,6 +326,7 @@ impl MessageProcessor {
         Self {
             outgoing,
             codex_message_processor,
+            thread_manager: Arc::clone(&thread_manager),
             config_api,
             external_agent_config_api,
             fs_api,
@@ -1131,8 +1136,65 @@ impl MessageProcessor {
         request_id: ConnectionRequestId,
         params: ExternalAgentConfigImportParams,
     ) {
+        let has_plugin_imports = params.migration_items.iter().any(|item| {
+            matches!(
+                item.item_type,
+                ExternalAgentConfigMigrationItemType::Plugins
+            )
+        });
         match self.external_agent_config_api.import(params).await {
-            Ok(response) => self.outgoing.send_response(request_id, response).await,
+            Ok(pending_plugin_imports) => {
+                if has_plugin_imports {
+                    self.handle_config_mutation().await;
+                }
+                self.outgoing
+                    .send_response(request_id, ExternalAgentConfigImportResponse {})
+                    .await;
+
+                if !has_plugin_imports {
+                    return;
+                }
+
+                if pending_plugin_imports.is_empty() {
+                    self.outgoing
+                        .send_server_notification(
+                            ServerNotification::ExternalAgentConfigImportCompleted(
+                                ExternalAgentConfigImportCompletedNotification {},
+                            ),
+                        )
+                        .await;
+                    return;
+                }
+
+                let external_agent_config_api = self.external_agent_config_api.clone();
+                let outgoing = Arc::clone(&self.outgoing);
+                let thread_manager = Arc::clone(&self.thread_manager);
+                tokio::spawn(async move {
+                    for pending_plugin_import in pending_plugin_imports {
+                        match external_agent_config_api
+                            .complete_pending_plugin_import(pending_plugin_import)
+                            .await
+                        {
+                            Ok(()) => {}
+                            Err(error) => {
+                                tracing::warn!(
+                                    error = %error.message,
+                                    "external agent config plugin import failed"
+                                );
+                            }
+                        }
+                    }
+                    thread_manager.plugins_manager().clear_cache();
+                    thread_manager.skills_manager().clear_cache();
+                    outgoing
+                        .send_server_notification(
+                            ServerNotification::ExternalAgentConfigImportCompleted(
+                                ExternalAgentConfigImportCompletedNotification {},
+                            ),
+                        )
+                        .await;
+                });
+            }
             Err(error) => self.outgoing.send_error(request_id, error).await,
         }
     }