Refresh session hooks after external plugin updates (#42990)

## Why

Plugin updates made by another process do not notify a loaded session's hook runtime, leaving it with stale hooks.

## What changed

Compare the current plugin hook sources and load warnings with those retained by the hook runtime during turn construction. Refresh hooks when either differs so existing sessions pick up external plugin updates.

## Testing

Add an app-server regression test that updates the shared plugin store without notifying the server and verifies that successive turns in the same thread run the installed hooks across an upgrade and a rollback.

GitOrigin-RevId: d3a2653ac4c069d9d8a02f30e21e4506d76a091c

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

@@ -26,7 +26,9 @@ use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStartResponse;
 use codex_app_server_protocol::UserInput as V2UserInput;
 use codex_core::config::set_project_trust_level;
+use codex_core_plugins::store::PluginStore;
 use codex_features::Feature;
+use codex_plugin::PluginId;
 use codex_protocol::config_types::TrustLevel;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use core_test_support::skip_if_host_windows;
@@ -550,6 +552,102 @@ async fn plugin_upgrade_refreshes_hook_runtime_for_loaded_session() -> Result<()
     Ok(())
 }
 
+#[tokio::test]
+async fn loaded_session_refreshes_externally_updated_plugin_hooks() -> Result<()> {
+    skip_if_host_windows!(Ok(()));
+    skip_if_remote!(Ok(()), "command hooks use host-local script and log paths");
+
+    let server = create_mock_responses_server_sequence_unchecked(vec![
+        create_final_assistant_message_sse_response("Version 1")?,
+        create_final_assistant_message_sse_response("Version 2")?,
+        create_final_assistant_message_sse_response("Rollback")?,
+    ])
+    .await;
+    let codex_home = TempDir::new()?;
+    let source = TempDir::new()?;
+    let hook_log_path = codex_home.path().join("plugin-hook-versions.log");
+    let plugin_id = PluginId::parse("demo@test")?;
+    let store = PluginStore::new(codex_home.path().to_path_buf());
+    write_versioned_plugin_hook(source.path(), "1.0.0", &hook_log_path)?;
+    store.install(AbsolutePathBuf::try_from(source.path())?, plugin_id.clone())?;
+    MockResponsesConfig::new(&server.uri())
+        .enable_feature(Feature::Plugins)
+        .enable_feature(Feature::CodexHooks)
+        .write(codex_home.path())?;
+    let mut config = std::fs::read_to_string(codex_home.path().join("config.toml"))?;
+    config.push_str("\n[plugins.\"demo@test\"]\nenabled = true\n");
+    std::fs::write(codex_home.path().join("config.toml"), config)?;
+
+    let mut mcp = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .build_initialized_with_timeout(DEFAULT_TIMEOUT)
+        .await?;
+    let list_id = mcp
+        .send_hooks_list_request(HooksListParams {
+            cwds: vec![codex_home.path().to_path_buf()],
+        })
+        .await?;
+    let HooksListResponse { data } = timeout(DEFAULT_TIMEOUT, mcp.read_response(list_id)).await??;
+    let trusted_hooks = data[0]
+        .hooks
+        .iter()
+        .map(|hook| {
+            (
+                hook.key.clone(),
+                serde_json::json!({ "trusted_hash": hook.current_hash }),
+            )
+        })
+        .collect::<serde_json::Map<String, serde_json::Value>>();
+    let trust_id = mcp
+        .send_config_batch_write_request(ConfigBatchWriteParams {
+            edits: vec![ConfigEdit {
+                key_path: "hooks.state".to_string(),
+                value: serde_json::Value::Object(trusted_hooks),
+                merge_strategy: MergeStrategy::Upsert,
+            }],
+            file_path: None,
+            expected_version: None,
+            reload_user_config: true,
+        })
+        .await?;
+    let _: codex_app_server_protocol::ConfigWriteResponse =
+        timeout(DEFAULT_TIMEOUT, mcp.read_response(trust_id)).await??;
+    let start_id = mcp
+        .send_thread_start_request_with_auto_env(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let ThreadStartResponse { thread, .. } =
+        timeout(DEFAULT_TIMEOUT, mcp.read_response(start_id)).await??;
+
+    let mut expected_versions = String::new();
+    for version in ["1.0.0", "2.0.0", "0.9.0"] {
+        // This test process updates the shared store without notifying app-server.
+        write_versioned_plugin_hook(source.path(), version, &hook_log_path)?;
+        store.install(AbsolutePathBuf::try_from(source.path())?, plugin_id.clone())?;
+        let turn_id = mcp
+            .send_turn_start_request(TurnStartParams {
+                thread_id: thread.id.clone(),
+                input: vec![V2UserInput::Text {
+                    text: format!("run version {version}"),
+                    text_elements: Vec::new(),
+                }],
+                ..Default::default()
+            })
+            .await?;
+        let _: TurnStartResponse = timeout(DEFAULT_TIMEOUT, mcp.read_response(turn_id)).await??;
+        timeout(
+            DEFAULT_TIMEOUT,
+            mcp.read_stream_until_notification_message("turn/completed"),
+        )
+        .await??;
+        expected_versions.push_str(&format!("{version}\n"));
+        assert_eq!(std::fs::read_to_string(&hook_log_path)?, expected_versions);
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn automatic_marketplace_upgrade_refreshes_hook_runtime_for_loaded_session() -> Result<()> {
     skip_if_host_windows!(Ok(()));

codex-rs/core/src/session/turn_context.rs

@@ -964,6 +964,17 @@ impl Session {
             .plugins_manager
             .plugins_for_config(&plugins_input)
             .await;
+        // Cache changes from another process do not notify this session's hook runtime.
+        if !self.hooks().matches_plugin_hooks(
+            plugin_outcome.iter_effective_plugin_hook_sources(),
+            plugin_outcome.iter_effective_plugin_hook_warnings(),
+        ) {
+            // Keep the refresh state out of the enclosing turn-construction future.
+            Box::pin(self.refresh_hooks(Arc::clone(
+                &session_configuration.original_config_do_not_use,
+            )))
+            .await;
+        }
         let trusted_plugin_roots = TrustedPluginRoots::from_plugin_load_outcome(
             &plugin_outcome,
             per_turn_config.codex_home.as_path(),

codex-rs/hooks/src/registry.rs

@@ -63,6 +63,8 @@ pub struct Hooks {
     environment: Arc<Vec<(OsString, OsString)>>,
     after_agent: Vec<Hook>,
     engine: ClaudeHooksEngine,
+    plugin_hook_sources: Vec<PluginHookSource>,
+    plugin_hook_load_warnings: Vec<String>,
 }
 
 impl Hooks {
@@ -98,6 +100,15 @@ impl Hooks {
         )
     }
 
+    pub fn matches_plugin_hooks<'a>(
+        &self,
+        sources: impl IntoIterator<Item = &'a PluginHookSource>,
+        warnings: impl IntoIterator<Item = &'a String>,
+    ) -> bool {
+        self.plugin_hook_sources.iter().eq(sources)
+            && self.plugin_hook_load_warnings.iter().eq(warnings)
+    }
+
     pub fn with_executor_hooks(&self, executor_hooks: Vec<ExecutorPluginHookSource>) -> Self {
         let mut hooks = self.clone();
         hooks.engine.set_executor_hooks(executor_hooks);
@@ -124,8 +135,8 @@ impl Hooks {
             config.feature_enabled,
             config.bypass_hook_trust,
             config.config_layer_stack.as_ref(),
-            config.plugin_hook_sources,
-            config.plugin_hook_load_warnings,
+            config.plugin_hook_sources.clone(),
+            config.plugin_hook_load_warnings.clone(),
             command_runtime,
             mcp_executor,
         );
@@ -133,6 +144,8 @@ impl Hooks {
             environment,
             after_agent,
             engine,
+            plugin_hook_sources: config.plugin_hook_sources,
+            plugin_hook_load_warnings: config.plugin_hook_load_warnings,
         }
     }
 

codex-rs/plugin/src/load_outcome.rs

@@ -169,19 +169,27 @@ impl<M: Clone> PluginLoadOutcome<M> {
     }
 
     pub fn effective_plugin_hook_sources(&self) -> Vec<PluginHookSource> {
+        self.iter_effective_plugin_hook_sources().cloned().collect()
+    }
+
+    pub fn iter_effective_plugin_hook_sources(&self) -> impl Iterator<Item = &PluginHookSource> {
         self.plugins
             .iter()
             .filter(|plugin| plugin.is_active())
-            .flat_map(|plugin| plugin.hook_sources.iter().cloned())
-            .collect()
+            .flat_map(|plugin| plugin.hook_sources.iter())
     }
 
     pub fn effective_plugin_hook_warnings(&self) -> Vec<String> {
+        self.iter_effective_plugin_hook_warnings()
+            .cloned()
+            .collect()
+    }
+
+    pub fn iter_effective_plugin_hook_warnings(&self) -> impl Iterator<Item = &String> {
         self.plugins
             .iter()
             .filter(|plugin| plugin.is_active())
-            .flat_map(|plugin| plugin.hook_load_warnings.iter().cloned())
-            .collect()
+            .flat_map(|plugin| plugin.hook_load_warnings.iter())
     }
 
     pub fn capability_summaries(&self) -> &[PluginCapabilitySummary] {