diff --git a/Cargo.lock b/Cargo.lock index abd0dc1..1a5eef8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4890,6 +4890,7 @@ dependencies = [ "anyhow", "base64", "chrono", + "futures-util", "serde", "serde_json", "sha2 0.11.0", diff --git a/apps/application/Cargo.toml b/apps/application/Cargo.toml index fc4954d..7ea8aa7 100644 --- a/apps/application/Cargo.toml +++ b/apps/application/Cargo.toml @@ -16,6 +16,7 @@ uuid.workspace = true anyhow.workspace = true tracing.workspace = true tokio.workspace = true +futures-util.workspace = true base64.workspace = true sha2.workspace = true url.workspace = true diff --git a/apps/application/src/agent/mod.rs b/apps/application/src/agent/mod.rs index f606543..99ce86f 100644 --- a/apps/application/src/agent/mod.rs +++ b/apps/application/src/agent/mod.rs @@ -11,6 +11,17 @@ pub trait ToolExecutor: Send + Sync { tool_name: &str, args: &serde_json::Value, ) -> impl Future> + Send; + + /// Whether a tool is *read-only* and therefore safe to run concurrently + /// with other read-only tools in the same assistant message. + /// + /// Defaults to `false` (sequential) so a caller that does not know the + /// tool surface stays conservative. Concrete executors that know their + /// tools override this — e.g. return `true` for `read`/`grep`/`glob`. + fn is_parallel_safe(&self, tool_name: &str) -> bool { + let _ = tool_name; + false + } } /// Service for running agent turns asynchronously. diff --git a/apps/application/src/agent/turn_service.rs b/apps/application/src/agent/turn_service.rs index 5c3de56..431e5ca 100644 --- a/apps/application/src/agent/turn_service.rs +++ b/apps/application/src/agent/turn_service.rs @@ -187,6 +187,53 @@ async fn execute_tool_call( output } +// --------------------------------------------------------------------------- +// Helper: bounded-parallel execution of read-only tool calls. +// --------------------------------------------------------------------------- + +/// Maximum number of read-only tool calls executed concurrently in a single +/// assistant batch. Models rarely emit more than a handful of reads per +/// message; this cap keeps resource usage bounded while still removing the +/// serial round-trip latency of many independent lookups. +const MAX_PARALLEL_TOOLS: usize = 8; + +fn tool_executor_ref(tool_executor: &T) -> &T { + tool_executor +} + +/// Execute a batch of *read-only* tool calls concurrently (bounded by +/// [`MAX_PARALLEL_TOOLS`]) and return their outputs **in the original call +/// order**. +/// +/// Order preservation matters: OpenAI/Anthropic tool-calling contracts expect +/// tool-result messages to appear in the same order as the `tool_calls` +/// emitted in the assistant message. Without it, the model sees shuffled +/// results and loses track of which result belongs to which call. +/// +/// Each call still pushes its `TurnEvent::ToolResult` (so the TUI shows each +/// tool as it completes) but the returned `Vec` is ordered by the input index. +async fn execute_tool_calls_in_parallel( + tool_executor: &T, + turn_events: &Arc>>, + tool_calls: &[zesdex_domain::core::ToolCall], +) -> Vec { + let semaphore = Arc::new(tokio::sync::Semaphore::new(MAX_PARALLEL_TOOLS)); + let executor_ref = tool_executor_ref(tool_executor); + + let futures = tool_calls.iter().map(|tc| { + let tc = tc.clone(); + let events = turn_events.clone(); + let sem = semaphore.clone(); + async move { + // Acquire a permit to bound concurrency across the batch. + let _permit = sem.acquire_owned().await; + execute_tool_call(executor_ref, &events, &tc).await + } + }); + + futures_util::future::join_all(futures).await +} + // --------------------------------------------------------------------------- // Helper: emit usage event from optional LLM response metadata. // --------------------------------------------------------------------------- @@ -384,10 +431,42 @@ impl super::AgentTurnService for AgentTurnS params.messages.push(assistant_msg); // ── Execute each tool call ────────────────────────── - for tc in &tool_calls { - let output = - execute_tool_call(self.tool_executor.as_ref(), ¶ms.turn_events, tc) - .await; + // + // If the whole batch is made of *read-only* tools + // (read/grep/glob/…), run it concurrently with bounded + // parallelism — a big latency win for coding turns that + // emit several independent lookups in one message. Any + // single mutating tool forces the whole batch back to the + // safe sequential path so writes never race. + // + // Results are always collected in the original call order + // to honour the tool-calling contract. + let parallel = tool_calls.len() > 1 + && tool_calls + .iter() + .all(|tc| self.tool_executor.is_parallel_safe(&tc.function.name)); + let outputs: Vec = if parallel { + execute_tool_calls_in_parallel( + self.tool_executor.as_ref(), + ¶ms.turn_events, + &tool_calls, + ) + .await + } else { + let mut sequential = Vec::with_capacity(tool_calls.len()); + for tc in &tool_calls { + let out = execute_tool_call( + self.tool_executor.as_ref(), + ¶ms.turn_events, + tc, + ) + .await; + sequential.push(out); + } + sequential + }; + + for (tc, output) in tool_calls.iter().zip(outputs) { if output.starts_with("Error:") { errors.record(&tc.function.name, &output, &mut params.messages); } @@ -479,7 +558,6 @@ pub async fn compact_messages_with_ai( #[cfg(test)] mod tests { use super::*; - #[test] fn truncate_short_output_is_unchanged() { let out = "short".to_string(); @@ -536,4 +614,83 @@ mod tests { ]; assert_eq!(conversation_chars(&messages), 3 + 11 + 6); } + + /// A fake executor that reports parallel-safety for read-only tools and + /// whose `execute` sleeps on the first call to prove the batch runs + /// concurrently (a sequential loop would pay the sleep per call). + struct FakeExecutor; + + impl ToolExecutor for FakeExecutor { + async fn execute(&self, name: &str, _args: &serde_json::Value) -> anyhow::Result { + if name == "read" { + // 30ms sleep on every read; a parallel batch of 3 would + // finish in ~30ms instead of ~90ms sequentially. + tokio::time::sleep(std::time::Duration::from_millis(30)).await; + } + Ok(format!("out:{name}")) + } + + fn is_parallel_safe(&self, name: &str) -> bool { + matches!(name, "read" | "grep") + } + } + + fn tc(name: &str, id: usize) -> zesdex_domain::core::ToolCall { + zesdex_domain::core::ToolCall { + id: format!("call_{id}"), + type_: "function".to_string(), + function: zesdex_domain::core::ToolFunction { + name: name.to_string(), + arguments: serde_json::Value::String(String::new()), + }, + } + } + + #[test] + fn parallel_batch_runs_concurrently_and_preserves_order() { + let executor = FakeExecutor; + let events = Arc::new(Mutex::new(VecDeque::new())); + let calls = vec![tc("read", 1), tc("grep", 2), tc("read", 3)]; + + // All three are parallel-safe. + assert!(calls + .iter() + .all(|c| executor.is_parallel_safe(&c.function.name))); + + let rt = tokio::runtime::Builder::new_multi_thread() + .enable_time() + .build() + .unwrap(); + let started = std::time::Instant::now(); + let outputs = rt.block_on(execute_tool_calls_in_parallel(&executor, &events, &calls)); + let elapsed = started.elapsed(); + + // Results are in *original* call order (read, grep, read). + assert_eq!( + outputs, + vec![ + "out:read".to_string(), + "out:grep".to_string(), + "out:read".to_string() + ] + ); + // Two reads sleep 30ms each; sequential would take ~60ms+ for the + // two reads, parallel keeps the whole batch under 60ms. + assert!( + elapsed < std::time::Duration::from_millis(60), + "batch took {elapsed:?}, expected parallel execution" + ); + assert!(elapsed >= std::time::Duration::from_millis(25)); + } + + #[test] + fn mutating_batch_falls_back_to_sequential_path() { + // A batch containing a mutating tool is not eligible for the parallel + // path, so the main loop keeps results ordered and side-effects safe. + let executor = FakeExecutor; + let calls = [tc("read", 1), tc("edit", 2)]; + assert!(!calls + .iter() + .all(|c| executor.is_parallel_safe(&c.function.name))); + } } diff --git a/apps/infrastructure/src/tools/executor.rs b/apps/infrastructure/src/tools/executor.rs index 4580e8f..10f0e29 100644 --- a/apps/infrastructure/src/tools/executor.rs +++ b/apps/infrastructure/src/tools/executor.rs @@ -15,6 +15,13 @@ impl InfrastructureToolExecutor { tools: all_tools(), } } + + /// Whether a tool is read-only and safe to execute concurrently with + /// other parallel-safe tools. Delegates to the registry so the main + /// turn loop and subagent engine share one source of truth. + pub fn is_parallel_safe(tool_name: &str) -> bool { + crate::tools::tool_is_parallel_safe(tool_name) + } } impl ToolExecutor for InfrastructureToolExecutor { @@ -37,4 +44,8 @@ impl ToolExecutor for InfrastructureToolExecutor { } } } + + fn is_parallel_safe(&self, tool_name: &str) -> bool { + Self::is_parallel_safe(tool_name) + } } diff --git a/apps/infrastructure/src/tools/mod.rs b/apps/infrastructure/src/tools/mod.rs index c7d2e03..a7064e3 100644 --- a/apps/infrastructure/src/tools/mod.rs +++ b/apps/infrastructure/src/tools/mod.rs @@ -59,7 +59,7 @@ pub mod workflow; // like `crate::tools::{Tool, ToolCtx}` continue to work. pub use context::{ToolCtx, ToolCtxBuilder}; pub use graduated::{check_graduated_checks, GraduatedCheck}; -pub use registry::{all_tools, tool_defs, tool_is_risky}; +pub use registry::{all_tools, tool_defs, tool_is_parallel_safe, tool_is_risky}; pub use util::{arg_str, execute_cmd, log_write_edit_tool, resolve_path}; /// Common interface every agent-invocable tool implements. diff --git a/apps/infrastructure/src/tools/registry.rs b/apps/infrastructure/src/tools/registry.rs index 1eca55c..d3a0d54 100644 --- a/apps/infrastructure/src/tools/registry.rs +++ b/apps/infrastructure/src/tools/registry.rs @@ -52,6 +52,34 @@ pub fn tool_is_risky(name: &str) -> bool { matches!(name, "write" | "delete" | "edit" | "bash" | "git_operator") } +/// Whether a tool by name is read-only and therefore safe to run *in +/// parallel* with other tool calls within the same assistant message. +/// +/// Read-only tools only inspect the workspace (read files, grep, glob, +/// semantic search, list symbols, recall memory, web search, directory +/// listing). They have no side effects, so concurrent execution cannot +/// create data races or conflicting writes. +/// +/// Everything else (edits, writes, deletes, shell, git, planning, memory +/// writes, agent/spawn orchestration) stays sequential to preserve +/// correctness. +pub fn tool_is_parallel_safe(name: &str) -> bool { + matches!( + name, + "read" + | "grep" + | "glob" + | "semantic_search" + | "list_symbols" + | "web_search" + | "recall" + | "dir_list" + | "pong" + | "seq_think" + | "dir_cache_update" + ) +} + /// Convert a list of tools into provider-facing `ToolDef` request schema. pub fn tool_defs(tools: &[Box]) -> Vec { tools @@ -66,3 +94,59 @@ pub fn tool_defs(tools: &[Box]) -> Vec