openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-12 21:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enable async hook delivery on accepted turns

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