Merge remote-tracking branch 'refs/remotes/origin/nornagon/parallel-tool-calls' into nornagon/parallel-tool-calls

This commit is contained in:
Jeremy Rose
2025-08-25 15:03:24 -07:00
2 changed files with 206 additions and 0 deletions
+156
View File
@@ -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<ProcessedResponseItem> = 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) => {
+50
View File
@@ -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<Line<'static>> =
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<AtomicBool>,
}
impl RunningMcpCell {
pub(crate) fn new(invocation: McpInvocation, is_running: Arc<AtomicBool>) -> Self {
Self {
invocation,
start_time: Instant::now(),
is_running,
}
}
}
impl HistoryCell for RunningMcpCell {
fn display_lines(&self) -> Vec<Line<'static>> {
let mut lines: Vec<Line<'static>> = 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(