mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 17:38:10 +00:00
3e3c6a24d2
- SDK PluginAPI (internal/plugin/sdk/): RegisterTool/RegisterStage/Subscribe/Publish - EventBus (internal/events/): system-level pub/sub with wildcard support - StageHost (internal/agent/core/stages.go): 7-stage message pipeline - Agent core: on_input/pre_action/post_action/before_toolcall/after_toolcall/before_output/after_output - Plugin Registry: SDK plugin registration and tool routing - GraphDB.MergeEntities: entity consolidation with relation redirection - memory_merge tool: allows LLM to merge similar entities - Consolidation task: heartbeat detects conflicts, enqueues via IO for LLM decision - _consolidation_ internal channel for system-level memory maintenance - Comprehensive documentation: ARCHITECTURE.md, PLAN.md, DESIGN.md, README.md - 54 tests across all packages, all passing
103 lines
1.9 KiB
Go
103 lines
1.9 KiB
Go
package events
|
|
|
|
import (
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestBusPublishSubscribe(t *testing.T) {
|
|
bus := NewBus()
|
|
var count int32
|
|
|
|
bus.Subscribe(EventRawInput, func(evt *Event) {
|
|
atomic.AddInt32(&count, 1)
|
|
})
|
|
|
|
bus.Publish(&Event{
|
|
Type: EventRawInput,
|
|
Source: "test",
|
|
Payload: map[string]interface{}{"content": "hello"},
|
|
})
|
|
|
|
if c := atomic.LoadInt32(&count); c != 1 {
|
|
t.Errorf("expected 1, got %d", c)
|
|
}
|
|
}
|
|
|
|
func TestBusWildcard(t *testing.T) {
|
|
bus := NewBus()
|
|
var count int32
|
|
|
|
bus.Subscribe(EventAll, func(evt *Event) {
|
|
atomic.AddInt32(&count, 1)
|
|
})
|
|
|
|
bus.Publish(&Event{Type: EventRawInput, Source: "test"})
|
|
bus.Publish(&Event{Type: EventToolCall, Source: "test"})
|
|
|
|
if c := atomic.LoadInt32(&count); c != 2 {
|
|
t.Errorf("expected 2, got %d", c)
|
|
}
|
|
}
|
|
|
|
func TestBusUnsubscribe(t *testing.T) {
|
|
bus := NewBus()
|
|
var count int32
|
|
|
|
handler := func(evt *Event) {
|
|
atomic.AddInt32(&count, 1)
|
|
}
|
|
unsub := bus.Subscribe(EventRawInput, handler)
|
|
|
|
bus.Publish(&Event{Type: EventRawInput, Source: "test"})
|
|
unsub()
|
|
bus.Publish(&Event{Type: EventRawInput, Source: "test"})
|
|
|
|
if c := atomic.LoadInt32(&count); c != 1 {
|
|
t.Errorf("expected 1 after unsub, got %d", c)
|
|
}
|
|
}
|
|
|
|
func TestBusNoMatch(t *testing.T) {
|
|
bus := NewBus()
|
|
var count int32
|
|
|
|
bus.Subscribe(EventRawInput, func(evt *Event) {
|
|
atomic.AddInt32(&count, 1)
|
|
})
|
|
|
|
bus.Publish(&Event{Type: EventAgentOutput, Source: "test"})
|
|
|
|
if c := atomic.LoadInt32(&count); c != 0 {
|
|
t.Errorf("expected 0, got %d", c)
|
|
}
|
|
}
|
|
|
|
func TestBusConcurrent(t *testing.T) {
|
|
bus := NewBus()
|
|
var count int32
|
|
|
|
bus.Subscribe(EventAll, func(evt *Event) {
|
|
atomic.AddInt32(&count, 1)
|
|
})
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
for i := 0; i < 100; i++ {
|
|
bus.Publish(&Event{Type: EventRawInput, Source: "test"})
|
|
}
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("timeout")
|
|
}
|
|
|
|
if c := atomic.LoadInt32(&count); c != 100 {
|
|
t.Errorf("expected 100, got %d", c)
|
|
}
|
|
}
|