Materialize remote installed plugin bundles during sync

codex-rs/app-server/tests/common/mcp_process.rs

@@ -133,6 +133,13 @@ impl McpProcess {
         Self::new_with_env_and_args(codex_home, &[], &[]).await
     }
 
+    pub async fn new_with_env_and_plugin_startup_tasks(
+        codex_home: &Path,
+        env_overrides: &[(&str, Option<&str>)],
+    ) -> anyhow::Result<Self> {
+        Self::new_with_env_and_args(codex_home, env_overrides, &[]).await
+    }
+
     pub async fn new_with_args(codex_home: &Path, args: &[&str]) -> anyhow::Result<Self> {
         let mut all_args = vec![DISABLE_PLUGIN_STARTUP_TASKS_ARG];
         all_args.extend_from_slice(args);

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

@@ -19,6 +19,8 @@ use codex_config::types::AuthCredentialsStoreMode;
 use codex_core::config::set_project_trust_level;
 use codex_protocol::config_types::TrustLevel;
 use codex_utils_absolute_path::AbsolutePathBuf;
+use flate2::Compression;
+use flate2::write::GzEncoder;
 use pretty_assertions::assert_eq;
 use tempfile::TempDir;
 use tokio::time::timeout;
@@ -32,6 +34,8 @@ use wiremock::matchers::query_param;
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
 const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567";
+const TEST_ALLOW_HTTP_REMOTE_PLUGIN_BUNDLE_DOWNLOADS: &str =
+    "CODEX_TEST_ALLOW_HTTP_REMOTE_PLUGIN_BUNDLE_DOWNLOADS";
 const ALTERNATE_MARKETPLACE_RELATIVE_PATH: &str = ".claude-plugin/marketplace.json";
 const ALTERNATE_PLUGIN_MANIFEST_RELATIVE_PATH: &str = ".claude-plugin/plugin.json";
 
@@ -1108,6 +1112,102 @@ async fn app_server_startup_refreshes_remote_installed_cache_each_process() -> R
     Ok(())
 }
 
+#[tokio::test]
+async fn app_server_startup_downloads_missing_remote_installed_plugin_bundle() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let server = MockServer::start().await;
+    write_remote_plugin_catalog_config(
+        codex_home.path(),
+        &format!("{}/backend-api/", server.uri()),
+    )?;
+    write_chatgpt_auth(
+        codex_home.path(),
+        ChatGptAuthFixture::new("chatgpt-token")
+            .account_id("account-123")
+            .chatgpt_user_id("user-123")
+            .chatgpt_account_id("account-123"),
+        AuthCredentialsStoreMode::File,
+    )?;
+
+    let bundle_url =
+        mount_remote_plugin_bundle(&server, remote_plugin_bundle_tar_gz_bytes("linear")?).await;
+    let global_installed_body = format!(
+        r#"{{
+  "plugins": [
+    {{
+      "id": "plugins~Plugin_linear",
+      "name": "linear",
+      "scope": "GLOBAL",
+      "installation_policy": "AVAILABLE",
+      "authentication_policy": "ON_USE",
+      "release": {{
+        "version": "1.2.3",
+        "bundle_download_url": "{bundle_url}",
+        "display_name": "Linear",
+        "description": "Track work in Linear",
+        "app_ids": [],
+        "interface": {{}},
+        "skills": []
+      }},
+      "enabled": true,
+      "disabled_skill_names": []
+    }}
+  ],
+  "pagination": {{
+    "limit": 50,
+    "next_page_token": null
+  }}
+}}"#
+    );
+    let empty_page_body = r#"{
+  "plugins": [],
+  "pagination": {
+    "limit": 50,
+    "next_page_token": null
+  }
+}"#;
+    for (scope, body) in [
+        ("GLOBAL", global_installed_body.as_str()),
+        ("WORKSPACE", empty_page_body),
+    ] {
+        Mock::given(method("GET"))
+            .and(path("/backend-api/ps/plugins/installed"))
+            .and(query_param("scope", scope))
+            .and(query_param("includeDownloadUrls", "true"))
+            .and(header("authorization", "Bearer chatgpt-token"))
+            .and(header("chatgpt-account-id", "account-123"))
+            .respond_with(ResponseTemplate::new(200).set_body_string(body))
+            .mount(&server)
+            .await;
+    }
+
+    let mut mcp = McpProcess::new_with_env_and_plugin_startup_tasks(
+        codex_home.path(),
+        &[(TEST_ALLOW_HTTP_REMOTE_PLUGIN_BUNDLE_DOWNLOADS, Some("1"))],
+    )
+    .await?;
+    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
+
+    let installed_manifest = codex_home
+        .path()
+        .join("plugins/cache/chatgpt-global/linear/1.2.3/.codex-plugin/plugin.json");
+    timeout(DEFAULT_TIMEOUT, async {
+        loop {
+            if installed_manifest.is_file() {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await?;
+    assert_eq!(
+        std::fs::read_to_string(installed_manifest)?,
+        r#"{"name":"linear"}"#
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn plugin_list_includes_remote_marketplaces_when_remote_plugin_enabled() -> Result<()> {
     let codex_home = TempDir::new()?;
@@ -1667,3 +1767,42 @@ fn write_openai_curated_marketplace(
     )?;
     Ok(())
 }
+
+async fn mount_remote_plugin_bundle(server: &MockServer, body: Vec<u8>) -> String {
+    Mock::given(method("GET"))
+        .and(path("/bundles/linear.tar.gz"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "application/gzip")
+                .set_body_bytes(body),
+        )
+        .mount(server)
+        .await;
+    format!("{}/bundles/linear.tar.gz", server.uri())
+}
+
+fn remote_plugin_bundle_tar_gz_bytes(plugin_name: &str) -> Result<Vec<u8>> {
+    let manifest = format!(r#"{{"name":"{plugin_name}"}}"#);
+    let skill = "# Plan Work\n\nTrack work in Linear.\n";
+    let encoder = GzEncoder::new(Vec::new(), Compression::default());
+    let mut tar = tar::Builder::new(encoder);
+    for (path, contents, mode) in [
+        (
+            ".codex-plugin/plugin.json",
+            manifest.as_bytes(),
+            /*mode*/ 0o644,
+        ),
+        (
+            "skills/plan-work/SKILL.md",
+            skill.as_bytes(),
+            /*mode*/ 0o644,
+        ),
+    ] {
+        let mut header = tar::Header::new_gnu();
+        header.set_size(contents.len() as u64);
+        header.set_mode(mode);
+        header.set_cksum();
+        tar.append_data(&mut header, path, contents)?;
+    }
+    Ok(tar.into_inner()?.finish()?)
+}

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

@@ -58,10 +58,12 @@ async fn list_tool_suggest_discoverable_plugins_returns_microsoft_curated_plugin
     );
     write_plugins_feature_config(codex_home.path());
 
-    let config = load_plugins_config(codex_home.path()).await;
-    let discoverable_plugins = list_tool_suggest_discoverable_plugins(&config)
-        .await
-        .unwrap();
+    let discoverable_plugins = list_tool_suggest_discoverable_plugins_for_test(
+        codex_home.path(),
+        /*plugins_enabled*/ true,
+    )
+    .await
+    .unwrap();
 
     assert_eq!(
         discoverable_plugins

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

@@ -170,9 +170,8 @@ pub fn remote_installed_plugins_to_config(
                         return None;
                     }
                 };
-            // TODO(remote plugins): download or update missing local bundles during remote
-            // installed reconciliation. Until then, only publish remote installed state for
-            // bundles already present in the local plugin cache.
+            // Remote installed refresh materializes bundles before publishing state. Keep this
+            // check so partial refresh failures do not make missing local bundles effective.
             store.active_plugin_root(&plugin_id)?;
             Some((
                 plugin_id.as_key(),

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

@@ -35,6 +35,8 @@ use crate::marketplace_upgrade::upgrade_configured_git_marketplaces;
 use crate::remote::RemoteInstalledPlugin;
 use crate::remote::RemotePluginCatalogError;
 use crate::remote::RemotePluginServiceConfig;
+use crate::remote_bundle::download_and_install_remote_plugin_bundle;
+use crate::remote_bundle::validate_remote_plugin_bundle;
 use crate::remote_legacy::RemotePluginFetchError;
 use crate::remote_legacy::RemotePluginMutationError;
 use crate::remote_startup_sync::RemoteStartupPluginSyncRequest;
@@ -1376,9 +1378,62 @@ impl PluginsManager {
     ) -> Result<bool, RemotePluginCatalogError> {
         let installed_plugins =
             crate::remote::fetch_remote_installed_plugins(service_config, auth).await?;
-        // TODO(remote plugins): reconcile missing or stale local bundles before
-        // publishing remote installed state as effective local plugin config.
-        Ok(self.write_remote_installed_plugins_cache(installed_plugins))
+        let mut bundles_changed = false;
+        for plugin in &installed_plugins {
+            let validated_bundle = match validate_remote_plugin_bundle(
+                &plugin.id,
+                &plugin.marketplace_name,
+                &plugin.name,
+                plugin.release_version.as_deref(),
+                plugin.bundle_download_url.as_deref(),
+            ) {
+                Ok(bundle) => bundle,
+                Err(err) => {
+                    warn!(
+                        remote_plugin_id = %plugin.id,
+                        marketplace = %plugin.marketplace_name,
+                        plugin = %plugin.name,
+                        error = %err,
+                        "failed to validate remote installed plugin bundle"
+                    );
+                    continue;
+                }
+            };
+            if self
+                .store
+                .active_plugin_version(&validated_bundle.plugin_id)
+                .as_deref()
+                == Some(validated_bundle.plugin_version.as_str())
+            {
+                continue;
+            }
+            match download_and_install_remote_plugin_bundle(
+                self.codex_home.clone(),
+                validated_bundle,
+            )
+            .await
+            {
+                Ok(result) => {
+                    bundles_changed = true;
+                    tracing::info!(
+                        plugin = %result.plugin_id.as_key(),
+                        version = %result.plugin_version,
+                        path = %result.installed_path.display(),
+                        "installed remote plugin bundle during installed-plugin refresh"
+                    );
+                }
+                Err(err) => {
+                    warn!(
+                        remote_plugin_id = %plugin.id,
+                        marketplace = %plugin.marketplace_name,
+                        plugin = %plugin.name,
+                        error = %err,
+                        "failed to install remote plugin bundle during installed-plugin refresh"
+                    );
+                }
+            }
+        }
+        Ok(self.write_remote_installed_plugins_cache(installed_plugins) || bundles_changed)
     }
 
     fn run_non_curated_plugin_cache_refresh_loop(self: Arc<Self>) {

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

@@ -581,6 +581,8 @@ remote_plugin = true
         id: "plugins~Plugin_linear".to_string(),
         name: "linear".to_string(),
         enabled: true,
+        release_version: Some("local".to_string()),
+        bundle_download_url: Some("https://example.com/linear.tar.gz".to_string()),
     }]);
 
     let outcome = manager.plugins_for_test_config(&config).await;
@@ -610,6 +612,8 @@ remote_plugin = true
         id: "plugins~Plugin_linear".to_string(),
         name: "linear".to_string(),
         enabled: true,
+        release_version: Some("local".to_string()),
+        bundle_download_url: Some("https://example.com/linear.tar.gz".to_string()),
     }]);
 
     let outcome = manager.plugins_for_test_config(&config).await;
@@ -638,6 +642,7 @@ async fn mount_remote_installed_plugin_pages(
     Mock::given(method("GET"))
         .and(path("/backend-api/ps/plugins/installed"))
         .and(query_param("scope", "GLOBAL"))
+        .and(query_param("includeDownloadUrls", "true"))
         .and(header("authorization", "Bearer Access Token"))
         .and(header("chatgpt-account-id", "account_id"))
         .respond_with(ResponseTemplate::new(200).set_body_string(body_for_plugins(global_plugins)))
@@ -646,6 +651,7 @@ async fn mount_remote_installed_plugin_pages(
     Mock::given(method("GET"))
         .and(path("/backend-api/ps/plugins/installed"))
         .and(query_param("scope", "WORKSPACE"))
+        .and(query_param("includeDownloadUrls", "true"))
         .and(header("authorization", "Bearer Access Token"))
         .and(header("chatgpt-account-id", "account_id"))
         .respond_with(
@@ -664,6 +670,8 @@ fn remote_installed_plugin_json(plugin_name: &str, enabled: bool) -> String {
       "installation_policy": "AVAILABLE",
       "authentication_policy": "ON_USE",
       "release": {{
+        "version": "local",
+        "bundle_download_url": "https://example.com/{plugin_name}.tar.gz",
         "display_name": "{plugin_name}",
         "description": "{plugin_name} plugin",
         "app_ids": [],
@@ -809,6 +817,8 @@ async fn remote_installed_plugins_cache_refresh_clears_stale_cache_when_auth_is_
         id: "plugins~Plugin_linear".to_string(),
         name: "linear".to_string(),
         enabled: true,
+        release_version: Some("local".to_string()),
+        bundle_download_url: Some("https://example.com/linear.tar.gz".to_string()),
     }]);
     let notification_count = Arc::new(AtomicUsize::new(0));
     let notification_count_for_callback = Arc::clone(&notification_count);

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

@@ -50,6 +50,8 @@ pub struct RemoteInstalledPlugin {
     pub id: String,
     pub name: String,
     pub enabled: bool,
+    pub release_version: Option<String>,
+    pub bundle_download_url: Option<String>,
 }
 
 #[derive(Debug, Clone, PartialEq)]
@@ -671,6 +673,8 @@ fn remote_installed_plugin_to_info(
         id: plugin.id.clone(),
         name: plugin.name.clone(),
         enabled: installed_plugin.enabled,
+        release_version: plugin.release.version.clone(),
+        bundle_download_url: plugin.release.bundle_download_url.clone(),
     }
 }
 
@@ -827,6 +831,7 @@ async fn get_remote_plugin_installed_page(
     let client = build_reqwest_client();
     let mut request = authenticated_request(client.get(&url), auth)?;
     request = request.query(&[("scope", scope.api_value())]);
+    request = request.query(&[("includeDownloadUrls", true)]);
     if let Some(page_token) = page_token {
         request = request.query(&[("pageToken", page_token)]);
     }

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

@@ -56,6 +56,8 @@ async fn mount_installed_plugins(server: &MockServer) {
       "installation_policy": "AVAILABLE",
       "authentication_policy": "ON_USE",
       "release": {
+        "version": "local",
+        "bundle_download_url": "https://example.com/linear.tar.gz",
         "display_name": "Linear",
         "description": "Track work in Linear",
         "app_ids": [],
@@ -78,6 +80,7 @@ async fn mount_installed_plugins(server: &MockServer) {
     Mock::given(method("GET"))
         .and(path("/backend-api/ps/plugins/installed"))
         .and(query_param("scope", "GLOBAL"))
+        .and(query_param("includeDownloadUrls", "true"))
         .and(header("authorization", "Bearer Access Token"))
         .and(header("chatgpt-account-id", "account_id"))
         .respond_with(ResponseTemplate::new(200).set_body_string(global_installed_body))
@@ -86,6 +89,7 @@ async fn mount_installed_plugins(server: &MockServer) {
     Mock::given(method("GET"))
         .and(path("/backend-api/ps/plugins/installed"))
         .and(query_param("scope", "WORKSPACE"))
+        .and(query_param("includeDownloadUrls", "true"))
         .and(header("authorization", "Bearer Access Token"))
         .and(header("chatgpt-account-id", "account_id"))
         .respond_with(ResponseTemplate::new(200).set_body_string(empty_page_body))