diff --git a/codex-rs/config/src/hook_config.rs b/codex-rs/config/src/hook_config.rs index 7f2d61bc47..7575fa5fcb 100644 --- a/codex-rs/config/src/hook_config.rs +++ b/codex-rs/config/src/hook_config.rs @@ -125,20 +125,27 @@ impl HookEventsToml { .sum() } - pub fn into_matcher_groups(self) -> [(HookEventName, Vec); 12] { + pub fn into_matcher_groups(mut self) -> [(HookEventName, Vec); 12] { + self.matcher_groups_mut() + .map(|(event, groups)| (event, std::mem::take(groups))) + } + + pub fn matcher_groups_mut(&mut self) -> [(HookEventName, &mut Vec); 12] { + use HookEventName as Event; + [ - (HookEventName::PreToolUse, self.pre_tool_use), - (HookEventName::PermissionRequest, self.permission_request), - (HookEventName::PostToolUse, self.post_tool_use), - (HookEventName::PreCompact, self.pre_compact), - (HookEventName::PostCompact, self.post_compact), - (HookEventName::SessionStart, self.session_start), - (HookEventName::SessionEnd, self.session_end), - (HookEventName::UserPromptSubmit, self.user_prompt_submit), - (HookEventName::SubagentStart, self.subagent_start), - (HookEventName::SubagentStop, self.subagent_stop), - (HookEventName::Stop, self.stop), - (HookEventName::Interrupt, self.interrupt), + (Event::PreToolUse, &mut self.pre_tool_use), + (Event::PermissionRequest, &mut self.permission_request), + (Event::PostToolUse, &mut self.post_tool_use), + (Event::PreCompact, &mut self.pre_compact), + (Event::PostCompact, &mut self.post_compact), + (Event::SessionStart, &mut self.session_start), + (Event::SessionEnd, &mut self.session_end), + (Event::UserPromptSubmit, &mut self.user_prompt_submit), + (Event::SubagentStart, &mut self.subagent_start), + (Event::SubagentStop, &mut self.subagent_stop), + (Event::Stop, &mut self.stop), + (Event::Interrupt, &mut self.interrupt), ] } } diff --git a/codex-rs/core-plugins/src/executor_hooks.rs b/codex-rs/core-plugins/src/executor_hooks.rs index 3bbd36cfb9..885d9e6e36 100644 --- a/codex-rs/core-plugins/src/executor_hooks.rs +++ b/codex-rs/core-plugins/src/executor_hooks.rs @@ -1,6 +1,4 @@ -use codex_config::HookEventsToml; use codex_config::HookHandlerConfig; -use codex_config::MatcherGroup; use codex_exec_server::ExecutorCapabilityDiscoverySnapshot; use codex_plugin::ExecutorPluginHookSource; use codex_plugin::PluginId; @@ -18,7 +16,7 @@ struct AllowlistedExecutorPluginHook { } // Executor plugin manifests are unsigned, so temporarily hardcode the expected -// bundled plugin identities and cleanup MCP target until plugin signing lands. +// bundled plugin identities, events, and MCP targets until plugin signing lands. const ALLOWLISTED_EXECUTOR_PLUGIN_HOOKS: &[AllowlistedExecutorPluginHook] = &[ AllowlistedExecutorPluginHook { plugin_id: "browser@openai-bundled", @@ -26,30 +24,60 @@ const ALLOWLISTED_EXECUTOR_PLUGIN_HOOKS: &[AllowlistedExecutorPluginHook] = &[ server: "node_repl", tool: "turn_ended", }, + AllowlistedExecutorPluginHook { + plugin_id: "browser@openai-bundled", + event: HookEventName::Interrupt, + server: "node_repl", + tool: "turn_ended", + }, AllowlistedExecutorPluginHook { plugin_id: "chrome@openai-bundled", event: HookEventName::Stop, server: "node_repl", tool: "turn_ended", }, + AllowlistedExecutorPluginHook { + plugin_id: "chrome@openai-bundled", + event: HookEventName::Interrupt, + server: "node_repl", + tool: "turn_ended", + }, AllowlistedExecutorPluginHook { plugin_id: "chrome-dev@openai-bundled", event: HookEventName::Stop, server: "node_repl", tool: "turn_ended", }, + AllowlistedExecutorPluginHook { + plugin_id: "chrome-dev@openai-bundled", + event: HookEventName::Interrupt, + server: "node_repl", + tool: "turn_ended", + }, AllowlistedExecutorPluginHook { plugin_id: "chrome-internal@openai-bundled", event: HookEventName::Stop, server: "node_repl", tool: "turn_ended", }, + AllowlistedExecutorPluginHook { + plugin_id: "chrome-internal@openai-bundled", + event: HookEventName::Interrupt, + server: "node_repl", + tool: "turn_ended", + }, AllowlistedExecutorPluginHook { plugin_id: "computer-use@openai-bundled", event: HookEventName::Stop, server: "node_repl", tool: "turn_ended", }, + AllowlistedExecutorPluginHook { + plugin_id: "computer-use@openai-bundled", + event: HookEventName::Interrupt, + server: "node_repl", + tool: "turn_ended", + }, ]; /// Returns accepted inline hook sources from executor-discovered plugin manifests. @@ -113,31 +141,27 @@ pub fn executor_plugin_hook_sources( } fn allowlisted_source(mut source: ExecutorPluginHookSource) -> Option { - let allowlisted_hook = ALLOWLISTED_EXECUTOR_PLUGIN_HOOKS.iter().find(|hook| { - hook.plugin_id == source.plugin_id.as_key() && hook.event == HookEventName::Stop - })?; - let handler = source - .hooks - .stop - .into_iter() - .filter(|group| group.matcher.is_none()) - .flat_map(|group| group.hooks) - .find(|handler| { - matches!( - handler, - HookHandlerConfig::McpTool { server, tool, .. } - if server == allowlisted_hook.server && tool == allowlisted_hook.tool - ) - })?; - - source.hooks = HookEventsToml { - stop: vec![MatcherGroup { - matcher: None, - hooks: vec![handler], - }], - ..Default::default() - }; - Some(source) + let plugin_id = source.plugin_id.as_key(); + for (event, groups) in source.hooks.matcher_groups_mut() { + groups.retain_mut(|group| { + if group.matcher.is_some() { + return false; + } + group.hooks.retain(|handler| { + let HookHandlerConfig::McpTool { server, tool, .. } = handler else { + return false; + }; + ALLOWLISTED_EXECUTOR_PLUGIN_HOOKS.iter().any(|hook| { + hook.plugin_id == plugin_id + && hook.event == event + && hook.server == server + && hook.tool == tool + }) + }); + !group.hooks.is_empty() + }); + } + (!source.hooks.is_empty()).then_some(source) } #[cfg(test)] diff --git a/codex-rs/core-plugins/src/executor_hooks_tests.rs b/codex-rs/core-plugins/src/executor_hooks_tests.rs index 60fe716f45..55c836b48f 100644 --- a/codex-rs/core-plugins/src/executor_hooks_tests.rs +++ b/codex-rs/core-plugins/src/executor_hooks_tests.rs @@ -1,6 +1,8 @@ use std::collections::HashMap; use std::sync::Arc; +use codex_config::HookEventsToml; +use codex_config::MatcherGroup; use codex_exec_server::CapabilityRootDiscovery; use codex_exec_server::CapabilityTextFile; use codex_exec_server::DiscoveredPluginFiles; @@ -105,24 +107,20 @@ fn expected_source(index: usize) -> ExecutorPluginHookSource { #[test] fn discovers_allowlisted_executor_plugin_hook_sources() { - let mut manifest = cleanup_hook_manifest(); - manifest["hooks"]["hooks"]["Stop"][0]["hooks"] - .as_array_mut() - .expect("stop hooks") - .push(json!({ - "type": "command", - "command": "echo ignored" - })); - manifest["hooks"]["hooks"]["UserPromptSubmit"] = json!([{ - "hooks": [{ - "type": "mcp_tool", - "server": "other", - "tool": "ignored" - }] - }]); - let second_hook_file = cleanup_hook_manifest()["hooks"].clone(); - let first_hook_file = manifest["hooks"].take(); - manifest["hooks"] = json!([first_hook_file, second_hook_file]); + let cleanup_hooks = cleanup_hook_manifest()["hooks"]["hooks"]["Stop"].clone(); + let manifest = json!({ + "name": "computer-use", + "hooks": [ + { "hooks": { "Stop": cleanup_hooks } }, + { "hooks": { "Interrupt": cleanup_hooks } }, + { "hooks": { + "Stop": cleanup_hooks, + "Interrupt": cleanup_hooks, + "UserPromptSubmit": cleanup_hooks + } }, + { "hooks": { "UserPromptSubmit": cleanup_hooks } } + ] + }); let snapshot = snapshot_for_manifest( "computer-use@openai-bundled", "executor-a", @@ -131,13 +129,55 @@ fn discovers_allowlisted_executor_plugin_hook_sources() { ); let sources = executor_plugin_hook_sources(&snapshot); + let mut interrupt_source = expected_source(/*index*/ 1); + interrupt_source.hooks.interrupt = std::mem::take(&mut interrupt_source.hooks.stop); + let mut stop_and_interrupt_source = expected_source(/*index*/ 2); + stop_and_interrupt_source.hooks.interrupt = stop_and_interrupt_source.hooks.stop.clone(); assert_eq!( sources, - vec![expected_source(/*index*/ 0), expected_source(/*index*/ 1)] + vec![ + expected_source(/*index*/ 0), + interrupt_source, + stop_and_interrupt_source, + ] ); } +#[test] +fn filters_mixed_handlers_without_rewriting_allowed_groups() { + let mut expected = expected_source(/*index*/ 0); + let mut second_handler = expected.hooks.stop[0].hooks[0].clone(); + let HookHandlerConfig::McpTool { input, .. } = &mut second_handler else { + panic!("expected an MCP tool hook"); + }; + input.insert("order".to_string(), json!(2)); + expected.hooks.stop[0].hooks.push(second_handler.clone()); + expected.hooks.stop.push(MatcherGroup { + matcher: None, + hooks: vec![second_handler], + }); + + let mut manifest = cleanup_hook_manifest(); + manifest["hooks"]["hooks"]["Stop"] = + serde_json::to_value(&expected.hooks.stop).expect("serialize stop hooks"); + manifest["hooks"]["hooks"]["Stop"][0]["hooks"] + .as_array_mut() + .expect("stop handlers") + .insert( + /*index*/ 0, + json!({ "type": "mcp_tool", "server": "node_repl", "tool": "other" }), + ); + let snapshot = snapshot_for_manifest( + "computer-use@openai-bundled", + "executor-a", + "file:///plugins/computer-use/.codex-plugin/plugin.json", + manifest, + ); + + assert_eq!(executor_plugin_hook_sources(&snapshot), vec![expected]); +} + #[test] fn preserves_allowlisted_executor_plugin_hook_options() { let mut manifest = cleanup_hook_manifest(); @@ -145,6 +185,7 @@ fn preserves_allowlisted_executor_plugin_hook_options() { handler["input"] = json!({ "untrusted": "manifest-provided input" }); handler["timeout"] = json!(30); handler["statusMessage"] = json!("Cleaning up Computer Use"); + manifest["hooks"]["hooks"]["Interrupt"] = manifest["hooks"]["hooks"]["Stop"].clone(); let snapshot = snapshot_for_manifest( "computer-use@openai-bundled", "executor-a", @@ -161,6 +202,7 @@ fn preserves_allowlisted_executor_plugin_hook_options() { timeout_sec: Some(30), status_message: Some("Cleaning up Computer Use".to_string()), }; + expected.hooks.interrupt = expected.hooks.stop.clone(); assert_eq!(executor_plugin_hook_sources(&snapshot), vec![expected]); } diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 3f0b6b4b22..a8e8308797 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -48,7 +48,9 @@ use codex_protocol::protocol::WarningEvent; use codex_rollout::state_db; use codex_thread_store::PersistContext; use codex_thread_store::ReadThreadParams; +use serde_json::Map; use serde_json::Value; +use tokio::sync::Mutex; use tracing::instrument; use crate::context::ContextualUserFragment; @@ -58,6 +60,7 @@ use crate::session::TurnInput; use crate::session::session::Session; use crate::session::step_context::StepContext; use crate::session::turn_context::TurnContext; +use crate::state::TurnState; use crate::tools::hook_names::HookToolName; use crate::tools::sandboxing::PermissionRequestPayload; use crate::turn_metadata::McpTurnMetadataContext; @@ -308,6 +311,29 @@ pub(crate) async fn run_post_tool_use_hooks( outcome } +fn build_request_metadata( + step_context: Option<&StepContext>, + turn_context: &TurnContext, +) -> Map { + let settings = step_context + .map(|step_context| step_context.settings.as_ref()) + .unwrap_or(turn_context.initial_settings.as_ref()); + turn_context + .turn_metadata_state + .current_meta_value_for_mcp_request(McpTurnMetadataContext { + model: settings.model_info.slug.as_str(), + reasoning_effort: settings.effective_reasoning_effort(), + node_repl_disabled: settings.model_info.node_repl_disabled, + }) + .map(|turn_metadata| { + Map::from_iter([( + crate::X_CODEX_TURN_METADATA_HEADER.to_string(), + turn_metadata, + )]) + }) + .unwrap_or_default() +} + #[instrument(level = "trace", skip_all)] pub(crate) async fn run_turn_stop_hooks( sess: &Arc, @@ -364,19 +390,7 @@ pub(crate) async fn run_turn_stop_hooks( ), _ => (StopHookTarget::Stop, sess.hook_transcript_path().await), }; - let request_metadata = turn_context - .turn_metadata_state - .current_meta_value_for_mcp_request(McpTurnMetadataContext { - model: step_context.settings.model_info.slug.as_str(), - reasoning_effort: step_context.settings.effective_reasoning_effort(), - node_repl_disabled: step_context.settings.model_info.node_repl_disabled, - }) - .map(|turn_metadata| { - serde_json::Map::from_iter([( - crate::X_CODEX_TURN_METADATA_HEADER.to_string(), - turn_metadata, - )]) - }); + let request_metadata = build_request_metadata(Some(step_context), turn_context); let request = codex_hooks::StopRequest { session_id: sess.session_id().into(), turn_id: turn_context.sub_id.clone(), @@ -385,7 +399,7 @@ pub(crate) async fn run_turn_stop_hooks( transcript_path, model: turn_context.model_info().slug.clone(), permission_mode: hook_permission_mode(turn_context), - request_metadata, + request_metadata: (!request_metadata.is_empty()).then_some(request_metadata), stop_hook_active, last_assistant_message, target, @@ -435,17 +449,30 @@ pub(crate) async fn run_session_end_hooks(sess: &Arc) { emit_hook_completed_events(sess, &turn_context, outcome.hook_events).await; } -pub(crate) async fn run_turn_interrupt_hooks(sess: &Arc, turn_context: &Arc) { +pub(crate) async fn run_turn_interrupt_hooks( + sess: &Arc, + turn_context: &Arc, + turn_state: &Mutex, +) { if matches!(&turn_context.session_source, SessionSource::SubAgent(_)) { return; } - let hooks = sess.hooks(); + // The active turn has already been detached. Reuse only its last executing step's discovery. + let last_known_step_context = turn_state.lock().await.last_known_step_context.clone(); + let executor_hook_sources = last_known_step_context + .as_ref() + .and_then(|step_context| step_context.executor_capability_discovery.as_deref()) + .map(executor_plugin_hook_sources) + .unwrap_or_default(); + let has_executor_hooks = !executor_hook_sources.is_empty(); + let hooks = sess.hooks().with_executor_hooks(executor_hook_sources); let preview_runs = hooks.preview_interrupt(); - if preview_runs.is_empty() { + if preview_runs.is_empty() && !has_executor_hooks { return; } + let request_metadata = build_request_metadata(last_known_step_context.as_deref(), turn_context); let request = InterruptRequest { session_id: sess.session_id().into(), turn_id: turn_context.sub_id.clone(), @@ -454,6 +481,7 @@ pub(crate) async fn run_turn_interrupt_hooks(sess: &Arc, turn_context: transcript_path: sess.hook_transcript_path().await, model: turn_context.model_info().slug.clone(), permission_mode: hook_permission_mode(turn_context), + request_metadata: (!request_metadata.is_empty()).then_some(request_metadata), }; if let Err(err) = sess.flush_rollout().await { tracing::warn!("failed to flush transcript before Interrupt hook: {err}"); diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 24bbae05a8..c51db874ee 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -12,6 +12,7 @@ use codex_diagnostics::Gauge; use codex_extension_api::ThreadIdleCause; use futures::future::BoxFuture; use tokio::select; +use tokio::sync::Mutex; use tokio::sync::Notify; use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; @@ -35,6 +36,7 @@ use crate::session::turn_context::TurnContext; use crate::state::ActiveTurn; use crate::state::RunningTask; use crate::state::TaskKind; +use crate::state::TurnState; use codex_analytics::TurnProfileFact; use codex_analytics::TurnTokenUsageFact; use codex_context_fragments::RenderedFragment; @@ -518,7 +520,8 @@ impl Session { aborted_turn = task.is_some(); turn_context = task.as_ref().map(|task| Arc::clone(&task.turn_context)); if let Some(task) = task { - self.handle_task_abort(task, reason.clone()).await; + self.handle_task_abort(task, reason.clone(), &active_turn.turn_state) + .await; } if aborted_turn { active_turn_to_clear = Some(active_turn); @@ -569,7 +572,8 @@ impl Session { let task = active_turn.task.take(); let turn_context = task.as_ref().map(|task| Arc::clone(&task.turn_context)); if let Some(task) = task { - self.handle_task_abort(task, reason.clone()).await; + self.handle_task_abort(task, reason.clone(), &active_turn.turn_state) + .await; } if let Some(turn_context) = turn_context.as_deref() { self.emit_turn_abort_lifecycle(reason.clone(), turn_context.extension_data.as_ref()) @@ -805,7 +809,7 @@ impl Session { }; let event = if let Some(reason) = abort_reason { if reason == TurnAbortReason::Interrupted { - run_turn_interrupt_hooks(self, &turn_context).await; + run_turn_interrupt_hooks(self, &turn_context, &turn_state).await; } self.emit_turn_abort_lifecycle(reason.clone(), turn_context.extension_data.as_ref()) .await; @@ -898,7 +902,12 @@ impl Session { .await } - async fn handle_task_abort(self: &Arc, task: RunningTask, reason: TurnAbortReason) { + async fn handle_task_abort( + self: &Arc, + task: RunningTask, + reason: TurnAbortReason, + turn_state: &Mutex, + ) { let sub_id = task.turn_context.sub_id.clone(); if task.cancellation_token.is_cancelled() { return; @@ -958,7 +967,7 @@ impl Session { } if reason == TurnAbortReason::Interrupted { - run_turn_interrupt_hooks(self, &task.turn_context).await; + run_turn_interrupt_hooks(self, &task.turn_context, turn_state).await; } let started_at = task diff --git a/codex-rs/core/tests/suite/hooks_executor.rs b/codex-rs/core/tests/suite/hooks_executor.rs index 2691e4866b..b8419e18cc 100644 --- a/codex-rs/core/tests/suite/hooks_executor.rs +++ b/codex-rs/core/tests/suite/hooks_executor.rs @@ -2,6 +2,7 @@ use anyhow::Context; use anyhow::Result; use codex_config::McpServerConfig; use codex_core::EnvironmentConfig; +use codex_core::TurnInputRequest; use codex_core::windows_sandbox::WindowsSandboxLevelExt; use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::ExecServerRuntimePaths; @@ -11,20 +12,26 @@ use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::config_types::WindowsSandboxLevel; use codex_protocol::models::PermissionProfileSnapshot; use codex_protocol::protocol::EnvironmentConfigState; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::TurnEnvironmentSelections; +use codex_protocol::user_input::UserInput; use core_test_support::responses::ResponseMock; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_response_created; -use core_test_support::responses::mount_sse_sequence; +use core_test_support::responses::mount_response_sequence; use core_test_support::responses::sse; +use core_test_support::responses::sse_response; use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; +use core_test_support::skip_if_remote; use core_test_support::submit_thread_settings; use core_test_support::test_codex::TestCodex; use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; use core_test_support::wait_for_mcp_server; use pretty_assertions::assert_eq; use serde_json::Value; @@ -99,6 +106,93 @@ async fn executor_stop_hook_runs_after_attachment() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn executor_interrupt_hook_runs_after_attachment() -> Result<()> { + skip_if_no_network!(Ok(())); + + let fixture = executor_hook_fixture(vec![ + completed_turn_response("turn").set_delay(Duration::from_secs(60)), + ]) + .await?; + fixture.attach().await?; + fixture + .test + .codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "interrupt this turn".to_string(), + text_elements: Vec::new(), + }])) + .await?; + + // The model request confirms discovery was saved, without requiring a target shell. + tokio::time::timeout(Duration::from_secs(10), async { + while fixture.responses.requests().is_empty() { + tokio::task::yield_now().await; + } + }) + .await + .context("interrupted turn should reach the model request")?; + fixture.test.codex.submit(Op::Interrupt).await?; + wait_for_event(&fixture.test.codex, |event| { + matches!(event, EventMsg::TurnAborted(_)) + }) + .await; + fixture.wait_for_hook_call().await?; + + let calls = fixture.calls().await?; + assert_eq!(calls.len(), 1); + let call = &calls[0]["params"]; + let response = fixture.responses.single_request().body_json(); + let turn_id = &response["client_metadata"]["turn_id"]; + assert_eq!(call["name"], "turn_ended"); + assert_eq!( + call["arguments"], + json!({ + "hook_event_name": "Interrupt", + "session_id": fixture.test.session_configured.thread_id.to_string(), + "turn_id": turn_id, + }) + ); + assert_eq!(call["_meta"]["x-codex-turn-metadata"]["turn_id"], *turn_id); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn executor_interrupt_hook_skips_turn_without_step_context() -> Result<()> { + skip_if_no_network!(Ok(())); + skip_if_remote!( + Ok(()), + "standalone user shell commands require a local environment" + ); + + let fixture = executor_hook_fixture(vec![completed_turn_response("first-turn")]).await?; + fixture.attach().await?; + fixture + .test + .submit_text_turn("populate executor discovery") + .await?; + fixture.wait_for_hook_call().await?; + + // Standalone shell turns have no model step, despite the previous turn's discovery. + fixture + .test + .codex + .submit(Op::RunUserShellCommand { + command: "sleep 60".to_string(), + timeout_ms: None, + }) + .await?; + fixture.interrupt_running_command().await?; + + assert!( + tokio::time::timeout(Duration::from_secs(1), fixture.hook_called.notified()) + .await + .is_err(), + "executor hooks must not reuse a previous turn's step context", + ); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn executor_stop_hook_stops_after_disconnection() -> Result<()> { skip_if_no_network!(Ok(())); @@ -256,7 +350,24 @@ async fn executor_stop_hook_rejects_mismatched_environment() -> Result<()> { Ok(()) } -async fn executor_stop_hook_fixture() -> Result { +fn completed_turn_response(id: &str) -> ResponseTemplate { + sse_response(sse(vec![ + ev_response_created(id), + ev_assistant_message(id, "done"), + ev_completed(id), + ])) +} + +async fn executor_stop_hook_fixture() -> Result { + executor_hook_fixture( + ["first-turn", "second-turn"] + .map(completed_turn_response) + .to_vec(), + ) + .await +} + +async fn executor_hook_fixture(responses: Vec) -> Result { let server = start_mock_server().await; let hook_called = Arc::new(Notify::new()); let hook_called_for_server = Arc::clone(&hook_called); @@ -337,37 +448,38 @@ async fn executor_stop_hook_fixture() -> Result { &manifest_path, serde_json::to_vec(&json!({ "name": "computer-use", - "hooks": { "hooks": { "Stop": [{ "hooks": [{ - "type": "mcp_tool", - "server": "node_repl", - "tool": "turn_ended", - "input": { - "hook_event_name": "${hook_event_name}", - "session_id": "${session_id}", - "turn_id": "${turn_id}", - }, - }] }] } }, + // Separate entries must both survive registration in the same environment. + "hooks": [ + { "hooks": { "Interrupt": [{ "hooks": [{ + "type": "mcp_tool", + "server": "node_repl", + "tool": "turn_ended", + "input": { + "hook_event_name": "${hook_event_name}", + "session_id": "${session_id}", + "turn_id": "${turn_id}", + }, + }] }] } }, + { "hooks": { "Stop": [{ "hooks": [{ + "type": "mcp_tool", + "server": "node_repl", + "tool": "turn_ended", + "input": { + "hook_event_name": "${hook_event_name}", + "session_id": "${session_id}", + "turn_id": "${turn_id}", + }, + }] }] } }, + ], }))?, Default::default(), /*sandbox*/ None, ) .await?; - let responses = mount_sse_sequence( - &server, - ["first-turn", "second-turn"] - .map(|id| { - sse(vec![ - ev_response_created(id), - ev_assistant_message(id, "done"), - ev_completed(id), - ]) - }) - .to_vec(), - ) - .await; + let responses = mount_response_sequence(&server, responses).await; - Ok(ExecutorStopHookFixture { + Ok(ExecutorHookFixture { server, test, responses, @@ -375,14 +487,14 @@ async fn executor_stop_hook_fixture() -> Result { }) } -struct ExecutorStopHookFixture { +struct ExecutorHookFixture { server: MockServer, test: TestCodex, responses: ResponseMock, hook_called: Arc, } -impl ExecutorStopHookFixture { +impl ExecutorHookFixture { async fn attach(&self) -> Result { let selection = self .test @@ -427,10 +539,23 @@ impl ExecutorStopHookFixture { Ok(selection) } + async fn interrupt_running_command(&self) -> Result<()> { + wait_for_event(&self.test.codex, |event| { + matches!(event, EventMsg::ExecCommandBegin(_)) + }) + .await; + self.test.codex.submit(Op::Interrupt).await?; + wait_for_event(&self.test.codex, |event| { + matches!(event, EventMsg::TurnAborted(_)) + }) + .await; + Ok(()) + } + async fn wait_for_hook_call(&self) -> Result<()> { tokio::time::timeout(Duration::from_secs(10), self.hook_called.notified()) .await - .context("attached executor Stop hook should call turn_ended")?; + .context("attached executor hook should call turn_ended")?; Ok(()) } diff --git a/codex-rs/hooks/src/engine/mod.rs b/codex-rs/hooks/src/engine/mod.rs index 5934f46322..a287e83c12 100644 --- a/codex-rs/hooks/src/engine/mod.rs +++ b/codex-rs/hooks/src/engine/mod.rs @@ -41,6 +41,7 @@ use codex_protocol::protocol::HookTrustStatus; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::PathUri; use std::collections::HashMap; +use std::collections::HashSet; use std::sync::Arc; use std::time::Duration; @@ -265,62 +266,73 @@ impl ClaudeHooksEngine { ) }); - // FIXME: Remove this restriction once executor hooks support multiple environments. - if executor_hooks.len() > 1 { - tracing::warn!( - executor_hook_count = executor_hooks.len(), - "multiple executor hooks found; only executing the first" - ); - } - let Some(source) = executor_hooks.into_iter().next() else { + // All events share the first environment, even if it has no handler for an event. + let Some(selected_environment_id) = executor_hooks + .first() + .map(|source| source.environment_id.clone()) + else { return; }; - let Some(handler) = source - .hooks - .stop - .into_iter() - .flat_map(|group| group.hooks) - .next() - else { - unreachable!("allowlisted executor hook source must contain a Stop handler"); - }; - let HookHandlerConfig::McpTool { - server, - tool, - input, - timeout_sec, - status_message, - } = handler - else { - unreachable!("allowlisted executor Stop handler must be an MCP tool"); - }; - - let display_order = self - .handlers + let environment_count = executor_hooks .iter() - .map(|handler| handler.display_order) - .max() - .map_or(0, |display_order| display_order.saturating_add(1)); - self.handlers.push(ConfiguredHandler { - event_name: HookEventName::Stop, - matcher: None, - timeout_sec: timeout_sec.unwrap_or(5).max(1), - status_message, - additional_context_limit: Default::default(), - source_path: HandlerSourcePath::ExecutorScoped { - plugin_id: source.plugin_id, - environment_id: source.environment_id, - manifest_path: source.manifest_path, - source_relative_path: source.source_relative_path, - }, - source: HookSource::Plugin, - display_order, - kind: ConfiguredHandlerKind::McpTool { - server, - tool, - input, - }, - }); + .map(|source| &source.environment_id) + .collect::>() + .len(); + // FIXME: Remove this restriction once executor hooks support multiple environments. + if environment_count > 1 { + tracing::warn!( + executor_environment_count = environment_count, + selected_environment_id = %selected_environment_id, + "multiple executor environments found; only executing hooks in the first" + ); + } + for source in executor_hooks { + if source.environment_id != selected_environment_id { + continue; + } + for (event_name, groups) in source.hooks.into_matcher_groups() { + let Some(handler) = groups.into_iter().flat_map(|group| group.hooks).next() else { + continue; + }; + let HookHandlerConfig::McpTool { + server, + tool, + input, + timeout_sec, + status_message, + } = handler + else { + unreachable!("allowlisted executor handler must be an MCP tool"); + }; + + let display_order = self + .handlers + .iter() + .map(|handler| handler.display_order) + .max() + .map_or(0, |display_order| display_order.saturating_add(1)); + self.handlers.push(ConfiguredHandler { + event_name, + matcher: None, + timeout_sec: timeout_sec.unwrap_or(5).max(1), + status_message, + additional_context_limit: Default::default(), + source_path: HandlerSourcePath::ExecutorScoped { + plugin_id: source.plugin_id.clone(), + environment_id: source.environment_id.clone(), + manifest_path: source.manifest_path.clone(), + source_relative_path: source.source_relative_path.clone(), + }, + source: HookSource::Plugin, + display_order, + kind: ConfiguredHandlerKind::McpTool { + server, + tool, + input, + }, + }); + } + } } pub(crate) fn required_load_errors(&self) -> &[String] { diff --git a/codex-rs/hooks/src/engine/mod_tests.rs b/codex-rs/hooks/src/engine/mod_tests.rs index ed25847148..3aba38c56d 100644 --- a/codex-rs/hooks/src/engine/mod_tests.rs +++ b/codex-rs/hooks/src/engine/mod_tests.rs @@ -2219,6 +2219,58 @@ fn executor_stop_hooks_register_only_the_first_environment_and_handler() { assert_eq!(engine.handlers, expected_handlers); } +#[tokio::test] +async fn executor_interrupt_hooks_register_and_run() { + let (mut engine, calls, stop_request, expected_stop_call, mut source) = + executor_stop_hook_fixture(); + source.hooks.interrupt = source.hooks.stop.clone(); + engine.set_executor_hooks(vec![source]); + + assert_eq!( + engine + .handlers + .iter() + .map(|handler| handler.event_name) + .collect::>(), + vec![HookEventName::Stop, HookEventName::Interrupt] + ); + assert_eq!(engine.preview_interrupt(), Vec::new()); + + let outcome = engine + .run_interrupt(InterruptRequest { + session_id: stop_request.session_id, + turn_id: stop_request.turn_id, + cwd: stop_request.cwd, + transcript_path: stop_request.transcript_path, + model: stop_request.model, + permission_mode: stop_request.permission_mode, + request_metadata: stop_request.request_metadata, + }) + .await; + wait_for_mcp_calls(&calls, /*count*/ 1).await; + + assert!(outcome.hook_events.is_empty()); + assert_eq!( + *calls.lock().expect("lock MCP calls"), + vec![expected_stop_call] + ); +} + +#[test] +fn executor_hooks_use_one_environment_for_all_events() { + let (mut engine, _, _, _, mut first_source) = executor_stop_hook_fixture(); + first_source.hooks.interrupt = std::mem::take(&mut first_source.hooks.stop); + engine.set_executor_hooks(vec![first_source.clone()]); + let expected_handlers = engine.handlers.clone(); + let mut second_source = first_source.clone(); + second_source.environment_id = "executor-b".to_string(); + second_source.hooks.stop = std::mem::take(&mut second_source.hooks.interrupt); + + engine.set_executor_hooks(vec![first_source, second_source]); + + assert_eq!(engine.handlers, expected_handlers); +} + #[tokio::test] async fn memory_consolidation_stop_preserves_policy_and_executor_cleanup() { for (source, runs_policy) in [ @@ -2482,6 +2534,7 @@ async fn mcp_interrupt_hooks_expand_event_input_and_bound_timeout() { transcript_path: None, model: "gpt-test".to_string(), permission_mode: "default".to_string(), + request_metadata: None, }) .await; diff --git a/codex-rs/hooks/src/events/interrupt.rs b/codex-rs/hooks/src/events/interrupt.rs index 3abea204bf..743002eb9c 100644 --- a/codex-rs/hooks/src/events/interrupt.rs +++ b/codex-rs/hooks/src/events/interrupt.rs @@ -8,11 +8,14 @@ use codex_protocol::protocol::HookOutputEntryKind; use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookRunSummary; use codex_utils_absolute_path::AbsolutePathBuf; +use serde_json::Map; +use serde_json::Value; use super::common; use crate::engine::ClaudeHooksEngine; use crate::engine::ConfiguredHandler; use crate::engine::HandlerRunResult; +use crate::engine::HandlerSourcePath; use crate::engine::dispatcher; use crate::engine::output_parser; use crate::schema::InterruptCommandInput; @@ -26,6 +29,7 @@ pub struct InterruptRequest { pub transcript_path: Option, pub model: String, pub permission_mode: String, + pub request_metadata: Option>, } #[derive(Debug, Default)] @@ -43,6 +47,8 @@ pub(crate) fn preview(handlers: &[ConfiguredHandler]) -> Vec { /*matcher_input*/ None, ) .into_iter() + // Executor-scoped hooks run asynchronously and do not emit public hook lifecycle events. + .filter(|handler| matches!(handler.source_path, HandlerSourcePath::Local(_))) .map(|handler| dispatcher::running_summary(&handler)) .collect() } @@ -64,6 +70,7 @@ pub(crate) async fn run(engine: &ClaudeHooksEngine, request: InterruptRequest) - transcript_path, model, permission_mode, + request_metadata, } = request; let input_json = match serde_json::to_string(&InterruptCommandInput { session_id: session_id.to_string(), @@ -86,12 +93,13 @@ pub(crate) async fn run(engine: &ClaudeHooksEngine, request: InterruptRequest) - } }; - let results = dispatcher::execute_handlers( + let results = dispatcher::execute_handlers_with_metadata( engine, matched, input_json, cwd.as_path(), Some(turn_id), + request_metadata.as_ref(), parse_completed, ) .await;