feat: add AgentEventListener for real-time tool/thinking notifications
This commit is contained in:
parent
b644bd1a24
commit
61cb340147
3 changed files with 318 additions and 0 deletions
58
pkg/agent/events.go
Normal file
58
pkg/agent/events.go
Normal file
|
|
@ -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)
|
||||||
|
}
|
||||||
216
pkg/agent/events_test.go
Normal file
216
pkg/agent/events_test.go
Normal file
|
|
@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -39,6 +39,7 @@ type AgentLoop struct {
|
||||||
fallback *providers.FallbackChain
|
fallback *providers.FallbackChain
|
||||||
channelManager *channels.Manager
|
channelManager *channels.Manager
|
||||||
permFuncFactory tools.PermissionFuncFactory
|
permFuncFactory tools.PermissionFuncFactory
|
||||||
|
eventListener AgentEventListener
|
||||||
}
|
}
|
||||||
|
|
||||||
// processOptions configures how a message is processed
|
// processOptions configures how a message is processed
|
||||||
|
|
@ -224,6 +225,19 @@ func (al *AgentLoop) SetPermissionFuncFactory(factory tools.PermissionFuncFactor
|
||||||
al.permFuncFactory = factory
|
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.
|
// RecordLastChannel records the last active channel for this workspace.
|
||||||
// This uses the atomic state save mechanism to prevent data loss on crash.
|
// This uses the atomic state save mechanism to prevent data loss on crash.
|
||||||
func (al *AgentLoop) RecordLastChannel(channel string) error {
|
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
|
// 4. Run LLM iteration loop
|
||||||
finalContent, iteration, err := al.runLLMIteration(ctx, agent, messages, opts)
|
finalContent, iteration, err := al.runLLMIteration(ctx, agent, messages, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
al.fireEvent(AgentEvent{
|
||||||
|
Type: EventError,
|
||||||
|
Data: ErrorData{Err: err},
|
||||||
|
})
|
||||||
return "", 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
|
// Retry loop for context/token errors
|
||||||
maxRetries := 2
|
maxRetries := 2
|
||||||
for retry := 0; retry <= maxRetries; retry++ {
|
for retry := 0; retry <= maxRetries; retry++ {
|
||||||
|
|
@ -611,6 +632,10 @@ func (al *AgentLoop) runLLMIteration(
|
||||||
"iteration": iteration,
|
"iteration": iteration,
|
||||||
"content_chars": len(finalContent),
|
"content_chars": len(finalContent),
|
||||||
})
|
})
|
||||||
|
al.fireEvent(AgentEvent{
|
||||||
|
Type: EventResponseComplete,
|
||||||
|
Data: ResponseCompleteData{Content: finalContent},
|
||||||
|
})
|
||||||
break
|
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(
|
toolResult := agent.Tools.ExecuteWithContext(
|
||||||
ctx,
|
ctx,
|
||||||
tc.Name,
|
tc.Name,
|
||||||
|
|
@ -701,6 +735,16 @@ func (al *AgentLoop) runLLMIteration(
|
||||||
asyncCallback,
|
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
|
// Send ForUser content to user immediately if not Silent
|
||||||
if !toolResult.Silent && toolResult.ForUser != "" && opts.SendResponse {
|
if !toolResult.Silent && toolResult.ForUser != "" && opts.SendResponse {
|
||||||
al.bus.PublishOutbound(bus.OutboundMessage{
|
al.bus.PublishOutbound(bus.OutboundMessage{
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue