diff --git a/codex-rs/hooks/Cargo.toml b/codex-rs/hooks/Cargo.toml index 1bd7f455f2..39151d4dfa 100644 --- a/codex-rs/hooks/Cargo.toml +++ b/codex-rs/hooks/Cargo.toml @@ -25,7 +25,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", "rt", "time"] } tracing = { workspace = true } uuid = { workspace = true, features = ["v4"] } diff --git a/codex-rs/hooks/src/engine/async_command.rs b/codex-rs/hooks/src/engine/async_command.rs index 2e9dbc9a44..3e1a0cd54b 100644 --- a/codex-rs/hooks/src/engine/async_command.rs +++ b/codex-rs/hooks/src/engine/async_command.rs @@ -5,8 +5,8 @@ //! snapshots the ready completions at turn entry, before any hooks for that //! turn run, so output that completes during the turn cannot race into it. //! -//! The runtime survives hook configuration refreshes and bounds concurrent -//! commands, queued completions, and the amount delivered to one request. +//! The runtime survives hook configuration refreshes and delivers all output +//! that was ready when an accepted user turn began. use std::collections::BTreeMap; use std::path::PathBuf; @@ -15,7 +15,6 @@ use std::sync::Mutex; use std::sync::MutexGuard; use codex_protocol::ThreadId; -use codex_utils_output_truncation::approx_token_count; use tokio::task::JoinSet; use super::CommandShell; @@ -24,11 +23,6 @@ use super::command_runner::run_command; use super::output_parser; use crate::output_spill::HookOutputSpiller; -const MAX_QUEUED_COMPLETIONS: usize = 64; -const MAX_IN_FLIGHT_COMMANDS: usize = 32; -const MAX_DELIVERED_COMPLETIONS_PER_REQUEST: usize = 8; -const MAX_DELIVERED_OUTPUT_TOKENS_PER_REQUEST: usize = 10_000; - /// Informational async hook output ready to be recorded for a model request. #[derive(Debug, Default, PartialEq, Eq)] pub struct AsyncHookDelivery { @@ -133,15 +127,6 @@ impl AsyncCommandRuntime { if state.shutting_down { return; } - if state.tasks.len() >= MAX_IN_FLIGHT_COMMANDS { - tracing::warn!( - event_name = ?handler.event_name, - hook_source = ?handler.source, - limit = MAX_IN_FLIGHT_COMMANDS, - "skipping async hook command after reaching the session concurrency limit" - ); - return; - } let launch_sequence = state.next_launch_sequence; state.next_launch_sequence = state.next_launch_sequence.saturating_add(1); @@ -177,49 +162,14 @@ impl AsyncCommandRuntime { return; } state.completions.insert(launch_sequence, output); - if state.completions.len() > MAX_QUEUED_COMPLETIONS - && let Some((oldest, _)) = state.completions.pop_first() - { - tracing::warn!( - launch_sequence = oldest, - "dropping queued async hook output after reaching the session limit" - ); - } }); } /// Drains output that was ready when an accepted user turn began. - /// - /// Delivery is bounded; remaining output stays queued for later accepted - /// user turns. fn drain_for_accepted_turn(&self, ready: Vec) -> AsyncHookDelivery { let mut state = self.inner.lock_state(); - let mut eligible = Vec::new(); - let mut output_tokens = 0usize; - for launch_sequence in ready { - let Some(completion) = state.completions.get(&launch_sequence) else { - continue; - }; - if eligible.len() >= MAX_DELIVERED_COMPLETIONS_PER_REQUEST { - break; - } - let completion_output_tokens = completion - .additional_context - .as_deref() - .into_iter() - .chain(completion.system_message.as_deref()) - .map(approx_token_count) - .fold(0, usize::saturating_add); - if output_tokens.saturating_add(completion_output_tokens) - > MAX_DELIVERED_OUTPUT_TOKENS_PER_REQUEST - { - break; - } - eligible.push(launch_sequence); - output_tokens = output_tokens.saturating_add(completion_output_tokens); - } let mut delivery = AsyncHookDelivery::default(); - for launch_sequence in eligible { + for launch_sequence in ready { let Some(completion) = state.completions.remove(&launch_sequence) else { continue; }; diff --git a/codex-rs/hooks/src/engine/async_command_tests.rs b/codex-rs/hooks/src/engine/async_command_tests.rs index e9b03f613f..2b97a8f2cc 100644 --- a/codex-rs/hooks/src/engine/async_command_tests.rs +++ b/codex-rs/hooks/src/engine/async_command_tests.rs @@ -2,16 +2,7 @@ use pretty_assertions::assert_eq; use super::AsyncCommandRuntime; use super::AsyncHookDelivery; -use super::MAX_DELIVERED_OUTPUT_TOKENS_PER_REQUEST; -use super::MAX_IN_FLIGHT_COMMANDS; -use crate::engine::CommandShell; -use crate::engine::ConfiguredHandler; use crate::engine::output_parser::AsyncInformationalOutput; -use codex_protocol::ThreadId; -use codex_protocol::protocol::HookEventName; -use codex_protocol::protocol::HookSource; -use codex_utils_absolute_path::AbsolutePathBuf; -use std::collections::HashMap; enum TestOutput<'a> { AdditionalContext(&'a str), @@ -69,65 +60,23 @@ fn unfinished_earlier_launch_does_not_block_ready_output() { assert_eq!(delivery, context_delivery("ready")); } -#[tokio::test] -async fn launch_is_skipped_at_session_concurrency_limit() { - let runtime = AsyncCommandRuntime::new(); - { - let mut state = runtime.inner.lock_state(); - for _ in 0..MAX_IN_FLIGHT_COMMANDS { - state.tasks.spawn(std::future::pending()); - } - } - - runtime.spawn( - CommandShell { - program: String::new(), - args: Vec::new(), - }, - ConfiguredHandler { - event_name: HookEventName::PreToolUse, - matcher: None, - command: "exit 0".to_string(), - timeout_sec: 5, - status_message: None, - source_path: AbsolutePathBuf::current_dir().expect("current dir"), - source: HookSource::User, - display_order: 0, - env: HashMap::new(), - }, - /*configured_order*/ 0, - String::new(), - std::env::current_dir().expect("current dir"), - ThreadId::new(), - ); - - assert_eq!(runtime.inner.lock_state().next_launch_sequence, 0); - runtime.shutdown().await; -} - #[test] -fn shared_output_budget_leaves_remaining_completions_queued() { +fn all_snapshotted_completions_are_delivered_together() { let runtime = AsyncCommandRuntime::new(); - let output = "x".repeat(MAX_DELIVERED_OUTPUT_TOKENS_PER_REQUEST); for (launch_sequence, output) in (0_u64..).zip([ - TestOutput::AdditionalContext(&output), - TestOutput::SystemMessage(&output), - TestOutput::AdditionalContext(&output), - TestOutput::SystemMessage(&output), - TestOutput::AdditionalContext(&output), + TestOutput::AdditionalContext("first"), + TestOutput::SystemMessage("notice"), + TestOutput::AdditionalContext("second"), ]) { complete(&runtime, launch_sequence, output); } - let first_delivery = runtime.pending_delivery().accept_turn(); + let delivery = runtime.pending_delivery().accept_turn(); assert_eq!( - first_delivery, + delivery, super::AsyncHookDelivery { - additional_contexts: vec![output.clone(), output.clone()], - system_messages: vec![output.clone(), output.clone()], + additional_contexts: vec!["first".to_string(), "second".to_string()], + system_messages: vec!["notice".to_string()], } ); - - let second_delivery = runtime.pending_delivery().accept_turn(); - assert_eq!(second_delivery, context_delivery(output)); }