From 076a110eb83445fd33fcd40f9a70f3ad84120f76 Mon Sep 17 00:00:00 2001 From: Abhinav Date: Sat, 11 Jul 2026 01:07:12 +0000 Subject: [PATCH] Trust hooks from materialized workspace plugins (#32301) ## What changed - Carry newly installed or updated remote plugin metadata through effective-plugin refresh callbacks, including coalesced refreshes. - After a successful refresh, record the current hook hashes for listed workspace plugins associated with the active account while preserving existing hook settings and unrelated state. - Serialize the background trust write with config mutations, and leave hooks untrusted if the write fails or the active account changes. ## Testing - Cover eligible plugin selection and escaped hook config keys. - Verify end-to-end trust for newly materialized plugin hooks, preservation of existing config, and fail-closed behavior when config cannot be written. GitOrigin-RevId: 0b150766415be6fccc117f7856a241e10ad456f2 --- .../app-server/src/effective_plugin_change.rs | 181 +++++++++ .../src/effective_plugin_change_tests.rs | 63 +++ codex-rs/app-server/src/lib.rs | 1 + codex-rs/app-server/src/message_processor.rs | 24 +- .../request_processors/account_processor.rs | 2 +- .../src/request_processors/plugins.rs | 38 +- .../app-server/src/request_serialization.rs | 31 +- .../app-server/tests/suite/v2/plugin_list.rs | 375 +++++++++++++++--- codex-rs/core-plugins/src/lib.rs | 1 + codex-rs/core-plugins/src/manager.rs | 89 ++++- codex-rs/core-plugins/src/manager_tests.rs | 76 ++++ codex-rs/core-plugins/src/remote.rs | 1 + .../remote/remote_installed_plugin_sync.rs | 72 +++- 13 files changed, 815 insertions(+), 139 deletions(-) create mode 100644 codex-rs/app-server/src/effective_plugin_change.rs create mode 100644 codex-rs/app-server/src/effective_plugin_change_tests.rs diff --git a/codex-rs/app-server/src/effective_plugin_change.rs b/codex-rs/app-server/src/effective_plugin_change.rs new file mode 100644 index 0000000000..c23572a103 --- /dev/null +++ b/codex-rs/app-server/src/effective_plugin_change.rs @@ -0,0 +1,181 @@ +use std::collections::BTreeSet; +use std::sync::Arc; + +use codex_app_server_protocol::ConfigBatchWriteParams; +use codex_app_server_protocol::ConfigEdit; +use codex_app_server_protocol::MergeStrategy; +use codex_core::ThreadManager; +use codex_core_plugins::EffectivePluginsChange; +use codex_core_plugins::remote::RemotePluginMaterialization; +use codex_core_plugins::remote::RemotePluginScope; +use codex_core_plugins::remote::RemotePluginShareDiscoverability; +use codex_login::AuthManager; +use serde_json::json; +use tracing::warn; + +use crate::config_manager::ConfigManager; +use crate::request_processors::ConfigRequestProcessor; +use crate::request_serialization::RequestSerializationAccess; +use crate::request_serialization::RequestSerializationQueueKey; +use crate::request_serialization::RequestSerializationQueues; + +/// Refresh plugin consumers and trust hooks from newly materialized Workspace + Listed bundles. +pub(crate) fn effective_plugins_changed_callback( + auth_manager: Arc, + thread_manager: Arc, + config_manager: ConfigManager, + config_processor: ConfigRequestProcessor, + request_serialization_queues: RequestSerializationQueues, +) -> Arc { + Arc::new(move |change| { + thread_manager.plugins_manager().clear_cache(); + thread_manager.skills_service().clear_cache(); + + let refresh_thread_manager = Arc::clone(&thread_manager); + let refresh_config_manager = config_manager.clone(); + tokio::spawn(async move { + if refresh_thread_manager.list_thread_ids().await.is_empty() { + return; + } + crate::mcp_refresh::queue_best_effort_refresh( + &refresh_thread_manager, + &refresh_config_manager, + ) + .await; + }); + + if change.materialized_remote_plugins.is_empty() { + return; + } + + let trust_auth_manager = Arc::clone(&auth_manager); + let trust_thread_manager = Arc::clone(&thread_manager); + let trust_config_manager = config_manager.clone(); + let trust_config_processor = config_processor.clone(); + let trust_request_serialization_queues = request_serialization_queues.clone(); + tokio::spawn(async move { + trust_request_serialization_queues + .enqueue_background( + RequestSerializationQueueKey::Global("config"), + RequestSerializationAccess::Exclusive, + async move { + if let Err(err) = trust_materialized_plugin_hooks( + change.materialized_remote_plugins, + &trust_auth_manager, + &trust_thread_manager, + &trust_config_manager, + &trust_config_processor, + ) + .await + { + warn!(error = %err, "failed to trust materialized plugin hooks"); + } + }, + ) + .await; + }); + }) +} + +fn workspace_listed_plugin_ids( + materializations: Vec, + current_account_id: &str, +) -> BTreeSet { + materializations + .into_iter() + .filter(|plugin| { + plugin.scope == RemotePluginScope::Workspace + && plugin.discoverability == Some(RemotePluginShareDiscoverability::Listed) + && plugin.authenticated_account_id.as_deref() == Some(current_account_id) + }) + .map(|plugin| plugin.plugin_id.as_key()) + .collect() +} + +fn hook_trusted_hash_edit(hook_key: &str, current_hash: &str) -> ConfigEdit { + let escaped_hook_key = hook_key.replace('\\', "\\\\").replace('"', "\\\""); + ConfigEdit { + key_path: format!(r#"hooks.state."{escaped_hook_key}".trusted_hash"#), + value: json!(current_hash), + merge_strategy: MergeStrategy::Replace, + } +} + +async fn trust_materialized_plugin_hooks( + materializations: Vec, + auth_manager: &AuthManager, + thread_manager: &ThreadManager, + config_manager: &ConfigManager, + config_processor: &ConfigRequestProcessor, +) -> Result<(), String> { + let Some(current_account_id) = auth_manager + .auth_cached() + .and_then(|auth| auth.get_account_id()) + else { + return Ok(()); + }; + let plugin_ids = workspace_listed_plugin_ids(materializations, ¤t_account_id); + if plugin_ids.is_empty() { + return Ok(()); + } + let config = config_manager + .load_latest_config(/*fallback_cwd*/ None) + .await + .map_err(|err| format!("failed to reload config: {err}"))?; + let plugin_outcome = thread_manager + .plugins_manager() + .plugins_for_config(&config.plugins_config_input()) + .await; + let hooks = codex_hooks::list_hooks(codex_hooks::HooksConfig { + feature_enabled: true, + bypass_hook_trust: config.bypass_hook_trust, + config_layer_stack: Some(config.config_layer_stack), + plugin_hook_sources: plugin_outcome.effective_plugin_hook_sources(), + plugin_hook_load_warnings: plugin_outcome.effective_plugin_hook_warnings(), + ..Default::default() + }); + if !hooks.warnings.is_empty() { + warn!( + warnings = ?hooks.warnings, + "hook discovery reported warnings while trusting materialized plugins" + ); + } + let edits = hooks + .hooks + .into_iter() + .filter(|hook| { + hook.plugin_id + .as_ref() + .is_some_and(|plugin_id| plugin_ids.contains(plugin_id)) + }) + .map(|hook| hook_trusted_hash_edit(&hook.key, &hook.current_hash)) + .collect::>(); + if edits.is_empty() { + return Ok(()); + } + if auth_manager + .auth_cached() + .and_then(|auth| auth.get_account_id()) + .as_deref() + != Some(current_account_id.as_str()) + { + warn!("skipping materialized plugin hook trust after account changed"); + return Ok(()); + } + + let params = ConfigBatchWriteParams { + edits, + file_path: None, + expected_version: None, + reload_user_config: true, + }; + config_processor + .batch_write(params) + .await + .map_err(|err| format!("failed to write hook trust: {}", err.message))?; + Ok(()) +} + +#[cfg(test)] +#[path = "effective_plugin_change_tests.rs"] +mod tests; diff --git a/codex-rs/app-server/src/effective_plugin_change_tests.rs b/codex-rs/app-server/src/effective_plugin_change_tests.rs new file mode 100644 index 0000000000..4ea5e9edf0 --- /dev/null +++ b/codex-rs/app-server/src/effective_plugin_change_tests.rs @@ -0,0 +1,63 @@ +use super::*; +use codex_plugin::PluginId; +use pretty_assertions::assert_eq; + +#[test] +fn only_workspace_listed_materializations_are_eligible() { + let materialization = + |name: &str, + scope: RemotePluginScope, + discoverability: Option| { + RemotePluginMaterialization { + plugin_id: PluginId::new(name.to_string(), "test".to_string()) + .expect("valid plugin id"), + scope, + discoverability, + authenticated_account_id: Some("account-123".to_string()), + } + }; + + let mut materializations = vec![ + materialization( + "eligible", + RemotePluginScope::Workspace, + Some(RemotePluginShareDiscoverability::Listed), + ), + materialization( + "unlisted", + RemotePluginScope::Workspace, + Some(RemotePluginShareDiscoverability::Unlisted), + ), + materialization( + "private", + RemotePluginScope::Workspace, + Some(RemotePluginShareDiscoverability::Private), + ), + materialization("workspace-missing", RemotePluginScope::Workspace, None), + materialization("global", RemotePluginScope::Global, None), + materialization("user", RemotePluginScope::User, None), + ]; + let mut wrong_account = materialization( + "wrong-account", + RemotePluginScope::Workspace, + Some(RemotePluginShareDiscoverability::Listed), + ); + wrong_account.authenticated_account_id = Some("account-456".to_string()); + materializations.push(wrong_account); + + let plugin_ids = workspace_listed_plugin_ids(materializations, "account-123"); + + assert_eq!(plugin_ids, BTreeSet::from(["eligible@test".to_string()])); +} + +#[test] +fn hook_trusted_hash_edit_targets_only_escaped_leaf() { + assert_eq!( + hook_trusted_hash_edit(r#"plugin."quoted"\path"#, "sha256:current"), + ConfigEdit { + key_path: r#"hooks.state."plugin.\"quoted\"\\path".trusted_hash"#.to_string(), + value: serde_json::json!("sha256:current"), + merge_strategy: MergeStrategy::Replace, + } + ); +} diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 8bb24c5ca1..9b7b26406a 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -94,6 +94,7 @@ mod connection_cleanup; mod connection_rpc_gate; mod current_time; mod dynamic_tools; +mod effective_plugin_change; mod error_code; mod extensions; mod external_auth; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 62eff72660..8cbfe4b69a 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -313,6 +313,21 @@ impl MessageProcessor { let workspace_settings_cache = Arc::new(workspace_settings::WorkspaceSettingsCache::default()); let app_list_shutdown_token = CancellationToken::new(); + let request_serialization_queues = RequestSerializationQueues::default(); + let config_processor = ConfigRequestProcessor::new( + outgoing.clone(), + config_manager.clone(), + thread_manager.clone(), + analytics_events_client.clone(), + ); + let on_effective_plugins_changed = + crate::effective_plugin_change::effective_plugins_changed_callback( + auth_manager.clone(), + Arc::clone(&thread_manager), + config_manager.clone(), + config_processor.clone(), + request_serialization_queues.clone(), + ); let account_processor = AccountRequestProcessor::new( auth_manager.clone(), Arc::clone(&thread_manager), @@ -382,6 +397,7 @@ impl MessageProcessor { analytics_events_client.clone(), config_manager.clone(), workspace_settings_cache, + on_effective_plugins_changed, ); let remote_control_processor = RemoteControlRequestProcessor::new(remote_control_handle); let search_processor = SearchRequestProcessor::new(outgoing.clone()); @@ -437,12 +453,6 @@ impl MessageProcessor { Some(on_effective_plugins_changed), ); } - let config_processor = ConfigRequestProcessor::new( - outgoing.clone(), - config_manager.clone(), - thread_manager.clone(), - analytics_events_client.clone(), - ); let external_agent_config_processor = ExternalAgentConfigRequestProcessor::new(ExternalAgentConfigRequestProcessorArgs { outgoing: outgoing.clone(), @@ -492,7 +502,7 @@ impl MessageProcessor { thread_processor, turn_processor, windows_sandbox_processor, - request_serialization_queues: RequestSerializationQueues::default(), + request_serialization_queues, } } diff --git a/codex-rs/app-server/src/request_processors/account_processor.rs b/codex-rs/app-server/src/request_processors/account_processor.rs index d98977ce58..93b1f3f10d 100644 --- a/codex-rs/app-server/src/request_processors/account_processor.rs +++ b/codex-rs/app-server/src/request_processors/account_processor.rs @@ -215,7 +215,7 @@ impl AccountRequestProcessor { .maybe_start_remote_plugin_caches_refresh( &config.plugins_config_input(), auth, - Some(Arc::new(move || { + Some(Arc::new(move |_change| { Self::spawn_effective_plugins_changed_task( Arc::clone(&refresh_thread_manager), refresh_config_manager.clone(), diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 507b63f0d0..5cf388de6b 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -36,6 +36,8 @@ pub(crate) struct PluginRequestProcessor { analytics_events_client: AnalyticsEventsClient, config_manager: ConfigManager, workspace_settings_cache: Arc, + on_effective_plugins_changed: + Arc, } fn plugin_skills_to_info( @@ -358,6 +360,9 @@ impl PluginRequestProcessor { analytics_events_client: AnalyticsEventsClient, config_manager: ConfigManager, workspace_settings_cache: Arc, + on_effective_plugins_changed: Arc< + dyn Fn(codex_core_plugins::EffectivePluginsChange) + Send + Sync, + >, ) -> Self { Self { auth_manager, @@ -366,6 +371,7 @@ impl PluginRequestProcessor { analytics_events_client, config_manager, workspace_settings_cache, + on_effective_plugins_changed, } } @@ -468,36 +474,14 @@ impl PluginRequestProcessor { .map(|response| Some(response.into())) } - pub(crate) fn effective_plugins_changed_callback(&self) -> Arc { - let thread_manager = Arc::clone(&self.thread_manager); - let config_manager = self.config_manager.clone(); - Arc::new(move || { - Self::spawn_effective_plugins_changed_task( - Arc::clone(&thread_manager), - config_manager.clone(), - ); - }) + pub(crate) fn effective_plugins_changed_callback( + &self, + ) -> Arc { + Arc::clone(&self.on_effective_plugins_changed) } fn on_effective_plugins_changed(&self) { - Self::spawn_effective_plugins_changed_task( - Arc::clone(&self.thread_manager), - self.config_manager.clone(), - ); - } - - fn spawn_effective_plugins_changed_task( - thread_manager: Arc, - config_manager: ConfigManager, - ) { - tokio::spawn(async move { - thread_manager.plugins_manager().clear_cache(); - thread_manager.skills_service().clear_cache(); - if thread_manager.list_thread_ids().await.is_empty() { - return; - } - crate::mcp_refresh::queue_best_effort_refresh(&thread_manager, &config_manager).await; - }); + (self.on_effective_plugins_changed)(Default::default()); } fn clear_plugin_related_caches(&self) { diff --git a/codex-rs/app-server/src/request_serialization.rs b/codex-rs/app-server/src/request_serialization.rs index 77ecfc8f56..e3038bbc30 100644 --- a/codex-rs/app-server/src/request_serialization.rs +++ b/codex-rs/app-server/src/request_serialization.rs @@ -104,7 +104,7 @@ impl RequestSerializationQueueKey { } pub(crate) struct QueuedInitializedRequest { - gate: Arc, + gate: Option>, future: BoxFutureUnit, } @@ -114,14 +114,24 @@ impl QueuedInitializedRequest { future: impl Future + Send + 'static, ) -> Self { Self { - gate, + gate: Some(gate), + future: Box::pin(future), + } + } + + fn new_background(future: impl Future + Send + 'static) -> Self { + Self { + gate: None, future: Box::pin(future), } } pub(crate) async fn run(self) { let Self { gate, future } = self; - gate.run(future).await; + match gate { + Some(gate) => gate.run(future).await, + None => future.await, + } } } @@ -136,6 +146,21 @@ pub(crate) struct RequestSerializationQueues { } impl RequestSerializationQueues { + /// Enqueue app-owned work alongside RPCs that mutate the same serialized resource. + pub(crate) async fn enqueue_background( + &self, + key: RequestSerializationQueueKey, + access: RequestSerializationAccess, + future: impl Future + Send + 'static, + ) { + self.enqueue( + key, + access, + QueuedInitializedRequest::new_background(future), + ) + .await; + } + pub(crate) async fn enqueue( &self, key: RequestSerializationQueueKey, diff --git a/codex-rs/app-server/tests/suite/v2/plugin_list.rs b/codex-rs/app-server/tests/suite/v2/plugin_list.rs index e42240ec65..28dea3c10a 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_list.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_list.rs @@ -6,6 +6,10 @@ use app_test_support::ChatGptAuthFixture; use app_test_support::TestAppServer; use app_test_support::to_response; use app_test_support::write_chatgpt_auth; +use codex_app_server_protocol::HookMetadata; +use codex_app_server_protocol::HookTrustStatus; +use codex_app_server_protocol::HooksListParams; +use codex_app_server_protocol::HooksListResponse; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::PluginAuthPolicy; use codex_app_server_protocol::PluginInstallPolicy; @@ -30,6 +34,7 @@ use flate2::Compression; use flate2::write::GzEncoder; use pretty_assertions::assert_eq; use tempfile::TempDir; +use tokio::time::sleep; use tokio::time::timeout; use wiremock::Mock; use wiremock::MockServer; @@ -1565,7 +1570,7 @@ async fn app_server_startup_sync_downloads_remote_installed_plugin_bundles() -> let bundle_url = mount_remote_plugin_bundle( &server, "linear", - remote_plugin_bundle_tar_gz_bytes("linear")?, + remote_plugin_bundle_tar_gz_bytes("linear", /*hooks_json*/ None)?, ) .await; let remote_app_manifest = serde_json::json!({ @@ -1637,7 +1642,7 @@ async fn plugin_list_sync_upgrades_and_removes_remote_installed_plugin_bundles() let bundle_url = mount_remote_plugin_bundle( &server, "linear", - remote_plugin_bundle_tar_gz_bytes("linear")?, + remote_plugin_bundle_tar_gz_bytes("linear", /*hooks_json*/ None)?, ) .await; let remote_app_manifest = serde_json::json!({ @@ -2767,7 +2772,7 @@ plugin_sharing = false let bundle_url = mount_remote_plugin_bundle( &server, "private-linear", - remote_plugin_bundle_tar_gz_bytes("private-linear")?, + remote_plugin_bundle_tar_gz_bytes("private-linear", /*hooks_json*/ None)?, ) .await; let mut user_installed_body: serde_json::Value = @@ -2829,42 +2834,54 @@ plugin_sharing = false } #[tokio::test] -async fn plugin_installed_starts_remote_installed_bundle_sync() -> Result<()> { +async fn plugin_installed_trusts_new_workspace_listed_plugin_hooks() -> Result<()> { let codex_home = TempDir::new()?; let server = MockServer::start().await; - std::fs::write( - codex_home.path().join("config.toml"), - format!( - r#"chatgpt_base_url = "{}/backend-api/" + let disabled_hook_key = "available-hooks@workspace-directory:hooks/hooks.json:pre_tool_use:0:0"; + let unrelated_hook_key = "unrelated@test:hooks/hooks.json:session_start:0:0"; + write_remote_plugin_hook_config( + codex_home.path(), + &format!("{}/backend-api/", server.uri()), + &format!( + r#" +[hooks.state."{disabled_hook_key}"] +enabled = false -[features] -plugins = true -plugin_sharing = false +[hooks.state."{unrelated_hook_key}"] +enabled = false +trusted_hash = "sha256:unrelated" "#, - 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, - )?; + write_remote_plugin_test_auth(codex_home.path())?; - let bundle_url = mount_remote_plugin_bundle( + let available_bundle_url = mount_remote_plugin_bundle_with_hooks( &server, - "linear", - remote_plugin_bundle_tar_gz_bytes("linear")?, + "available-hooks", + Some( + r#"{"hooks":{"PreToolUse":[{"matcher":"Bash","hooks":[{"type":"command","command":"echo available"}]}]}}"#, + ), + ) + .await?; + let default_bundle_url = mount_remote_plugin_bundle_with_hooks( + &server, + "default-hooks", + Some( + r#"{"hooks":{"SessionStart":[{"hooks":[{"type":"command","command":"echo default"}]}]}}"#, + ), + ) + .await?; + let no_hooks_bundle_url = + mount_remote_plugin_bundle_with_hooks(&server, "no-hooks", /*hooks_json*/ None).await?; + mount_workspace_bundle_sync( + &server, + &[ + ("available-hooks", "AVAILABLE", &available_bundle_url), + ("default-hooks", "INSTALLED_BY_DEFAULT", &default_bundle_url), + ("no-hooks", "AVAILABLE", &no_hooks_bundle_url), + ], ) .await; - let global_installed_body = - remote_installed_plugin_body(&bundle_url, "1.2.3", /*enabled*/ true); - mount_remote_installed_plugins(&server, "GLOBAL", &global_installed_body).await; - mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body()) - .await; - mount_empty_user_installed_plugins(&server).await; let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -2874,36 +2891,128 @@ plugin_sharing = false .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; - let plugin_installed_request_id = mcp - .send_plugin_installed_request(PluginInstalledParams { - cwds: None, - install_suggestion_plugin_names: None, - }) - .await?; - let response: PluginInstalledResponse = to_response( - timeout( - DEFAULT_TIMEOUT, - mcp.read_stream_until_response_message(RequestId::Integer(plugin_installed_request_id)), - ) - .await??, - )?; - - assert_eq!(response.marketplaces.len(), 1); - assert_eq!(response.marketplaces[0].name, "openai-curated-remote"); - assert_eq!( - response.marketplaces[0] - .plugins + trigger_plugin_installed_sync(&mut mcp).await?; + let plugin_ids = [ + "available-hooks@workspace-directory", + "default-hooks@workspace-directory", + ]; + let hooks = wait_for_plugin_hooks( + &mut mcp, + codex_home.path(), + &plugin_ids, + HookTrustStatus::Trusted, + ) + .await?; + for plugin_id in plugin_ids { + assert!( + hooks + .iter() + .any(|hook| hook.plugin_id.as_deref() == Some(plugin_id)) + ); + } + wait_for_path_exists( + &codex_home + .path() + .join("plugins/cache/workspace-directory/no-hooks/1.2.3/.codex-plugin/plugin.json"), + ) + .await?; + let config: toml::Value = toml::from_str(&std::fs::read_to_string( + codex_home.path().join("config.toml"), + )?)?; + let hook_states = config["hooks"]["state"].as_table().expect("hook states"); + for hook in &hooks { + assert_eq!( + hook_states[hook.key.as_str()]["trusted_hash"].as_str(), + Some(hook.current_hash.as_str()) + ); + } + assert!( + !hooks .iter() - .map(|plugin| (plugin.id.clone(), plugin.installed, plugin.enabled)) - .collect::>(), - vec![("linear@openai-curated-remote".to_string(), true, true)] + .find(|hook| hook.key == disabled_hook_key) + .expect("disabled hook") + .enabled ); - let installed_path = codex_home - .path() - .join("plugins/cache/openai-curated-remote/linear/1.2.3/.codex-plugin/plugin.json"); - wait_for_path_exists(&installed_path).await?; - wait_for_remote_installed_scope_request(&server, "GLOBAL").await?; - wait_for_remote_installed_scope_request(&server, "WORKSPACE").await?; + assert_eq!( + hook_states[disabled_hook_key]["enabled"].as_bool(), + Some(false) + ); + assert_eq!( + hook_states[unrelated_hook_key]["trusted_hash"].as_str(), + Some("sha256:unrelated") + ); + assert_eq!( + hook_states[unrelated_hook_key]["enabled"].as_bool(), + Some(false) + ); + Ok(()) +} + +#[cfg(unix)] +#[tokio::test] +async fn plugin_installed_hook_trust_write_failure_stays_untrusted() -> Result<()> { + use std::os::unix::fs::PermissionsExt; + use std::os::unix::fs::symlink; + + let codex_home = TempDir::new()?; + let config_target_dir = TempDir::new()?; + let config_target = config_target_dir.path().join("config.toml"); + let server = MockServer::start().await; + write_remote_plugin_hook_config( + config_target_dir.path(), + &format!("{}/backend-api/", server.uri()), + "", + )?; + symlink(&config_target, codex_home.path().join("config.toml"))?; + write_remote_plugin_test_auth(codex_home.path())?; + + let bundle_url = mount_remote_plugin_bundle_with_hooks( + &server, + "failed-trust", + Some( + r#"{"hooks":{"SessionStart":[{"hooks":[{"type":"command","command":"echo fail closed"}]}]}}"#, + ), + ) + .await?; + mount_workspace_bundle_sync(&server, &[("failed-trust", "AVAILABLE", &bundle_url)]).await; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .with_env_overrides(&[(TEST_ALLOW_HTTP_REMOTE_PLUGIN_BUNDLE_DOWNLOADS, Some("1"))]) + .build() + .await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let original_permissions = std::fs::metadata(config_target_dir.path())?.permissions(); + let _permission_guard = RestorePermissions( + config_target_dir.path().to_path_buf(), + original_permissions.clone(), + ); + let mut read_only_permissions = original_permissions; + read_only_permissions.set_mode(read_only_permissions.mode() & !0o222); + std::fs::set_permissions(config_target_dir.path(), read_only_permissions)?; + + trigger_plugin_installed_sync(&mut mcp).await?; + let plugin_ids = ["failed-trust@workspace-directory"]; + let before = wait_for_plugin_hooks( + &mut mcp, + codex_home.path(), + &plugin_ids, + HookTrustStatus::Untrusted, + ) + .await?; + sleep(Duration::from_millis(300)).await; + let after = wait_for_plugin_hooks( + &mut mcp, + codex_home.path(), + &plugin_ids, + HookTrustStatus::Untrusted, + ) + .await?; + + assert_eq!(after[0].current_hash, before[0].current_hash); + assert!(!std::fs::read_to_string(config_target)?.contains("trusted_hash")); Ok(()) } @@ -3867,6 +3976,61 @@ async fn wait_for_path_exists(path: &std::path::Path) -> Result<()> { Ok(()) } +async fn trigger_plugin_installed_sync(mcp: &mut TestAppServer) -> Result<()> { + let request_id = mcp + .send_plugin_installed_request(PluginInstalledParams { + cwds: None, + install_suggestion_plugin_names: None, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let _: PluginInstalledResponse = to_response(response)?; + Ok(()) +} + +async fn wait_for_plugin_hooks( + mcp: &mut TestAppServer, + cwd: &std::path::Path, + plugin_ids: &[&str], + expected_status: HookTrustStatus, +) -> Result> { + timeout(DEFAULT_TIMEOUT, async { + loop { + let request_id = mcp + .send_hooks_list_request(HooksListParams { + cwds: vec![cwd.to_path_buf()], + }) + .await?; + let response: JSONRPCResponse = mcp + .read_stream_until_response_message(RequestId::Integer(request_id)) + .await?; + let HooksListResponse { data } = to_response(response)?; + let hooks = data + .into_iter() + .flat_map(|entry| entry.hooks) + .filter(|hook| { + hook.plugin_id + .as_deref() + .is_some_and(|plugin_id| plugin_ids.contains(&plugin_id)) + }) + .collect::>(); + if hooks.len() == plugin_ids.len() + && hooks + .iter() + .all(|hook| hook.trust_status == expected_status) + { + return Ok::<_, anyhow::Error>(hooks); + } + sleep(Duration::from_millis(10)).await; + } + }) + .await? +} + async fn wait_for_path_missing(path: &std::path::Path) -> Result<()> { timeout(DEFAULT_TIMEOUT, async { loop { @@ -4034,6 +4198,34 @@ fn workspace_remote_plugin_page_body( ) } +async fn mount_workspace_bundle_sync(server: &MockServer, plugins: &[(&str, &str, &str)]) { + let plugins = plugins + .iter() + .map(|(name, install_policy, bundle_url)| { + let body: serde_json::Value = serde_json::from_str(&workspace_remote_plugin_page_body( + &format!("plugins~Plugin_{name}"), + name, + name, + "LISTED", + /*enabled*/ Some(true), + )) + .expect("workspace plugin body"); + let mut plugin = body["plugins"][0].clone(); + plugin["installation_policy"] = serde_json::json!(install_policy); + plugin["release"]["bundle_download_url"] = serde_json::json!(bundle_url); + plugin + }) + .collect::>(); + let body = serde_json::json!({ + "plugins": plugins, + "pagination": {"next_page_token": null}, + }) + .to_string(); + mount_remote_installed_plugins(server, "GLOBAL", empty_remote_installed_plugins_body()).await; + mount_remote_installed_plugins(server, "WORKSPACE", &body).await; + mount_empty_user_installed_plugins(server).await; +} + fn user_remote_plugin_page_body( remote_plugin_id: &str, plugin_name: &str, @@ -4137,12 +4329,28 @@ async fn mount_remote_plugin_bundle( format!("{}{bundle_path}", server.uri()) } -fn remote_plugin_bundle_tar_gz_bytes(plugin_name: &str) -> Result> { +async fn mount_remote_plugin_bundle_with_hooks( + server: &MockServer, + plugin_name: &str, + hooks_json: Option<&str>, +) -> Result { + Ok(mount_remote_plugin_bundle( + server, + plugin_name, + remote_plugin_bundle_tar_gz_bytes(plugin_name, hooks_json)?, + ) + .await) +} + +fn remote_plugin_bundle_tar_gz_bytes( + plugin_name: &str, + hooks_json: Option<&str>, +) -> Result> { let manifest = format!(r#"{{"name":"{plugin_name}"}}"#); let skill = "---\nname: plan-work\ndescription: Track work in Linear.\n---\n\n# Plan Work\n"; let encoder = GzEncoder::new(Vec::new(), Compression::default()); let mut tar = tar::Builder::new(encoder); - for (path, contents, mode) in [ + let mut entries = vec![ ( ".codex-plugin/plugin.json", manifest.as_bytes(), @@ -4153,7 +4361,15 @@ fn remote_plugin_bundle_tar_gz_bytes(plugin_name: &str) -> Result> { skill.as_bytes(), /*mode*/ 0o644, ), - ] { + ]; + if let Some(hooks_json) = hooks_json { + entries.push(( + "hooks/hooks.json", + hooks_json.as_bytes(), + /*mode*/ 0o644, + )); + } + for (path, contents, mode) in entries { let mut header = tar::Header::new_gnu(); header.set_size(contents.len() as u64); header.set_mode(mode); @@ -4232,6 +4448,35 @@ plugins = true ) } +fn write_remote_plugin_hook_config( + codex_home: &std::path::Path, + base_url: &str, + hook_state: &str, +) -> std::io::Result<()> { + std::fs::write( + codex_home.join("config.toml"), + format!( + r#"chatgpt_base_url = "{base_url}" + +[features] +plugins = true +hooks = true +{hook_state}"#, + ), + ) +} + +fn write_remote_plugin_test_auth(codex_home: &std::path::Path) -> Result<()> { + write_chatgpt_auth( + codex_home, + ChatGptAuthFixture::new("chatgpt-token") + .account_id("account-123") + .chatgpt_user_id("user-123") + .chatgpt_account_id("account-123"), + AuthCredentialsStoreMode::File, + ) +} + fn write_openai_curated_marketplace( codex_home: &std::path::Path, plugin_names: &[&str], @@ -4341,3 +4586,13 @@ fn write_plugin_share_local_path_mapping( format!("{contents}\n"), ) } + +#[cfg(unix)] +struct RestorePermissions(std::path::PathBuf, std::fs::Permissions); + +#[cfg(unix)] +impl Drop for RestorePermissions { + fn drop(&mut self) { + let _ = std::fs::set_permissions(&self.0, self.1.clone()); + } +} diff --git a/codex-rs/core-plugins/src/lib.rs b/codex-rs/core-plugins/src/lib.rs index df30305c38..2241eda43c 100644 --- a/codex-rs/core-plugins/src/lib.rs +++ b/codex-rs/core-plugins/src/lib.rs @@ -43,6 +43,7 @@ pub use loader::PluginHookLoadOutcome; pub use manager::ConfiguredMarketplace; pub use manager::ConfiguredMarketplaceListOutcome; pub use manager::ConfiguredMarketplacePlugin; +pub use manager::EffectivePluginsChange; pub use manager::PluginDetail; pub use manager::PluginDetailsUnavailableReason; pub use manager::PluginInstallError; diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 16fb4c8663..010455e5c2 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -44,7 +44,9 @@ use crate::marketplace_upgrade::upgrade_configured_git_marketplaces; use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME; use crate::remote::RecommendedPluginsMode; use crate::remote::RemoteInstalledPlugin; +use crate::remote::RemoteInstalledPluginBundleSyncOutcome; use crate::remote::RemotePluginCatalogError; +use crate::remote::RemotePluginMaterialization; use crate::remote::RemotePluginServiceConfig; use crate::remote_legacy::RemotePluginFetchError; use crate::remote_legacy::RemotePluginMutationError; @@ -104,6 +106,8 @@ static CURATED_REPO_SYNC_STARTED: AtomicBool = AtomicBool::new(false); const FEATURED_PLUGIN_IDS_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(60 * 60 * 3); +type EffectivePluginsChangedCallback = Arc; + #[derive(Debug, Clone)] pub struct PluginsConfigInput { pub config_layer_stack: ConfigLayerStack, @@ -128,6 +132,13 @@ impl PluginsConfigInput { } } +/// Effective-plugin changes that downstream composition layers may act on. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct EffectivePluginsChange { + /// Remote bundles installed or updated by background installed-plugin sync. + pub materialized_remote_plugins: Vec, +} + /// Inputs used to select endpoint-backed plugin install candidates. pub struct RecommendedPluginCandidatesInput<'a> { pub plugins_config: &'a PluginsConfigInput, @@ -163,7 +174,8 @@ struct RemoteInstalledPluginsCacheRefreshRequest { notify: RemoteInstalledPluginsCacheRefreshNotify, // App-server attaches side effects such as skills metadata invalidation and MCP refreshes when // remote installed state changes. - on_effective_plugins_changed: Option>, + on_effective_plugins_changed: Option, + change: EffectivePluginsChange, } #[derive(Clone, Copy)] @@ -879,7 +891,7 @@ impl PluginsManager { config: &PluginsConfigInput, auth: Option<&CodexAuth>, visible_marketplaces: &[&str], - on_effective_plugins_changed: Option>, + on_effective_plugins_changed: Option, ) -> Result, RemotePluginCatalogError> { let plugins = crate::remote::fetch_remote_installed_plugins( &remote_plugin_service_config(config), @@ -892,7 +904,7 @@ impl PluginsManager { ); let changed = self.write_remote_installed_plugins_cache(plugins); if changed && let Some(on_effective_plugins_changed) = on_effective_plugins_changed { - on_effective_plugins_changed(); + on_effective_plugins_changed(EffectivePluginsChange::default()); } Ok(marketplaces) } @@ -929,13 +941,14 @@ impl PluginsManager { self: &Arc, config: &PluginsConfigInput, auth: Option, - on_effective_plugins_changed: Option>, + on_effective_plugins_changed: Option, ) { self.maybe_start_remote_installed_plugins_cache_refresh_with_notify( config, auth.clone(), RemoteInstalledPluginsCacheRefreshNotify::IfCacheChanged, on_effective_plugins_changed, + EffectivePluginsChange::default(), ); let manager = Arc::clone(self); @@ -951,13 +964,14 @@ impl PluginsManager { self: &Arc, config: &PluginsConfigInput, auth: Option, - on_effective_plugins_changed: Option>, + on_effective_plugins_changed: Option, ) { self.maybe_start_remote_installed_plugins_cache_refresh_with_notify( config, auth, RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh, on_effective_plugins_changed, + EffectivePluginsChange::default(), ); } @@ -966,7 +980,8 @@ impl PluginsManager { config: &PluginsConfigInput, auth: Option, notify: RemoteInstalledPluginsCacheRefreshNotify, - on_effective_plugins_changed: Option>, + on_effective_plugins_changed: Option, + change: EffectivePluginsChange, ) { if !config.plugins_enabled { return; @@ -978,6 +993,7 @@ impl PluginsManager { auth, notify, on_effective_plugins_changed, + change, }, ); } @@ -986,7 +1002,7 @@ impl PluginsManager { self: &Arc, config: &PluginsConfigInput, auth: Option, - on_effective_plugins_changed: Option>, + on_effective_plugins_changed: Option, ) { if !config.plugins_enabled { return; @@ -995,13 +1011,18 @@ impl PluginsManager { let manager = Arc::clone(self); let config_for_refresh = config.clone(); let auth_for_refresh = auth.clone(); - let on_local_cache_changed = Arc::new(move || { - manager.maybe_start_remote_installed_plugins_cache_refresh_after_mutation( - &config_for_refresh, - auth_for_refresh.clone(), - on_effective_plugins_changed.clone(), - ); - }); + let on_local_cache_changed = + Arc::new(move |outcome: RemoteInstalledPluginBundleSyncOutcome| { + manager.maybe_start_remote_installed_plugins_cache_refresh_with_notify( + &config_for_refresh, + auth_for_refresh.clone(), + RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh, + on_effective_plugins_changed.clone(), + EffectivePluginsChange { + materialized_remote_plugins: outcome.materialized_remote_plugins, + }, + ); + }); crate::remote::maybe_start_remote_installed_plugin_bundle_sync( self.codex_home.clone(), @@ -1032,7 +1053,7 @@ impl PluginsManager { auth: Option, roots: &[AbsolutePathBuf], options: PluginListBackgroundTaskOptions, - on_effective_plugins_changed: Option>, + on_effective_plugins_changed: Option, ) { self.maybe_start_non_curated_plugin_cache_refresh(config, roots); if options.refresh_global_remote_catalog_cache { @@ -1936,7 +1957,7 @@ impl PluginsManager { self: &Arc, config: &PluginsConfigInput, auth_manager: Arc, - on_effective_plugins_changed: Option>, + on_effective_plugins_changed: Option, ) { if config.plugins_enabled { let use_remote_global_catalog = @@ -2139,10 +2160,35 @@ impl PluginsManager { request.notify = RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh; } - if request.on_effective_plugins_changed.is_none() { + if !existing_request + .change + .materialized_remote_plugins + .is_empty() + && let Some(existing_callback) = + existing_request.on_effective_plugins_changed.as_ref() + { + request.on_effective_plugins_changed = Some(Arc::clone(existing_callback)); + } else if request.on_effective_plugins_changed.is_none() { request.on_effective_plugins_changed = existing_request.on_effective_plugins_changed.clone(); } + for materialization in &existing_request.change.materialized_remote_plugins { + if !request + .change + .materialized_remote_plugins + .iter() + .any(|pending| pending.plugin_id == materialization.plugin_id) + { + request + .change + .materialized_remote_plugins + .push(materialization.clone()); + } + } + request + .change + .materialized_remote_plugins + .sort_by_key(|materialization| materialization.plugin_id.as_key()); } state.requested = Some(request); if state.in_flight { @@ -2366,6 +2412,7 @@ impl PluginsManager { // publishing remote installed state as effective local plugin config. let changed = self.write_remote_installed_plugins_cache(installed_plugins); let should_notify = changed + || !request.change.materialized_remote_plugins.is_empty() || matches!( request.notify, RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh @@ -2374,7 +2421,7 @@ impl PluginsManager { && let Some(on_effective_plugins_changed) = request.on_effective_plugins_changed { - on_effective_plugins_changed(); + on_effective_plugins_changed(request.change); } } Err( @@ -2386,12 +2433,16 @@ impl PluginsManager { && let Some(on_effective_plugins_changed) = request.on_effective_plugins_changed { - on_effective_plugins_changed(); + on_effective_plugins_changed(EffectivePluginsChange::default()); } } Err(err) => { warn!( error = %err, + materialized_remote_plugin_count = request + .change + .materialized_remote_plugins + .len(), "failed to refresh remote installed plugins cache" ); } diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index 5e743416f2..151d2d3768 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -5874,3 +5874,79 @@ async fn plugin_hooks_for_layer_stack_loads_configured_plugin_hooks() { ); assert_eq!(outcome.hook_load_warnings, Vec::::new()); } + +#[test] +fn remote_installed_plugins_cache_refresh_coalesces_materializations() { + let tmp = TempDir::new().unwrap(); + let manager = std::sync::Arc::new(PluginsManager::new(tmp.path().to_path_buf())); + let materialization_callback_count = + std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let unrelated_callback_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + manager + .remote_installed_plugins_cache_refresh_state + .write() + .expect("refresh state lock") + .in_flight = true; + let materialization = |name: &str| RemotePluginMaterialization { + plugin_id: PluginId::new( + name.to_string(), + REMOTE_WORKSPACE_MARKETPLACE_NAME.to_string(), + ) + .expect("valid plugin id"), + scope: crate::remote::RemotePluginScope::Workspace, + discoverability: Some(crate::remote::RemotePluginShareDiscoverability::Listed), + authenticated_account_id: Some("account-123".to_string()), + }; + let change = |name: &str| EffectivePluginsChange { + materialized_remote_plugins: vec![materialization(name)], + }; + let callback = |count: std::sync::Arc| { + let callback: EffectivePluginsChangedCallback = std::sync::Arc::new(move |_change| { + count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + }); + callback + }; + let request = + |change, on_effective_plugins_changed| RemoteInstalledPluginsCacheRefreshRequest { + service_config: RemotePluginServiceConfig { + chatgpt_base_url: "https://example.com".to_string(), + }, + auth: None, + notify: RemoteInstalledPluginsCacheRefreshNotify::IfCacheChanged, + on_effective_plugins_changed: Some(on_effective_plugins_changed), + change, + }; + + manager.schedule_remote_installed_plugins_cache_refresh(request( + change("beta"), + callback(std::sync::Arc::clone(&materialization_callback_count)), + )); + manager.schedule_remote_installed_plugins_cache_refresh(request( + change("alpha"), + callback(std::sync::Arc::clone(&unrelated_callback_count)), + )); + + let state = manager + .remote_installed_plugins_cache_refresh_state + .read() + .expect("refresh state lock"); + let request = state.requested.as_ref().expect("pending refresh"); + assert_eq!( + request.change, + EffectivePluginsChange { + materialized_remote_plugins: vec![materialization("alpha"), materialization("beta"),], + } + ); + request + .on_effective_plugins_changed + .as_ref() + .expect("pending callback")(request.change.clone()); + assert_eq!( + materialization_callback_count.load(std::sync::atomic::Ordering::Relaxed), + 1 + ); + assert_eq!( + unrelated_callback_count.load(std::sync::atomic::Ordering::Relaxed), + 0 + ); +} diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index e5aef0bb47..52a57330c8 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -45,6 +45,7 @@ mod tests; pub use remote_installed_plugin_sync::RemoteInstalledPluginBundleSyncError; pub use remote_installed_plugin_sync::RemoteInstalledPluginBundleSyncOutcome; pub use remote_installed_plugin_sync::RemotePluginCacheMutationGuard; +pub use remote_installed_plugin_sync::RemotePluginMaterialization; pub use remote_installed_plugin_sync::mark_remote_plugin_cache_mutation_in_flight; pub(crate) use remote_installed_plugin_sync::maybe_start_remote_installed_plugin_bundle_sync; pub use remote_installed_plugin_sync::sync_remote_installed_plugin_bundles_once; diff --git a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs index 2a23b066ed..e168698722 100644 --- a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs +++ b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs @@ -7,6 +7,7 @@ use super::REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME; use super::RemotePluginCatalogError; use super::RemotePluginScope; use super::RemotePluginServiceConfig; +use super::RemotePluginShareDiscoverability; use super::ensure_chatgpt_auth; use super::fetch_installed_plugins_for_scope_with_download_url; use super::remote_plugin_canonical_marketplace_name; @@ -35,16 +36,25 @@ static REMOTE_PLUGIN_CACHE_MUTATIONS_IN_FLIGHT: OnceLock< Mutex>, > = OnceLock::new(); +/// A remote plugin bundle newly installed or updated from an authenticated snapshot. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RemotePluginMaterialization { + pub plugin_id: PluginId, + pub scope: RemotePluginScope, + pub discoverability: Option, + pub authenticated_account_id: Option, +} + #[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct RemoteInstalledPluginBundleSyncOutcome { - pub installed_plugin_ids: Vec, + pub materialized_remote_plugins: Vec, pub removed_cache_plugin_ids: Vec, pub failed_remote_plugin_ids: Vec, } impl RemoteInstalledPluginBundleSyncOutcome { pub fn changed_local_cache(&self) -> bool { - !self.installed_plugin_ids.is_empty() || !self.removed_cache_plugin_ids.is_empty() + !self.materialized_remote_plugins.is_empty() || !self.removed_cache_plugin_ids.is_empty() } } @@ -55,12 +65,6 @@ pub enum RemoteInstalledPluginBundleSyncError { #[error("{0}")] Store(#[from] PluginStoreError), - - #[error("failed to join stale remote plugin cache cleanup task: {0}")] - Join(#[from] tokio::task::JoinError), - - #[error("failed to remove stale remote plugin cache entries: {0}")] - CacheRemove(String), } #[derive(Debug, Clone, PartialEq, Eq, Hash)] @@ -83,7 +87,9 @@ pub(crate) fn maybe_start_remote_installed_plugin_bundle_sync( codex_home: PathBuf, config: RemotePluginServiceConfig, auth: Option, - on_local_cache_changed: Option>, + on_local_cache_changed: Option< + Arc, + >, ) { let Some(auth) = auth else { return; @@ -100,17 +106,17 @@ pub(crate) fn maybe_start_remote_installed_plugin_bundle_sync( sync_remote_installed_plugin_bundles_once(codex_home, &config, Some(&auth)).await; match result { Ok(outcome) => { - if outcome.changed_local_cache() - && let Some(on_local_cache_changed) = on_local_cache_changed - { - on_local_cache_changed(); - } info!( - installed_plugin_ids = ?outcome.installed_plugin_ids, + materialized_remote_plugins = ?outcome.materialized_remote_plugins, removed_cache_plugin_ids = ?outcome.removed_cache_plugin_ids, failed_remote_plugin_ids = ?outcome.failed_remote_plugin_ids, "completed remote installed plugin bundle sync" ); + if outcome.changed_local_cache() + && let Some(on_local_cache_changed) = on_local_cache_changed + { + on_local_cache_changed(outcome); + } } Err(err) => { warn!( @@ -129,6 +135,7 @@ pub async fn sync_remote_installed_plugin_bundles_once( auth: Option<&CodexAuth>, ) -> Result { let auth = ensure_chatgpt_auth(auth)?; + let authenticated_account_id = auth.get_account_id(); let global = async { let scope = RemotePluginScope::Global; let installed_plugins = fetch_installed_plugins_for_scope_with_download_url( @@ -180,12 +187,14 @@ pub async fn sync_remote_installed_plugin_bundles_once( BTreeSet::new(), ), ]); - let mut installed_plugin_ids = BTreeSet::new(); + let mut materialized_remote_plugins = BTreeMap::new(); let mut failed_remote_plugin_ids = BTreeSet::new(); for (_scope, installed_plugins) in [global, workspace, user] { for installed_plugin in installed_plugins { let plugin = installed_plugin.plugin; + let scope = plugin.scope; + let discoverability = plugin.discoverability; let marketplace_name = remote_plugin_canonical_marketplace_name(&plugin)?.to_string(); installed_plugin_names_by_marketplace .entry(marketplace_name.clone()) @@ -254,7 +263,16 @@ pub async fn sync_remote_installed_plugin_bundles_once( .await { Ok(result) => { - installed_plugin_ids.insert(result.plugin_id.as_key()); + let plugin_id = result.plugin_id; + materialized_remote_plugins.insert( + plugin_id.as_key(), + RemotePluginMaterialization { + plugin_id, + scope, + discoverability, + authenticated_account_id: authenticated_account_id.clone(), + }, + ); } Err(err) => { warn!( @@ -270,17 +288,27 @@ pub async fn sync_remote_installed_plugin_bundles_once( } } - let removed_cache_plugin_ids = tokio::task::spawn_blocking(move || { + let stale_cache_cleanup = tokio::task::spawn_blocking(move || { remove_stale_remote_plugin_caches( codex_home.as_path(), &installed_plugin_names_by_marketplace, ) }) - .await? - .map_err(RemoteInstalledPluginBundleSyncError::CacheRemove)?; + .await; + let removed_cache_plugin_ids = match stale_cache_cleanup { + Ok(Ok(removed_cache_plugin_ids)) => removed_cache_plugin_ids, + Ok(Err(err)) => { + warn!(error = %err, "failed to remove stale remote plugin cache entries"); + Vec::new() + } + Err(err) => { + warn!(error = %err, "failed to join stale remote plugin cache cleanup task"); + Vec::new() + } + }; Ok(RemoteInstalledPluginBundleSyncOutcome { - installed_plugin_ids: installed_plugin_ids.into_iter().collect(), + materialized_remote_plugins: materialized_remote_plugins.into_values().collect(), removed_cache_plugin_ids, failed_remote_plugin_ids: failed_remote_plugin_ids.into_iter().collect(), }) @@ -479,7 +507,7 @@ mod tests { } #[tokio::test] - async fn sync_backfills_remote_plugin_install_metadata_for_current_bundle() { + async fn sync_same_version_backfills_metadata_without_materialization() { let server = MockServer::start().await; let codex_home = tempfile::tempdir().expect("create codex home"); let cached_manifest = codex_home