vv_agent/runtime/tool_call_runner/
runner.rs1use crate::runtime::hooks::RuntimeHookManager;
2use crate::tools::{ToolError, ToolOrchestrator, ToolRegistry, ToolRunOptions};
3use crate::types::{CompletionReason, ToolDirective};
4
5use super::outcome::ToolRunOutcome;
6use super::request::ToolRunRequest;
7use super::results::{
8 apply_tool_use_behavior, image_notification_from_tool_result, needs_tool_call_id,
9 skipped_tool_result,
10};
11
12pub struct ToolCallRunner {
13 tool_registry: ToolRegistry,
14 hook_manager: RuntimeHookManager,
15}
16
17impl ToolCallRunner {
18 pub fn new(tool_registry: ToolRegistry) -> Self {
19 Self {
20 tool_registry,
21 hook_manager: RuntimeHookManager::default(),
22 }
23 }
24
25 pub fn with_hook_manager(mut self, hook_manager: RuntimeHookManager) -> Self {
26 self.hook_manager = hook_manager;
27 self
28 }
29
30 pub fn run(&self, mut request: ToolRunRequest<'_>) -> Result<ToolRunOutcome, String> {
31 let mut directive_result = None;
32 let mut completion_reason = None;
33 let mut completion_tool_name = None;
34 let mut interruption_messages = Vec::new();
35 let mut image_notifications = Vec::new();
36 let orchestrator = ToolOrchestrator::from_tools(self.tool_registry.executors());
37
38 for (index, call) in request.tool_calls.iter().enumerate() {
39 if let Some(context) = request.execution_context {
40 context
41 .check_cancelled()
42 .map_err(|error| error.to_string())?;
43 }
44 let (patched_call, short_circuit_result) = self.hook_manager.apply_before_tool_call(
45 request.task,
46 request.context.cycle_index,
47 call.clone(),
48 request.context,
49 );
50 let options = ToolRunOptions::default();
51 let mut execution = match short_circuit_result {
52 Some(mut result) => {
53 if needs_tool_call_id(&result.tool_call_id) {
54 result.tool_call_id = call.id.clone();
55 }
56 orchestrator.observe_result_without_execution(
57 patched_call.clone(),
58 result,
59 &options,
60 )
61 }
62 None => {
63 block_on_tool_run(orchestrator.run_one_with_approval_and_metadata_deferred(
64 patched_call.clone(),
65 request.context,
66 options,
67 |_call, _requirement, _context, _metadata| None,
68 ))?
69 }
70 };
71 let mut result = execution.result().clone();
72 if needs_tool_call_id(&result.tool_call_id) {
73 result.tool_call_id = patched_call.id.clone();
74 }
75 result = self.hook_manager.apply_after_tool_call(
76 request.task,
77 request.context.cycle_index,
78 &patched_call,
79 request.context,
80 result,
81 );
82 if needs_tool_call_id(&result.tool_call_id) {
83 result.tool_call_id = patched_call.id.clone();
84 }
85 let behavior_reason = apply_tool_use_behavior(request.task, &patched_call, &mut result);
86 execution.replace_result(result);
87 let result = execution.complete();
88
89 request.messages.push(result.to_message());
90 if let Some(image_notification) =
91 image_notification_from_tool_result(&result, request.task.native_multimodal)
92 {
93 image_notifications.push(image_notification);
94 }
95 request.cycle_record.tool_results.push(result.clone());
96 if let Some(callback) = request.on_tool_result.as_deref_mut() {
97 callback(call, &result);
98 }
99
100 if result.directive != ToolDirective::Continue {
101 directive_result = Some(result.clone());
102 completion_reason = behavior_reason.or(Some(match result.directive {
103 ToolDirective::WaitUser => CompletionReason::WaitUser,
104 ToolDirective::Finish => CompletionReason::ToolFinish,
105 ToolDirective::Continue => unreachable!(),
106 }));
107 completion_tool_name = Some(patched_call.name.clone());
108 let (error_code, message) = match result.directive {
109 ToolDirective::WaitUser => (
110 "skipped_due_to_wait_user",
111 "Tool skipped because a previous tool requested user input.",
112 ),
113 ToolDirective::Finish => (
114 "skipped_due_to_finish",
115 "Tool skipped because a previous tool finished the task.",
116 ),
117 ToolDirective::Continue => ("skipped_due_to_directive", "Tool skipped."),
118 };
119 for skipped_call in request.tool_calls.iter().skip(index + 1) {
120 let skipped = skipped_tool_result(skipped_call, error_code, message);
121 request.messages.push(skipped.to_message());
122 request.cycle_record.tool_results.push(skipped.clone());
123 if let Some(callback) = request.on_tool_result.as_deref_mut() {
124 callback(skipped_call, &skipped);
125 }
126 }
127 break;
128 }
129
130 if let Some(provider) = request.interruption_provider {
131 let pending_messages = provider();
132 if !pending_messages.is_empty() {
133 interruption_messages.extend(pending_messages);
134 for skipped_call in request.tool_calls.iter().skip(index + 1) {
135 let skipped = skipped_tool_result(
136 skipped_call,
137 "skipped_due_to_steering",
138 "Tool skipped due to queued steering message.",
139 );
140 request.messages.push(skipped.to_message());
141 request.cycle_record.tool_results.push(skipped.clone());
142 if let Some(callback) = request.on_tool_result.as_deref_mut() {
143 callback(skipped_call, &skipped);
144 }
145 }
146 break;
147 }
148 }
149 }
150
151 request.messages.extend(image_notifications);
152 Ok(ToolRunOutcome {
153 directive_result,
154 completion_reason,
155 completion_tool_name,
156 interruption_messages,
157 })
158 }
159}
160
161fn block_on_tool_run<'a, T>(
162 future: impl std::future::Future<Output = Result<T, ToolError>> + 'a,
163) -> Result<T, String> {
164 if let Ok(handle) = tokio::runtime::Handle::try_current() {
165 if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread {
166 tokio::task::block_in_place(|| handle.block_on(future))
167 .map_err(|error| error.to_string())
168 } else {
169 handle.block_on(future).map_err(|error| error.to_string())
170 }
171 } else {
172 tokio::runtime::Builder::new_current_thread()
173 .enable_all()
174 .build()
175 .map_err(|error| error.to_string())?
176 .block_on(future)
177 .map_err(|error| error.to_string())
178 }
179}