diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index ec5bfebf7b..0d09cf38bb 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3088,6 +3088,7 @@ dependencies = [ "serde_json", "tempfile", "tokio", + "tokio-util", "tracing", "uuid", ] diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 7730a30ad7..235f831795 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -625,6 +625,7 @@ pub async fn set_thread_memory_mode(sess: &Arc, sub_id: String, mode: T async fn shutdown_session_runtime(sess: &Arc) { sess.abort_all_tasks(TurnAbortReason::Interrupted).await; + sess.services.async_hook_output_queue.shutdown().await; let _ = sess.conversation.shutdown().await; sess.services .unified_exec_manager diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 37d55256f5..db4cb05862 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1514,6 +1514,7 @@ impl Session { config.as_ref(), self.services.plugins_manager.as_ref(), self.services.user_shell.as_ref(), + self.services.async_hook_output_queue.clone(), ) .await; @@ -3403,6 +3404,7 @@ async fn build_hooks_for_config( config: &Config, plugins_manager: &PluginsManager, user_shell: &crate::shell::Shell, + async_output_queue: codex_hooks::AsyncHookOutputQueue, ) -> Hooks { let mut hook_shell_argv = user_shell.derive_exec_args("", /*use_login_shell*/ false); let hook_shell_program = hook_shell_argv.remove(0); @@ -3420,6 +3422,7 @@ async fn build_hooks_for_config( plugin_hook_load_warnings, shell_program: Some(hook_shell_program), shell_args: hook_shell_argv, + async_output_queue, }) } diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 92f9737310..15a4c78dd4 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -925,8 +925,14 @@ impl Session { (None, None) }; - let hooks = - build_hooks_for_config(&config, plugins_manager.as_ref(), &default_shell).await; + let async_hook_output_queue = codex_hooks::AsyncHookOutputQueue::default(); + let hooks = build_hooks_for_config( + &config, + plugins_manager.as_ref(), + &default_shell, + async_hook_output_queue.clone(), + ) + .await; for warning in hooks.startup_warnings() { post_session_configured_events.push(Event { id: INITIAL_SUBMIT_ID.to_owned(), @@ -991,6 +997,7 @@ impl Session { main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(), analytics_events_client, hooks: arc_swap::ArcSwap::from_pointee(hooks), + async_hook_output_queue, rollout_thread_trace, user_shell: Arc::new(default_shell), shell_snapshot_tx, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 4d914ff985..a41c8a6275 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -105,6 +105,7 @@ use codex_protocol::protocol::CompactedItem; use codex_protocol::protocol::ConversationAudioParams; use codex_protocol::protocol::CreditsSnapshot; use codex_protocol::protocol::GranularApprovalConfig; +use codex_protocol::protocol::HookEventName; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::MultiAgentVersion; @@ -4776,6 +4777,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { .expect("create environment"), ); + let async_hook_output_queue = codex_hooks::AsyncHookOutputQueue::default(); let services = SessionServices { mcp_connection_manager: Arc::new(RwLock::new( McpConnectionManager::new_uninitialized_with_permission_profile( @@ -4797,8 +4799,10 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { ), hooks: arc_swap::ArcSwap::from_pointee(Hooks::new(HooksConfig { legacy_notify_argv: config.notify.clone(), + async_output_queue: async_hook_output_queue.clone(), ..HooksConfig::default() })), + async_hook_output_queue, rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell: Arc::new(default_user_shell()), shell_snapshot_tx: watch::channel(None).0, @@ -6853,6 +6857,7 @@ where .expect("create environment"), ); + let async_hook_output_queue = codex_hooks::AsyncHookOutputQueue::default(); let services = SessionServices { mcp_connection_manager: Arc::new(RwLock::new( McpConnectionManager::new_uninitialized_with_permission_profile( @@ -6874,8 +6879,10 @@ where ), hooks: arc_swap::ArcSwap::from_pointee(Hooks::new(HooksConfig { legacy_notify_argv: config.notify.clone(), + async_output_queue: async_hook_output_queue.clone(), ..HooksConfig::default() })), + async_hook_output_queue, rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell: Arc::new(default_user_shell()), shell_snapshot_tx: watch::channel(None).0, @@ -8559,6 +8566,10 @@ async fn abort_gracefully_emits_marker_before_turn_aborted() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() { let (sess, tc, rx) = make_session_and_context_with_rx().await; + let queued_context = "queued async hook context"; + sess.services + .async_hook_output_queue + .push(HookEventName::PostToolUse, queued_context.to_string()); let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), @@ -8612,6 +8623,32 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() history.raw_items().iter().any(|item| item == &expected), "expected pending input to be persisted into history on turn completion" ); + let user_index = history + .raw_items() + .iter() + .position(|item| item == &expected) + .expect("pending user input in history"); + let developer_index = history + .raw_items() + .iter() + .position(|item| { + matches!( + item, + ResponseItem::Message { role, content, .. } + if role == "developer" + && content.iter().any(|content| { + matches!(content, ContentItem::InputText { text } if text.contains(queued_context)) + }) + ) + }) + .expect("queued async hook context in history"); + assert!(user_index < developer_index); + assert!( + sess.services + .async_hook_output_queue + .pending_batch() + .is_none() + ); let first = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await @@ -8663,11 +8700,17 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() )); let fifth = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) + .await + .expect("expected async hook context event") + .expect("channel open"); + assert!(matches!(fifth.msg, EventMsg::RawResponseItem(_))); + + let sixth = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected turn complete event") .expect("channel open"); assert!(matches!( - fifth.msg, + sixth.msg, EventMsg::TurnComplete(TurnCompleteEvent { turn_id, last_agent_message: None, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 9d4e7d1e82..e659375cf1 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -396,20 +396,34 @@ async fn turn_diff_display_roots(turn_context: &TurnContext) -> Vec<(String, Pat display_roots } -async fn run_hooks_and_record_inputs( +pub(crate) 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; + let mut async_hook_output_batch = None; + let mut async_hook_output_batch_prepared = false; for input_item in input { - let hook_outcome = inspect_pending_input(sess, turn_context, input_item).await; + let is_real_user_input = + matches!(input_item, TurnInput::UserInput { content, .. } if !content.is_empty()); + if is_real_user_input && !async_hook_output_batch_prepared { + async_hook_output_batch = sess.services.async_hook_output_queue.pending_batch(); + async_hook_output_batch_prepared = true; + } + let mut 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()) { + if is_real_user_input && let Some(batch) = async_hook_output_batch.take() { + sess.services.async_hook_output_queue.commit(&batch); + hook_outcome + .additional_contexts + .insert(0, batch.into_text()); + } + if is_real_user_input { accepted_user_input = true; } record_pending_input( diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 59def9c7ab..72ca673d67 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -22,6 +22,7 @@ use codex_core_plugins::PluginsManager; use codex_exec_server::EnvironmentManager; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionRegistry; +use codex_hooks::AsyncHookOutputQueue; use codex_hooks::Hooks; use codex_login::AuthManager; use codex_mcp::McpConnectionManager; @@ -48,6 +49,7 @@ pub(crate) struct SessionServices { pub(crate) main_execve_wrapper_exe: Option, pub(crate) analytics_events_client: AnalyticsEventsClient, pub(crate) hooks: ArcSwap, + pub(crate) async_hook_output_queue: AsyncHookOutputQueue, pub(crate) rollout_thread_trace: ThreadTraceContext, pub(crate) user_shell: Arc, pub(crate) shell_snapshot_tx: watch::Sender>>, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 911f30b319..9957920814 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -23,11 +23,9 @@ use tracing::warn; 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::session::TurnInput; use crate::session::session::Session; +use crate::session::turn::run_hooks_and_record_inputs; use crate::session::turn_context::TurnContext; use crate::state::ActiveTurn; use crate::state::RunningTask; @@ -583,26 +581,7 @@ impl Session { ) }; if !pending_input.is_empty() { - 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_hooks_and_record_inputs(self, &turn_context, &pending_input).await; } // Emit token usage metrics. { diff --git a/codex-rs/core/tests/suite/async_hooks.rs b/codex-rs/core/tests/suite/async_hooks.rs new file mode 100644 index 0000000000..20b9c1fa95 --- /dev/null +++ b/codex-rs/core/tests/suite/async_hooks.rs @@ -0,0 +1,209 @@ +use std::fs; +use std::path::Path; +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use codex_core::CodexThread; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; +use codex_protocol::user_input::UserInput; +use core_test_support::hooks::trust_discovered_hooks; +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::sse; +use core_test_support::responses::start_mock_server; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use pretty_assertions::assert_eq; +use serde_json::Value; +use tokio::time::sleep; +use tokio::time::timeout; + +const FIRST_PROMPT: &str = "start the detached hooks"; +const SECOND_PROMPT: &str = "use the detached hook output"; +const FIRST_CONTEXT: &str = "first detached hook context"; +const SECOND_CONTEXT: &str = "second detached hook context"; + +fn write_gated_async_hooks(home: &Path) -> Result<()> { + let script_path = home.join("async_hook.py"); + let script = format!( + r#"import json +from pathlib import Path +import sys +import time + +name = sys.argv[1] +if json.load(sys.stdin).get("prompt") != {FIRST_PROMPT:?}: + raise SystemExit(0) +root = Path(__file__).parent +if name == "sync": + completed = root / "first.completed" + deadline = time.time() + 10 + while not completed.exists() and time.time() < deadline: + time.sleep(0.01) + if not completed.exists(): + raise RuntimeError("timed out waiting for first async hook") + raise SystemExit(0) +release = root / f"{{name}}.release" +if name == "second": + while not release.exists(): + time.sleep(0.01) +context = {{"first": {FIRST_CONTEXT:?}, "second": {SECOND_CONTEXT:?}}}[name] +print(json.dumps({{"hookSpecificOutput": {{ + "hookEventName": "UserPromptSubmit", + "additionalContext": context, +}}}})) +(root / f"{{name}}.completed").write_text("done", encoding="utf-8") +"#, + ); + let hooks = serde_json::json!({ + "hooks": {"UserPromptSubmit": [{"hooks": [ + {"type": "command", "command": format!("python3 {} first", script_path.display()), "async": true}, + {"type": "command", "command": format!("python3 {} second", script_path.display()), "async": true}, + {"type": "command", "command": format!("python3 {} sync", script_path.display())}, + ]}]} + }); + fs::write(script_path, script).context("write async hook script")?; + fs::write(home.join("hooks.json"), hooks.to_string()).context("write hooks.json")?; + Ok(()) +} + +async fn submit_user_turn(codex: &CodexThread, prompt: &str) -> Result<()> { + codex + .submit(Op::UserInput { + environments: None, + items: vec![UserInput::Text { + text: 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?; + Ok(()) +} + +async fn hook_lifecycle_before_turn_complete(codex: &CodexThread) -> Result<(usize, usize)> { + let mut started = 0; + let mut completed = 0; + loop { + let event = timeout(Duration::from_secs(10), codex.next_event()) + .await + .context("timed out waiting for Codex event")??; + match event.msg { + EventMsg::HookStarted(_) => started += 1, + EventMsg::HookCompleted(_) => completed += 1, + EventMsg::TurnComplete(_) => return Ok((started, completed)), + _ => {} + } + } +} + +async fn wait_for_hook(home: &Path, name: &str) -> Result<()> { + let completed = home.join(format!("{name}.completed")); + timeout(Duration::from_secs(10), async { + while !completed.exists() { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .with_context(|| format!("timed out waiting for {name} async hook"))?; + sleep(Duration::from_millis(100)).await; + Ok(()) +} + +async fn release_hook(home: &Path, name: &str) -> Result<()> { + fs::write(home.join(format!("{name}.release")), "release")?; + wait_for_hook(home, name).await +} + +fn message_index(input: &[Value], role: &str, text: &str) -> Option { + input.iter().position(|item| { + item.get("role").and_then(Value::as_str) == Some(role) && item.to_string().contains(text) + }) +} + +#[tokio::test] +async fn async_command_hooks_deliver_ordered_output_on_the_next_user_turn() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let responses = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "first turn complete"), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-2", "async output received"), + ev_completed("resp-2"), + ]), + ], + ) + .await; + let test = test_codex() + .with_pre_build_hook(|home| write_gated_async_hooks(home).unwrap()) + .with_config(trust_discovered_hooks) + .build(&server) + .await?; + + submit_user_turn(&test.codex, FIRST_PROMPT).await?; + assert_eq!( + hook_lifecycle_before_turn_complete(&test.codex).await?, + (1, 1) + ); + assert_eq!(responses.requests().len(), 1); + wait_for_hook(test.codex_home_path(), "first").await?; + assert!(!test.codex_home_path().join("second.completed").exists()); + + release_hook(test.codex_home_path(), "second").await?; + assert_eq!( + responses.requests().len(), + 1, + "async completion must not sample while the session is idle", + ); + + submit_user_turn(&test.codex, SECOND_PROMPT).await?; + assert_eq!( + hook_lifecycle_before_turn_complete(&test.codex).await?, + (1, 1) + ); + + let requests = responses.requests(); + assert_eq!(requests.len(), 2); + assert!( + requests[0] + .message_input_texts("developer") + .iter() + .all(|text| !text.contains("")) + ); + let messages = requests[1] + .message_input_texts("developer") + .into_iter() + .filter(|text| text.contains("")) + .collect::>(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].matches("").context("async context")?; + assert!(user < developer); + Ok(()) +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 234fb8a2f0..f6b25c1c85 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -37,6 +37,8 @@ mod agents_md; mod apply_patch_cli; #[cfg(not(target_os = "windows"))] mod approvals; +#[cfg(not(target_os = "windows"))] +mod async_hooks; mod auto_review; mod cli_stream; mod client; diff --git a/codex-rs/hooks/Cargo.toml b/codex-rs/hooks/Cargo.toml index 1bd7f455f2..2e6ea59bba 100644 --- a/codex-rs/hooks/Cargo.toml +++ b/codex-rs/hooks/Cargo.toml @@ -25,7 +25,8 @@ regex = { workspace = true } schemars = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } -tokio = { workspace = true, features = ["fs", "io-util", "process", "time"] } +tokio = { workspace = true, features = ["fs", "io-util", "process", "rt", "time"] } +tokio-util = { workspace = true, features = ["rt"] } tracing = { workspace = true } uuid = { workspace = true, features = ["v4"] } diff --git a/codex-rs/hooks/src/engine/async_output.rs b/codex-rs/hooks/src/engine/async_output.rs new file mode 100644 index 0000000000..614b950cc3 --- /dev/null +++ b/codex-rs/hooks/src/engine/async_output.rs @@ -0,0 +1,204 @@ +use std::collections::VecDeque; +use std::future::Future; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::MutexGuard; + +use codex_protocol::protocol::HookEventName; +use codex_utils_output_truncation::TruncationPolicy; +use codex_utils_output_truncation::approx_token_count; +use codex_utils_output_truncation::formatted_truncate_text; +use codex_utils_output_truncation::truncate_text; +use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; + +use super::command_runner::CommandRunResult; +use super::output_parser; + +const ASYNC_HOOK_COMPLETION_TOKEN_LIMIT: usize = 500; +const ASYNC_HOOK_COMPLETION_TRUNCATION_TOKEN_LIMIT: usize = 450; +const ASYNC_HOOK_FLUSH_TOKEN_LIMIT: usize = 1_000; + +#[derive(Debug, Clone, PartialEq, Eq)] +struct AsyncHookCompletion { + event_name: HookEventName, + text: String, +} + +/// Session-scoped FIFO for informational output from detached command hooks. +#[derive(Clone)] +pub struct AsyncHookOutputQueue { + state: Arc, +} + +struct AsyncHookOutputState { + pending: Mutex>, + cancellation: CancellationToken, + tasks: TaskTracker, +} + +/// A bounded prefix prepared for one accepted user turn. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AsyncHookOutputBatch { + completion_count: usize, + text: String, +} + +impl AsyncHookOutputBatch { + /// Returns the merged developer context for this batch. + pub fn into_text(self) -> String { + self.text + } +} + +impl AsyncHookOutputQueue { + /// Records one completed hook's bounded informational output. + pub fn push(&self, event_name: HookEventName, text: String) { + if text.trim().is_empty() { + return; + } + let text = formatted_truncate_text( + &text, + TruncationPolicy::Tokens(ASYNC_HOOK_COMPLETION_TRUNCATION_TOKEN_LIMIT), + ); + let text = if approx_token_count(&text) > ASYNC_HOOK_COMPLETION_TOKEN_LIMIT { + truncate_text( + &text, + TruncationPolicy::Tokens(ASYNC_HOOK_COMPLETION_TOKEN_LIMIT), + ) + } else { + text + }; + self.lock_pending() + .push_back(AsyncHookCompletion { event_name, text }); + } + + /// Prepares a bounded FIFO prefix without removing it from the queue. + pub fn pending_batch(&self) -> Option { + let pending = self.lock_pending(); + let mut selected = Vec::new(); + for completion in pending.iter() { + let mut candidate = selected.clone(); + candidate.push(completion.clone()); + let text = render_batch(&candidate); + if approx_token_count(&text) > ASYNC_HOOK_FLUSH_TOKEN_LIMIT { + break; + } + selected = candidate; + } + (!selected.is_empty()).then(|| AsyncHookOutputBatch { + completion_count: selected.len(), + text: render_batch(&selected), + }) + } + + /// Removes a previously prepared prefix after its user input is accepted. + pub fn commit(&self, batch: &AsyncHookOutputBatch) { + self.lock_pending().drain(..batch.completion_count); + } + + /// Cancels detached hook commands and waits for their tasks to stop. + pub async fn shutdown(&self) { + self.state.cancellation.cancel(); + self.state.tasks.close(); + self.state.tasks.wait().await; + } + + pub(crate) fn cancellation_token(&self) -> CancellationToken { + self.state.cancellation.clone() + } + + pub(crate) fn spawn(&self, future: impl Future + Send + 'static) { + self.state.tasks.spawn(future); + } + + fn lock_pending(&self) -> MutexGuard<'_, VecDeque> { + self.state + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +impl Default for AsyncHookOutputQueue { + fn default() -> Self { + Self { + state: Arc::new(AsyncHookOutputState { + pending: Mutex::new(VecDeque::new()), + cancellation: CancellationToken::new(), + tasks: TaskTracker::new(), + }), + } + } +} + +fn render_batch(completions: &[AsyncHookCompletion]) -> String { + let outputs = completions + .iter() + .map(|completion| { + format!( + "\n{}\n", + completion.event_name, completion.text + ) + }) + .collect::>() + .join("\n"); + format!("\n{outputs}\n") +} + +pub(crate) fn deliverable_output( + event_name: HookEventName, + run_result: &CommandRunResult, +) -> Option { + if let Some(error) = run_result.error.as_deref() { + return Some(format!("Async hook failed to run: {error}")); + } + + match run_result.exit_code { + Some(0) => parse_stdout(event_name, &run_result.stdout), + Some(exit_code) => Some(format!("Async hook exited with code {exit_code}")), + None => Some("Async hook process terminated without an exit code".to_string()), + } +} + +fn parse_stdout(event_name: HookEventName, stdout: &str) -> Option { + let trimmed = stdout.trim(); + if trimmed.is_empty() { + return None; + } + + let additional_context = match event_name { + HookEventName::SessionStart => { + output_parser::parse_session_start(trimmed).map(|output| output.additional_context) + } + HookEventName::SubagentStart => { + output_parser::parse_subagent_start(trimmed).map(|output| output.additional_context) + } + HookEventName::PreToolUse => { + output_parser::parse_pre_tool_use(trimmed).map(|output| output.additional_context) + } + HookEventName::PostToolUse => { + output_parser::parse_post_tool_use(trimmed).map(|output| output.additional_context) + } + HookEventName::UserPromptSubmit => { + output_parser::parse_user_prompt_submit(trimmed).map(|output| output.additional_context) + } + HookEventName::PermissionRequest => { + output_parser::parse_permission_request(trimmed).map(|_| None) + } + HookEventName::PreCompact => output_parser::parse_pre_compact(trimmed).map(|_| None), + HookEventName::PostCompact => output_parser::parse_post_compact(trimmed).map(|_| None), + HookEventName::SubagentStop => output_parser::parse_subagent_stop(trimmed).map(|_| None), + HookEventName::Stop => output_parser::parse_stop(trimmed).map(|_| None), + }; + match additional_context { + Some(additional_context) => additional_context.filter(|context| !context.trim().is_empty()), + None => Some(format!( + "Async {event_name:?} hook returned invalid JSON output" + )), + } +} + +#[cfg(test)] +#[path = "async_output_tests.rs"] +mod tests; diff --git a/codex-rs/hooks/src/engine/async_output_tests.rs b/codex-rs/hooks/src/engine/async_output_tests.rs new file mode 100644 index 0000000000..e11e0d6aeb --- /dev/null +++ b/codex-rs/hooks/src/engine/async_output_tests.rs @@ -0,0 +1,326 @@ +use std::fs; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; +use std::time::Duration; + +use codex_protocol::ThreadId; +use codex_protocol::protocol::HookEventName; +use codex_protocol::protocol::HookSource; +use codex_utils_absolute_path::AbsolutePathBuf; +use codex_utils_output_truncation::approx_token_count; +use pretty_assertions::assert_eq; +use tempfile::tempdir; +use tokio::time::sleep; +use tokio::time::timeout; + +use super::ASYNC_HOOK_COMPLETION_TOKEN_LIMIT; +use super::ASYNC_HOOK_FLUSH_TOKEN_LIMIT; +use super::AsyncHookOutputQueue; +use super::deliverable_output; +use crate::engine::CommandShell; +use crate::engine::ConfiguredHandler; +use crate::engine::command_runner::CommandRunResult; +use crate::events::permission_request; +use crate::events::post_tool_use; +use crate::events::pre_tool_use; + +#[test] +fn async_output_delivers_only_event_context() { + let pre = output( + HookEventName::PreToolUse, + r#"{"continue":false,"systemMessage":"ignore","decision":"block","reason":"ignore","hookSpecificOutput":{"hookEventName":"PreToolUse","permissionDecision":"deny","updatedInput":{"command":"rewrite"},"additionalContext":"pre context"}}"#, + ); + let permission = output( + HookEventName::PermissionRequest, + r#"{"continue":false,"hookSpecificOutput":{"hookEventName":"PermissionRequest","decision":{"behavior":"deny","message":"ignore"}}}"#, + ); + let post = output( + HookEventName::PostToolUse, + r#"{"continue":false,"decision":"block","reason":"ignore","hookSpecificOutput":{"hookEventName":"PostToolUse","additionalContext":"post context"}}"#, + ); + + assert_eq!( + (pre.as_deref(), permission, post.as_deref()), + (Some("pre context"), None, Some("post context")) + ); +} + +#[test] +fn async_output_surfaces_parse_and_runtime_failures() { + let invalid = deliverable_output( + HookEventName::PreToolUse, + &result(Some(0), "plain stdout", "", /*error*/ None), + ); + let nonzero = deliverable_output( + HookEventName::PreToolUse, + &result(Some(2), "", "denied", /*error*/ None), + ); + let runtime = deliverable_output( + HookEventName::PreToolUse, + &result( + /*exit_code*/ None, + "", + "", + Some("spawn failed".to_string()), + ), + ); + + assert_eq!( + (invalid, nonzero, runtime), + ( + Some("Async PreToolUse hook returned invalid JSON output".to_string()), + Some("Async hook exited with code 2".to_string()), + Some("Async hook failed to run: spawn failed".to_string()), + ) + ); +} + +#[cfg(not(target_os = "windows"))] +#[tokio::test] +async fn async_tool_hooks_do_not_return_control_or_lifecycle() { + let temp = tempdir().expect("create temp dir"); + let script_path = temp.path().join("async_controls.py"); + fs::write( + &script_path, + r#"import json +from pathlib import Path +import sys + +name = sys.argv[1] +outputs = { + "pre_block": {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": "blocked", "additionalContext": "pre block context"}}, + "pre_rewrite": {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "allow", "updatedInput": {"command": "rewrite"}, "additionalContext": "pre rewrite context"}}, + "permission": {"hookSpecificOutput": {"hookEventName": "PermissionRequest", "decision": {"behavior": "deny", "message": "denied"}}}, + "post": {"continue": False, "decision": "block", "reason": "stop", "hookSpecificOutput": {"hookEventName": "PostToolUse", "additionalContext": "post context"}}, +} +json.load(sys.stdin) +print(json.dumps(outputs[name])) +(Path(__file__).parent / f"{name}.completed").write_text("done", encoding="utf-8") +"#, + ) + .expect("write async controls script"); + let cwd = AbsolutePathBuf::try_from(temp.path().to_path_buf()).expect("absolute temp dir"); + let queue = AsyncHookOutputQueue::default(); + let shell = CommandShell { + program: String::new(), + args: Vec::new(), + }; + let pre_request = pre_tool_use::PreToolUseRequest { + session_id: ThreadId::new(), + turn_id: "turn-1".to_string(), + subagent: None, + cwd: cwd.clone(), + transcript_path: None, + model: "gpt-test".to_string(), + permission_mode: "default".to_string(), + tool_name: "Bash".to_string(), + matcher_aliases: Vec::new(), + tool_use_id: "tool-1".to_string(), + tool_input: serde_json::json!({"command": "original"}), + }; + let pre_handlers = [ + async_handler(HookEventName::PreToolUse, &script_path, "pre_block"), + async_handler(HookEventName::PreToolUse, &script_path, "pre_rewrite"), + ]; + assert!(pre_tool_use::preview(&pre_handlers, &pre_request).is_empty()); + let pre = pre_tool_use::run(&pre_handlers, &shell, &queue, pre_request).await; + assert_eq!( + ( + pre.hook_events, + pre.should_block, + pre.block_reason, + pre.additional_contexts, + pre.updated_input, + ), + (Vec::new(), false, None, Vec::new(), None) + ); + + let permission_request = permission_request::PermissionRequestRequest { + session_id: ThreadId::new(), + turn_id: "turn-1".to_string(), + subagent: None, + cwd: temp.path().to_path_buf(), + transcript_path: None, + model: "gpt-test".to_string(), + permission_mode: "default".to_string(), + tool_name: "Bash".to_string(), + matcher_aliases: Vec::new(), + run_id_suffix: "tool-1".to_string(), + tool_input: serde_json::json!({"command": "original"}), + }; + let permission_handlers = [async_handler( + HookEventName::PermissionRequest, + &script_path, + "permission", + )]; + assert!(permission_request::preview(&permission_handlers, &permission_request).is_empty()); + let permission = + permission_request::run(&permission_handlers, &shell, &queue, permission_request).await; + assert_eq!( + (permission.hook_events, permission.decision), + (Vec::new(), None) + ); + + let post_request = post_tool_use::PostToolUseRequest { + session_id: ThreadId::new(), + turn_id: "turn-1".to_string(), + subagent: None, + cwd, + transcript_path: None, + model: "gpt-test".to_string(), + permission_mode: "default".to_string(), + tool_name: "Bash".to_string(), + matcher_aliases: Vec::new(), + tool_use_id: "tool-1".to_string(), + tool_input: serde_json::json!({"command": "original"}), + tool_response: serde_json::json!({"ok": true}), + }; + let post_handlers = [async_handler( + HookEventName::PostToolUse, + &script_path, + "post", + )]; + assert!(post_tool_use::preview(&post_handlers, &post_request).is_empty()); + let post = post_tool_use::run(&post_handlers, &shell, &queue, post_request).await; + assert_eq!( + ( + post.hook_events, + post.should_stop, + post.stop_reason, + post.additional_contexts, + post.feedback_message, + ), + (Vec::new(), false, None, Vec::new(), None) + ); + + timeout(Duration::from_secs(10), async { + for name in ["pre_block", "pre_rewrite", "permission", "post"] { + while !temp.path().join(format!("{name}.completed")).exists() { + sleep(Duration::from_millis(10)).await; + } + } + }) + .await + .expect("async control hooks complete"); + let delivered = queue + .pending_batch() + .expect("informational async outputs") + .into_text(); + assert!(delivered.contains("pre block context")); + assert!(delivered.contains("pre rewrite context")); + assert!(delivered.contains("post context")); + assert!(!delivered.contains("denied")); + queue.shutdown().await; +} + +#[test] +fn queue_preserves_duplicate_completions_until_commit() { + let queue = AsyncHookOutputQueue::default(); + queue.push(HookEventName::PreToolUse, "same".to_string()); + queue.push(HookEventName::PostToolUse, "same".to_string()); + queue.push(HookEventName::Stop, "last".to_string()); + + let batch = queue.pending_batch().expect("queued output batch"); + assert_eq!(queue.pending_batch(), Some(batch.clone())); + let text = batch.clone().into_text(); + assert_eq!(text.matches("same").count(), 2); + assert!( + ["PreToolUse", "PostToolUse", "last"] + .map(|needle| text.find(needle).expect("queued output")) + .is_sorted() + ); + queue.commit(&batch); + assert!(queue.pending_batch().is_none()); +} + +#[test] +fn queue_bounds_items_and_flushes_a_contiguous_prefix() { + let queue = AsyncHookOutputQueue::default(); + let large = "large-output ".repeat(ASYNC_HOOK_COMPLETION_TOKEN_LIMIT); + for event_name in [ + HookEventName::PreToolUse, + HookEventName::PostToolUse, + HookEventName::Stop, + ] { + queue.push(event_name, large.clone()); + } + queue.push(HookEventName::SessionStart, "small-tail".to_string()); + assert!(queue.lock_pending().iter().all(|completion| { + approx_token_count(&completion.text) <= ASYNC_HOOK_COMPLETION_TOKEN_LIMIT + })); + + let mut completed = 0; + loop { + let batch = queue.pending_batch().expect("bounded output batch"); + let text = batch.clone().into_text(); + assert!(approx_token_count(&text) <= ASYNC_HOOK_FLUSH_TOKEN_LIMIT); + completed += batch.completion_count; + queue.commit(&batch); + if let Some(tail) = text.find("small-tail") { + if let Some(large) = text.find("tokens truncated") { + assert!(large < tail); + } + assert_eq!(completed, 4); + break; + } + } + assert!(queue.pending_batch().is_none()); +} + +#[tokio::test] +async fn shutdown_cancels_and_joins_detached_tasks() { + let queue = AsyncHookOutputQueue::default(); + let cancellation = queue.cancellation_token(); + let stopped = Arc::new(AtomicBool::new(false)); + let task_stopped = Arc::clone(&stopped); + queue.spawn(async move { + cancellation.cancelled().await; + task_stopped.store(true, Ordering::SeqCst); + }); + + queue.shutdown().await; + + assert!(stopped.load(Ordering::SeqCst)); +} + +fn output(event_name: HookEventName, stdout: &str) -> Option { + deliverable_output(event_name, &result(Some(0), stdout, "", /*error*/ None)) +} + +#[cfg(not(target_os = "windows"))] +fn async_handler( + event_name: HookEventName, + script_path: &std::path::Path, + name: &str, +) -> ConfiguredHandler { + ConfiguredHandler { + event_name, + matcher: Some("^Bash$".to_string()), + command: format!("python3 {} {name}", script_path.display()), + timeout_sec: 5, + r#async: true, + status_message: None, + source_path: AbsolutePathBuf::try_from(script_path.to_path_buf()).expect("absolute script"), + source: HookSource::User, + display_order: 0, + env: std::collections::HashMap::new(), + } +} + +fn result( + exit_code: Option, + stdout: &str, + stderr: &str, + error: Option, +) -> CommandRunResult { + CommandRunResult { + started_at: 1, + completed_at: 2, + duration_ms: 1, + exit_code, + stdout: stdout.to_string(), + stderr: stderr.to_string(), + error, + } +} diff --git a/codex-rs/hooks/src/engine/discovery.rs b/codex-rs/hooks/src/engine/discovery.rs index 92d17fbabd..ac59038f2f 100644 --- a/codex-rs/hooks/src/engine/discovery.rs +++ b/codex-rs/hooks/src/engine/discovery.rs @@ -465,13 +465,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 {}", @@ -529,6 +522,7 @@ fn append_matcher_groups( matcher: matcher.map(ToOwned::to_owned), command, timeout_sec, + r#async, status_message, source_path: source.path.clone(), source: source.source, @@ -780,6 +774,7 @@ mod tests { matcher: None, command: "echo hello".to_string(), timeout_sec: 600, + r#async: false, status_message: None, source_path: source_path.clone(), source: hook_source(), @@ -790,8 +785,9 @@ mod tests { } #[test] - fn pre_tool_use_keeps_valid_matcher_during_discovery() { + fn async_pre_tool_use_is_discovered_without_warning() { let mut handlers = Vec::new(); + let mut hook_entries = Vec::new(); let mut warnings = Vec::new(); let mut display_order = 0; let source_path = source_path(); @@ -799,12 +795,21 @@ mod tests { append_matcher_groups( &mut handlers, - &mut Vec::new(), + &mut hook_entries, &mut warnings, &mut display_order, &hook_handler_source(&source_path, &hook_states), HookEventName::PreToolUse, - vec![command_group(Some("^Bash$"))], + vec![MatcherGroup { + matcher: Some("^Bash$".to_string()), + hooks: vec![HookHandlerConfig::Command { + command: "echo hello".to_string(), + command_windows: None, + timeout_sec: None, + r#async: true, + status_message: None, + }], + }], ); assert_eq!(warnings, Vec::::new()); @@ -815,6 +820,7 @@ mod tests { matcher: Some("^Bash$".to_string()), command: "echo hello".to_string(), timeout_sec: 600, + r#async: true, status_message: None, source_path: source_path.clone(), source: hook_source(), @@ -822,6 +828,7 @@ mod tests { env: std::collections::HashMap::new(), }] ); + assert_eq!(hook_entries.len(), 1); } #[test] diff --git a/codex-rs/hooks/src/engine/dispatcher.rs b/codex-rs/hooks/src/engine/dispatcher.rs index 50822bfc96..2c5d2d4a66 100644 --- a/codex-rs/hooks/src/engine/dispatcher.rs +++ b/codex-rs/hooks/src/engine/dispatcher.rs @@ -1,4 +1,5 @@ use std::path::Path; +use std::path::PathBuf; use futures::StreamExt; use futures::stream::FuturesUnordered; @@ -13,6 +14,8 @@ use codex_protocol::protocol::HookScope; use super::CommandShell; use super::ConfiguredHandler; +use super::async_output::AsyncHookOutputQueue; +use super::async_output::deliverable_output; use super::command_runner::CommandRunResult; use super::command_runner::run_command; use crate::events::common::matches_matcher; @@ -33,6 +36,17 @@ pub(crate) fn select_handlers( select_handlers_for_matcher_inputs(handlers, event_name, &matcher_inputs) } +pub(crate) fn select_sync_handlers( + handlers: &[ConfiguredHandler], + event_name: HookEventName, + matcher_input: Option<&str>, +) -> Vec { + select_handlers(handlers, event_name, matcher_input) + .into_iter() + .filter(|handler| !handler.r#async) + .collect() +} + pub(crate) fn select_handlers_for_matcher_inputs( handlers: &[ConfiguredHandler], event_name: HookEventName, @@ -67,6 +81,80 @@ pub(crate) fn select_handlers_for_matcher_inputs( .collect() } +pub(crate) fn select_sync_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.r#async) + .collect() +} + +fn partition_handlers( + handlers: Vec, +) -> (Vec, Vec) { + handlers.into_iter().partition(|handler| !handler.r#async) +} + +fn spawn_async_handlers( + shell: &CommandShell, + output_queue: &AsyncHookOutputQueue, + handlers: Vec, + input_json: String, + cwd: &Path, +) { + for handler in handlers { + let shell = shell.clone(); + let output_queue = output_queue.clone(); + let cancellation = output_queue.cancellation_token(); + let input_json = input_json.clone(); + let cwd = PathBuf::from(cwd); + output_queue.clone().spawn(async move { + tokio::select! { + _ = cancellation.cancelled() => {} + run_result = run_command(&shell, &handler, &input_json, &cwd) => { + if let Some(output) = deliverable_output(handler.event_name, &run_result) { + output_queue.push(handler.event_name, output); + } + } + } + }); + } +} + +fn spawn_async_failures( + output_queue: &AsyncHookOutputQueue, + handlers: Vec, + error: String, +) { + for handler in handlers { + let output_queue = output_queue.clone(); + let cancellation = output_queue.cancellation_token(); + let error = error.clone(); + output_queue.clone().spawn(async move { + if !cancellation.is_cancelled() { + output_queue.push(handler.event_name, error); + } + }); + } +} + +pub(crate) fn synchronous_handlers_after_serialization_failure( + output_queue: &AsyncHookOutputQueue, + handlers: Vec, + error: &str, +) -> Vec { + let (synchronous, asynchronous) = partition_handlers(handlers); + spawn_async_failures( + output_queue, + asynchronous, + format!("Async hook input serialization failed: {error}"), + ); + synchronous +} + pub(crate) fn running_summary(handler: &ConfiguredHandler) -> HookRunSummary { HookRunSummary { id: handler.run_id(), @@ -88,12 +176,16 @@ pub(crate) fn running_summary(handler: &ConfiguredHandler) -> HookRunSummary { pub(crate) async fn execute_handlers( shell: &CommandShell, + output_queue: &AsyncHookOutputQueue, handlers: Vec, input_json: String, cwd: &Path, turn_id: Option, parse: fn(&ConfiguredHandler, CommandRunResult, Option) -> ParsedHandler, ) -> Vec> { + let (handlers, asynchronous) = partition_handlers(handlers); + spawn_async_handlers(shell, output_queue, asynchronous, input_json.clone(), cwd); + let mut pending = FuturesUnordered::new(); for (configured_order, handler) in handlers.into_iter().enumerate() { let input_json = input_json.clone(); @@ -175,6 +267,7 @@ mod tests { matcher: matcher.map(str::to_owned), command: command.to_string(), timeout_sec: 5, + r#async: false, status_message: None, source_path: test_path_buf("/tmp/hooks.json").abs(), source: HookSource::User, @@ -416,7 +509,7 @@ mod tests { #[test] fn select_handlers_preserves_declaration_order() { - let handlers = vec![ + let mut handlers = vec![ make_handler( HookEventName::Stop, /*matcher*/ None, @@ -436,12 +529,35 @@ mod tests { /*display_order*/ 2, ), ]; + handlers[1].r#async = true; let selected = select_handlers(&handlers, HookEventName::Stop, /*matcher_input*/ None); + let synchronous = super::select_sync_handlers( + &handlers, + HookEventName::Stop, + /*matcher_input*/ None, + ); + let (_, asynchronous) = super::partition_handlers(selected.clone()); assert_eq!(selected.len(), 3); assert_eq!(selected[0].command, "first"); assert_eq!(selected[1].command, "second"); assert_eq!(selected[2].command, "third"); + assert_eq!( + ( + synchronous + .into_iter() + .map(|handler| handler.command) + .collect::>(), + asynchronous + .into_iter() + .map(|handler| handler.command) + .collect::>(), + ), + ( + vec!["first".to_string(), "third".to_string()], + vec!["second".to_string()], + ) + ); } } diff --git a/codex-rs/hooks/src/engine/mod.rs b/codex-rs/hooks/src/engine/mod.rs index 859fc54069..a90d84f3f3 100644 --- a/codex-rs/hooks/src/engine/mod.rs +++ b/codex-rs/hooks/src/engine/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod async_output; pub(crate) mod command_runner; pub(crate) mod discovery; pub(crate) mod dispatcher; @@ -32,6 +33,9 @@ use codex_protocol::protocol::HookTrustStatus; use codex_utils_absolute_path::AbsolutePathBuf; use std::collections::HashMap; +pub use async_output::AsyncHookOutputBatch; +pub use async_output::AsyncHookOutputQueue; + #[derive(Debug, Clone)] pub(crate) struct CommandShell { pub program: String, @@ -44,6 +48,7 @@ pub(crate) struct ConfiguredHandler { pub matcher: Option, pub command: String, pub timeout_sec: u64, + pub r#async: bool, pub status_message: Option, pub source_path: AbsolutePathBuf, pub source: HookSource, @@ -101,10 +106,12 @@ pub(crate) struct ClaudeHooksEngine { handlers: Vec, warnings: Vec, shell: CommandShell, + async_output_queue: AsyncHookOutputQueue, output_spiller: HookOutputSpiller, } impl ClaudeHooksEngine { + #[cfg(test)] pub(crate) fn new( enabled: bool, bypass_hook_trust: bool, @@ -112,12 +119,33 @@ impl ClaudeHooksEngine { plugin_hook_sources: Vec, plugin_hook_load_warnings: Vec, shell: CommandShell, + ) -> Self { + Self::new_with_async_output_queue( + enabled, + bypass_hook_trust, + config_layer_stack, + plugin_hook_sources, + plugin_hook_load_warnings, + shell, + AsyncHookOutputQueue::default(), + ) + } + + pub(crate) fn new_with_async_output_queue( + enabled: bool, + bypass_hook_trust: bool, + config_layer_stack: Option<&ConfigLayerStack>, + plugin_hook_sources: Vec, + plugin_hook_load_warnings: Vec, + shell: CommandShell, + async_output_queue: AsyncHookOutputQueue, ) -> Self { if !enabled { return Self { handlers: Vec::new(), warnings: Vec::new(), shell, + async_output_queue, output_spiller: HookOutputSpiller::new(), }; } @@ -133,6 +161,7 @@ impl ClaudeHooksEngine { handlers: discovered.handlers, warnings: discovered.warnings, shell, + async_output_queue, output_spiller: HookOutputSpiller::new(), } } @@ -172,8 +201,14 @@ impl ClaudeHooksEngine { turn_id: Option, ) -> SessionStartOutcome { let session_id = request.session_id; - let mut outcome = - crate::events::session_start::run(&self.handlers, &self.shell, request, turn_id).await; + let mut outcome = crate::events::session_start::run( + &self.handlers, + &self.shell, + &self.async_output_queue, + request, + turn_id, + ) + .await; outcome.additional_contexts = self .maybe_spill_texts(session_id, outcome.additional_contexts) .await; @@ -182,8 +217,13 @@ 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; + let mut outcome = crate::events::pre_tool_use::run( + &self.handlers, + &self.shell, + &self.async_output_queue, + request, + ) + .await; outcome.additional_contexts = self .maybe_spill_texts(session_id, outcome.additional_contexts) .await; @@ -194,7 +234,13 @@ 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.shell, + &self.async_output_queue, + request, + ) + .await } pub(crate) async fn run_post_tool_use( @@ -202,8 +248,13 @@ impl ClaudeHooksEngine { request: PostToolUseRequest, ) -> PostToolUseOutcome { let session_id = request.session_id; - let mut outcome = - crate::events::post_tool_use::run(&self.handlers, &self.shell, request).await; + let mut outcome = crate::events::post_tool_use::run( + &self.handlers, + &self.shell, + &self.async_output_queue, + request, + ) + .await; outcome.additional_contexts = self .maybe_spill_texts(session_id, outcome.additional_contexts) .await; @@ -218,7 +269,13 @@ 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.shell, + &self.async_output_queue, + request, + ) + .await } pub(crate) fn preview_post_compact(&self, request: &PostCompactRequest) -> Vec { @@ -229,7 +286,13 @@ 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.shell, + &self.async_output_queue, + request, + ) + .await } pub(crate) fn preview_user_prompt_submit( @@ -244,8 +307,13 @@ impl ClaudeHooksEngine { request: UserPromptSubmitRequest, ) -> UserPromptSubmitOutcome { let session_id = request.session_id; - let mut outcome = - crate::events::user_prompt_submit::run(&self.handlers, &self.shell, request).await; + let mut outcome = crate::events::user_prompt_submit::run( + &self.handlers, + &self.shell, + &self.async_output_queue, + request, + ) + .await; outcome.additional_contexts = self .maybe_spill_texts(session_id, outcome.additional_contexts) .await; @@ -258,7 +326,13 @@ 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.shell, + &self.async_output_queue, + request, + ) + .await; outcome.continuation_fragments = self .maybe_spill_prompt_fragments(session_id, outcome.continuation_fragments) .await; diff --git a/codex-rs/hooks/src/engine/mod_tests.rs b/codex-rs/hooks/src/engine/mod_tests.rs index 4b7697b34f..40c66cd40c 100644 --- a/codex-rs/hooks/src/engine/mod_tests.rs +++ b/codex-rs/hooks/src/engine/mod_tests.rs @@ -221,6 +221,7 @@ with Path(r"{log_path}").open("a", encoding="utf-8") as handle: plugin_hook_load_warnings: Vec::new(), shell_program: None, shell_args: Vec::new(), + async_output_queue: Default::default(), }); assert!(listed.hooks[0].is_managed); let cwd = cwd(); @@ -1219,6 +1220,7 @@ print(json.dumps({ plugin_hook_load_warnings: Vec::new(), shell_program: None, shell_args: Vec::new(), + async_output_queue: Default::default(), }); assert_eq!( listed.hooks[0].plugin_id.as_deref(), diff --git a/codex-rs/hooks/src/events/compact.rs b/codex-rs/hooks/src/events/compact.rs index cb3080219a..e4a949ff45 100644 --- a/codex-rs/hooks/src/events/compact.rs +++ b/codex-rs/hooks/src/events/compact.rs @@ -12,6 +12,7 @@ use codex_utils_absolute_path::AbsolutePathBuf; use super::common; use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; +use crate::engine::async_output::AsyncHookOutputQueue; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; use crate::engine::output_parser; @@ -59,7 +60,7 @@ pub(crate) fn preview_pre( handlers: &[ConfiguredHandler], request: &PreCompactRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::select_sync_handlers( handlers, HookEventName::PreCompact, Some(request.trigger.as_str()), @@ -72,6 +73,7 @@ pub(crate) fn preview_pre( pub(crate) async fn run_pre( handlers: &[ConfiguredHandler], shell: &CommandShell, + async_output_queue: &AsyncHookOutputQueue, request: PreCompactRequest, ) -> PreCompactOutcome { let matched = dispatcher::select_handlers( @@ -90,11 +92,17 @@ pub(crate) async fn run_pre( let input_json = match pre_command_input_json(&request) { Ok(input_json) => input_json, Err(error) => { + let error_message = format!("failed to serialize pre compact hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); return PreCompactOutcome { hook_events: common::serialization_failure_hook_events( matched, Some(request.turn_id), - format!("failed to serialize pre compact hook input: {error}"), + error_message, ), should_stop: false, stop_reason: None, @@ -104,6 +112,7 @@ pub(crate) async fn run_pre( let results = dispatcher::execute_handlers( shell, + async_output_queue, matched, input_json, request.cwd.as_path(), @@ -141,7 +150,7 @@ pub(crate) fn preview_post( handlers: &[ConfiguredHandler], request: &PostCompactRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::select_sync_handlers( handlers, HookEventName::PostCompact, Some(request.trigger.as_str()), @@ -154,6 +163,7 @@ pub(crate) fn preview_post( pub(crate) async fn run_post( handlers: &[ConfiguredHandler], shell: &CommandShell, + async_output_queue: &AsyncHookOutputQueue, request: PostCompactRequest, ) -> StatelessHookOutcome { let matched = dispatcher::select_handlers( @@ -172,11 +182,17 @@ pub(crate) async fn run_post( let input_json = match post_command_input_json(&request) { Ok(input_json) => input_json, Err(error) => { + let error_message = format!("failed to serialize post compact hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); return StatelessHookOutcome { hook_events: common::serialization_failure_hook_events( matched, Some(request.turn_id), - format!("failed to serialize post compact hook input: {error}"), + error_message, ), should_stop: false, stop_reason: None, @@ -186,6 +202,7 @@ pub(crate) async fn run_post( let results = dispatcher::execute_handlers( shell, + async_output_queue, matched, input_json, request.cwd.as_path(), @@ -599,6 +616,7 @@ mod tests { matcher: None, command: "python3 compact_hook.py".to_string(), timeout_sec: 5, + r#async: false, status_message: Some("running compact hook".to_string()), source_path: test_path_buf("/tmp/hooks.json").abs(), source: codex_protocol::protocol::HookSource::User, diff --git a/codex-rs/hooks/src/events/permission_request.rs b/codex-rs/hooks/src/events/permission_request.rs index db7970f02d..7cce1ff711 100644 --- a/codex-rs/hooks/src/events/permission_request.rs +++ b/codex-rs/hooks/src/events/permission_request.rs @@ -18,6 +18,7 @@ use std::path::PathBuf; use super::common; use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; +use crate::engine::async_output::AsyncHookOutputQueue; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; use crate::engine::output_parser; @@ -69,7 +70,7 @@ 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::select_sync_handlers_for_matcher_inputs( handlers, HookEventName::PermissionRequest, &matcher_inputs, @@ -87,6 +88,7 @@ pub(crate) fn preview( pub(crate) async fn run( handlers: &[ConfiguredHandler], shell: &CommandShell, + async_output_queue: &AsyncHookOutputQueue, request: PermissionRequestRequest, ) -> PermissionRequestOutcome { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); @@ -105,10 +107,17 @@ pub(crate) async fn run( let input_json = match serde_json::to_string(&build_command_input(&request)) { Ok(input_json) => input_json, Err(error) => { + let error_message = + format!("failed to serialize permission request hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); let hook_events = common::serialization_failure_hook_events_for_tool_use( matched, Some(request.turn_id.clone()), - format!("failed to serialize permission request hook input: {error}"), + error_message, &request.run_id_suffix, ); return PermissionRequestOutcome { @@ -120,6 +129,7 @@ pub(crate) async fn run( let results = dispatcher::execute_handlers( shell, + async_output_queue, matched, input_json, request.cwd.as_path(), diff --git a/codex-rs/hooks/src/events/post_tool_use.rs b/codex-rs/hooks/src/events/post_tool_use.rs index f096de0110..1c01a8abe4 100644 --- a/codex-rs/hooks/src/events/post_tool_use.rs +++ b/codex-rs/hooks/src/events/post_tool_use.rs @@ -13,6 +13,7 @@ use serde_json::Value; use super::common; use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; +use crate::engine::async_output::AsyncHookOutputQueue; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; use crate::engine::output_parser; @@ -57,7 +58,7 @@ 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::select_sync_handlers_for_matcher_inputs( handlers, HookEventName::PostToolUse, &matcher_inputs, @@ -72,6 +73,7 @@ pub(crate) fn preview( pub(crate) async fn run( handlers: &[ConfiguredHandler], shell: &CommandShell, + async_output_queue: &AsyncHookOutputQueue, request: PostToolUseRequest, ) -> PostToolUseOutcome { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); @@ -93,10 +95,16 @@ pub(crate) async fn run( let input_json = match command_input_json(&request) { Ok(input_json) => input_json, Err(error) => { + let error_message = format!("failed to serialize post tool use hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); let hook_events = common::serialization_failure_hook_events_for_tool_use( matched, Some(request.turn_id.clone()), - format!("failed to serialize post tool use hook input: {error}"), + error_message, &request.tool_use_id, ); return serialization_failure_outcome(hook_events); @@ -105,6 +113,7 @@ pub(crate) async fn run( let results = dispatcher::execute_handlers( shell, + async_output_queue, matched, input_json, request.cwd.as_path(), @@ -552,6 +561,7 @@ mod tests { matcher: Some("^Bash$".to_string()), command: "python3 post_tool_use_hook.py".to_string(), timeout_sec: 5, + r#async: false, status_message: Some("running post tool use hook".to_string()), source_path: test_path_buf("/tmp/hooks.json").abs(), source: codex_protocol::protocol::HookSource::User, diff --git a/codex-rs/hooks/src/events/pre_tool_use.rs b/codex-rs/hooks/src/events/pre_tool_use.rs index b3579aba82..7162143346 100644 --- a/codex-rs/hooks/src/events/pre_tool_use.rs +++ b/codex-rs/hooks/src/events/pre_tool_use.rs @@ -13,6 +13,7 @@ use serde_json::Value; use super::common; use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; +use crate::engine::async_output::AsyncHookOutputQueue; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; use crate::engine::output_parser; @@ -56,7 +57,7 @@ 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::select_sync_handlers_for_matcher_inputs( handlers, HookEventName::PreToolUse, &matcher_inputs, @@ -71,6 +72,7 @@ pub(crate) fn preview( pub(crate) async fn run( handlers: &[ConfiguredHandler], shell: &CommandShell, + async_output_queue: &AsyncHookOutputQueue, request: PreToolUseRequest, ) -> PreToolUseOutcome { let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); @@ -92,10 +94,16 @@ pub(crate) async fn run( let input_json = match command_input_json(&request) { Ok(input_json) => input_json, Err(error) => { + let error_message = format!("failed to serialize pre tool use hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); let hook_events = common::serialization_failure_hook_events_for_tool_use( matched, Some(request.turn_id.clone()), - format!("failed to serialize pre tool use hook input: {error}"), + error_message, &request.tool_use_id, ); return serialization_failure_outcome(hook_events); @@ -104,6 +112,7 @@ pub(crate) async fn run( let results = dispatcher::execute_handlers( shell, + async_output_queue, matched, input_json, request.cwd.as_path(), @@ -744,6 +753,7 @@ mod tests { matcher: Some("^Bash$".to_string()), command: "echo hook".to_string(), timeout_sec: 5, + r#async: false, status_message: None, source_path: test_path_buf("/tmp/hooks.json").abs(), source: codex_protocol::protocol::HookSource::User, diff --git a/codex-rs/hooks/src/events/session_start.rs b/codex-rs/hooks/src/events/session_start.rs index bd1aa2096f..b05403e27d 100644 --- a/codex-rs/hooks/src/events/session_start.rs +++ b/codex-rs/hooks/src/events/session_start.rs @@ -12,6 +12,7 @@ use codex_utils_absolute_path::AbsolutePathBuf; use super::common; use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; +use crate::engine::async_output::AsyncHookOutputQueue; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; use crate::engine::output_parser; @@ -95,7 +96,7 @@ pub(crate) fn preview( handlers: &[ConfiguredHandler], request: &SessionStartRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::select_sync_handlers( handlers, request.target.event_name(), Some(request.target.matcher_input()), @@ -108,6 +109,7 @@ pub(crate) fn preview( pub(crate) async fn run( handlers: &[ConfiguredHandler], shell: &CommandShell, + async_output_queue: &AsyncHookOutputQueue, request: SessionStartRequest, turn_id: Option, ) -> SessionStartOutcome { @@ -137,12 +139,15 @@ pub(crate) async fn run( )) { Ok(input_json) => input_json, Err(error) => { + let error_message = + format!("failed to serialize session start hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); return serialization_failure_outcome( - common::serialization_failure_hook_events( - matched, - turn_id, - format!("failed to serialize session start hook input: {error}"), - ), + common::serialization_failure_hook_events(matched, turn_id, error_message), ); } }; @@ -167,11 +172,18 @@ pub(crate) async fn run( let input_json = match serde_json::to_string(&input) { Ok(input_json) => input_json, Err(error) => { + let error_message = + format!("failed to serialize subagent start hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); return serialization_failure_outcome( common::serialization_failure_hook_events( matched, Some(subagent_turn_id), - format!("failed to serialize subagent start hook input: {error}"), + error_message, ), ); } @@ -182,6 +194,7 @@ pub(crate) async fn run( let results = dispatcher::execute_handlers( shell, + async_output_queue, matched, input_json, request.cwd.as_path(), @@ -518,6 +531,7 @@ mod tests { matcher: None, command: "echo hook".to_string(), timeout_sec: 600, + r#async: false, status_message: None, source_path: test_path_buf("/tmp/hooks.json").abs(), source: codex_protocol::protocol::HookSource::User, diff --git a/codex-rs/hooks/src/events/stop.rs b/codex-rs/hooks/src/events/stop.rs index 24920c569b..0ee83572a6 100644 --- a/codex-rs/hooks/src/events/stop.rs +++ b/codex-rs/hooks/src/events/stop.rs @@ -13,6 +13,7 @@ use codex_utils_absolute_path::AbsolutePathBuf; use super::common; use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; +use crate::engine::async_output::AsyncHookOutputQueue; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; use crate::engine::output_parser; @@ -82,7 +83,7 @@ pub(crate) fn preview( handlers: &[ConfiguredHandler], request: &StopRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::select_sync_handlers( handlers, request.target.event_name(), request.target.matcher_input(), @@ -95,6 +96,7 @@ pub(crate) fn preview( pub(crate) async fn run( handlers: &[ConfiguredHandler], shell: &CommandShell, + async_output_queue: &AsyncHookOutputQueue, request: StopRequest, ) -> StopOutcome { let matched = dispatcher::select_handlers( @@ -131,11 +133,17 @@ pub(crate) async fn run( match serde_json::to_string(&input) { Ok(input_json) => input_json, Err(error) => { + let error_message = format!("failed to serialize stop hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); return serialization_failure_outcome( common::serialization_failure_hook_events( matched, Some(request.turn_id), - format!("failed to serialize stop hook input: {error}"), + error_message, ), ); } @@ -165,11 +173,18 @@ pub(crate) async fn run( match serde_json::to_string(&input) { Ok(input_json) => input_json, Err(error) => { + let error_message = + format!("failed to serialize subagent stop hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); return serialization_failure_outcome( common::serialization_failure_hook_events( matched, Some(request.turn_id), - format!("failed to serialize subagent stop hook input: {error}"), + error_message, ), ); } @@ -179,6 +194,7 @@ pub(crate) async fn run( let results = dispatcher::execute_handlers( shell, + async_output_queue, matched, input_json, request.cwd.as_path(), @@ -634,6 +650,7 @@ mod tests { matcher: None, command: "echo hook".to_string(), timeout_sec: 600, + r#async: false, status_message: None, source_path: test_path_buf("/tmp/hooks.json").abs(), source: codex_protocol::protocol::HookSource::User, diff --git a/codex-rs/hooks/src/events/user_prompt_submit.rs b/codex-rs/hooks/src/events/user_prompt_submit.rs index 2934bd3523..177b462667 100644 --- a/codex-rs/hooks/src/events/user_prompt_submit.rs +++ b/codex-rs/hooks/src/events/user_prompt_submit.rs @@ -12,6 +12,7 @@ use codex_utils_absolute_path::AbsolutePathBuf; use super::common; use crate::engine::CommandShell; use crate::engine::ConfiguredHandler; +use crate::engine::async_output::AsyncHookOutputQueue; use crate::engine::command_runner::CommandRunResult; use crate::engine::dispatcher; use crate::engine::output_parser; @@ -50,7 +51,7 @@ pub(crate) fn preview( handlers: &[ConfiguredHandler], _request: &UserPromptSubmitRequest, ) -> Vec { - dispatcher::select_handlers( + dispatcher::select_sync_handlers( handlers, HookEventName::UserPromptSubmit, /*matcher_input*/ None, @@ -63,6 +64,7 @@ pub(crate) fn preview( pub(crate) async fn run( handlers: &[ConfiguredHandler], shell: &CommandShell, + async_output_queue: &AsyncHookOutputQueue, request: UserPromptSubmitRequest, ) -> UserPromptSubmitOutcome { let matched = dispatcher::select_handlers( @@ -94,16 +96,24 @@ pub(crate) async fn run( }) { Ok(input_json) => input_json, Err(error) => { + let error_message = + format!("failed to serialize user prompt submit hook input: {error}"); + let matched = dispatcher::synchronous_handlers_after_serialization_failure( + async_output_queue, + matched, + &error_message, + ); return serialization_failure_outcome(common::serialization_failure_hook_events( matched, Some(request.turn_id), - format!("failed to serialize user prompt submit hook input: {error}"), + error_message, )); } }; let results = dispatcher::execute_handlers( shell, + async_output_queue, matched, input_json, request.cwd.as_path(), @@ -423,6 +433,7 @@ mod tests { matcher: None, command: "echo hook".to_string(), timeout_sec: 5, + r#async: false, status_message: None, source_path: test_path_buf("/tmp/hooks.json").abs(), source: codex_protocol::protocol::HookSource::User, diff --git a/codex-rs/hooks/src/lib.rs b/codex-rs/hooks/src/lib.rs index 11300802d8..3bfe6dbc4a 100644 --- a/codex-rs/hooks/src/lib.rs +++ b/codex-rs/hooks/src/lib.rs @@ -13,6 +13,8 @@ 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::AsyncHookOutputBatch; +pub use engine::AsyncHookOutputQueue; pub use engine::HookListEntry; pub use events::common::SubagentHookContext; /// Hook event names as they appear in hooks JSON and config files. diff --git a/codex-rs/hooks/src/registry.rs b/codex-rs/hooks/src/registry.rs index 1f4e01aa59..4341604133 100644 --- a/codex-rs/hooks/src/registry.rs +++ b/codex-rs/hooks/src/registry.rs @@ -2,6 +2,7 @@ use codex_config::ConfigLayerStack; use codex_plugin::PluginHookSource; use tokio::process::Command; +use crate::engine::AsyncHookOutputQueue; use crate::engine::ClaudeHooksEngine; use crate::engine::CommandShell; use crate::engine::HookListEntry; @@ -36,6 +37,7 @@ pub struct HooksConfig { pub plugin_hook_load_warnings: Vec, pub shell_program: Option, pub shell_args: Vec, + pub async_output_queue: AsyncHookOutputQueue, } #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -64,7 +66,7 @@ impl Hooks { .map(crate::notify_hook) .into_iter() .collect(); - let engine = ClaudeHooksEngine::new( + let engine = ClaudeHooksEngine::new_with_async_output_queue( config.feature_enabled, config.bypass_hook_trust, config.config_layer_stack.as_ref(), @@ -74,6 +76,7 @@ impl Hooks { program: config.shell_program.unwrap_or_default(), args: config.shell_args, }, + config.async_output_queue, ); Self { after_agent,