From 61cb340147210443ad081a4f22ff8b9becd38f72 Mon Sep 17 00:00:00 2001 From: Rahul Bansal Date: Sat, 21 Feb 2026 17:13:32 +0530 Subject: [PATCH] feat: add AgentEventListener for real-time tool/thinking notifications --- pkg/agent/events.go | 58 +++++++++++ pkg/agent/events_test.go | 216 +++++++++++++++++++++++++++++++++++++++ pkg/agent/loop.go | 44 ++++++++ 3 files changed, 318 insertions(+) create mode 100644 pkg/agent/events.go create mode 100644 pkg/agent/events_test.go diff --git a/pkg/agent/events.go b/pkg/agent/events.go new file mode 100644 index 000000000..bf4c04a3c --- /dev/null +++ b/pkg/agent/events.go @@ -0,0 +1,58 @@ +// PicoClaw - Ultra-lightweight personal AI agent +// License: MIT + +package agent + +// AgentEventType identifies the kind of agent lifecycle event +type AgentEventType int + +const ( + // EventThinkingStarted fires before calling the LLM + EventThinkingStarted AgentEventType = iota + // EventToolCallStarted fires before executing a tool + EventToolCallStarted + // EventToolCallCompleted fires after a tool execution finishes + EventToolCallCompleted + // EventResponseComplete fires when the agent produces a final text response + EventResponseComplete + // EventError fires when the agent loop encounters an error + EventError +) + +// AgentEvent represents a lifecycle event emitted by the agent loop +type AgentEvent struct { + Type AgentEventType + Data any +} + +// ToolCallStartedData carries information about a tool call that is about to execute +type ToolCallStartedData struct { + ID string + Name string + Args map[string]any +} + +// ToolCallCompletedData carries information about a completed tool call +type ToolCallCompletedData struct { + ID string + Name string + Result string + IsError bool +} + +// ResponseCompleteData carries the final response content from the agent +type ResponseCompleteData struct { + Content string +} + +// ErrorData carries error information from the agent loop +type ErrorData struct { + Err error +} + +// AgentEventListener receives lifecycle events from the agent loop. +// Implementations must be safe for concurrent use, as events fire from +// the agent loop goroutine while the caller may run on a different goroutine. +type AgentEventListener interface { + OnEvent(event AgentEvent) +} diff --git a/pkg/agent/events_test.go b/pkg/agent/events_test.go new file mode 100644 index 000000000..6a9c29662 --- /dev/null +++ b/pkg/agent/events_test.go @@ -0,0 +1,216 @@ +package agent + +import ( + "fmt" + "sync" + "testing" +) + +// mockEventListener records events for test assertions +type mockEventListener struct { + mu sync.Mutex + events []AgentEvent +} + +func (m *mockEventListener) OnEvent(event AgentEvent) { + m.mu.Lock() + defer m.mu.Unlock() + m.events = append(m.events, event) +} + +func (m *mockEventListener) getEvents() []AgentEvent { + m.mu.Lock() + defer m.mu.Unlock() + copied := make([]AgentEvent, len(m.events)) + copy(copied, m.events) + return copied +} + +func TestAgentEventTypes_AreDistinct(t *testing.T) { + types := []AgentEventType{ + EventThinkingStarted, + EventToolCallStarted, + EventToolCallCompleted, + EventResponseComplete, + EventError, + } + + seen := make(map[AgentEventType]bool) + for _, et := range types { + if seen[et] { + t.Errorf("duplicate event type value: %d", et) + } + seen[et] = true + } + + if len(seen) != 5 { + t.Errorf("expected 5 distinct event types, got %d", len(seen)) + } +} + +func TestMockEventListener_ReceivesEvents(t *testing.T) { + listener := &mockEventListener{} + + tests := []struct { + name string + event AgentEvent + checkData func(t *testing.T, data any) + }{ + { + name: "thinking started", + event: AgentEvent{ + Type: EventThinkingStarted, + Data: nil, + }, + checkData: func(t *testing.T, data any) { + if data != nil { + t.Errorf("expected nil data for ThinkingStarted, got %v", data) + } + }, + }, + { + name: "tool call started", + event: AgentEvent{ + Type: EventToolCallStarted, + Data: ToolCallStartedData{ + ID: "call_123", + Name: "exec", + Args: map[string]any{"command": "ls"}, + }, + }, + checkData: func(t *testing.T, data any) { + d, ok := data.(ToolCallStartedData) + if !ok { + t.Fatalf("expected ToolCallStartedData, got %T", data) + } + if d.ID != "call_123" { + t.Errorf("expected ID 'call_123', got %q", d.ID) + } + if d.Name != "exec" { + t.Errorf("expected Name 'exec', got %q", d.Name) + } + if d.Args["command"] != "ls" { + t.Errorf("expected Args[command]='ls', got %v", d.Args["command"]) + } + }, + }, + { + name: "tool call completed", + event: AgentEvent{ + Type: EventToolCallCompleted, + Data: ToolCallCompletedData{ + ID: "call_123", + Name: "exec", + Result: "file1.txt\nfile2.txt", + IsError: false, + }, + }, + checkData: func(t *testing.T, data any) { + d, ok := data.(ToolCallCompletedData) + if !ok { + t.Fatalf("expected ToolCallCompletedData, got %T", data) + } + if d.ID != "call_123" { + t.Errorf("expected ID 'call_123', got %q", d.ID) + } + if d.IsError { + t.Error("expected IsError=false") + } + if d.Result != "file1.txt\nfile2.txt" { + t.Errorf("unexpected Result: %q", d.Result) + } + }, + }, + { + name: "tool call completed with error", + event: AgentEvent{ + Type: EventToolCallCompleted, + Data: ToolCallCompletedData{ + ID: "call_456", + Name: "read_file", + Result: "file not found", + IsError: true, + }, + }, + checkData: func(t *testing.T, data any) { + d, ok := data.(ToolCallCompletedData) + if !ok { + t.Fatalf("expected ToolCallCompletedData, got %T", data) + } + if !d.IsError { + t.Error("expected IsError=true") + } + }, + }, + { + name: "response complete", + event: AgentEvent{ + Type: EventResponseComplete, + Data: ResponseCompleteData{ + Content: "Here is the answer.", + }, + }, + checkData: func(t *testing.T, data any) { + d, ok := data.(ResponseCompleteData) + if !ok { + t.Fatalf("expected ResponseCompleteData, got %T", data) + } + if d.Content != "Here is the answer." { + t.Errorf("expected content 'Here is the answer.', got %q", d.Content) + } + }, + }, + { + name: "error", + event: AgentEvent{ + Type: EventError, + Data: ErrorData{ + Err: fmt.Errorf("LLM call failed"), + }, + }, + checkData: func(t *testing.T, data any) { + d, ok := data.(ErrorData) + if !ok { + t.Fatalf("expected ErrorData, got %T", data) + } + if d.Err == nil { + t.Fatal("expected non-nil error") + } + if d.Err.Error() != "LLM call failed" { + t.Errorf("expected error 'LLM call failed', got %q", d.Err.Error()) + } + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + listener.OnEvent(tt.event) + }) + } + + // Verify all events were received in order + events := listener.getEvents() + if len(events) != len(tests) { + t.Fatalf("expected %d events, got %d", len(tests), len(events)) + } + + for i, tt := range tests { + t.Run(tt.name+"/verify", func(t *testing.T) { + if events[i].Type != tt.event.Type { + t.Errorf("event %d: expected type %d, got %d", i, tt.event.Type, events[i].Type) + } + tt.checkData(t, events[i].Data) + }) + } +} + +func TestAgentEventListener_NilSafe(t *testing.T) { + // Verify that fireEvent with nil listener doesn't panic + // This is tested indirectly through the AgentLoop, but we can + // verify the interface contract here + var listener AgentEventListener + if listener != nil { + t.Error("expected nil listener") + } +} diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 8b38636d5..619ed3a58 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -39,6 +39,7 @@ type AgentLoop struct { fallback *providers.FallbackChain channelManager *channels.Manager permFuncFactory tools.PermissionFuncFactory + eventListener AgentEventListener } // processOptions configures how a message is processed @@ -224,6 +225,19 @@ func (al *AgentLoop) SetPermissionFuncFactory(factory tools.PermissionFuncFactor al.permFuncFactory = factory } +// SetEventListener sets the listener that receives agent lifecycle events. +// The listener must be safe for concurrent use. +func (al *AgentLoop) SetEventListener(listener AgentEventListener) { + al.eventListener = listener +} + +// fireEvent dispatches an event to the listener if one is set +func (al *AgentLoop) fireEvent(event AgentEvent) { + if al.eventListener != nil { + al.eventListener.OnEvent(event) + } +} + // RecordLastChannel records the last active channel for this workspace. // This uses the atomic state save mechanism to prevent data loss on crash. func (al *AgentLoop) RecordLastChannel(channel string) error { @@ -436,6 +450,10 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt // 4. Run LLM iteration loop finalContent, iteration, err := al.runLLMIteration(ctx, agent, messages, opts) if err != nil { + al.fireEvent(AgentEvent{ + Type: EventError, + Data: ErrorData{Err: err}, + }) return "", err } @@ -552,6 +570,9 @@ func (al *AgentLoop) runLLMIteration( }) } + // Notify listener that we're about to call the LLM + al.fireEvent(AgentEvent{Type: EventThinkingStarted}) + // Retry loop for context/token errors maxRetries := 2 for retry := 0; retry <= maxRetries; retry++ { @@ -611,6 +632,10 @@ func (al *AgentLoop) runLLMIteration( "iteration": iteration, "content_chars": len(finalContent), }) + al.fireEvent(AgentEvent{ + Type: EventResponseComplete, + Data: ResponseCompleteData{Content: finalContent}, + }) break } @@ -692,6 +717,15 @@ func (al *AgentLoop) runLLMIteration( } } + al.fireEvent(AgentEvent{ + Type: EventToolCallStarted, + Data: ToolCallStartedData{ + ID: tc.ID, + Name: tc.Name, + Args: tc.Arguments, + }, + }) + toolResult := agent.Tools.ExecuteWithContext( ctx, tc.Name, @@ -701,6 +735,16 @@ func (al *AgentLoop) runLLMIteration( asyncCallback, ) + al.fireEvent(AgentEvent{ + Type: EventToolCallCompleted, + Data: ToolCallCompletedData{ + ID: tc.ID, + Name: tc.Name, + Result: toolResult.ForLLM, + IsError: toolResult.Err != nil, + }, + }) + // Send ForUser content to user immediately if not Silent if !toolResult.Silent && toolResult.ForUser != "" && opts.SendResponse { al.bus.PublishOutbound(bus.OutboundMessage{