From 484bcc7b7160fce0e8ed0bd846f0e45d35253268 Mon Sep 17 00:00:00 2001 From: Duy /zuey/ Date: Sun, 31 May 2026 12:24:00 +0700 Subject: [PATCH] fix: harden parallel tool scheduling Squash merge PR #103 after validation. --- docs/01-agent-loop.md | 7 +- docs/03-tools-system.md | 2 + internal/agent/loop.go | 22 + internal/agent/loop_pipeline_adapter.go | 59 +-- .../loop_pipeline_tool_callbacks_test.go | 73 ++- internal/pipeline/deps.go | 29 +- internal/pipeline/stages_test.go | 439 +++++++++++++++++- internal/pipeline/tool_stage.go | 169 +++++-- ...characterize-current-parallel-semantics.md | 66 +++ ...02-harden-eligibility-hooks-and-budgets.md | 94 ++++ ...3-add-concurrency-cap-and-observability.md | 82 ++++ .../phase-04-validate-document-and-ship.md | 77 +++ .../plan.md | 75 +++ .../reports/brainstorm-summary.md | 67 +++ 14 files changed, 1170 insertions(+), 91 deletions(-) create mode 100644 plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-01-characterize-current-parallel-semantics.md create mode 100644 plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-02-harden-eligibility-hooks-and-budgets.md create mode 100644 plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-03-add-concurrency-cap-and-observability.md create mode 100644 plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-04-validate-document-and-ship.md create mode 100644 plans/260528-1833-issue-81-parallel-tool-call-hardening/plan.md create mode 100644 plans/260528-1833-issue-81-parallel-tool-call-hardening/reports/brainstorm-summary.md diff --git a/docs/01-agent-loop.md b/docs/01-agent-loop.md index 0bdd0291..e88e3d78 100644 --- a/docs/01-agent-loop.md +++ b/docs/01-agent-loop.md @@ -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. diff --git a/docs/03-tools-system.md b/docs/03-tools-system.md index 97dd65e8..7bf929d5 100644 --- a/docs/03-tools-system.md +++ b/docs/03-tools-system.md @@ -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 diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 3cc398ad..cd61a67d 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -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 diff --git a/internal/agent/loop_pipeline_adapter.go b/internal/agent/loop_pipeline_adapter.go index 13f6325b..29e8e76b 100644 --- a/internal/agent/loop_pipeline_adapter.go +++ b/internal/agent/loop_pipeline_adapter.go @@ -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, } } diff --git a/internal/agent/loop_pipeline_tool_callbacks_test.go b/internal/agent/loop_pipeline_tool_callbacks_test.go index e1df4dd9..ce4b432a 100644 --- a/internal/agent/loop_pipeline_tool_callbacks_test.go +++ b/internal/agent/loop_pipeline_tool_callbacks_test.go @@ -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) { diff --git a/internal/pipeline/deps.go b/internal/pipeline/deps.go index a8054ca2..2f713f2f 100644 --- a/internal/pipeline/deps.go +++ b/internal/pipeline/deps.go @@ -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 diff --git a/internal/pipeline/stages_test.go b/internal/pipeline/stages_test.go index 8c1204ed..c9678d2b 100644 --- a/internal/pipeline/stages_test.go +++ b/internal/pipeline/stages_test.go @@ -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) diff --git a/internal/pipeline/tool_stage.go b/internal/pipeline/tool_stage.go index ac6c7a1e..9327d2fd 100644 --- a/internal/pipeline/tool_stage.go +++ b/internal/pipeline/tool_stage.go @@ -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, }) } diff --git a/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-01-characterize-current-parallel-semantics.md b/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-01-characterize-current-parallel-semantics.md new file mode 100644 index 00000000..8cf30c4c --- /dev/null +++ b/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-01-characterize-current-parallel-semantics.md @@ -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. diff --git a/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-02-harden-eligibility-hooks-and-budgets.md b/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-02-harden-eligibility-hooks-and-budgets.md new file mode 100644 index 00000000..e3ae72b0 --- /dev/null +++ b/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-02-harden-eligibility-hooks-and-budgets.md @@ -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. diff --git a/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-03-add-concurrency-cap-and-observability.md b/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-03-add-concurrency-cap-and-observability.md new file mode 100644 index 00000000..4637186d --- /dev/null +++ b/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-03-add-concurrency-cap-and-observability.md @@ -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. diff --git a/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-04-validate-document-and-ship.md b/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-04-validate-document-and-ship.md new file mode 100644 index 00000000..ae59029f --- /dev/null +++ b/plans/260528-1833-issue-81-parallel-tool-call-hardening/phase-04-validate-document-and-ship.md @@ -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. diff --git a/plans/260528-1833-issue-81-parallel-tool-call-hardening/plan.md b/plans/260528-1833-issue-81-parallel-tool-call-hardening/plan.md new file mode 100644 index 00000000..dbcf7d11 --- /dev/null +++ b/plans/260528-1833-issue-81-parallel-tool-call-hardening/plan.md @@ -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 ./...` diff --git a/plans/260528-1833-issue-81-parallel-tool-call-hardening/reports/brainstorm-summary.md b/plans/260528-1833-issue-81-parallel-tool-call-hardening/reports/brainstorm-summary.md new file mode 100644 index 00000000..21cda06c --- /dev/null +++ b/plans/260528-1833-issue-81-parallel-tool-call-hardening/reports/brainstorm-summary.md @@ -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.