diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 1ac95c8f38..a24e12a084 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -1979,6 +1979,162 @@ async fn try_run_turn( let _ = sess.tx_event.send(event).await; } + // Run MCP, LocalShell, and non-apply exec tool calls fully in parallel. + let mut mcp_futs = Vec::new(); + let mut local_futs = Vec::new(); + let mut exec_parallel_futs = Vec::new(); + // Keep apply_patch exec calls sequential for diff tracking. + let mut exec_sequential_calls = Vec::new(); + + for sc in staged { + match sc { + StagedCall::Mcp { + item, + server, + tool, + args, + call_id, + } => { + let fut = handle_mcp_tool_call( + sess, + sub_id, + call_id.clone(), + server, + tool, + args, + None, + ); + mcp_futs.push(async move { + let response = fut.await; + ProcessedResponseItem { + item, + response: Some(response), + } + }); + } + StagedCall::LocalShell { + item, + id, + call_id, + action, + } => { + let effective_call_id = match (call_id, id) { + (Some(call_id), _) => call_id, + (None, Some(id)) => id, + (None, None) => String::new(), + }; + let LocalShellAction::Exec(action) = action; + let params = ShellToolCallParams { + command: action.command, + workdir: action.working_directory, + timeout_ms: action.timeout_ms, + with_escalated_permissions: None, + justification: None, + }; + let exec_params = to_exec_params(params, turn_context); + let fut = async move { + // Use a fresh tracker – non-apply_patch calls won't emit diffs anyway. + let mut tdt = TurnDiffTracker::new(); + let response = handle_container_exec_with_params( + exec_params, + sess, + turn_context, + &mut tdt, + sub_id.to_string(), + effective_call_id, + ) + .await; + ProcessedResponseItem { + item, + response: Some(response), + } + }; + local_futs.push(fut); + } + StagedCall::Exec { + item, + name, + args, + call_id, + } => { + if name == "apply_patch" { + exec_sequential_calls.push(StagedCall::Exec { + item, + name, + args, + call_id, + }); + } else { + let fut = async move { + let response = match parse_container_exec_arguments( + args, + turn_context, + &call_id, + ) { + Ok(params) => { + let mut tdt = TurnDiffTracker::new(); + handle_container_exec_with_params( + params, + sess, + turn_context, + &mut tdt, + sub_id.to_string(), + call_id, + ) + .await + } + Err(output) => *output, + }; + ProcessedResponseItem { + item, + response: Some(response), + } + }; + exec_parallel_futs.push(fut); + } + } + other => exec_sequential_calls.push(other), + } + } + + let mut processed: Vec = join_all(mcp_futs).await; + processed.extend(join_all(local_futs).await); + processed.extend(join_all(exec_parallel_futs).await); + + // Now handle exec/shell calls one by one so we can use the turn_diff_tracker safely. + for sc in exec_sequential_calls { + match sc { + StagedCall::Exec { + item, + name, + args, + call_id, + } => { + // Reuse existing logic for exec tools. + let response = handle_function_call( + sess, + turn_context, + turn_diff_tracker, + sub_id.to_string(), + name, + args, + call_id, + ) + .await; + processed.push(ProcessedResponseItem { + item, + response: Some(response), + }); + } + StagedCall::LocalShell { .. } => {} + StagedCall::Mcp { .. } => {} + } + } + + // Preserve original ordering: non-tool items were already pushed to `output` as they arrived; + // now append the processed tool-call items in their staged order. + output.extend(processed); + return Ok(output); } ResponseEvent::OutputTextDelta(delta) => { diff --git a/codex-rs/tui/src/history_cell.rs b/codex-rs/tui/src/history_cell.rs index 9e7481ccab..0299f58469 100644 --- a/codex-rs/tui/src/history_cell.rs +++ b/codex-rs/tui/src/history_cell.rs @@ -446,6 +446,7 @@ pub(crate) fn new_active_mcp_tool_call(invocation: McpInvocation) -> PlainHistor title_line, format_mcp_invocation(invocation), ]; +<<<<<<< HEAD PlainHistoryCell { lines } } @@ -497,9 +498,58 @@ impl HistoryCell for RunningMcpCell { pub(crate) fn new_web_search_call(query: String) -> PlainHistoryCell { let lines: Vec> = vec![Line::from(""), Line::from(vec!["🌐 ".into(), query.into()])]; +||||||| 097782c7 + +======= +>>>>>>> refs/remotes/origin/nornagon/parallel-tool-calls PlainHistoryCell { lines } } +#[derive(Debug)] +pub(crate) struct RunningMcpCell { + invocation: McpInvocation, + start_time: Instant, + is_running: Arc, +} + +impl RunningMcpCell { + pub(crate) fn new(invocation: McpInvocation, is_running: Arc) -> Self { + Self { + invocation, + start_time: Instant::now(), + is_running, + } + } +} + +impl HistoryCell for RunningMcpCell { + fn display_lines(&self) -> Vec> { + let mut lines: Vec> = Vec::new(); + lines.push(Line::from("")); + + // Spinner or final status marker + let running = self.is_running.load(Ordering::Relaxed); + let status = if running { + const FRAMES: &[char] = &['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']; + let idx = ((self.start_time.elapsed().as_millis() / 100) as usize) % FRAMES.len(); + Span::raw(format!("{}", FRAMES[idx])) + } else { + "✓".green() + }; + + let title = Line::from(vec![ + "tool".magenta(), + " ".into(), + status, + " ".into(), + "running...".dim(), + ]); + lines.push(title); + lines.push(format_mcp_invocation(self.invocation.clone())); + lines + } +} + /// If the first content is an image, return a new cell with the image. /// TODO(rgwood-dd): Handle images properly even if they're not the first result. fn try_new_completed_mcp_tool_call_with_image_output(