core: avoid HOL blocking interrupted tool results

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<String>,
 }
 
+#[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<BoxFuture<'static, CodexResult<ToolDispatchOutput>>>,
+    in_flight: &mut FuturesUnordered<BoxFuture<'static, CodexResult<IndexedToolDispatchOutput>>>,
     sess: Arc<Session>,
     turn_context: Arc<TurnContext>,
 ) -> CodexResult<bool> {
-    let mut interrupted_tool_result = false;
+    let mut next_seq = 0usize;
+    let mut ready = BTreeMap::<usize, ToolDispatchOutput>::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<BoxFuture<'static, CodexResult<ToolDispatchOutput>>> =
-        FuturesOrdered::new();
+    let mut in_flight: FuturesUnordered<
+        BoxFuture<'static, CodexResult<IndexedToolDispatchOutput>>,
+    > = FuturesUnordered::new();
+    let mut next_in_flight_seq = 0usize;
     let mut needs_follow_up = false;
     let mut last_agent_message: Option<String> = None;
     let mut active_item: Option<TurnItem> = 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);

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<F>(mode_name: &str, build_mode: F) -> anyhow::Result<()>
 where
     F: FnOnce(String) -> CollaborationMode,