From b721077974b2f5467db738d2bd7d5272b59c4677 Mon Sep 17 00:00:00 2001 From: Abhinav Vedmala Date: Fri, 12 Jun 2026 05:12:01 +0000 Subject: [PATCH] Enable async hook delivery on accepted turns --- codex-rs/core/src/hook_runtime.rs | 88 +++- codex-rs/core/src/session/handlers.rs | 1 + codex-rs/core/src/session/mod.rs | 19 +- codex-rs/core/src/session/session.rs | 14 +- codex-rs/core/src/session/turn.rs | 48 +- codex-rs/core/src/tasks/mod.rs | 30 +- codex-rs/core/tests/suite/hooks.rs | 471 ++++++++++++++++++ codex-rs/hooks/src/engine/command_runner.rs | 3 +- codex-rs/hooks/src/engine/discovery.rs | 17 +- codex-rs/hooks/src/engine/dispatcher.rs | 138 +++-- codex-rs/hooks/src/engine/mod.rs | 59 ++- codex-rs/hooks/src/events/common.rs | 2 + codex-rs/hooks/src/events/compact.rs | 55 +- .../hooks/src/events/permission_request.rs | 32 +- codex-rs/hooks/src/events/post_tool_use.rs | 30 +- codex-rs/hooks/src/events/pre_tool_use.rs | 30 +- codex-rs/hooks/src/events/session_start.rs | 29 +- codex-rs/hooks/src/events/stop.rs | 29 +- .../hooks/src/events/user_prompt_submit.rs | 29 +- codex-rs/hooks/src/lib.rs | 2 + codex-rs/hooks/src/registry.rs | 15 + 21 files changed, 882 insertions(+), 259 deletions(-) diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 342e4d4b8f..63d912b7e6 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -5,6 +5,8 @@ use std::time::Duration; use codex_analytics::CompactionTrigger; use codex_analytics::HookRunFact; use codex_analytics::build_track_events_context; +use codex_hooks::Hooks; +use codex_hooks::PendingAsyncHookDelivery; use codex_hooks::PermissionRequestDecision; use codex_hooks::PermissionRequestOutcome; use codex_hooks::PermissionRequestRequest; @@ -35,6 +37,7 @@ use codex_protocol::protocol::HookSource; use codex_protocol::protocol::HookStartedEvent; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::WarningEvent; use codex_thread_store::ReadThreadParams; use serde_json::Value; use tracing::instrument; @@ -48,9 +51,9 @@ use crate::session::turn_context::TurnContext; use crate::tools::hook_names::HookToolName; use crate::tools::sandboxing::PermissionRequestPayload; -pub(crate) struct HookRuntimeOutcome { - pub should_stop: bool, - pub additional_contexts: Vec, +struct HookRuntimeOutcome { + should_stop: bool, + additional_contexts: Vec, } pub(crate) enum PreToolUseHookResult { @@ -497,7 +500,81 @@ pub(crate) async fn run_legacy_after_agent_hook( true } -pub(crate) async fn inspect_pending_input( +/// Runs prompt hooks for the initial turn input and delivers async hook output +/// that was ready when the turn began. +#[instrument(level = "trace", skip_all)] +pub(crate) async fn run_initial_user_prompt_submit_hooks_and_record_inputs( + sess: &Arc, + turn_context: &Arc, + input: &[TurnInput], + pending_async_output: PendingAsyncHookDelivery, +) -> bool { + let can_accept_user_input = + input.iter().any(|input_item| { + matches!(input_item, TurnInput::UserInput { content, .. } if !content.is_empty()) + }); + let blocked_input = + run_user_prompt_submit_hooks_and_record_inputs(sess, turn_context, input).await; + if !blocked_input && can_accept_user_input { + let delivery = pending_async_output.accept_turn(); + record_additional_contexts(sess, turn_context, delivery.additional_contexts).await; + for message in delivery.system_messages { + sess.send_event(turn_context, EventMsg::Warning(WarningEvent { message })) + .await; + } + } + blocked_input +} + +/// Runs prompt hooks and records accepted input without delivering async output. +#[instrument(level = "trace", skip_all)] +pub(crate) async fn run_user_prompt_submit_hooks_and_record_inputs( + sess: &Arc, + turn_context: &Arc, + input: &[TurnInput], +) -> bool { + let hooks = sess.hooks(); + let mut blocked_input = false; + let mut accepted_user_input = false; + for input_item in input { + let can_accept_user_input = + matches!(input_item, TurnInput::UserInput { content, .. } if !content.is_empty()); + if run_user_prompt_submit_hook_and_record_input(&hooks, sess, turn_context, input_item) + .await + { + blocked_input = true; + } else if can_accept_user_input { + accepted_user_input = true; + } + } + blocked_input && !accepted_user_input +} + +/// Runs the prompt hook and records one input, returning whether it was blocked. +pub(crate) async fn run_user_prompt_submit_hook_and_record_input( + hooks: &Hooks, + sess: &Arc, + turn_context: &Arc, + input: &TurnInput, +) -> bool { + let hook_outcome = inspect_pending_input(hooks, sess, turn_context, input).await; + if hook_outcome.should_stop { + record_additional_contexts(sess, turn_context, hook_outcome.additional_contexts).await; + true + } else { + record_pending_input( + sess, + turn_context, + input.clone(), + hook_outcome.additional_contexts, + ) + .await; + false + } +} + +async fn inspect_pending_input( + hooks: &Hooks, sess: &Arc, turn_context: &Arc, pending_input_item: &TurnInput, @@ -515,7 +592,6 @@ pub(crate) async fn inspect_pending_input( permission_mode: hook_permission_mode(turn_context), prompt: UserMessageItem::new(content).message(), }; - let hooks = sess.hooks(); let preview_runs = hooks.preview_user_prompt_submit(&request); run_context_injecting_hook( sess, @@ -536,7 +612,7 @@ pub(crate) async fn inspect_pending_input( } } -pub(crate) async fn record_pending_input( +async fn record_pending_input( sess: &Arc, turn_context: &Arc, pending_input: TurnInput, diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index cc21755524..94af659a2e 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -602,6 +602,7 @@ async fn shutdown_session_runtime(sess: &Arc) { } let _ = sess.conversation.shutdown().await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; + sess.hooks().shutdown().await; sess.services .unified_exec_manager .terminate_all_processes() diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 9339413b33..1bd4d13360 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1620,12 +1620,15 @@ impl Session { self.services.skills_service.clear_cache(); self.services.plugins_manager.clear_cache(); let environments = self.services.turn_environments.snapshot().await; - let hooks = build_hooks_for_config( + let hooks_config = build_hooks_config( config.as_ref(), self.services.plugins_manager.as_ref(), environments.single_local_environment(), ) .await; + // Reconfigure the existing hook service so background commands, queued + // output, and async delivery state survive a config refresh. + let hooks = self.hooks().reconfigured(hooks_config); let state = self.state.lock().await; // A newer refresh may have updated the config while this hook build was in flight. @@ -4042,12 +4045,16 @@ pub(crate) fn emit_subagent_session_started( }); } -/// Builds the hook engine for one config snapshot, including any enabled plugin hooks. -async fn build_hooks_for_config( +/// Builds the reloadable hook configuration for one config snapshot. +/// +/// Runtime state is intentionally not handled here. Initial session setup passes +/// this config to `Hooks::new`, while config refresh passes it to +/// `Hooks::reconfigured` so the hook crate can preserve its own in-flight work. +async fn build_hooks_config( config: &Config, plugins_manager: &PluginsManager, environment: Option<&TurnEnvironment>, -) -> Hooks { +) -> HooksConfig { let (hook_shell_program, hook_shell_argv) = environment .and_then(|environment| environment.shell.as_ref()) .map(|shell| { @@ -4061,7 +4068,7 @@ async fn build_hooks_for_config( let plugin_outcome = plugins_manager.plugins_for_config(&plugins_input).await; let plugin_hook_sources = plugin_outcome.effective_plugin_hook_sources(); let plugin_hook_load_warnings = plugin_outcome.effective_plugin_hook_warnings(); - Hooks::new(HooksConfig { + HooksConfig { legacy_notify_argv: config.notify.clone(), feature_enabled: config.features.enabled(Feature::CodexHooks), bypass_hook_trust: config.bypass_hook_trust, @@ -4070,7 +4077,7 @@ async fn build_hooks_for_config( plugin_hook_load_warnings, shell_program: hook_shell_program, shell_args: hook_shell_argv, - }) + } } #[cfg(test)] diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 7c59ec22b7..d39ca253b4 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1001,12 +1001,14 @@ impl Session { (None, None) }; - let hooks = build_hooks_for_config( - &config, - plugins_manager.as_ref(), - resolved_environments.single_local_environment(), - ) - .await; + let hooks = Hooks::new( + build_hooks_config( + &config, + plugins_manager.as_ref(), + resolved_environments.single_local_environment(), + ) + .await, + ); for warning in hooks.startup_warnings() { post_session_configured_events.push(Event { id: INITIAL_SUBMIT_ID.to_owned(), diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index bafb368b32..f7137cf4d8 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -18,12 +18,11 @@ use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_ use crate::connectors; use crate::context::ContextualUserFragment; use crate::feedback_tags; -use crate::hook_runtime::inspect_pending_input; -use crate::hook_runtime::record_additional_contexts; -use crate::hook_runtime::record_pending_input; +use crate::hook_runtime::run_initial_user_prompt_submit_hooks_and_record_inputs; use crate::hook_runtime::run_legacy_after_agent_hook; use crate::hook_runtime::run_pending_session_start_hooks; use crate::hook_runtime::run_turn_stop_hooks; +use crate::hook_runtime::run_user_prompt_submit_hooks_and_record_inputs; use crate::injection::ToolMentionKind; use crate::injection::app_id_from_path; use crate::injection::tool_kind_for_path; @@ -147,6 +146,7 @@ pub(crate) async fn run_turn( prewarmed_client_session: Option, cancellation_token: CancellationToken, ) -> CodexResult> { + let pending_async_output = sess.hooks().pending_async_delivery(); let mut client_session = prewarmed_client_session.unwrap_or_else(|| sess.services.model_client.new_session()); // TODO(ccunningham): Pre-turn compaction runs before context updates and the @@ -187,7 +187,14 @@ pub(crate) async fn run_turn( return Ok(None); } let mut can_drain_pending_input = input.is_empty(); - if run_hooks_and_record_inputs(&sess, &turn_context, &input).await { + if run_initial_user_prompt_submit_hooks_and_record_inputs( + &sess, + &turn_context, + &input, + pending_async_output, + ) + .await + { return Ok(None); } @@ -232,7 +239,9 @@ pub(crate) async fn run_turn( Vec::new() }; - if run_hooks_and_record_inputs(&sess, &turn_context, &pending_input).await { + if run_user_prompt_submit_hooks_and_record_inputs(&sess, &turn_context, &pending_input) + .await + { break; } @@ -478,35 +487,6 @@ async fn turn_diff_display_roots(turn_context: &TurnContext) -> Vec<(String, Pat display_roots } -#[instrument(level = "trace", skip_all)] -async fn run_hooks_and_record_inputs( - sess: &Arc, - turn_context: &Arc, - input: &[TurnInput], -) -> bool { - let mut blocked_input = false; - let mut accepted_user_input = false; - for input_item in input { - let hook_outcome = inspect_pending_input(sess, turn_context, input_item).await; - if hook_outcome.should_stop { - blocked_input = true; - record_additional_contexts(sess, turn_context, hook_outcome.additional_contexts).await; - } else { - if matches!(input_item, TurnInput::UserInput { content, .. } if !content.is_empty()) { - accepted_user_input = true; - } - record_pending_input( - sess, - turn_context, - input_item.clone(), - hook_outcome.additional_contexts, - ) - .await; - } - } - blocked_input && !accepted_user_input -} - #[instrument(level = "trace", skip_all)] async fn build_skills_and_plugins( sess: &Arc, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 7a930bb578..8640d527dc 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -25,9 +25,7 @@ use tracing::warn; use crate::codex_thread::BackgroundTerminalInfo; use crate::config::Config; use crate::context::ContextualUserFragment; -use crate::hook_runtime::inspect_pending_input; -use crate::hook_runtime::record_additional_contexts; -use crate::hook_runtime::record_pending_input; +use crate::hook_runtime::run_user_prompt_submit_hook_and_record_input; use crate::session::TurnInput; use crate::session::session::Session; use crate::session::turn_context::TurnContext; @@ -601,25 +599,15 @@ impl Session { ) }; if !pending_input.is_empty() { + let hooks = self.hooks(); for pending_input_item in pending_input { - let hook_outcome = - inspect_pending_input(self, &turn_context, &pending_input_item).await; - if hook_outcome.should_stop { - record_additional_contexts( - self, - &turn_context, - hook_outcome.additional_contexts, - ) - .await; - } else { - record_pending_input( - self, - &turn_context, - pending_input_item, - hook_outcome.additional_contexts, - ) - .await; - } + run_user_prompt_submit_hook_and_record_input( + &hooks, + self, + &turn_context, + &pending_input_item, + ) + .await; } } // Emit token usage metrics. diff --git a/codex-rs/core/tests/suite/hooks.rs b/codex-rs/core/tests/suite/hooks.rs index a5c82d5767..46337aaefe 100644 --- a/codex-rs/core/tests/suite/hooks.rs +++ b/codex-rs/core/tests/suite/hooks.rs @@ -25,6 +25,7 @@ use codex_utils_absolute_path::AbsolutePathBuf; use core_test_support::hooks::trust_discovered_hooks; use core_test_support::hooks::trust_hooks; use core_test_support::managed_network_requirements_loader; +use core_test_support::responses::ResponseMock; use core_test_support::responses::ev_apply_patch_custom_tool_call; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; @@ -319,6 +320,199 @@ with Path(r"{log_path}").open("a", encoding="utf-8") as handle: Ok(()) } +struct TestCommandHook { + filename: &'static str, + source: String, + asynchronous: bool, +} + +fn write_command_hooks(home: &Path, event_name: &str, scripts: Vec) -> Result<()> { + let mut handlers = Vec::with_capacity(scripts.len()); + for script in scripts { + let script_path = home.join(script.filename); + fs::write(&script_path, script.source) + .with_context(|| format!("write test hook {}", script_path.display()))?; + handlers.push(serde_json::json!({ + "type": "command", + "command": format!("python3 {}", script_path.display()), + "async": script.asynchronous, + })); + } + let mut events = serde_json::Map::new(); + events.insert( + event_name.to_string(), + serde_json::json!([{ "hooks": handlers }]), + ); + fs::write( + home.join("hooks.json"), + serde_json::json!({ "hooks": events }).to_string(), + ) + .context("write hooks.json")?; + Ok(()) +} + +fn async_user_prompt_submit_hook(home: &Path) -> TestCommandHook { + let ready_path = home.join("async_user_prompt_submit_ready"); + let source = format!( + r#"import json +from pathlib import Path +import sys + +payload = json.load(sys.stdin) +prompt = payload.get("prompt") +print(json.dumps({{ + "hookSpecificOutput": {{ + "hookEventName": "UserPromptSubmit", + "additionalContext": f"async context for {{prompt}}" + }} +}})) +Path(r"{ready_path}").write_text(prompt, encoding="utf-8") +"#, + ready_path = ready_path.display(), + ); + TestCommandHook { + filename: "async_user_prompt_submit_hook.py", + source, + asynchronous: true, + } +} + +fn write_async_user_prompt_submit_hook(home: &Path) -> Result<()> { + write_command_hooks( + home, + "UserPromptSubmit", + vec![async_user_prompt_submit_hook(home)], + ) +} + +fn write_blocking_async_user_prompt_submit_hooks(home: &Path) -> Result<()> { + let blocking_script = r#"import json +import sys +import time + +payload = json.load(sys.stdin) +if payload.get("prompt") == "blocked first prompt": + print(json.dumps({"decision": "block", "reason": "blocked synchronously"})) +else: + time.sleep(0.2) +"# + .to_string(); + write_command_hooks( + home, + "UserPromptSubmit", + vec![ + TestCommandHook { + filename: "blocking_user_prompt_submit_hook.py", + source: blocking_script, + asynchronous: false, + }, + async_user_prompt_submit_hook(home), + ], + ) +} + +fn write_gated_async_user_prompt_submit_hook(home: &Path) -> Result<()> { + let started_path = home.join("gated_async_user_prompt_submit_started"); + let release_path = home.join("gated_async_user_prompt_submit_release"); + let ready_path = home.join("gated_async_user_prompt_submit_ready"); + let script = format!( + r#"import json +from pathlib import Path +import sys +import time + +payload = json.load(sys.stdin) +prompt = payload.get("prompt") +Path(r"{started_path}").write_text(prompt, encoding="utf-8") +release_path = Path(r"{release_path}") +while not release_path.exists(): + time.sleep(0.01) +print(json.dumps({{ + "hookSpecificOutput": {{ + "hookEventName": "UserPromptSubmit", + "additionalContext": f"async context surviving refresh for {{prompt}}" + }} +}})) +Path(r"{ready_path}").write_text(prompt, encoding="utf-8") +"#, + started_path = started_path.display(), + release_path = release_path.display(), + ready_path = ready_path.display(), + ); + write_command_hooks( + home, + "UserPromptSubmit", + vec![TestCommandHook { + filename: "gated_async_user_prompt_submit_hook.py", + source: script, + asynchronous: true, + }], + ) +} + +fn write_async_session_start_hook(home: &Path) -> Result<()> { + let ready_path = home.join("async_session_start_ready"); + let script = format!( + r#"import json +from pathlib import Path +import sys + +json.load(sys.stdin) +print(json.dumps({{ + "systemMessage": "async startup message", + "hookSpecificOutput": {{ + "hookEventName": "SessionStart", + "additionalContext": "async startup context" + }} +}})) +Path(r"{ready_path}").write_text("ready", encoding="utf-8") +"#, + ready_path = ready_path.display(), + ); + write_command_hooks( + home, + "SessionStart", + vec![TestCommandHook { + filename: "async_session_start_hook.py", + source: script, + asynchronous: true, + }], + ) +} + +async fn wait_for_async_hook(path: &Path) -> Result<()> { + timeout(Duration::from_secs(5), async { + while !path.exists() { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .context("timed out waiting for async hook")?; + // Test scripts write their marker immediately before exiting. Give the + // parent task time to observe process completion and queue its output. + sleep(Duration::from_millis(100)).await; + Ok(()) +} + +async fn mount_two_turn_responses(server: &wiremock::MockServer) -> ResponseMock { + mount_sse_sequence( + server, + ["first", "second"] + .into_iter() + .enumerate() + .map(|(index, message)| { + let number = index + 1; + sse(vec![ + ev_response_created(&format!("resp-{number}")), + ev_assistant_message(&format!("msg-{number}"), message), + ev_completed(&format!("resp-{number}")), + ]) + }) + .collect(), + ) + .await +} + fn write_pre_tool_use_hook( home: &Path, matcher: Option<&str>, @@ -1259,6 +1453,283 @@ async fn session_start_runs_before_user_prompt_submit_on_first_turn() -> Result< Ok(()) } +#[tokio::test] +async fn async_user_prompt_output_waits_for_next_accepted_request() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let responses = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "accepted"), + ev_completed("resp-1"), + ]), + ) + .await; + + let mut builder = test_codex() + .with_pre_build_hook(|home| { + write_blocking_async_user_prompt_submit_hooks(home) + .expect("write async user prompt submit hooks"); + }) + .with_config(trust_discovered_hooks); + let test = builder.build(&server).await?; + + test.submit_turn("blocked first prompt").await?; + wait_for_async_hook( + &test + .codex_home_path() + .join("async_user_prompt_submit_ready"), + ) + .await?; + assert_eq!(responses.requests().len(), 0); + + test.submit_turn("accepted second prompt").await?; + + let request = responses.single_request(); + let developer_messages = request.message_input_texts("developer"); + assert!(developer_messages.contains(&"async context for blocked first prompt".to_string())); + assert!(!developer_messages.contains(&"async context for accepted second prompt".to_string())); + assert!( + request + .message_input_texts("user") + .iter() + .all(|text| !text.contains("blocked first prompt")) + ); + + Ok(()) +} + +#[tokio::test] +async fn async_user_prompt_output_waits_for_fresh_turn_after_steer() -> Result<()> { + skip_if_no_network!(Ok(())); + + let (gate_completed_tx, gate_completed_rx) = oneshot::channel(); + let first_chunks = vec![ + StreamingSseChunk { + gate: None, + body: sse_event(ev_response_created("resp-1")), + }, + StreamingSseChunk { + gate: None, + body: sse_event(ev_message_item_added("msg-1", "")), + }, + StreamingSseChunk { + gate: None, + body: sse_event(ev_output_text_delta("first response")), + }, + StreamingSseChunk { + gate: None, + body: sse_event(ev_message_item_done("msg-1", "first response")), + }, + StreamingSseChunk { + gate: Some(gate_completed_rx), + body: sse_event(ev_completed("resp-1")), + }, + ]; + let second_chunks = vec![StreamingSseChunk { + gate: None, + body: sse(vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-2", "steer handled"), + ev_completed("resp-2"), + ]), + }]; + let third_chunks = vec![StreamingSseChunk { + gate: None, + body: sse(vec![ + ev_response_created("resp-3"), + ev_assistant_message("msg-3", "fresh turn handled"), + ev_completed("resp-3"), + ]), + }]; + let (server, _completions) = + start_streaming_sse_server(vec![first_chunks, second_chunks, third_chunks]).await; + + let mut builder = test_codex() + .with_pre_build_hook(|home| { + write_async_user_prompt_submit_hook(home).expect("write async user prompt submit hook"); + }) + .with_config(trust_discovered_hooks); + let test = builder.build_with_streaming_server(&server).await?; + + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "initial prompt".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::AgentMessageContentDelta(_)) + }) + .await; + wait_for_async_hook( + &test + .codex_home_path() + .join("async_user_prompt_submit_ready"), + ) + .await?; + test.codex + .steer_input( + vec![UserInput::Text { + text: "steered prompt".to_string(), + text_elements: Vec::new(), + }], + /*additional_context*/ Default::default(), + /*expected_turn_id*/ None, + /*client_user_message_id*/ None, + /*responsesapi_client_metadata*/ None, + ) + .await + .unwrap_or_else(|err| panic!("steer user input: {err:?}")); + let _ = gate_completed_tx.send(()); + + timeout( + Duration::from_secs(30), + server.wait_for_request_count(/*count*/ 2), + ) + .await + .expect("second request should arrive"); + let requests = server.requests().await; + assert_eq!(requests.len(), 2); + assert!( + !request_message_input_texts(&requests[1], "developer") + .contains(&"async context for initial prompt".to_string()) + ); + + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.submit_turn("following fresh prompt").await?; + + let requests = server.requests().await; + assert_eq!(requests.len(), 3); + assert!( + request_message_input_texts(&requests[2], "developer") + .contains(&"async context for initial prompt".to_string()) + ); + + Ok(()) +} + +#[tokio::test] +async fn async_startup_output_skips_first_accepted_request() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let responses = mount_two_turn_responses(&server).await; + + let mut builder = test_codex() + .with_pre_build_hook(|home| { + write_async_session_start_hook(home).expect("write async session start hook"); + }) + .with_config(trust_discovered_hooks); + let test = builder.build(&server).await?; + + test.submit_turn("first prompt").await?; + wait_for_async_hook(&test.codex_home_path().join("async_session_start_ready")).await?; + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "second prompt".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + let warning = wait_for_event(&test.codex, |event| matches!(event, EventMsg::Warning(_))).await; + let EventMsg::Warning(warning) = warning else { + unreachable!("waited for warning event") + }; + assert_eq!(warning.message, "async startup message"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let requests = responses.requests(); + assert_eq!(requests.len(), 2); + assert!( + !requests[0] + .message_input_texts("developer") + .contains(&"async startup context".to_string()) + ); + assert!( + requests[1] + .message_input_texts("developer") + .contains(&"async startup context".to_string()) + ); + assert!(requests.iter().all(|request| { + !request + .body_json() + .to_string() + .contains("async startup message") + })); + + Ok(()) +} + +#[tokio::test] +async fn async_output_survives_runtime_config_refresh() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let responses = mount_two_turn_responses(&server).await; + + let mut builder = test_codex() + .with_pre_build_hook(|home| { + write_gated_async_user_prompt_submit_hook(home) + .expect("write gated async user prompt submit hook"); + }) + .with_config(trust_discovered_hooks); + let test = builder.build(&server).await?; + + test.submit_turn("before refresh").await?; + wait_for_async_hook( + &test + .codex_home_path() + .join("gated_async_user_prompt_submit_started"), + ) + .await?; + + test.codex.refresh_runtime_config(test.config.clone()).await; + fs::write( + test.codex_home_path() + .join("gated_async_user_prompt_submit_release"), + "release", + )?; + wait_for_async_hook( + &test + .codex_home_path() + .join("gated_async_user_prompt_submit_ready"), + ) + .await?; + + test.submit_turn("after refresh").await?; + + let requests = responses.requests(); + assert_eq!(requests.len(), 2); + assert!( + requests[1] + .message_input_texts("developer") + .contains(&"async context surviving refresh for before refresh".to_string()) + ); + + Ok(()) +} + #[tokio::test] async fn session_start_hook_spills_large_additional_context() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/hooks/src/engine/command_runner.rs b/codex-rs/hooks/src/engine/command_runner.rs index 9f4c4e6059..889ed3f194 100644 --- a/codex-rs/hooks/src/engine/command_runner.rs +++ b/codex-rs/hooks/src/engine/command_runner.rs @@ -16,7 +16,6 @@ use super::dispatcher::hook_handler_type_label; use super::dispatcher::hook_scope_label; use super::dispatcher::hook_source_label; use super::dispatcher::scope_for_event; -use codex_protocol::protocol::HookExecutionMode; use codex_protocol::protocol::HookHandlerType; #[derive(Debug)] @@ -37,7 +36,7 @@ pub(crate) struct CommandRunResult { fields( hook.event_name = hook_event_name_label(handler.event_name), hook.handler_type = hook_handler_type_label(HookHandlerType::Command), - hook.execution_mode = hook_execution_mode_label(HookExecutionMode::Sync), + hook.execution_mode = hook_execution_mode_label(handler.execution_mode), hook.scope = hook_scope_label(scope_for_event(handler.event_name)), hook.source = hook_source_label(handler.source), hook.display_order = handler.display_order, diff --git a/codex-rs/hooks/src/engine/discovery.rs b/codex-rs/hooks/src/engine/discovery.rs index cab28b64a1..179c153cfe 100644 --- a/codex-rs/hooks/src/engine/discovery.rs +++ b/codex-rs/hooks/src/engine/discovery.rs @@ -27,6 +27,7 @@ use super::HookListEntry; use crate::config_rules::hook_states_from_stack; use crate::events::common::matcher_pattern_for_event; use crate::events::common::validate_matcher_pattern; +use codex_protocol::protocol::HookExecutionMode; use codex_protocol::protocol::HookHandlerType; use codex_protocol::protocol::HookSource; use codex_protocol::protocol::HookTrustStatus; @@ -472,13 +473,6 @@ fn append_matcher_groups( } else { command }; - if r#async { - warnings.push(format!( - "skipping async hook in {}: async hooks are not supported yet", - source.path.display() - )); - continue; - } if command.trim().is_empty() { warnings.push(format!( "skipping empty hook command in {}", @@ -507,6 +501,11 @@ fn append_matcher_groups( let trusted_hash = hook_trusted_hash(source.is_managed, state); let trust_status = hook_trust_status(source.is_managed, ¤t_hash, trusted_hash); + let execution_mode = if r#async { + HookExecutionMode::Async + } else { + HookExecutionMode::Sync + }; hook_entries.push(HookListEntry { key, event_name, @@ -541,6 +540,7 @@ fn append_matcher_groups( source: source.source, display_order: *display_order, env: source.env.clone(), + execution_mode, }); } *display_order += 1; @@ -658,6 +658,7 @@ mod tests { use codex_config::HookEventsToml; use codex_config::RequirementSource; use codex_protocol::protocol::HookEventName; + use codex_protocol::protocol::HookExecutionMode; use codex_protocol::protocol::HookSource; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::test_support::PathBufExt; @@ -792,6 +793,7 @@ mod tests { source: hook_source(), display_order: 0, env: std::collections::HashMap::new(), + execution_mode: HookExecutionMode::Sync, }] ); } @@ -827,6 +829,7 @@ mod tests { source: hook_source(), display_order: 0, env: std::collections::HashMap::new(), + execution_mode: HookExecutionMode::Sync, }] ); } diff --git a/codex-rs/hooks/src/engine/dispatcher.rs b/codex-rs/hooks/src/engine/dispatcher.rs index ccf5c2c8e6..0325a03c14 100644 --- a/codex-rs/hooks/src/engine/dispatcher.rs +++ b/codex-rs/hooks/src/engine/dispatcher.rs @@ -3,6 +3,7 @@ use std::path::Path; use futures::StreamExt; use futures::stream::FuturesUnordered; +use codex_protocol::ThreadId; use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::HookEventName; use codex_protocol::protocol::HookExecutionMode; @@ -13,9 +14,12 @@ use codex_protocol::protocol::HookScope; use super::CommandShell; use super::ConfiguredHandler; +use super::async_command::AsyncCommandRuntime; +use super::async_command::PendingAsyncHookDelivery; use super::command_runner::CommandRunResult; use super::command_runner::run_command; use crate::events::common::matches_matcher; +use crate::output_spill::HookOutputSpiller; #[derive(Debug)] pub(crate) struct ParsedHandler { @@ -24,13 +28,108 @@ pub(crate) struct ParsedHandler { pub completion_order: usize, } +/// Executes command hooks while hiding sync/async scheduling from event modules. +#[derive(Clone)] +pub(crate) struct CommandHookExecutor { + shell: CommandShell, + async_runtime: AsyncCommandRuntime, +} + +impl CommandHookExecutor { + pub(crate) fn new(shell: CommandShell) -> Self { + Self { + shell, + async_runtime: AsyncCommandRuntime::new(), + } + } + + pub(crate) fn preserve_runtime_from(&mut self, previous: &Self) { + self.async_runtime = previous.async_runtime.clone(); + } + + pub(crate) fn output_spiller(&self) -> &HookOutputSpiller { + self.async_runtime.output_spiller() + } + + pub(crate) fn pending_async_delivery(&self) -> PendingAsyncHookDelivery { + self.async_runtime.pending_delivery() + } + + pub(crate) async fn shutdown(&self) { + self.async_runtime.shutdown().await; + } + + pub(crate) async fn execute( + &self, + handlers: Vec, + input_json: String, + cwd: &Path, + turn_id: Option, + session_id: ThreadId, + parse: fn(&ConfiguredHandler, CommandRunResult, Option) -> ParsedHandler, + ) -> Vec> { + let mut pending = FuturesUnordered::new(); + for (configured_order, handler) in handlers.into_iter().enumerate() { + let input_json = input_json.clone(); + let turn_id = turn_id.clone(); + match handler.execution_mode { + HookExecutionMode::Sync => { + pending.push(async move { + let result = + run_command(&self.shell, &handler, configured_order, &input_json, cwd) + .await; + (configured_order, parse(&handler, result, turn_id)) + }); + } + HookExecutionMode::Async => self.async_runtime.spawn( + self.shell.clone(), + handler, + configured_order, + input_json, + cwd.to_path_buf(), + session_id, + ), + } + } + + let mut completed = Vec::new(); + let mut completion_order = 0; + while let Some((configured_order, mut parsed)) = pending.next().await { + parsed.completion_order = completion_order; + completion_order += 1; + completed.push((configured_order, parsed)); + } + completed.sort_by_key(|(configured_order, _)| *configured_order); + completed.into_iter().map(|(_, parsed)| parsed).collect() + } +} + pub(crate) fn select_handlers( handlers: &[ConfiguredHandler], event_name: HookEventName, matcher_input: Option<&str>, ) -> Vec { - let matcher_inputs = matcher_input.into_iter().collect::>(); - select_handlers_for_matcher_inputs(handlers, event_name, &matcher_inputs) + select_handlers_for_matcher_inputs(handlers, event_name, matcher_input.as_slice()) +} + +pub(crate) fn preview_handlers( + handlers: &[ConfiguredHandler], + event_name: HookEventName, + matcher_input: Option<&str>, +) -> Vec { + preview_handlers_for_matcher_inputs(handlers, event_name, matcher_input.as_slice()) +} + +pub(crate) fn preview_handlers_for_matcher_inputs( + handlers: &[ConfiguredHandler], + event_name: HookEventName, + matcher_inputs: &[&str], +) -> Vec { + select_handlers_for_matcher_inputs(handlers, event_name, matcher_inputs) + .into_iter() + .filter(|handler| handler.execution_mode == HookExecutionMode::Sync) + .map(|handler| running_summary(&handler)) + .collect() } pub(crate) fn select_handlers_for_matcher_inputs( @@ -72,7 +171,7 @@ pub(crate) fn running_summary(handler: &ConfiguredHandler) -> HookRunSummary { id: handler.run_id(), event_name: handler.event_name, handler_type: HookHandlerType::Command, - execution_mode: HookExecutionMode::Sync, + execution_mode: handler.execution_mode, scope: scope_for_event(handler.event_name), source_path: handler.source_path.clone(), source: handler.source, @@ -86,35 +185,6 @@ pub(crate) fn running_summary(handler: &ConfiguredHandler) -> HookRunSummary { } } -pub(crate) async fn execute_handlers( - shell: &CommandShell, - handlers: Vec, - input_json: String, - cwd: &Path, - turn_id: Option, - parse: fn(&ConfiguredHandler, CommandRunResult, Option) -> ParsedHandler, -) -> Vec> { - let mut pending = FuturesUnordered::new(); - for (configured_order, handler) in handlers.into_iter().enumerate() { - let input_json = input_json.clone(); - let turn_id = turn_id.clone(); - pending.push(async move { - let result = run_command(shell, &handler, configured_order, &input_json, cwd).await; - (configured_order, parse(&handler, result, turn_id)) - }); - } - - let mut completed = Vec::new(); - let mut completion_order = 0; - while let Some((configured_order, mut parsed)) = pending.next().await { - parsed.completion_order = completion_order; - completion_order += 1; - completed.push((configured_order, parsed)); - } - completed.sort_by_key(|(configured_order, _)| *configured_order); - completed.into_iter().map(|(_, parsed)| parsed).collect() -} - pub(crate) fn completed_summary( handler: &ConfiguredHandler, run_result: &CommandRunResult, @@ -125,7 +195,7 @@ pub(crate) fn completed_summary( id: handler.run_id(), event_name: handler.event_name, handler_type: HookHandlerType::Command, - execution_mode: HookExecutionMode::Sync, + execution_mode: handler.execution_mode, scope: scope_for_event(handler.event_name), source_path: handler.source_path.clone(), source: handler.source, @@ -211,6 +281,7 @@ pub(crate) fn hook_source_label(source: codex_protocol::protocol::HookSource) -> #[cfg(test)] mod tests { use codex_protocol::protocol::HookEventName; + use codex_protocol::protocol::HookExecutionMode; use codex_protocol::protocol::HookSource; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; @@ -236,6 +307,7 @@ mod tests { source: HookSource::User, display_order, env: std::collections::HashMap::new(), + execution_mode: HookExecutionMode::Sync, } } diff --git a/codex-rs/hooks/src/engine/mod.rs b/codex-rs/hooks/src/engine/mod.rs index 171e292414..6d1a2e8042 100644 --- a/codex-rs/hooks/src/engine/mod.rs +++ b/codex-rs/hooks/src/engine/mod.rs @@ -1,4 +1,3 @@ -#[allow(dead_code)] pub(crate) mod async_command; pub(crate) mod command_runner; pub(crate) mod discovery; @@ -22,7 +21,6 @@ use crate::events::stop::StopOutcome; use crate::events::stop::StopRequest; use crate::events::user_prompt_submit::UserPromptSubmitOutcome; use crate::events::user_prompt_submit::UserPromptSubmitRequest; -use crate::output_spill::HookOutputSpiller; use codex_config::ConfigLayerStack; use codex_plugin::PluginHookSource; use codex_protocol::ThreadId; @@ -34,6 +32,9 @@ use codex_protocol::protocol::HookTrustStatus; use codex_utils_absolute_path::AbsolutePathBuf; use std::collections::HashMap; +pub use async_command::AsyncHookDelivery; +pub use async_command::PendingAsyncHookDelivery; +use dispatcher::CommandHookExecutor; #[derive(Debug, Clone)] pub(crate) struct CommandShell { pub program: String, @@ -51,6 +52,7 @@ pub(crate) struct ConfiguredHandler { pub source: HookSource, pub display_order: i64, pub env: HashMap, + pub execution_mode: codex_protocol::protocol::HookExecutionMode, } impl ConfiguredHandler { @@ -102,8 +104,7 @@ pub struct HookListEntry { pub(crate) struct ClaudeHooksEngine { handlers: Vec, warnings: Vec, - shell: CommandShell, - output_spiller: HookOutputSpiller, + executor: CommandHookExecutor, } impl ClaudeHooksEngine { @@ -115,12 +116,12 @@ impl ClaudeHooksEngine { plugin_hook_load_warnings: Vec, shell: CommandShell, ) -> Self { + let executor = CommandHookExecutor::new(shell); if !enabled { return Self { handlers: Vec::new(), warnings: Vec::new(), - shell, - output_spiller: HookOutputSpiller::new(), + executor, }; } @@ -134,11 +135,22 @@ impl ClaudeHooksEngine { Self { handlers: discovered.handlers, warnings: discovered.warnings, - shell, - output_spiller: HookOutputSpiller::new(), + executor, } } + pub(crate) fn preserve_runtime_from(&mut self, previous: &Self) { + self.executor.preserve_runtime_from(&previous.executor); + } + + pub(crate) fn pending_async_delivery(&self) -> PendingAsyncHookDelivery { + self.executor.pending_async_delivery() + } + + pub(crate) async fn shutdown(&self) { + self.executor.shutdown().await; + } + pub(crate) fn warnings(&self) -> &[String] { &self.warnings } @@ -175,7 +187,8 @@ impl ClaudeHooksEngine { ) -> SessionStartOutcome { let session_id = request.session_id; let mut outcome = - crate::events::session_start::run(&self.handlers, &self.shell, request, turn_id).await; + crate::events::session_start::run(&self.handlers, &self.executor, request, turn_id) + .await; outcome.additional_contexts = self .maybe_spill_texts(session_id, outcome.additional_contexts) .await; @@ -185,7 +198,7 @@ impl ClaudeHooksEngine { pub(crate) async fn run_pre_tool_use(&self, request: PreToolUseRequest) -> PreToolUseOutcome { let session_id = request.session_id; let mut outcome = - crate::events::pre_tool_use::run(&self.handlers, &self.shell, request).await; + crate::events::pre_tool_use::run(&self.handlers, &self.executor, request).await; outcome.additional_contexts = self .maybe_spill_texts(session_id, outcome.additional_contexts) .await; @@ -196,7 +209,7 @@ impl ClaudeHooksEngine { &self, request: PermissionRequestRequest, ) -> PermissionRequestOutcome { - crate::events::permission_request::run(&self.handlers, &self.shell, request).await + crate::events::permission_request::run(&self.handlers, &self.executor, request).await } pub(crate) async fn run_post_tool_use( @@ -205,7 +218,7 @@ impl ClaudeHooksEngine { ) -> PostToolUseOutcome { let session_id = request.session_id; let mut outcome = - crate::events::post_tool_use::run(&self.handlers, &self.shell, request).await; + crate::events::post_tool_use::run(&self.handlers, &self.executor, request).await; outcome.additional_contexts = self .maybe_spill_texts(session_id, outcome.additional_contexts) .await; @@ -220,7 +233,7 @@ impl ClaudeHooksEngine { } pub(crate) async fn run_pre_compact(&self, request: PreCompactRequest) -> PreCompactOutcome { - crate::events::compact::run_pre(&self.handlers, &self.shell, request).await + crate::events::compact::run_pre(&self.handlers, &self.executor, request).await } pub(crate) fn preview_post_compact(&self, request: &PostCompactRequest) -> Vec { @@ -231,7 +244,7 @@ impl ClaudeHooksEngine { &self, request: PostCompactRequest, ) -> StatelessHookOutcome { - crate::events::compact::run_post(&self.handlers, &self.shell, request).await + crate::events::compact::run_post(&self.handlers, &self.executor, request).await } pub(crate) fn preview_user_prompt_submit( @@ -247,7 +260,7 @@ impl ClaudeHooksEngine { ) -> UserPromptSubmitOutcome { let session_id = request.session_id; let mut outcome = - crate::events::user_prompt_submit::run(&self.handlers, &self.shell, request).await; + crate::events::user_prompt_submit::run(&self.handlers, &self.executor, request).await; outcome.additional_contexts = self .maybe_spill_texts(session_id, outcome.additional_contexts) .await; @@ -260,7 +273,7 @@ impl ClaudeHooksEngine { pub(crate) async fn run_stop(&self, request: StopRequest) -> StopOutcome { let session_id = request.session_id; - let mut outcome = crate::events::stop::run(&self.handlers, &self.shell, request).await; + let mut outcome = crate::events::stop::run(&self.handlers, &self.executor, request).await; outcome.continuation_fragments = self .maybe_spill_prompt_fragments(session_id, outcome.continuation_fragments) .await; @@ -268,16 +281,17 @@ impl ClaudeHooksEngine { } async fn maybe_spill_texts(&self, session_id: ThreadId, texts: Vec) -> Vec { - self.output_spiller + self.executor + .output_spiller() .maybe_spill_texts(session_id, texts) .await } async fn maybe_spill_text(&self, session_id: ThreadId, text: Option) -> Option { - match text { - Some(text) => Some(self.output_spiller.maybe_spill_text(session_id, text).await), - None => None, - } + self.executor + .output_spiller() + .maybe_spill_optional_text(session_id, text) + .await } async fn maybe_spill_prompt_fragments( @@ -285,7 +299,8 @@ impl ClaudeHooksEngine { session_id: ThreadId, fragments: Vec, ) -> Vec { - self.output_spiller + self.executor + .output_spiller() .maybe_spill_prompt_fragments(session_id, fragments) .await } diff --git a/codex-rs/hooks/src/events/common.rs b/codex-rs/hooks/src/events/common.rs index 997eac139f..9322ef824c 100644 --- a/codex-rs/hooks/src/events/common.rs +++ b/codex-rs/hooks/src/events/common.rs @@ -1,5 +1,6 @@ use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::HookEventName; +use codex_protocol::protocol::HookExecutionMode; use codex_protocol::protocol::HookOutputEntry; use codex_protocol::protocol::HookOutputEntryKind; use codex_protocol::protocol::HookRunStatus; @@ -60,6 +61,7 @@ pub(crate) fn serialization_failure_hook_events( ) -> Vec { handlers .into_iter() + .filter(|handler| handler.execution_mode == HookExecutionMode::Sync) .map(|handler| { let mut run = dispatcher::running_summary(&handler); run.status = HookRunStatus::Failed; diff --git a/codex-rs/hooks/src/events/compact.rs b/codex-rs/hooks/src/events/compact.rs index cb3080219a..ad332ac76a 100644 --- a/codex-rs/hooks/src/events/compact.rs +++ b/codex-rs/hooks/src/events/compact.rs @@ -10,10 +10,10 @@ use codex_protocol::protocol::HookRunSummary; use codex_utils_absolute_path::AbsolutePathBuf; use super::common; -use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; +use crate::engine::dispatcher::CommandHookExecutor; use crate::engine::output_parser; use crate::schema::PostCompactCommandInput; use crate::schema::PreCompactCommandInput; @@ -59,19 +59,16 @@ pub(crate) fn preview_pre( handlers: &[ConfiguredHandler], request: &PreCompactRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::preview_handlers( handlers, HookEventName::PreCompact, Some(request.trigger.as_str()), ) - .into_iter() - .map(|handler| dispatcher::running_summary(&handler)) - .collect() } pub(crate) async fn run_pre( handlers: &[ConfiguredHandler], - shell: &CommandShell, + executor: &CommandHookExecutor, request: PreCompactRequest, ) -> PreCompactOutcome { let matched = dispatcher::select_handlers( @@ -102,15 +99,16 @@ pub(crate) async fn run_pre( } }; - let results = dispatcher::execute_handlers( - shell, - matched, - input_json, - request.cwd.as_path(), - Some(request.turn_id), - parse_pre_completed, - ) - .await; + let results = executor + .execute( + matched, + input_json, + request.cwd.as_path(), + Some(request.turn_id), + request.session_id, + parse_pre_completed, + ) + .await; let should_stop = results.iter().any(|result| result.data.should_stop); let stop_reason = results .iter() @@ -141,19 +139,16 @@ pub(crate) fn preview_post( handlers: &[ConfiguredHandler], request: &PostCompactRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::preview_handlers( handlers, HookEventName::PostCompact, Some(request.trigger.as_str()), ) - .into_iter() - .map(|handler| dispatcher::running_summary(&handler)) - .collect() } pub(crate) async fn run_post( handlers: &[ConfiguredHandler], - shell: &CommandShell, + executor: &CommandHookExecutor, request: PostCompactRequest, ) -> StatelessHookOutcome { let matched = dispatcher::select_handlers( @@ -184,15 +179,16 @@ pub(crate) async fn run_post( } }; - let results = dispatcher::execute_handlers( - shell, - matched, - input_json, - request.cwd.as_path(), - Some(request.turn_id), - parse_post_completed, - ) - .await; + let results = executor + .execute( + matched, + input_json, + request.cwd.as_path(), + Some(request.turn_id), + request.session_id, + parse_post_completed, + ) + .await; let should_stop = results.iter().any(|result| result.data.should_stop); let stop_reason = results .iter() @@ -604,6 +600,7 @@ mod tests { source: codex_protocol::protocol::HookSource::User, display_order: 0, env: std::collections::HashMap::new(), + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, } } diff --git a/codex-rs/hooks/src/events/permission_request.rs b/codex-rs/hooks/src/events/permission_request.rs index db7970f02d..97626afcc0 100644 --- a/codex-rs/hooks/src/events/permission_request.rs +++ b/codex-rs/hooks/src/events/permission_request.rs @@ -16,10 +16,10 @@ use std::path::PathBuf; use super::common; -use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; +use crate::engine::dispatcher::CommandHookExecutor; use crate::engine::output_parser; use crate::schema::PermissionRequestCommandInput; use crate::schema::SubagentCommandInputFields; @@ -69,24 +69,19 @@ pub(crate) fn preview( request: &PermissionRequestRequest, ) -> Vec { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); - dispatcher::select_handlers_for_matcher_inputs( + dispatcher::preview_handlers_for_matcher_inputs( handlers, HookEventName::PermissionRequest, &matcher_inputs, ) .into_iter() - .map(|handler| { - common::hook_run_for_tool_use( - dispatcher::running_summary(&handler), - &request.run_id_suffix, - ) - }) + .map(|run| common::hook_run_for_tool_use(run, &request.run_id_suffix)) .collect() } pub(crate) async fn run( handlers: &[ConfiguredHandler], - shell: &CommandShell, + executor: &CommandHookExecutor, request: PermissionRequestRequest, ) -> PermissionRequestOutcome { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); @@ -118,15 +113,16 @@ pub(crate) async fn run( } }; - let results = dispatcher::execute_handlers( - shell, - matched, - input_json, - request.cwd.as_path(), - Some(request.turn_id.clone()), - parse_completed, - ) - .await; + let results = executor + .execute( + matched, + input_json, + request.cwd.as_path(), + Some(request.turn_id.clone()), + request.session_id, + parse_completed, + ) + .await; // Preserve the most specific matching allow, but treat any deny as final so // broader policy layers cannot accidentally overrule a more specific block. diff --git a/codex-rs/hooks/src/events/post_tool_use.rs b/codex-rs/hooks/src/events/post_tool_use.rs index 71adb11cda..f2cefb12e1 100644 --- a/codex-rs/hooks/src/events/post_tool_use.rs +++ b/codex-rs/hooks/src/events/post_tool_use.rs @@ -11,10 +11,10 @@ use codex_utils_absolute_path::AbsolutePathBuf; use serde_json::Value; use super::common; -use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; +use crate::engine::dispatcher::CommandHookExecutor; use crate::engine::output_parser; use crate::schema::PostToolUseCommandInput; use crate::schema::SubagentCommandInputFields; @@ -55,21 +55,19 @@ pub(crate) fn preview( request: &PostToolUseRequest, ) -> Vec { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); - dispatcher::select_handlers_for_matcher_inputs( + dispatcher::preview_handlers_for_matcher_inputs( handlers, HookEventName::PostToolUse, &matcher_inputs, ) .into_iter() - .map(|handler| { - common::hook_run_for_tool_use(dispatcher::running_summary(&handler), &request.tool_use_id) - }) + .map(|run| common::hook_run_for_tool_use(run, &request.tool_use_id)) .collect() } pub(crate) async fn run( handlers: &[ConfiguredHandler], - shell: &CommandShell, + executor: &CommandHookExecutor, request: PostToolUseRequest, ) -> PostToolUseOutcome { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); @@ -100,15 +98,16 @@ pub(crate) async fn run( } }; - let results = dispatcher::execute_handlers( - shell, - matched, - input_json, - request.cwd.as_path(), - Some(request.turn_id.clone()), - parse_completed, - ) - .await; + let results = executor + .execute( + matched, + input_json, + request.cwd.as_path(), + Some(request.turn_id.clone()), + request.session_id, + parse_completed, + ) + .await; let additional_contexts = common::flatten_additional_contexts( results @@ -568,6 +567,7 @@ mod tests { source: codex_protocol::protocol::HookSource::User, display_order: 0, env: std::collections::HashMap::new(), + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, } } diff --git a/codex-rs/hooks/src/events/pre_tool_use.rs b/codex-rs/hooks/src/events/pre_tool_use.rs index b3579aba82..b365bab8c7 100644 --- a/codex-rs/hooks/src/events/pre_tool_use.rs +++ b/codex-rs/hooks/src/events/pre_tool_use.rs @@ -11,10 +11,10 @@ use codex_utils_absolute_path::AbsolutePathBuf; use serde_json::Value; use super::common; -use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; +use crate::engine::dispatcher::CommandHookExecutor; use crate::engine::output_parser; use crate::schema::PreToolUseCommandInput; use crate::schema::SubagentCommandInputFields; @@ -56,21 +56,19 @@ pub(crate) fn preview( request: &PreToolUseRequest, ) -> Vec { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); - dispatcher::select_handlers_for_matcher_inputs( + dispatcher::preview_handlers_for_matcher_inputs( handlers, HookEventName::PreToolUse, &matcher_inputs, ) .into_iter() - .map(|handler| { - common::hook_run_for_tool_use(dispatcher::running_summary(&handler), &request.tool_use_id) - }) + .map(|run| common::hook_run_for_tool_use(run, &request.tool_use_id)) .collect() } pub(crate) async fn run( handlers: &[ConfiguredHandler], - shell: &CommandShell, + executor: &CommandHookExecutor, request: PreToolUseRequest, ) -> PreToolUseOutcome { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); @@ -102,15 +100,16 @@ pub(crate) async fn run( } }; - let results = dispatcher::execute_handlers( - shell, - matched, - input_json, - request.cwd.as_path(), - Some(request.turn_id.clone()), - parse_completed, - ) - .await; + let results = executor + .execute( + matched, + input_json, + request.cwd.as_path(), + Some(request.turn_id.clone()), + request.session_id, + parse_completed, + ) + .await; let should_block = results.iter().any(|result| result.data.should_block); let block_reason = results @@ -749,6 +748,7 @@ mod tests { source: codex_protocol::protocol::HookSource::User, display_order: 0, env: std::collections::HashMap::new(), + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, } } diff --git a/codex-rs/hooks/src/events/session_start.rs b/codex-rs/hooks/src/events/session_start.rs index bd1aa2096f..9368da1744 100644 --- a/codex-rs/hooks/src/events/session_start.rs +++ b/codex-rs/hooks/src/events/session_start.rs @@ -10,10 +10,10 @@ use codex_protocol::protocol::HookRunSummary; use codex_utils_absolute_path::AbsolutePathBuf; use super::common; -use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; +use crate::engine::dispatcher::CommandHookExecutor; use crate::engine::output_parser; use crate::schema::NullableString; use crate::schema::SessionStartCommandInput; @@ -95,19 +95,16 @@ pub(crate) fn preview( handlers: &[ConfiguredHandler], request: &SessionStartRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::preview_handlers( handlers, request.target.event_name(), Some(request.target.matcher_input()), ) - .into_iter() - .map(|handler| dispatcher::running_summary(&handler)) - .collect() } pub(crate) async fn run( handlers: &[ConfiguredHandler], - shell: &CommandShell, + executor: &CommandHookExecutor, request: SessionStartRequest, turn_id: Option, ) -> SessionStartOutcome { @@ -180,15 +177,16 @@ pub(crate) async fn run( } }; - let results = dispatcher::execute_handlers( - shell, - matched, - input_json, - request.cwd.as_path(), - turn_id, - parse_completed, - ) - .await; + let results = executor + .execute( + matched, + input_json, + request.cwd.as_path(), + turn_id, + request.session_id, + parse_completed, + ) + .await; let should_stop = results.iter().any(|result| result.data.should_stop); let stop_reason = results @@ -523,6 +521,7 @@ mod tests { source: codex_protocol::protocol::HookSource::User, display_order: 0, env: std::collections::HashMap::new(), + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, } } diff --git a/codex-rs/hooks/src/events/stop.rs b/codex-rs/hooks/src/events/stop.rs index 24920c569b..7f0ff96679 100644 --- a/codex-rs/hooks/src/events/stop.rs +++ b/codex-rs/hooks/src/events/stop.rs @@ -11,10 +11,10 @@ use codex_protocol::protocol::HookRunSummary; use codex_utils_absolute_path::AbsolutePathBuf; use super::common; -use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; +use crate::engine::dispatcher::CommandHookExecutor; use crate::engine::output_parser; use crate::schema::NullableString; use crate::schema::StopCommandInput; @@ -82,19 +82,16 @@ pub(crate) fn preview( handlers: &[ConfiguredHandler], request: &StopRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::preview_handlers( handlers, request.target.event_name(), request.target.matcher_input(), ) - .into_iter() - .map(|handler| dispatcher::running_summary(&handler)) - .collect() } pub(crate) async fn run( handlers: &[ConfiguredHandler], - shell: &CommandShell, + executor: &CommandHookExecutor, request: StopRequest, ) -> StopOutcome { let matched = dispatcher::select_handlers( @@ -177,15 +174,16 @@ pub(crate) async fn run( } }; - let results = dispatcher::execute_handlers( - shell, - matched, - input_json, - request.cwd.as_path(), - Some(request.turn_id), - parse_completed, - ) - .await; + let results = executor + .execute( + matched, + input_json, + request.cwd.as_path(), + Some(request.turn_id), + request.session_id, + parse_completed, + ) + .await; let aggregate = aggregate_results(results.iter().map(|result| &result.data)); @@ -639,6 +637,7 @@ mod tests { source: codex_protocol::protocol::HookSource::User, display_order: 0, env: std::collections::HashMap::new(), + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, } } diff --git a/codex-rs/hooks/src/events/user_prompt_submit.rs b/codex-rs/hooks/src/events/user_prompt_submit.rs index 2934bd3523..f28eb7d32c 100644 --- a/codex-rs/hooks/src/events/user_prompt_submit.rs +++ b/codex-rs/hooks/src/events/user_prompt_submit.rs @@ -10,10 +10,10 @@ use codex_protocol::protocol::HookRunSummary; use codex_utils_absolute_path::AbsolutePathBuf; use super::common; -use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; +use crate::engine::dispatcher::CommandHookExecutor; use crate::engine::output_parser; use crate::schema::NullableString; use crate::schema::SubagentCommandInputFields; @@ -50,19 +50,16 @@ pub(crate) fn preview( handlers: &[ConfiguredHandler], _request: &UserPromptSubmitRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::preview_handlers( handlers, HookEventName::UserPromptSubmit, /*matcher_input*/ None, ) - .into_iter() - .map(|handler| dispatcher::running_summary(&handler)) - .collect() } pub(crate) async fn run( handlers: &[ConfiguredHandler], - shell: &CommandShell, + executor: &CommandHookExecutor, request: UserPromptSubmitRequest, ) -> UserPromptSubmitOutcome { let matched = dispatcher::select_handlers( @@ -102,15 +99,16 @@ pub(crate) async fn run( } }; - let results = dispatcher::execute_handlers( - shell, - matched, - input_json, - request.cwd.as_path(), - Some(request.turn_id), - parse_completed, - ) - .await; + let results = executor + .execute( + matched, + input_json, + request.cwd.as_path(), + Some(request.turn_id), + request.session_id, + parse_completed, + ) + .await; let should_stop = results.iter().any(|result| result.data.should_stop); let stop_reason = results @@ -428,6 +426,7 @@ mod tests { source: codex_protocol::protocol::HookSource::User, display_order: 0, env: std::collections::HashMap::new(), + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, } } diff --git a/codex-rs/hooks/src/lib.rs b/codex-rs/hooks/src/lib.rs index 11300802d8..f5a56067ea 100644 --- a/codex-rs/hooks/src/lib.rs +++ b/codex-rs/hooks/src/lib.rs @@ -13,7 +13,9 @@ use codex_protocol::protocol::HookEventName; pub use config_rules::hook_states_from_stack; pub use declarations::PluginHookDeclaration; pub use declarations::plugin_hook_declarations; +pub use engine::AsyncHookDelivery; pub use engine::HookListEntry; +pub use engine::PendingAsyncHookDelivery; pub use events::common::SubagentHookContext; /// Hook event names as they appear in hooks JSON and config files. pub const HOOK_EVENT_NAMES: [&str; 10] = [ diff --git a/codex-rs/hooks/src/registry.rs b/codex-rs/hooks/src/registry.rs index 1f4e01aa59..c4a66d37c3 100644 --- a/codex-rs/hooks/src/registry.rs +++ b/codex-rs/hooks/src/registry.rs @@ -5,6 +5,7 @@ use tokio::process::Command; use crate::engine::ClaudeHooksEngine; use crate::engine::CommandShell; use crate::engine::HookListEntry; +use crate::engine::PendingAsyncHookDelivery; use crate::events::compact::PostCompactRequest; use crate::events::compact::PreCompactOutcome; use crate::events::compact::PreCompactRequest; @@ -81,6 +82,20 @@ impl Hooks { } } + pub fn reconfigured(&self, config: HooksConfig) -> Self { + let mut reconfigured = Self::new(config); + reconfigured.engine.preserve_runtime_from(&self.engine); + reconfigured + } + + pub fn pending_async_delivery(&self) -> PendingAsyncHookDelivery { + self.engine.pending_async_delivery() + } + + pub async fn shutdown(&self) { + self.engine.shutdown().await; + } + pub fn startup_warnings(&self) -> &[String] { self.engine.warnings() }