perf(agent): eksekusi tool read-only paralel seperti Claude Code

Sebelumnya loop utama mengeksekusi semua tool call satu-per-satu
(sequential for loop). Seperti Claude Code, tool read-only yang
independen dalam satu pesan assistant (read/grep/glob/semantic_search
dsb.) kini dijalankan konkuren dengan bounded parallelism (max 8),
mengurangi latensi per turn secara signifikan untuk beban coding.

- feat(registry): tool_is_parallel_safe() — whitelist tool read-only
  yang aman dijalankan paralel; tool mutating/shell tetap sequential
- feat(executor): is_parallel_safe() delegasi ke registry; ToolExecutor
  trait Default=false (konservatif)
- fix(application): execute_tool_calls_in_parallel() — join_all +
  semaphore bounded 8, hasil dikumpulkan dalam URUTAN panggilan asli
  (kontrak OpenAI/Anthropic tool-result ordering)
- loop utama: batch paralel hanya jika SEMUA tool parallel-safe; jika
  ada satu tool mutating, jatuh balik ke jalur sequential aman
- test: +2 registry test, +2 application test (konkurensi & urutan,
  fallback batch mutating)

Cargo.lock

@@ -4890,6 +4890,7 @@ dependencies = [
  "anyhow",
  "base64",
  "chrono",
+ "futures-util",
  "serde",
  "serde_json",
  "sha2 0.11.0",

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

apps/application/src/agent/mod.rs

@@ -11,6 +11,17 @@ pub trait ToolExecutor: Send + Sync {
         tool_name: &str,
         args: &serde_json::Value,
     ) -> impl Future<Output = Result<String>> + 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.

apps/application/src/agent/turn_service.rs

@@ -187,6 +187,53 @@ async fn execute_tool_call<T: ToolExecutor>(
     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<T: ToolExecutor>(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<T: ToolExecutor>(
+    tool_executor: &T,
+    turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
+    tool_calls: &[zesdex_domain::core::ToolCall],
+) -> Vec<String> {
+    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<P: ProviderService, T: ToolExecutor> super::AgentTurnService for AgentTurnS
                     params.messages.push(assistant_msg);
 
                     // ── Execute each tool call ──────────────────────────
-                    for tc in &tool_calls {
+                    //
-                        let output =
+                    // If the whole batch is made of *read-only* tools
-                            execute_tool_call(self.tool_executor.as_ref(), &params.turn_events, tc)
+                    // (read/grep/glob/…), run it concurrently with bounded
-                                .await;
+                    // 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<String> = if parallel {
+                        execute_tool_calls_in_parallel(
+                            self.tool_executor.as_ref(),
+                            &params.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(),
+                                &params.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<P: ProviderService>(
 #[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<String> {
+            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)));
+    }
 }

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)
+    }
 }

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.

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<dyn super::Tool>]) -> Vec<zesdex_domain::core::ToolDef> {
     tools
@@ -66,3 +94,59 @@ pub fn tool_defs(tools: &[Box<dyn super::Tool>]) -> Vec<zesdex_domain::core::Too
         })
         .collect()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn read_only_tools_are_parallel_safe() {
+        for name in [
+            "read",
+            "grep",
+            "glob",
+            "semantic_search",
+            "list_symbols",
+            "web_search",
+            "recall",
+            "dir_list",
+            "pong",
+            "seq_think",
+            "dir_cache_update",
+        ] {
+            assert!(
+                tool_is_parallel_safe(name),
+                "{name} should be parallel-safe"
+            );
+        }
+    }
+
+    #[test]
+    fn mutating_and_shell_tools_are_not_parallel_safe() {
+        for name in [
+            "edit",
+            "write",
+            "delete",
+            "bash",
+            "git_operator",
+            "git_worktree",
+            "remember",
+            "forget",
+            "todowrite",
+            "todofinish",
+            "plan_enter",
+            "plan_ready",
+            "workflow_run",
+            "hive_mind",
+            "spawn_agents",
+            "spawn_pipeline",
+            "parallel_delegate",
+            "explore_codebase",
+        ] {
+            assert!(
+                !tool_is_parallel_safe(name),
+                "{name} should NOT be parallel-safe"
+            );
+        }
+    }
+}