Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions AUTHORS
Original file line number Diff line number Diff line change
Expand Up @@ -10,3 +10,4 @@ Bassam Adnan <mailbassam@gmail.com>
Saleh Enab <salehenab850@gmail.com>
Tejas Tyagi <tejastyagi.tt@gmail.com>
abishekcs <abishekdascs@gmail.com>
Youssef Ali <joe92228@gmail.com>
8 changes: 5 additions & 3 deletions doc/manual/en/30-tools.md
Original file line number Diff line number Diff line change
Expand Up @@ -387,9 +387,11 @@ answer, so the main conversation's context stays small. Each subagent:
`use_ssh`, which only copies files
- runs at the same time as the other subagents the model starts in the same
turn, up to four at once; their answers come back in the order the model
asked for them. Other tools in the turn still run one after another, in
order. Subagents that write share the workspace, so give concurrent writers
separate files
asked for them. Read-only tools the model starts together in one turn also
run at the same time, and `run_agent` asked to `list` counts as one of them;
a tool that writes, or a subagent beside a read, keeps the order the model
asked for. Subagents that write share the workspace, so give concurrent
writers separate files
- has a fixed budget of model requests, whatever `max_tool_rounds` says
- runs silently: no streaming; `/tasks` shows the tool it is on and the
tokens it has used
Expand Down
258 changes: 247 additions & 11 deletions src/session.rs
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,9 @@ fn render_tool_result(result: Result<String>) -> String {
}
}

/// Runs subagent `calls` concurrently, at most [`MAX_PARALLEL_SUBAGENTS`] at
/// a time, and returns their results in call order. Each runs on its own
/// clone of `tools` (clones share their state), and is aborted if the turn
/// is.
/// Runs `calls` concurrently, at most [`MAX_PARALLEL_SUBAGENTS`] at a time,
/// and returns their results in call order. Each runs on its own clone of
/// `tools` (clones share their state), and is aborted if the turn is.
async fn execute_concurrently(tools: &ToolExecutor, calls: &[crate::llm::ToolCall]) -> Vec<String> {
let calls = calls
.iter()
Expand All @@ -54,6 +53,77 @@ async fn execute_concurrently(tools: &ToolExecutor, calls: &[crate::llm::ToolCal
.await
}

/// What a run of calls that may overlap has in common.
///
/// A run is homogeneous: subagent starts overlap with subagent starts, and
/// read-only calls with read-only calls, but never across the two. A read
/// beside a writer would otherwise see a workspace the model had not asked it
/// to see yet.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum BatchKind {
Subagents,
ReadsOnly,
}

/// Whether `call` is local and read-only, so it may overlap another such
/// call. `run_agent` counts only when it lists agents.
fn is_read_only(tools: &ToolExecutor, call: &crate::llm::ToolCall) -> bool {
tools.parallel_safe(&call.function.name)
|| tools.lists_agents(&call.function.name, &call.function.arguments)
}

/// Whether `call` starts a subagent, as opposed to only listing agents.
fn starts_subagent(tools: &ToolExecutor, call: &crate::llm::ToolCall) -> bool {
tools.runs_subagent(&call.function.name)
&& !tools.lists_agents(&call.function.name, &call.function.arguments)
}

/// The end (exclusive) of the run starting at `start`, and what the run is,
/// or `None` when that call must run alone.
fn parallel_run(
tools: &ToolExecutor,
calls: &[crate::llm::ToolCall],
start: usize,
) -> Option<(usize, BatchKind)> {
let first = &calls[start];
let kind = if starts_subagent(tools, first) {
BatchKind::Subagents
} else if is_read_only(tools, first) && !tools.has_agent_hooks() {
BatchKind::ReadsOnly
} else {
return None;
};

let mut continued_tasks: Vec<&str> = Vec::new();
let mut end = start;
while end < calls.len() {
let call = &calls[end];
let same_kind = match kind {
BatchKind::Subagents => starts_subagent(tools, call),
BatchKind::ReadsOnly => is_read_only(tools, call),
};
if !same_kind {
break;
}
// Two continuations of one task are not independent: the first
// restart wins and the second finds it running. Keep them in order.
if call.function.name == "continue_agent" {
let task = call
.function
.arguments
.get("task")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
if continued_tasks.contains(&task) {
break;
}
continued_tasks.push(task);
}
end += 1;
}
Some((end, kind))
}

/// Runs `run` on every call, at most `limit` at a time, each as its own
/// task; the rendered results come back in call order. A task that panics
/// becomes an error result rather than taking the turn down.
Expand Down Expand Up @@ -619,13 +689,13 @@ impl ChatSession {
on_tool_running(true);
let mut index = 0;
while index < tool_calls.len() {
// Consecutive subagent calls are independent of
// each other: run them together. Everything else
// Consecutive calls of one kind are independent
// of each other: subagent starts, or read-only
// tools. Run those together; everything else
// runs in order, as the model asked.
let run_end = (index..tool_calls.len())
.find(|&i| !tools.runs_subagent(&tool_calls[i].function.name))
.unwrap_or(tool_calls.len());
if run_end - index >= 2 {
if let Some((run_end, _)) = parallel_run(tools, &tool_calls, index)
&& run_end - index >= 2
{
let batch = &tool_calls[index..run_end];
for tool_call in batch {
on_tool_call(tool_call);
Expand Down Expand Up @@ -719,7 +789,7 @@ impl std::error::Error for RoundsExhausted {}

#[cfg(test)]
mod tests {
use super::{ChatSession, run_concurrently, strip_stale_fragment};
use super::{BatchKind, ChatSession, parallel_run, run_concurrently, strip_stale_fragment};
use crate::config::LlmConfiguration;
use crate::context::fragments::ContextualFragment;
use crate::test_support::serve_chat_responses;
Expand Down Expand Up @@ -1268,6 +1338,172 @@ data: [DONE]
assert_eq!(server.join().expect("server").len(), 2);
}

fn tool_call(id: &str, name: &str, arguments: serde_json::Value) -> crate::llm::ToolCall {
crate::llm::ToolCall {
id: id.to_string(),
tool_type: "function".to_string(),
function: crate::llm::ToolCallFunction {
name: name.to_string(),
arguments: arguments
.as_object()
.cloned()
.unwrap_or_default()
.into_iter()
.collect(),
},
}
}

/// Read-only calls are independent, so a turn that asks for several files
/// runs them together. Anything that writes keeps the order the model
/// asked for.
#[test]
fn read_only_calls_overlap_but_writes_do_not() {
let workspace = tempfile::tempdir().unwrap();
let tools = crate::tools::ToolExecutor::new(workspace.path());

let reads = vec![
tool_call("a", "show_file", serde_json::json!({"path": "a.rs"})),
tool_call("b", "list_directory", serde_json::json!({})),
];
assert_eq!(
parallel_run(&tools, &reads, 0),
Some((2, BatchKind::ReadsOnly))
);

let mixed = vec![
tool_call("a", "show_file", serde_json::json!({"path": "a.rs"})),
tool_call("b", "create_file", serde_json::json!({"path": "b.rs"})),
];
assert_eq!(
parallel_run(&tools, &mixed, 0),
Some((1, BatchKind::ReadsOnly))
);
}

/// A read and a subagent in one turn are not overlapped: the subagent may
/// write, and the read has to see what the model asked it to see.
#[test]
fn a_read_and_a_subagent_do_not_overlap() {
let workspace = tempfile::tempdir().unwrap();
let tools = crate::tools::ToolExecutor::new(workspace.path());
let calls = vec![
tool_call("a", "show_file", serde_json::json!({"path": "a.rs"})),
tool_call("b", "explore_repository", serde_json::json!({"query": "x"})),
];
assert_eq!(
parallel_run(&tools, &calls, 0),
Some((1, BatchKind::ReadsOnly))
);
}

/// Two continuations of one task are one task's work, not two: the second
/// has to wait for the first. Different tasks may overlap.
#[test]
fn continuations_of_one_task_stay_in_order() {
let workspace = tempfile::tempdir().unwrap();
let tools = crate::tools::ToolExecutor::new(workspace.path());
let same = vec![
tool_call(
"a",
"continue_agent",
serde_json::json!({"task": "t1", "message": "one"}),
),
tool_call(
"b",
"continue_agent",
serde_json::json!({"task": "t1", "message": "two"}),
),
];
assert_eq!(
parallel_run(&tools, &same, 0),
Some((1, BatchKind::Subagents))
);

let different = vec![
tool_call(
"a",
"continue_agent",
serde_json::json!({"task": "t1", "message": "one"}),
),
tool_call(
"b",
"continue_agent",
serde_json::json!({"task": "t2", "message": "two"}),
),
];
assert_eq!(
parallel_run(&tools, &different, 0),
Some((2, BatchKind::Subagents))
);
}

/// `run_agent` with `agent: "list"` only searches the registry, so it
/// batches with other reads. Starting an agent still does not.
#[test]
fn listing_agents_is_read_only_but_starting_one_is_not() {
let workspace = tempfile::tempdir().unwrap();
let tools = crate::tools::ToolExecutor::new(workspace.path());

let listings = vec![
tool_call(
"a",
"run_agent",
serde_json::json!({"agent": "list", "task": "review"}),
),
tool_call("b", "show_file", serde_json::json!({"path": "a.rs"})),
];
assert_eq!(
parallel_run(&tools, &listings, 0),
Some((2, BatchKind::ReadsOnly))
);

// Case and surrounding spaces do not change what the call is.
let upper = vec![
tool_call(
"a",
"run_agent",
serde_json::json!({"agent": " LIST ", "task": "review"}),
),
tool_call("b", "list_directory", serde_json::json!({})),
];
assert_eq!(
parallel_run(&tools, &upper, 0),
Some((2, BatchKind::ReadsOnly))
);

let starting = vec![
tool_call(
"a",
"run_agent",
serde_json::json!({"agent": "reviewer", "task": "review"}),
),
tool_call("b", "show_file", serde_json::json!({"path": "a.rs"})),
];
assert_eq!(
parallel_run(&tools, &starting, 0),
Some((1, BatchKind::Subagents))
);

// A listing and a start are different kinds, so they stay ordered.
let mixed = vec![
tool_call(
"a",
"run_agent",
serde_json::json!({"agent": "list", "task": "review"}),
),
tool_call(
"b",
"run_agent",
serde_json::json!({"agent": "reviewer", "task": "review"}),
),
];
assert_eq!(
parallel_run(&tools, &mixed, 0),
Some((1, BatchKind::ReadsOnly))
);
}

/// Running out of rounds is a typed error. By default the turn is rolled
/// back; a session told to keep it still has the work in its transcript,
/// which is what a subagent's partial answer is made from.
Expand Down
44 changes: 44 additions & 0 deletions src/tools.rs
Original file line number Diff line number Diff line change
Expand Up @@ -961,6 +961,50 @@ impl ToolExecutor {
|| crate::subagents::find(name).is_some()
}

/// Whether `name` is a read-only tool that may share one turn with
/// another such call.
///
/// A name-level allowlist: a tool whose safety depends on its arguments
/// has its own test (see [`Self::lists_agents`]). `browse` is stateful
/// (link numbers refer to the page it last showed), the graph tools may
/// build or refresh a cache, and MCP tools have unknown side effects, so
/// none of those qualify. Subagent starters are [`Self::runs_subagent`]'s
/// business, not this list's.
pub fn parallel_safe(&self, name: &str) -> bool {
matches!(
name,
"show_file"
| "list_directory"
| "search_text"
| "expand_context"
| "fetch_url"
| "web_search"
| "use_skill"
)
}

/// Whether this call asks `run_agent` to list matching agents. That is a
/// local search, not a subagent start, so it may share a turn with other
/// read-only calls.
pub fn lists_agents(
&self,
name: &str,
arguments: &std::collections::HashMap<String, serde_json::Value>,
) -> bool {
name == "run_agent"
&& arguments
.get("agent")
.and_then(serde_json::Value::as_str)
.is_some_and(|agent| agent.trim().eq_ignore_ascii_case("list"))
}

/// Whether a hook runs around every tool call in this executor. Hooks can
/// observe call order, so the chat loop does not overlap otherwise
/// parallel-safe calls when one is installed.
pub fn has_agent_hooks(&self) -> bool {
self.agent_hooks.is_some()
}

/// Lets this conversation's subagents run in the background: an
/// interactive client, which hands their reports to the model later.
pub fn with_background_tasks(mut self) -> Self {
Expand Down