mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 14:17:40 +00:00
fix(agent): emit tool.call event in parallel tool execution path
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.
This commit is contained in:
1 parent
619b253b82
commit
221cd78fcf
3 files changed
+262
No files matched your search
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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{
|
||||
|
||||
Reference in new issue
Block a user