use super::*;
use crate::agent::tool_lifecycle::parallel;
use crate::code_mode::engine::{
ResourceLimitExceeded, SdkRequest, serialized_size, value_reservation,
};
struct PreparedCall {
call: ToolCall,
sequence: u64,
available: bool,
context: ToolDispatchContext,
}
#[expect(clippy::too_many_arguments)]
pub(super) fn execute(
tools: &ToolRuntime,
hooks: Option<&HookRuntime>,
nested: &mut SessionPersistence<'_>,
sink: &mut NestedSink<'_, '_>,
mut instructions: Option<&mut SubdirInstructionState>,
parent: &ToolCall,
context: &ToolDispatchContext,
catalog: &Catalog,
request: SdkRequest,
) -> anyhow::Result<Value> {
let entries = request
.arguments
.as_array()
.ok_or_else(|| anyhow::anyhow!("Expected batch array"))?;
let mut prepared = Vec::with_capacity(entries.len());
for entry in entries {
nested.code_mode_sequence += 1;
let sequence = nested.code_mode_sequence;
let name = entry["name"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("Expected tool name"))?;
let canonical = catalog.canonical_name(name);
let call = ToolCall {
id: format!("{}:inner:{sequence}", parent.id),
name: canonical.clone().unwrap_or_else(|| name.to_owned()),
arguments: entry["args"].clone(),
};
nested.record_required(
SessionEventKind::ToolCall,
json!({"id":call.id,"name":call.name,"arguments":call.arguments}),
)?;
nested.require_code_mode_history_headroom()?;
prepared.push(PreparedCall {
call,
sequence,
available: canonical.is_some(),
context: context.clone(),
});
}
let final_sequence = nested.code_mode_sequence;
let result = (|| {
let mut inner_sink: Option<&mut dyn AgentOutputSink> = Some(sink);
for entry in &mut prepared {
let activity = tool_lifecycle::emit_tool_started(
&mut inner_sink,
0,
entry.sequence as usize,
&entry.call,
)?;
entry.context.parent_activity_id = Some(activity.activity_id);
entry.context.provider_call_id = Some(entry.call.id.clone());
}
let mut results = Vec::with_capacity(prepared.len());
let mut reservations = Vec::with_capacity(prepared.len());
let mut remaining_bytes = request.remaining_result_bytes;
let mut first_error = None;
let mut next = 0;
while next < prepared.len() {
let entry = &prepared[next];
let eligible = |entry: &PreparedCall| {
entry.available && parallel::eligible_call(hooks, &entry.call, false)
};
let mut end = next + 1;
if first_error.is_none() && !context.cancellation.is_canceled() && eligible(entry) {
while end < prepared.len() && eligible(&prepared[end]) {
end += 1;
}
let calls: Vec<_> = prepared[next..end]
.iter()
.map(|entry| (entry.call.clone(), entry.context.clone()))
.collect();
let dispatch = parallel::dispatch_ordered(tools, &calls, |index, outcome| {
let entry = &prepared[next + index];
nested.code_mode_sequence = entry.sequence;
let completed = tool_lifecycle::record_tool_dispatch(
Some(tools),
None,
nested,
&mut inner_sink,
instructions.as_deref_mut(),
entry.call.clone(),
entry.context.clone(),
outcome.into(),
)
.and_then(|outcome| {
finish_inner_call(
tools,
nested,
&mut inner_sink,
&entry.call,
entry
.context
.parent_activity_id
.clone()
.expect("started activity"),
context,
&request.memory,
outcome,
)
});
collect_result(
completed,
&request,
context,
&mut remaining_bytes,
&mut results,
&mut reservations,
&mut first_error,
);
Ok(())
});
if let Err(error) = dispatch {
first_error.get_or_insert(error);
}
} else {
nested.code_mode_sequence = entry.sequence;
let stopped = first_error.is_some() || context.cancellation.is_canceled();
let completed = if stopped || !entry.available {
let message = if stopped {
"Code Mode batch stopped before dispatch"
} else {
"Capability unavailable in Code Mode"
};
record_failed_call(nested, &mut inner_sink, entry, message).map(|()| json!({
"success":false, "error":{"code":if stopped { "batch_stopped" } else { "unavailable_tool" }, "message":message}
}))
} else {
let outcome = tool_lifecycle::execute_tool_call(
Some(tools),
hooks,
nested,
&mut inner_sink,
instructions.as_deref_mut(),
entry.call.clone(),
entry.context.clone(),
);
match outcome {
Ok(outcome) => finish_inner_call(
tools,
nested,
&mut inner_sink,
&entry.call,
entry
.context
.parent_activity_id
.clone()
.expect("started activity"),
context,
&request.memory,
outcome,
),
Err(error) => {
if !nested.code_mode_audit.borrow().has_result(entry.sequence) {
let cleanup = record_failed_call(
nested,
&mut inner_sink,
entry,
"Code Mode guarded host execution stopped",
);
if let Err(cleanup) = cleanup {
first_error.get_or_insert(cleanup);
}
}
Err(error)
}
}
};
collect_result(
completed,
&request,
context,
&mut remaining_bytes,
&mut results,
&mut reservations,
&mut first_error,
);
}
next = end;
}
context.cancellation.check()?;
if let Some(error) = first_error {
return Err(error);
}
Ok(Value::Array(results))
})();
nested.code_mode_sequence = final_sequence;
result
}
fn collect_result(
result: anyhow::Result<Value>,
request: &SdkRequest,
context: &ToolDispatchContext,
remaining_bytes: &mut usize,
results: &mut Vec<Value>,
reservations: &mut Vec<crate::code_mode::engine::MemoryReservation>,
first_error: &mut Option<anyhow::Error>,
) {
let result = result.and_then(|value| {
if first_error.is_some() {
return Ok(());
}
let limit = (*remaining_bytes).min(request.max_inner_result_bytes);
let size = serialized_size(&value, limit, &context.cancellation).map_err(|error| {
if error.downcast_ref::<ResourceLimitExceeded>().is_some() {
anyhow::Error::from(ResourceLimitExceeded(
if *remaining_bytes < request.max_inner_result_bytes {
"total result bytes"
} else {
"inner result bytes"
},
))
} else {
error
}
})?;
let reservation = request
.memory
.reserve(value_reservation(&value, &context.cancellation)?)?;
*remaining_bytes -= size;
reservations.push(reservation);
results.push(value);
Ok(())
});
if let Err(error) = result {
first_error.get_or_insert(error);
}
}
fn record_failed_call(
nested: &SessionPersistence<'_>,
sink: &mut Option<&mut dyn AgentOutputSink>,
entry: &PreparedCall,
message: &str,
) -> anyhow::Result<()> {
let result = ToolResult {
tool_name: entry.call.name.clone(),
success: false,
content: message.to_owned(),
metadata: json!({}),
display: Default::default(),
};
nested.record_required(
SessionEventKind::ToolResult,
json!({"call_id":entry.call.id,"result":result}),
)?;
if let Some(sink) = sink.as_deref_mut() {
sink.output_event(OutputEvent::ToolResult {
call: Box::new(entry.call.clone()),
result: Box::new(result.clone()),
changed_paths: Vec::new(),
summary: Box::new(crate::output::tool_display_summary(&entry.call, &result)),
})?;
}
tool_lifecycle::emit_tool_finished(
sink,
entry
.context
.parent_activity_id
.clone()
.expect("started activity"),
&entry.call,
&result,
)
}