From 221cd78fcf1a97185fcd4095072d57e13a8a3b80 Mon Sep 17 00:00:00 2001 From: viettranx Date: Tue, 14 Apr 2026 10:09:45 +0700 Subject: [PATCH] fix(agent): emit tool.call event in parallel tool execution path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v3 pipeline's parallel path (makeExecuteToolRaw) skipped the tool.call WebSocket event, so web UI and desktop UI silently dropped tool cards during real-time streaming. Only page refresh (which reloads history) revealed the tool calls. Both UIs rely on tool.call to create entries that later tool.result events can update. Fix: mirror the sequential path's emission in makeExecuteToolRaw. Bus.Broadcast is RWMutex-guarded, safe to call from parallel goroutines. Add tests at two layers to prevent regression: - Pipeline layer (stages_test.go): guards the dispatch contract — multiple tool calls route through ExecuteToolRaw + ProcessToolResult rather than ExecuteToolCall. Previously the parallel path had zero test coverage, which is why this bug escaped. - Agent layer (loop_pipeline_tool_callbacks_test.go): guards the emission contract — both sequential and parallel wrappers emit tool.call with correct payload and routing context. Mutation-verified. --- .../agent/loop_pipeline_tool_callbacks.go | 13 ++ .../loop_pipeline_tool_callbacks_test.go | 188 ++++++++++++++++++ internal/pipeline/stages_test.go | 61 ++++++ 3 files changed, 262 insertions(+) create mode 100644 internal/agent/loop_pipeline_tool_callbacks_test.go diff --git a/internal/agent/loop_pipeline_tool_callbacks.go b/internal/agent/loop_pipeline_tool_callbacks.go index 3012b5a6..b7743f3e 100644 --- a/internal/agent/loop_pipeline_tool_callbacks.go +++ b/internal/agent/loop_pipeline_tool_callbacks.go @@ -63,9 +63,22 @@ type toolRawResult struct { // makeExecuteToolRaw wraps tool I/O only (parallel-safe, no state mutation). // Returns tool message + toolRawResult (with timing + spanID) as opaque raw data for ProcessToolResult. func (l *Loop) makeExecuteToolRaw(req *RunRequest) func(ctx context.Context, tc providers.ToolCall) (providers.Message, any, error) { + emitRun := makeToolEmitRun(l, req) return func(ctx context.Context, tc providers.ToolCall) (providers.Message, any, error) { registryName := l.resolveToolCallName(tc.Name) argsJSON, _ := json.Marshal(tc.Arguments) + slog.Info("tool call", "agent", l.id, "tool", tc.Name, "args_len", len(argsJSON)) + + // Emit tool.call event at I/O start — parity with sequential path (makeExecuteToolCall). + // Without this, parallel tool execution (2+ concurrent tools) never notifies UI of + // tool invocation, so `tool.result` arrives with no matching `tool.call` to update. + // Bus.Broadcast is RWMutex-guarded; safe to call from parallel goroutines. + emitRun(AgentEvent{ + Type: protocol.AgentEventToolCall, + AgentID: l.id, + RunID: req.RunID, + Payload: map[string]any{"name": tc.Name, "id": tc.ID, "arguments": tc.Arguments}, + }) // Emit tool span start (goroutine-safe: channel send only). start := time.Now().UTC() diff --git a/internal/agent/loop_pipeline_tool_callbacks_test.go b/internal/agent/loop_pipeline_tool_callbacks_test.go new file mode 100644 index 00000000..866806c5 --- /dev/null +++ b/internal/agent/loop_pipeline_tool_callbacks_test.go @@ -0,0 +1,188 @@ +package agent + +import ( + "context" + "sync" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/pipeline" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/tools" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +// stubExecutor implements tools.ToolExecutor with a canned successful Result. +// Used to isolate the tool-callback wrappers from real tool registry wiring. +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) 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 } + +// eventCollector buffers AgentEvents for inspection in tests. +// Safe for concurrent appends from parallel goroutines. +type eventCollector struct { + mu sync.Mutex + events []AgentEvent +} + +func (c *eventCollector) onEvent(e AgentEvent) { + c.mu.Lock() + c.events = append(c.events, e) + c.mu.Unlock() +} + +func (c *eventCollector) filter(typ string) []AgentEvent { + c.mu.Lock() + defer c.mu.Unlock() + var out []AgentEvent + for _, e := range c.events { + if e.Type == typ { + out = append(out, e) + } + } + return out +} + +// newTestLoopForToolCallbacks builds a minimal Loop instance sufficient to +// exercise makeExecuteToolCall / makeExecuteToolRaw. All optional subsystems +// (tracing, metrics, input guard) are left nil and hit early-return paths. +func newTestLoopForToolCallbacks(onEvent func(AgentEvent)) *Loop { + return &Loop{ + id: "test-agent", + tools: &stubExecutor{}, + onEvent: onEvent, + } +} + +// TestMakeExecuteToolCall_EmitsToolCallEvent verifies the sequential wrapper +// emits a tool.call event before running tool I/O. +func TestMakeExecuteToolCall_EmitsToolCallEvent(t *testing.T) { + t.Parallel() + col := &eventCollector{} + l := newTestLoopForToolCallbacks(col.onEvent) + + req := &RunRequest{ + RunID: "run-1", + SessionKey: "sess-A", + UserID: "u-1", + Channel: "ws", + RunKind: "", + } + state := &pipeline.RunState{RunID: "run-1"} + tc := providers.ToolCall{ID: "tc-1", Name: "read_file", Arguments: map[string]any{"path": "/tmp/x"}} + + _, err := l.makeExecuteToolCall(req, &runState{})(context.Background(), state, tc) + if err != nil { + t.Fatalf("makeExecuteToolCall returned error: %v", err) + } + + calls := col.filter(protocol.AgentEventToolCall) + if len(calls) != 1 { + t.Fatalf("expected 1 tool.call event, got %d (all events: %+v)", len(calls), col.events) + } + assertToolCallPayload(t, calls[0], tc, req) +} + +// TestMakeExecuteToolRaw_EmitsToolCallEvent is the PRIMARY regression guard. +// The original bug: parallel path (makeExecuteToolRaw) did not emit tool.call, +// so web UI and desktop UI silently dropped tool info during real-time streaming. +// Mutation-verify: remove emitRun(...) from makeExecuteToolRaw — this test must fail. +func TestMakeExecuteToolRaw_EmitsToolCallEvent(t *testing.T) { + t.Parallel() + col := &eventCollector{} + l := newTestLoopForToolCallbacks(col.onEvent) + + req := &RunRequest{ + RunID: "run-2", + SessionKey: "sess-B", + UserID: "u-2", + Channel: "ws", + RunKind: "", + } + tc := providers.ToolCall{ID: "tc-2", Name: "write_file", Arguments: map[string]any{"path": "/tmp/y"}} + + msg, raw, err := l.makeExecuteToolRaw(req)(context.Background(), tc) + if err != nil { + t.Fatalf("makeExecuteToolRaw returned error: %v", err) + } + if msg.Role != "tool" || msg.ToolCallID != tc.ID { + t.Errorf("unexpected tool message: %+v", msg) + } + if raw == nil { + t.Error("expected non-nil raw data (toolRawResult)") + } + + calls := col.filter(protocol.AgentEventToolCall) + if len(calls) != 1 { + t.Fatalf("expected 1 tool.call event, got %d (all events: %+v)", len(calls), col.events) + } + assertToolCallPayload(t, calls[0], tc, req) +} + +// TestMakeExecuteToolRaw_ConcurrentCallsEmitAllEvents confirms the parallel +// wrapper is safe to invoke from multiple goroutines — mirrors the real +// executeParallel dispatch in pipeline/tool_stage.go. +func TestMakeExecuteToolRaw_ConcurrentCallsEmitAllEvents(t *testing.T) { + t.Parallel() + col := &eventCollector{} + l := newTestLoopForToolCallbacks(col.onEvent) + + req := &RunRequest{RunID: "run-3", SessionKey: "sess-C", UserID: "u-3", Channel: "ws"} + exec := l.makeExecuteToolRaw(req) + + const n = 5 + var wg sync.WaitGroup + for i := range n { + wg.Add(1) + go func(idx int) { + defer wg.Done() + tc := providers.ToolCall{ID: "tc-" + string(rune('a'+idx)), Name: "t", Arguments: nil} + if _, _, err := exec(context.Background(), tc); err != nil { + t.Errorf("goroutine %d: %v", idx, err) + } + }(i) + } + wg.Wait() + + calls := col.filter(protocol.AgentEventToolCall) + if len(calls) != n { + t.Fatalf("expected %d tool.call events, got %d", n, len(calls)) + } +} + +// 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) { + t.Helper() + if ev.AgentID != "test-agent" { + t.Errorf("AgentID: got %q, want test-agent", ev.AgentID) + } + if ev.RunID != req.RunID { + t.Errorf("RunID: got %q, want %q", ev.RunID, req.RunID) + } + if ev.SessionKey != req.SessionKey { + t.Errorf("SessionKey: got %q, want %q", ev.SessionKey, req.SessionKey) + } + if ev.Channel != req.Channel { + t.Errorf("Channel: got %q, want %q", ev.Channel, req.Channel) + } + if ev.UserID != req.UserID { + t.Errorf("UserID: got %q, want %q", ev.UserID, req.UserID) + } + payload, ok := ev.Payload.(map[string]any) + if !ok { + t.Fatalf("Payload is not map[string]any: %T", ev.Payload) + } + if payload["id"] != tc.ID { + t.Errorf("payload.id: got %v, want %q", payload["id"], tc.ID) + } + if payload["name"] != tc.Name { + t.Errorf("payload.name: got %v, want %q", payload["name"], tc.Name) + } +} diff --git a/internal/pipeline/stages_test.go b/internal/pipeline/stages_test.go index b778b1f1..7ab33a2a 100644 --- a/internal/pipeline/stages_test.go +++ b/internal/pipeline/stages_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "github.com/nextlevelbuilder/goclaw/internal/bootstrap" @@ -789,6 +790,66 @@ func TestToolStage_MultipleTools_Sequential_MessagesInOrder(t *testing.T) { } } +// Regression: v3 parallel path invokes ExecuteToolRaw + ProcessToolResult +// for every tool call. If this breaks, the `tool.call` WS event emitted +// inside makeExecuteToolRaw (loop_pipeline_tool_callbacks.go) stops firing +// and UIs go silent during real-time tool execution. +func TestToolStage_MultipleTools_ParallelPath_InvokesRawAndProcessForEach(t *testing.T) { + t.Parallel() + var rawMu, procMu sync.Mutex + rawCalls := []string{} + procCalls := []string{} + deps := &PipelineDeps{ + // Parallel path requires all three callbacks. ExecuteToolCall is required + // upfront (nil-guard) even though the parallel branch won't invoke it. + ExecuteToolCall: func(_ context.Context, _ *RunState, _ providers.ToolCall) ([]providers.Message, error) { + t.Fatal("ExecuteToolCall must NOT be called when parallel path is active") + return nil, nil + }, + ExecuteToolRaw: func(_ context.Context, tc providers.ToolCall) (providers.Message, any, error) { + rawMu.Lock() + rawCalls = append(rawCalls, tc.ID) + rawMu.Unlock() + return providers.Message{Role: "tool", Content: "raw:" + tc.Name, ToolCallID: tc.ID}, nil, nil + }, + ProcessToolResult: func(_ context.Context, _ *RunState, tc providers.ToolCall, rawMsg providers.Message, _ any) []providers.Message { + procMu.Lock() + procCalls = append(procCalls, tc.ID) + procMu.Unlock() + return []providers.Message{rawMsg} + }, + } + stage := NewToolStage(deps) + state := defaultState() + state.Think.LastResponse = &providers.ChatResponse{ + ToolCalls: []providers.ToolCall{ + {ID: "1", Name: "tool_a"}, + {ID: "2", Name: "tool_b"}, + {ID: "3", Name: "tool_c"}, + }, + } + + if err := stage.Execute(context.Background(), state); err != nil { + t.Fatalf("Execute() error: %v", err) + } + + // Each tool must be passed through BOTH ExecuteToolRaw (where tool.call emits) + // AND ProcessToolResult (where tool.result emits). Missing either breaks UIs. + if len(rawCalls) != 3 { + t.Errorf("ExecuteToolRaw called %d times, want 3 (one per tool)", len(rawCalls)) + } + if len(procCalls) != 3 { + t.Errorf("ProcessToolResult called %d times, want 3", len(procCalls)) + } + // ProcessToolResult must run sequentially in original order (deterministic state mutation) + if len(procCalls) == 3 && (procCalls[0] != "1" || procCalls[1] != "2" || procCalls[2] != "3") { + t.Errorf("ProcessToolResult order = %v, want [1 2 3]", procCalls) + } + if state.Tool.TotalToolCalls != 3 { + t.Errorf("TotalToolCalls = %d, want 3", state.Tool.TotalToolCalls) + } +} + func TestToolStage_LoopKilled_ReturnsBreakLoop(t *testing.T) { t.Parallel() deps := &PipelineDeps{