1use super::execution_context::RuntimeExecutionContext;
2use crate::tool_dispatch::{
3 ToolCallLaunch, ToolDispatchOutcome, ToolPreparationOutcome,
4 dispatch_granted_prepared_tool_attempt_launch_with_execution_context,
5 dispatch_granted_prepared_tool_call_launch_with_execution_context,
6 dispatch_prepared_tool_attempt_launch_with_execution_context,
7 dispatch_prepared_tool_call_launch_with_execution_context,
8 finalize_tool_result_with_execution_context, mark_retry_exhausted,
9 prepare_granted_tool_call_with_context, prepare_tool_call_with_context, retry_after_ms,
10 schedule_tool_batch,
11};
12use crate::{
13 ModelToolReturn, SessionEvent, ToolCallOutput, ToolCallRecord, ToolCancellation, ToolFailure,
14 ToolFailureClass, ToolResult, TurnActivityId, TurnEvent,
15};
16use std::collections::HashMap;
17
18#[derive(Clone)]
19pub struct ToolInvocation {
20 pub id: String,
21 pub tool_id: crate::ToolId,
22 pub args: serde_json::Value,
23 pub execution_grant: Option<Box<crate::ToolExecutionGrant>>,
24 pub child_execution_trace_hook: Option<crate::ToolChildExecutionTraceHook>,
25}
26
27impl ToolInvocation {
28 pub fn new(id: impl Into<String>, tool_id: crate::ToolId, args: serde_json::Value) -> Self {
29 Self {
30 id: id.into(),
31 tool_id,
32 args,
33 execution_grant: None,
34 child_execution_trace_hook: None,
35 }
36 }
37
38 pub fn with_execution_grant(mut self, grant: crate::ToolExecutionGrant) -> Self {
39 self.execution_grant = Some(Box::new(grant));
40 self
41 }
42
43 pub fn label(&self) -> String {
44 self.tool_id.to_string()
45 }
46
47 pub fn with_child_execution_trace_hook(
48 mut self,
49 hook: crate::ToolChildExecutionTraceHook,
50 ) -> Self {
51 self.child_execution_trace_hook = Some(hook);
52 self
53 }
54}
55
56impl std::fmt::Debug for ToolInvocation {
57 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58 f.debug_struct("ToolInvocation")
59 .field("id", &self.id)
60 .field("tool_id", &self.tool_id)
61 .field("args", &self.args)
62 .field(
63 "execution_grant",
64 &self.execution_grant.as_ref().map(|_| "<grant>"),
65 )
66 .field(
67 "child_execution_trace_hook",
68 &self.child_execution_trace_hook.as_ref().map(|_| "<hook>"),
69 )
70 .finish()
71 }
72}
73
74#[derive(Clone, Debug)]
75pub struct ToolInvocationReply {
76 pub output: ToolCallOutput,
77 pub record: Option<ToolCallRecord>,
78}
79
80impl ToolInvocationReply {
81 pub fn success(value: serde_json::Value) -> Self {
82 Self {
83 output: ToolCallOutput::success(value),
84 record: None,
85 }
86 }
87
88 pub fn error(value: serde_json::Value) -> Self {
89 let message = value
90 .as_str()
91 .map(ToOwned::to_owned)
92 .unwrap_or_else(|| value.to_string());
93 let mut failure = ToolFailure::tool(ToolFailureClass::Execution, "tool_error", message);
94 failure.raw =
95 Some(serde_json::from_value(value).unwrap_or_else(|_| {
96 crate::ToolValue::String("unserializable tool error".to_string())
97 }));
98 Self {
99 output: ToolCallOutput::failure(failure),
100 record: None,
101 }
102 }
103
104 pub fn from_output(output: ToolCallOutput) -> Self {
105 Self {
106 output,
107 record: None,
108 }
109 }
110
111 pub fn cancelled(message: impl Into<String>) -> Self {
112 Self::from_output(ToolCallOutput::cancelled(ToolCancellation::runtime(
113 message,
114 )))
115 }
116
117 pub(crate) fn with_record(mut self, record: ToolCallRecord) -> Self {
118 self.record = Some(record);
119 self
120 }
121}
122
123#[derive(Clone, Debug)]
124pub(crate) struct CompletedProtocolToolCall {
125 pub completed: crate::sansio::CompletedToolCall,
126 pub record: ToolCallRecord,
127}
128
129fn cancelled_runtime_tool_call_launch(
130 call_id: String,
131 tool_name: String,
132 args: serde_json::Value,
133 replay: Option<crate::llm::types::ProviderReplayMeta>,
134) -> crate::runtime::ToolCallLaunch {
135 crate::runtime::ToolCallLaunch::Done {
136 result: cancelled_completed_tool_call(call_id, tool_name, args, replay),
137 }
138}
139
140fn cancelled_completed_tool_call(
141 call_id: String,
142 tool_name: String,
143 args: serde_json::Value,
144 replay: Option<crate::llm::types::ProviderReplayMeta>,
145) -> crate::sansio::CompletedToolCall {
146 let output = ToolCallOutput::cancelled(ToolCancellation::runtime("tool call cancelled"));
147 crate::sansio::CompletedToolCall {
148 call_id: call_id.clone(),
149 tool_name: tool_name.clone(),
150 args,
151 model_return: ModelToolReturn {
152 call_id,
153 tool_name,
154 parts: vec![crate::ModelToolReturnPart::text(
155 "[Tool execution cancelled]\ntool call cancelled".to_string(),
156 )],
157 },
158 output,
159 duration_ms: 0,
160 replay,
161 }
162}
163
164fn runtime_failure_dispatch_outcome(
165 call_id: Option<String>,
166 tool_name: String,
167 args: serde_json::Value,
168 code: impl Into<String>,
169 message: impl Into<String>,
170) -> ToolDispatchOutcome {
171 ToolDispatchOutcome {
172 record: ToolCallRecord {
173 call_id,
174 tool: tool_name,
175 args,
176 output: ToolCallOutput::failure(ToolFailure::runtime(
177 ToolFailureClass::Internal,
178 code,
179 message,
180 )),
181 duration_ms: 0,
182 },
183 }
184}
185
186fn deterministic_tool_invocation_batch_id(calls: &[ToolInvocation]) -> String {
187 let identity = calls
188 .iter()
189 .map(|call| {
190 serde_json::json!({
191 "id": call.id.clone(),
192 "tool_id": call.tool_id.to_string(),
193 "args": call.args.clone(),
194 "execution_grant": call.execution_grant.as_ref().map(|grant| serde_json::json!({
195 "tool_id": grant.manifest.id.to_string(),
196 "source_id": grant.source_id.clone(),
197 "execution_binding": grant.execution_binding.clone(),
198 })),
199 })
200 })
201 .collect::<Vec<_>>();
202 let digest = crate::stable_hash::stable_json_sha256_hex(&identity)
203 .unwrap_or_else(|_| format!("len-{}", calls.len()));
204 format!("tool-batch:{digest}")
205}
206
207struct CoordinatedToolLaunch {
208 launch: crate::runtime::ToolCallLaunch,
209 triggers: Vec<crate::tool_dispatch::ToolTriggerEffectOutcome>,
210}
211
212impl RuntimeExecutionContext<'_> {
213 fn tool_batch_invocation(&self, batch_id: &str) -> crate::RuntimeInvocation {
214 let suffix = format!("tool-batch:{batch_id}");
215 if let Some(parent) = self.parent_invocation.as_ref() {
216 let parent_effect_id = parent.effect_id().unwrap_or("effect");
217 return crate::runtime::causal::child_effect_invocation(
218 parent,
219 format!("{parent_effect_id}:{suffix}"),
220 crate::RuntimeEffectKind::ToolBatch,
221 suffix,
222 );
223 }
224 let replay_key = format!("{}:{suffix}", self.execution_scope_id());
225 crate::RuntimeInvocation::effect(
226 crate::RuntimeScope::new(self.session_id.clone()),
227 suffix,
228 crate::RuntimeEffectKind::ToolBatch,
229 replay_key,
230 )
231 }
232
233 pub(crate) async fn execute_prepared_tool_batch_launches(
234 &self,
235 batch: crate::PreparedToolBatch,
236 parent_invocation: crate::RuntimeInvocation,
237 child_trace_hooks: HashMap<String, crate::ToolChildExecutionTraceHook>,
238 ) -> Result<crate::ToolBatchEffectOutcome, crate::RuntimeEffectControllerError> {
239 let indexed_tools = batch.calls.into_iter().enumerate().collect::<Vec<_>>();
240 let cancellation = self.cancellation_token.clone().unwrap_or_default();
241 let tool_cancel = cancellation.child_token();
242 let child_trace_hooks = std::sync::Arc::new(child_trace_hooks);
243 if !self
244 .dispatch
245 .effect_controller
246 .controller()
247 .supports_concurrent_effects()
248 {
249 let mut launches = Vec::with_capacity(indexed_tools.len());
250 let mut triggers = Vec::new();
251 let mut context = self.clone().with_cancellation_token(tool_cancel.clone());
252 for (index, child) in indexed_tools {
253 if cancellation.is_cancelled() {
254 tool_cancel.cancel();
255 launches.push(cancelled_runtime_tool_call_launch(
256 child.call.call_id,
257 child.call.tool_name,
258 child.call.args,
259 child.call.replay,
260 ));
261 continue;
262 }
263 let child_execution_trace_hook =
264 child_trace_hooks.get(&child.call.call_id).cloned();
265 let outcome = context
266 .execute_prepared_tool_batch_child(
267 child,
268 index,
269 parent_invocation.clone(),
270 child_execution_trace_hook,
271 )
272 .await;
273 launches.push(outcome.launch);
274 triggers.extend(outcome.triggers);
275 context = context.with_cancellation_token(tool_cancel.clone());
276 }
277 return Ok(crate::ToolBatchEffectOutcome { launches, triggers });
278 }
279 let child_outcomes = schedule_tool_batch(
280 indexed_tools,
281 |(index, _)| *index,
282 |(_, child)| self.tool_scheduling(&child.call.tool_name),
283 {
284 let context = self.clone();
285 let cancellation = cancellation.clone();
286 let tool_cancel = tool_cancel.clone();
287 let child_trace_hooks = std::sync::Arc::clone(&child_trace_hooks);
288 move |(index, child)| {
289 let context = context.clone().with_cancellation_token(tool_cancel.clone());
290 let cancellation = cancellation.clone();
291 let tool_cancel = tool_cancel.clone();
292 let parent_invocation = parent_invocation.clone();
293 let cancelled_tool = child.call.clone();
294 let child_execution_trace_hook =
295 child_trace_hooks.get(&child.call.call_id).cloned();
296 async move {
297 let tool_call = context.execute_prepared_tool_batch_child(
298 child,
299 index,
300 parent_invocation,
301 child_execution_trace_hook,
302 );
303 tokio::pin!(tool_call);
304 tokio::select! {
305 biased;
306 _ = cancellation.cancelled() => {
307 tool_cancel.cancel();
308 let grace = context
309 .dispatch
310 .clock
311 .sleep(std::time::Duration::from_millis(50));
312 tokio::pin!(grace);
313 tokio::select! {
314 biased;
315 outcome = &mut tool_call => outcome,
316 _ = &mut grace => CoordinatedToolLaunch {
317 launch: cancelled_runtime_tool_call_launch(
318 cancelled_tool.call_id,
319 cancelled_tool.tool_name,
320 cancelled_tool.args,
321 cancelled_tool.replay,
322 ),
323 triggers: Vec::new(),
324 },
325 }
326 }
327 outcome = &mut tool_call => outcome,
328 }
329 }
330 }
331 },
332 )
333 .await;
334
335 let mut launches = Vec::with_capacity(child_outcomes.len());
336 let mut triggers = Vec::new();
337 for outcome in child_outcomes {
338 launches.push(outcome.launch);
339 triggers.extend(outcome.triggers);
340 }
341 Ok(crate::ToolBatchEffectOutcome { launches, triggers })
342 }
343
344 async fn execute_prepared_tool_batch_child(
345 &self,
346 child: crate::PreparedToolBatchCall,
347 index: usize,
348 parent_invocation: crate::RuntimeInvocation,
349 child_execution_trace_hook: Option<crate::ToolChildExecutionTraceHook>,
350 ) -> CoordinatedToolLaunch {
351 let call_id = child.call.call_id.clone();
352 let tool_name = child.call.tool_name.clone();
353 let args = child.call.args.clone();
354 let replay = child.call.replay.clone();
355 let activity_id = TurnActivityId::new(format!("tool:{call_id}"));
356 self.emit_tool_call_started(&call_id, &tool_name, args.clone(), activity_id.clone())
357 .await;
358
359 let retry_policy = crate::tool_dispatch::resolve_callable_manifest_by_id(
360 self.dispatch.as_ref(),
361 &child.call.tool_id,
362 )
363 .map(|manifest| manifest.retry_policy)
364 .or_else(|| {
365 child
366 .execution_grant
367 .as_ref()
368 .map(|grant| grant.manifest.retry_policy)
369 })
370 .unwrap_or(crate::ToolRetryPolicy::Never);
371 let max_attempts = retry_policy.max_attempts().max(1);
372 let mut triggers = Vec::new();
373
374 for attempt in 1..=max_attempts {
375 let attempt_invocation =
376 self.tool_attempt_invocation(&parent_invocation, &child.replay_suffix, attempt);
377 let attempt_outcome = self
378 .dispatch
379 .effect_controller
380 .controller()
381 .execute_effect(
382 crate::RuntimeEffectEnvelope::new(
383 attempt_invocation,
384 crate::RuntimeEffectCommand::ToolAttempt {
385 call: child.call.clone(),
386 execution_grant: child.execution_grant.clone(),
387 attempt,
388 max_attempts,
389 },
390 ),
391 crate::RuntimeEffectLocalExecutor::tool_batch(
392 self.clone(),
393 child_execution_trace_hook
394 .clone()
395 .map(|hook| {
396 std::iter::once((child.call.call_id.clone(), hook)).collect()
397 })
398 .unwrap_or_default(),
399 ),
400 )
401 .await;
402 let attempt_outcome = match attempt_outcome {
403 Ok(outcome) => match outcome.into_tool_attempt_effect() {
404 Ok(outcome) => outcome,
405 Err(err) => {
406 let completed = self
407 .complete_tool_call(
408 index,
409 call_id.clone(),
410 replay,
411 runtime_failure_dispatch_outcome(
412 Some(call_id.clone()),
413 tool_name,
414 args,
415 "tool_attempt_failed",
416 err.to_string(),
417 ),
418 activity_id,
419 )
420 .await;
421 return CoordinatedToolLaunch {
422 launch: crate::runtime::ToolCallLaunch::Done {
423 result: completed.completed,
424 },
425 triggers,
426 };
427 }
428 },
429 Err(err) => {
430 let completed = self
431 .complete_tool_call(
432 index,
433 call_id.clone(),
434 replay,
435 runtime_failure_dispatch_outcome(
436 Some(call_id.clone()),
437 tool_name,
438 args,
439 "tool_attempt_failed",
440 err.to_string(),
441 ),
442 activity_id,
443 )
444 .await;
445 return CoordinatedToolLaunch {
446 launch: crate::runtime::ToolCallLaunch::Done {
447 result: completed.completed,
448 },
449 triggers,
450 };
451 }
452 };
453 triggers.extend(attempt_outcome.triggers);
454 match attempt_outcome.launch {
455 crate::ToolAttemptLaunch::Pending {
456 key,
457 pending,
458 duration_ms,
459 } => {
460 let dispatch_outcome = self
461 .await_pending_tool_dispatch_outcome_with_suffix(
462 &call_id,
463 Some(parent_invocation.clone()),
464 format!("{}:await", child.replay_suffix),
465 crate::tool_dispatch::PendingToolDispatchOutcome {
466 tool_name: child.call.tool_name.clone(),
467 args: child.call.args.clone(),
468 key,
469 pending,
470 duration_ms,
471 },
472 self.cancellation_token.clone(),
473 )
474 .await;
475 let completed = self
476 .complete_tool_call(
477 index,
478 call_id.clone(),
479 child.call.replay.clone(),
480 dispatch_outcome,
481 activity_id,
482 )
483 .await;
484 return CoordinatedToolLaunch {
485 launch: crate::runtime::ToolCallLaunch::Done {
486 result: completed.completed,
487 },
488 triggers,
489 };
490 }
491 crate::ToolAttemptLaunch::Done { mut record } => {
492 record.call_id = Some(call_id.clone());
493 let retry_after = retry_after_ms(
494 &ToolResult::from_output(record.output.clone()),
495 retry_policy,
496 attempt - 1,
497 );
498 let Some(retry_after) = retry_after else {
499 let completed = self
500 .complete_tool_call(
501 index,
502 call_id.clone(),
503 child.call.replay.clone(),
504 ToolDispatchOutcome { record },
505 activity_id,
506 )
507 .await;
508 return CoordinatedToolLaunch {
509 launch: crate::runtime::ToolCallLaunch::Done {
510 result: completed.completed,
511 },
512 triggers,
513 };
514 };
515 if attempt >= max_attempts {
516 let exhausted =
517 mark_retry_exhausted(ToolResult::from_output(record.output), attempt);
518 record.output = exhausted.into_done_output().unwrap_or_else(|_| {
519 ToolCallOutput::failure(ToolFailure::runtime(
520 ToolFailureClass::Internal,
521 "tool_retry_exhaustion_failed",
522 "retry exhaustion produced a pending output",
523 ))
524 });
525 let completed = self
526 .complete_tool_call(
527 index,
528 call_id.clone(),
529 child.call.replay.clone(),
530 ToolDispatchOutcome { record },
531 activity_id,
532 )
533 .await;
534 return CoordinatedToolLaunch {
535 launch: crate::runtime::ToolCallLaunch::Done {
536 result: completed.completed,
537 },
538 triggers,
539 };
540 }
541 if retry_after > 0
542 && let Err(err) = self
543 .sleep_before_tool_retry(
544 &parent_invocation,
545 &child.replay_suffix,
546 attempt,
547 retry_after,
548 )
549 .await
550 {
551 let completed = self
552 .complete_tool_call(
553 index,
554 call_id.clone(),
555 child.call.replay.clone(),
556 runtime_failure_dispatch_outcome(
557 Some(call_id.clone()),
558 child.call.tool_name.clone(),
559 child.call.args.clone(),
560 "tool_retry_sleep_failed",
561 format!(
562 "retry sleep for tool `{}` failed after attempt {attempt}: {err}",
563 child.call.tool_name
564 ),
565 ),
566 activity_id,
567 )
568 .await;
569 return CoordinatedToolLaunch {
570 launch: crate::runtime::ToolCallLaunch::Done {
571 result: completed.completed,
572 },
573 triggers,
574 };
575 }
576 }
577 }
578 }
579
580 let completed = self
581 .complete_tool_call(
582 index,
583 call_id.clone(),
584 child.call.replay,
585 runtime_failure_dispatch_outcome(
586 Some(call_id),
587 child.call.tool_name,
588 child.call.args,
589 "tool_retry_loop_failed",
590 "tool retry loop exited without a terminal result",
591 ),
592 activity_id,
593 )
594 .await;
595 CoordinatedToolLaunch {
596 launch: crate::runtime::ToolCallLaunch::Done {
597 result: completed.completed,
598 },
599 triggers,
600 }
601 }
602
603 async fn emit_tool_call_started(
604 &self,
605 call_id: &str,
606 name: &str,
607 args: serde_json::Value,
608 activity_id: TurnActivityId,
609 ) {
610 let _ = self
611 .dispatch
612 .event_tx
613 .send(SessionEvent::ToolCallStart {
614 call_id: Some(call_id.to_string()),
615 name: name.to_string(),
616 args: args.clone(),
617 })
618 .await;
619 self.emit_turn_activity(
620 activity_id,
621 TurnEvent::ToolCallStarted {
622 call_id: Some(call_id.to_string()),
623 name: name.to_string(),
624 args,
625 },
626 )
627 .await;
628 }
629
630 fn tool_attempt_invocation(
631 &self,
632 parent_invocation: &crate::RuntimeInvocation,
633 child_replay_suffix: &str,
634 attempt: u32,
635 ) -> crate::RuntimeInvocation {
636 let suffix = format!("{child_replay_suffix}:attempt:{attempt}");
637 let parent_effect_id = parent_invocation.effect_id().unwrap_or("tool-batch");
638 crate::runtime::causal::child_effect_invocation(
639 parent_invocation,
640 format!("{parent_effect_id}:{suffix}"),
641 crate::RuntimeEffectKind::ToolAttempt,
642 suffix,
643 )
644 }
645
646 async fn sleep_before_tool_retry(
647 &self,
648 parent_invocation: &crate::RuntimeInvocation,
649 child_replay_suffix: &str,
650 attempt: u32,
651 retry_after_ms: u64,
652 ) -> Result<(), crate::RuntimeEffectControllerError> {
653 let suffix = format!("{child_replay_suffix}:attempt:{attempt}:sleep");
654 let parent_effect_id = parent_invocation.effect_id().unwrap_or("tool-batch");
655 let invocation = crate::runtime::causal::child_effect_invocation(
656 parent_invocation,
657 format!("{parent_effect_id}:{suffix}"),
658 crate::RuntimeEffectKind::Sleep,
659 suffix,
660 );
661 let cancellation = self.cancellation_token.clone().unwrap_or_default();
662 let outcome = self
663 .dispatch
664 .effect_controller
665 .controller()
666 .execute_effect(
667 crate::RuntimeEffectEnvelope::new(
668 invocation,
669 crate::RuntimeEffectCommand::Sleep {
670 duration_ms: retry_after_ms,
671 },
672 ),
673 crate::RuntimeEffectLocalExecutor::sleep_with_clock(
674 cancellation,
675 std::sync::Arc::clone(&self.dispatch.clock),
676 ),
677 )
678 .await?;
679 match outcome {
680 crate::RuntimeEffectOutcome::Sleep => Ok(()),
681 other => Err(crate::RuntimeEffectControllerError::new(
682 "runtime_effect_wrong_outcome",
683 format!("expected sleep outcome, got {}", other.kind().as_str()),
684 )),
685 }
686 }
687
688 #[expect(
689 clippy::too_many_arguments,
690 reason = "tool execution carries explicit runtime call metadata"
691 )]
692 pub(crate) async fn execute_tool_call(
693 &self,
694 call_id: String,
695 name: String,
696 args: serde_json::Value,
697 index: usize,
698 replay: Option<crate::llm::types::ProviderReplayMeta>,
699 parent_invocation: Option<crate::RuntimeInvocation>,
700 child_execution_trace_hook: Option<crate::ToolChildExecutionTraceHook>,
701 ) -> CompletedProtocolToolCall {
702 let _ = self
703 .dispatch
704 .event_tx
705 .send(SessionEvent::ToolCallStart {
706 call_id: Some(call_id.clone()),
707 name: name.clone(),
708 args: args.clone(),
709 })
710 .await;
711 let tool_correlation_id = TurnActivityId::new(format!("tool:{call_id}"));
712 self.emit_turn_activity(
713 tool_correlation_id.clone(),
714 TurnEvent::ToolCallStarted {
715 call_id: Some(call_id.clone()),
716 name: name.clone(),
717 args: args.clone(),
718 },
719 )
720 .await;
721
722 let parent_invocation = parent_invocation.or_else(|| self.parent_invocation.clone());
723 let mut dispatch = (*self.dispatch).clone();
724 dispatch.parent_invocation = parent_invocation.clone();
725 let pending = crate::sansio::PendingToolCall {
726 call_id: call_id.clone(),
727 tool_name: name,
728 args,
729 replay: replay.clone(),
730 };
731 let launch = match prepare_tool_call_with_context(&dispatch, pending, Some(call_id.clone()))
732 .await
733 {
734 ToolPreparationOutcome::Prepared(prepared) => {
735 let dispatch_context = std::sync::Arc::new(dispatch.clone());
736 let runtime_context = if let Some(parent_invocation) = parent_invocation.clone() {
737 self.clone().with_parent_invocation(parent_invocation)
738 } else {
739 self.clone()
740 };
741 let mut tool_context =
742 crate::ToolContext::from_dispatch(std::sync::Arc::clone(&dispatch_context))
743 .runtime_execution_context(runtime_context)
744 .prepared_call(&prepared)
745 .cancellation_token(self.cancellation_token.clone())
746 .runtime_process_id(self.runtime_process_id.clone())
747 .parent_invocation(parent_invocation.clone())
748 .child_execution_trace_hook(child_execution_trace_hook.clone());
749 if let Some(process_events) = self.process_event_context.as_ref() {
750 tool_context = tool_context.process_events(
751 process_events.process_id.clone(),
752 std::sync::Arc::clone(&process_events.registry),
753 process_events.awaiter.clone(),
754 process_events.store.clone(),
755 process_events.session_store_factory.clone(),
756 process_events.queued_work_driver.clone(),
757 );
758 }
759 let tool_context = tool_context.build();
760 dispatch_prepared_tool_call_launch_with_execution_context(
761 dispatch_context.as_ref(),
762 prepared,
763 None,
764 tool_context,
765 )
766 .await
767 }
768 ToolPreparationOutcome::Completed(outcome) => ToolCallLaunch::Done(*outcome),
769 };
770 let mut outcome = match launch {
771 ToolCallLaunch::Done(outcome) => outcome,
772 ToolCallLaunch::Pending(pending) => {
773 self.await_pending_tool_dispatch_outcome(
774 &call_id,
775 parent_invocation.clone(),
776 pending,
777 self.cancellation_token.clone(),
778 )
779 .await
780 }
781 };
782 outcome.record.call_id = Some(call_id.clone());
783
784 self.complete_tool_call(index, call_id, replay, outcome, tool_correlation_id)
785 .await
786 }
787
788 #[expect(
789 clippy::too_many_arguments,
790 reason = "tool execution carries explicit runtime call metadata"
791 )]
792 pub(crate) async fn execute_tool_call_by_id(
793 &self,
794 call_id: String,
795 tool_id: crate::ToolId,
796 args: serde_json::Value,
797 index: usize,
798 replay: Option<crate::llm::types::ProviderReplayMeta>,
799 parent_invocation: Option<crate::RuntimeInvocation>,
800 child_execution_trace_hook: Option<crate::ToolChildExecutionTraceHook>,
801 ) -> CompletedProtocolToolCall {
802 let Some(manifest) =
803 crate::tool_dispatch::resolve_callable_manifest_by_id(self.dispatch.as_ref(), &tool_id)
804 else {
805 let outcome = ToolDispatchOutcome {
806 record: ToolCallRecord {
807 call_id: Some(call_id.clone()),
808 tool: tool_id.to_string(),
809 args,
810 output: ToolCallOutput::failure(ToolFailure::runtime(
811 ToolFailureClass::Unavailable,
812 "tool_unavailable",
813 format!("Tool id `{tool_id}` is unavailable in this session"),
814 )),
815 duration_ms: 0,
816 },
817 };
818 let activity_id = TurnActivityId::new(format!("tool:{call_id}"));
819 return self
820 .complete_tool_call(index, call_id, replay, outcome, activity_id)
821 .await;
822 };
823 self.execute_tool_call(
824 call_id,
825 manifest.name,
826 args,
827 index,
828 replay,
829 parent_invocation,
830 child_execution_trace_hook,
831 )
832 .await
833 }
834
835 #[expect(
836 clippy::too_many_arguments,
837 reason = "tool execution carries explicit runtime call metadata"
838 )]
839 pub(crate) async fn execute_tool_call_by_grant(
840 &self,
841 call_id: String,
842 grant: crate::ToolExecutionGrant,
843 args: serde_json::Value,
844 index: usize,
845 replay: Option<crate::llm::types::ProviderReplayMeta>,
846 parent_invocation: Option<crate::RuntimeInvocation>,
847 child_execution_trace_hook: Option<crate::ToolChildExecutionTraceHook>,
848 ) -> CompletedProtocolToolCall {
849 let name = grant.manifest.name.clone();
850 let _ = self
851 .dispatch
852 .event_tx
853 .send(SessionEvent::ToolCallStart {
854 call_id: Some(call_id.clone()),
855 name: name.clone(),
856 args: args.clone(),
857 })
858 .await;
859 let tool_correlation_id = TurnActivityId::new(format!("tool:{call_id}"));
860 self.emit_turn_activity(
861 tool_correlation_id.clone(),
862 TurnEvent::ToolCallStarted {
863 call_id: Some(call_id.clone()),
864 name: name.clone(),
865 args: args.clone(),
866 },
867 )
868 .await;
869
870 let parent_invocation = parent_invocation.or_else(|| self.parent_invocation.clone());
871 let mut dispatch = (*self.dispatch).clone();
872 dispatch.parent_invocation = parent_invocation.clone();
873 let pending = crate::sansio::PendingToolCall {
874 call_id: call_id.clone(),
875 tool_name: name,
876 args,
877 replay: replay.clone(),
878 };
879 let launch = match prepare_granted_tool_call_with_context(
880 &dispatch,
881 &grant,
882 pending,
883 Some(call_id.clone()),
884 )
885 .await
886 {
887 ToolPreparationOutcome::Prepared(prepared) => {
888 let dispatch_context = std::sync::Arc::new(dispatch.clone());
889 let runtime_context = if let Some(parent_invocation) = parent_invocation.clone() {
890 self.clone().with_parent_invocation(parent_invocation)
891 } else {
892 self.clone()
893 };
894 let mut tool_context =
895 crate::ToolContext::from_dispatch(std::sync::Arc::clone(&dispatch_context))
896 .runtime_execution_context(runtime_context)
897 .prepared_call(&prepared)
898 .tool_execution_binding(grant.execution_binding.clone())
899 .cancellation_token(self.cancellation_token.clone())
900 .runtime_process_id(self.runtime_process_id.clone())
901 .parent_invocation(parent_invocation.clone())
902 .child_execution_trace_hook(child_execution_trace_hook.clone());
903 if let Some(process_events) = self.process_event_context.as_ref() {
904 tool_context = tool_context.process_events(
905 process_events.process_id.clone(),
906 std::sync::Arc::clone(&process_events.registry),
907 process_events.awaiter.clone(),
908 process_events.store.clone(),
909 process_events.session_store_factory.clone(),
910 process_events.queued_work_driver.clone(),
911 );
912 }
913 let tool_context = tool_context.build();
914 dispatch_granted_prepared_tool_call_launch_with_execution_context(
915 dispatch_context.as_ref(),
916 &grant,
917 prepared,
918 None,
919 tool_context,
920 )
921 .await
922 }
923 ToolPreparationOutcome::Completed(outcome) => ToolCallLaunch::Done(*outcome),
924 };
925 let mut outcome = match launch {
926 ToolCallLaunch::Done(outcome) => outcome,
927 ToolCallLaunch::Pending(pending) => {
928 self.await_pending_tool_dispatch_outcome(
929 &call_id,
930 parent_invocation.clone(),
931 pending,
932 self.cancellation_token.clone(),
933 )
934 .await
935 }
936 };
937 outcome.record.call_id = Some(call_id.clone());
938
939 self.complete_tool_call(index, call_id, replay, outcome, tool_correlation_id)
940 .await
941 }
942
943 pub(crate) async fn prepare_tool_call(
944 &self,
945 pending: crate::sansio::PendingToolCall,
946 ) -> ToolPreparationOutcome {
947 let call_id = Some(pending.call_id.clone());
948 prepare_tool_call_with_context(self.dispatch.as_ref(), pending, call_id).await
949 }
950
951 pub(crate) async fn execute_prepared_tool_attempt_effect(
952 &self,
953 prepared: crate::PreparedToolCall,
954 execution_grant: Option<Box<crate::ToolExecutionGrant>>,
955 attempt: u32,
956 max_attempts: u32,
957 attempt_invocation: crate::RuntimeInvocation,
958 child_execution_trace_hook: Option<crate::ToolChildExecutionTraceHook>,
959 ) -> Result<crate::ToolAttemptEffectOutcome, crate::RuntimeEffectControllerError> {
960 let call_id = prepared.call_id.clone();
961 let mut attempt_dispatch = (*self.dispatch).clone();
962 attempt_dispatch.parent_invocation = Some(attempt_invocation.clone());
963 attempt_dispatch.trigger_outcomes =
964 crate::tool_dispatch::ToolTriggerOutcomeBuffer::default();
965 let attempt_dispatch = std::sync::Arc::new(attempt_dispatch);
966 let mut attempt_context = self.clone();
967 attempt_context.dispatch = std::sync::Arc::clone(&attempt_dispatch);
968 attempt_context.parent_invocation = Some(attempt_invocation.clone());
969
970 let mut tool_context =
971 crate::ToolContext::from_dispatch(std::sync::Arc::clone(&attempt_dispatch))
972 .runtime_execution_context(attempt_context.clone())
973 .prepared_call(&prepared)
974 .cancellation_token(self.cancellation_token.clone())
975 .runtime_process_id(self.runtime_process_id.clone())
976 .parent_invocation(Some(attempt_invocation))
977 .child_execution_trace_hook(child_execution_trace_hook);
978 if let Some(process_events) = self.process_event_context.as_ref() {
979 tool_context = tool_context.process_events(
980 process_events.process_id.clone(),
981 std::sync::Arc::clone(&process_events.registry),
982 process_events.awaiter.clone(),
983 process_events.store.clone(),
984 process_events.session_store_factory.clone(),
985 process_events.queued_work_driver.clone(),
986 );
987 }
988 let tool_context = tool_context.build();
989 let launch = match Box::pin(async {
990 if let Some(grant) = execution_grant.as_ref() {
991 dispatch_granted_prepared_tool_attempt_launch_with_execution_context(
992 attempt_dispatch.as_ref(),
993 grant,
994 prepared,
995 attempt,
996 max_attempts,
997 None,
998 tool_context,
999 )
1000 .await
1001 } else {
1002 dispatch_prepared_tool_attempt_launch_with_execution_context(
1003 attempt_dispatch.as_ref(),
1004 prepared,
1005 attempt,
1006 max_attempts,
1007 None,
1008 tool_context,
1009 )
1010 .await
1011 }
1012 })
1013 .await
1014 {
1015 ToolCallLaunch::Done(outcome) => {
1016 let mut record = outcome.record;
1017 record.call_id = Some(call_id);
1018 crate::ToolAttemptLaunch::Done { record }
1019 }
1020 ToolCallLaunch::Pending(pending) => crate::ToolAttemptLaunch::Pending {
1021 key: pending.key,
1022 pending: pending.pending,
1023 duration_ms: pending.duration_ms,
1024 },
1025 };
1026 let triggers = attempt_context
1027 .drain_tool_trigger_outcomes()
1028 .map_err(|err| {
1029 crate::RuntimeEffectControllerError::new(
1030 "tool_trigger_outcome_drain",
1031 err.to_string(),
1032 )
1033 })?;
1034 Ok(crate::ToolAttemptEffectOutcome { launch, triggers })
1035 }
1036
1037 pub(super) async fn await_process_with_cancellation(
1038 &self,
1039 process_id: &str,
1040 parent_invocation: Option<crate::RuntimeInvocation>,
1041 cancellation: Option<tokio_util::sync::CancellationToken>,
1042 ) -> Result<crate::ProcessAwaitOutput, crate::PluginError> {
1043 let _phase = self.named_phase("process.await_handle");
1044 if let Some(cancellation) = cancellation {
1045 tokio::select! {
1046 result = self.dispatch.processes.await_process(
1047 process_id,
1048 self.process_scope(parent_invocation.clone()),
1049 ) => result,
1050 _ = cancellation.cancelled() => {
1051 let _ = self.dispatch.processes.cancel(
1052 &self.dispatch.session_id,
1053 process_id,
1054 self.process_scope(parent_invocation.clone()),
1055 ).await;
1056 self.dispatch.processes.await_process(
1057 process_id,
1058 self.process_scope(parent_invocation),
1059 ).await
1060 }
1061 }
1062 } else {
1063 self.dispatch
1064 .processes
1065 .await_process(process_id, self.process_scope(parent_invocation))
1066 .await
1067 }
1068 }
1069
1070 pub(crate) async fn complete_tool_call(
1071 &self,
1072 _index: usize,
1073 call_id: String,
1074 replay: Option<crate::llm::types::ProviderReplayMeta>,
1075 outcome: ToolDispatchOutcome,
1076 tool_correlation_id: TurnActivityId,
1077 ) -> CompletedProtocolToolCall {
1078 let output = outcome.record.output.clone();
1079 let projection_output = output.clone();
1080 let projection_tool_name = outcome.record.tool.clone();
1081 let projection_args = outcome.record.args.clone();
1082 let projection_duration_ms = outcome.record.duration_ms;
1083 let projection_call_id = call_id.clone();
1084 tokio::task::yield_now().await;
1085 let plugins = std::sync::Arc::clone(&self.dispatch.plugins);
1086 let projection_context = crate::plugin::ToolResultProjectionContext {
1087 session_id: self.dispatch.session_id.clone(),
1088 tool_name: projection_tool_name,
1089 args: projection_args,
1090 output: projection_output,
1091 duration_ms: projection_duration_ms,
1092 call_id: projection_call_id,
1093 };
1094 let model_return = match plugins.project_tool_result(projection_context).await {
1095 Ok(projected) => projected,
1096 Err(err) => ModelToolReturn::text(
1097 call_id.clone(),
1098 outcome.record.tool.clone(),
1099 err.to_string(),
1100 ),
1101 };
1102
1103 self.emit_turn_activity(
1104 tool_correlation_id,
1105 TurnEvent::ToolCallCompleted {
1106 call_id: Some(call_id.clone()),
1107 name: outcome.record.tool.clone(),
1108 args: outcome.record.args.clone(),
1109 output: output.clone(),
1110 duration_ms: outcome.record.duration_ms,
1111 },
1112 )
1113 .await;
1114
1115 let record = ToolCallRecord {
1116 call_id: Some(call_id.clone()),
1117 tool: outcome.record.tool.clone(),
1118 args: outcome.record.args.clone(),
1119 output: output.clone(),
1120 duration_ms: outcome.record.duration_ms,
1121 };
1122 CompletedProtocolToolCall {
1123 completed: crate::sansio::CompletedToolCall {
1124 call_id,
1125 tool_name: outcome.record.tool,
1126 args: outcome.record.args,
1127 output,
1128 model_return,
1129 duration_ms: outcome.record.duration_ms,
1130 replay,
1131 },
1132 record,
1133 }
1134 }
1135
1136 pub(crate) async fn pending_completion_dispatch_outcome(
1137 &self,
1138 tool_name: String,
1139 args: serde_json::Value,
1140 resolution: crate::Resolution,
1141 duration_ms: u64,
1142 ) -> ToolDispatchOutcome {
1143 let output = crate::tool_result::tool_output_from_completion_resolution(resolution);
1144 let result = finalize_tool_result_with_execution_context(
1145 self.dispatch.as_ref(),
1146 &tool_name,
1147 &args,
1148 ToolResult::from_output(output),
1149 duration_ms,
1150 )
1151 .await;
1152 let output = result.into_done_output().unwrap_or_else(|_| {
1153 ToolCallOutput::failure(ToolFailure::runtime(
1154 ToolFailureClass::Internal,
1155 "pending_tool_not_finalized",
1156 "pending tool result reached a completed-output projection path",
1157 ))
1158 });
1159 ToolDispatchOutcome {
1160 record: ToolCallRecord {
1161 call_id: None,
1162 tool: tool_name,
1163 args,
1164 output,
1165 duration_ms,
1166 },
1167 }
1168 }
1169
1170 async fn await_pending_tool_dispatch_outcome(
1171 &self,
1172 call_id: &str,
1173 parent_invocation: Option<crate::RuntimeInvocation>,
1174 pending: crate::tool_dispatch::PendingToolDispatchOutcome,
1175 cancellation: Option<tokio_util::sync::CancellationToken>,
1176 ) -> ToolDispatchOutcome {
1177 self.await_pending_tool_dispatch_outcome_with_suffix(
1178 call_id,
1179 parent_invocation,
1180 format!("{call_id}:await"),
1181 pending,
1182 cancellation,
1183 )
1184 .await
1185 }
1186
1187 async fn await_pending_tool_dispatch_outcome_with_suffix(
1188 &self,
1189 call_id: &str,
1190 parent_invocation: Option<crate::RuntimeInvocation>,
1191 replay_suffix: String,
1192 pending: crate::tool_dispatch::PendingToolDispatchOutcome,
1193 cancellation: Option<tokio_util::sync::CancellationToken>,
1194 ) -> ToolDispatchOutcome {
1195 let fallback;
1196 let parent = if let Some(parent) = parent_invocation.as_ref() {
1197 parent
1198 } else {
1199 fallback = crate::RuntimeInvocation::effect(
1200 crate::RuntimeScope::new(&self.dispatch.session_id),
1201 format!("tool:{call_id}:await"),
1202 crate::RuntimeEffectKind::AwaitEvent,
1203 format!("tool:{call_id}:await"),
1204 );
1205 &fallback
1206 };
1207 let parent_effect_id = parent.effect_id().unwrap_or("tool");
1208 let invocation = crate::runtime::causal::child_effect_invocation(
1209 parent,
1210 format!("{parent_effect_id}:{replay_suffix}"),
1211 crate::RuntimeEffectKind::AwaitEvent,
1212 replay_suffix,
1213 );
1214 let cancellation = cancellation.unwrap_or_default();
1215 let deadline = pending
1216 .pending
1217 .deadline
1218 .map(|duration| self.dispatch.clock.now() + duration);
1219 let outcome = self
1220 .dispatch
1221 .effect_controller
1222 .controller()
1223 .execute_effect(
1224 crate::RuntimeEffectEnvelope::new(
1225 invocation,
1226 crate::RuntimeEffectCommand::AwaitEvent { key: pending.key },
1227 ),
1228 crate::RuntimeEffectLocalExecutor::await_event_with_clock(
1229 cancellation,
1230 deadline,
1231 std::sync::Arc::clone(&self.dispatch.clock),
1232 ),
1233 )
1234 .await;
1235 let resolution = match outcome.and_then(crate::RuntimeEffectOutcome::into_await_event) {
1236 Ok(resolution) => resolution,
1237 Err(err) => {
1238 return ToolDispatchOutcome {
1239 record: ToolCallRecord {
1240 call_id: None,
1241 tool: pending.tool_name,
1242 args: pending.args,
1243 output: ToolCallOutput::failure(ToolFailure::runtime(
1244 ToolFailureClass::Internal,
1245 "pending_tool_completion_failed",
1246 err.to_string(),
1247 )),
1248 duration_ms: pending.duration_ms,
1249 },
1250 };
1251 }
1252 };
1253 self.pending_completion_dispatch_outcome(
1254 pending.tool_name,
1255 pending.args,
1256 resolution,
1257 pending.duration_ms,
1258 )
1259 .await
1260 }
1261
1262 pub async fn call_tool_by_id(
1263 &self,
1264 call_id: String,
1265 tool_id: crate::ToolId,
1266 args: serde_json::Value,
1267 index: usize,
1268 ) -> ToolInvocationReply {
1269 let executed = self
1270 .execute_tool_call_by_id(call_id, tool_id, args, index, None, None, None)
1271 .await;
1272 let reply = ToolInvocationReply::from_output(executed.completed.output);
1273 reply.with_record(executed.record)
1274 }
1275
1276 pub async fn call_tool_by_id_with_child_execution_trace_hook(
1277 &self,
1278 call_id: String,
1279 tool_id: crate::ToolId,
1280 args: serde_json::Value,
1281 index: usize,
1282 trace_hook: crate::ToolChildExecutionTraceHook,
1283 ) -> ToolInvocationReply {
1284 let executed = self
1285 .execute_tool_call_by_id(call_id, tool_id, args, index, None, None, Some(trace_hook))
1286 .await;
1287 let reply = ToolInvocationReply::from_output(executed.completed.output);
1288 reply.with_record(executed.record)
1289 }
1290
1291 pub async fn call_tool_with_execution_grant(
1292 &self,
1293 call_id: String,
1294 grant: crate::ToolExecutionGrant,
1295 args: serde_json::Value,
1296 index: usize,
1297 ) -> ToolInvocationReply {
1298 let executed = self
1299 .execute_tool_call_by_grant(call_id, grant, args, index, None, None, None)
1300 .await;
1301 let reply = ToolInvocationReply::from_output(executed.completed.output);
1302 reply.with_record(executed.record)
1303 }
1304
1305 pub async fn call_tool_with_execution_grant_and_child_execution_trace_hook(
1306 &self,
1307 call_id: String,
1308 grant: crate::ToolExecutionGrant,
1309 args: serde_json::Value,
1310 index: usize,
1311 trace_hook: crate::ToolChildExecutionTraceHook,
1312 ) -> ToolInvocationReply {
1313 let executed = self
1314 .execute_tool_call_by_grant(call_id, grant, args, index, None, None, Some(trace_hook))
1315 .await;
1316 let reply = ToolInvocationReply::from_output(executed.completed.output);
1317 reply.with_record(executed.record)
1318 }
1319
1320 pub async fn call_tool_batch(&self, calls: Vec<ToolInvocation>) -> Vec<ToolInvocationReply> {
1321 if calls.is_empty() {
1322 return Vec::new();
1323 }
1324
1325 let batch_id = deterministic_tool_invocation_batch_id(&calls);
1326 let mut replies = vec![None; calls.len()];
1327 let mut prepared_entries = Vec::new();
1328
1329 for (index, call) in calls.into_iter().enumerate() {
1330 let preparation = if let Some(grant) = call.execution_grant.as_deref().cloned() {
1331 let pending = crate::sansio::PendingToolCall {
1332 call_id: call.id.clone(),
1333 tool_name: grant.manifest.name.clone(),
1334 args: call.args,
1335 replay: None,
1336 };
1337 (
1338 Some(grant.clone()),
1339 prepare_granted_tool_call_with_context(
1340 self.dispatch.as_ref(),
1341 &grant,
1342 pending,
1343 Some(call.id.clone()),
1344 )
1345 .await,
1346 )
1347 } else {
1348 let Some(manifest) = crate::tool_dispatch::resolve_callable_manifest_by_id(
1349 self.dispatch.as_ref(),
1350 &call.tool_id,
1351 ) else {
1352 let outcome = ToolDispatchOutcome {
1353 record: ToolCallRecord {
1354 call_id: Some(call.id.clone()),
1355 tool: call.tool_id.to_string(),
1356 args: call.args,
1357 output: ToolCallOutput::failure(ToolFailure::runtime(
1358 ToolFailureClass::Unavailable,
1359 "tool_unavailable",
1360 format!(
1361 "Tool id `{}` is unavailable in this session",
1362 call.tool_id
1363 ),
1364 )),
1365 duration_ms: 0,
1366 },
1367 };
1368 let completed = self
1369 .complete_tool_call(
1370 index,
1371 call.id,
1372 None,
1373 outcome,
1374 TurnActivityId::new(format!("tool:{}", batch_id)),
1375 )
1376 .await;
1377 replies[index] = Some(
1378 ToolInvocationReply::from_output(completed.completed.output)
1379 .with_record(completed.record),
1380 );
1381 continue;
1382 };
1383
1384 let pending = crate::sansio::PendingToolCall {
1385 call_id: call.id.clone(),
1386 tool_name: manifest.name,
1387 args: call.args,
1388 replay: None,
1389 };
1390 (None, self.prepare_tool_call(pending).await)
1391 };
1392 let (execution_grant, preparation) = preparation;
1393 match preparation {
1394 ToolPreparationOutcome::Prepared(prepared) => {
1395 prepared_entries.push((
1396 index,
1397 prepared,
1398 execution_grant,
1399 call.child_execution_trace_hook,
1400 ));
1401 }
1402 ToolPreparationOutcome::Completed(outcome) => {
1403 let completed = self
1404 .complete_tool_call(
1405 index,
1406 call.id,
1407 None,
1408 *outcome,
1409 TurnActivityId::new(format!("tool:{}", batch_id)),
1410 )
1411 .await;
1412 replies[index] = Some(
1413 ToolInvocationReply::from_output(completed.completed.output)
1414 .with_record(completed.record),
1415 );
1416 }
1417 }
1418 }
1419
1420 if !prepared_entries.is_empty() {
1421 let invocation = self.tool_batch_invocation(&batch_id);
1422 let batch = crate::PreparedToolBatch::new_with_grants(
1423 batch_id.clone(),
1424 prepared_entries
1425 .iter()
1426 .map(|(_, prepared, grant, _)| (prepared.clone(), grant.clone()))
1427 .collect(),
1428 );
1429 let child_trace_hooks = prepared_entries
1430 .iter()
1431 .filter_map(|(_, prepared, _, hook)| {
1432 hook.clone().map(|hook| (prepared.call_id.clone(), hook))
1433 })
1434 .collect();
1435 let envelope = crate::RuntimeEffectEnvelope::new(
1436 invocation.clone(),
1437 crate::RuntimeEffectCommand::ToolBatch { batch },
1438 );
1439 let local_executor =
1440 crate::RuntimeEffectLocalExecutor::tool_batch(self.clone(), child_trace_hooks);
1441 let raw_outcome = self
1442 .dispatch
1443 .effect_controller
1444 .controller()
1445 .execute_effect(envelope, local_executor)
1446 .await;
1447 let outcome =
1448 match raw_outcome.and_then(crate::RuntimeEffectOutcome::into_tool_batch_effect) {
1449 Ok(outcome) => outcome,
1450 Err(err) => {
1451 for (index, prepared, _, _) in prepared_entries {
1452 replies[index] = Some(ToolInvocationReply::error(serde_json::json!(
1453 format!("tool batch failed: {err}")
1454 )));
1455 let _ = prepared;
1456 }
1457 return replies
1458 .into_iter()
1459 .map(|reply| reply.expect("every batch reply slot should be filled"))
1460 .collect();
1461 }
1462 };
1463 if outcome.launches.len() != prepared_entries.len() {
1464 let message = format!(
1465 "tool batch returned {} launches for {} prepared calls",
1466 outcome.launches.len(),
1467 prepared_entries.len()
1468 );
1469 for (index, _, _, _) in prepared_entries {
1470 replies[index] = Some(ToolInvocationReply::error(serde_json::json!(message)));
1471 }
1472 } else {
1473 for ((index, prepared, _, _), launch) in
1474 prepared_entries.into_iter().zip(outcome.launches)
1475 {
1476 let call_id = prepared.call_id.clone();
1477 let reply = match launch {
1478 crate::runtime::ToolCallLaunch::Done { result } => {
1479 let record = ToolCallRecord {
1480 call_id: Some(result.call_id.clone()),
1481 tool: result.tool_name.clone(),
1482 args: result.args.clone(),
1483 output: result.output.clone(),
1484 duration_ms: result.duration_ms,
1485 };
1486 ToolInvocationReply::from_output(result.output).with_record(record)
1487 }
1488 crate::runtime::ToolCallLaunch::Pending {
1489 key,
1490 pending,
1491 duration_ms,
1492 } => {
1493 let dispatch_outcome = self
1494 .await_pending_tool_dispatch_outcome(
1495 &call_id,
1496 Some(invocation.clone()),
1497 crate::tool_dispatch::PendingToolDispatchOutcome {
1498 tool_name: prepared.tool_name.clone(),
1499 args: prepared.args.clone(),
1500 key,
1501 pending,
1502 duration_ms,
1503 },
1504 self.cancellation_token.clone(),
1505 )
1506 .await;
1507 let completed = self
1508 .complete_tool_call(
1509 index,
1510 call_id.clone(),
1511 prepared.replay.clone(),
1512 dispatch_outcome,
1513 TurnActivityId::new(format!("tool:{call_id}")),
1514 )
1515 .await;
1516 ToolInvocationReply::from_output(completed.completed.output)
1517 .with_record(completed.record)
1518 }
1519 };
1520 replies[index] = Some(reply);
1521 }
1522 }
1523 }
1524
1525 replies
1526 .into_iter()
1527 .map(|reply| reply.expect("every batch reply slot should be filled"))
1528 .collect()
1529 }
1530
1531 pub async fn start_tool_call(
1532 &self,
1533 call_id: String,
1534 name: String,
1535 args: serde_json::Value,
1536 ) -> ToolInvocationReply {
1537 self.start_tool_process(call_id, name, args).await
1538 }
1539
1540 pub async fn await_tool_handle(
1541 &self,
1542 call_id: String,
1543 handle: serde_json::Value,
1544 ) -> ToolInvocationReply {
1545 self.await_process_handle(call_id, handle).await
1546 }
1547
1548 pub async fn cancel_tool_handle(
1549 &self,
1550 call_id: String,
1551 handle: serde_json::Value,
1552 ) -> ToolInvocationReply {
1553 self.cancel_process_handle(call_id, handle).await
1554 }
1555
1556 pub async fn signal_tool_handle(
1557 &self,
1558 call_id: String,
1559 handle: serde_json::Value,
1560 signal_name: String,
1561 payload: serde_json::Value,
1562 ) -> ToolInvocationReply {
1563 self.signal_process_handle(call_id, handle, signal_name, payload)
1564 .await
1565 }
1566}