mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-20 00:15:16 +00:00
fix(chat): support zero debounce and agent overrides
This commit is contained in:
1 parent
8af0e4ae09
commit
2c699f31de
29 files changed
+498
-54
No files matched your search
@@ -82,19 +82,17 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
|
||||
// Inbound debounce: merge rapid messages from the same sender before processing.
|
||||
// Matching TS createInboundDebouncer from src/auto-reply/inbound-debounce.ts.
|
||||
debounceMs := cfg.Gateway.InboundDebounceMs
|
||||
if debounceMs == 0 {
|
||||
debounceMs = 1000 // default: 1000ms
|
||||
}
|
||||
debouncer := bus.NewInboundDebouncer(
|
||||
time.Duration(debounceMs)*time.Millisecond,
|
||||
debouncer := bus.NewInboundDebouncerFunc(
|
||||
func(msg bus.InboundMessage) time.Duration {
|
||||
return resolveInboundDebounceDelay(ctx, msg, deps)
|
||||
},
|
||||
func(msg bus.InboundMessage) {
|
||||
processNormalMessage(ctx, msg, deps)
|
||||
},
|
||||
)
|
||||
defer debouncer.Stop()
|
||||
|
||||
slog.Info("inbound debounce configured", "debounce_ms", debounceMs)
|
||||
slog.Info("inbound debounce configured", "global_debounce_ms", cfg.Gateway.InboundDebounceMs, "agent_override", true)
|
||||
|
||||
// Track background goroutines (subagent announces, teammate messages)
|
||||
// so shutdown can wait for in-flight work to complete.
|
||||
@@ -139,6 +137,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
}
|
||||
|
||||
// --- Normal messages: route through debouncer ---
|
||||
prepareInboundDebounceMessage(&msg, deps)
|
||||
debouncer.Push(msg)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
func prepareInboundDebounceMessage(msg *bus.InboundMessage, deps *ConsumerDeps) {
|
||||
if msg == nil || deps == nil || deps.Cfg == nil || msg.AgentID != "" {
|
||||
return
|
||||
}
|
||||
msg.AgentID = resolveAgentRoute(deps.Cfg, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
}
|
||||
|
||||
func resolveInboundDebounceDelay(ctx context.Context, msg bus.InboundMessage, deps *ConsumerDeps) time.Duration {
|
||||
debounceMs := 0
|
||||
if deps != nil && deps.Cfg != nil {
|
||||
debounceMs = deps.Cfg.Gateway.InboundDebounceMs
|
||||
}
|
||||
if deps == nil || deps.AgentStore == nil || msg.AgentID == "" {
|
||||
return inboundDebounceDuration(debounceMs)
|
||||
}
|
||||
|
||||
agentCtx := ctx
|
||||
if msg.TenantID != uuid.Nil {
|
||||
agentCtx = store.WithTenantID(agentCtx, msg.TenantID)
|
||||
} else {
|
||||
agentCtx = store.WithTenantID(agentCtx, store.MasterTenantID)
|
||||
}
|
||||
|
||||
agentData, err := getInboundDebounceAgent(agentCtx, deps.AgentStore, msg.AgentID)
|
||||
if err != nil || agentData == nil {
|
||||
if err != nil {
|
||||
slog.Debug("inbound debounce: agent config unavailable", "agent", msg.AgentID, "error", err)
|
||||
}
|
||||
return inboundDebounceDuration(debounceMs)
|
||||
}
|
||||
if overrideMs, ok := agentData.ParseInboundDebounceMs(); ok {
|
||||
debounceMs = overrideMs
|
||||
}
|
||||
return inboundDebounceDuration(debounceMs)
|
||||
}
|
||||
|
||||
func getInboundDebounceAgent(ctx context.Context, agentStore store.AgentStore, agentID string) (*store.AgentData, error) {
|
||||
if parsed, err := uuid.Parse(agentID); err == nil && parsed != uuid.Nil {
|
||||
return agentStore.GetByID(ctx, parsed)
|
||||
}
|
||||
return agentStore.GetByKey(ctx, agentID)
|
||||
}
|
||||
|
||||
func inboundDebounceDuration(ms int) time.Duration {
|
||||
if ms <= 0 {
|
||||
return 0
|
||||
}
|
||||
return time.Duration(ms) * time.Millisecond
|
||||
}
|
||||
@@ -78,6 +78,9 @@ func seedConfigForContext(ctx context.Context, sc store.SystemConfigStore, cfg *
|
||||
set(key, fmt.Sprintf("%d", val))
|
||||
}
|
||||
}
|
||||
setIntAllowZero := func(key string, val int) {
|
||||
set(key, fmt.Sprintf("%d", val))
|
||||
}
|
||||
setBool := func(key string, val *bool) {
|
||||
if val != nil {
|
||||
set(key, fmt.Sprintf("%t", *val))
|
||||
@@ -102,7 +105,7 @@ func seedConfigForContext(ctx context.Context, sc store.SystemConfigStore, cfg *
|
||||
setInt("gateway.rate_limit_rpm", cfg.Gateway.RateLimitRPM)
|
||||
setInt("gateway.max_message_chars", cfg.Gateway.MaxMessageChars)
|
||||
set("gateway.injection_action", cfg.Gateway.InjectionAction)
|
||||
setInt("gateway.inbound_debounce_ms", cfg.Gateway.InboundDebounceMs)
|
||||
setIntAllowZero("gateway.inbound_debounce_ms", cfg.Gateway.InboundDebounceMs)
|
||||
setBool("gateway.block_reply", cfg.Gateway.BlockReply)
|
||||
setBool("gateway.tool_status", cfg.Gateway.ToolStatus)
|
||||
setInt("gateway.task_recovery_interval_sec", cfg.Gateway.TaskRecoveryIntervalSec)
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"maps"
|
||||
"testing"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
type captureSystemConfigStore struct {
|
||||
data map[string]string
|
||||
}
|
||||
|
||||
func (s *captureSystemConfigStore) Get(_ context.Context, key string) (string, error) {
|
||||
return s.data[key], nil
|
||||
}
|
||||
|
||||
func (s *captureSystemConfigStore) Set(_ context.Context, key, value string) error {
|
||||
s.data[key] = value
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *captureSystemConfigStore) Delete(_ context.Context, key string) error {
|
||||
delete(s.data, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *captureSystemConfigStore) List(_ context.Context) (map[string]string, error) {
|
||||
out := make(map[string]string, len(s.data))
|
||||
maps.Copy(out, s.data)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func TestSeedConfigForContextPersistsZeroInboundDebounce(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
sc := &captureSystemConfigStore{data: map[string]string{}}
|
||||
cfg := config.Default()
|
||||
cfg.Gateway.InboundDebounceMs = 0
|
||||
|
||||
seedConfigForContext(store.WithTenantID(context.Background(), store.MasterTenantID), sc, cfg, false)
|
||||
|
||||
if got := sc.data["gateway.inbound_debounce_ms"]; got != "0" {
|
||||
t.Fatalf("gateway.inbound_debounce_ms = %q, want 0", got)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user