lkn
/
ollie-core
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

core: loop-level parallel tool execution for read-safe tools 

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 <noreply@anthropic.com>
This commit is contained in:
Levi Neely 2026-04-28 13:05:15 +02:00
parent efac665ffe
commit bc59f6c40f
4 changed files with 138 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
pkg/agent/core.go
View File

@ -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 != "" {

179
pkg/agent/loop.go
View File

@ -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<system-user-interruption>\n" + injected + "\n</system-user-interruption>"
}
}
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<system-user-interruption>\n" + injected + "\n</system-user-interruption>"
}
}
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<system-user-interruption>\n" + injected + "\n</system-user-interruption>"
}
}
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)

10
pkg/tools/execute/flock.go
View File

@ -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 {

7
pkg/tools/tools.go
View File

@ -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
}