package agent import ( "context" "sync" "testing" "time" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/eventbus" ) // recordingBus captures every published DomainEvent for assertion. // Distinct from the eventbus package's own test bus — this one does no // dispatch, it only records what the emit callback published. type recordingBus struct { mu sync.Mutex published []eventbus.DomainEvent } func (r *recordingBus) Publish(event eventbus.DomainEvent) { r.mu.Lock() defer r.mu.Unlock() r.published = append(r.published, event) } func (r *recordingBus) Subscribe(_ eventbus.EventType, _ eventbus.DomainEventHandler) func() { return func() {} } func (r *recordingBus) Start(_ context.Context) {} func (r *recordingBus) Drain(_ time.Duration) error { return nil } func (r *recordingBus) events() []eventbus.DomainEvent { r.mu.Lock() defer r.mu.Unlock() out := make([]eventbus.DomainEvent, len(r.published)) copy(out, r.published) return out } // --------------------------------------------------------------------------- // V1-A: emit reads the CUMULATIVE session compaction count, not the per-run 0. // --------------------------------------------------------------------------- func TestEmitSessionCompleted_UsesCumulativeCount(t *testing.T) { bus := &recordingBus{} sessions := &nopSessionStore{ compactionCount: 5, // cumulative session count summary: "prior-cycle summary", } loop := &Loop{ id: "test-agent", agentUUID: uuid.New(), tenantID: uuid.New(), domainBus: bus, sessions: sessions, } loop.emitSessionCompleted(context.Background(), "sess-1", "user-1", 42, 9000) evs := bus.events() if len(evs) != 1 { t.Fatalf("published %d events, want 1", len(evs)) } ev := evs[0] // Bug A: payload.CompactionCount (the field the worker uses for idempotency) // must be the cumulative count (5), not the per-run 0. pl, ok := ev.Payload.(*eventbus.SessionCompletedPayload) if !ok { t.Fatalf("payload type = %T, want *SessionCompletedPayload", ev.Payload) } if pl.CompactionCount != 5 { t.Errorf("payload.CompactionCount = %d, want 5 (cumulative)", pl.CompactionCount) } // Bug C: SourceID embeds the cumulative count. if ev.SourceID != "sess-1:5" { t.Errorf("SourceID = %q, want %q", ev.SourceID, "sess-1:5") } // Summary from the previous cycle is attached when count > 0. if pl.Summary != "prior-cycle summary" { t.Errorf("payload.Summary = %q, want prior-cycle summary", pl.Summary) } if pl.MessageCount != 42 || pl.TokensUsed != 9000 { t.Errorf("msgCount/tokens = %d/%d, want 42/9000", pl.MessageCount, pl.TokensUsed) } } func TestEmitSessionCompleted_ZeroCountOmitsSummary(t *testing.T) { bus := &recordingBus{} // count == 0 → no prior cycle → summary must NOT be fetched/attached // (worker would otherwise get a stale/empty summary and skip the LLM path). sessions := &nopSessionStore{ compactionCount: 0, summary: "should-not-be-attached", } loop := &Loop{ id: "test-agent", agentUUID: uuid.New(), tenantID: uuid.New(), domainBus: bus, sessions: sessions, } loop.emitSessionCompleted(context.Background(), "sess-1", "user-1", 1, 100) evs := bus.events() if len(evs) != 1 { t.Fatalf("published %d events, want 1", len(evs)) } pl := evs[0].Payload.(*eventbus.SessionCompletedPayload) if pl.CompactionCount != 0 { t.Errorf("payload.CompactionCount = %d, want 0", pl.CompactionCount) } if pl.Summary != "" { t.Errorf("payload.Summary = %q, want empty (count==0)", pl.Summary) } if evs[0].SourceID != "sess-1:0" { t.Errorf("SourceID = %q, want sess-1:0", evs[0].SourceID) } } func TestEmitSessionCompleted_NoBusIsNoop(t *testing.T) { loop := &Loop{ id: "test-agent", agentUUID: uuid.New(), tenantID: uuid.New(), domainBus: nil, // no bus sessions: &nopSessionStore{compactionCount: 3}, } // Must not panic when domainBus is nil. loop.emitSessionCompleted(context.Background(), "sess-1", "user-1", 1, 100) } // --------------------------------------------------------------------------- // V1-C: eventbus dedup — distinct counts both pass, same count deduped. // Uses the REAL bus + real dedupSet (not the recording double) so we exercise // the actual dedup key = Type + ":" + SourceID. // --------------------------------------------------------------------------- func TestEventbusDedup_DistinctCompactionCountsPass(t *testing.T) { bus := eventbus.NewDomainEventBus(eventbus.Config{ QueueSize: 100, WorkerCount: 2, RetryAttempts: 1, RetryDelay: time.Millisecond, DedupTTL: time.Minute, }) bus.Start(context.Background()) defer func() { _ = bus.Drain(time.Second) }() var mu sync.Mutex var seen []string bus.Subscribe(eventbus.EventSessionCompleted, func(_ context.Context, e eventbus.DomainEvent) error { mu.Lock() seen = append(seen, e.SourceID) mu.Unlock() return nil }) // Same session, three DIFFERENT compaction cycles → all three must pass dedup. bus.Publish(eventbus.DomainEvent{Type: eventbus.EventSessionCompleted, SourceID: "sess-1:1"}) bus.Publish(eventbus.DomainEvent{Type: eventbus.EventSessionCompleted, SourceID: "sess-1:2"}) bus.Publish(eventbus.DomainEvent{Type: eventbus.EventSessionCompleted, SourceID: "sess-1:3"}) // Rapid duplicate of cycle 2 within TTL → must be deduped. bus.Publish(eventbus.DomainEvent{Type: eventbus.EventSessionCompleted, SourceID: "sess-1:2"}) time.Sleep(80 * time.Millisecond) mu.Lock() got := len(seen) mu.Unlock() if got != 3 { t.Errorf("handler saw %d events, want 3 (cycles 1,2,3; duplicate of 2 deduped)", got) } } func TestEventbusDedup_SameCompactionCountDeduped(t *testing.T) { bus := eventbus.NewDomainEventBus(eventbus.Config{ QueueSize: 100, WorkerCount: 2, RetryAttempts: 1, RetryDelay: time.Millisecond, DedupTTL: time.Minute, }) bus.Start(context.Background()) defer func() { _ = bus.Drain(time.Second) }() var count int var mu sync.Mutex bus.Subscribe(eventbus.EventSessionCompleted, func(_ context.Context, _ eventbus.DomainEvent) error { mu.Lock() count++ mu.Unlock() return nil }) // Same SourceID (same session, same cycle) published 3× → only 1 passes. for range 3 { bus.Publish(eventbus.DomainEvent{Type: eventbus.EventSessionCompleted, SourceID: "sess-1:7"}) } time.Sleep(80 * time.Millisecond) mu.Lock() got := count mu.Unlock() if got != 1 { t.Errorf("handler called %d times, want 1 (same cycle deduped)", got) } }