diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 39654f1d7d..f3a2f5d684 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3288,6 +3288,7 @@ name = "codex-hooks" version = "0.0.0" dependencies = [ "anyhow", + "async-channel", "chrono", "codex-config", "codex-plugin", diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index ee4684a8c6..ba49e994e0 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -29,12 +29,15 @@ use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::HookEventName; +use codex_protocol::protocol::HookExecutionMode; +use codex_protocol::protocol::HookOutputEntryKind; use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookRunSummary; 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; @@ -678,6 +681,51 @@ pub(crate) async fn reject_pending_input( } } +/// Processes finished async hook results at a safe turn boundary. +/// +/// Before the user prompt, records additional context directly into conversation +/// history so results from a previous turn appear before the new prompt. After +/// sampling, injects context into the active turn's pending-input queue so it +/// reaches the next sampling request. Warnings and telemetry are handled in both +/// cases. +pub(crate) async fn drain_async_hook_results( + sess: &Arc, + turn_context: &Arc, + before_user_prompt: bool, +) { + while let Ok(result) = sess.async_hook_results.try_recv() { + let additional_contexts = result + .run + .entries + .iter() + .filter(|entry| entry.kind == HookOutputEntryKind::Context) + .map(|entry| entry.text.clone()) + .collect::>(); + + if before_user_prompt { + record_additional_contexts(sess, turn_context, additional_contexts).await; + } else if !additional_contexts.is_empty() { + let _ = sess + .inject_if_running(additional_context_messages(additional_contexts)) + .await; + } + + for entry in &result.run.entries { + if entry.kind == HookOutputEntryKind::Warning { + sess.send_event( + turn_context, + EventMsg::Warning(WarningEvent { + message: entry.text.clone(), + }), + ) + .await; + } + } + + emit_hook_completed_events(sess, turn_context, vec![result]).await; + } +} + async fn run_context_injecting_hook( sess: &Arc, turn_context: &Arc, @@ -734,7 +782,10 @@ async fn emit_hook_started_events( turn_context: &Arc, preview_runs: Vec, ) { - for run in preview_runs { + for run in preview_runs + .into_iter() + .filter(|run| run.execution_mode == HookExecutionMode::Sync) + { sess.send_event( turn_context, EventMsg::HookStarted(HookStartedEvent { @@ -754,8 +805,10 @@ pub(crate) async fn emit_hook_completed_events( for completed in completed_events { emit_hook_completed_metrics(turn_context, &completed); track_hook_completed_analytics(sess, turn_context, &completed); - sess.send_event(turn_context, EventMsg::HookCompleted(completed)) - .await; + if completed.run.execution_mode == HookExecutionMode::Sync { + sess.send_event(turn_context, EventMsg::HookCompleted(completed)) + .await; + } } } @@ -902,9 +955,12 @@ mod tests { use pretty_assertions::assert_eq; use super::additional_context_messages; + use super::emit_hook_completed_events; + use super::emit_hook_started_events; use super::hook_run_analytics_payload; use super::hook_run_metric_tags; use crate::session::tests::make_session_and_context; + use crate::session::tests::make_session_and_context_with_rx; use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::HookRunSummary; use codex_utils_absolute_path::test_support::PathBufExt; @@ -946,6 +1002,57 @@ mod tests { ); } + #[tokio::test] + async fn hook_lifecycle_notifications_only_report_synchronous_runs() { + let (session, turn_context, events) = make_session_and_context_with_rx().await; + let mut synchronous_run = sample_hook_run(HookRunStatus::Running, HookSource::User); + synchronous_run.id = "synchronous-hook".to_string(); + let mut asynchronous_run = synchronous_run.clone(); + asynchronous_run.id = "asynchronous-hook".to_string(); + asynchronous_run.execution_mode = HookExecutionMode::Async; + + emit_hook_started_events( + &session, + &turn_context, + vec![asynchronous_run.clone(), synchronous_run.clone()], + ) + .await; + + let started = events.try_recv().expect("synchronous hook should start"); + assert!(matches!( + started.msg, + codex_protocol::protocol::EventMsg::HookStarted(event) + if event.run.id == synchronous_run.id + )); + assert!(events.try_recv().is_err()); + + asynchronous_run.status = HookRunStatus::Completed; + synchronous_run.status = HookRunStatus::Completed; + emit_hook_completed_events( + &session, + &turn_context, + vec![ + HookCompletedEvent { + turn_id: Some(turn_context.sub_id.clone()), + run: asynchronous_run, + }, + HookCompletedEvent { + turn_id: Some(turn_context.sub_id.clone()), + run: synchronous_run.clone(), + }, + ], + ) + .await; + + let completed = events.try_recv().expect("synchronous hook should complete"); + assert!(matches!( + completed.msg, + codex_protocol::protocol::EventMsg::HookCompleted(event) + if event.run.id == synchronous_run.id + )); + assert!(events.try_recv().is_err()); + } + #[tokio::test] async fn hook_run_analytics_payload_uses_completed_turn_id() { let (_session, turn_context) = make_session_and_context().await; diff --git a/codex-rs/core/src/mcp_tool_call_tests.rs b/codex-rs/core/src/mcp_tool_call_tests.rs index 7def794800..626b9a2cd7 100644 --- a/codex-rs/core/src/mcp_tool_call_tests.rs +++ b/codex-rs/core/src/mcp_tool_call_tests.rs @@ -22,7 +22,6 @@ use codex_config::types::AppsConfigToml; use codex_config::types::McpServerConfig; use codex_config::types::McpServerToolConfig; use codex_features::Features; -use codex_hooks::Hooks; use codex_hooks::HooksConfig; use codex_model_provider::create_model_provider; use codex_protocol::models::PermissionProfile; @@ -203,20 +202,18 @@ print({hook_output:?}) hook_list.hooks, ); - session - .services - .hooks - .store(Arc::new(Hooks::new(HooksConfig { - feature_enabled: true, - config_layer_stack: Some(trusted_config_layer_stack), - shell_program: (!cfg!(windows)).then_some("/bin/sh".to_string()), - shell_args: if cfg!(windows) { - Vec::new() - } else { - vec!["-c".to_string()] - }, - ..HooksConfig::default() - }))); + let hooks = session.hooks().reconfigured(HooksConfig { + feature_enabled: true, + config_layer_stack: Some(trusted_config_layer_stack), + shell_program: (!cfg!(windows)).then_some("/bin/sh".to_string()), + shell_args: if cfg!(windows) { + Vec::new() + } else { + vec!["-c".to_string()] + }, + ..HooksConfig::default() + }); + session.services.hooks.store(Arc::new(hooks)); log_path.to_path_buf() } diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 8ed7a7badd..09fedfb125 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -590,6 +590,9 @@ async fn shutdown_session_runtime(sess: &Arc) { } let _ = sess.conversation.shutdown().await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; + sess.hooks().shutdown().await; + sess.async_hook_results.close(); + while sess.async_hook_results.try_recv().is_ok() {} 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 1180a94a4d..e2b39499ea 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1742,7 +1742,7 @@ impl Session { self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref()); self.schedule_mcp_prewarm(); 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(), @@ -1756,6 +1756,7 @@ impl Session { &state.session_configuration.original_config_do_not_use, &config, ) { + let hooks = self.hooks().reconfigured(hooks_config); self.services.hooks.store(Arc::new(hooks)); } } @@ -4203,12 +4204,12 @@ 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 hook configuration for one config snapshot, including any enabled plugin hooks. +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| { @@ -4222,7 +4223,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, @@ -4231,7 +4232,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 3676d4aaab..912a2e850a 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -21,6 +21,7 @@ use codex_protocol::config_types::ServiceTier; use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::permissions::FileSystemPath; use codex_protocol::permissions::FileSystemSpecialPath; +use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSource; @@ -55,6 +56,7 @@ pub(crate) struct Session { pub(super) mcp_prewarm_task: std::sync::Mutex>>, pub(crate) conversation: Arc, pub(crate) active_turn: Mutex>, + pub(crate) async_hook_results: async_channel::Receiver, pub(crate) pending_user_message_admissions: crate::user_message_admission::PendingUserMessageAdmissions, pub(crate) input_queue: InputQueue, @@ -1169,12 +1171,13 @@ impl Session { (None, None) }; - let hooks = build_hooks_for_config( + let hooks_config = build_hooks_config( &config, plugins_manager.as_ref(), resolved_environments.single_local_environment(), ) .await; + let (hooks, async_hook_results) = Hooks::new(hooks_config, thread_id); for warning in hooks.startup_warnings() { post_session_configured_events.push(Event { id: INITIAL_SUBMIT_ID.to_owned(), @@ -1320,6 +1323,7 @@ impl Session { mcp_prewarm_task: std::sync::Mutex::new(None), conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), + async_hook_results, pending_user_message_admissions: Default::default(), input_queue: InputQueue::new(), guardian_review_session: GuardianReviewSessionManager::default(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 7ebe44e3ee..c25947df89 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -762,15 +762,19 @@ async fn write_project_trust_config( async fn preview_session_start_hooks( config: &crate::config::Config, ) -> std::io::Result> { - let hooks = Hooks::new(HooksConfig { - feature_enabled: true, - config_layer_stack: Some(config.config_layer_stack.clone()), - ..HooksConfig::default() - }); + let thread_id = ThreadId::new(); + let (hooks, _result_receiver) = Hooks::new( + HooksConfig { + feature_enabled: true, + config_layer_stack: Some(config.config_layer_stack.clone()), + ..HooksConfig::default() + }, + thread_id, + ); Ok( hooks.preview_session_start(&codex_hooks::SessionStartRequest { - session_id: ThreadId::new(), + session_id: thread_id, cwd: config.cwd.clone(), transcript_path: None, model: "gpt-5.2".to_string(), @@ -5755,6 +5759,13 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { .features .enabled(Feature::ExecutedToolCallMetadata) .then(|| Arc::new(crate::state::ExecutedToolCallRecorder::default())); + let (hooks, async_hook_results) = Hooks::new( + HooksConfig { + legacy_notify_argv: config.notify.clone(), + ..HooksConfig::default() + }, + thread_id, + ); let services = SessionServices { mcp_runtime, mcp_handler_cache: Default::default(), @@ -5769,10 +5780,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { config.chatgpt_base_url.trim_end_matches('/').to_string(), config.analytics_enabled, ), - hooks: arc_swap::ArcSwap::from_pointee(Hooks::new(HooksConfig { - legacy_notify_argv: config.notify.clone(), - ..HooksConfig::default() - })), + hooks: arc_swap::ArcSwap::from_pointee(hooks), rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell: Arc::new(default_user_shell()), show_raw_agent_reasoning: config.show_raw_agent_reasoning, @@ -5896,6 +5904,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { mcp_prewarm_task: std::sync::Mutex::new(None), conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), + async_hook_results, pending_user_message_admissions: Default::default(), input_queue: super::input_queue::InputQueue::new(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), @@ -7433,9 +7442,41 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() { .expect("create thread persistence"); session.services.thread_store = thread_store; session.services.live_thread = Some(live_thread); + let (result_sender, result_receiver) = async_channel::unbounded(); + result_sender + .try_send( + serde_json::from_value::(json!({ + "turn_id": "turn-1", + "run": { + "id": "user_prompt_submit:0:hooks.json", + "event_name": "user_prompt_submit", + "handler_type": "command", + "execution_mode": "async", + "scope": "turn", + "source_path": config.cwd.join("hooks.json"), + "source": "user", + "display_order": 0, + "status": "completed", + "status_message": null, + "started_at": 0, + "completed_at": 1, + "duration_ms": 1, + "entries": [{ + "kind": "context", + "text": "must not be persisted during shutdown" + }] + } + })) + .expect("valid buffered async hook result"), + ) + .expect("buffer an async hook result before shutdown"); + session.async_hook_results = result_receiver; let session = Arc::new(session); assert!(handlers::shutdown(&session, "sub-1".to_string()).await); + assert!(session.async_hook_results.is_closed()); + assert!(session.async_hook_results.is_empty()); + assert!(result_sender.is_closed()); assert_eq!( codex_thread_store::InMemoryThreadStoreCalls { @@ -7977,6 +8018,13 @@ where .features .enabled(Feature::ExecutedToolCallMetadata) .then(|| Arc::new(crate::state::ExecutedToolCallRecorder::default())); + let (hooks, async_hook_results) = Hooks::new( + HooksConfig { + legacy_notify_argv: config.notify.clone(), + ..HooksConfig::default() + }, + thread_id, + ); let services = SessionServices { mcp_runtime, mcp_handler_cache: Default::default(), @@ -7991,10 +8039,7 @@ where config.chatgpt_base_url.trim_end_matches('/').to_string(), config.analytics_enabled, ), - hooks: arc_swap::ArcSwap::from_pointee(Hooks::new(HooksConfig { - legacy_notify_argv: config.notify.clone(), - ..HooksConfig::default() - })), + hooks: arc_swap::ArcSwap::from_pointee(hooks), rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell: Arc::new(default_user_shell()), show_raw_agent_reasoning: config.show_raw_agent_reasoning, @@ -8118,6 +8163,7 @@ where mcp_prewarm_task: std::sync::Mutex::new(None), conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), + async_hook_results, pending_user_message_admissions: Default::default(), input_queue: super::input_queue::InputQueue::new(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index b7fc003fbc..869d6ab312 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -15,6 +15,7 @@ use crate::connectors; use crate::context::ContextualUserFragment; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::feedback_tags; +use crate::hook_runtime::drain_async_hook_results; use crate::hook_runtime::inspect_pending_input; use crate::hook_runtime::record_additional_contexts; use crate::hook_runtime::record_pending_input; @@ -156,6 +157,9 @@ pub(crate) async fn run_turn( prewarmed_client_session: Option, cancellation_token: CancellationToken, ) -> CodexResult> { + // Record results from hooks that finished after the previous turn before this turn's user prompt. + drain_async_hook_results(&sess, &turn_context, /*before_user_prompt*/ true).await; + 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 @@ -374,6 +378,8 @@ pub(crate) async fn run_turn( .await; } can_drain_pending_input = true; + // Process async hooks only after sampling and its tools have finished. + drain_async_hook_results(&sess, &turn_context, /*before_user_prompt*/ false).await; let (has_pending_input, token_status) = async { let has_pending_input = sess.input_queue.has_pending_input(&sess.active_turn).await; diff --git a/codex-rs/core/src/tools/runtimes/shell/unix_escalation_tests.rs b/codex-rs/core/src/tools/runtimes/shell/unix_escalation_tests.rs index 7834ddc7c3..cfd95e5f73 100644 --- a/codex-rs/core/src/tools/runtimes/shell/unix_escalation_tests.rs +++ b/codex-rs/core/src/tools/runtimes/shell/unix_escalation_tests.rs @@ -15,7 +15,6 @@ use codex_execpolicy::Decision; use codex_execpolicy::Evaluation; use codex_execpolicy::PolicyParser; use codex_execpolicy::RuleMatch; -use codex_hooks::Hooks; use codex_hooks::HooksConfig; use codex_network_proxy::PROXY_ACTIVE_ENV_KEY; use codex_network_proxy::PROXY_ENV_KEYS; @@ -536,16 +535,14 @@ async fn execve_permission_request_hook_short_circuits_prompt() -> anyhow::Resul .derive_exec_args("", /*use_login_shell*/ false); let hook_shell_program = hook_shell_argv.remove(0); let _ = hook_shell_argv.pop(); - session - .services - .hooks - .store(Arc::new(Hooks::new(HooksConfig { - feature_enabled: true, - config_layer_stack: Some(trusted_config_layer_stack), - shell_program: Some(hook_shell_program), - shell_args: hook_shell_argv, - ..HooksConfig::default() - }))); + let hooks = session.hooks().reconfigured(HooksConfig { + feature_enabled: true, + config_layer_stack: Some(trusted_config_layer_stack), + shell_program: Some(hook_shell_program), + shell_args: hook_shell_argv, + ..HooksConfig::default() + }); + session.services.hooks.store(Arc::new(hooks)); Arc::make_mut(&mut turn_context.config) .permissions diff --git a/codex-rs/core/tests/suite/hooks.rs b/codex-rs/core/tests/suite/hooks.rs index 9302955307..5fe2cc882a 100644 --- a/codex-rs/core/tests/suite/hooks.rs +++ b/codex-rs/core/tests/suite/hooks.rs @@ -32,6 +32,7 @@ use codex_protocol::user_input::UserInput; use codex_thread_store::InMemoryThreadStore; use codex_utils_absolute_path::AbsolutePathBuf; use core_test_support::TestTargetOs; +use core_test_support::fs_wait; use core_test_support::hooks::trust_discovered_hooks; use core_test_support::hooks::trust_hooks; use core_test_support::managed_network_requirements_loader; @@ -307,6 +308,55 @@ if payload.get("prompt") == {blocked_prompt_json}: Ok(()) } +fn write_async_user_prompt_submit_hook(home: &Path, gated: bool) -> Result<()> { + let script_path = home.join("async_user_prompt_submit_hook.py"); + let started_path = home.join("async_user_prompt_submit_started"); + let finished_path = home.join("async_user_prompt_submit_finished"); + let release_path = home.join("async_user_prompt_submit_release"); + let script = format!( + r#"import json +from pathlib import Path +import sys +import time + +prompt = json.load(sys.stdin).get("prompt") +Path(r"{started_path}").write_text(prompt, encoding="utf-8") +while {gated} and not Path(r"{release_path}").exists(): + time.sleep(0.01) +print(json.dumps({{ + "continue": False, + "decision": "block", + "reason": "an async hook cannot block its launching operation", + "systemMessage": f"async warning for {{prompt}}", + "hookSpecificOutput": {{ + "hookEventName": "UserPromptSubmit", + "additionalContext": f"async context for {{prompt}}" + }} +}}), flush=True) +Path(r"{finished_path}").write_text(prompt, encoding="utf-8") +"#, + started_path = started_path.display(), + finished_path = finished_path.display(), + release_path = release_path.display(), + gated = if gated { "True" } else { "False" }, + ); + let hooks = serde_json::json!({ + "hooks": { + "UserPromptSubmit": [{ + "hooks": [{ + "type": "command", + "command": format!("python3 {}", script_path.display()), + "async": true, + }] + }] + } + }); + + fs::write(&script_path, script).context("write async user prompt submit hook script")?; + fs::write(home.join("hooks.json"), hooks.to_string()).context("write async hooks.json")?; + Ok(()) +} + fn write_session_start_and_user_prompt_submit_order_hooks(home: &Path) -> Result<()> { let session_start_script_path = home.join("session_start_order_hook.py"); let user_prompt_submit_script_path = home.join("user_prompt_submit_order_hook.py"); @@ -1519,6 +1569,255 @@ async fn session_start_runs_before_user_prompt_submit_on_first_turn() -> Result< Ok(()) } +#[tokio::test] +async fn async_hook_context_is_injected_into_the_active_turn() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let gate = TempDir::new()?; + let release_path = gate.path().join("release"); + let call_id = "async-hook-context-gated-shell-command"; + let args = serde_json::json!({ + "command": format!( + r#"python3 -c 'import time; from pathlib import Path; gate = Path(r"{}"); exec("while not gate.exists(): time.sleep(0.01)")'"#, + release_path.display() + ) + }); + let responses = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_function_call(call_id, "shell_command", &serde_json::to_string(&args)?), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-1", "async context observed in this turn"), + ev_completed("resp-2"), + ]), + ], + ) + .await; + + let test = test_codex() + .with_pre_build_hook(|home| { + write_async_user_prompt_submit_hook(home, /*gated*/ false) + .expect("write immediate async user prompt submit hook"); + }) + .with_config(trust_discovered_hooks) + .build(&server) + .await?; + + let (sandbox_policy, permission_profile) = + turn_permission_fields(PermissionProfile::Disabled, test.config.cwd.as_path()); + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "observe async context immediately".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: codex_protocol::protocol::ThreadSettingsOverrides { + approval_policy: Some(AskForApproval::Never), + sandbox_policy: Some(sandbox_policy), + permission_profile, + ..Default::default() + }, + }) + .await?; + + let finished_path = test + .codex_home_path() + .join("async_user_prompt_submit_finished"); + fs_wait::wait_for_path_exists(finished_path, Duration::from_secs(5)) + .await + .context("timed out waiting for the async hook to finish")?; + assert!( + timeout( + Duration::from_millis(150), + wait_for_event(&test.codex, |event| { + matches!( + event, + EventMsg::Warning(warning) + if warning.message == "async warning for observe async context immediately" + ) + }), + ) + .await + .is_err(), + "async hook output must not interrupt the active sampling request" + ); + fs::write(release_path, "ready").context("release gated shell command")?; + timeout( + Duration::from_secs(5), + wait_for_event(&test.codex, |event| { + matches!( + event, + EventMsg::Warning(warning) + if warning.message == "async warning for observe async context immediately" + ) + }), + ) + .await + .context("timed out waiting for the async hook warning after sampling")?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let requests = responses.requests(); + assert_eq!(requests.len(), 2); + assert!( + requests[1] + .message_input_texts("developer") + .contains(&"async context for observe async context immediately".to_string()), + "an async hook should steer its context into the same active turn" + ); + + Ok(()) +} + +#[tokio::test] +async fn async_hook_finishing_while_idle_waits_for_the_next_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 completed"), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-2", "idle async context observed"), + ev_completed("resp-2"), + ]), + ], + ) + .await; + + let test = test_codex() + .with_pre_build_hook(|home| { + write_async_user_prompt_submit_hook(home, /*gated*/ true) + .expect("write gated async user prompt submit hook"); + }) + .with_config(trust_discovered_hooks) + .build(&server) + .await?; + + test.submit_turn("finish before the async hook").await?; + let first_turn_id = responses.requests()[0].body_json()["client_metadata"]["turn_id"] + .as_str() + .context("first model request should include its turn ID")? + .to_string(); + let started_path = test + .codex_home_path() + .join("async_user_prompt_submit_started"); + fs_wait::wait_for_path_exists(started_path, Duration::from_secs(5)) + .await + .context("timed out waiting for the async hook to start")?; + + fs::write( + test.codex_home_path() + .join("async_user_prompt_submit_release"), + "ready", + ) + .context("release gated async hook")?; + + let finished_path = test + .codex_home_path() + .join("async_user_prompt_submit_finished"); + fs_wait::wait_for_path_exists(finished_path, Duration::from_secs(5)) + .await + .context("timed out waiting for the async hook to finish")?; + + assert!( + timeout(Duration::from_millis(150), test.codex.next_event()) + .await + .is_err(), + "an async hook result from the previous turn must not emit warnings or raw response items" + ); + + assert_eq!( + responses.requests().len(), + 1, + "an async hook result from the previous turn must not start a model turn" + ); + + let next_prompt = "observe the buffered async context"; + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: next_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 mut warning_event = None; + timeout(Duration::from_secs(5), async { + loop { + let event = test.codex.next_event().await?; + if matches!( + &event.msg, + EventMsg::Warning(warning) + if warning.message == "async warning for finish before the async hook" + ) { + warning_event = Some(event); + continue; + } + if matches!(event.msg, EventMsg::TurnComplete(_)) { + return Ok::<_, anyhow::Error>(()); + } + } + }) + .await + .context("timed out waiting for the next turn to complete")??; + + let requests = responses.requests(); + assert_eq!(requests.len(), 2); + let second_turn_id = requests[1].body_json()["client_metadata"]["turn_id"] + .as_str() + .context("second model request should include its turn ID")? + .to_string(); + assert_ne!(first_turn_id, second_turn_id); + assert_eq!( + warning_event + .context("buffered async hook warning should be delivered during the next turn")? + .id, + second_turn_id + ); + + let input = requests[1].input(); + let context_index = input + .iter() + .position(|item| { + item.to_string() + .contains("async context for finish before the async hook") + }) + .context("buffered async hook context should be present in the next model request")?; + let prompt_index = input + .iter() + .position(|item| item.to_string().contains(next_prompt)) + .context("next user prompt should be present in the next model request")?; + assert!( + context_index < prompt_index, + "buffered async hook context should precede the next user prompt" + ); + + Ok(()) +} + #[tokio::test] async fn session_start_hook_spills_large_additional_context() -> Result<()> { skip_if_no_network!(Ok(())); @@ -2591,7 +2890,8 @@ async fn permission_request_hook_allow_bypasses_strict_auto_review() -> Result<( .enable(Feature::RequestPermissionsTool) .expect("test config should allow feature update"); }); - let test = builder.build_with_auto_env(&server).await?; + // Command hooks currently need a local executor even when the tool executor is remote. + let test = builder.build_with_remote_and_local_env(&server).await?; let marker = test .executor_environment() @@ -3204,6 +3504,169 @@ async fn blocked_pre_tool_use_records_additional_context_for_shell_command() -> Ok(()) } +#[tokio::test] +async fn async_pre_tool_use_cannot_block_or_rewrite_and_still_records_additional_context() +-> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let gate = TempDir::new()?; + let release_path = gate.path().join("release"); + let hook_finished_path = gate.path().join("hook-finished"); + let hook_finished_path_for_fixture = hook_finished_path.clone(); + let original_marker = gate.path().join("original"); + let rewritten_marker = gate.path().join("rewritten"); + let call_id = "async-pretooluse-shell-command"; + let original_command = format!( + r#"python3 -c 'import time; from pathlib import Path; gate = Path(r"{}"); exec("while not gate.exists(): time.sleep(0.01)"); Path(r"{}").write_text("original"); print("original-output")'"#, + release_path.display(), + original_marker.display() + ); + let args = serde_json::json!({ "command": original_command }); + let responses = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_function_call(call_id, "shell_command", &serde_json::to_string(&args)?), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-1", "async pre-tool hook context observed"), + ev_completed("resp-2"), + ]), + ], + ) + .await; + + let pre_context = "async pre-tool hook cannot block its launching command"; + let updated_input = serde_json::json!({ + "command": format!("printf rewritten > {}", rewritten_marker.display()) + }); + let test = test_codex() + .with_pre_build_hook(move |home| { + let denying_script = home.join("async_deny_pre_tool_use_hook.py"); + fs::write( + &denying_script, + format!( + r#"import json, sys +from pathlib import Path +json.load(sys.stdin) +print(json.dumps({{ + "systemMessage": "{pre_context}", + "hookSpecificOutput": {{ + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "{pre_context}", + "additionalContext": "{pre_context}" + }} +}}), flush=True) +Path(r"{hook_finished_path}").write_text("finished", encoding="utf-8") +"#, + hook_finished_path = hook_finished_path_for_fixture.display(), + ), + ) + .expect("write async denying pre-tool hook fixture"); + write_updating_pre_tool_use_hook(home, "^Bash$", &updated_input) + .expect("write async rewriting pre-tool hook fixture"); + + let hooks_path = home.join("hooks.json"); + let mut hooks: Value = serde_json::from_slice( + &fs::read(&hooks_path).expect("read async pre-tool hook config"), + ) + .expect("parse async pre-tool hook config"); + let handlers = hooks["hooks"]["PreToolUse"][0]["hooks"] + .as_array_mut() + .expect("pre-tool hook handlers"); + handlers[0]["async"] = Value::Bool(true); + handlers.push(serde_json::json!({ + "type": "command", + "command": format!("python3 {}", denying_script.display()), + "async": true, + })); + fs::write(hooks_path, hooks.to_string()).expect("write async pre-tool hook config"); + }) + .with_config(trust_discovered_hooks) + .build(&server) + .await?; + + let (sandbox_policy, permission_profile) = + turn_permission_fields(PermissionProfile::Disabled, test.config.cwd.as_path()); + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "run the original command with async pre-tool hooks".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: codex_protocol::protocol::ThreadSettingsOverrides { + approval_policy: Some(AskForApproval::Never), + sandbox_policy: Some(sandbox_policy), + permission_profile, + ..Default::default() + }, + }) + .await?; + + fs_wait::wait_for_path_exists(hook_finished_path, Duration::from_secs(5)) + .await + .context("timed out waiting for the async pre-tool hook to finish")?; + assert!( + timeout( + Duration::from_millis(150), + wait_for_event( + &test.codex, + |event| matches!(event, EventMsg::Warning(warning) if warning.message == pre_context), + ), + ) + .await + .is_err(), + "async pre-tool output must not interrupt the launching command" + ); + fs::write(release_path, "ready").context("release original shell command")?; + timeout( + Duration::from_secs(5), + wait_for_event( + &test.codex, + |event| matches!(event, EventMsg::Warning(warning) if warning.message == pre_context), + ), + ) + .await + .context("timed out waiting for the async pre-tool warning after sampling")?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let requests = responses.requests(); + assert_eq!(requests.len(), 2); + assert!( + requests[1] + .message_input_texts("developer") + .contains(&pre_context.to_string()), + "async pre-tool hook context should reach the follow-up model request", + ); + let output_item = requests[1].function_call_output(call_id); + let output = output_item + .get("output") + .and_then(Value::as_str) + .expect("shell command output string"); + assert!( + output.contains("original-output"), + "async pre-tool hooks should not block the original command", + ); + assert!(original_marker.exists(), "original command should execute"); + assert!( + !rewritten_marker.exists(), + "async pre-tool hooks should not rewrite the original command", + ); + + Ok(()) +} + #[derive(Clone, Copy)] enum BashRewriteSurface { ExecCommand, diff --git a/codex-rs/hooks/Cargo.toml b/codex-rs/hooks/Cargo.toml index 5f5a180766..32149dc650 100644 --- a/codex-rs/hooks/Cargo.toml +++ b/codex-rs/hooks/Cargo.toml @@ -14,6 +14,7 @@ workspace = true [dependencies] anyhow = { workspace = true } +async-channel = { workspace = true } chrono = { workspace = true, features = ["serde"] } codex-config = { workspace = true } codex-plugin = { workspace = true } @@ -25,7 +26,7 @@ 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", "sync", "time"] } tracing = { workspace = true } uuid = { workspace = true, features = ["v4"] } diff --git a/codex-rs/hooks/src/engine/command_runner.rs b/codex-rs/hooks/src/engine/command_runner.rs index 635defb831..aec7e723b3 100644 --- a/codex-rs/hooks/src/engine/command_runner.rs +++ b/codex-rs/hooks/src/engine/command_runner.rs @@ -1,46 +1,163 @@ use std::io::ErrorKind; use std::path::Path; use std::process::Stdio; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::MutexGuard; use std::time::Duration; use std::time::Instant; +use async_channel::Sender; #[cfg(windows)] use codex_utils_pty::JobObject; use tokio::io::AsyncWriteExt; use tokio::process::Command; +use tokio::sync::Semaphore; +use tokio::task::JoinSet; use tokio::time::timeout; use tracing::Span; use super::CommandShell; use super::ConfiguredHandler; +use super::dispatcher::ParsedHandler; use super::dispatcher::hook_event_name_label; use super::dispatcher::hook_execution_mode_label; 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 crate::output_spill::AdditionalContext; use crate::output_spill::HookOutputSpiller; -use codex_protocol::protocol::HookExecutionMode; +use codex_protocol::ThreadId; +use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::HookHandlerType; +use codex_protocol::protocol::HookOutputEntry; +use codex_protocol::protocol::HookOutputEntryKind; -/// Owns the shell and output spiller shared by a session's command hooks. +const MAX_CONCURRENT_ASYNC_HOOKS: usize = 8; + +/// Owns command execution and bounded asynchronous work for one session. #[derive(Clone)] pub(crate) struct CommandHookRuntime { shell: CommandShell, + result_sender: Sender, + state: Arc>, output_spiller: HookOutputSpiller, } +struct CommandHookRuntimeState { + concurrency_limit: Arc, + tasks: JoinSet<()>, +} + +impl Default for CommandHookRuntimeState { + fn default() -> Self { + Self { + concurrency_limit: Arc::new(Semaphore::new(MAX_CONCURRENT_ASYNC_HOOKS)), + tasks: JoinSet::new(), + } + } +} + impl CommandHookRuntime { - pub(crate) fn new(shell: CommandShell) -> Self { + pub(crate) fn new( + shell: CommandShell, + thread_id: ThreadId, + result_sender: Sender, + ) -> Self { Self { shell, - output_spiller: HookOutputSpiller::new(), + result_sender, + state: Arc::new(Mutex::new(CommandHookRuntimeState::default())), + output_spiller: HookOutputSpiller::new(thread_id), + } + } + + fn lock_state(&self) -> MutexGuard<'_, CommandHookRuntimeState> { + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + pub(crate) fn reconfigured(&self, shell: CommandShell) -> Self { + Self { + shell, + result_sender: self.result_sender.clone(), + state: Arc::clone(&self.state), + output_spiller: self.output_spiller.clone(), } } pub(crate) fn output_spiller(&self) -> &HookOutputSpiller { &self.output_spiller } + + pub(crate) fn schedule_async_hook( + &self, + handler: ConfiguredHandler, + input_json: String, + cwd: std::path::PathBuf, + turn_id: Option, + parse: fn(&ConfiguredHandler, CommandRunResult, Option) -> ParsedHandler, + ) { + let mut state = self.lock_state(); + if self.result_sender.is_closed() || state.concurrency_limit.is_closed() { + return; + } + + while state.tasks.try_join_next().is_some() {} + let result_sender = self.result_sender.clone(); + let concurrency_limit = Arc::clone(&state.concurrency_limit); + let runtime = self.clone(); + state.tasks.spawn(async move { + let Ok(_permit) = concurrency_limit.acquire_owned().await else { + return; + }; + let result = run_command(&runtime, &handler, &input_json, &cwd).await; + let mut hook_result = parse(&handler, result, turn_id).completed; + let mut entries = Vec::new(); + let mut warnings = Vec::new(); + + for entry in std::mem::take(&mut hook_result.run.entries) { + match entry.kind { + HookOutputEntryKind::Context => { + if let Some(text) = runtime + .output_spiller + .maybe_spill_additional_contexts(vec![AdditionalContext { + text: entry.text, + limit: handler.additional_context_limit, + }]) + .await + .into_iter() + .next() + { + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Context, + text, + }); + } + } + HookOutputEntryKind::Warning => warnings.push(entry), + HookOutputEntryKind::Error => entries.push(entry), + HookOutputEntryKind::Stop | HookOutputEntryKind::Feedback => {} + } + } + + entries.extend(warnings); + hook_result.run.entries = entries; + let _ = result_sender.try_send(hook_result); + }); + } + + pub(crate) async fn shutdown(&self) { + let mut tasks = { + let mut state = self.lock_state(); + state.concurrency_limit.close(); + std::mem::take(&mut state.tasks) + }; + tasks.abort_all(); + while tasks.join_next().await.is_some() {} + } } #[derive(Debug)] @@ -61,11 +178,10 @@ 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, - hook.configured_order = configured_order, hook.timeout_sec = handler.timeout_sec, hook.command_outcome = tracing::field::Empty, ) @@ -73,7 +189,6 @@ pub(crate) struct CommandRunResult { pub(crate) async fn run_command( runtime: &CommandHookRuntime, handler: &ConfiguredHandler, - configured_order: usize, input_json: &str, cwd: &Path, ) -> CommandRunResult { diff --git a/codex-rs/hooks/src/engine/command_runner_tests.rs b/codex-rs/hooks/src/engine/command_runner_tests.rs index 6a22fd84a3..177b0f0a60 100644 --- a/codex-rs/hooks/src/engine/command_runner_tests.rs +++ b/codex-rs/hooks/src/engine/command_runner_tests.rs @@ -1,17 +1,31 @@ use std::collections::HashMap; #[cfg(windows)] use std::fs; +use std::path::Path; +use std::time::Duration; +use async_channel::Receiver; +use codex_protocol::ThreadId; +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; use codex_protocol::protocol::HookSource; use codex_utils_absolute_path::AbsolutePathBuf; use pretty_assertions::assert_eq; +use tempfile::TempDir; use tempfile::tempdir; +use tokio::time::sleep; +use tokio::time::timeout; use super::CommandHookRuntime; use super::CommandShell; use super::ConfiguredHandler; +use super::MAX_CONCURRENT_ASYNC_HOOKS; use super::run_command; +use crate::events::user_prompt_submit::UserPromptSubmitRequest; #[cfg(windows)] #[tokio::test] @@ -29,6 +43,7 @@ async fn cmd_shell_runs_quoted_hook_command_path() { AbsolutePathBuf::try_from(hook_path.clone()).expect("absolute hook command path"); let handler = ConfiguredHandler { event_name: HookEventName::SessionStart, + execution_mode: HookExecutionMode::Sync, matcher: None, command: format!(r#""{}" notify"#, hook_path.display()), timeout_sec: 10, @@ -51,14 +66,9 @@ async fn cmd_shell_runs_quoted_hook_command_path() { ]; for shell in shells { - let result = run_command( - &CommandHookRuntime::new(shell), - &handler, - /*configured_order*/ 0, - "{}", - temp.path(), - ) - .await; + let (result_sender, _result_receiver) = async_channel::unbounded(); + let runtime = CommandHookRuntime::new(shell, ThreadId::new(), result_sender); + let result = run_command(&runtime, &handler, "{}", temp.path()).await; assert_eq!(result.exit_code, Some(0), "stderr: {}", result.stderr); assert_eq!(result.stdout.trim(), "hook-ran"); @@ -73,6 +83,7 @@ async fn fast_exiting_hook_preserves_stdout_when_stdin_is_not_consumed() { .expect("absolute hook configuration path"); let handler = ConfiguredHandler { event_name: HookEventName::SessionStart, + execution_mode: HookExecutionMode::Sync, matcher: None, command: "echo hook-ran".to_string(), timeout_sec: 10, @@ -83,22 +94,248 @@ async fn fast_exiting_hook_preserves_stdout_when_stdin_is_not_consumed() { display_order: 0, env: HashMap::new(), }; - let shell = CommandShell { - program: String::new(), - args: Vec::new(), - }; let input_json = format!(r#"{{"padding":"{}"}}"#, "x".repeat(1024 * 1024)); + let (runtime, _result_receiver) = runtime(); - let result = run_command( - &CommandHookRuntime::new(shell), - &handler, - /*configured_order*/ 0, - &input_json, - temp.path(), - ) - .await; + let result = run_command(&runtime, &handler, &input_json, temp.path()).await; assert_eq!(result.exit_code, Some(0), "stderr: {}", result.stderr); assert_eq!(result.stdout.trim(), "hook-ran"); assert_eq!(result.error, None); } + +const ASYNC_HOOK_TEST_TIMEOUT: Duration = Duration::from_secs(30); + +fn runtime() -> (CommandHookRuntime, Receiver) { + let thread_id = ThreadId::new(); + let (result_sender, result_receiver) = async_channel::unbounded(); + let runtime = CommandHookRuntime::new( + CommandShell { + program: String::new(), + args: Vec::new(), + }, + thread_id, + result_sender, + ); + (runtime, result_receiver) +} + +fn write_handler(temp: &TempDir, source: &str) -> ConfiguredHandler { + let script_path = temp.path().join("async_hook.py"); + std::fs::write(&script_path, source).expect("write async test hook"); + ConfiguredHandler { + event_name: HookEventName::UserPromptSubmit, + execution_mode: HookExecutionMode::Async, + matcher: None, + command: format!("python3 {}", script_path.display()), + timeout_sec: 10, + status_message: None, + additional_context_limit: Default::default(), + source_path: AbsolutePathBuf::try_from(temp.path().join("hooks.json")) + .expect("absolute test hook path"), + source: HookSource::User, + display_order: 0, + env: HashMap::new(), + } +} + +async fn schedule(runtime: &CommandHookRuntime, handler: ConfiguredHandler, cwd: &Path) { + crate::events::user_prompt_submit::run( + &[handler], + runtime, + UserPromptSubmitRequest { + session_id: ThreadId::new(), + turn_id: "async-test-turn".to_string(), + subagent: None, + cwd: AbsolutePathBuf::try_from(cwd.to_path_buf()).expect("absolute test hook cwd"), + transcript_path: None, + model: "test-model".to_string(), + permission_mode: "default".to_string(), + prompt: "test prompt".to_string(), + }, + ) + .await; +} + +#[tokio::test] +async fn async_hook_marks_invalid_structured_output_as_failed() { + let temp = TempDir::new().expect("async test directory"); + let (runtime, results) = runtime(); + let handler = write_handler( + &temp, + r#"import sys + +sys.stdin.read() +print('{"systemMessage": 123}') +"#, + ); + + schedule(&runtime, handler, temp.path()).await; + let hook_result = timeout(ASYNC_HOOK_TEST_TIMEOUT, results.recv()) + .await + .expect("invalid async output should still finish its background task") + .expect("result receiver should remain open"); + + assert_eq!(hook_result.run.status, HookRunStatus::Failed); + assert_eq!( + hook_result.run.entries, + vec![HookOutputEntry { + kind: HookOutputEntryKind::Error, + text: "hook returned invalid user prompt submit JSON output".to_string(), + }] + ); + + runtime.shutdown().await; +} + +#[tokio::test] +async fn async_hook_result_survives_runtime_reconfiguration() { + let temp = TempDir::new().expect("async test directory"); + let (previous, results) = runtime(); + let release_path = temp.path().join("release"); + let handler = write_handler( + &temp, + &format!( + r#"import json +from pathlib import Path +import sys +import time + +json.load(sys.stdin) +while not Path(r"{}").exists(): + time.sleep(0.01) +print("survived reconfiguration") +"#, + release_path.display() + ), + ); + schedule(&previous, handler, temp.path()).await; + + let reconfigured = previous.reconfigured(CommandShell { + program: String::new(), + args: Vec::new(), + }); + std::fs::write(release_path, "ready").expect("release async hook"); + + let hook_result = timeout(ASYNC_HOOK_TEST_TIMEOUT, results.recv()) + .await + .expect("in-flight hook should survive runtime reconfiguration") + .expect("result receiver should remain open"); + assert_eq!( + hook_result.run.entries, + vec![HookOutputEntry { + kind: HookOutputEntryKind::Context, + text: "survived reconfiguration".to_string(), + }] + ); + + reconfigured.shutdown().await; +} + +#[tokio::test] +async fn async_hooks_limit_concurrent_processes_without_dropping_waiting_jobs() { + let temp = TempDir::new().expect("async test directory"); + let (runtime, results) = runtime(); + let started_dir = temp.path().join("started"); + let release_path = temp.path().join("release"); + std::fs::create_dir(&started_dir).expect("create hook marker directory"); + let mut handler = write_handler( + &temp, + &format!( + r#"import os +from pathlib import Path +import sys +import time + +sys.stdin.read() +Path(r"{started}", str(os.getpid())).touch() +while not Path(r"{release}").exists(): + time.sleep(0.01) +print("{{}}") +"#, + started = started_dir.display(), + release = release_path.display(), + ), + ); + handler.timeout_sec = ASYNC_HOOK_TEST_TIMEOUT.as_secs(); + + for _ in 0..=MAX_CONCURRENT_ASYNC_HOOKS { + schedule(&runtime, handler.clone(), temp.path()).await; + } + + let started_count = || { + std::fs::read_dir(&started_dir) + .expect("read hook marker directory") + .count() + }; + timeout(ASYNC_HOOK_TEST_TIMEOUT, async { + while started_count() < MAX_CONCURRENT_ASYNC_HOOKS { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("all available async hook slots should start"); + assert_eq!(started_count(), MAX_CONCURRENT_ASYNC_HOOKS); + + std::fs::write(release_path, "ready").expect("release running async hooks"); + for _ in 0..=MAX_CONCURRENT_ASYNC_HOOKS { + timeout(ASYNC_HOOK_TEST_TIMEOUT, results.recv()) + .await + .expect("waiting async hook should eventually finish") + .expect("result receiver should remain open"); + } + assert_eq!(started_count(), MAX_CONCURRENT_ASYNC_HOOKS + 1); + + runtime.shutdown().await; +} + +#[tokio::test] +async fn shutdown_aborts_in_flight_async_hooks_without_delivering_context() { + let temp = TempDir::new().expect("async test directory"); + let (runtime, results) = runtime(); + let started_path = temp.path().join("started"); + let release_path = temp.path().join("release"); + let handler = write_handler( + &temp, + &format!( + r#"import json +from pathlib import Path +import sys +import time + +json.load(sys.stdin) +Path(r"{started}").write_text("started", encoding="utf-8") +while not Path(r"{release}").exists(): + time.sleep(0.01) +print(json.dumps({{ + "hookSpecificOutput": {{ + "hookEventName": "UserPromptSubmit", + "additionalContext": "must not be delivered after shutdown" + }} +}})) +"#, + started = started_path.display(), + release = release_path.display(), + ), + ); + schedule(&runtime, handler, temp.path()).await; + + timeout(ASYNC_HOOK_TEST_TIMEOUT, async { + while !started_path.exists() { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("async hook should start before runtime shutdown"); + + runtime.shutdown().await; + drop(runtime); + std::fs::write(release_path, "ready").expect("release shutdown hook"); + assert!( + timeout(Duration::from_millis(150), results.recv()) + .await + .expect("shutdown should close the result channel") + .is_err(), + "shutdown must not deliver a late async result" + ); +} diff --git a/codex-rs/hooks/src/engine/discovery.rs b/codex-rs/hooks/src/engine/discovery.rs index 9ed6a903ef..3d5af4d60a 100644 --- a/codex-rs/hooks/src/engine/discovery.rs +++ b/codex-rs/hooks/src/engine/discovery.rs @@ -30,6 +30,7 @@ use crate::events::session_end::SESSION_END_DEFAULT_TIMEOUT_SEC; use crate::events::session_end::SESSION_END_MAX_TIMEOUT_SEC; use crate::output_spill::AdditionalContextLimit; use crate::output_spill::DEFAULT_HOOK_OUTPUT_TOKEN_LIMIT; +use codex_protocol::protocol::HookExecutionMode; use codex_protocol::protocol::HookHandlerType; use codex_protocol::protocol::HookSource; use codex_protocol::protocol::HookTrustStatus; @@ -473,14 +474,6 @@ fn append_matcher_groups( } else { command }; - if r#async && event_name != codex_protocol::protocol::HookEventName::SessionEnd - { - 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 {}", @@ -494,7 +487,14 @@ fn append_matcher_groups( source.path.as_path(), warnings, ); - if r#async { + let execution_mode = if r#async + && event_name != codex_protocol::protocol::HookEventName::SessionEnd + { + HookExecutionMode::Async + } else { + HookExecutionMode::Sync + }; + if r#async && execution_mode == HookExecutionMode::Sync { warnings.push(format!( "running async SessionEnd hook synchronously in {}", source.path.display() @@ -568,6 +568,7 @@ fn append_matcher_groups( { handlers.push(ConfiguredHandler { event_name, + execution_mode, matcher: matcher.map(ToOwned::to_owned), command, timeout_sec, @@ -956,6 +957,7 @@ mod tests { handlers, vec![ConfiguredHandler { event_name: HookEventName::UserPromptSubmit, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: None, command: "echo hello".to_string(), timeout_sec: 600, @@ -992,6 +994,7 @@ mod tests { handlers, vec![ConfiguredHandler { event_name: HookEventName::PreToolUse, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: Some("^Bash$".to_string()), command: "echo hello".to_string(), timeout_sec: 600, diff --git a/codex-rs/hooks/src/engine/dispatcher.rs b/codex-rs/hooks/src/engine/dispatcher.rs index 4855bbf333..2caabf04e3 100644 --- a/codex-rs/hooks/src/engine/dispatcher.rs +++ b/codex-rs/hooks/src/engine/dispatcher.rs @@ -73,7 +73,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, @@ -87,7 +87,7 @@ pub(crate) fn running_summary(handler: &ConfiguredHandler) -> HookRunSummary { } } -pub(crate) async fn execute_handlers( +pub(crate) async fn execute_handlers( runtime: &CommandHookRuntime, handlers: Vec, input_json: String, @@ -97,10 +97,20 @@ pub(crate) async fn execute_handlers( ) -> Vec> { let mut pending = FuturesUnordered::new(); for (configured_order, handler) in handlers.into_iter().enumerate() { + if handler.execution_mode == HookExecutionMode::Async { + runtime.schedule_async_hook( + handler, + input_json.clone(), + cwd.to_path_buf(), + turn_id.clone(), + parse, + ); + continue; + } let input_json = input_json.clone(); let turn_id = turn_id.clone(); pending.push(async move { - let result = run_command(runtime, &handler, configured_order, &input_json, cwd).await; + let result = run_command(runtime, &handler, &input_json, cwd).await; (configured_order, parse(&handler, result, turn_id)) }); } @@ -126,7 +136,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, @@ -232,6 +242,7 @@ mod tests { ) -> ConfiguredHandler { ConfiguredHandler { event_name, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: matcher.map(str::to_owned), command: command.to_string(), timeout_sec: 5, diff --git a/codex-rs/hooks/src/engine/mod.rs b/codex-rs/hooks/src/engine/mod.rs index cfa87a239c..38c94e8128 100644 --- a/codex-rs/hooks/src/engine/mod.rs +++ b/codex-rs/hooks/src/engine/mod.rs @@ -26,6 +26,7 @@ use crate::output_spill::AdditionalContextLimit; use codex_config::ConfigLayerStack; use codex_plugin::PluginHookSource; use codex_protocol::protocol::HookEventName; +use codex_protocol::protocol::HookExecutionMode; use codex_protocol::protocol::HookHandlerType; use codex_protocol::protocol::HookRunSummary; use codex_protocol::protocol::HookSource; @@ -45,6 +46,7 @@ pub(crate) struct CommandShell { #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct ConfiguredHandler { pub event_name: codex_protocol::protocol::HookEventName, + pub execution_mode: HookExecutionMode, pub matcher: Option, pub command: String, pub timeout_sec: u64, @@ -57,6 +59,11 @@ pub(crate) struct ConfiguredHandler { } impl ConfiguredHandler { + /// Only synchronous hooks can apply control effects. + pub(crate) fn can_apply_control_effects(&self) -> bool { + self.execution_mode == HookExecutionMode::Sync + } + pub fn run_id(&self) -> String { format!( "{}:{}:{}", @@ -107,7 +114,7 @@ pub struct HookListEntry { pub(crate) struct ClaudeHooksEngine { handlers: Vec, warnings: Vec, - command_runtime: CommandHookRuntime, + pub(crate) command_runtime: CommandHookRuntime, } impl ClaudeHooksEngine { @@ -168,7 +175,10 @@ impl ClaudeHooksEngine { Duration::from_secs( self.handlers .iter() - .filter(|handler| handler.event_name == HookEventName::PermissionRequest) + .filter(|handler| { + handler.event_name == HookEventName::PermissionRequest + && handler.execution_mode == HookExecutionMode::Sync + }) .map(|handler| handler.timeout_sec) .max() .unwrap_or_default(), @@ -206,14 +216,13 @@ impl ClaudeHooksEngine { &self, request: PostToolUseRequest, ) -> PostToolUseOutcome { - let session_id = request.session_id; let mut outcome = crate::events::post_tool_use::run(&self.handlers, &self.command_runtime, request).await; if let Some(feedback_message) = outcome.feedback_message.take() { outcome.feedback_message = Some( self.command_runtime .output_spiller() - .maybe_spill_text(session_id, feedback_message) + .maybe_spill_text(feedback_message) .await, ); } @@ -266,13 +275,12 @@ 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.command_runtime, request).await; outcome.continuation_fragments = self .command_runtime .output_spiller() - .maybe_spill_prompt_fragments(session_id, outcome.continuation_fragments) + .maybe_spill_prompt_fragments(outcome.continuation_fragments) .await; outcome } diff --git a/codex-rs/hooks/src/engine/mod_tests.rs b/codex-rs/hooks/src/engine/mod_tests.rs index 11612a984a..3fede5ac5e 100644 --- a/codex-rs/hooks/src/engine/mod_tests.rs +++ b/codex-rs/hooks/src/engine/mod_tests.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; use std::fs; use std::path::Path; +use std::time::Duration; use codex_config::AbsolutePathBuf; use codex_config::ConfigLayerEntry; @@ -20,6 +21,8 @@ use codex_config::TomlValue; use codex_plugin::PluginHookSource; use codex_plugin::PluginId; use codex_protocol::ThreadId; +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; @@ -31,12 +34,58 @@ use tempfile::tempdir; use super::ClaudeHooksEngine; use super::CommandHookRuntime; use super::CommandShell; +use super::ConfiguredHandler; use crate::events::pre_tool_use::PreToolUseRequest; fn cwd() -> AbsolutePathBuf { AbsolutePathBuf::current_dir().expect("current dir") } +fn command_runtime(shell: CommandShell) -> CommandHookRuntime { + let (result_sender, _result_receiver) = async_channel::unbounded(); + CommandHookRuntime::new(shell, ThreadId::new(), result_sender) +} + +#[test] +fn permission_request_timeout_only_counts_synchronous_handlers() { + let mut engine = ClaudeHooksEngine::new( + /*enabled*/ true, + /*bypass_hook_trust*/ false, + /*config_layer_stack*/ None, + Vec::new(), + Vec::new(), + command_runtime(CommandShell { + program: String::new(), + args: Vec::new(), + }), + ); + let synchronous_handler = ConfiguredHandler { + event_name: HookEventName::PermissionRequest, + execution_mode: HookExecutionMode::Sync, + matcher: None, + command: "echo synchronous permission hook".to_string(), + timeout_sec: 5, + status_message: None, + additional_context_limit: Default::default(), + source_path: cwd().join("hooks.json"), + source: HookSource::User, + display_order: 0, + env: HashMap::new(), + }; + let mut asynchronous_handler = synchronous_handler.clone(); + asynchronous_handler.execution_mode = HookExecutionMode::Async; + asynchronous_handler.timeout_sec = 600; + + engine.handlers = vec![synchronous_handler, asynchronous_handler.clone()]; + assert_eq!( + engine.max_permission_request_timeout(), + Duration::from_secs(5) + ); + + engine.handlers = vec![asynchronous_handler]; + assert_eq!(engine.max_permission_request_timeout(), Duration::ZERO); +} + fn managed_hooks_for_current_platform( managed_dir: impl AsRef, hooks: HookEventsToml, @@ -203,7 +252,7 @@ with Path(r"{log_path}").open("a", encoding="utf-8") as handle: Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -310,7 +359,7 @@ async fn requirements_managed_hooks_execute_windows_command_override() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -390,7 +439,7 @@ fn unknown_requirement_source_hooks_stay_managed() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -473,7 +522,7 @@ fn user_disablement_filters_non_managed_hooks_but_not_managed_hooks() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -539,7 +588,7 @@ fn user_disablement_does_not_filter_managed_layer_hooks() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -701,7 +750,7 @@ fn requirements_managed_hooks_load_when_managed_dir_is_missing() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -757,7 +806,7 @@ fn allow_managed_hooks_only_false_keeps_unmanaged_hooks() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -811,7 +860,7 @@ fn allow_managed_hooks_only_in_config_toml_does_not_enable_policy() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -881,7 +930,7 @@ fn allow_managed_hooks_only_skips_unmanaged_json_and_toml_hooks() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -920,7 +969,7 @@ fn allow_managed_hooks_only_skips_unmanaged_plugin_hooks() { Some(&config_layer_stack), plugin_hook_sources, Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -992,7 +1041,7 @@ fn allow_managed_hooks_only_keeps_managed_requirement_and_config_layer_hooks() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -1102,7 +1151,7 @@ fn discovers_hooks_from_json_and_toml_in_the_same_layer() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -1197,7 +1246,7 @@ fn profile_user_layers_load_shared_hooks_json_once() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -1271,7 +1320,7 @@ fn malformed_hooks_json_is_reported_as_startup_warning() { Some(&config_layer_stack), Vec::new(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -1343,7 +1392,7 @@ print(json.dumps({ Some(&config_layer_stack), plugin_hook_sources.clone(), Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -1463,7 +1512,7 @@ fn plugin_hook_sources_expand_plugin_placeholders() { Some(&config_layer_stack), plugin_hook_sources, Vec::new(), - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), @@ -1507,7 +1556,7 @@ fn plugin_hook_load_warnings_are_startup_warnings() { /*config_layer_stack*/ None, Vec::new(), vec!["failed plugin hook".to_string()], - CommandHookRuntime::new(CommandShell { + command_runtime(CommandShell { program: String::new(), args: Vec::new(), }), diff --git a/codex-rs/hooks/src/engine/output_parser.rs b/codex-rs/hooks/src/engine/output_parser.rs index b4f214146a..d4c71a172b 100644 --- a/codex-rs/hooks/src/engine/output_parser.rs +++ b/codex-rs/hooks/src/engine/output_parser.rs @@ -61,12 +61,6 @@ pub(crate) struct StopOutput { pub invalid_block_reason: Option, } -#[derive(Debug, Clone)] -pub(crate) struct PreCompactOutput { - pub universal: UniversalOutput, - pub invalid_reason: Option, -} - #[derive(Debug, Clone)] pub(crate) struct StatelessHookOutput { pub universal: UniversalOutput, @@ -238,10 +232,10 @@ pub(crate) fn parse_post_tool_use(stdout: &str) -> Option { }) } -pub(crate) fn parse_pre_compact(stdout: &str) -> Option { +pub(crate) fn parse_pre_compact(stdout: &str) -> Option { let wire: PreCompactCommandOutputWire = parse_json(stdout)?; let universal = UniversalOutput::from(wire.universal); - Some(PreCompactOutput { + Some(StatelessHookOutput { universal, invalid_reason: None, }) @@ -519,6 +513,25 @@ mod tests { use serde_json::json; use super::parse_permission_request; + use super::parse_user_prompt_submit; + + #[test] + fn structured_output_rejects_invalid_shapes_and_types() { + for stdout in [ + "[]", + r#"{"systemMessage":123}"#, + r#"{"hookSpecificOutput":{"hookEventName":"UserPromptSubmit","additionalContext":123}}"#, + r#"{"hookSpecificOutput":{"additionalContext":"missing event name"}}"#, + r#"{"hookSpecificOutput":{"hookEventName":"UserPromptSubmit","updatedInput":{}}}"#, + r#"{"unexpectedField":true}"#, + "{", + ] { + assert!( + parse_user_prompt_submit(stdout).is_none(), + "invalid structured output should fail: {stdout}", + ); + } + } #[test] fn permission_request_rejects_reserved_updated_input_field() { diff --git a/codex-rs/hooks/src/events/compact.rs b/codex-rs/hooks/src/events/compact.rs index b6765f9835..86bc360735 100644 --- a/codex-rs/hooks/src/events/compact.rs +++ b/codex-rs/hooks/src/events/compact.rs @@ -230,86 +230,13 @@ fn parse_pre_completed( run_result: CommandRunResult, turn_id: Option, ) -> dispatcher::ParsedHandler { - let mut entries = Vec::new(); - let mut status = HookRunStatus::Completed; - let mut should_stop = false; - let mut stop_reason = None; - - match run_result.error.as_deref() { - Some(error) => { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: error.to_string(), - }); - } - None => match run_result.exit_code { - Some(0) => { - let trimmed_stdout = run_result.stdout.trim(); - if trimmed_stdout.is_empty() { - } else if let Some(parsed) = output_parser::parse_pre_compact(&run_result.stdout) { - if let Some(system_message) = parsed.universal.system_message { - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Warning, - text: system_message, - }); - } - let _ = parsed.universal.suppress_output; - if !parsed.universal.continue_processing { - status = HookRunStatus::Stopped; - should_stop = true; - stop_reason = parsed.universal.stop_reason.clone(); - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Stop, - text: parsed - .universal - .stop_reason - .unwrap_or_else(|| "PreCompact hook stopped execution".to_string()), - }); - } else if let Some(invalid_reason) = parsed.invalid_reason { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: invalid_reason, - }); - } - } else if output_parser::looks_like_json(&run_result.stdout) { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: "hook returned invalid PreCompact hook JSON output".to_string(), - }); - } - } - Some(code) => { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: common::trimmed_non_empty(&run_result.stderr) - .unwrap_or_else(|| format!("hook exited with code {code}")), - }); - } - None => { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: "hook process terminated without an exit code".to_string(), - }); - } - }, - } - - dispatcher::ParsedHandler { - completed: HookCompletedEvent { - turn_id, - run: dispatcher::completed_summary(handler, &run_result, status, entries), - }, - data: CompactHandlerData { - should_stop, - stop_reason, - }, - completion_order: 0, - } + parse_completed( + handler, + run_result, + turn_id, + "PreCompact", + output_parser::parse_pre_compact, + ) } fn parse_post_completed( @@ -358,23 +285,24 @@ fn parse_completed( }); } let _ = parsed.universal.suppress_output; - if !parsed.universal.continue_processing { - status = HookRunStatus::Stopped; - should_stop = true; - stop_reason = parsed.universal.stop_reason.clone(); - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Stop, - text: parsed - .universal - .stop_reason - .unwrap_or_else(|| format!("{event_label} hook stopped execution")), - }); - } else if let Some(invalid_reason) = parsed.invalid_reason { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: invalid_reason, - }); + if handler.can_apply_control_effects() { + if !parsed.universal.continue_processing { + status = HookRunStatus::Stopped; + should_stop = true; + stop_reason = parsed.universal.stop_reason.clone(); + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Stop, + text: parsed.universal.stop_reason.unwrap_or_else(|| { + format!("{event_label} hook stopped execution") + }), + }); + } else if let Some(invalid_reason) = parsed.invalid_reason { + status = HookRunStatus::Failed; + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Error, + text: invalid_reason, + }); + } } } else if output_parser::looks_like_json(&run_result.stdout) { status = HookRunStatus::Failed; @@ -598,6 +526,7 @@ mod tests { fn handler(event_name: HookEventName) -> ConfiguredHandler { ConfiguredHandler { event_name, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: None, command: "python3 compact_hook.py".to_string(), timeout_sec: 5, diff --git a/codex-rs/hooks/src/events/permission_request.rs b/codex-rs/hooks/src/events/permission_request.rs index 91b50a2f2e..1c8e32dcbd 100644 --- a/codex-rs/hooks/src/events/permission_request.rs +++ b/codex-rs/hooks/src/events/permission_request.rs @@ -216,24 +216,26 @@ fn parse_completed( text: system_message, }); } - if let Some(invalid_reason) = parsed.invalid_reason { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: invalid_reason, - }); - } else if let Some(parsed_decision) = parsed.decision { - match parsed_decision { - output_parser::PermissionRequestDecision::Allow => { - decision = Some(PermissionRequestDecision::Allow); - } - output_parser::PermissionRequestDecision::Deny { message } => { - status = HookRunStatus::Blocked; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Feedback, - text: message.clone(), - }); - decision = Some(PermissionRequestDecision::Deny { message }); + if handler.can_apply_control_effects() { + if let Some(invalid_reason) = parsed.invalid_reason { + status = HookRunStatus::Failed; + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Error, + text: invalid_reason, + }); + } else if let Some(parsed_decision) = parsed.decision { + match parsed_decision { + output_parser::PermissionRequestDecision::Allow => { + decision = Some(PermissionRequestDecision::Allow); + } + output_parser::PermissionRequestDecision::Deny { message } => { + status = HookRunStatus::Blocked; + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Feedback, + text: message.clone(), + }); + decision = Some(PermissionRequestDecision::Deny { message }); + } } } } @@ -245,7 +247,7 @@ fn parse_completed( }); } } - Some(2) => { + Some(2) if handler.can_apply_control_effects() => { if let Some(message) = common::trimmed_non_empty(&run_result.stderr) { status = HookRunStatus::Blocked; entries.push(HookOutputEntry { diff --git a/codex-rs/hooks/src/events/post_tool_use.rs b/codex-rs/hooks/src/events/post_tool_use.rs index 2cac61f9ee..9db324c137 100644 --- a/codex-rs/hooks/src/events/post_tool_use.rs +++ b/codex-rs/hooks/src/events/post_tool_use.rs @@ -73,7 +73,6 @@ pub(crate) async fn run( runtime: &CommandHookRuntime, request: PostToolUseRequest, ) -> PostToolUseOutcome { - let session_id = request.session_id; let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); let matched = dispatcher::select_handlers_for_matcher_inputs( handlers, @@ -119,7 +118,7 @@ pub(crate) async fn run( ); let additional_contexts = runtime .output_spiller() - .maybe_spill_additional_contexts(session_id, additional_contexts) + .maybe_spill_additional_contexts(additional_contexts) .await; let should_block = results.iter().any(|result| result.data.should_block); let feedback_message = common::join_text_chunks( @@ -198,8 +197,8 @@ fn parse_completed( text: system_message, }); } - if parsed.invalid_reason.is_none() - && parsed.invalid_block_reason.is_none() + if (!handler.can_apply_control_effects() + || parsed.invalid_reason.is_none() && parsed.invalid_block_reason.is_none()) && let Some(additional_context) = parsed.additional_context { common::append_additional_context( @@ -209,43 +208,44 @@ fn parse_completed( additional_context, ); } - if !parsed.universal.continue_processing { - status = HookRunStatus::Stopped; - let stop_text = parsed - .universal - .stop_reason - .unwrap_or_else(|| "PostToolUse hook stopped execution".to_string()); - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Stop, - text: stop_text.clone(), - }); - let model_feedback = parsed - .reason - .as_deref() - .and_then(common::trimmed_non_empty) - .unwrap_or(stop_text); - feedback_messages_for_model.push(model_feedback); - } else if let Some(invalid_reason) = parsed.invalid_reason { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: invalid_reason, - }); - } else if let Some(invalid_block_reason) = parsed.invalid_block_reason { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: invalid_block_reason, - }); - } else if parsed.should_block { - status = HookRunStatus::Blocked; - should_block = true; - if let Some(reason) = parsed.reason { - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Feedback, - text: reason.clone(), + if handler.can_apply_control_effects() { + if !parsed.universal.continue_processing { + status = HookRunStatus::Stopped; + let stop_text = parsed.universal.stop_reason.unwrap_or_else(|| { + "PostToolUse hook stopped execution".to_string() }); - feedback_messages_for_model.push(reason); + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Stop, + text: stop_text.clone(), + }); + let model_feedback = parsed + .reason + .as_deref() + .and_then(common::trimmed_non_empty) + .unwrap_or(stop_text); + feedback_messages_for_model.push(model_feedback); + } else if let Some(invalid_reason) = parsed.invalid_reason { + status = HookRunStatus::Failed; + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Error, + text: invalid_reason, + }); + } else if let Some(invalid_block_reason) = parsed.invalid_block_reason { + status = HookRunStatus::Failed; + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Error, + text: invalid_block_reason, + }); + } else if parsed.should_block { + status = HookRunStatus::Blocked; + should_block = true; + if let Some(reason) = parsed.reason { + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Feedback, + text: reason.clone(), + }); + feedback_messages_for_model.push(reason); + } } } } else if output_parser::looks_like_json(&run_result.stdout) { @@ -256,7 +256,7 @@ fn parse_completed( }); } } - Some(2) => { + Some(2) if handler.can_apply_control_effects() => { if let Some(reason) = common::trimmed_non_empty(&run_result.stderr) { status = HookRunStatus::Blocked; should_block = true; @@ -319,6 +319,7 @@ fn serialization_failure_outcome(hook_events: Vec) -> PostTo mod tests { use codex_protocol::ThreadId; 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; @@ -408,13 +409,10 @@ mod tests { #[test] fn unsupported_updated_mcp_tool_output_fails_open() { + let stdout = r#"{"hookSpecificOutput":{"hookEventName":"PostToolUse","updatedMCPToolOutput":{"ok":true},"additionalContext":"preserved"}}"#; let parsed = parse_completed( &handler(), - run_result( - Some(0), - r#"{"hookSpecificOutput":{"hookEventName":"PostToolUse","updatedMCPToolOutput":{"ok":true}}}"#, - "", - ), + run_result(Some(0), stdout, ""), Some("turn-1".to_string()), ); @@ -434,6 +432,22 @@ mod tests { text: "PostToolUse hook returned unsupported updatedMCPToolOutput".to_string(), }] ); + + let mut async_handler = handler(); + async_handler.execution_mode = HookExecutionMode::Async; + let parsed = parse_completed( + &async_handler, + run_result(Some(0), stdout, ""), + Some("turn-1".to_string()), + ); + assert_eq!(parsed.completed.run.status, HookRunStatus::Completed); + assert_eq!( + parsed.completed.run.entries, + vec![HookOutputEntry { + kind: HookOutputEntryKind::Context, + text: "preserved".to_string(), + }], + ); } #[test] @@ -574,6 +588,7 @@ mod tests { fn handler() -> ConfiguredHandler { ConfiguredHandler { event_name: HookEventName::PostToolUse, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: Some("^Bash$".to_string()), command: "python3 post_tool_use_hook.py".to_string(), timeout_sec: 5, diff --git a/codex-rs/hooks/src/events/pre_tool_use.rs b/codex-rs/hooks/src/events/pre_tool_use.rs index 273ff9b44a..64394c7a23 100644 --- a/codex-rs/hooks/src/events/pre_tool_use.rs +++ b/codex-rs/hooks/src/events/pre_tool_use.rs @@ -74,7 +74,6 @@ pub(crate) async fn run( runtime: &CommandHookRuntime, request: PreToolUseRequest, ) -> PreToolUseOutcome { - let session_id = request.session_id; let matcher_inputs = common::matcher_inputs(&request.tool_name, &request.matcher_aliases); let matched = dispatcher::select_handlers_for_matcher_inputs( handlers, @@ -125,7 +124,7 @@ pub(crate) async fn run( ); let additional_contexts = runtime .output_spiller() - .maybe_spill_additional_contexts(session_id, additional_contexts) + .maybe_spill_additional_contexts(additional_contexts) .await; let updated_input = if should_block { None @@ -222,22 +221,24 @@ fn parse_completed( text: system_message, }); } - if let Some(invalid_reason) = parsed.invalid_reason { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: invalid_reason, - }); - } else { - if let Some(additional_context) = parsed.additional_context { - common::append_additional_context( - &mut entries, - &mut additional_contexts_for_model, - handler, - additional_context, - ); - } - if let Some(reason) = parsed.block_reason { + if (!handler.can_apply_control_effects() || parsed.invalid_reason.is_none()) + && let Some(additional_context) = parsed.additional_context + { + common::append_additional_context( + &mut entries, + &mut additional_contexts_for_model, + handler, + additional_context, + ); + } + if handler.can_apply_control_effects() { + if let Some(invalid_reason) = parsed.invalid_reason { + status = HookRunStatus::Failed; + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Error, + text: invalid_reason, + }); + } else if let Some(reason) = parsed.block_reason { status = HookRunStatus::Blocked; should_block = true; block_reason = Some(reason.clone()); @@ -245,8 +246,7 @@ fn parse_completed( kind: HookOutputEntryKind::Feedback, text: reason, }); - } - if !should_block { + } else { updated_input = parsed.updated_input; } } @@ -258,7 +258,7 @@ fn parse_completed( }); } } - Some(2) => { + Some(2) if handler.can_apply_control_effects() => { if let Some(reason) = common::trimmed_non_empty(&run_result.stderr) { status = HookRunStatus::Blocked; should_block = true; @@ -323,6 +323,7 @@ fn serialization_failure_outcome(hook_events: Vec) -> PreToo mod tests { use codex_protocol::ThreadId; 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; @@ -440,13 +441,10 @@ mod tests { #[test] fn permission_decision_allow_without_updated_input_fails_open() { + let stdout = r#"{"hookSpecificOutput":{"hookEventName":"PreToolUse","permissionDecision":"allow","additionalContext":"preserved"}}"#; let parsed = parse_completed( &handler(), - run_result( - Some(0), - r#"{"hookSpecificOutput":{"hookEventName":"PreToolUse","permissionDecision":"allow"}}"#, - "", - ), + run_result(Some(0), stdout, ""), Some("turn-1".to_string()), ); @@ -467,6 +465,22 @@ mod tests { text: "PreToolUse hook returned unsupported permissionDecision:allow".to_string(), }] ); + + let mut async_handler = handler(); + async_handler.execution_mode = HookExecutionMode::Async; + let parsed = parse_completed( + &async_handler, + run_result(Some(0), stdout, ""), + Some("turn-1".to_string()), + ); + assert_eq!(parsed.completed.run.status, HookRunStatus::Completed); + assert_eq!( + parsed.completed.run.entries, + vec![HookOutputEntry { + kind: HookOutputEntryKind::Context, + text: "preserved".to_string(), + }], + ); } #[test] @@ -758,6 +772,7 @@ mod tests { fn handler() -> ConfiguredHandler { ConfiguredHandler { event_name: HookEventName::PreToolUse, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: Some("^Bash$".to_string()), command: "echo hook".to_string(), timeout_sec: 5, diff --git a/codex-rs/hooks/src/events/session_end_tests.rs b/codex-rs/hooks/src/events/session_end_tests.rs index f3228f9dc9..1b9a44682f 100644 --- a/codex-rs/hooks/src/events/session_end_tests.rs +++ b/codex-rs/hooks/src/events/session_end_tests.rs @@ -61,6 +61,7 @@ fn session_end_ignores_successful_output() { fn handler(matcher: Option<&str>) -> ConfiguredHandler { ConfiguredHandler { event_name: HookEventName::SessionEnd, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: matcher.map(str::to_string), command: "echo hook".to_string(), timeout_sec: 2, diff --git a/codex-rs/hooks/src/events/session_start.rs b/codex-rs/hooks/src/events/session_start.rs index 2386354edc..2a74fad014 100644 --- a/codex-rs/hooks/src/events/session_start.rs +++ b/codex-rs/hooks/src/events/session_start.rs @@ -112,7 +112,6 @@ pub(crate) async fn run( request: SessionStartRequest, turn_id: Option, ) -> SessionStartOutcome { - let session_id = request.session_id; let matched = dispatcher::select_handlers( handlers, request.target.event_name(), @@ -203,7 +202,7 @@ pub(crate) async fn run( ); let additional_contexts = runtime .output_spiller() - .maybe_spill_additional_contexts(session_id, additional_contexts) + .maybe_spill_additional_contexts(additional_contexts) .await; SessionStartOutcome { @@ -269,7 +268,8 @@ fn parse_completed( ); } let _ = parsed.universal.suppress_output; - if handler.event_name == HookEventName::SessionStart + if handler.can_apply_control_effects() + && handler.event_name == HookEventName::SessionStart && !parsed.universal.continue_processing { status = HookRunStatus::Stopped; @@ -541,6 +541,7 @@ mod tests { fn handler_for(event_name: HookEventName) -> ConfiguredHandler { ConfiguredHandler { event_name, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: None, command: "echo hook".to_string(), timeout_sec: 600, diff --git a/codex-rs/hooks/src/events/stop.rs b/codex-rs/hooks/src/events/stop.rs index 64e6e04473..408e78fb6c 100644 --- a/codex-rs/hooks/src/events/stop.rs +++ b/codex-rs/hooks/src/events/stop.rs @@ -244,25 +244,26 @@ fn parse_completed( }); } let _ = parsed.universal.suppress_output; - if !parsed.universal.continue_processing { - status = HookRunStatus::Stopped; - should_stop = true; - stop_reason = parsed.universal.stop_reason.clone(); - if let Some(stop_reason_text) = parsed.universal.stop_reason { + if handler.can_apply_control_effects() { + if !parsed.universal.continue_processing { + status = HookRunStatus::Stopped; + should_stop = true; + stop_reason = parsed.universal.stop_reason.clone(); + if let Some(stop_reason_text) = parsed.universal.stop_reason { + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Stop, + text: stop_reason_text, + }); + } + } else if let Some(invalid_block_reason) = parsed.invalid_block_reason { + status = HookRunStatus::Failed; entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Stop, - text: stop_reason_text, + kind: HookOutputEntryKind::Error, + text: invalid_block_reason, }); - } - } else if let Some(invalid_block_reason) = parsed.invalid_block_reason { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: invalid_block_reason, - }); - } else if parsed.should_block { - if let Some(reason) = - parsed.reason.as_deref().and_then(common::trimmed_non_empty) + } else if parsed.should_block + && let Some(reason) = + parsed.reason.as_deref().and_then(common::trimmed_non_empty) { status = HookRunStatus::Blocked; should_block = true; @@ -272,20 +273,11 @@ fn parse_completed( kind: HookOutputEntryKind::Feedback, text: reason, }); - } else { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: match hook_event_name { - HookEventName::Stop => "Stop hook returned decision:block without a non-empty reason", - HookEventName::SubagentStop => "SubagentStop hook returned decision:block without a non-empty reason", - _ => unreachable!("validated stop hook event"), - } - .to_string(), - }); } } - } else { + } else if handler.can_apply_control_effects() + || output_parser::looks_like_json(&run_result.stdout) + { status = HookRunStatus::Failed; entries.push(HookOutputEntry { kind: HookOutputEntryKind::Error, @@ -300,7 +292,7 @@ fn parse_completed( }); } } - Some(2) => { + Some(2) if handler.can_apply_control_effects() => { if let Some(reason) = common::trimmed_non_empty(&run_result.stderr) { status = HookRunStatus::Blocked; should_block = true; @@ -420,6 +412,7 @@ fn serialization_failure_outcome(hook_events: Vec) -> StopOu #[cfg(test)] mod tests { 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; @@ -480,6 +473,16 @@ mod tests { text: "Stop hook returned decision:block without a non-empty reason".to_string(), }] ); + + let mut async_handler = handler(); + async_handler.execution_mode = HookExecutionMode::Async; + let parsed = parse_completed( + &async_handler, + run_result(Some(0), r#"{"decision":"block"}"#, ""), + Some("turn-1".to_string()), + ); + assert_eq!(parsed.completed.run.status, HookRunStatus::Completed); + assert_eq!(parsed.completed.run.entries, Vec::new()); } #[test] @@ -588,6 +591,16 @@ mod tests { text: "hook returned invalid stop hook JSON output".to_string(), }] ); + + let mut async_handler = handler(); + async_handler.execution_mode = HookExecutionMode::Async; + let parsed = parse_completed( + &async_handler, + run_result(Some(0), "not json", ""), + Some("turn-1".to_string()), + ); + assert_eq!(parsed.completed.run.status, HookRunStatus::Completed); + assert_eq!(parsed.completed.run.entries, Vec::new()); } #[test] @@ -631,6 +644,7 @@ mod tests { fn handler() -> ConfiguredHandler { ConfiguredHandler { event_name: HookEventName::Stop, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: None, command: "echo hook".to_string(), timeout_sec: 600, diff --git a/codex-rs/hooks/src/events/user_prompt_submit.rs b/codex-rs/hooks/src/events/user_prompt_submit.rs index 5cec1d63e3..f6b0a36fbf 100644 --- a/codex-rs/hooks/src/events/user_prompt_submit.rs +++ b/codex-rs/hooks/src/events/user_prompt_submit.rs @@ -66,7 +66,6 @@ pub(crate) async fn run( runtime: &CommandHookRuntime, request: UserPromptSubmitRequest, ) -> UserPromptSubmitOutcome { - let session_id = request.session_id; let matched = dispatcher::select_handlers( handlers, HookEventName::UserPromptSubmit, @@ -125,7 +124,7 @@ pub(crate) async fn run( ); let additional_contexts = runtime .output_spiller() - .maybe_spill_additional_contexts(session_id, additional_contexts) + .maybe_spill_additional_contexts(additional_contexts) .await; UserPromptSubmitOutcome { @@ -168,7 +167,8 @@ fn parse_completed( text: system_message, }); } - if parsed.invalid_block_reason.is_none() + if (!handler.can_apply_control_effects() + || parsed.invalid_block_reason.is_none()) && let Some(additional_context) = parsed.additional_context { common::append_additional_context( @@ -179,31 +179,33 @@ fn parse_completed( ); } let _ = parsed.universal.suppress_output; - if !parsed.universal.continue_processing { - status = HookRunStatus::Stopped; - should_stop = true; - stop_reason = parsed.universal.stop_reason.clone(); - if let Some(stop_reason_text) = parsed.universal.stop_reason { + if handler.can_apply_control_effects() { + if !parsed.universal.continue_processing { + status = HookRunStatus::Stopped; + should_stop = true; + stop_reason = parsed.universal.stop_reason.clone(); + if let Some(stop_reason_text) = parsed.universal.stop_reason { + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Stop, + text: stop_reason_text, + }); + } + } else if let Some(invalid_block_reason) = parsed.invalid_block_reason { + status = HookRunStatus::Failed; entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Stop, - text: stop_reason_text, - }); - } - } else if let Some(invalid_block_reason) = parsed.invalid_block_reason { - status = HookRunStatus::Failed; - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Error, - text: invalid_block_reason, - }); - } else if parsed.should_block { - status = HookRunStatus::Blocked; - should_stop = true; - stop_reason = parsed.reason.clone(); - if let Some(reason) = parsed.reason { - entries.push(HookOutputEntry { - kind: HookOutputEntryKind::Feedback, - text: reason, + kind: HookOutputEntryKind::Error, + text: invalid_block_reason, }); + } else if parsed.should_block { + status = HookRunStatus::Blocked; + should_stop = true; + stop_reason = parsed.reason.clone(); + if let Some(reason) = parsed.reason { + entries.push(HookOutputEntry { + kind: HookOutputEntryKind::Feedback, + text: reason, + }); + } } } } else if output_parser::looks_like_json(&run_result.stdout) { @@ -222,7 +224,7 @@ fn parse_completed( ); } } - Some(2) => { + Some(2) if handler.can_apply_control_effects() => { if let Some(reason) = common::trimmed_non_empty(&run_result.stderr) { status = HookRunStatus::Blocked; should_stop = true; @@ -284,6 +286,7 @@ fn serialization_failure_outcome(hook_events: Vec) -> UserPr #[cfg(test)] mod tests { 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; @@ -377,13 +380,10 @@ mod tests { #[test] fn claude_block_decision_requires_reason() { + let stdout = r#"{"decision":"block","hookSpecificOutput":{"hookEventName":"UserPromptSubmit","additionalContext":"do not inject"}}"#; let parsed = parse_completed( &handler(), - run_result( - Some(0), - r#"{"decision":"block","hookSpecificOutput":{"hookEventName":"UserPromptSubmit","additionalContext":"do not inject"}}"#, - "", - ), + run_result(Some(0), stdout, ""), Some("turn-1".to_string()), ); @@ -404,6 +404,22 @@ mod tests { .to_string(), }] ); + + let mut async_handler = handler(); + async_handler.execution_mode = HookExecutionMode::Async; + let parsed = parse_completed( + &async_handler, + run_result(Some(0), stdout, ""), + Some("turn-1".to_string()), + ); + assert_eq!(parsed.completed.run.status, HookRunStatus::Completed); + assert_eq!( + parsed.completed.run.entries, + vec![HookOutputEntry { + kind: HookOutputEntryKind::Context, + text: "do not inject".to_string(), + }], + ); } #[test] @@ -430,11 +446,22 @@ mod tests { text: "blocked by policy".to_string(), }] ); + + let mut async_handler = handler(); + async_handler.execution_mode = HookExecutionMode::Async; + let parsed = parse_completed( + &async_handler, + run_result(Some(2), "", "blocked by policy\n"), + Some("turn-1".to_string()), + ); + assert_eq!(parsed.completed.run.status, HookRunStatus::Failed); + assert!(!parsed.data.should_stop); } fn handler() -> ConfiguredHandler { ConfiguredHandler { event_name: HookEventName::UserPromptSubmit, + execution_mode: codex_protocol::protocol::HookExecutionMode::Sync, matcher: None, command: "echo hook".to_string(), timeout_sec: 5, diff --git a/codex-rs/hooks/src/output_spill.rs b/codex-rs/hooks/src/output_spill.rs index 5297d9d256..b60ac97399 100644 --- a/codex-rs/hooks/src/output_spill.rs +++ b/codex-rs/hooks/src/output_spill.rs @@ -42,10 +42,11 @@ pub(crate) struct HookOutputSpiller { } impl HookOutputSpiller { - pub(crate) fn new() -> Self { + pub(crate) fn new(thread_id: ThreadId) -> Self { Self { output_dir: AbsolutePathBuf::resolve_path_against_base(std::env::temp_dir(), "/") - .join(HOOK_OUTPUTS_DIR), + .join(HOOK_OUTPUTS_DIR) + .join(thread_id.to_string()), } } @@ -55,14 +56,13 @@ impl HookOutputSpiller { /// `/hook_outputs//` /// and replaced with the same head/tail preview style used for other truncated /// output, plus a path back to the preserved full text. - pub(crate) async fn maybe_spill_text(&self, thread_id: ThreadId, text: String) -> String { - self.maybe_spill_text_with_limit(thread_id, text, AdditionalContextLimit::default()) + pub(crate) async fn maybe_spill_text(&self, text: String) -> String { + self.maybe_spill_text_with_limit(text, AdditionalContextLimit::default()) .await } async fn maybe_spill_text_with_limit( &self, - thread_id: ThreadId, text: String, limit: AdditionalContextLimit, ) -> String { @@ -71,7 +71,7 @@ impl HookOutputSpiller { return text; } - let path = hook_output_path(&self.output_dir, thread_id); + let path = self.output_dir.join(format!("{}.txt", Uuid::new_v4())); if let Some(parent) = path.parent() && let Err(err) = fs::create_dir_all(parent.as_ref()).await { @@ -92,13 +92,12 @@ impl HookOutputSpiller { pub(crate) async fn maybe_spill_additional_contexts( &self, - thread_id: ThreadId, contexts: Vec, ) -> Vec { let mut spilled = Vec::with_capacity(contexts.len()); for context in contexts { spilled.push( - self.maybe_spill_text_with_limit(thread_id, context.text, context.limit) + self.maybe_spill_text_with_limit(context.text, context.limit) .await, ); } @@ -107,13 +106,12 @@ impl HookOutputSpiller { pub(crate) async fn maybe_spill_prompt_fragments( &self, - thread_id: ThreadId, fragments: Vec, ) -> Vec { let mut spilled = Vec::with_capacity(fragments.len()); for fragment in fragments { spilled.push(HookPromptFragment { - text: self.maybe_spill_text(thread_id, fragment.text).await, + text: self.maybe_spill_text(fragment.text).await, hook_run_id: fragment.hook_run_id, }); } @@ -121,12 +119,6 @@ impl HookOutputSpiller { } } -fn hook_output_path(output_dir: &AbsolutePathBuf, thread_id: ThreadId) -> AbsolutePathBuf { - output_dir - .join(thread_id.to_string()) - .join(format!("{}.txt", Uuid::new_v4())) -} - /// Builds the model-visible replacement for a spilled hook output. /// /// The path footer is budgeted before truncation so adding the recovery path diff --git a/codex-rs/hooks/src/output_spill_tests.rs b/codex-rs/hooks/src/output_spill_tests.rs index 253d37c841..6b72c956ce 100644 --- a/codex-rs/hooks/src/output_spill_tests.rs +++ b/codex-rs/hooks/src/output_spill_tests.rs @@ -3,18 +3,28 @@ use anyhow::Context; use anyhow::Result; use tempfile::tempdir; +#[test] +fn hook_output_spiller_is_scoped_to_its_thread() { + let thread_id = ThreadId::new(); + let spiller = HookOutputSpiller::new(thread_id); + + assert_eq!( + spiller.output_dir, + AbsolutePathBuf::resolve_path_against_base(std::env::temp_dir(), "/") + .join(HOOK_OUTPUTS_DIR) + .join(thread_id.to_string()) + ); +} + #[tokio::test] async fn small_hook_output_remains_inline() -> Result<()> { let dir = tempdir()?; let output_dir = AbsolutePathBuf::from_absolute_path(dir.path())?.join(HOOK_OUTPUTS_DIR); - let thread_id = ThreadId::new(); let spiller = HookOutputSpiller { output_dir: output_dir.clone(), }; - let output = spiller - .maybe_spill_text(thread_id, "short".to_string()) - .await; + let output = spiller.maybe_spill_text("short".to_string()).await; assert_eq!(output, "short"); assert!(!output_dir.exists()); @@ -28,9 +38,7 @@ async fn large_hook_output_spills_to_file() -> Result<()> { let output_dir = AbsolutePathBuf::from_absolute_path(dir.path())?.join(HOOK_OUTPUTS_DIR); let spiller = HookOutputSpiller { output_dir }; - let output = spiller - .maybe_spill_text(ThreadId::new(), text.clone()) - .await; + let output = spiller.maybe_spill_text(text.clone()).await; assert!(output.contains("tokens truncated")); let path = output @@ -50,23 +58,20 @@ async fn additional_contexts_apply_limits_individually() -> Result<()> { let output_dir = AbsolutePathBuf::from_absolute_path(dir.path())?.join(HOOK_OUTPUTS_DIR); let spiller = HookOutputSpiller { output_dir }; let output = spiller - .maybe_spill_additional_contexts( - ThreadId::new(), - vec![ - AdditionalContext { - text: limited_text.clone(), - limit: AdditionalContextLimit::from_config(Some(1)), - }, - AdditionalContext { - text: unlimited_text.clone(), - limit: AdditionalContextLimit::from_config(Some(0)), - }, - AdditionalContext { - text: unlimited_text.clone(), - limit: AdditionalContextLimit::from_config(Some(usize::MAX)), - }, - ], - ) + .maybe_spill_additional_contexts(vec![ + AdditionalContext { + text: limited_text.clone(), + limit: AdditionalContextLimit::from_config(Some(1)), + }, + AdditionalContext { + text: unlimited_text.clone(), + limit: AdditionalContextLimit::from_config(Some(0)), + }, + AdditionalContext { + text: unlimited_text.clone(), + limit: AdditionalContextLimit::from_config(Some(usize::MAX)), + }, + ]) .await; let [limited_output, zero_limit_output, high_limit_output] = output.as_slice() else { panic!("expected one output for each additional context"); diff --git a/codex-rs/hooks/src/registry.rs b/codex-rs/hooks/src/registry.rs index 2506782d24..3d0e47584e 100644 --- a/codex-rs/hooks/src/registry.rs +++ b/codex-rs/hooks/src/registry.rs @@ -1,8 +1,3 @@ -use codex_config::ConfigLayerStack; -use codex_plugin::PluginHookSource; -use std::time::Duration; -use tokio::process::Command; - use crate::engine::ClaudeHooksEngine; use crate::engine::CommandShell; use crate::engine::HookListEntry; @@ -29,6 +24,12 @@ use crate::types::Hook; use crate::types::HookEvent; use crate::types::HookPayload; use crate::types::HookResponse; +use async_channel::Receiver; +use codex_config::ConfigLayerStack; +use codex_plugin::PluginHookSource; +use codex_protocol::ThreadId; +use std::time::Duration; +use tokio::process::Command; #[derive(Default, Clone)] pub struct HooksConfig { @@ -54,21 +55,37 @@ pub struct Hooks { engine: ClaudeHooksEngine, } -impl Default for Hooks { - fn default() -> Self { - Self::new(HooksConfig::default()) - } -} - impl Hooks { - pub fn new(config: HooksConfig) -> Self { + /// Bind this session's hook runtime and output files to its thread. + pub fn new( + config: HooksConfig, + thread_id: ThreadId, + ) -> (Self, Receiver) { + let (result_sender, result_receiver) = async_channel::unbounded(); + let hooks = Self::from_config(config, |shell| { + CommandHookRuntime::new(shell, thread_id, result_sender) + }); + (hooks, result_receiver) + } + + /// Preserve in-flight background hooks while applying a refreshed configuration. + pub fn reconfigured(&self, config: HooksConfig) -> Self { + Self::from_config(config, |shell| { + self.engine.command_runtime.reconfigured(shell) + }) + } + + fn from_config( + config: HooksConfig, + build_runtime: impl FnOnce(CommandShell) -> CommandHookRuntime, + ) -> Self { let after_agent = config .legacy_notify_argv .filter(|argv| !argv.is_empty() && !argv[0].is_empty()) .map(crate::notify_hook) .into_iter() .collect(); - let command_runtime = CommandHookRuntime::new(CommandShell { + let command_runtime = build_runtime(CommandShell { program: config.shell_program.unwrap_or_default(), args: config.shell_args, }); @@ -86,6 +103,11 @@ impl Hooks { } } + /// Abort and join outstanding async hooks during session shutdown. + pub async fn shutdown(&self) { + self.engine.command_runtime.shutdown().await; + } + pub fn startup_warnings(&self) -> &[String] { self.engine.warnings() }