import claude/types/content.{type ContentBlock, ToolUse} import claude/types/tool.{type Registry} import gleam/erlang/process import gleam/list import gleam/string /// Result type for a single tool execution: (tool_use_id, result_or_error). pub type ToolResult = #(String, Result(String, String)) /// Internal type to distinguish success messages from crash notifications /// when collecting tool results via a selector. type ToolMessage { ToolSuccess(id: String, result: Result(String, String)) ToolCrashed(reason: process.ExitReason) } /// Execute tool calls concurrently using BEAM processes. /// /// Spawns one unlinked, monitored process per tool call, then collects /// results with a timeout. Results are returned in the same order as the /// input tool_calls. Non-ToolUse blocks in the input list are silently /// ignored. /// /// If a tool handler exceeds the timeout, an Error("Tool execution timed out") /// is returned for that tool. If a tool handler crashes, the monitor detects /// the exit and an Error describing the crash reason is returned immediately /// rather than waiting for the full timeout. pub fn execute_concurrent( tool_calls tool_calls: List(ContentBlock), tools tools: Registry, timeout_ms timeout: Int, ) -> List(ToolResult) { // Filter to only ToolUse blocks let calls = list.filter_map(tool_calls, fn(block) { case block { ToolUse(id: id, name: name, input: input) -> Ok(#(id, name, input)) _ -> Error(Nil) } }) // Spawn an unlinked, monitored process for each tool call let tasks = list.map(calls, fn(call) { let #(id, name, input) = call let subject = process.new_subject() // Use spawn_unlinked so a crash does not propagate to the caller let pid = process.spawn_unlinked(fn() { let result = tool.dispatch(tools, name, input) process.send(subject, #(id, result)) }) // Monitor the process so we receive a Down message if it crashes let mon = process.monitor(pid) #(id, subject, mon) }) // Collect results with timeout, maintaining order. // For each task, build a selector that listens for either: // 1. A successful result on the subject, or // 2. A Down message from the monitor (process crashed) list.map(tasks, fn(task) { let #(id, subject, mon) = task let selector = process.new_selector() |> process.select_map(subject, fn(msg) { let #(tool_id, tool_result) = msg ToolSuccess(tool_id, tool_result) }) |> process.select_specific_monitor(mon, fn(down) { ToolCrashed(down.reason) }) let result = case process.selector_receive(selector, timeout) { Ok(ToolSuccess(_, tool_result)) -> #(id, tool_result) Ok(ToolCrashed(reason)) -> #(id, Error( "Tool execution crashed: " <> format_exit_reason(reason), )) Error(Nil) -> #(id, Error("Tool execution timed out")) } // Clean up: demonitor to avoid stale Down messages in the mailbox process.demonitor_process(mon) result }) } /// Format an exit reason into a human-readable string. fn format_exit_reason(reason: process.ExitReason) -> String { case reason { process.Normal -> "normal exit" process.Killed -> "killed" process.Abnormal(reason) -> "abnormal: " <> string.inspect(reason) } }