agent: resource-based parallel tool execution
Replace binary ReadOnly/not batching with path-based conflict detection. Tool calls that operate on different file paths now execute in parallel, while same-path writes and shell (global barrier) serialize. Conflict rules: - ReadOnly tools: never conflict (unchanged from before) - Write tools with path arg: conflict only on same path - Shell / unknown (no path): global barrier, runs alone This parallelizes the common refactoring case where the model edits multiple files in one turn (e.g., 5 file_edits on different files complete in 1 round-trip instead of 5). Safety argument: - LLM API is turn-based: model reads files first, plans edits based on known content, emits writes in a subsequent turn - file_edit carries old_string context: self-contained per-file - Same-path conflicts serialize (detected via extractFilePath) - Shell is conservative: always a full serialization barrier
This commit is contained in:
Ollie Agent
parent
bb58c42f4f
commit
3dccb36e82
|
|
@ -343,8 +343,10 @@ func streamResponse(rt *Runtime, ctx TurnCtx, history []backend.Message, respons
|
|||
|
||||
// ── execToolCalls ───────────────────────────────────────────────────────────
|
||||
|
||||
// execToolCalls dispatches all tool calls, running consecutive read-safe
|
||||
// calls in parallel. Returns the results and whether the loop was interrupted.
|
||||
// execToolCalls dispatches all tool calls, grouping non-conflicting calls
|
||||
// into parallel batches. Conflict detection uses file paths from tool
|
||||
// arguments: writes on the same path serialize, writes on different paths
|
||||
// parallelize, shell/unknown is a full serialization barrier.
|
||||
func execToolCalls(rt *Runtime, ctx TurnCtx, toolCalls []backend.ToolCall, resultCache *sync.Map) ([]toolResult, bool) {
|
||||
results := make([]toolResult, 0, len(toolCalls))
|
||||
|
||||
|
|
@ -365,24 +367,30 @@ func execToolCalls(rt *Runtime, ctx TurnCtx, toolCalls []backend.ToolCall, resul
|
|||
}
|
||||
}
|
||||
|
||||
isParallelSafe := func(name string) bool {
|
||||
return name != "" && rt.ToolMeta[name].ReadOnly
|
||||
}
|
||||
|
||||
for i := 0; i < len(toolCalls); {
|
||||
if ctx.Err() != nil {
|
||||
fillCancelled(toolCalls[i:])
|
||||
return results, true
|
||||
}
|
||||
|
||||
// Collect a run of consecutive parallel-safe calls.
|
||||
// Build a batch of non-conflicting calls using greedy grouping.
|
||||
batch := []backend.ToolCall{toolCalls[i]}
|
||||
batchWrites := toolConflictKeys(rt, toolCalls[i])
|
||||
j := i + 1
|
||||
if isParallelSafe(toolCalls[i].Name) {
|
||||
for j < len(toolCalls) && isParallelSafe(toolCalls[j].Name) {
|
||||
j++
|
||||
|
||||
for j < len(toolCalls) {
|
||||
// A global barrier seals the batch — nothing else can join.
|
||||
if hasGlobal(batchWrites) {
|
||||
break
|
||||
}
|
||||
keys := toolConflictKeys(rt, toolCalls[j])
|
||||
if conflictsWithSet(keys, batchWrites) {
|
||||
break
|
||||
}
|
||||
batch = append(batch, toolCalls[j])
|
||||
batchWrites = mergeKeys(batchWrites, keys)
|
||||
j++
|
||||
}
|
||||
batch := toolCalls[i:j]
|
||||
|
||||
if len(batch) == 1 {
|
||||
tr, wasInt := execOne(rt, ctx, batch[0], resultCache)
|
||||
|
|
@ -405,6 +413,62 @@ func execToolCalls(rt *Runtime, ctx TurnCtx, toolCalls []backend.ToolCall, resul
|
|||
return results, false
|
||||
}
|
||||
|
||||
// conflictKeyGlobal is a sentinel indicating a tool conflicts with everything.
|
||||
const conflictKeyGlobal = "\x00GLOBAL"
|
||||
|
||||
// toolConflictKeys returns the set of resource keys a tool call touches.
|
||||
// ReadOnly tools return nil (never conflict). Write tools return their file
|
||||
// path. Shell and unknown tools return the global sentinel.
|
||||
func toolConflictKeys(rt *Runtime, call backend.ToolCall) []string {
|
||||
if rt.ToolMeta[call.Name].ReadOnly {
|
||||
return nil // reads never conflict
|
||||
}
|
||||
// File tools with a path argument: conflict on that path
|
||||
if p := extractFilePath(call.Arguments); p != "" {
|
||||
return []string{p}
|
||||
}
|
||||
// Shell or tools without a detectable path: global barrier
|
||||
return []string{conflictKeyGlobal}
|
||||
}
|
||||
|
||||
// conflictsWithSet returns true if keys conflicts with the accumulated set.
|
||||
// nil keys (ReadOnly) never conflict. Global sentinel conflicts with any
|
||||
// non-empty set. Path keys conflict if the same path exists in the set.
|
||||
func conflictsWithSet(keys []string, set []string) bool {
|
||||
if len(keys) == 0 {
|
||||
return false // ReadOnly: never conflicts
|
||||
}
|
||||
if len(set) == 0 {
|
||||
return false // nothing accumulated yet
|
||||
}
|
||||
for _, k := range keys {
|
||||
if k == conflictKeyGlobal {
|
||||
return true // global conflicts with any non-empty set
|
||||
}
|
||||
for _, s := range set {
|
||||
if s == conflictKeyGlobal || s == k {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// mergeKeys appends keys into set (no dedup needed for small sets).
|
||||
func mergeKeys(set, keys []string) []string {
|
||||
return append(set, keys...)
|
||||
}
|
||||
|
||||
// hasGlobal returns true if set contains the global barrier sentinel.
|
||||
func hasGlobal(set []string) bool {
|
||||
for _, s := range set {
|
||||
if s == conflictKeyGlobal {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// execBatch fans out a batch of parallel-safe calls, deduplicating identical ones.
|
||||
func execBatch(rt *Runtime, ctx TurnCtx, batch []backend.ToolCall, resultCache *sync.Map) ([]toolResult, bool) {
|
||||
type inflightResult struct {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,191 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"ollie/cmd/olliesrv/internal/backend"
|
||||
"ollie/toolsrv"
|
||||
)
|
||||
|
||||
func TestToolConflictKeys(t *testing.T) {
|
||||
rt := &Runtime{
|
||||
ToolMeta: map[string]toolsrv.ToolInfo{
|
||||
"file_read": {Name: "file_read", ReadOnly: true},
|
||||
"file_grep": {Name: "file_grep", ReadOnly: true},
|
||||
"file_edit": {Name: "file_edit", ReadOnly: false},
|
||||
"file_write": {Name: "file_write", ReadOnly: false},
|
||||
"shell": {Name: "shell", ReadOnly: false},
|
||||
},
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
call backend.ToolCall
|
||||
want []string
|
||||
}{
|
||||
{"read-only returns nil", backend.ToolCall{Name: "file_read", Arguments: json.RawMessage(`{"path":"/a"}`)}, nil},
|
||||
{"write with path", backend.ToolCall{Name: "file_edit", Arguments: json.RawMessage(`{"path":"/a","old_string":"x","new_string":"y"}`)}, []string{"/a"}},
|
||||
{"shell is global", backend.ToolCall{Name: "shell", Arguments: json.RawMessage(`{"cmd":"ls"}`)}, []string{conflictKeyGlobal}},
|
||||
{"write no path is global", backend.ToolCall{Name: "file_write", Arguments: json.RawMessage(`{"content":"hi"}`)}, []string{conflictKeyGlobal}},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := toolConflictKeys(rt, tt.call)
|
||||
if len(got) != len(tt.want) {
|
||||
t.Fatalf("got %v, want %v", got, tt.want)
|
||||
}
|
||||
for i := range got {
|
||||
if got[i] != tt.want[i] {
|
||||
t.Errorf("got[%d]=%q, want %q", i, got[i], tt.want[i])
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestConflictsWithSet(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
keys []string
|
||||
set []string
|
||||
want bool
|
||||
}{
|
||||
{"nil keys never conflict", nil, []string{"/a"}, false},
|
||||
{"empty set never conflicts", []string{"/a"}, nil, false},
|
||||
{"same path conflicts", []string{"/a"}, []string{"/a"}, true},
|
||||
{"different paths no conflict", []string{"/b"}, []string{"/a"}, false},
|
||||
{"global conflicts with any", []string{conflictKeyGlobal}, []string{"/a"}, true},
|
||||
{"any conflicts with global", []string{"/a"}, []string{conflictKeyGlobal}, true},
|
||||
{"global vs global", []string{conflictKeyGlobal}, []string{conflictKeyGlobal}, true},
|
||||
{"multiple paths partial conflict", []string{"/b"}, []string{"/a", "/b"}, true},
|
||||
{"multiple paths no conflict", []string{"/c"}, []string{"/a", "/b"}, false},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := conflictsWithSet(tt.keys, tt.set)
|
||||
if got != tt.want {
|
||||
t.Errorf("conflictsWithSet(%v, %v) = %v, want %v", tt.keys, tt.set, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecToolCallsBatching(t *testing.T) {
|
||||
// Verify the batching logic groups correctly without actually executing.
|
||||
// We test by inspecting which calls would be grouped together.
|
||||
rt := &Runtime{
|
||||
ToolMeta: map[string]toolsrv.ToolInfo{
|
||||
"file_read": {Name: "file_read", ReadOnly: true},
|
||||
"file_edit": {Name: "file_edit", ReadOnly: false},
|
||||
"file_write": {Name: "file_write", ReadOnly: false},
|
||||
"shell": {Name: "shell", ReadOnly: false},
|
||||
},
|
||||
}
|
||||
|
||||
calls := []backend.ToolCall{
|
||||
{ID: "1", Name: "file_edit", Arguments: json.RawMessage(`{"path":"/a"}`)},
|
||||
{ID: "2", Name: "file_edit", Arguments: json.RawMessage(`{"path":"/b"}`)},
|
||||
{ID: "3", Name: "file_read", Arguments: json.RawMessage(`{"path":"/a"}`)},
|
||||
{ID: "4", Name: "file_edit", Arguments: json.RawMessage(`{"path":"/c"}`)},
|
||||
{ID: "5", Name: "shell", Arguments: json.RawMessage(`{"cmd":"make"}`)},
|
||||
{ID: "6", Name: "file_read", Arguments: json.RawMessage(`{"path":"/d"}`)},
|
||||
}
|
||||
|
||||
// Simulate batching: track which batches form
|
||||
var batches [][]string
|
||||
i := 0
|
||||
for i < len(calls) {
|
||||
batch := []backend.ToolCall{calls[i]}
|
||||
batchWrites := toolConflictKeys(rt, calls[i])
|
||||
j := i + 1
|
||||
for j < len(calls) {
|
||||
if hasGlobal(batchWrites) {
|
||||
break
|
||||
}
|
||||
keys := toolConflictKeys(rt, calls[j])
|
||||
if conflictsWithSet(keys, batchWrites) {
|
||||
break
|
||||
}
|
||||
batch = append(batch, calls[j])
|
||||
batchWrites = mergeKeys(batchWrites, keys)
|
||||
j++
|
||||
}
|
||||
var ids []string
|
||||
for _, c := range batch {
|
||||
ids = append(ids, c.ID)
|
||||
}
|
||||
batches = append(batches, ids)
|
||||
i = j
|
||||
}
|
||||
|
||||
// Expected batches:
|
||||
// Batch 1: [1:/a, 2:/b, 3:read/a(no conflict), 4:/c] — all different write paths, reads don't conflict
|
||||
// Batch 2: [5:shell(GLOBAL)] — shell is barrier, seals its own batch
|
||||
// Batch 3: [6:read/d] — after barrier, continues
|
||||
if len(batches) != 3 {
|
||||
t.Fatalf("expected 3 batches, got %d: %v", len(batches), batches)
|
||||
}
|
||||
if len(batches[0]) != 4 {
|
||||
t.Errorf("batch 0: expected 4 calls, got %v", batches[0])
|
||||
}
|
||||
if len(batches[1]) != 1 || batches[1][0] != "5" {
|
||||
t.Errorf("batch 1: expected [5], got %v", batches[1])
|
||||
}
|
||||
if len(batches[2]) != 1 || batches[2][0] != "6" {
|
||||
t.Errorf("batch 2: expected [6], got %v", batches[2])
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecToolCallsSamePathSerializes(t *testing.T) {
|
||||
rt := &Runtime{
|
||||
ToolMeta: map[string]toolsrv.ToolInfo{
|
||||
"file_edit": {Name: "file_edit", ReadOnly: false},
|
||||
},
|
||||
}
|
||||
|
||||
calls := []backend.ToolCall{
|
||||
{ID: "1", Name: "file_edit", Arguments: json.RawMessage(`{"path":"/a"}`)},
|
||||
{ID: "2", Name: "file_edit", Arguments: json.RawMessage(`{"path":"/a"}`)},
|
||||
{ID: "3", Name: "file_edit", Arguments: json.RawMessage(`{"path":"/b"}`)},
|
||||
}
|
||||
|
||||
var batches [][]string
|
||||
i := 0
|
||||
for i < len(calls) {
|
||||
batch := []backend.ToolCall{calls[i]}
|
||||
batchWrites := toolConflictKeys(rt, calls[i])
|
||||
j := i + 1
|
||||
for j < len(calls) {
|
||||
if hasGlobal(batchWrites) {
|
||||
break
|
||||
}
|
||||
keys := toolConflictKeys(rt, calls[j])
|
||||
if conflictsWithSet(keys, batchWrites) {
|
||||
break
|
||||
}
|
||||
batch = append(batch, calls[j])
|
||||
batchWrites = mergeKeys(batchWrites, keys)
|
||||
j++
|
||||
}
|
||||
var ids []string
|
||||
for _, c := range batch {
|
||||
ids = append(ids, c.ID)
|
||||
}
|
||||
batches = append(batches, ids)
|
||||
i = j
|
||||
}
|
||||
|
||||
// Expected: [1:/a] then [2:/a, 3:/b] — second /a conflicts with first batch's /a
|
||||
if len(batches) != 2 {
|
||||
t.Fatalf("expected 2 batches, got %d: %v", len(batches), batches)
|
||||
}
|
||||
if len(batches[0]) != 1 || batches[0][0] != "1" {
|
||||
t.Errorf("batch 0: expected [1], got %v", batches[0])
|
||||
}
|
||||
if len(batches[1]) != 2 {
|
||||
t.Errorf("batch 1: expected [2,3], got %v", batches[1])
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue