Remove async hook runtime limits

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"] }
 

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<u64>) -> 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;
             };

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));
 }