vv_agent/runtime/tool_call_runner/
runner.rs

1use 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 mut result = match short_circuit_result {
51                Some(mut result) => {
52                    if needs_tool_call_id(&result.tool_call_id) {
53                        result.tool_call_id = call.id.clone();
54                    }
55                    result
56                }
57                None => {
58                    let mut result = block_on_tool_run(orchestrator.run_one(
59                        patched_call.clone(),
60                        request.context,
61                        ToolRunOptions::default(),
62                    ))?;
63                    if needs_tool_call_id(&result.tool_call_id) {
64                        result.tool_call_id = patched_call.id.clone();
65                    }
66                    result
67                }
68            };
69            result = self.hook_manager.apply_after_tool_call(
70                request.task,
71                request.context.cycle_index,
72                &patched_call,
73                request.context,
74                result,
75            );
76            if needs_tool_call_id(&result.tool_call_id) {
77                result.tool_call_id = patched_call.id.clone();
78            }
79            let behavior_reason = apply_tool_use_behavior(request.task, &patched_call, &mut result);
80
81            request.messages.push(result.to_message());
82            if let Some(image_notification) =
83                image_notification_from_tool_result(&result, request.task.native_multimodal)
84            {
85                image_notifications.push(image_notification);
86            }
87            request.cycle_record.tool_results.push(result.clone());
88            if let Some(callback) = request.on_tool_result.as_deref_mut() {
89                callback(call, &result);
90            }
91
92            if result.directive != ToolDirective::Continue {
93                directive_result = Some(result.clone());
94                completion_reason = behavior_reason.or(Some(match result.directive {
95                    ToolDirective::WaitUser => CompletionReason::WaitUser,
96                    ToolDirective::Finish => CompletionReason::ToolFinish,
97                    ToolDirective::Continue => unreachable!(),
98                }));
99                completion_tool_name = Some(patched_call.name.clone());
100                let (error_code, message) = match result.directive {
101                    ToolDirective::WaitUser => (
102                        "skipped_due_to_wait_user",
103                        "Tool skipped because a previous tool requested user input.",
104                    ),
105                    ToolDirective::Finish => (
106                        "skipped_due_to_finish",
107                        "Tool skipped because a previous tool finished the task.",
108                    ),
109                    ToolDirective::Continue => ("skipped_due_to_directive", "Tool skipped."),
110                };
111                for skipped_call in request.tool_calls.iter().skip(index + 1) {
112                    let skipped = skipped_tool_result(skipped_call, error_code, message);
113                    request.messages.push(skipped.to_message());
114                    request.cycle_record.tool_results.push(skipped.clone());
115                    if let Some(callback) = request.on_tool_result.as_deref_mut() {
116                        callback(skipped_call, &skipped);
117                    }
118                }
119                break;
120            }
121
122            if let Some(provider) = request.interruption_provider {
123                let pending_messages = provider();
124                if !pending_messages.is_empty() {
125                    interruption_messages.extend(pending_messages);
126                    for skipped_call in request.tool_calls.iter().skip(index + 1) {
127                        let skipped = skipped_tool_result(
128                            skipped_call,
129                            "skipped_due_to_steering",
130                            "Tool skipped due to queued steering message.",
131                        );
132                        request.messages.push(skipped.to_message());
133                        request.cycle_record.tool_results.push(skipped.clone());
134                        if let Some(callback) = request.on_tool_result.as_deref_mut() {
135                            callback(skipped_call, &skipped);
136                        }
137                    }
138                    break;
139                }
140            }
141        }
142
143        request.messages.extend(image_notifications);
144        Ok(ToolRunOutcome {
145            directive_result,
146            completion_reason,
147            completion_tool_name,
148            interruption_messages,
149        })
150    }
151}
152
153fn block_on_tool_run<'a>(
154    future: impl std::future::Future<Output = Result<crate::types::ToolExecutionResult, ToolError>> + 'a,
155) -> Result<crate::types::ToolExecutionResult, String> {
156    if let Ok(handle) = tokio::runtime::Handle::try_current() {
157        if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread {
158            tokio::task::block_in_place(|| handle.block_on(future))
159                .map_err(|error| error.to_string())
160        } else {
161            handle.block_on(future).map_err(|error| error.to_string())
162        }
163    } else {
164        tokio::runtime::Builder::new_current_thread()
165            .enable_all()
166            .build()
167            .map_err(|error| error.to_string())?
168            .block_on(future)
169            .map_err(|error| error.to_string())
170    }
171}
vv_agent/runtime/tool_call_runner/runner.rs

vv_agent/runtime/tool_call_runner/
runner.rs
