From 8d064684dd79ebe5acd8f3949bb13da3dfd7dc9f Mon Sep 17 00:00:00 2001 From: Charles Cunningham Date: Thu, 19 Feb 2026 22:14:10 -0800 Subject: [PATCH] core: avoid HOL blocking interrupted tool results --- codex-rs/core/src/codex.rs | 53 +++++++-- .../core/tests/suite/request_user_input.rs | 102 ++++++++++++++++++ 2 files changed, 144 insertions(+), 11 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 9cbee6c1a8..1e3286210a 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -1,3 +1,4 @@ +use std::collections::BTreeMap; use std::collections::HashMap; use std::collections::HashSet; use std::fmt::Debug; @@ -105,7 +106,7 @@ use codex_utils_stream_parser::extract_proposed_plan_text; use codex_utils_stream_parser::strip_citations; use futures::future::BoxFuture; use futures::prelude::*; -use futures::stream::FuturesOrdered; +use futures::stream::FuturesUnordered; use rmcp::model::ListResourceTemplatesResult; use rmcp::model::ListResourcesResult; use rmcp::model::PaginatedRequestParams; @@ -5650,6 +5651,12 @@ struct SamplingRequestResult { last_agent_message: Option, } +#[derive(Debug)] +struct IndexedToolDispatchOutput { + seq: usize, + output: ToolDispatchOutput, +} + /// Ephemeral per-response state for streaming a single proposed plan. /// This is intentionally not persisted or stored in session/state since it /// only exists while a response is actively streaming. The final plan text @@ -6162,24 +6169,41 @@ async fn handle_assistant_item_done_in_plan_mode( } async fn drain_in_flight( - in_flight: &mut FuturesOrdered>>, + in_flight: &mut FuturesUnordered>>, sess: Arc, turn_context: Arc, ) -> CodexResult { - let mut interrupted_tool_result = false; + let mut next_seq = 0usize; + let mut ready = BTreeMap::::new(); while let Some(res) = in_flight.next().await { match res { - Ok(output) => { - sess.record_conversation_items(&turn_context, &[output.response_input.into()]) - .await; - interrupted_tool_result |= output.interrupt_turn; + Ok(indexed) => { + let IndexedToolDispatchOutput { seq, output } = indexed; + if output.interrupt_turn { + sess.record_conversation_items(&turn_context, &[output.response_input.into()]) + .await; + return Ok(true); + } + + ready.insert(seq, output); + while let Some(output) = ready.remove(&next_seq) { + sess.record_conversation_items(&turn_context, &[output.response_input.into()]) + .await; + next_seq += 1; + } } Err(err) => { error_or_panic(format!("in-flight tool future failed during drain: {err}")); } } } - Ok(interrupted_tool_result) + if !ready.is_empty() { + for (_seq, output) in ready { + sess.record_conversation_items(&turn_context, &[output.response_input.into()]) + .await; + } + } + Ok(false) } #[allow(clippy::too_many_arguments)] @@ -6228,8 +6252,10 @@ async fn try_run_sampling_request( Arc::clone(&turn_context), Arc::clone(&turn_diff_tracker), ); - let mut in_flight: FuturesOrdered>> = - FuturesOrdered::new(); + let mut in_flight: FuturesUnordered< + BoxFuture<'static, CodexResult>, + > = FuturesUnordered::new(); + let mut next_in_flight_seq = 0usize; let mut needs_follow_up = false; let mut last_agent_message: Option = None; let mut active_item: Option = None; @@ -6313,7 +6339,12 @@ async fn try_run_sampling_request( .instrument(handle_responses) .await?; if let Some(tool_future) = output_result.tool_future { - in_flight.push_back(tool_future); + let seq = next_in_flight_seq; + next_in_flight_seq += 1; + in_flight.push(Box::pin(async move { + let output = tool_future.await?; + Ok(IndexedToolDispatchOutput { seq, output }) + })); } if let Some(agent_message) = output_result.last_agent_message { last_agent_message = Some(agent_message); diff --git a/codex-rs/core/tests/suite/request_user_input.rs b/codex-rs/core/tests/suite/request_user_input.rs index 49f4bd2b88..0bddc28dfc 100644 --- a/codex-rs/core/tests/suite/request_user_input.rs +++ b/codex-rs/core/tests/suite/request_user_input.rs @@ -1,6 +1,7 @@ #![allow(clippy::unwrap_used)] use std::collections::HashMap; +use std::time::Duration; use codex_core::features::Feature; use codex_protocol::config_types::CollaborationMode; @@ -352,6 +353,107 @@ async fn request_user_input_interrupted_response_preserves_tool_output() -> anyh Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn request_user_input_interrupt_not_blocked_by_earlier_tool() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + + let builder = test_codex().with_model("test-gpt-5.1-codex"); + let TestCodex { + codex, + cwd, + session_configured, + .. + } = builder + .with_config(|config| { + config.features.enable(Feature::CollaborationModes); + }) + .build(&server) + .await?; + + let call_id = "user-input-call-fast-interrupt"; + let slow_tool_args = json!({ + "sleep_after_ms": 2_000 + }) + .to_string(); + let request_args = json!({ + "questions": [{ + "id": "confirm_path", + "header": "Confirm", + "question": "Proceed with the plan?", + "options": [{ + "label": "Yes (Recommended)", + "description": "Continue the current plan." + }, { + "label": "No", + "description": "Stop and revisit the approach." + }] + }] + }) + .to_string(); + + let first_response = sse(vec![ + ev_response_created("resp-1"), + ev_function_call("slow-call-1", "test_sync_tool", &slow_tool_args), + ev_function_call(call_id, "request_user_input", &request_args), + ev_completed("resp-1"), + ]); + responses::mount_sse_once(&server, first_response).await; + + let session_model = session_configured.model.clone(); + + codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "please confirm".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + cwd: cwd.path().to_path_buf(), + approval_policy: AskForApproval::Never, + sandbox_policy: SandboxPolicy::DangerFullAccess, + model: session_model, + effort: None, + summary: ReasoningSummary::Auto, + collaboration_mode: Some(CollaborationMode { + mode: ModeKind::Plan, + settings: Settings { + model: session_configured.model.clone(), + reasoning_effort: None, + developer_instructions: None, + }, + }), + personality: None, + }) + .await?; + + let request = wait_for_event_match(&codex, |event| match event { + EventMsg::RequestUserInput(request) => Some(request.clone()), + _ => None, + }) + .await; + assert_eq!(request.call_id, call_id); + + codex + .submit(Op::UserInputAnswer { + id: request.turn_id.clone(), + response: RequestUserInputResponse { + answers: HashMap::new(), + interrupted: true, + }, + }) + .await?; + + tokio::time::timeout(Duration::from_millis(750), async { + wait_for_event(&codex, |event| matches!(event, EventMsg::TurnAborted(_))).await; + }) + .await + .expect("interrupting request_user_input should abort promptly"); + + Ok(()) +} + async fn assert_request_user_input_rejected(mode_name: &str, build_mode: F) -> anyhow::Result<()> where F: FnOnce(String) -> CollaborationMode,