mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-19 00:15:05 +00:00
fix: harden parallel tool scheduling
Squash merge PR #103 after validation.
This commit is contained in:
1 parent
4de94c1d7c
commit
484bcc7b71
14 files changed
+1170
-91
No files matched your search
@@ -73,7 +73,8 @@ Finalize (runs once, uses background context if cancelled)
|
||||
|
||||
**ToolStage**
|
||||
- Execute single tool sequentially (no goroutine overhead)
|
||||
- Execute multiple tools in parallel via goroutines, sort results by index
|
||||
- Execute eligible read-only multi-tool batches in bounded parallel, then process results in original order
|
||||
- Keep mutating, async, MCP-bridged, `exec`/`bash`, `wait`, and unknown tools sequential
|
||||
- Emit `tool.call` before, `tool.result` after
|
||||
- Record tool span
|
||||
- Append tool messages to buffer
|
||||
@@ -271,7 +272,9 @@ flowchart TD
|
||||
|
||||
- Append the assistant message (with tool calls) to the message list.
|
||||
- **Single tool call**: execute sequentially (no goroutine overhead).
|
||||
- **Multiple tool calls**: launch parallel goroutines, collect all results, sort by original index, then process sequentially.
|
||||
- **Multiple eligible read-only tool calls**: run raw I/O through a bounded goroutine pool, collect results, then process sequentially in original assistant order.
|
||||
- **Unsafe or mixed batches**: run sequentially when any call is mutating, async, MCP-bridged, `exec`/`bash`, `wait`, unknown, or over the remaining tool-call budget.
|
||||
- Run synchronous `PreToolUse` hooks before any raw parallel I/O; blocked calls append synthetic tool messages and are not executed.
|
||||
- Emit `tool.call` before execution and `tool.result` after.
|
||||
- Record a tool span for each call. Track async tools (spawn, cron) separately.
|
||||
- Save tool messages to the session.
|
||||
|
||||
@@ -64,6 +64,8 @@ Each tool carries structured metadata describing side-effect class, group member
|
||||
|
||||
Capabilities are inferred from tool name when no explicit metadata is registered. The policy engine can use capability class to gate entire sets (e.g. restrict an agent to read-only tools).
|
||||
|
||||
The agent loop also uses this metadata for parallel tool-call scheduling. Only registered read-only tools are eligible for bounded parallel raw I/O. Mutating, async, MCP-bridged, `exec`/`bash`, `wait`, and unknown tools stay sequential by default. `PreToolUse` hooks run before any parallel I/O so hooks can block or rewrite arguments consistently.
|
||||
|
||||
---
|
||||
|
||||
## 4. Built-in Tool Inventory
|
||||
|
||||
@@ -34,6 +34,28 @@ func (l *Loop) resolveToolCallName(name string) string {
|
||||
return name
|
||||
}
|
||||
|
||||
func (l *Loop) parallelEligibleToolCall(tc providers.ToolCall) bool {
|
||||
name := l.resolveToolCallName(tc.Name)
|
||||
switch {
|
||||
case name == "exec", name == "bash", name == "wait":
|
||||
return false
|
||||
case strings.HasPrefix(name, "mcp_"):
|
||||
return false
|
||||
case l.registry == nil:
|
||||
return false
|
||||
}
|
||||
tool, ok := l.registry.Get(name)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
|
||||
meta := l.registry.GetMetadata(tool.Name())
|
||||
return meta.IsReadOnly() &&
|
||||
!meta.HasCapability(tools.CapMutating) &&
|
||||
!meta.HasCapability(tools.CapAsync) &&
|
||||
!meta.HasCapability(tools.CapMCPBridged)
|
||||
}
|
||||
|
||||
// normalizeToolCall rewrites malformed MCP pseudo-calls that some models emit
|
||||
// as `exec` with `{action:"mcp_xxx", code|command:"..."}`.
|
||||
// We recover the intended MCP tool name from `action` and map payload to MCP
|
||||
|
||||
@@ -144,7 +144,8 @@ func (l *Loop) buildPipelineDeps(req *RunRequest, bridgeRS *runState) pipeline.P
|
||||
SequentialToolCall: func(tc providers.ToolCall) bool {
|
||||
return l.resolveToolCallName(tc.Name) == "wait"
|
||||
},
|
||||
CheckReadOnly: cb.checkReadOnly,
|
||||
ParallelEligibleToolCall: l.parallelEligibleToolCall,
|
||||
CheckReadOnly: cb.checkReadOnly,
|
||||
|
||||
// Observe: drain InjectCh
|
||||
DrainInjectCh: func() []providers.Message {
|
||||
@@ -207,38 +208,38 @@ func (l *Loop) buildPipelineDeps(req *RunRequest, bridgeRS *runState) pipeline.P
|
||||
// convertRunInput converts agent.RunRequest to pipeline.RunInput.
|
||||
func convertRunInput(req *RunRequest) *pipeline.RunInput {
|
||||
return &pipeline.RunInput{
|
||||
SessionKey: req.SessionKey,
|
||||
Message: req.Message,
|
||||
Media: req.Media,
|
||||
ForwardMedia: req.ForwardMedia,
|
||||
SessionKey: req.SessionKey,
|
||||
Message: req.Message,
|
||||
Media: req.Media,
|
||||
ForwardMedia: req.ForwardMedia,
|
||||
Channel: req.Channel,
|
||||
ChannelType: req.ChannelType,
|
||||
BitrixPortalDomain: req.BitrixPortalDomain,
|
||||
ChatTitle: req.ChatTitle,
|
||||
ChatID: req.ChatID,
|
||||
PeerKind: req.PeerKind,
|
||||
RunID: req.RunID,
|
||||
UserID: req.UserID,
|
||||
SenderID: req.SenderID,
|
||||
Stream: req.Stream,
|
||||
ExtraSystemPrompt: req.ExtraSystemPrompt,
|
||||
SkillFilter: req.SkillFilter,
|
||||
HistoryLimit: req.HistoryLimit,
|
||||
ToolAllow: req.ToolAllow,
|
||||
LightContext: req.LightContext,
|
||||
RunKind: req.RunKind,
|
||||
DelegationID: req.DelegationID,
|
||||
TeamID: req.TeamID,
|
||||
TeamTaskID: req.TeamTaskID,
|
||||
ParentAgentID: req.ParentAgentID,
|
||||
MaxIterations: req.MaxIterations,
|
||||
ModelOverride: req.ModelOverride,
|
||||
HideInput: req.HideInput,
|
||||
ContentSuffix: req.ContentSuffix,
|
||||
LeaderAgentID: req.LeaderAgentID,
|
||||
WorkspaceChannel: req.WorkspaceChannel,
|
||||
WorkspaceChatID: req.WorkspaceChatID,
|
||||
TeamWorkspace: req.TeamWorkspace,
|
||||
ChatID: req.ChatID,
|
||||
PeerKind: req.PeerKind,
|
||||
RunID: req.RunID,
|
||||
UserID: req.UserID,
|
||||
SenderID: req.SenderID,
|
||||
Stream: req.Stream,
|
||||
ExtraSystemPrompt: req.ExtraSystemPrompt,
|
||||
SkillFilter: req.SkillFilter,
|
||||
HistoryLimit: req.HistoryLimit,
|
||||
ToolAllow: req.ToolAllow,
|
||||
LightContext: req.LightContext,
|
||||
RunKind: req.RunKind,
|
||||
DelegationID: req.DelegationID,
|
||||
TeamID: req.TeamID,
|
||||
TeamTaskID: req.TeamTaskID,
|
||||
ParentAgentID: req.ParentAgentID,
|
||||
MaxIterations: req.MaxIterations,
|
||||
ModelOverride: req.ModelOverride,
|
||||
HideInput: req.HideInput,
|
||||
ContentSuffix: req.ContentSuffix,
|
||||
LeaderAgentID: req.LeaderAgentID,
|
||||
WorkspaceChannel: req.WorkspaceChannel,
|
||||
WorkspaceChatID: req.WorkspaceChatID,
|
||||
TeamWorkspace: req.TeamWorkspace,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/pipeline"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/providers"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
@@ -18,11 +19,22 @@ type stubExecutor struct{}
|
||||
func (s *stubExecutor) ExecuteWithContext(_ context.Context, _ string, _ map[string]any, _, _, _, _ string, _ tools.AsyncCallback) *tools.Result {
|
||||
return &tools.Result{ForLLM: "ok", IsError: false}
|
||||
}
|
||||
func (s *stubExecutor) TryActivateDeferred(string) bool { return false }
|
||||
func (s *stubExecutor) TryActivateDeferred(string) bool { return false }
|
||||
func (s *stubExecutor) ProviderDefs() []providers.ToolDefinition { return nil }
|
||||
func (s *stubExecutor) Get(string) (tools.Tool, bool) { return nil, false }
|
||||
func (s *stubExecutor) List() []string { return nil }
|
||||
func (s *stubExecutor) Aliases() map[string]string { return nil }
|
||||
func (s *stubExecutor) Get(string) (tools.Tool, bool) { return nil, false }
|
||||
func (s *stubExecutor) List() []string { return nil }
|
||||
func (s *stubExecutor) Aliases() map[string]string { return nil }
|
||||
|
||||
type metadataTestTool struct {
|
||||
name string
|
||||
}
|
||||
|
||||
func (t metadataTestTool) Name() string { return t.name }
|
||||
func (t metadataTestTool) Description() string { return "test tool" }
|
||||
func (t metadataTestTool) Parameters() map[string]any { return nil }
|
||||
func (t metadataTestTool) Execute(context.Context, map[string]any) *tools.Result {
|
||||
return &tools.Result{ForLLM: "ok"}
|
||||
}
|
||||
|
||||
// eventCollector buffers AgentEvents for inspection in tests.
|
||||
// Safe for concurrent appends from parallel goroutines.
|
||||
@@ -158,6 +170,59 @@ func TestMakeExecuteToolRaw_ConcurrentCallsEmitAllEvents(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestParallelEligibleToolCall_OnlyAllowsRegisteredReadOnlyTools(t *testing.T) {
|
||||
t.Parallel()
|
||||
registry := tools.NewRegistry()
|
||||
registry.RegisterWithMetadata(metadataTestTool{name: "read_file"}, tools.ToolMetadata{Capabilities: []tools.ToolCapability{tools.CapReadOnly}})
|
||||
registry.RegisterWithMetadata(metadataTestTool{name: "write_file"}, tools.ToolMetadata{Capabilities: []tools.ToolCapability{tools.CapMutating}})
|
||||
registry.RegisterWithMetadata(metadataTestTool{name: "spawn"}, tools.ToolMetadata{Capabilities: []tools.ToolCapability{tools.CapAsync}})
|
||||
registry.RegisterWithMetadata(metadataTestTool{name: "mcp_search"}, tools.ToolMetadata{Capabilities: []tools.ToolCapability{tools.CapReadOnly, tools.CapMCPBridged}})
|
||||
registry.RegisterWithMetadata(metadataTestTool{name: "web_fetch"}, tools.ToolMetadata{Capabilities: []tools.ToolCapability{tools.CapReadOnly}})
|
||||
registry.RegisterAlias("read_alias", "read_file")
|
||||
|
||||
l := &Loop{registry: registry}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
tc providers.ToolCall
|
||||
want bool
|
||||
}{
|
||||
{name: "registered read only", tc: providers.ToolCall{Name: "read_file"}, want: true},
|
||||
{name: "alias read only", tc: providers.ToolCall{Name: "read_alias"}, want: true},
|
||||
{name: "mutating", tc: providers.ToolCall{Name: "write_file"}, want: false},
|
||||
{name: "async", tc: providers.ToolCall{Name: "spawn"}, want: false},
|
||||
{name: "mcp prefix", tc: providers.ToolCall{Name: "mcp_search"}, want: false},
|
||||
{name: "exec excluded", tc: providers.ToolCall{Name: "exec"}, want: false},
|
||||
{name: "wait excluded", tc: providers.ToolCall{Name: "wait"}, want: false},
|
||||
{name: "unknown inferred read only still blocked", tc: providers.ToolCall{Name: "web_search"}, want: false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
if got := l.parallelEligibleToolCall(tt.tc); got != tt.want {
|
||||
t.Fatalf("parallelEligibleToolCall(%q) = %v, want %v", tt.tc.Name, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestParallelEligibleToolCall_StripsAgentToolPrefixBeforeMetadataLookup(t *testing.T) {
|
||||
t.Parallel()
|
||||
registry := tools.NewRegistry()
|
||||
registry.RegisterWithMetadata(metadataTestTool{name: "web_fetch"}, tools.ToolMetadata{Capabilities: []tools.ToolCapability{tools.CapReadOnly}})
|
||||
|
||||
l := &Loop{
|
||||
registry: registry,
|
||||
agentToolPolicy: &config.ToolPolicySpec{
|
||||
ToolCallPrefix: "agent_",
|
||||
},
|
||||
}
|
||||
|
||||
if !l.parallelEligibleToolCall(providers.ToolCall{Name: "agent_web_fetch"}) {
|
||||
t.Fatal("expected prefixed registered read-only tool to be parallel eligible")
|
||||
}
|
||||
}
|
||||
|
||||
// assertToolCallPayload verifies the event carries the expected tc identity
|
||||
// and routing context from RunRequest.
|
||||
func assertToolCallPayload(t *testing.T, ev AgentEvent, tc providers.ToolCall, req *RunRequest) {
|
||||
|
||||
+16
-13
@@ -61,10 +61,10 @@ type PipelineDeps struct {
|
||||
InjectReminders func(ctx context.Context, input *RunInput, msgs []providers.Message) []providers.Message
|
||||
|
||||
// Think callbacks (ThinkStage)
|
||||
BuildFilteredTools func(state *RunState) ([]providers.ToolDefinition, error)
|
||||
CallLLM func(ctx context.Context, state *RunState, req providers.ChatRequest) (*providers.ChatResponse, error)
|
||||
UniqueToolCallIDs func(calls []providers.ToolCall, runID string, iteration int) []providers.ToolCall
|
||||
EmitBlockReply func(content string) // emit block.reply for intermediate assistant content
|
||||
BuildFilteredTools func(state *RunState) ([]providers.ToolDefinition, error)
|
||||
CallLLM func(ctx context.Context, state *RunState, req providers.ChatRequest) (*providers.ChatResponse, error)
|
||||
UniqueToolCallIDs func(calls []providers.ToolCall, runID string, iteration int) []providers.ToolCall
|
||||
EmitBlockReply func(content string) // emit block.reply for intermediate assistant content
|
||||
|
||||
// Prune callbacks (PruneStage)
|
||||
PruneMessages func(msgs []providers.Message, budget int) ([]providers.Message, PruneStats)
|
||||
@@ -72,10 +72,10 @@ type PipelineDeps struct {
|
||||
CompactMessages func(ctx context.Context, msgs []providers.Message, model string) ([]providers.Message, error)
|
||||
|
||||
// Cache-TTL gate callbacks (Phase 06). All optional (nil = feature disabled).
|
||||
GetProviderCaps func() providers.ProviderCapabilities // provider capabilities for cache detection
|
||||
GetPruningConfig func() *config.ContextPruningConfig // pruning config (TTL field)
|
||||
GetCacheTouch func(sessionKey string) time.Time // per-session last prune-mutation timestamp
|
||||
MarkCacheTouched func(sessionKey string) // record mutation timestamp AFTER prune mutates
|
||||
GetProviderCaps func() providers.ProviderCapabilities // provider capabilities for cache detection
|
||||
GetPruningConfig func() *config.ContextPruningConfig // pruning config (TTL field)
|
||||
GetCacheTouch func(sessionKey string) time.Time // per-session last prune-mutation timestamp
|
||||
MarkCacheTouched func(sessionKey string) // record mutation timestamp AFTER prune mutates
|
||||
|
||||
// Memory flush callbacks (MemoryFlushStage, invoked by PruneStage)
|
||||
RunMemoryFlush func(ctx context.Context, state *RunState) error
|
||||
@@ -93,6 +93,9 @@ type PipelineDeps struct {
|
||||
// When any tool call in a batch matches, ToolStage uses ExecuteToolCall for the
|
||||
// whole batch instead of parallel raw execution.
|
||||
SequentialToolCall func(tc providers.ToolCall) bool
|
||||
// ParallelEligibleToolCall returns true when a tool call is safe to execute
|
||||
// through the parallel raw I/O path. Nil = no parallel raw execution.
|
||||
ParallelEligibleToolCall func(tc providers.ToolCall) bool
|
||||
// CheckReadOnly checks read-only streak. Returns warning message (if any) and whether to break.
|
||||
CheckReadOnly func(state *RunState) (*providers.Message, bool)
|
||||
|
||||
@@ -107,11 +110,11 @@ type PipelineDeps struct {
|
||||
// response to workspace disk, appends MediaRefs, and clears inline base64.
|
||||
// Called BEFORE building the assistant message for session persistence.
|
||||
// nil = feature disabled (no Codex image gen or no workspace).
|
||||
PersistAssistantImages func(msg *providers.Message, workspace string)
|
||||
SkillPostscript func(ctx context.Context, content string, totalToolCalls int) string // skill evolution nudge (nil = disabled)
|
||||
SanitizeContent func(content string) string
|
||||
StripMessageDirectives func(content string) string
|
||||
DeduplicateMediaSuffix func(content, suffix string) string
|
||||
PersistAssistantImages func(msg *providers.Message, workspace string)
|
||||
SkillPostscript func(ctx context.Context, content string, totalToolCalls int) string // skill evolution nudge (nil = disabled)
|
||||
SanitizeContent func(content string) string
|
||||
StripMessageDirectives func(content string) string
|
||||
DeduplicateMediaSuffix func(content, suffix string) string
|
||||
IsSilentReply func(content string) bool
|
||||
EmitSessionCompleted func(ctx context.Context, sessionKey string, msgCount, tokensUsed, compactionCount int)
|
||||
UpdateMetadata func(ctx context.Context, sessionKey string, usage providers.Usage) error
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -14,6 +15,7 @@ import (
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bootstrap"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/hooks"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/providers"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/workspace"
|
||||
)
|
||||
@@ -37,6 +39,17 @@ func defaultState() *RunState {
|
||||
return stateWithInput(minimalInput())
|
||||
}
|
||||
|
||||
type toolStageHookDispatcher struct {
|
||||
fire func(context.Context, hooks.Event) (hooks.FireResult, error)
|
||||
}
|
||||
|
||||
func (d toolStageHookDispatcher) Fire(ctx context.Context, ev hooks.Event) (hooks.FireResult, error) {
|
||||
if d.fire == nil {
|
||||
return hooks.FireResult{Decision: hooks.DecisionAllow}, nil
|
||||
}
|
||||
return d.fire(ctx, ev)
|
||||
}
|
||||
|
||||
// mockTokenCounter returns a fixed count for every call.
|
||||
type mockTokenCounter struct {
|
||||
countPerMessage int
|
||||
@@ -47,7 +60,7 @@ func (m *mockTokenCounter) CountMessages(_ string, msgs []providers.Message) int
|
||||
return len(msgs) * m.countPerMessage
|
||||
}
|
||||
func (m *mockTokenCounter) CountToolSchemas(_ string, _ []providers.ToolDefinition) int { return 0 }
|
||||
func (m *mockTokenCounter) ModelContextWindow(_ string) int { return 200_000 }
|
||||
func (m *mockTokenCounter) ModelContextWindow(_ string) int { return 200_000 }
|
||||
|
||||
// --- ThinkStage tests ---
|
||||
|
||||
@@ -935,6 +948,7 @@ func TestToolStage_MultipleTools_ParallelPath_InvokesRawAndProcessForEach(t *tes
|
||||
procMu.Unlock()
|
||||
return []providers.Message{rawMsg}
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
@@ -967,6 +981,425 @@ func TestToolStage_MultipleTools_ParallelPath_InvokesRawAndProcessForEach(t *tes
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_ParallelPath_PreservesOriginalOrderWhenRawCompletesOutOfOrder(t *testing.T) {
|
||||
t.Parallel()
|
||||
release := map[string]chan struct{}{
|
||||
"1": make(chan struct{}),
|
||||
"2": make(chan struct{}),
|
||||
"3": make(chan struct{}),
|
||||
}
|
||||
started := make(chan string, 3)
|
||||
done := make(chan error, 1)
|
||||
|
||||
deps := &PipelineDeps{
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, _ providers.ToolCall) ([]providers.Message, error) {
|
||||
t.Fatal("ExecuteToolCall must not run for eligible parallel batch")
|
||||
return nil, nil
|
||||
},
|
||||
ExecuteToolRaw: func(_ context.Context, tc providers.ToolCall) (providers.Message, any, error) {
|
||||
started <- tc.ID
|
||||
<-release[tc.ID]
|
||||
return providers.Message{Role: "tool", Content: "raw:" + tc.ID, ToolCallID: tc.ID}, nil, nil
|
||||
},
|
||||
ProcessToolResult: func(_ context.Context, _ *RunState, _ providers.ToolCall, rawMsg providers.Message, _ any) []providers.Message {
|
||||
return []providers.Message{rawMsg}
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: []providers.ToolCall{
|
||||
{ID: "1", Name: "read_file"},
|
||||
{ID: "2", Name: "read_file"},
|
||||
{ID: "3", Name: "read_file"},
|
||||
}}
|
||||
|
||||
go func() {
|
||||
done <- stage.Execute(context.Background(), state)
|
||||
}()
|
||||
|
||||
for range 3 {
|
||||
select {
|
||||
case <-started:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for raw calls to start")
|
||||
}
|
||||
}
|
||||
close(release["3"])
|
||||
close(release["2"])
|
||||
close(release["1"])
|
||||
|
||||
if err := <-done; err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
got := []string{}
|
||||
for _, msg := range state.Messages.Pending() {
|
||||
got = append(got, msg.ToolCallID)
|
||||
}
|
||||
if want := []string{"1", "2", "3"}; !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("pending order = %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_ParallelPath_RunsPreToolUseBeforeRawAndAppliesRewrite(t *testing.T) {
|
||||
t.Parallel()
|
||||
rawArgsByID := map[string]map[string]any{}
|
||||
var rawMu sync.Mutex
|
||||
deps := &PipelineDeps{
|
||||
Hooks: toolStageHookDispatcher{fire: func(_ context.Context, ev hooks.Event) (hooks.FireResult, error) {
|
||||
if ev.HookEvent == hooks.EventPreToolUse && ev.ToolName == "read_file" {
|
||||
return hooks.FireResult{
|
||||
Decision: hooks.DecisionAllow,
|
||||
UpdatedToolInput: map[string]any{"path": "/safe"},
|
||||
}, nil
|
||||
}
|
||||
return hooks.FireResult{Decision: hooks.DecisionAllow}, nil
|
||||
}},
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, _ providers.ToolCall) ([]providers.Message, error) {
|
||||
t.Fatal("ExecuteToolCall must not run for eligible parallel batch")
|
||||
return nil, nil
|
||||
},
|
||||
ExecuteToolRaw: func(_ context.Context, tc providers.ToolCall) (providers.Message, any, error) {
|
||||
rawMu.Lock()
|
||||
rawArgsByID[tc.ID] = tc.Arguments
|
||||
rawMu.Unlock()
|
||||
return providers.Message{Role: "tool", Content: "ok", ToolCallID: tc.ID}, nil, nil
|
||||
},
|
||||
ProcessToolResult: func(_ context.Context, _ *RunState, _ providers.ToolCall, rawMsg providers.Message, _ any) []providers.Message {
|
||||
return []providers.Message{rawMsg}
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: []providers.ToolCall{
|
||||
{ID: "1", Name: "read_file", Arguments: map[string]any{"path": "/unsafe"}},
|
||||
{ID: "2", Name: "web_fetch", Arguments: map[string]any{"url": "https://example.com"}},
|
||||
}}
|
||||
|
||||
if err := stage.Execute(context.Background(), state); err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
rawMu.Lock()
|
||||
rawArgs := rawArgsByID["1"]
|
||||
rawMu.Unlock()
|
||||
if got, _ := rawArgs["path"].(string); got != "/safe" {
|
||||
t.Fatalf("raw path = %q, want hook rewrite /safe", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_ParallelPath_PreservesOriginalOrderWithDuplicateToolCallIDs(t *testing.T) {
|
||||
t.Parallel()
|
||||
deps := &PipelineDeps{
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, _ providers.ToolCall) ([]providers.Message, error) {
|
||||
t.Fatal("ExecuteToolCall must not run for eligible parallel batch")
|
||||
return nil, nil
|
||||
},
|
||||
ExecuteToolRaw: func(_ context.Context, tc providers.ToolCall) (providers.Message, any, error) {
|
||||
slot, _ := tc.Arguments["slot"].(string)
|
||||
return providers.Message{Role: "tool", Content: "raw:" + slot, ToolCallID: tc.ID}, nil, nil
|
||||
},
|
||||
ProcessToolResult: func(_ context.Context, _ *RunState, _ providers.ToolCall, rawMsg providers.Message, _ any) []providers.Message {
|
||||
return []providers.Message{rawMsg}
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: []providers.ToolCall{
|
||||
{ID: "dup", Name: "read_file", Arguments: map[string]any{"slot": "1"}},
|
||||
{ID: "dup", Name: "read_file", Arguments: map[string]any{"slot": "2"}},
|
||||
{ID: "dup", Name: "read_file", Arguments: map[string]any{"slot": "3"}},
|
||||
}}
|
||||
|
||||
if err := stage.Execute(context.Background(), state); err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
got := []string{}
|
||||
for _, msg := range state.Messages.Pending() {
|
||||
got = append(got, msg.Content)
|
||||
}
|
||||
if want := []string{"raw:1", "raw:2", "raw:3"}; !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("pending contents = %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_ParallelPath_BlockedPreToolUseSkipsRawExecution(t *testing.T) {
|
||||
t.Parallel()
|
||||
rawCalls := []string{}
|
||||
deps := &PipelineDeps{
|
||||
Hooks: toolStageHookDispatcher{fire: func(_ context.Context, ev hooks.Event) (hooks.FireResult, error) {
|
||||
if ev.HookEvent == hooks.EventPreToolUse && ev.ToolName == "read_file" {
|
||||
return hooks.FireResult{Decision: hooks.DecisionBlock}, nil
|
||||
}
|
||||
return hooks.FireResult{Decision: hooks.DecisionAllow}, nil
|
||||
}},
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, _ providers.ToolCall) ([]providers.Message, error) {
|
||||
t.Fatal("ExecuteToolCall must not run when remaining calls are parallel eligible")
|
||||
return nil, nil
|
||||
},
|
||||
ExecuteToolRaw: func(_ context.Context, tc providers.ToolCall) (providers.Message, any, error) {
|
||||
rawCalls = append(rawCalls, tc.ID)
|
||||
return providers.Message{Role: "tool", Content: "raw:" + tc.Name, ToolCallID: tc.ID}, nil, nil
|
||||
},
|
||||
ProcessToolResult: func(_ context.Context, _ *RunState, _ providers.ToolCall, rawMsg providers.Message, _ any) []providers.Message {
|
||||
return []providers.Message{rawMsg}
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: []providers.ToolCall{
|
||||
{ID: "1", Name: "read_file"},
|
||||
{ID: "2", Name: "web_fetch"},
|
||||
}}
|
||||
|
||||
if err := stage.Execute(context.Background(), state); err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(rawCalls, []string{"2"}) {
|
||||
t.Fatalf("raw calls = %v, want only unblocked call 2", rawCalls)
|
||||
}
|
||||
pending := state.Messages.Pending()
|
||||
if len(pending) != 2 {
|
||||
t.Fatalf("pending len = %d, want 2", len(pending))
|
||||
}
|
||||
if pending[0].ToolCallID != "1" || !strings.Contains(pending[0].Content, "Hook blocked") {
|
||||
t.Fatalf("pending[0] = %+v, want blocked message for call 1", pending[0])
|
||||
}
|
||||
if pending[1].ToolCallID != "2" {
|
||||
t.Fatalf("pending[1] = %+v, want raw result for call 2", pending[1])
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_MultipleTools_IneligibleBatchUsesSequentialExecution(t *testing.T) {
|
||||
t.Parallel()
|
||||
calls := []string{}
|
||||
deps := &PipelineDeps{
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, tc providers.ToolCall) ([]providers.Message, error) {
|
||||
calls = append(calls, tc.Name)
|
||||
return []providers.Message{{Role: "tool", Content: "result:" + tc.Name, ToolCallID: tc.ID}}, nil
|
||||
},
|
||||
ExecuteToolRaw: func(_ context.Context, _ providers.ToolCall) (providers.Message, any, error) {
|
||||
t.Fatal("ExecuteToolRaw must not run when any call is ineligible")
|
||||
return providers.Message{}, nil, nil
|
||||
},
|
||||
ProcessToolResult: func(_ context.Context, _ *RunState, _ providers.ToolCall, _ providers.Message, _ any) []providers.Message {
|
||||
t.Fatal("ProcessToolResult must not run when any call is ineligible")
|
||||
return nil
|
||||
},
|
||||
ParallelEligibleToolCall: func(tc providers.ToolCall) bool {
|
||||
return tc.Name == "read_file" || tc.Name == "web_fetch"
|
||||
},
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: []providers.ToolCall{
|
||||
{ID: "1", Name: "read_file"},
|
||||
{ID: "2", Name: "write_file"},
|
||||
{ID: "3", Name: "web_fetch"},
|
||||
}}
|
||||
|
||||
if err := stage.Execute(context.Background(), state); err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
if want := []string{"read_file", "write_file", "web_fetch"}; !reflect.DeepEqual(calls, want) {
|
||||
t.Fatalf("ExecuteToolCall order = %v, want %v", calls, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_ParallelBatchFallsBackToSequentialWhenBudgetCannotFitBatch(t *testing.T) {
|
||||
t.Parallel()
|
||||
calls := []string{}
|
||||
deps := &PipelineDeps{
|
||||
Config: PipelineConfig{MaxToolCalls: 2},
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, tc providers.ToolCall) ([]providers.Message, error) {
|
||||
calls = append(calls, tc.ID)
|
||||
return []providers.Message{{Role: "tool", Content: "result:" + tc.Name, ToolCallID: tc.ID}}, nil
|
||||
},
|
||||
ExecuteToolRaw: func(_ context.Context, _ providers.ToolCall) (providers.Message, any, error) {
|
||||
t.Fatal("ExecuteToolRaw must not run when full batch exceeds budget")
|
||||
return providers.Message{}, nil, nil
|
||||
},
|
||||
ProcessToolResult: func(_ context.Context, _ *RunState, _ providers.ToolCall, _ providers.Message, _ any) []providers.Message {
|
||||
t.Fatal("ProcessToolResult must not run when full batch exceeds budget")
|
||||
return nil
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
state.Tool.TotalToolCalls = 1
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: []providers.ToolCall{
|
||||
{ID: "1", Name: "read_file"},
|
||||
{ID: "2", Name: "web_fetch"},
|
||||
}}
|
||||
|
||||
if err := stage.Execute(context.Background(), state); err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(calls, []string{"1"}) {
|
||||
t.Fatalf("ExecuteToolCall calls = %v, want [1]", calls)
|
||||
}
|
||||
if stage.Result() != BreakLoop {
|
||||
t.Fatalf("Result() = %v, want BreakLoop", stage.Result())
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_SequentialFallback_DoesNotPreflightCallsSkippedByBudget(t *testing.T) {
|
||||
t.Parallel()
|
||||
hookCalls := []string{}
|
||||
deps := &PipelineDeps{
|
||||
Config: PipelineConfig{MaxToolCalls: 1},
|
||||
Hooks: toolStageHookDispatcher{fire: func(_ context.Context, ev hooks.Event) (hooks.FireResult, error) {
|
||||
if ev.HookEvent == hooks.EventPreToolUse {
|
||||
hookCalls = append(hookCalls, ev.ToolName)
|
||||
}
|
||||
return hooks.FireResult{Decision: hooks.DecisionAllow}, nil
|
||||
}},
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, tc providers.ToolCall) ([]providers.Message, error) {
|
||||
return []providers.Message{{Role: "tool", Content: "result:" + tc.Name, ToolCallID: tc.ID}}, nil
|
||||
},
|
||||
ExecuteToolRaw: func(_ context.Context, _ providers.ToolCall) (providers.Message, any, error) {
|
||||
t.Fatal("ExecuteToolRaw must not run when full batch exceeds budget")
|
||||
return providers.Message{}, nil, nil
|
||||
},
|
||||
ProcessToolResult: func(_ context.Context, _ *RunState, _ providers.ToolCall, _ providers.Message, _ any) []providers.Message {
|
||||
t.Fatal("ProcessToolResult must not run when full batch exceeds budget")
|
||||
return nil
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: []providers.ToolCall{
|
||||
{ID: "1", Name: "read_file"},
|
||||
{ID: "2", Name: "web_fetch"},
|
||||
}}
|
||||
|
||||
if err := stage.Execute(context.Background(), state); err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(hookCalls, []string{"read_file"}) {
|
||||
t.Fatalf("pre_tool_use calls = %v, want only first executable call", hookCalls)
|
||||
}
|
||||
if state.Tool.TotalToolCalls != 1 {
|
||||
t.Fatalf("TotalToolCalls = %d, want 1", state.Tool.TotalToolCalls)
|
||||
}
|
||||
if stage.Result() != BreakLoop {
|
||||
t.Fatalf("Result() = %v, want BreakLoop", stage.Result())
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_SequentialFallback_DoesNotOvershootBudgetWithBlockedHook(t *testing.T) {
|
||||
t.Parallel()
|
||||
hookCalls := []string{}
|
||||
deps := &PipelineDeps{
|
||||
Config: PipelineConfig{MaxToolCalls: 1},
|
||||
Hooks: toolStageHookDispatcher{fire: func(_ context.Context, ev hooks.Event) (hooks.FireResult, error) {
|
||||
if ev.HookEvent == hooks.EventPreToolUse {
|
||||
hookCalls = append(hookCalls, ev.ToolName)
|
||||
return hooks.FireResult{Decision: hooks.DecisionBlock}, nil
|
||||
}
|
||||
return hooks.FireResult{Decision: hooks.DecisionAllow}, nil
|
||||
}},
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, _ providers.ToolCall) ([]providers.Message, error) {
|
||||
t.Fatal("ExecuteToolCall must not run when hook blocks the only budgeted call")
|
||||
return nil, nil
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: []providers.ToolCall{
|
||||
{ID: "1", Name: "read_file"},
|
||||
{ID: "2", Name: "web_fetch"},
|
||||
}}
|
||||
|
||||
if err := stage.Execute(context.Background(), state); err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(hookCalls, []string{"read_file"}) {
|
||||
t.Fatalf("pre_tool_use calls = %v, want only first blocked call", hookCalls)
|
||||
}
|
||||
if state.Tool.TotalToolCalls != 1 {
|
||||
t.Fatalf("TotalToolCalls = %d, want 1", state.Tool.TotalToolCalls)
|
||||
}
|
||||
pending := state.Messages.Pending()
|
||||
if len(pending) != 1 || pending[0].ToolCallID != "1" || !strings.Contains(pending[0].Content, "Hook blocked") {
|
||||
t.Fatalf("pending = %+v, want one blocked message for first call", pending)
|
||||
}
|
||||
if stage.Result() != BreakLoop {
|
||||
t.Fatalf("Result() = %v, want BreakLoop", stage.Result())
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_ParallelPath_BoundsRawConcurrency(t *testing.T) {
|
||||
t.Parallel()
|
||||
const totalCalls = 8
|
||||
started := make(chan struct{}, totalCalls)
|
||||
release := make(chan struct{})
|
||||
done := make(chan error, 1)
|
||||
var current int32
|
||||
var peak int32
|
||||
|
||||
deps := &PipelineDeps{
|
||||
ExecuteToolCall: func(_ context.Context, _ *RunState, _ providers.ToolCall) ([]providers.Message, error) {
|
||||
t.Fatal("ExecuteToolCall must not run for eligible parallel batch")
|
||||
return nil, nil
|
||||
},
|
||||
ExecuteToolRaw: func(_ context.Context, tc providers.ToolCall) (providers.Message, any, error) {
|
||||
now := atomic.AddInt32(¤t, 1)
|
||||
for {
|
||||
prev := atomic.LoadInt32(&peak)
|
||||
if now <= prev || atomic.CompareAndSwapInt32(&peak, prev, now) {
|
||||
break
|
||||
}
|
||||
}
|
||||
started <- struct{}{}
|
||||
<-release
|
||||
atomic.AddInt32(¤t, -1)
|
||||
return providers.Message{Role: "tool", Content: "raw:" + tc.Name, ToolCallID: tc.ID}, nil, nil
|
||||
},
|
||||
ProcessToolResult: func(_ context.Context, _ *RunState, _ providers.ToolCall, rawMsg providers.Message, _ any) []providers.Message {
|
||||
return []providers.Message{rawMsg}
|
||||
},
|
||||
ParallelEligibleToolCall: func(providers.ToolCall) bool { return true },
|
||||
}
|
||||
stage := NewToolStage(deps)
|
||||
state := defaultState()
|
||||
calls := make([]providers.ToolCall, 0, totalCalls)
|
||||
for i := range totalCalls {
|
||||
id := strconv.Itoa(i + 1)
|
||||
calls = append(calls, providers.ToolCall{ID: id, Name: "read_file"})
|
||||
}
|
||||
state.Think.LastResponse = &providers.ChatResponse{ToolCalls: calls}
|
||||
|
||||
go func() {
|
||||
done <- stage.Execute(context.Background(), state)
|
||||
}()
|
||||
|
||||
for range defaultParallelToolCallLimit {
|
||||
select {
|
||||
case <-started:
|
||||
case <-time.After(time.Second):
|
||||
close(release)
|
||||
t.Fatal("timed out waiting for capped raw calls to start")
|
||||
}
|
||||
}
|
||||
if got := atomic.LoadInt32(&peak); got > int32(defaultParallelToolCallLimit) {
|
||||
close(release)
|
||||
t.Fatalf("peak raw concurrency = %d, want <= %d", got, defaultParallelToolCallLimit)
|
||||
}
|
||||
close(release)
|
||||
if err := <-done; err != nil {
|
||||
t.Fatalf("Execute() error: %v", err)
|
||||
}
|
||||
if got := atomic.LoadInt32(&peak); got > int32(defaultParallelToolCallLimit) {
|
||||
t.Fatalf("peak raw concurrency = %d, want <= %d", got, defaultParallelToolCallLimit)
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolStage_MultipleTools_SequentialBarrierSkipsParallelRawPath(t *testing.T) {
|
||||
t.Parallel()
|
||||
calls := []string{}
|
||||
@@ -2306,8 +2739,8 @@ func TestParseTTL_ValidInputs(t *testing.T) {
|
||||
{"5m", 5 * time.Minute},
|
||||
{"30s", 30 * time.Second},
|
||||
{"1h30m", 90 * time.Minute},
|
||||
{"bogus", 5 * time.Minute}, // invalid → fallback
|
||||
{"-1m", 5 * time.Minute}, // negative → fallback
|
||||
{"bogus", 5 * time.Minute}, // invalid → fallback
|
||||
{"-1m", 5 * time.Minute}, // negative → fallback
|
||||
}
|
||||
for _, tc := range cases {
|
||||
got := parseTTL(tc.in)
|
||||
|
||||
+129
-40
@@ -4,8 +4,11 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"slices"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/hooks"
|
||||
@@ -13,7 +16,10 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
const maxSequentialWaitBatchMs = 300000
|
||||
const (
|
||||
maxSequentialWaitBatchMs = 300000
|
||||
defaultParallelToolCallLimit = 4
|
||||
)
|
||||
|
||||
// ToolStage runs per iteration after PruneStage. Executes tool calls from
|
||||
// ThinkState.LastResponse, checks exit conditions (loop kill, read-only streak, budget).
|
||||
@@ -46,8 +52,9 @@ func (s *ToolStage) Execute(ctx context.Context, state *RunState) error {
|
||||
|
||||
// Parallel path: separate I/O (parallel) from state mutation (sequential).
|
||||
// Requires both ExecuteToolRaw and ProcessToolResult callbacks.
|
||||
if len(toolCalls) > 1 && s.deps.ExecuteToolRaw != nil && s.deps.ProcessToolResult != nil && !s.requiresSequential(toolCalls) {
|
||||
return s.executeParallel(ctx, state, toolCalls)
|
||||
if len(toolCalls) > 1 && s.canExecuteParallel(toolCalls) && !s.batchExceedsBudget(state, toolCalls) {
|
||||
preflight := s.preflightToolCalls(ctx, state, toolCalls)
|
||||
return s.executeParallel(ctx, state, preflight)
|
||||
}
|
||||
|
||||
// Sequential fallback: ExecuteToolCall handles both I/O and state mutation.
|
||||
@@ -64,28 +71,11 @@ func (s *ToolStage) Execute(ctx context.Context, state *RunState) error {
|
||||
}
|
||||
}
|
||||
|
||||
// Hook: sync PreToolUse — block if hook denies. Builtin-source hooks may
|
||||
// rewrite tc.Arguments via UpdatedToolInput (e.g. path-sanitizer); apply
|
||||
// before ExecuteToolCall so the rewrite is authoritative.
|
||||
if r, _ := s.deps.FireHook(ctx, hooks.Event{
|
||||
EventID: uuid.NewString(),
|
||||
SessionID: state.Input.SessionKey,
|
||||
TenantID: store.TenantIDFromContext(ctx),
|
||||
AgentID: store.AgentIDFromContext(ctx),
|
||||
ToolName: tc.Name,
|
||||
ToolInput: tc.Arguments,
|
||||
HookEvent: hooks.EventPreToolUse,
|
||||
}); r.Decision == hooks.DecisionBlock {
|
||||
// Inject synthetic blocked tool message and skip actual execution.
|
||||
state.Messages.AppendPending(providers.Message{
|
||||
Role: "tool",
|
||||
Content: "Hook blocked: pre_tool_use",
|
||||
ToolCallID: tc.ID,
|
||||
})
|
||||
tc, blocked := s.preflightToolCall(ctx, state, tc)
|
||||
if blocked != nil {
|
||||
state.Messages.AppendPending(*blocked)
|
||||
state.Tool.TotalToolCalls++
|
||||
continue
|
||||
} else if r.UpdatedToolInput != nil {
|
||||
tc.Arguments = r.UpdatedToolInput
|
||||
}
|
||||
|
||||
msgs, err := s.deps.ExecuteToolCall(ctx, state, tc)
|
||||
@@ -129,12 +119,75 @@ func (s *ToolStage) requiresSequential(toolCalls []providers.ToolCall) bool {
|
||||
if s.deps.SequentialToolCall == nil {
|
||||
return false
|
||||
}
|
||||
for _, tc := range toolCalls {
|
||||
if s.deps.SequentialToolCall(tc) {
|
||||
return true
|
||||
return slices.ContainsFunc(toolCalls, s.deps.SequentialToolCall)
|
||||
}
|
||||
|
||||
type toolCallPreflightItem struct {
|
||||
index int
|
||||
tc providers.ToolCall
|
||||
blocked *providers.Message
|
||||
}
|
||||
|
||||
type toolCallPreflight struct {
|
||||
ordered []toolCallPreflightItem
|
||||
executable []toolCallPreflightItem
|
||||
}
|
||||
|
||||
func (s *ToolStage) preflightToolCalls(ctx context.Context, state *RunState, toolCalls []providers.ToolCall) toolCallPreflight {
|
||||
out := toolCallPreflight{
|
||||
ordered: make([]toolCallPreflightItem, 0, len(toolCalls)),
|
||||
}
|
||||
for i, tc := range toolCalls {
|
||||
tc, blocked := s.preflightToolCall(ctx, state, tc)
|
||||
item := toolCallPreflightItem{index: i, tc: tc, blocked: blocked}
|
||||
out.ordered = append(out.ordered, item)
|
||||
if blocked == nil {
|
||||
out.executable = append(out.executable, item)
|
||||
}
|
||||
}
|
||||
return false
|
||||
return out
|
||||
}
|
||||
|
||||
func (s *ToolStage) preflightToolCall(ctx context.Context, state *RunState, tc providers.ToolCall) (providers.ToolCall, *providers.Message) {
|
||||
r, _ := s.deps.FireHook(ctx, hooks.Event{
|
||||
EventID: uuid.NewString(),
|
||||
SessionID: state.Input.SessionKey,
|
||||
TenantID: store.TenantIDFromContext(ctx),
|
||||
AgentID: store.AgentIDFromContext(ctx),
|
||||
ToolName: tc.Name,
|
||||
ToolInput: tc.Arguments,
|
||||
HookEvent: hooks.EventPreToolUse,
|
||||
})
|
||||
if r.Decision == hooks.DecisionBlock {
|
||||
return tc, &providers.Message{
|
||||
Role: "tool",
|
||||
Content: "Hook blocked: pre_tool_use",
|
||||
ToolCallID: tc.ID,
|
||||
}
|
||||
}
|
||||
if r.UpdatedToolInput != nil {
|
||||
tc.Arguments = r.UpdatedToolInput
|
||||
}
|
||||
return tc, nil
|
||||
}
|
||||
|
||||
func (s *ToolStage) canExecuteParallel(toolCalls []providers.ToolCall) bool {
|
||||
if s.deps.ExecuteToolRaw == nil || s.deps.ProcessToolResult == nil || s.deps.ParallelEligibleToolCall == nil {
|
||||
return false
|
||||
}
|
||||
if s.requiresSequential(toolCalls) {
|
||||
return false
|
||||
}
|
||||
for _, tc := range toolCalls {
|
||||
if !s.deps.ParallelEligibleToolCall(tc) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (s *ToolStage) batchExceedsBudget(state *RunState, toolCalls []providers.ToolCall) bool {
|
||||
return s.deps.Config.MaxToolCalls > 0 && state.Tool.TotalToolCalls+len(toolCalls) > s.deps.Config.MaxToolCalls
|
||||
}
|
||||
|
||||
func (s *ToolStage) shouldStopBeforeTool(ctx context.Context, state *RunState) bool {
|
||||
@@ -171,40 +224,76 @@ func toolCallTimeMs(tc providers.ToolCall) int {
|
||||
}
|
||||
|
||||
// executeParallel runs tool I/O concurrently, then processes results sequentially.
|
||||
func (s *ToolStage) executeParallel(ctx context.Context, state *RunState, toolCalls []providers.ToolCall) error {
|
||||
func (s *ToolStage) executeParallel(ctx context.Context, state *RunState, preflight toolCallPreflight) error {
|
||||
type rawResult struct {
|
||||
index int
|
||||
tc providers.ToolCall
|
||||
msg providers.Message
|
||||
rawData any
|
||||
err error
|
||||
}
|
||||
|
||||
startedAt := time.Now()
|
||||
slog.Info("tool.parallel.batch.start",
|
||||
"session", state.Input.SessionKey,
|
||||
"run_id", state.RunID,
|
||||
"count", len(preflight.executable),
|
||||
"limit", defaultParallelToolCallLimit)
|
||||
|
||||
// Phase 1: parallel I/O (no state mutation)
|
||||
results := make([]rawResult, len(toolCalls))
|
||||
results := make([]rawResult, len(preflight.executable))
|
||||
sem := make(chan struct{}, defaultParallelToolCallLimit)
|
||||
var wg sync.WaitGroup
|
||||
for i, tc := range toolCalls {
|
||||
for i, item := range preflight.executable {
|
||||
wg.Add(1)
|
||||
go func(idx int, tc providers.ToolCall) {
|
||||
go func(idx int, item toolCallPreflightItem) {
|
||||
defer wg.Done()
|
||||
msg, rawData, err := s.deps.ExecuteToolRaw(ctx, tc)
|
||||
results[idx] = rawResult{tc: tc, msg: msg, rawData: rawData, err: err}
|
||||
}(i, tc)
|
||||
select {
|
||||
case sem <- struct{}{}:
|
||||
defer func() { <-sem }()
|
||||
case <-ctx.Done():
|
||||
results[idx] = rawResult{index: item.index, tc: item.tc, err: ctx.Err()}
|
||||
return
|
||||
}
|
||||
msg, rawData, err := s.deps.ExecuteToolRaw(ctx, item.tc)
|
||||
results[idx] = rawResult{index: item.index, tc: item.tc, msg: msg, rawData: rawData, err: err}
|
||||
}(i, item)
|
||||
}
|
||||
wg.Wait()
|
||||
slog.Info("tool.parallel.batch.end",
|
||||
"session", state.Input.SessionKey,
|
||||
"run_id", state.RunID,
|
||||
"count", len(preflight.executable),
|
||||
"limit", defaultParallelToolCallLimit,
|
||||
"duration_ms", time.Since(startedAt).Milliseconds())
|
||||
|
||||
// Phase 2: sequential state mutation (safe, deterministic order)
|
||||
resultByIndex := make(map[int]rawResult, len(results))
|
||||
for _, r := range results {
|
||||
if r.err != nil {
|
||||
return fmt.Errorf("execute tool %s: %w", r.tc.Name, r.err)
|
||||
resultByIndex[r.index] = r
|
||||
}
|
||||
for _, item := range preflight.ordered {
|
||||
tc := item.tc
|
||||
if item.blocked != nil {
|
||||
state.Messages.AppendPending(*item.blocked)
|
||||
state.Tool.TotalToolCalls++
|
||||
continue
|
||||
}
|
||||
processed := s.deps.ProcessToolResult(ctx, state, r.tc, r.msg, r.rawData)
|
||||
r, ok := resultByIndex[item.index]
|
||||
if !ok {
|
||||
return fmt.Errorf("execute tool %s: missing parallel result", tc.Name)
|
||||
}
|
||||
if r.err != nil {
|
||||
return fmt.Errorf("execute tool %s: %w", tc.Name, r.err)
|
||||
}
|
||||
processed := s.deps.ProcessToolResult(ctx, state, tc, r.msg, r.rawData)
|
||||
for _, msg := range processed {
|
||||
state.Messages.AppendPending(msg)
|
||||
}
|
||||
state.Tool.TotalToolCalls++
|
||||
|
||||
// Hook: async PostToolUse for parallel path — fire and forget.
|
||||
// PreToolUse is not instrumented in the parallel path (TODO: add when parallel path matures).
|
||||
// PreToolUse already ran in preflight before any raw I/O was scheduled.
|
||||
if s.deps.Hooks != nil {
|
||||
detached := context.WithoutCancel(ctx)
|
||||
go s.deps.FireHook(detached, hooks.Event{ //nolint:errcheck
|
||||
@@ -212,8 +301,8 @@ func (s *ToolStage) executeParallel(ctx context.Context, state *RunState, toolCa
|
||||
SessionID: state.Input.SessionKey,
|
||||
TenantID: store.TenantIDFromContext(ctx),
|
||||
AgentID: store.AgentIDFromContext(ctx),
|
||||
ToolName: r.tc.Name,
|
||||
ToolInput: r.tc.Arguments,
|
||||
ToolName: tc.Name,
|
||||
ToolInput: tc.Arguments,
|
||||
HookEvent: hooks.EventPostToolUse,
|
||||
})
|
||||
}
|
||||
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
---
|
||||
phase: 1
|
||||
title: "Characterize Current Parallel Semantics"
|
||||
status: complete
|
||||
priority: P1
|
||||
effort: ""
|
||||
dependencies: []
|
||||
---
|
||||
|
||||
# Phase 1: Characterize Current Parallel Semantics
|
||||
|
||||
## Context Links
|
||||
|
||||
- Plan: [plan.md](./plan.md)
|
||||
- Existing executor: `internal/pipeline/tool_stage.go`
|
||||
- Existing tests: `internal/pipeline/stages_test.go`
|
||||
- Loop callbacks: `internal/agent/loop_pipeline_tool_callbacks.go`
|
||||
|
||||
## Overview
|
||||
|
||||
Add tests first around the current behavior and the desired safety boundaries. This phase should fail on the missing hardening cases before implementation starts.
|
||||
|
||||
## Requirements
|
||||
|
||||
- Functional: capture current two-phase parallel behavior, stable ordered processing, and sequential fallback.
|
||||
- Functional: prove current gaps around hook parity, mutating eligibility, and tool budget.
|
||||
- Non-functional: tests deterministic, no sleep-based timing flake where avoidable.
|
||||
|
||||
## Architecture
|
||||
|
||||
Use `ToolStage` tests with fake callbacks. Keep tests at pipeline layer first because this is where scheduling, hook firing, budget, and result ordering live. Use small channels/atomic counters only where concurrency must be proven.
|
||||
|
||||
## Related Code Files
|
||||
|
||||
- Modify: `internal/pipeline/stages_test.go`
|
||||
- Read: `internal/pipeline/tool_stage.go`
|
||||
- Read: `internal/pipeline/deps.go`
|
||||
- Read: `internal/agent/loop_pipeline_adapter.go`
|
||||
|
||||
## Implementation Steps
|
||||
|
||||
1. Add a test proving parallel processing keeps original result order even when raw execution completes out of order.
|
||||
2. Add a test proving `PreToolUse` must run before parallel raw execution and can rewrite args.
|
||||
3. Add a test proving blocked `PreToolUse` calls append synthetic blocked tool messages and do not call raw execution.
|
||||
4. Add a test proving a batch containing a mutating tool uses sequential execution.
|
||||
5. Add a test proving a batch containing `exec`, `bash`, `wait`, MCP/unknown, or async tools uses sequential execution.
|
||||
6. Add a test proving `MaxToolCalls` prevents scheduling extra calls in a parallel batch.
|
||||
7. Add a test proving fixed concurrency cap bounds peak raw executions.
|
||||
|
||||
## Success Criteria
|
||||
|
||||
- [x] New tests fail against current implementation for the real gaps.
|
||||
- [x] Existing parallel path regression remains covered.
|
||||
- [x] No implementation changes in this phase except test helpers.
|
||||
|
||||
## Risk Assessment
|
||||
|
||||
Risk: concurrency tests can become flaky. Mitigation: use controlled channels and atomic peak counters, not wall-clock timing.
|
||||
|
||||
## Security Considerations
|
||||
|
||||
Hook blocking and mutating-tool fallback are security boundaries. Tests must assert denial happens before tool execution.
|
||||
|
||||
## Next Steps
|
||||
|
||||
Proceed to Phase 2 only after the tests express all approved safety decisions.
|
||||
+94
@@ -0,0 +1,94 @@
|
||||
---
|
||||
phase: 2
|
||||
title: "Harden Eligibility Hooks and Budgets"
|
||||
status: complete
|
||||
priority: P1
|
||||
effort: ""
|
||||
dependencies: [1]
|
||||
---
|
||||
|
||||
# Phase 2: Harden Eligibility Hooks and Budgets
|
||||
|
||||
## Context Links
|
||||
|
||||
- Phase 1 tests: [phase-01-characterize-current-parallel-semantics.md](./phase-01-characterize-current-parallel-semantics.md)
|
||||
- Tool metadata: `internal/tools/capability.go`
|
||||
- Registry metadata lookup: `internal/tools/registry.go`
|
||||
- Pipeline deps: `internal/pipeline/deps.go`
|
||||
|
||||
## Overview
|
||||
|
||||
Implement conservative eligibility and pre-scheduling safety gates. Keep the existing raw/process split, but only enter it for fully safe batches.
|
||||
|
||||
## Requirements
|
||||
|
||||
- Functional: parallel path only for calls marked read-only by metadata.
|
||||
- Functional: `exec`, `bash`, `wait`, mutating, async, MCP-bridged, and unknown tools remain sequential.
|
||||
- Functional: `PreToolUse` runs before parallel I/O and can block or rewrite args.
|
||||
- Functional: `MaxToolCalls` is enforced before scheduling a parallel batch.
|
||||
- Non-functional: single-tool and sequential behavior unchanged.
|
||||
|
||||
## Architecture
|
||||
|
||||
Add a small callback in `PipelineDeps`, likely `ParallelEligibleToolCall func(tc providers.ToolCall) bool`. Wire it from `agent.Loop` using `l.tools.GetMetadata(l.resolveToolCallName(tc.Name))`.
|
||||
|
||||
Suggested rule:
|
||||
|
||||
```go
|
||||
func eligible(tc providers.ToolCall) bool {
|
||||
name := resolve(tc.Name)
|
||||
if name == "exec" || name == "bash" || name == "wait" || strings.HasPrefix(name, "mcp_") {
|
||||
return false
|
||||
}
|
||||
meta := registry.GetMetadata(name)
|
||||
return meta.IsReadOnly() &&
|
||||
!meta.HasCapability(tools.CapMutating) &&
|
||||
!meta.HasCapability(tools.CapAsync) &&
|
||||
!meta.HasCapability(tools.CapMCPBridged)
|
||||
}
|
||||
```
|
||||
|
||||
Unknown tools should be treated as sequential. If `inferMetadata` currently defaults unknown tools to mutating, preserve that behavior.
|
||||
|
||||
## Related Code Files
|
||||
|
||||
- Modify: `internal/pipeline/deps.go`
|
||||
- Modify: `internal/pipeline/tool_stage.go`
|
||||
- Modify: `internal/agent/loop_pipeline_adapter.go`
|
||||
- Modify: `internal/pipeline/stages_test.go`
|
||||
- Read: `internal/tools/capability.go`
|
||||
- Read: `internal/tools/registry.go`
|
||||
|
||||
## Implementation Steps
|
||||
|
||||
1. Add `ParallelEligibleToolCall` to `PipelineDeps`.
|
||||
2. Replace `requiresSequential` with a clearer decision helper:
|
||||
- return sequential when any `SequentialToolCall` matches
|
||||
- return sequential when any call is not parallel eligible
|
||||
- return sequential when required callbacks are nil
|
||||
3. Move `PreToolUse` processing into a shared preflight helper used before both sequential execution and parallel scheduling.
|
||||
4. For blocked hooks, append blocked tool messages, increment tool count, and exclude those calls from raw execution.
|
||||
5. Apply hook argument rewrites before eligibility and scheduling decisions.
|
||||
6. Enforce `MaxToolCalls` against the whole executable batch before launching goroutines.
|
||||
7. Keep `PostToolUse` after result processing, preserving existing async semantics.
|
||||
|
||||
## Success Criteria
|
||||
|
||||
- [x] Phase 1 tests for hook parity, eligibility, and budget pass.
|
||||
- [x] Existing tests for sequential `wait` barrier still pass.
|
||||
- [x] Single-tool calls still use `ExecuteToolCall`.
|
||||
- [x] No mutating tool can enter `ExecuteToolRaw` by default.
|
||||
|
||||
## Risk Assessment
|
||||
|
||||
Risk: a tool inferred read-only is actually stateful. Mitigation: start with explicit conservative checks and no UI override.
|
||||
|
||||
Risk: hook rewrite changes eligibility. Mitigation: apply rewrite before eligibility decision.
|
||||
|
||||
## Security Considerations
|
||||
|
||||
Do not run any tool raw path before `PreToolUse` has allowed it. A blocked hook must be a hard stop for that call.
|
||||
|
||||
## Next Steps
|
||||
|
||||
Proceed to Phase 3 after eligibility and hook behavior pass focused tests.
|
||||
+82
@@ -0,0 +1,82 @@
|
||||
---
|
||||
phase: 3
|
||||
title: "Add Concurrency Cap and Observability"
|
||||
status: complete
|
||||
priority: P2
|
||||
effort: ""
|
||||
dependencies: [2]
|
||||
---
|
||||
|
||||
# Phase 3: Add Concurrency Cap and Observability
|
||||
|
||||
## Context Links
|
||||
|
||||
- Phase 2 safety gates: [phase-02-harden-eligibility-hooks-and-budgets.md](./phase-02-harden-eligibility-hooks-and-budgets.md)
|
||||
- Tool spans: `internal/agent/loop_pipeline_tool_callbacks.go`
|
||||
- Tracing docs: `docs/10-tracing-observability.md`
|
||||
|
||||
## Overview
|
||||
|
||||
Bound parallel execution with a fixed default and make batches debuggable in logs/traces without adding new settings UI.
|
||||
|
||||
## Requirements
|
||||
|
||||
- Functional: parallel raw I/O uses fixed bounded concurrency.
|
||||
- Functional: stable result aggregation/order remains unchanged.
|
||||
- Functional: partial per-tool errors stay visible as per-call tool results when represented as `tools.Result`.
|
||||
- Non-functional: no user-facing config or web UI in this phase.
|
||||
- Non-functional: log fields must avoid full arguments/secrets.
|
||||
|
||||
## Architecture
|
||||
|
||||
Add a package-level constant in `internal/pipeline/tool_stage.go`, for example:
|
||||
|
||||
```go
|
||||
const defaultParallelToolCallLimit = 4
|
||||
```
|
||||
|
||||
Use a semaphore channel inside `executeParallel`. Store results by original index as today. Do not reorder by finish time.
|
||||
|
||||
Observability should be low-noise:
|
||||
- batch start/end debug/info logs with run/session, count, limit
|
||||
- per-call index/name/id in existing `tool.call` and span data
|
||||
- avoid full argument logging beyond existing `args_len`
|
||||
|
||||
If trace attributes are easy through existing span APIs, add batch index/count metadata. If not, keep this phase to logs and existing per-tool spans to avoid a tracing schema detour.
|
||||
|
||||
## Related Code Files
|
||||
|
||||
- Modify: `internal/pipeline/tool_stage.go`
|
||||
- Modify: `internal/pipeline/stages_test.go`
|
||||
- Optional modify: `internal/agent/loop_pipeline_tool_callbacks.go`
|
||||
- Optional modify: `docs/10-tracing-observability.md`
|
||||
|
||||
## Implementation Steps
|
||||
|
||||
1. Add `defaultParallelToolCallLimit`.
|
||||
2. Gate raw goroutines through a semaphore.
|
||||
3. Add a unit test that raw execution peak never exceeds the fixed limit.
|
||||
4. Add batch start/end logs with safe metadata.
|
||||
5. Verify cancellation behavior still waits for launched goroutines and does not orphan spans.
|
||||
6. If simple, enrich tool span metadata with batch index/count; otherwise document why logs + existing spans are sufficient for v1.
|
||||
|
||||
## Success Criteria
|
||||
|
||||
- [x] Peak raw execution count is bounded in test.
|
||||
- [x] Result order remains original assistant tool-call order.
|
||||
- [x] Logs indicate when a parallel batch starts and completes.
|
||||
- [x] No settings schema, migration, API, or web UI changes introduced.
|
||||
|
||||
## Risk Assessment
|
||||
|
||||
Risk: too low cap leaves latency wins on table. Mitigation: fixed `4` is conservative; make configurable later only with evidence.
|
||||
|
||||
Risk: logs leak sensitive args. Mitigation: log tool names, ids, counts, durations, args length only.
|
||||
|
||||
## Security Considerations
|
||||
|
||||
Do not emit raw tool args or outputs in new logs. Existing trace verbose mode controls larger previews.
|
||||
|
||||
## Next Steps
|
||||
|
||||
Proceed to Phase 4 after bounded parallelism is verified.
|
||||
+77
@@ -0,0 +1,77 @@
|
||||
---
|
||||
phase: 4
|
||||
title: "Validate Document and Ship"
|
||||
status: in_progress
|
||||
priority: P1
|
||||
effort: ""
|
||||
dependencies: [3]
|
||||
---
|
||||
|
||||
# Phase 4: Validate Document and Ship
|
||||
|
||||
## Context Links
|
||||
|
||||
- Project docs: `docs/03-tools-system.md`, `docs/01-agent-loop.md`, `docs/10-tracing-observability.md`
|
||||
- Post-implementation checklist: `CLAUDE.md`
|
||||
- GitHub issue: digitopvn/goclaw#81
|
||||
|
||||
## Overview
|
||||
|
||||
Run focused and repo-level verification, update docs if behavior wording changes, then prepare the branch for PR/issue handoff.
|
||||
|
||||
## Requirements
|
||||
|
||||
- Functional: all acceptance criteria from issue #81 covered by tests or documented constraints.
|
||||
- Non-functional: no flaky load/stress/benchmark tests.
|
||||
- Non-functional: docs describe conservative read-only-only default.
|
||||
|
||||
## Architecture
|
||||
|
||||
Validation stays layered:
|
||||
- focused unit tests first
|
||||
- agent/tool package tests next
|
||||
- compile checks for PostgreSQL and SQLite builds
|
||||
- docs update only for changed runtime semantics
|
||||
|
||||
## Related Code Files
|
||||
|
||||
- Modify if needed: `docs/03-tools-system.md`
|
||||
- Modify if needed: `docs/01-agent-loop.md`
|
||||
- Modify if needed: `docs/10-tracing-observability.md`
|
||||
- Read: `CLAUDE.md`
|
||||
|
||||
## Implementation Steps
|
||||
|
||||
1. Run focused tests:
|
||||
- `go test ./internal/pipeline`
|
||||
- `go test ./internal/agent`
|
||||
- `go test ./internal/tools`
|
||||
2. Run compile checks:
|
||||
- `go build ./...`
|
||||
- `go build -tags sqliteonly ./...`
|
||||
3. Run `go vet ./...` if compile checks pass.
|
||||
4. Update docs only where current docs overstate behavior or omit new conservative eligibility.
|
||||
5. Review diff for accidental schema/UI/config changes.
|
||||
6. Commit and push implementation.
|
||||
7. Open PR against `dev`.
|
||||
8. Reply to issue #81 with PR link, validation proof, and any out-of-scope follow-up.
|
||||
|
||||
## Success Criteria
|
||||
|
||||
- [x] Focused tests pass.
|
||||
- [x] PostgreSQL and SQLite builds pass.
|
||||
- [x] `go vet ./...` passes or any existing unrelated failure is documented with evidence.
|
||||
- [x] Docs match shipped behavior.
|
||||
- [ ] GitHub issue has implementation summary and PR link.
|
||||
|
||||
## Risk Assessment
|
||||
|
||||
Risk: full integration tests require external PostgreSQL/pgvector. Mitigation: do focused unit/build checks here; run integration only if a touched path warrants DB coverage and environment exists.
|
||||
|
||||
## Security Considerations
|
||||
|
||||
Verify no new config, secrets, or credential values are committed. Keep mutating tool fallback conservative.
|
||||
|
||||
## Next Steps
|
||||
|
||||
After PR is open, watch CI and fix failures. Do not mark issue complete until PR is merged or maintainer accepts the plan.
|
||||
@@ -0,0 +1,75 @@
|
||||
---
|
||||
title: "Issue 81 Parallel Tool Call Hardening"
|
||||
description: "Harden existing multi-tool parallel execution so only safe independent read-only calls run concurrently."
|
||||
status: in_progress
|
||||
priority: P1
|
||||
issue: 81
|
||||
branch: "codex/issue-81-parallel-tool-calls"
|
||||
tags: [tools, pipeline, parallel, tdd, issue-81]
|
||||
blockedBy: []
|
||||
blocks: []
|
||||
created: "2026-05-28T11:33:10.426Z"
|
||||
createdBy: "ck:plan"
|
||||
source: skill
|
||||
---
|
||||
|
||||
# Issue 81 Parallel Tool Call Hardening
|
||||
|
||||
## Overview
|
||||
|
||||
Issue #81 asks GoClaw to execute independent tool calls in parallel while preserving deterministic transcript order, rate limits, safety, and observability.
|
||||
|
||||
Current `dev` already has a parallel path in `internal/pipeline/tool_stage.go`: multi-tool batches call `ExecuteToolRaw` concurrently, then process results sequentially by original index. That is the right architecture. The plan is to harden it, not rewrite it.
|
||||
|
||||
Approved decisions:
|
||||
- Keep the existing two-phase parallel path.
|
||||
- Parallelize read-only tools only.
|
||||
- Keep `exec`, `bash`, `wait`, mutating tools, async tools, MCP-bridged tools, and unknown tools sequential by default.
|
||||
- Use a fixed in-code concurrency default now. Web/UI configurability is out of scope until users need it.
|
||||
|
||||
Main gaps to close:
|
||||
- Parallel path currently skips synchronous `PreToolUse` hooks.
|
||||
- Any non-`wait` multi-tool batch can enter the parallel path, including mutating tools.
|
||||
- No fixed batch concurrency cap.
|
||||
- `MaxToolCalls` is checked per sequential call, but parallel scheduling can overshoot unless gated before launch.
|
||||
|
||||
Design summary: [reports/brainstorm-summary.md](./reports/brainstorm-summary.md)
|
||||
|
||||
## Phases
|
||||
|
||||
| Phase | Name | Status |
|
||||
|-------|------|--------|
|
||||
| 1 | [Characterize Current Parallel Semantics](./phase-01-characterize-current-parallel-semantics.md) | Complete |
|
||||
| 2 | [Harden Eligibility Hooks and Budgets](./phase-02-harden-eligibility-hooks-and-budgets.md) | Complete |
|
||||
| 3 | [Add Concurrency Cap and Observability](./phase-03-add-concurrency-cap-and-observability.md) | Complete |
|
||||
| 4 | [Validate Document and Ship](./phase-04-validate-document-and-ship.md) | In Progress |
|
||||
|
||||
## Dependencies
|
||||
|
||||
- GitHub issue: digitopvn/goclaw#81
|
||||
- Existing executor: `internal/pipeline/tool_stage.go`
|
||||
- Pipeline dependency contract: `internal/pipeline/deps.go`
|
||||
- Loop wiring: `internal/agent/loop_pipeline_adapter.go`
|
||||
- Tool metadata: `internal/tools/capability.go`, `internal/tools/registry.go`
|
||||
- Existing regression tests: `internal/pipeline/stages_test.go`
|
||||
|
||||
No unfinished project plan currently blocks this plan.
|
||||
|
||||
## Success Criteria
|
||||
|
||||
- [x] Read-only multi-tool batches execute concurrently with stable result order.
|
||||
- [x] Mutating, async, MCP/unknown, `exec`/`bash`, and `wait` batches stay sequential by default.
|
||||
- [x] Sync `PreToolUse` hooks run before any parallel I/O and can block or rewrite arguments.
|
||||
- [x] Tool-call budget is respected before scheduling a parallel batch.
|
||||
- [x] Parallel I/O uses a fixed bounded concurrency default.
|
||||
- [x] Logs/traces expose enough batch metadata to debug concurrent execution.
|
||||
- [x] Focused unit tests prove both parallel and sequential safety paths.
|
||||
- [x] No production behavior change for single-tool calls.
|
||||
|
||||
## Validation
|
||||
|
||||
- `go test ./internal/pipeline`
|
||||
- `go test ./internal/agent`
|
||||
- `go test ./internal/tools`
|
||||
- `go build ./...`
|
||||
- `go build -tags sqliteonly ./...`
|
||||
@@ -0,0 +1,67 @@
|
||||
---
|
||||
title: "Issue 81 Parallel Tool Call Hardening Brainstorm Summary"
|
||||
status: approved
|
||||
source: ck:brainstorm
|
||||
issue: 81
|
||||
---
|
||||
|
||||
# Issue 81 Parallel Tool Call Hardening Brainstorm Summary
|
||||
|
||||
## Problem
|
||||
|
||||
GoClaw should reduce latency by running independent tool calls concurrently. Current `dev` already has a parallel path, but it is too broad: any multi-tool batch except `wait` can parallelize, sync `PreToolUse` hooks are skipped, and there is no fixed raw-execution concurrency cap.
|
||||
|
||||
## Requirements
|
||||
|
||||
- Preserve sequential execution when calls depend on earlier results.
|
||||
- Run safe read-only multi-tool batches concurrently.
|
||||
- Keep deterministic result order in transcript/UI.
|
||||
- Respect tool budget, rate limits, hooks, and cancellation.
|
||||
- Avoid parallelizing destructive operations by default.
|
||||
- Make behavior debuggable in logs/traces.
|
||||
|
||||
## Approved Decisions
|
||||
|
||||
- Harden existing implementation, do not rewrite.
|
||||
- Parallel eligibility is read-only tools only.
|
||||
- `exec`, `bash`, `wait`, mutating tools, async tools, MCP/unknown tools stay sequential by default.
|
||||
- Use a fixed default concurrency limit now.
|
||||
- Web UI/configurability for concurrency is deferred until user demand exists.
|
||||
|
||||
## Evaluated Approaches
|
||||
|
||||
| Approach | Verdict | Reason |
|
||||
|---|---|---|
|
||||
| Eligibility guard only | Reject | Too narrow; leaves hook and cap gaps. |
|
||||
| Hardened existing executor | Accept | Smallest change that satisfies safety and acceptance criteria. |
|
||||
| Full dependency planner | Reject | Over-engineered; dependency inference from tool args is unreliable. |
|
||||
|
||||
## Recommended Solution
|
||||
|
||||
Keep `ToolStage` two-phase design:
|
||||
|
||||
1. Preflight sync hooks sequentially.
|
||||
2. Block or rewrite calls before scheduling.
|
||||
3. Check every remaining call is parallel-eligible.
|
||||
4. Enforce budget before launch.
|
||||
5. Run raw I/O with fixed semaphore limit.
|
||||
6. Process results sequentially in original order.
|
||||
|
||||
## Risks
|
||||
|
||||
- Some tools inferred read-only may still mutate internal caches. Mitigate by conservative eligibility and explicit exclusions.
|
||||
- Concurrency tests may flake. Mitigate with controlled channels and atomic peak counters.
|
||||
- Logs could leak data. Mitigate by logging counts, names, ids, durations, and `args_len`, not raw args.
|
||||
|
||||
## Validation Criteria
|
||||
|
||||
- Read-only multi-tool batch proves concurrent raw execution.
|
||||
- Mutating/exec/wait/MCP/unknown batches prove sequential fallback.
|
||||
- Hook block/rewrite proves no raw execution happens before sync hook approval.
|
||||
- Result order stays stable.
|
||||
- Fixed concurrency cap is enforced.
|
||||
- Focused tests and build checks pass.
|
||||
|
||||
## Unresolved Questions
|
||||
|
||||
None.
|
||||
Reference in new issue
Block a user