vv_agent/runtime/tool_call_runner/
runner.rs

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

vv_agent/runtime/tool_call_runner/
runner.rs
