| 1 | //! Tool dispatch — plan/execute helpers for the per-turn tool batch. |
| 2 | //! |
| 3 | //! Extracted from `core/engine.rs` (P1.3). The high-level ordering still |
| 4 | //! lives in `Engine::handle_deepseek_turn`; this module owns: |
| 5 | //! |
| 6 | //! * Streaming-buffer parsing into a finalized `serde_json::Value` tool input |
| 7 | //! (`final_tool_input`, `parse_tool_input`, fenced/JSON segment helpers). |
| 8 | //! * The `multi_tool_use.parallel` payload parser. |
| 9 | //! * Policy predicates the turn loop consults — when a batch can run in |
| 10 | //! parallel and the small set of read-only MCP tools that are safe to run |
| 11 | //! in parallel. |
| 12 | //! * The tool execution plan/outcome types the batch driver passes around. |
| 13 | //! |
| 14 | //! All items are `pub(super)`-only: the public engine surface (Op/Event, |
| 15 | //! `EngineHandle`, `spawn_engine`) stays in `core/engine.rs`. |
| 16 | |
| 17 | use std::collections::HashMap; |
| 18 | |
| 19 | use serde_json::json; |
| 20 | |
| 21 | use crate::models::{Tool, ToolCaller}; |
| 22 | use crate::tools::spec::{ |
| 23 | ResourceClaim, ToolError, ToolExecutionOutcome, ToolResult, schedule_non_conflicting, |
| 24 | }; |
| 25 | |
| 26 | use super::ToolUseState; |
| 27 | use super::read_repeat_guard::{ReadRepeatGuard, ReadRepeatOccurrence}; |
| 28 | |
| 29 | // === Types ============================================================ |
| 30 | |
| 31 | #[allow(dead_code)] // `index` mirrors batch order for diagnostic ergonomics. |
| 32 | pub(super) struct ToolExecOutcome { |
| 33 | pub(super) index: usize, |
| 34 | pub(super) id: String, |
| 35 | pub(super) name: String, |
| 36 | pub(super) input: serde_json::Value, |
| 37 | pub(super) started_at: std::time::Instant, |
| 38 | pub(super) terminal: ToolExecutionOutcome, |
| 39 | } |
| 40 | |
| 41 | #[derive(Debug, Clone)] |
| 42 | pub(super) struct ToolExecutionPlan { |
| 43 | pub(super) index: usize, |
| 44 | pub(super) id: String, |
| 45 | pub(super) name: String, |
| 46 | pub(super) input: serde_json::Value, |
| 47 | pub(super) caller: Option<ToolCaller>, |
| 48 | pub(super) interactive: bool, |
| 49 | pub(super) approval_required: bool, |
| 50 | pub(super) approval_description: String, |
| 51 | pub(super) approval_force_prompt: bool, |
| 52 | pub(super) supports_parallel: bool, |
| 53 | pub(super) read_only: bool, |
| 54 | pub(super) detached_start: bool, |
| 55 | pub(super) resources: Vec<ResourceClaim>, |
| 56 | pub(super) blocked_error: Option<ToolError>, |
| 57 | pub(super) guard_result: Option<ToolResult>, |
| 58 | } |
| 59 | |
| 60 | pub(super) enum ToolExecutionBatch { |
| 61 | Parallel(Vec<ToolExecutionPlan>), |
| 62 | Serial(Box<ToolExecutionPlan>), |
| 63 | } |
| 64 | |
| 65 | pub(super) struct CoalescedReadPlan { |
| 66 | pub(super) leader_index: usize, |
| 67 | pub(super) follower: ToolExecutionPlan, |
| 68 | pub(super) occurrence: ReadRepeatOccurrence, |
| 69 | } |
| 70 | |
| 71 | pub(super) struct ReadRepeatExecutionPlan { |
| 72 | pub(super) executable: Vec<ToolExecutionPlan>, |
| 73 | pub(super) coalesced: Vec<CoalescedReadPlan>, |
| 74 | pub(super) occurrences: HashMap<usize, ReadRepeatOccurrence>, |
| 75 | } |
| 76 | |
| 77 | #[derive(Debug, serde::Serialize)] |
| 78 | pub(super) struct ParallelToolResultEntry { |
| 79 | pub(super) tool_name: String, |
| 80 | pub(super) success: bool, |
| 81 | pub(super) content: String, |
| 82 | #[serde(skip_serializing_if = "Option::is_none")] |
| 83 | pub(super) error: Option<String>, |
| 84 | } |
| 85 | |
| 86 | #[derive(Debug, serde::Serialize)] |
| 87 | pub(super) struct ParallelToolResult { |
| 88 | pub(super) results: Vec<ParallelToolResultEntry>, |
| 89 | } |
| 90 | |
| 91 | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 92 | pub(super) enum ToolApprovalStamp { |
| 93 | ApprovedByUser, |
| 94 | ApprovedWithPolicy, |
| 95 | } |
| 96 | |
| 97 | impl ToolApprovalStamp { |
| 98 | fn decision(self) -> &'static str { |
| 99 | match self { |
| 100 | Self::ApprovedByUser => "approved_by_user", |
| 101 | Self::ApprovedWithPolicy => "approved_with_policy", |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | fn model_visible_note(self) -> &'static str { |
| 106 | match self { |
| 107 | Self::ApprovedByUser => { |
| 108 | "[approval] This tool call required approval and was approved by the user before execution." |
| 109 | } |
| 110 | Self::ApprovedWithPolicy => { |
| 111 | "[approval] This tool call required approval and was approved by the user with an adjusted execution policy before execution." |
| 112 | } |
| 113 | } |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | pub(super) fn stamp_tool_result_approval(result: &mut ToolResult, approval: ToolApprovalStamp) { |
| 118 | let approval_metadata = json!({ |
| 119 | "required": true, |
| 120 | "decision": approval.decision(), |
| 121 | "model_visible": true, |
| 122 | }); |
| 123 | let metadata = result.metadata.get_or_insert_with(|| json!({})); |
| 124 | if let Some(object) = metadata.as_object_mut() { |
| 125 | object.insert("approval".to_string(), approval_metadata); |
| 126 | } else { |
| 127 | let prior = std::mem::replace(metadata, json!({})); |
| 128 | if let Some(object) = metadata.as_object_mut() { |
| 129 | object.insert("_prior".to_string(), prior); |
| 130 | object.insert("approval".to_string(), approval_metadata); |
| 131 | } |
| 132 | } |
| 133 | |
| 134 | let note = approval.model_visible_note(); |
| 135 | if result.content.starts_with("[approval] ") { |
| 136 | return; |
| 137 | } |
| 138 | if result.content.is_empty() { |
| 139 | result.content = note.to_string(); |
| 140 | } else { |
| 141 | result.content = format!("{note}\n\n{}", result.content); |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | // Hold the lock guard for the duration of a tool execution. |
| 146 | // The inner guards are held for RAII purposes (dropped when the guard is dropped). |
| 147 | pub(super) enum ToolExecGuard<'a> { |
| 148 | Read(#[allow(dead_code)] tokio::sync::RwLockReadGuard<'a, ()>), |
| 149 | Write(#[allow(dead_code)] tokio::sync::RwLockWriteGuard<'a, ()>), |
| 150 | } |
| 151 | |
| 152 | // === Caller policy and errors ======================================== |
| 153 | |
| 154 | pub(super) fn caller_type_for_tool_use(caller: Option<&ToolCaller>) -> &str { |
| 155 | caller.map_or("direct", |c| c.caller_type.as_str()) |
| 156 | } |
| 157 | |
| 158 | pub(super) fn caller_allowed_for_tool( |
| 159 | caller: Option<&ToolCaller>, |
| 160 | tool_def: Option<&Tool>, |
| 161 | ) -> bool { |
| 162 | let requested = caller_type_for_tool_use(caller); |
| 163 | if let Some(def) = tool_def |
| 164 | && let Some(allowed) = &def.allowed_callers |
| 165 | { |
| 166 | if allowed.is_empty() { |
| 167 | return requested == "direct"; |
| 168 | } |
| 169 | return allowed.iter().any(|item| item == requested); |
| 170 | } |
| 171 | requested == "direct" |
| 172 | } |
| 173 | |
| 174 | /// Whole-word check for "mode"/"modes" — a plain `contains("mode")` also |
| 175 | /// matched "model", letting provider model errors skip the actionable-hint |
| 176 | /// suffix (#3020). |
| 177 | fn mentions_mode_word(lower: &str) -> bool { |
| 178 | lower |
| 179 | .split(|ch: char| !ch.is_ascii_alphanumeric()) |
| 180 | .any(|word| word == "mode" || word == "modes") |
| 181 | } |
| 182 | |
| 183 | #[cfg(test)] |
| 184 | pub(super) fn format_tool_error(err: &ToolError, tool_name: &str) -> String { |
| 185 | format_tool_error_with_schema(err, tool_name, None) |
| 186 | } |
| 187 | |
| 188 | pub(super) fn format_tool_error_with_schema( |
| 189 | err: &ToolError, |
| 190 | tool_name: &str, |
| 191 | input_schema: Option<&serde_json::Value>, |
| 192 | ) -> String { |
| 193 | let message = match err { |
| 194 | ToolError::InvalidInput { message } => { |
| 195 | format!("Invalid input for tool '{tool_name}': {message}") |
| 196 | } |
| 197 | ToolError::MissingField { field } => { |
| 198 | format!("Tool '{tool_name}' is missing required field '{field}'") |
| 199 | } |
| 200 | ToolError::PathEscape { path } => format!( |
| 201 | "Path escapes workspace: {}. Use a workspace-relative path or enable trust mode.", |
| 202 | path.display() |
| 203 | ), |
| 204 | ToolError::ExecutionFailed { message } => message.clone(), |
| 205 | ToolError::Timeout { seconds } => format!( |
| 206 | "Tool '{tool_name}' timed out after {seconds}s. Try a narrower scope or a longer timeout." |
| 207 | ), |
| 208 | ToolError::Cancelled { message } => message.clone(), |
| 209 | ToolError::NotAvailable { message } => { |
| 210 | let lower = message.to_ascii_lowercase(); |
| 211 | // #3020: Pass through self-explanatory messages that already name the |
| 212 | // cause (mode switch, allow_shell, feature flag). Avoids appending a |
| 213 | // conflicting "Check mode, feature flags" suffix on top of |
| 214 | // "switch to Act mode" which already gives the recovery path. |
| 215 | if lower.contains("current tool catalog") |
| 216 | || lower.contains("did you mean:") |
| 217 | || mentions_mode_word(&lower) |
| 218 | || lower.contains("allow_shell") |
| 219 | || lower.contains("feature flag") |
| 220 | { |
| 221 | message.clone() |
| 222 | } else { |
| 223 | format!( |
| 224 | "Tool '{tool_name}' is not available: {message}. Check mode, feature flags, or tool name." |
| 225 | ) |
| 226 | } |
| 227 | } |
| 228 | ToolError::PermissionDenied { message } => { |
| 229 | let lower = message.to_ascii_lowercase(); |
| 230 | // #3020: Pass through messages that already name the denial cause. |
| 231 | if mentions_mode_word(&lower) |
| 232 | || lower.contains("allow_shell") |
| 233 | || lower.contains("denied by user") |
| 234 | { |
| 235 | message.clone() |
| 236 | } else { |
| 237 | format!( |
| 238 | "Tool '{tool_name}' was denied: {message}. Adjust approval mode or request permission." |
| 239 | ) |
| 240 | } |
| 241 | } |
| 242 | }; |
| 243 | |
| 244 | let message = with_transient_tool_fallback_hint(message, err, tool_name); |
| 245 | let (category, bad_field) = match err { |
| 246 | ToolError::InvalidInput { .. } => ("invalid_input", None), |
| 247 | ToolError::MissingField { field } => ("missing_field", Some(field.as_str())), |
| 248 | ToolError::PathEscape { .. } => ("path_escape", Some("path")), |
| 249 | ToolError::NotAvailable { .. } => ("tool_not_available", Some("tool_name")), |
| 250 | _ => return message, |
| 251 | }; |
| 252 | let valid_shape = input_schema.cloned().unwrap_or_else(|| { |
| 253 | serde_json::json!({ |
| 254 | "type": "object", |
| 255 | "guidance": format!("Use the advertised input schema for '{tool_name}'") |
| 256 | }) |
| 257 | }); |
| 258 | let feedback = serde_json::json!({ |
| 259 | "category": category, |
| 260 | "bad_field": bad_field, |
| 261 | "valid_shape": valid_shape, |
| 262 | "retryable": true, |
| 263 | "side_effect_status": "not_started" |
| 264 | }); |
| 265 | format!("{message}\nTool validation feedback: {feedback}") |
| 266 | } |
| 267 | |
| 268 | fn with_transient_tool_fallback_hint(message: String, err: &ToolError, tool_name: &str) -> String { |
| 269 | if message_already_has_recovery_hint(&message) { |
| 270 | return message; |
| 271 | } |
| 272 | |
| 273 | let Some(hint) = transient_tool_fallback_hint(err, tool_name, &message) else { |
| 274 | return message; |
| 275 | }; |
| 276 | |
| 277 | format!("{message} Fallback: {hint}") |
| 278 | } |
| 279 | |
| 280 | fn message_already_has_recovery_hint(message: &str) -> bool { |
| 281 | let lower = message.to_ascii_lowercase(); |
| 282 | lower.contains("recovery:") || lower.contains("fallback:") |
| 283 | } |
| 284 | |
| 285 | fn transient_tool_fallback_hint( |
| 286 | err: &ToolError, |
| 287 | tool_name: &str, |
| 288 | formatted_message: &str, |
| 289 | ) -> Option<&'static str> { |
| 290 | if !is_transient_tool_failure(err, formatted_message) { |
| 291 | return None; |
| 292 | } |
| 293 | |
| 294 | let lower_tool = tool_name.to_ascii_lowercase(); |
| 295 | if lower_tool.contains("web_search") |
| 296 | || lower_tool.contains("web_run") |
| 297 | || lower_tool == "web.run" |
| 298 | { |
| 299 | return Some( |
| 300 | "after one retry, switch to a direct URL/open/fetch path or cached context instead of repeating the same search.", |
| 301 | ); |
| 302 | } |
| 303 | |
| 304 | if lower_tool.contains("fetch_url") { |
| 305 | return Some( |
| 306 | "after one retry, try a narrower URL/source, use search results or cached context, or state the access limit instead of repeating the same request.", |
| 307 | ); |
| 308 | } |
| 309 | |
| 310 | if lower_tool.contains("file_search") || lower_tool.contains("grep") { |
| 311 | return Some( |
| 312 | "after one retry, narrow the query/path or inspect likely files directly instead of repeating the same search unchanged.", |
| 313 | ); |
| 314 | } |
| 315 | |
| 316 | if lower_tool.contains("exec_shell") |
| 317 | || lower_tool.contains("run_tests") |
| 318 | || lower_tool.contains("run_verifiers") |
| 319 | { |
| 320 | return Some( |
| 321 | "after one retry, narrow the command/scope, increase timeout only for expected long runs, or switch to file-level evidence.", |
| 322 | ); |
| 323 | } |
| 324 | |
| 325 | if lower_tool.contains("agent") { |
| 326 | return Some( |
| 327 | "after one retry, reduce delegated scope or continue in the parent context instead of repeatedly spawning the same agent.", |
| 328 | ); |
| 329 | } |
| 330 | |
| 331 | Some( |
| 332 | "after one retry, choose a different tool or narrower strategy instead of repeating the same call unchanged.", |
| 333 | ) |
| 334 | } |
| 335 | |
| 336 | fn is_transient_tool_failure(err: &ToolError, formatted_message: &str) -> bool { |
| 337 | if matches!(err, ToolError::Timeout { .. }) { |
| 338 | return true; |
| 339 | } |
| 340 | |
| 341 | if !matches!(err, ToolError::ExecutionFailed { .. }) { |
| 342 | return false; |
| 343 | } |
| 344 | |
| 345 | let lower = formatted_message.to_ascii_lowercase(); |
| 346 | [ |
| 347 | "timeout", |
| 348 | "timed out", |
| 349 | "request failed", |
| 350 | "connection", |
| 351 | "network", |
| 352 | "http 429", |
| 353 | "rate limit", |
| 354 | "http 5", |
| 355 | "anti-bot", |
| 356 | "captcha", |
| 357 | ] |
| 358 | .iter() |
| 359 | .any(|needle| lower.contains(needle)) |
| 360 | } |
| 361 | |
| 362 | // === Streaming-buffer parsing ========================================= |
| 363 | |
| 364 | /// Promote a streaming `ToolUseState` to a finalized JSON input. |
| 365 | /// |
| 366 | /// Order of preference: |
| 367 | /// |
| 368 | /// 1. `input_buffer` (the raw streamed delta concatenation) — parsed as |
| 369 | /// JSON. This is the most authoritative because it's what the model |
| 370 | /// actually emitted. |
| 371 | /// 2. `input` (the per-delta best-effort parse mirror) — used when the |
| 372 | /// buffer is empty (pre-streaming tool calls take this path). |
| 373 | /// 3. `input_buffer` non-empty but unparseable → fall back to `input` |
| 374 | /// (the per-delta parser has already mirrored the most recent valid |
| 375 | /// partial parse into `tool_state.input`). |
| 376 | pub(super) fn final_tool_input(state: &ToolUseState) -> serde_json::Value { |
| 377 | if state.input_parse_error.is_some() { |
| 378 | return malformed_tool_arguments_input(&state.input_buffer); |
| 379 | } |
| 380 | if !state.input_buffer.trim().is_empty() |
| 381 | && let Some(parsed) = parse_tool_input(&state.input_buffer) |
| 382 | { |
| 383 | return parsed; |
| 384 | } |
| 385 | state.input.clone() |
| 386 | } |
| 387 | |
| 388 | pub(super) fn parse_tool_input(buffer: &str) -> Option<serde_json::Value> { |
| 389 | let trimmed = buffer.trim(); |
| 390 | if trimmed.is_empty() { |
| 391 | return None; |
| 392 | } |
| 393 | // Try the deterministic arg-repair ladder first (handles trailing commas, |
| 394 | // unclosed braces, embedded control chars, etc.) |
| 395 | if let Ok(value) = crate::tools::arg_repair::repair(trimmed) { |
| 396 | return Some(value); |
| 397 | } |
| 398 | // Fall back to existing strategies for code-fenced, double-encoded, and |
| 399 | // segment-extraction patterns that the repair ladder doesn't cover. |
| 400 | if let Some(stripped) = strip_code_fences(trimmed) |
| 401 | && let Ok(value) = serde_json::from_str::<serde_json::Value>(&stripped) |
| 402 | { |
| 403 | return Some(value); |
| 404 | } |
| 405 | if let Ok(serde_json::Value::String(inner)) = serde_json::from_str::<serde_json::Value>(trimmed) |
| 406 | && let Ok(value) = serde_json::from_str::<serde_json::Value>(&inner) |
| 407 | { |
| 408 | return Some(value); |
| 409 | } |
| 410 | extract_json_segment(trimmed) |
| 411 | .and_then(|segment| serde_json::from_str::<serde_json::Value>(&segment).ok()) |
| 412 | } |
| 413 | |
| 414 | pub(super) fn malformed_tool_arguments_input(buffer: &str) -> serde_json::Value { |
| 415 | json!({ "raw_arguments": buffer }) |
| 416 | } |
| 417 | |
| 418 | pub(super) fn malformed_tool_arguments_error(buffer: &str) -> String { |
| 419 | format!("malformed tool arguments from model: expected valid JSON, got {buffer:?}") |
| 420 | } |
| 421 | |
| 422 | fn strip_code_fences(text: &str) -> Option<String> { |
| 423 | if !text.contains("```") { |
| 424 | return None; |
| 425 | } |
| 426 | let line_count = text.lines().count(); |
| 427 | let mut lines = Vec::with_capacity(line_count); |
| 428 | for line in text.lines() { |
| 429 | if line.trim_start().starts_with("```") { |
| 430 | continue; |
| 431 | } |
| 432 | lines.push(line); |
| 433 | } |
| 434 | let stripped = lines.join("\n"); |
| 435 | let stripped = stripped.trim(); |
| 436 | if stripped.is_empty() { |
| 437 | None |
| 438 | } else { |
| 439 | Some(stripped.to_string()) |
| 440 | } |
| 441 | } |
| 442 | |
| 443 | fn extract_json_segment(text: &str) -> Option<String> { |
| 444 | extract_balanced_segment(text, '{', '}').or_else(|| extract_balanced_segment(text, '[', ']')) |
| 445 | } |
| 446 | |
| 447 | fn extract_balanced_segment(text: &str, open: char, close: char) -> Option<String> { |
| 448 | let start = text.find(open)?; |
| 449 | let mut depth = 0i32; |
| 450 | let mut end = None; |
| 451 | for (offset, ch) in text[start..].char_indices() { |
| 452 | if ch == open { |
| 453 | depth += 1; |
| 454 | } else if ch == close { |
| 455 | depth -= 1; |
| 456 | if depth == 0 { |
| 457 | end = Some(start + offset + ch.len_utf8()); |
| 458 | break; |
| 459 | } |
| 460 | } |
| 461 | } |
| 462 | end.map(|end_idx| text[start..end_idx].to_string()) |
| 463 | } |
| 464 | |
| 465 | fn normalize_parallel_tool_name(raw: &str) -> String { |
| 466 | let mut name = raw.trim(); |
| 467 | for prefix in ["functions.", "tools.", "tool."] { |
| 468 | if let Some(stripped) = name.strip_prefix(prefix) { |
| 469 | name = stripped; |
| 470 | break; |
| 471 | } |
| 472 | } |
| 473 | name.to_string() |
| 474 | } |
| 475 | |
| 476 | pub(super) fn parse_parallel_tool_calls( |
| 477 | input: &serde_json::Value, |
| 478 | ) -> Result<Vec<(String, serde_json::Value)>, ToolError> { |
| 479 | let tool_uses = input |
| 480 | .get("tool_uses") |
| 481 | .and_then(|v| v.as_array()) |
| 482 | .ok_or_else(|| ToolError::missing_field("tool_uses"))?; |
| 483 | if tool_uses.is_empty() { |
| 484 | return Err(ToolError::invalid_input( |
| 485 | "multi_tool_use.parallel requires at least one tool call", |
| 486 | )); |
| 487 | } |
| 488 | |
| 489 | let mut calls = Vec::with_capacity(tool_uses.len()); |
| 490 | for item in tool_uses { |
| 491 | let name = item |
| 492 | .get("recipient_name") |
| 493 | .or_else(|| item.get("tool_name")) |
| 494 | .or_else(|| item.get("name")) |
| 495 | .or_else(|| item.get("tool")) |
| 496 | .and_then(|v| v.as_str()) |
| 497 | .ok_or_else(|| ToolError::missing_field("recipient_name"))?; |
| 498 | let params = item |
| 499 | .get("parameters") |
| 500 | .or_else(|| item.get("input")) |
| 501 | .or_else(|| item.get("args")) |
| 502 | .or_else(|| item.get("arguments")) |
| 503 | .cloned() |
| 504 | .unwrap_or_else(|| json!({})); |
| 505 | calls.push((normalize_parallel_tool_name(name), params)); |
| 506 | } |
| 507 | |
| 508 | Ok(calls) |
| 509 | } |
| 510 | |
| 511 | // === Dispatch policy ================================================== |
| 512 | |
| 513 | #[cfg(test)] |
| 514 | pub(super) fn should_parallelize_tool_batch(plans: &[ToolExecutionPlan]) -> bool { |
| 515 | if plans.is_empty() || !plans.iter().all(tool_plan_can_join_parallel_batch) { |
| 516 | return false; |
| 517 | } |
| 518 | schedule_non_conflicting( |
| 519 | plans |
| 520 | .iter() |
| 521 | .map(|plan| ((), plan.resources.clone())) |
| 522 | .collect(), |
| 523 | ) |
| 524 | .len() |
| 525 | == 1 |
| 526 | } |
| 527 | |
| 528 | pub(super) fn tool_plan_is_parallel_safe(plan: &ToolExecutionPlan) -> bool { |
| 529 | plan.read_only && plan.supports_parallel && !plan.approval_required && !plan.interactive |
| 530 | } |
| 531 | |
| 532 | pub(super) fn tool_plan_can_join_parallel_batch(plan: &ToolExecutionPlan) -> bool { |
| 533 | plan.blocked_error.is_none() |
| 534 | && (tool_plan_is_parallel_safe(plan) |
| 535 | || (plan.detached_start && !plan.approval_required && !plan.interactive)) |
| 536 | } |
| 537 | |
| 538 | /// Register finalized read-only calls and remove same-batch duplicates from |
| 539 | /// physical execution. Every removed follower is retained so the turn loop can |
| 540 | /// fan the leader's terminal result back out under the follower's own tool ID. |
| 541 | pub(super) fn plan_read_repeat_execution( |
| 542 | plans: Vec<ToolExecutionPlan>, |
| 543 | guard: &mut ReadRepeatGuard, |
| 544 | ) -> ReadRepeatExecutionPlan { |
| 545 | let mut executable = Vec::with_capacity(plans.len()); |
| 546 | let mut coalesced = Vec::new(); |
| 547 | let mut occurrences = HashMap::new(); |
| 548 | let mut leaders = HashMap::new(); |
| 549 | |
| 550 | for mut plan in plans { |
| 551 | let eligible = plan.read_only |
| 552 | && !plan.interactive |
| 553 | && !plan.detached_start |
| 554 | && plan.blocked_error.is_none() |
| 555 | && plan.guard_result.is_none(); |
| 556 | if !eligible { |
| 557 | // A write, interactive call, detached start, denial, or other |
| 558 | // execution barrier may change what a later read observes. Do not |
| 559 | // subscribe a post-barrier read to a pre-barrier result. |
| 560 | leaders.clear(); |
| 561 | executable.push(plan); |
| 562 | continue; |
| 563 | } |
| 564 | |
| 565 | let occurrence = guard.register(&plan.name, &plan.input); |
| 566 | occurrences.insert(plan.index, occurrence.clone()); |
| 567 | |
| 568 | if let Some(receipt) = guard.prior_receipt(&occurrence) { |
| 569 | plan.guard_result = Some(receipt); |
| 570 | executable.push(plan); |
| 571 | continue; |
| 572 | } |
| 573 | |
| 574 | if let Some(leader_index) = leaders.get(&occurrence.key).copied() { |
| 575 | coalesced.push(CoalescedReadPlan { |
| 576 | leader_index, |
| 577 | follower: plan, |
| 578 | occurrence, |
| 579 | }); |
| 580 | } else { |
| 581 | leaders.insert(occurrence.key.clone(), plan.index); |
| 582 | executable.push(plan); |
| 583 | } |
| 584 | } |
| 585 | |
| 586 | ReadRepeatExecutionPlan { |
| 587 | executable, |
| 588 | coalesced, |
| 589 | occurrences, |
| 590 | } |
| 591 | } |
| 592 | |
| 593 | pub(super) fn plan_tool_execution_batches( |
| 594 | plans: Vec<ToolExecutionPlan>, |
| 595 | ) -> Vec<ToolExecutionBatch> { |
| 596 | let mut batches = Vec::new(); |
| 597 | let mut parallel_candidates = Vec::new(); |
| 598 | |
| 599 | let flush_parallel = |parallel_candidates: &mut Vec<_>, |
| 600 | batches: &mut Vec<ToolExecutionBatch>| { |
| 601 | for chunk in schedule_non_conflicting(std::mem::take(parallel_candidates)) { |
| 602 | batches.push(ToolExecutionBatch::Parallel(chunk)); |
| 603 | } |
| 604 | }; |
| 605 | |
| 606 | for plan in plans { |
| 607 | if tool_plan_can_join_parallel_batch(&plan) { |
| 608 | let resources = plan.resources.clone(); |
| 609 | parallel_candidates.push((plan, resources)); |
| 610 | continue; |
| 611 | } |
| 612 | |
| 613 | flush_parallel(&mut parallel_candidates, &mut batches); |
| 614 | batches.push(ToolExecutionBatch::Serial(Box::new(plan))); |
| 615 | } |
| 616 | |
| 617 | flush_parallel(&mut parallel_candidates, &mut batches); |
| 618 | |
| 619 | batches |
| 620 | } |
| 621 | |
| 622 | pub(super) fn mcp_tool_is_parallel_safe(name: &str) -> bool { |
| 623 | matches!( |
| 624 | name, |
| 625 | "list_mcp_resources" |
| 626 | | "list_mcp_resource_templates" |
| 627 | | "mcp_read_resource" |
| 628 | | "read_mcp_resource" |
| 629 | | "mcp_get_prompt" |
| 630 | ) |
| 631 | } |
| 632 | |
| 633 | pub(super) fn mcp_tool_is_read_only(name: &str) -> bool { |
| 634 | matches!( |
| 635 | name, |
| 636 | "list_mcp_resources" |
| 637 | | "list_mcp_resource_templates" |
| 638 | | "mcp_read_resource" |
| 639 | | "read_mcp_resource" |
| 640 | | "mcp_get_prompt" |
| 641 | ) |
| 642 | } |
| 643 | |
| 644 | pub(super) fn mcp_tool_approval_description(name: &str) -> String { |
| 645 | if mcp_tool_is_read_only(name) { |
| 646 | format!("Read-only MCP tool '{name}'") |
| 647 | } else { |
| 648 | format!("MCP tool '{name}' may have side effects") |
| 649 | } |
| 650 | } |
| 651 |