From bc59f6c40f846d039b44b80bc373abe7f9ce20b7 Mon Sep 17 00:00:00 2001 From: Levi Neely Date: Tue, 28 Apr 2026 13:05:15 +0200 Subject: [PATCH] core: loop-level parallel tool execution for read-safe tools MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tools annotated with "# ollie:parallel read" now run concurrently when the LLM issues multiple such calls in a single round, without needing an execute_code pipeline. - tools.ParallelClassifier interface; execute.Server implements it via IsParallelRead, which reads the tool script and checks the annotation - BuildAgentEnv wires IsParallelRead → AgentEnv.classifyTool → agentConfig.ClassifyTool (nil when execute server absent) - loop.go: consecutive parallel-safe calls are fanned out with a WaitGroup; results are collected in submission order; cancellation fills remaining calls with interrupted results; serial tools and batches of one skip the goroutine overhead Co-Authored-By: Claude Sonnet 4.6 --- pkg/agent/core.go | 28 ++++-- pkg/agent/loop.go | 179 +++++++++++++++++++++---------------- pkg/tools/execute/flock.go | 10 +++ pkg/tools/tools.go | 7 ++ 4 files changed, 138 insertions(+), 86 deletions(-) diff --git a/pkg/agent/core.go b/pkg/agent/core.go index 13a652e..e27f47a 100644 --- a/pkg/agent/core.go +++ b/pkg/agent/core.go @@ -24,13 +24,18 @@ import ( "ollie/pkg/tools" ) +// toolClassifier reports whether a named tool is safe to run concurrently +// with other read-class tools. nil means treat all tools as serial. +type toolClassifier func(name string) bool + // AgentEnv holds the runtime state derived from an agent config file. type AgentEnv struct { dispatcher tools.Dispatcher tools []backend.Tool exec toolExecutor + classifyTool toolClassifier Hooks Hooks - preamble string + preamble string genParams backend.GenerationParams Messages []string } @@ -95,12 +100,20 @@ func BuildAgentEnv(cfg *config.Config, d tools.Dispatcher, cwd string) AgentEnv return text, nil } + var classify toolClassifier + if srv, ok := d.GetServer("execute"); ok { + if pc, ok := srv.(tools.ParallelClassifier); ok { + classify = pc.IsParallelRead + } + } + return AgentEnv{ - dispatcher: d, - tools: allTools, - exec: exec, - Hooks: hooks, - preamble: preamble, + dispatcher: d, + tools: allTools, + exec: exec, + classifyTool: classify, + Hooks: hooks, + preamble: preamble, genParams: genParams, Messages: messages, } @@ -261,9 +274,10 @@ func NewAgentCore(cfg AgentCoreConfig) Core { } run := agentConfig{ Backend: cfg.Backend, - preamble: cfg.Env.preamble, + preamble: cfg.Env.preamble, Tools: cfg.Env.tools, Exec: cfg.Env.exec, + ClassifyTool: cfg.Env.classifyTool, GenerationParams: cfg.Env.genParams, } if cfg.SessionID != "" { diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 70fe903..fc1b83f 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "strings" + "sync" "time" "ollie/pkg/backend" @@ -19,13 +20,14 @@ type agentConfig struct { Backend backend.Backend Tools []backend.Tool Exec toolExecutor + ClassifyTool func(name string) bool // nil=treat all as serial; true=parallel-read-safe Output EventHandler preamble string // compiled system+agent prompt sent as the system role GenerationParams backend.GenerationParams - PopInject func() string // returns and clears pending inject, or "" - AutoCompact func(ctx context.Context) // called after each tool round; may compact in-place - SaveSession func() // called after each state.update(); persists mid-turn progress - PreTool func(ctx context.Context, name string, args json.RawMessage) HookResult // called before each tool; exit 2 blocks execution + PopInject func() string // returns and clears pending inject, or "" + AutoCompact func(ctx context.Context) // called after each tool round; may compact in-place + SaveSession func() // called after each state.update(); persists mid-turn progress + PreTool func(ctx context.Context, name string, args json.RawMessage) HookResult // called before each tool; exit 2 blocks execution PostTool func(ctx context.Context, name string, args json.RawMessage, result string) HookResult // called after each tool; exit 0 appends, exit 2 replaces result } @@ -158,57 +160,41 @@ func run(ctx context.Context, cfg agentConfig, state state) error { return fmt.Errorf("step %d: %s", step, stopReason) } - // Execute tool calls, handling mid-execution interruption. + // Execute tool calls, running consecutive parallel-read-safe tools concurrently. msg := backend.Message{Role: "assistant", Content: content.String(), ToolCalls: toolCalls} - var results []toolResult + results := make([]toolResult, 0, len(toolCalls)) interrupted := false - for i, tc := range toolCalls { + cancelledResult := func(tc backend.ToolCall) toolResult { + return toolResult{ + ToolCallID: tc.ID, + Name: tc.Name, + Content: `{"status":"cancelled","error":"interrupted"}`, + IsError: true, + } + } + + // execOne runs a single tool call end-to-end and reports whether the + // context was cancelled during execution. + execOne := func(tc backend.ToolCall) (toolResult, bool) { if tc.Name == "" { - results = append(results, toolResult{ - ToolCallID: tc.ID, - Name: tc.Name, - Content: "error: empty tool name", - IsError: true, - }) - continue + return toolResult{ToolCallID: tc.ID, Name: tc.Name, Content: "error: empty tool name", IsError: true}, false } - - // Check for cancellation before executing. if ctx.Err() != nil { - // Fill synthetic results for this and all remaining tool calls. - for _, remaining := range toolCalls[i:] { - results = append(results, toolResult{ - ToolCallID: remaining.ID, - Name: remaining.Name, - Content: `{"status":"cancelled","error":"interrupted"}`, - IsError: true, - }) - } - interrupted = true - break + return cancelledResult(tc), true } - emit(cfg, Event{Role: "call", Name: tc.Name, Content: string(tc.Arguments)}) - if cfg.PreTool != nil { hr := cfg.PreTool(ctx, tc.Name, tc.Arguments) if hr.Blocked { - msg := hr.Context - if msg == "" { - msg = fmt.Sprintf("tool %q blocked by hook", tc.Name) + blocked := hr.Context + if blocked == "" { + blocked = fmt.Sprintf("tool %q blocked by hook", tc.Name) } - results = append(results, toolResult{ - ToolCallID: tc.ID, - Name: tc.Name, - Content: msg, - IsError: true, - }) - emit(cfg, Event{Role: "tool", Name: tc.Name, Content: msg}) - continue + emit(cfg, Event{Role: "tool", Name: tc.Name, Content: blocked}) + return toolResult{ToolCallID: tc.ID, Name: tc.Name, Content: blocked, IsError: true}, false } } - var result string var isErr bool if cfg.Exec != nil { @@ -216,29 +202,14 @@ func run(ctx context.Context, cfg agentConfig, state state) error { if err != nil { result = fmt.Sprintf("error: %v", err) isErr = true - // If cancelled during execution, fill remaining and break. if ctx.Err() != nil { if cfg.PopInject != nil { if injected := cfg.PopInject(); injected != "" { result += "\n\n\n" + injected + "\n" } } - results = append(results, toolResult{ - ToolCallID: tc.ID, - Name: tc.Name, - Content: result, - IsError: true, - }) - for _, remaining := range toolCalls[i+1:] { - results = append(results, toolResult{ - ToolCallID: remaining.ID, - Name: remaining.Name, - Content: `{"status":"cancelled","error":"interrupted"}`, - IsError: true, - }) - } - interrupted = true - break + emit(cfg, Event{Role: "tool", Name: tc.Name, Content: result}) + return toolResult{ToolCallID: tc.ID, Name: tc.Name, Content: result, IsError: true}, true } } else { result = out @@ -247,29 +218,79 @@ func run(ctx context.Context, cfg agentConfig, state state) error { result = "error: no tool executor configured" isErr = true } - - if !interrupted { - if cfg.PostTool != nil { - hr := cfg.PostTool(ctx, tc.Name, tc.Arguments, result) - if hr.Blocked { - result = hr.Context - } else if hr.Context != "" { - result += "\n" + hr.Context - } + if cfg.PostTool != nil { + hr := cfg.PostTool(ctx, tc.Name, tc.Arguments, result) + if hr.Blocked { + result = hr.Context + } else if hr.Context != "" { + result += "\n" + hr.Context } - if cfg.PopInject != nil { - if injected := cfg.PopInject(); injected != "" { - result += "\n\n\n" + injected + "\n" - } - } - results = append(results, toolResult{ - ToolCallID: tc.ID, - Name: tc.Name, - Content: result, - IsError: isErr, - }) - emit(cfg, Event{Role: "tool", Name: tc.Name, Content: result}) } + if cfg.PopInject != nil { + if injected := cfg.PopInject(); injected != "" { + result += "\n\n\n" + injected + "\n" + } + } + emit(cfg, Event{Role: "tool", Name: tc.Name, Content: result}) + return toolResult{ToolCallID: tc.ID, Name: tc.Name, Content: result, IsError: isErr}, false + } + + isParallelSafe := func(name string) bool { + return name != "" && cfg.ClassifyTool != nil && cfg.ClassifyTool(name) + } + + for i := 0; i < len(toolCalls) && !interrupted; { + if ctx.Err() != nil { + for _, remaining := range toolCalls[i:] { + results = append(results, cancelledResult(remaining)) + } + interrupted = true + break + } + + // Collect a run of consecutive parallel-safe calls. + j := i + 1 + if isParallelSafe(toolCalls[i].Name) { + for j < len(toolCalls) && isParallelSafe(toolCalls[j].Name) { + j++ + } + } + batch := toolCalls[i:j] + + if len(batch) == 1 { + tr, wasInt := execOne(batch[0]) + results = append(results, tr) + if wasInt { + for _, remaining := range toolCalls[j:] { + results = append(results, cancelledResult(remaining)) + } + interrupted = true + } + } else { + // Fan out the batch concurrently; preserve submission order in results. + batchResults := make([]toolResult, len(batch)) + batchInt := make([]bool, len(batch)) + var wg sync.WaitGroup + for k, tc := range batch { + wg.Add(1) + go func(k int, tc backend.ToolCall) { + defer wg.Done() + batchResults[k], batchInt[k] = execOne(tc) + }(k, tc) + } + wg.Wait() + for k, tr := range batchResults { + results = append(results, tr) + if batchInt[k] { + for _, remaining := range toolCalls[j:] { + results = append(results, cancelledResult(remaining)) + } + interrupted = true + break + } + } + } + i = j } state.update(msg, results) diff --git a/pkg/tools/execute/flock.go b/pkg/tools/execute/flock.go index e45d9b5..4391803 100644 --- a/pkg/tools/execute/flock.go +++ b/pkg/tools/execute/flock.go @@ -87,6 +87,16 @@ func acquireFlock(dir, name string, exclusive bool) (*os.File, error) { return f, nil } +// IsParallelRead implements tools.ParallelClassifier. Returns true when the +// named tool script carries a "# ollie:parallel read" annotation. +func (e *Server) IsParallelRead(name string) bool { + code, err := ReadTool(name) + if err != nil { + return false + } + return detectParallelClass(code) == lockClassRead +} + func sanitizeLockName(s string) string { var b strings.Builder for _, r := range s { diff --git a/pkg/tools/tools.go b/pkg/tools/tools.go index e71d38a..5016a2e 100644 --- a/pkg/tools/tools.go +++ b/pkg/tools/tools.go @@ -112,3 +112,10 @@ type TrustedToolsSetter interface { type LockDirSetter interface { SetLockDir(dir string) } + +// ParallelClassifier is implemented by tool servers that can report whether a +// named tool is safe to run concurrently with other read-class tools. +// Returns false for unknown tools (conservative default). +type ParallelClassifier interface { + IsParallelRead(name string) bool +}