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