diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 9bd5f6a5a3..7e92cf5ae2 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -171,6 +171,8 @@ use codex_protocol::models::NetworkPermissions as CoreNetworkPermissions; use codex_protocol::models::PermissionProfile as CorePermissionProfile; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::HookEventName; +use codex_protocol::protocol::HookExecutionMode; +use codex_protocol::protocol::HookHandlerType; use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookSource; use codex_protocol::protocol::SessionSource; @@ -3745,6 +3747,8 @@ fn hook_run_event_serializes_expected_shape() { HookRunFact { event_name: HookEventName::PreToolUse, hook_source: HookSource::User, + handler_type: HookHandlerType::McpTool, + execution_mode: HookExecutionMode::Sync, status: HookRunStatus::Completed, }, ), @@ -3763,6 +3767,8 @@ fn hook_run_event_serializes_expected_shape() { "model_slug": "gpt-5", "hook_name": "PreToolUse", "hook_source": "user", + "handler_type": "mcp_tool", + "execution_mode": "sync", "status": "completed" } }) @@ -3778,6 +3784,8 @@ fn hook_run_metadata_maps_sources_and_statuses() { HookRunFact { event_name: HookEventName::SessionStart, hook_source: HookSource::System, + handler_type: HookHandlerType::Command, + execution_mode: HookExecutionMode::Sync, status: HookRunStatus::Completed, }, )) @@ -3787,6 +3795,8 @@ fn hook_run_metadata_maps_sources_and_statuses() { HookRunFact { event_name: HookEventName::Stop, hook_source: HookSource::Project, + handler_type: HookHandlerType::Prompt, + execution_mode: HookExecutionMode::Async, status: HookRunStatus::Blocked, }, )) @@ -3796,6 +3806,8 @@ fn hook_run_metadata_maps_sources_and_statuses() { HookRunFact { event_name: HookEventName::Stop, hook_source: HookSource::CloudRequirements, + handler_type: HookHandlerType::Agent, + execution_mode: HookExecutionMode::Sync, status: HookRunStatus::Blocked, }, )) @@ -3805,6 +3817,8 @@ fn hook_run_metadata_maps_sources_and_statuses() { HookRunFact { event_name: HookEventName::UserPromptSubmit, hook_source: HookSource::Unknown, + handler_type: HookHandlerType::Command, + execution_mode: HookExecutionMode::Async, status: HookRunStatus::Failed, }, )) @@ -3829,6 +3843,8 @@ fn hook_run_metadata_maps_stopped_status() { HookRunFact { event_name: HookEventName::Stop, hook_source: HookSource::User, + handler_type: HookHandlerType::Command, + execution_mode: HookExecutionMode::Sync, status: HookRunStatus::Stopped, }, )) @@ -4015,6 +4031,8 @@ async fn reducer_ingests_hook_run_fact() { hook: HookRunFact { event_name: HookEventName::PostToolUse, hook_source: HookSource::Unknown, + handler_type: HookHandlerType::Agent, + execution_mode: HookExecutionMode::Async, status: HookRunStatus::Failed, }, })), @@ -4027,6 +4045,8 @@ async fn reducer_ingests_hook_run_fact() { assert_eq!(payload[0]["event_type"], "codex_hook_run"); assert_eq!(payload[0]["event_params"]["hook_name"], "PostToolUse"); assert_eq!(payload[0]["event_params"]["hook_source"], "unknown"); + assert_eq!(payload[0]["event_params"]["handler_type"], "agent"); + assert_eq!(payload[0]["event_params"]["execution_mode"], "async"); assert_eq!(payload[0]["event_params"]["status"], "failed"); } diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 7080aa20b7..af431721eb 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -39,6 +39,8 @@ use codex_protocol::protocol::GuardianCommandSource; use codex_protocol::protocol::GuardianRiskLevel; use codex_protocol::protocol::GuardianUserAuthorization; use codex_protocol::protocol::HookEventName; +use codex_protocol::protocol::HookExecutionMode; +use codex_protocol::protocol::HookHandlerType; use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookSource; use codex_protocol::protocol::SubAgentSource; @@ -905,6 +907,8 @@ pub(crate) struct CodexHookRunMetadata { pub(crate) model_slug: Option, pub(crate) hook_name: Option, pub(crate) hook_source: Option<&'static str>, + pub(crate) handler_type: Option, + pub(crate) execution_mode: Option, pub(crate) status: Option, } @@ -1359,6 +1363,8 @@ pub(crate) fn codex_hook_run_metadata( model_slug: Some(tracking.model_slug.clone()), hook_name: Some(analytics_hook_event_name(hook.event_name).to_owned()), hook_source: Some(analytics_hook_source(hook.hook_source)), + handler_type: Some(hook.handler_type), + execution_mode: Some(hook.execution_mode), status: Some(analytics_hook_status(hook.status)), } } diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 111442e5b2..f43acd9b06 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -21,6 +21,8 @@ use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::HookEventName; +use codex_protocol::protocol::HookExecutionMode; +use codex_protocol::protocol::HookHandlerType; use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookSource; use codex_protocol::protocol::SessionSource; @@ -587,6 +589,8 @@ pub(crate) struct HookRunInput { pub struct HookRunFact { pub event_name: HookEventName, pub hook_source: HookSource, + pub handler_type: HookHandlerType, + pub execution_mode: HookExecutionMode, pub status: HookRunStatus, } diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs index eb3e8abedc..03f9122b84 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -1205,9 +1205,7 @@ async fn thread_start_fails_when_managed_hook_handler_is_unsupported() -> Result matcher = "^Bash$" [[hooks.PreToolUse.hooks]] -type = "mcp_tool" -server = "security" -tool = "scan" +type = "prompt" "#, )?; @@ -1225,7 +1223,7 @@ tool = "scan" .await??; assert!( - error.error.message.contains("MCP tool hook"), + error.error.message.contains("prompt hook"), "unexpected error message: {}", error.error.message ); @@ -1233,6 +1231,38 @@ tool = "scan" Ok(()) } +#[tokio::test] +async fn thread_start_succeeds_when_managed_mcp_hook_is_supported() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?; + std::fs::write( + codex_home.path().join("requirements.toml"), + r#"[hooks] + +[[hooks.PreToolUse]] +matcher = "^Bash$" + +[[hooks.PreToolUse.hooks]] +type = "mcp_tool" +server = "security" +tool = "scan" +"#, + )?; + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let response = app_server + .start_thread(ThreadStartParams::default()) + .await?; + + assert!(!response.thread.id.is_empty()); + + Ok(()) +} + #[tokio::test] async fn thread_start_emits_mcp_server_status_updated_notifications() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/codex-mcp/src/binding.rs b/codex-rs/codex-mcp/src/binding.rs index c1e9373ef7..862558de5b 100644 --- a/codex-rs/codex-mcp/src/binding.rs +++ b/codex-rs/codex-mcp/src/binding.rs @@ -4,6 +4,7 @@ use std::collections::HashMap; use std::fmt; use std::future::Future; use std::sync::Arc; +use std::time::Duration; use anyhow::Context; use anyhow::Result; @@ -260,22 +261,34 @@ impl PreparedMcpCall { &self, arguments: Option, meta: Option, + timeout: Option, ) -> Result { - self.call_with_preparation(|| async move { Ok((arguments, meta)) }) + self.call_with_preparation(timeout, || async move { Ok((arguments, meta)) }) .await } /// Runs irreversible call preparation and execution under the authority of /// this call's exact catalog revision and the extensions owned by the Codex session. + /// A caller-supplied timeout can further restrict the server's configured timeout. #[expect( clippy::await_holding_invalid_type, reason = "catalog replacement must remain serialized with call preparation and execution" )] - pub async fn call_with_preparation(&self, prepare: F) -> Result + pub async fn call_with_preparation( + &self, + requested_timeout: Option, + prepare: F, + ) -> Result where F: FnOnce() -> Fut, Fut: Future, Option)>>, { + let effective_timeout = match (self.client.tool_timeout, requested_timeout) { + (Some(server_timeout), Some(requested_timeout)) => { + Some(server_timeout.min(requested_timeout)) + } + (server_timeout, requested_timeout) => server_timeout.or(requested_timeout), + }; let tool_name = self.tool_info.tool.name.to_string(); let current_revision = self.catalog_revision_source.read().await; if *current_revision != self.catalog_revision { @@ -288,7 +301,7 @@ impl PreparedMcpCall { let result = self .client .client - .call_tool(tool_name.clone(), arguments, meta, self.client.tool_timeout) + .call_tool(tool_name.clone(), arguments, meta, effective_timeout) .await .with_context(|| format!("tool call failed for `{}/{tool_name}`", self.server_name))?; drop(current_revision); diff --git a/codex-rs/codex-mcp/src/binding_tests.rs b/codex-rs/codex-mcp/src/binding_tests.rs index c33d5673ff..2d04479492 100644 --- a/codex-rs/codex-mcp/src/binding_tests.rs +++ b/codex-rs/codex-mcp/src/binding_tests.rs @@ -269,6 +269,7 @@ async fn prepared_call_does_not_reroute_after_captured_connection_closes() { .call( Some(serde_json::json!({"query": "codex"})), /*meta*/ None, + /*timeout*/ None, ) .await .expect_err("a call bound to a closed connection must fail"); @@ -297,6 +298,7 @@ async fn prepared_call_is_rejected_after_catalog_refresh() { .call( Some(serde_json::json!({"query": "codex"})), /*meta*/ None, + /*timeout*/ None, ) .await .expect_err("a call from an older catalog must be rejected"); @@ -323,7 +325,7 @@ async fn stale_prepared_call_does_not_run_preparation() { let marker = Arc::clone(&prepared_side_effect_ran); prepared - .call_with_preparation(|| async move { + .call_with_preparation(/*requested_timeout*/ None, || async move { marker.store(true, Ordering::SeqCst); Ok((None, None)) }) @@ -351,7 +353,7 @@ async fn preparation_holds_catalog_authority_until_it_finishes() { let finish = Arc::clone(&finish_preparation); let call = tokio::spawn(async move { prepared - .call_with_preparation(|| async move { + .call_with_preparation(/*requested_timeout*/ None, || async move { started.notify_one(); finish.notified().await; Err(anyhow::anyhow!("stop after preparation")) diff --git a/codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs b/codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs index 4075bd616a..287b2fe675 100644 --- a/codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs +++ b/codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs @@ -325,6 +325,82 @@ impl McpConnectionSet { ) } + /// Captures an exact live client and filtered tool without starting or reconnecting a server. + /// + /// The prepared call retains the current catalog revision so subsequent execution cannot + /// silently drift to a different connection, tool catalog, or managed configuration. + pub(crate) async fn prepare_connected_call( + self: &Arc, + config: Arc, + server: &str, + tool: &str, + ) -> Option { + // Only consider servers and tools allowed by the current configuration. + let view = self.servers.get(server)?; + if !view.tool_filter.allows(tool) { + return None; + } + + // Reuse an existing connection without starting or reconnecting the server. + let mut client = view.connection.client.ready_client()?; + if client.client.is_closed().await { + return None; + } + client.tool_timeout = view.tool_timeout; + + // Snapshot the revision before reading tools so concurrent refreshes invalidate the call. + let revision = *self.tool_catalog_revision.read().await; + + // Codex Apps catalogs can refresh independently of their MCP connection. + let server_tools = if server == CODEX_APPS_MCP_SERVER_NAME { + self.codex_apps_tools_override + .read() + .await + .clone() + .unwrap_or_else(|| client.tools.clone()) + } else { + client.tools.clone() + }; + + // Filter and annotate tools using the same rules as the rest of the MCP runtime. + let server_tools = filter_tools(server_tools, &view.tool_filter); + let server_tools = if server == CODEX_APPS_MCP_SERVER_NAME { + prepare_codex_apps_tools_for_model(server_tools, &self.tool_plugin_provenance) + } else { + crate::rmcp_client::prepare_regular_mcp_tools_for_model( + server_tools, + &self.tool_plugin_provenance, + ) + }; + + // Look up the requested tool and attach its server's policy metadata. + let tool_info = server_tools + .into_iter() + .find(|tool_info| tool_info.tool.name.as_ref() == tool) + .map(|tool_info| Self::with_server_metadata(tool_info, &view.metadata))?; + + // Reject tools from connectors that an administrator disabled in managed requirements. + if server == CODEX_APPS_MCP_SERVER_NAME + && tool_info + .connector_id + .as_deref() + .and_then(|connector_id| { + config + .config_layer_stack + .requirements_toml() + .apps + .as_ref()? + .apps + .get(connector_id) + }) + .is_some_and(|requirements| requirements.enabled == Some(false)) + { + return None; + } + + self.prepare_call(&tool_info, Arc::new(client), config, revision) + } + fn prepare_call( self: &Arc, tool_info: &ToolInfo, diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index 97a5fe9a5f..be7fbc791c 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -22,12 +22,20 @@ use crate::tools::ToolFilter; use crate::tools::ToolInfo; use crate::tools::filter_tools; use crate::tools::normalize_tools_for_model_with_prefix; +use codex_config::AbsolutePathBuf; +use codex_config::AppRequirementToml; use codex_config::AppToolApproval; +use codex_config::AppsRequirementsToml; +use codex_config::CONFIG_TOML_FILE; +use codex_config::ConfigLayerStack; +use codex_config::ConfigRequirements; +use codex_config::ConfigRequirementsToml; use codex_config::Constrained; use codex_config::McpServerAuth; use codex_config::McpServerConfig; use codex_config::McpServerEnvVar; use codex_config::McpServerToolConfig; +use codex_config::TomlValue; use codex_config::types::AuthKeyringBackendKind; use codex_config::types::OAuthCredentialsStoreMode; use codex_connectors::ConnectorRuntimeContext; @@ -71,6 +79,7 @@ use rmcp::model::ServerCapabilities; use rmcp::model::ServerInfo; use rmcp::model::Tool; use rmcp::service::RequestContext; +use std::collections::BTreeMap; use std::collections::HashMap; use std::collections::HashSet; use std::io; @@ -2636,6 +2645,202 @@ async fn list_all_tools_applies_legacy_mcp_prefix_by_default() { ); } +#[tokio::test] +async fn prepare_connected_call_fails_immediately_until_server_startup_finishes() { + let client = create_test_managed_client(vec![create_test_tool("docs", "search")]).await; + let (client, startup_started, release_startup) = create_gated_async_managed_client(client); + let startup_client = client.clone(); + let approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + let permission_profile = Constrained::allow_any(PermissionProfile::default()); + let mut manager = McpConnectionSet::new_uninitialized( + &approval_policy, + &permission_profile, + /*prefix_mcp_tool_names*/ true, + ); + manager.insert_test_client("docs", client); + let manager = Arc::new(manager); + let config = Arc::new(crate::mcp::tests::test_mcp_config(std::env::temp_dir())); + + let pending_call = tokio::time::timeout( + Duration::from_millis(50), + manager.prepare_connected_call(Arc::clone(&config), "docs", "search"), + ) + .await + .expect("ready-only preparation must not wait for pending server startup"); + assert!(pending_call.is_none()); + + let startup = tokio::spawn(async move { startup_client.client().await }); + startup_started.await.expect("server startup should begin"); + release_startup.send(()).expect("release server startup"); + startup + .await + .expect("startup task should finish") + .expect("server startup should succeed"); + + let prepared_call = manager + .prepare_connected_call(config, "docs", "search") + .await + .expect("ready server should prepare its exact configured tool"); + assert_eq!( + ( + prepared_call.server_name(), + prepared_call.tool_info().tool.name.as_ref(), + ), + ("docs", "search") + ); +} + +#[tokio::test] +async fn prepare_connected_call_respects_server_tool_filters() { + let client = create_ready_async_managed_client(vec![create_test_tool("docs", "search")]).await; + client.client().await.expect("server should be ready"); + let approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + let permission_profile = Constrained::allow_any(PermissionProfile::default()); + let mut manager = McpConnectionSet::new_uninitialized( + &approval_policy, + &permission_profile, + /*prefix_mcp_tool_names*/ true, + ); + manager.insert_test_client("docs", client); + manager + .servers + .get_mut("docs") + .expect("test server should exist") + .tool_filter + .disabled + .insert("search".to_string()); + + assert!( + Arc::new(manager) + .prepare_connected_call( + Arc::new(crate::mcp::tests::test_mcp_config(std::env::temp_dir())), + "docs", + "search", + ) + .await + .is_none() + ); +} + +#[tokio::test] +async fn prepare_connected_call_captures_revision_before_codex_apps_catalog() { + let client = create_ready_async_managed_client(vec![create_test_tool( + CODEX_APPS_MCP_SERVER_NAME, + "search", + )]) + .await; + client.client().await.expect("server should be ready"); + let approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + let permission_profile = Constrained::allow_any(PermissionProfile::default()); + let mut manager = McpConnectionSet::new_uninitialized( + &approval_policy, + &permission_profile, + /*prefix_mcp_tool_names*/ true, + ); + manager.insert_test_client(CODEX_APPS_MCP_SERVER_NAME, client); + let manager = Arc::new(manager); + let config = Arc::new(crate::mcp::tests::test_mcp_config(std::env::temp_dir())); + + let catalog_override = manager.codex_apps_tools_override.write().await; + let mut preparation = + Box::pin(manager.prepare_connected_call(config, CODEX_APPS_MCP_SERVER_NAME, "search")); + + assert!( + preparation.as_mut().now_or_never().is_none(), + "preparation should wait for the Codex Apps catalog" + ); + *manager + .tool_catalog_revision + .try_write() + .expect("catalog lookup must not hold the revision lock across an await") += 1; + + drop(catalog_override); + let prepared_call = preparation + .await + .expect("ready server should prepare its exact configured tool"); + let error = prepared_call + .call( + /*arguments*/ None, /*meta*/ None, /*timeout*/ None, + ) + .await + .expect_err("catalog refresh during lookup must invalidate the prepared call"); + assert!(error.to_string().contains("catalog changed")); +} + +#[tokio::test] +async fn prepare_connected_call_rejects_only_administrator_disabled_connectors() { + let mut administrator_disabled_tool = + create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "administrator_disabled"); + administrator_disabled_tool.connector_id = Some("administrator_disabled".to_string()); + let mut user_disabled_tool = create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "user_disabled"); + user_disabled_tool.connector_id = Some("user_disabled".to_string()); + let client = + create_ready_async_managed_client(vec![administrator_disabled_tool, user_disabled_tool]) + .await; + client.client().await.expect("server should be ready"); + + let approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + let permission_profile = Constrained::allow_any(PermissionProfile::default()); + let mut manager = McpConnectionSet::new_uninitialized( + &approval_policy, + &permission_profile, + /*prefix_mcp_tool_names*/ true, + ); + manager.insert_test_client(CODEX_APPS_MCP_SERVER_NAME, client); + let manager = Arc::new(manager); + + let requirements = ConfigRequirementsToml { + apps: Some(AppsRequirementsToml { + apps: BTreeMap::from([( + "administrator_disabled".to_string(), + AppRequirementToml { + enabled: Some(false), + tools: None, + }, + )]), + }), + ..Default::default() + }; + let config_layer_stack = + ConfigLayerStack::new(Vec::new(), ConfigRequirements::default(), requirements) + .expect("config layer stack"); + let user_config: TomlValue = serde_json::from_value(serde_json::json!({ + "apps": { + "user_disabled": { + "enabled": false, + "tools": { + "user_disabled": { "enabled": false } + } + } + } + })) + .expect("user app configuration"); + let config_toml_path = AbsolutePathBuf::try_from(std::env::temp_dir().join(CONFIG_TOML_FILE)) + .expect("absolute config path"); + let mut config = crate::mcp::tests::test_mcp_config(std::env::temp_dir()); + config.config_layer_stack = config_layer_stack + .with_user_config(&config_toml_path, user_config) + .expect("user app configuration should be valid"); + let config = Arc::new(config); + + assert!( + manager + .prepare_connected_call( + Arc::clone(&config), + CODEX_APPS_MCP_SERVER_NAME, + "administrator_disabled", + ) + .await + .is_none() + ); + assert!( + manager + .prepare_connected_call(config, CODEX_APPS_MCP_SERVER_NAME, "user_disabled") + .await + .is_some() + ); +} + #[tokio::test] async fn list_all_tools_resolves_server_catalogs_concurrently() { let first_client = create_test_managed_client(vec![create_test_tool("first", "search")]).await; diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 0b56a7171f..42e17bf6ce 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -493,22 +493,22 @@ impl AsyncManagedClient { self.client.clone().await } + /// Returns the current ready client, including its tool catalog and metadata, + /// without waiting for startup or initiating a reconnection. + pub(crate) fn ready_client(&self) -> Option { + self.startup_reconnect + .as_ref() + .and_then(|reconnect| reconnect.current_client()) + .or_else(|| { + self.client + .peek() + .and_then(|result| result.as_ref().ok()) + .cloned() + }) + } + pub(crate) fn ready_transport(&self) -> Option> { - let recovered = self.startup_reconnect.as_ref().and_then(|reconnect| { - reconnect - .state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .current_client - .as_ref() - .map(|client| Arc::clone(&client.client)) - }); - recovered.or_else(|| { - self.client - .peek() - .and_then(|result| result.as_ref().ok()) - .map(|client| Arc::clone(&client.client)) - }) + self.ready_client().map(|client| client.client) } pub(crate) async fn reconnect_failed_startup(&self) { diff --git a/codex-rs/codex-mcp/src/runtime.rs b/codex-rs/codex-mcp/src/runtime.rs index 3dbfb460ed..b9ca091ee6 100644 --- a/codex-rs/codex-mcp/src/runtime.rs +++ b/codex-rs/codex-mcp/src/runtime.rs @@ -45,6 +45,7 @@ use tokio_util::sync::CancellationToken; use crate::McpConfig; use crate::binding::McpBinding; +use crate::binding::PreparedMcpCall; use crate::connection_manager::McpConnectionSet; use crate::elicitation::ElicitationLifecycle; use crate::elicitation::ElicitationRequestRouter; @@ -408,6 +409,24 @@ impl McpRuntime { Self::binding_from_published_runtime(current, /*required_servers*/ &[]).await } + /// Prepares an exact MCP tool call only when its server is already connected. + /// + /// Missing, disconnected, filtered, or uncataloged tools return `None` immediately rather + /// than starting a server, waiting for startup, initiating OAuth, or reconnecting. Successful + /// calls capture the live client, effective tool policy, and current catalog revision together. + pub async fn prepare_call_if_connected( + &self, + server: &str, + tool: &str, + ) -> Option { + let current = self.current.load_full(); + let config = Arc::clone(current.config.as_ref()?); + current + .connections + .prepare_connected_call(config, server, tool) + .await + } + /// Returns the latest published configuration without waiting for clients. pub fn current_config(&self) -> Option> { self.current.load().config.clone() diff --git a/codex-rs/core/src/hook_mcp_executor.rs b/codex-rs/core/src/hook_mcp_executor.rs new file mode 100644 index 0000000000..68523234bc --- /dev/null +++ b/codex-rs/core/src/hook_mcp_executor.rs @@ -0,0 +1,58 @@ +use std::sync::Arc; + +use anyhow::Context; +use anyhow::bail; +use codex_hooks::HookMcpCall; +use codex_hooks::HookMcpExecutor; +use codex_mcp::McpRuntime; +use codex_protocol::ThreadId; +use futures::FutureExt; +use futures::future::BoxFuture; +use serde_json::Value; + +pub(crate) struct CoreHookMcpExecutor { + pub(crate) runtime: Arc, + // Session-scoped MCP tools require the owning thread ID in request metadata. + pub(crate) thread_id: ThreadId, +} + +impl HookMcpExecutor for CoreHookMcpExecutor { + fn execute(&self, call: HookMcpCall) -> BoxFuture<'_, anyhow::Result> { + async move { + let prepared_call = self + .runtime + .prepare_call_if_connected(&call.server, &call.tool) + .await + .with_context(|| { + format!( + "MCP server `{}` or tool `{}` is not connected and available", + call.server, call.tool + ) + })?; + + let result = prepared_call + .call( + Some(Value::Object(call.input)), + Some(serde_json::json!({ "threadId": self.thread_id.to_string() })), + Some(call.timeout), + ) + .await?; + let text = result + .content + .iter() + .filter_map(|content| { + (content.get("type").and_then(Value::as_str) == Some("text")) + .then(|| content.get("text").and_then(Value::as_str)) + .flatten() + }) + .collect::>() + .join("\n"); + if result.is_error == Some(true) { + bail!("MCP tool returned an error: {text}"); + } + + Ok(text) + } + .boxed() + } +} diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 279edbf9d2..af2a31e944 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -19,6 +19,8 @@ use codex_hooks::StopOutcome; use codex_hooks::SubagentHookContext; use codex_hooks::UserPromptSubmitOutcome; use codex_hooks::UserPromptSubmitRequest; +use codex_hooks::hook_execution_mode_label; +use codex_hooks::hook_handler_type_label; use codex_otel::HOOK_RUN_DURATION_METRIC; use codex_otel::HOOK_RUN_METRIC; use codex_protocol::items::TurnItem; @@ -30,6 +32,7 @@ use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::HookEventName; use codex_protocol::protocol::HookExecutionMode; +use codex_protocol::protocol::HookHandlerType; use codex_protocol::protocol::HookOutputEntryKind; use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookRunSummary; @@ -38,6 +41,7 @@ use codex_protocol::protocol::HookStartedEvent; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::WarningEvent; +use codex_rollout::state_db; use codex_thread_store::PersistContext; use codex_thread_store::ReadThreadParams; use serde_json::Value; @@ -722,6 +726,23 @@ pub(crate) async fn emit_hook_completed_events( turn_context: &Arc, completed_events: Vec, ) { + if turn_context.config.memories.disable_on_external_context + && completed_events.iter().any(|completed| { + completed.run.handler_type == HookHandlerType::McpTool + && matches!( + completed.run.status, + HookRunStatus::Completed | HookRunStatus::Blocked | HookRunStatus::Stopped + ) + }) + { + state_db::mark_thread_memory_mode_polluted( + sess.services.state_db.as_deref(), + sess.thread_id, + "mcp_tool_hook", + ) + .await; + } + for completed in completed_events { emit_hook_completed_metrics(turn_context, &completed); track_hook_completed_analytics(sess, turn_context, &completed); @@ -778,12 +799,14 @@ fn hook_run_analytics_payload( HookRunFact { event_name: completed.run.event_name, hook_source: completed.run.source, + handler_type: completed.run.handler_type, + execution_mode: completed.run.execution_mode, status: completed.run.status, }, ) } -fn hook_run_metric_tags(run: &HookRunSummary) -> [(&'static str, &'static str); 3] { +fn hook_run_metric_tags(run: &HookRunSummary) -> [(&'static str, &'static str); 5] { let hook_name = match run.event_name { HookEventName::PreToolUse => "PreToolUse", HookEventName::PermissionRequest => "PermissionRequest", @@ -817,11 +840,15 @@ fn hook_run_metric_tags(run: &HookRunSummary) -> [(&'static str, &'static str); HookRunStatus::Blocked => "blocked", HookRunStatus::Stopped => "stopped", }; - [ ("hook_name", hook_name), ("source", hook_source), ("status", status), + ("handler_type", hook_handler_type_label(run.handler_type)), + ( + "execution_mode", + hook_execution_mode_label(run.execution_mode), + ), ] } @@ -988,6 +1015,8 @@ mod tests { assert_eq!(tracking.turn_id, "turn-from-hook"); assert_eq!(tracking.model_slug, turn_context.model_info.slug); assert_eq!(hook.event_name, HookEventName::Stop); + assert_eq!(hook.handler_type, HookHandlerType::Command); + assert_eq!(hook.execution_mode, HookExecutionMode::Sync); assert_eq!(hook.hook_source, HookSource::Project); assert_eq!(hook.status, HookRunStatus::Blocked); } @@ -995,22 +1024,25 @@ mod tests { #[tokio::test] async fn hook_run_analytics_payload_falls_back_to_turn_context_id() { let (_session, turn_context) = make_session_and_context().await; - let completed = HookCompletedEvent { - turn_id: None, - run: sample_hook_run(HookRunStatus::Failed, HookSource::Unknown), - }; + let mut run = sample_hook_run(HookRunStatus::Failed, HookSource::Unknown); + run.handler_type = HookHandlerType::Prompt; + run.execution_mode = HookExecutionMode::Async; + let completed = HookCompletedEvent { turn_id: None, run }; let (tracking, hook) = hook_run_analytics_payload("thread-123".to_string(), &turn_context, &completed); assert_eq!(tracking.turn_id, turn_context.sub_id); + assert_eq!(hook.handler_type, HookHandlerType::Prompt); + assert_eq!(hook.execution_mode, HookExecutionMode::Async); assert_eq!(hook.hook_source, HookSource::Unknown); assert_eq!(hook.status, HookRunStatus::Failed); } #[test] fn hook_run_metric_tags_match_analytics_shape() { - let run = sample_hook_run(HookRunStatus::Blocked, HookSource::Project); + let mut run = sample_hook_run(HookRunStatus::Blocked, HookSource::Project); + run.handler_type = HookHandlerType::McpTool; assert_eq!( hook_run_metric_tags(&run), @@ -1018,6 +1050,8 @@ mod tests { ("hook_name", "Stop"), ("source", "project"), ("status", "blocked"), + ("handler_type", "mcp_tool"), + ("execution_mode", "sync"), ] ); @@ -1030,13 +1064,16 @@ mod tests { ("hook_name", "Stop"), ("source", "cloud_requirements"), ("status", "blocked"), + ("handler_type", "command"), + ("execution_mode", "sync"), ] ); } #[test] fn hook_run_metric_tags_include_expanded_hook_sources() { - let run = sample_hook_run(HookRunStatus::Completed, HookSource::LegacyManagedConfigMdm); + let mut run = sample_hook_run(HookRunStatus::Completed, HookSource::LegacyManagedConfigMdm); + run.execution_mode = HookExecutionMode::Async; assert_eq!( hook_run_metric_tags(&run), @@ -1044,6 +1081,8 @@ mod tests { ("hook_name", "Stop"), ("source", "legacy_managed_config_mdm"), ("status", "completed"), + ("handler_type", "command"), + ("execution_mode", "async"), ] ); } diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 43e3bc1e02..b1dc960722 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -61,6 +61,7 @@ mod exec_policy; #[cfg(test)] mod git_info_tests; mod guardian; +mod hook_mcp_executor; mod hook_runtime; mod image_preparation; mod installation_id; diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index 73decd4504..58ecdac2e5 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -409,7 +409,7 @@ async fn handle_approved_mcp_tool_call( let result = async { let result = async { let result = prepared_call - .call_with_preparation(|| async { + .call_with_preparation(/*requested_timeout*/ None, || async { if let McpToolApprovalApplication::Apply { decision, policy } = &approval_application { diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 3cd72aabb7..bde2fda59a 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -4188,7 +4188,6 @@ async fn build_hooks_config( plugin_hook_load_warnings, shell_program: hook_shell_program, shell_args: hook_shell_argv, - mcp_executor: None, } } diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 4a4d6dcfee..0a74430fd5 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -5,6 +5,7 @@ use crate::agents_md_manager::AgentsMdManager; use crate::config::ConstraintError; use crate::environment_selection::ThreadEnvironments; use crate::environment_selection::TurnEnvironmentSnapshot; +use crate::hook_mcp_executor::CoreHookMcpExecutor; use crate::shell_snapshot::ShellSnapshot; use crate::state::ActiveTurn; use codex_extension_api::ExtensionDataInit; @@ -1196,13 +1197,24 @@ impl Session { (None, None) }; + // Hooks and extensions share one stable thread-owned MCP runtime handle. + let mcp_runtime = Arc::new(McpRuntime::empty( + mcp_projection.config.prefix_mcp_tool_names, + )); let hooks_config = build_hooks_config( &config, plugins_manager.as_ref(), resolved_environments.single_local_environment(), ) .await; - let (hooks, async_hook_results) = Hooks::new(hooks_config, thread_id)?; + let (hooks, async_hook_results) = Hooks::new( + hooks_config, + thread_id, + Arc::new(CoreHookMcpExecutor { + runtime: Arc::clone(&mcp_runtime), + thread_id, + }), + )?; for warning in hooks.startup_warnings() { post_session_configured_events.push(Event { id: INITIAL_SUBMIT_ID.to_owned(), @@ -1219,10 +1231,6 @@ impl Session { config.analytics_enabled, ) }); - // Extensions need a stable thread-owned resource client before the Session exists. - let mcp_runtime = Arc::new(McpRuntime::empty( - mcp_projection.config.prefix_mcp_tool_names, - )); for item in initial_history.get_rollout_items() { match item { RolloutItem::Compacted(compacted) => { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 222ea7d639..8e0638e609 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -11,6 +11,7 @@ use crate::environment_selection::EnvironmentConfigOrigin; use crate::environment_selection::ThreadEnvironments; use crate::environment_selection::TurnEnvironmentState; use crate::function_tool::FunctionCallError; +use crate::hook_mcp_executor::CoreHookMcpExecutor; use crate::plugins::plugins_manager_for_config; use crate::session::step_context::StepContext; use crate::shell::default_user_shell; @@ -814,6 +815,10 @@ async fn preview_session_start_hooks( ..HooksConfig::default() }, thread_id, + Arc::new(CoreHookMcpExecutor { + runtime: Arc::new(McpRuntime::empty(config.prefix_mcp_tool_names())), + thread_id, + }), ) .expect("initialize hooks for session-start preview"); @@ -5888,6 +5893,10 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { ..HooksConfig::default() }, thread_id, + Arc::new(CoreHookMcpExecutor { + runtime: Arc::clone(&mcp_runtime), + thread_id, + }), ) .expect("initialize test hooks"); let services = SessionServices { @@ -8093,6 +8102,10 @@ where ..HooksConfig::default() }, thread_id, + Arc::new(CoreHookMcpExecutor { + runtime: Arc::clone(&mcp_runtime), + thread_id, + }), ) .expect("initialize test hooks"); let services = SessionServices { diff --git a/codex-rs/core/tests/suite/hooks_mcp.rs b/codex-rs/core/tests/suite/hooks_mcp.rs index 9ea3b10c37..133a0d6d92 100644 --- a/codex-rs/core/tests/suite/hooks_mcp.rs +++ b/codex-rs/core/tests/suite/hooks_mcp.rs @@ -233,6 +233,33 @@ print({python_output_literal}) Ok(()) } +fn write_mcp_tool_hook( + home: &Path, + event_name: &str, + matcher: Option<&str>, + server: &str, + output: &str, +) -> Result<()> { + let hooks = json!({ + "hooks": { + event_name: [{ + "matcher": matcher, + "hooks": [{ + "type": "mcp_tool", + "server": server, + "tool": "image_scenario", + "input": { + "scenario": "text_only", + "caption": output, + }, + "timeout": 5, + }], + }], + }, + }); + fs::write(home.join("hooks.json"), hooks.to_string()).context("write MCP tool hook config") +} + fn read_hook_inputs(home: &Path, log_name: &str) -> Result> { fs::read_to_string(home.join(log_name)) .with_context(|| format!("read {log_name}"))? @@ -421,6 +448,297 @@ async fn run_mcp_permission_request_hook_test(outcome: PermissionRequestHookOutc Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mcp_tool_hook_interpolates_prompt_and_runs_without_tool_approval() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let response = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "hook context received"), + ev_completed("resp-1"), + ]), + ) + .await; + let rmcp_test_server_bin = stdio_server_bin()?; + let test = test_codex() + .with_pre_build_hook(|home| { + write_mcp_tool_hook( + home, + "UserPromptSubmit", + /*matcher*/ None, + RMCP_SERVER, + r#"{"hookSpecificOutput":{"hookEventName":"UserPromptSubmit","additionalContext":"MCP scanner checked ${prompt}"}}"#, + ) + .expect("write MCP prompt hook fixture"); + }) + .with_config(move |config| { + enable_hooks_and_rmcp_server( + config, + rmcp_test_server_bin, + AppToolApproval::Prompt, + /*prefix_mcp_tool_names*/ true, + ); + }) + .build(&server) + .await?; + wait_for_mcp_server(&test.codex, RMCP_SERVER).await?; + + tokio::time::timeout( + Duration::from_secs(15), + test.submit_turn("review checkout.rs"), + ) + .await + .context("MCP hook must not wait for model-tool approval")??; + + assert!( + response + .single_request() + .message_input_texts("developer") + .contains(&"MCP scanner checked review checkout.rs".to_string()) + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mcp_tool_hook_passes_thread_metadata_to_model_hidden_tools() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let response = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "thread context received"), + ev_completed("resp-1"), + ]), + ) + .await; + let rmcp_test_server_bin = stdio_server_bin()?; + let test = test_codex() + .with_pre_build_hook(|home| { + let hooks = json!({ + "hooks": { + "UserPromptSubmit": [{ + "hooks": [{ + "type": "mcp_tool", + "server": RMCP_SERVER, + "tool": "thread_hint", + "input": {}, + "timeout": 5, + }], + }], + }, + }); + fs::write(home.join("hooks.json"), hooks.to_string()) + .expect("write model-hidden MCP tool hook config"); + }) + .with_config(move |config| { + enable_hooks_and_rmcp_server( + config, + rmcp_test_server_bin, + AppToolApproval::Prompt, + /*prefix_mcp_tool_names*/ true, + ); + }) + .build(&server) + .await?; + wait_for_mcp_server(&test.codex, RMCP_SERVER).await?; + + test.submit_turn("load thread context").await?; + + let request = response.single_request(); + let expected_context = format!( + "manual history hint for thread {}\nunstructured notes/thread_hint fixture result", + test.session_configured.thread_id, + ); + assert!( + request + .message_input_texts("developer") + .contains(&expected_context) + ); + assert!( + request + .tool_by_name(RMCP_PREFIXED_NAMESPACE, "thread_hint") + .is_none(), + "the hook tool must remain hidden from the model" + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mcp_tool_hook_marks_thread_memory_mode_polluted_when_configured() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "hook context received"), + ev_completed("resp-1"), + ]), + ) + .await; + let rmcp_test_server_bin = stdio_server_bin()?; + let test = test_codex() + .with_pre_build_hook(|home| { + write_mcp_tool_hook( + home, + "UserPromptSubmit", + /*matcher*/ None, + RMCP_SERVER, + r#"{"hookSpecificOutput":{"hookEventName":"UserPromptSubmit","additionalContext":"external MCP context"}}"#, + ) + .expect("write MCP prompt hook fixture"); + }) + .with_config(move |config| { + enable_hooks_and_rmcp_server( + config, + rmcp_test_server_bin, + AppToolApproval::Prompt, + /*prefix_mcp_tool_names*/ true, + ); + config + .features + .enable(Feature::Sqlite) + .expect("test config should allow feature update"); + config.memories.disable_on_external_context = true; + }) + .build(&server) + .await?; + wait_for_mcp_server(&test.codex, RMCP_SERVER).await?; + + let db = test.codex.state_db().expect("state db enabled"); + let thread_id = test.session_configured.thread_id; + test.submit_turn("review checkout.rs").await?; + + assert_eq!( + db.get_thread_memory_mode(thread_id).await?, + Some("polluted".to_string()) + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mcp_tool_hook_blocks_model_tool_without_recursive_hooks_or_approval() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let call_id = "mcp-hook-denied-echo"; + let arguments = json!({ "message": RMCP_ECHO_MESSAGE }).to_string(); + let responses = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_function_call_with_namespace( + call_id, + RMCP_PREFIXED_NAMESPACE, + "echo", + &arguments, + ), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-1", "the MCP hook blocked that tool"), + ev_completed("resp-2"), + ]), + ], + ) + .await; + let rmcp_test_server_bin = stdio_server_bin()?; + let test = test_codex() + .with_pre_build_hook(|home| { + write_mcp_tool_hook( + home, + "PreToolUse", + Some("mcp__rmcp__.*"), + RMCP_SERVER, + r#"{"hookSpecificOutput":{"hookEventName":"PreToolUse","permissionDecision":"deny","permissionDecisionReason":"MCP blocked ${tool_input.message}"}}"#, + ) + .expect("write MCP denial hook fixture"); + }) + .with_config(move |config| { + enable_hooks_and_rmcp_server( + config, + rmcp_test_server_bin, + AppToolApproval::Prompt, + /*prefix_mcp_tool_names*/ true, + ); + }) + .build(&server) + .await?; + wait_for_mcp_server(&test.codex, RMCP_SERVER).await?; + + tokio::time::timeout( + Duration::from_secs(15), + test.submit_turn("call the blocked MCP tool"), + ) + .await + .context("MCP hook must not recurse or request model-tool approval")??; + + let requests = responses.requests(); + assert_eq!(requests.len(), 2); + let output = requests[1].function_call_output(call_id); + let output = output["output"] + .as_str() + .context("blocked model tool output")?; + assert!(output.contains("MCP blocked hook e2e ping")); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mcp_tool_hook_fails_open_when_server_is_unavailable() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let response = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "continued without the missing server"), + ev_completed("resp-1"), + ]), + ) + .await; + let test = test_codex() + .with_pre_build_hook(|home| { + write_mcp_tool_hook( + home, + "UserPromptSubmit", + /*matcher*/ None, + "missing-server", + r#"{"hookSpecificOutput":{"hookEventName":"UserPromptSubmit","additionalContext":"this must not appear"}}"#, + ) + .expect("write unavailable MCP server fixture"); + }) + .with_config(trust_discovered_hooks) + .build(&server) + .await?; + + tokio::time::timeout(Duration::from_secs(5), test.submit_turn("continue")) + .await + .context("unavailable MCP hook server must fail immediately")??; + + assert!( + !response + .single_request() + .message_input_texts("developer") + .iter() + .any(|message| message.contains("this must not appear")) + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn pre_tool_use_blocks_mcp_tool_before_execution_with_legacy_prefixed_names() -> Result<()> { pre_tool_use_blocks_mcp_tool_before_execution( diff --git a/codex-rs/hooks/src/engine/command_runner_tests.rs b/codex-rs/hooks/src/engine/command_runner_tests.rs index 6cee21aede..eae36b45ee 100644 --- a/codex-rs/hooks/src/engine/command_runner_tests.rs +++ b/codex-rs/hooks/src/engine/command_runner_tests.rs @@ -22,6 +22,7 @@ use tokio::time::timeout; use super::super::ClaudeHooksEngine; use super::super::ConfiguredHandlerKind; +use super::super::tests::mcp_executor; use super::CommandHookRuntime; use super::CommandShell; use super::ConfiguredHandler; @@ -199,7 +200,7 @@ async fn schedule(runtime: &CommandHookRuntime, handler: ConfiguredHandler, cwd: warnings: Vec::new(), required_load_errors: Vec::new(), command_runtime: runtime.clone(), - mcp_executor: None, + mcp_executor: mcp_executor(), }; engine .run_user_prompt_submit(UserPromptSubmitRequest { diff --git a/codex-rs/hooks/src/engine/discovery.rs b/codex-rs/hooks/src/engine/discovery.rs index 2c730a8a09..7fad77da2e 100644 --- a/codex-rs/hooks/src/engine/discovery.rs +++ b/codex-rs/hooks/src/engine/discovery.rs @@ -603,17 +603,6 @@ fn append_matcher_groups( ); continue; } - if matches!(&source.requirement, HookRequirement::Required(_)) { - source.record_load_failure( - format!( - "skipping MCP tool hook in {}: MCP tool hooks are not supported yet", - source.path.display() - ), - warnings, - ); - continue; - } - let timeout_sec = timeout_sec.unwrap_or(600).max(1); let config = HookHandlerConfig::McpTool { server: server.clone(), diff --git a/codex-rs/hooks/src/engine/dispatcher.rs b/codex-rs/hooks/src/engine/dispatcher.rs index 8361de803a..a024f75939 100644 --- a/codex-rs/hooks/src/engine/dispatcher.rs +++ b/codex-rs/hooks/src/engine/dispatcher.rs @@ -129,20 +129,24 @@ pub(crate) async fn execute_handlers( tool, input, } => { - let executor = engine.mcp_executor.as_deref()?; - run_mcp_tool(executor, &handler, server, tool, input, &input_json).await + run_mcp_tool( + engine.mcp_executor.as_ref(), + &handler, + server, + tool, + input, + &input_json, + ) + .await } }; - Some((configured_order, parse(&handler, result, turn_id))) + (configured_order, parse(&handler, result, turn_id)) }); } let mut completed = Vec::new(); let mut completion_order = 0; - while let Some(result) = pending.next().await { - let Some((configured_order, mut parsed)) = result else { - continue; - }; + while let Some((configured_order, mut parsed)) = pending.next().await { parsed.completion_order = completion_order; completion_order += 1; completed.push((configured_order, parsed)); @@ -207,14 +211,16 @@ pub(crate) fn hook_event_name_label(event_name: HookEventName) -> &'static str { } } -pub(crate) fn hook_execution_mode_label(mode: HookExecutionMode) -> &'static str { +/// Returns the canonical label for a hook execution mode. +pub fn hook_execution_mode_label(mode: HookExecutionMode) -> &'static str { match mode { HookExecutionMode::Sync => "sync", HookExecutionMode::Async => "async", } } -pub(crate) fn hook_handler_type_label(handler_type: HookHandlerType) -> &'static str { +/// Returns the canonical label for a hook handler type. +pub fn hook_handler_type_label(handler_type: HookHandlerType) -> &'static str { match handler_type { HookHandlerType::Command => "command", HookHandlerType::McpTool => "mcp_tool", diff --git a/codex-rs/hooks/src/engine/mod.rs b/codex-rs/hooks/src/engine/mod.rs index c9f44ece9e..513dc10fac 100644 --- a/codex-rs/hooks/src/engine/mod.rs +++ b/codex-rs/hooks/src/engine/mod.rs @@ -162,7 +162,7 @@ pub(crate) struct ClaudeHooksEngine { warnings: Vec, required_load_errors: Vec, pub(crate) command_runtime: CommandHookRuntime, - mcp_executor: Option>, + pub(crate) mcp_executor: Arc, } impl ClaudeHooksEngine { @@ -173,7 +173,7 @@ impl ClaudeHooksEngine { plugin_hook_sources: Vec, plugin_hook_load_warnings: Vec, command_runtime: CommandHookRuntime, - mcp_executor: Option>, + mcp_executor: Arc, ) -> Self { if !enabled { return Self { @@ -186,25 +186,12 @@ impl ClaudeHooksEngine { } let _ = schema_loader::generated_hook_schemas(); - let mut discovered = discovery::discover_handlers( + let discovered = discovery::discover_handlers( config_layer_stack, plugin_hook_sources, plugin_hook_load_warnings, bypass_hook_trust, ); - if mcp_executor.is_none() { - discovered.handlers.retain(|handler| { - if matches!(&handler.kind, ConfiguredHandlerKind::McpTool { .. }) { - discovered.warnings.push(format!( - "skipping MCP tool hook in {}: MCP invocation is not available yet", - handler.source_path.display() - )); - false - } else { - true - } - }); - } Self { handlers: discovered.handlers, warnings: discovered.warnings, diff --git a/codex-rs/hooks/src/engine/mod_tests.rs b/codex-rs/hooks/src/engine/mod_tests.rs index baf486d914..80b470563b 100644 --- a/codex-rs/hooks/src/engine/mod_tests.rs +++ b/codex-rs/hooks/src/engine/mod_tests.rs @@ -54,6 +54,13 @@ fn command_runtime(shell: CommandShell) -> CommandHookRuntime { CommandHookRuntime::new(shell, ThreadId::new(), result_sender) } +pub(crate) fn mcp_executor() -> Arc { + Arc::new(StaticMcpExecutor { + calls: Arc::new(Mutex::new(Vec::new())), + output: String::new(), + }) +} + #[test] fn permission_request_timeout_only_counts_synchronous_handlers() { let mut engine = ClaudeHooksEngine::new( @@ -66,7 +73,7 @@ fn permission_request_timeout_only_counts_synchronous_handlers() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); let command = "echo synchronous permission hook"; let synchronous_handler = ConfiguredHandler { @@ -245,6 +252,7 @@ fn required_managed_hooks_allow_disabled_hooks_feature() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .expect("disabled hooks feature should not enforce managed requirements hooks"); @@ -268,6 +276,7 @@ fn required_managed_hooks_reject_invalid_matchers() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .err() .expect("invalid required managed matcher should reject startup"); @@ -295,6 +304,7 @@ fn required_managed_hooks_allow_invalid_matchers_without_handlers() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .expect("an empty matcher group should not prevent required managed hooks from loading"); @@ -317,6 +327,7 @@ fn required_managed_hooks_reject_empty_commands() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .err() .expect("empty required managed command should reject startup"); @@ -330,13 +341,7 @@ fn required_managed_hooks_reject_unsupported_handler_types() { let events = HookEventsToml { pre_tool_use: vec![MatcherGroup { matcher: Some("^Bash$".to_string()), - hooks: vec![HookHandlerConfig::McpTool { - server: "policy".to_string(), - tool: "check".to_string(), - input: Default::default(), - timeout_sec: None, - status_message: None, - }], + hooks: vec![HookHandlerConfig::Prompt {}], }], ..Default::default() }; @@ -352,6 +357,7 @@ fn required_managed_hooks_reject_unsupported_handler_types() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .err() .expect("unsupported required managed handler should reject startup"); @@ -359,7 +365,7 @@ fn required_managed_hooks_reject_unsupported_handler_types() { assert!( error .to_string() - .contains("MCP tool hooks are not supported yet") + .contains("prompt hooks are not supported yet") ); } @@ -391,6 +397,7 @@ fn required_managed_mcp_hooks_reject_empty_targets() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .err() .expect("invalid required managed MCP hook should reject startup"); @@ -430,6 +437,7 @@ fn required_managed_session_end_mcp_hooks_reject_startup() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .err() .expect("required managed SessionEnd MCP hook should reject startup"); @@ -456,6 +464,7 @@ fn required_managed_hooks_with_unknown_source_still_reject_discovery_failures() ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .err() .expect("unknown-source managed requirements hook should still reject startup"); @@ -478,6 +487,7 @@ fn valid_required_managed_hooks_allow_startup() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .expect("valid managed requirements hook should allow startup"); @@ -506,6 +516,7 @@ fn managed_config_layer_hook_failures_remain_startup_warnings() { ..Default::default() }, ThreadId::new(), + mcp_executor(), ) .expect("managed config layer hooks should remain optional"); @@ -580,7 +591,7 @@ with Path(r"{log_path}").open("a", encoding="utf-8") as handle: program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.warnings().is_empty()); @@ -598,7 +609,6 @@ with Path(r"{log_path}").open("a", encoding="utf-8") as handle: plugin_hook_load_warnings: Vec::new(), shell_program: None, shell_args: Vec::new(), - mcp_executor: None, }); assert!(listed.hooks[0].is_managed); let cwd = cwd(); @@ -689,7 +699,7 @@ async fn requirements_managed_hooks_execute_windows_command_override() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); let outcome = engine @@ -770,7 +780,7 @@ fn unknown_requirement_source_hooks_stay_managed() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert_eq!(engine.handlers.len(), 1); @@ -854,7 +864,7 @@ fn user_disablement_filters_non_managed_hooks_but_not_managed_hooks() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert_eq!(engine.handlers.len(), 1); @@ -921,7 +931,7 @@ fn user_disablement_does_not_filter_managed_layer_hooks() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert_eq!(engine.handlers.len(), 1); @@ -1084,7 +1094,7 @@ fn requirements_managed_hooks_load_when_managed_dir_is_missing() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.warnings().is_empty()); @@ -1148,7 +1158,7 @@ fn allow_managed_hooks_only_false_keeps_unmanaged_hooks() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.warnings().is_empty()); @@ -1206,7 +1216,7 @@ fn allow_managed_hooks_only_in_config_toml_does_not_enable_policy() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.warnings().is_empty()); @@ -1280,7 +1290,7 @@ fn allow_managed_hooks_only_skips_unmanaged_json_and_toml_hooks() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.handlers.is_empty()); @@ -1320,7 +1330,7 @@ fn allow_managed_hooks_only_skips_unmanaged_plugin_hooks() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.handlers.is_empty()); @@ -1393,7 +1403,7 @@ fn allow_managed_hooks_only_keeps_managed_requirement_and_config_layer_hooks() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.warnings().is_empty()); @@ -1507,7 +1517,7 @@ fn discovers_hooks_from_json_and_toml_in_the_same_layer() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.warnings().iter().any(|warning| { @@ -1603,7 +1613,7 @@ fn profile_user_layers_load_shared_hooks_json_once() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.warnings().is_empty()); @@ -1678,7 +1688,7 @@ fn malformed_hooks_json_is_reported_as_startup_warning() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert!(engine.handlers.is_empty()); @@ -1751,7 +1761,7 @@ print(json.dumps({ program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); let preview = engine.preview_pre_tool_use(&PreToolUseRequest { @@ -1779,7 +1789,6 @@ print(json.dumps({ plugin_hook_load_warnings: Vec::new(), shell_program: None, shell_args: Vec::new(), - mcp_executor: None, }); assert_eq!( listed.hooks[0].plugin_id.as_deref(), @@ -1873,7 +1882,7 @@ fn plugin_hook_sources_expand_plugin_placeholders() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert_eq!( @@ -1918,7 +1927,7 @@ fn plugin_hook_load_warnings_are_startup_warnings() { program: String::new(), args: Vec::new(), }), - /*mcp_executor*/ None, + mcp_executor(), ); assert_eq!(engine.warnings(), &["failed plugin hook".to_string()]); @@ -1976,22 +1985,19 @@ async fn mcp_tool_hooks_expand_event_input_and_apply_pre_tool_decisions() { ) .expect("config layer stack"); - let unavailable = ClaudeHooksEngine::new( - /*enabled*/ true, - /*bypass_hook_trust*/ true, - Some(&config_layer_stack), - Vec::new(), - Vec::new(), - command_runtime(CommandShell { - program: String::new(), - args: Vec::new(), - }), - /*mcp_executor*/ None, - ); - assert!(unavailable.handlers.is_empty()); - assert_eq!(unavailable.warnings().len(), 1); - assert!(unavailable.warnings()[0].contains("MCP invocation is not available")); - + let request = PreToolUseRequest { + session_id: ThreadId::new(), + turn_id: "turn-1".to_string(), + subagent: None, + cwd: cwd(), + transcript_path: None, + model: "gpt-test".to_string(), + permission_mode: "default".to_string(), + tool_name: "Bash".to_string(), + matcher_aliases: Vec::new(), + tool_use_id: "tool-1".to_string(), + tool_input: serde_json::json!({ "command": "rm important.txt" }), + }; let calls = Arc::new(Mutex::new(Vec::new())); let executor = StaticMcpExecutor { calls: Arc::clone(&calls), @@ -2014,23 +2020,9 @@ async fn mcp_tool_hooks_expand_event_input_and_apply_pre_tool_decisions() { program: String::new(), args: Vec::new(), }), - Some(Arc::new(executor)), + Arc::new(executor), ); - let outcome = engine - .run_pre_tool_use(PreToolUseRequest { - session_id: ThreadId::new(), - turn_id: "turn-1".to_string(), - subagent: None, - cwd: cwd(), - transcript_path: None, - model: "gpt-test".to_string(), - permission_mode: "default".to_string(), - tool_name: "Bash".to_string(), - matcher_aliases: Vec::new(), - tool_use_id: "tool-1".to_string(), - tool_input: serde_json::json!({ "command": "rm important.txt" }), - }) - .await; + let outcome = engine.run_pre_tool_use(request).await; assert!(outcome.should_block); assert_eq!( diff --git a/codex-rs/hooks/src/lib.rs b/codex-rs/hooks/src/lib.rs index 7536bda0b4..a2cd3f3b03 100644 --- a/codex-rs/hooks/src/lib.rs +++ b/codex-rs/hooks/src/lib.rs @@ -16,6 +16,8 @@ pub use declarations::PluginHookDeclaration; pub use declarations::plugin_hook_declarations; pub use engine::HookListEntry; pub use engine::HookListEntryHandler; +pub use engine::dispatcher::hook_execution_mode_label; +pub use engine::dispatcher::hook_handler_type_label; pub use events::common::SubagentHookContext; /// Hook event names as they appear in hooks JSON and config files. pub const HOOK_EVENT_NAMES: [&str; 11] = [ diff --git a/codex-rs/hooks/src/registry.rs b/codex-rs/hooks/src/registry.rs index f45a645e3d..62648c7b28 100644 --- a/codex-rs/hooks/src/registry.rs +++ b/codex-rs/hooks/src/registry.rs @@ -44,7 +44,6 @@ pub struct HooksConfig { pub plugin_hook_load_warnings: Vec, pub shell_program: Option, pub shell_args: Vec, - pub mcp_executor: Option>, } #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -65,9 +64,10 @@ impl Hooks { pub fn new( config: HooksConfig, thread_id: ThreadId, + mcp_executor: Arc, ) -> anyhow::Result<(Self, Receiver)> { let (result_sender, result_receiver) = async_channel::unbounded(); - let hooks = Self::from_config(config, |shell| { + let hooks = Self::from_config(config, mcp_executor, |shell| { CommandHookRuntime::new(shell, thread_id, result_sender) }); let required_load_errors = hooks.engine.required_load_errors(); @@ -82,13 +82,14 @@ impl Hooks { /// Preserve in-flight background hooks while applying a refreshed configuration. pub fn reconfigured(&self, config: HooksConfig) -> Self { - Self::from_config(config, |shell| { + Self::from_config(config, Arc::clone(&self.engine.mcp_executor), |shell| { self.engine.command_runtime.reconfigured(shell) }) } fn from_config( config: HooksConfig, + mcp_executor: Arc, build_runtime: impl FnOnce(CommandShell) -> CommandHookRuntime, ) -> Self { let after_agent = config @@ -108,7 +109,7 @@ impl Hooks { config.plugin_hook_sources, config.plugin_hook_load_warnings, command_runtime, - config.mcp_executor, + mcp_executor, ); Self { after_agent,