mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-11 03:13:24 +00:00
fix(discord): preserve channel agent routing (#1380)
Co-authored-by: Collective Developer <man@collective.dev>
This commit is contained in:
1 parent
f826738ee6
commit
2a082f4edf
8 files changed
+233
-45
No files matched your search
@@ -147,7 +147,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
}
|
||||
|
||||
// --- Normal messages: route through debouncer ---
|
||||
prepareInboundDebounceMessage(&msg, deps)
|
||||
prepareInboundDebounceMessage(ctx, &msg, deps)
|
||||
debouncer.Push(msg)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,11 +18,12 @@ import (
|
||||
// debouncing (issue #63). See plans/260528-1351-multi-attachment-debounce/.
|
||||
const mediaDebounceFloorMs = 1000
|
||||
|
||||
func prepareInboundDebounceMessage(msg *bus.InboundMessage, deps *ConsumerDeps) {
|
||||
func prepareInboundDebounceMessage(ctx context.Context, 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)
|
||||
routeCtx := inboundMessageTenantContext(ctx, *msg)
|
||||
msg.AgentID = resolveAgentRouteForInbound(routeCtx, deps.Cfg, deps.AgentStore, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
}
|
||||
|
||||
func resolveInboundDebounceDelay(ctx context.Context, msg bus.InboundMessage, deps *ConsumerDeps) time.Duration {
|
||||
|
||||
@@ -401,7 +401,8 @@ func handleResetCommand(
|
||||
|
||||
agentID := msg.AgentID
|
||||
if agentID == "" {
|
||||
agentID = resolveAgentRoute(deps.Cfg, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
ctx := inboundMessageTenantContext(context.Background(), msg)
|
||||
agentID = resolveAgentRouteForInbound(ctx, deps.Cfg, deps.AgentStore, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
}
|
||||
peerKind := msg.PeerKind
|
||||
if peerKind == "" {
|
||||
@@ -437,7 +438,8 @@ func handleStopCommand(
|
||||
|
||||
agentID := msg.AgentID
|
||||
if agentID == "" {
|
||||
agentID = resolveAgentRoute(deps.Cfg, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
ctx := inboundMessageTenantContext(context.Background(), msg)
|
||||
agentID = resolveAgentRouteForInbound(ctx, deps.Cfg, deps.AgentStore, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
}
|
||||
peerKind := msg.PeerKind
|
||||
if peerKind == "" {
|
||||
|
||||
@@ -1,16 +1,20 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"mime"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/agent"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/sessions"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
)
|
||||
|
||||
@@ -19,26 +23,60 @@ import (
|
||||
// Matching TS resolve-route.ts binding resolution.
|
||||
func resolveAgentRoute(cfg *config.Config, channel, chatID, peerKind string) string {
|
||||
for _, binding := range cfg.Bindings {
|
||||
match := binding.Match
|
||||
if match.Channel != channel {
|
||||
continue
|
||||
if bindingMatchesInbound(binding, channel, chatID, peerKind) {
|
||||
return config.NormalizeAgentID(binding.AgentID)
|
||||
}
|
||||
|
||||
// Peer-level match (most specific)
|
||||
if match.Peer != nil {
|
||||
if match.Peer.Kind == peerKind && match.Peer.ID == chatID {
|
||||
return config.NormalizeAgentID(binding.AgentID)
|
||||
}
|
||||
continue // has peer constraint but doesn't match — skip
|
||||
}
|
||||
|
||||
// Channel-level match (least specific, no peer constraint)
|
||||
return config.NormalizeAgentID(binding.AgentID)
|
||||
}
|
||||
|
||||
return cfg.ResolveDefaultAgentID()
|
||||
}
|
||||
|
||||
type defaultAgentGetter interface {
|
||||
GetDefault(ctx context.Context) (*store.AgentData, error)
|
||||
}
|
||||
|
||||
func resolveAgentRouteForInbound(ctx context.Context, cfg *config.Config, agentStore defaultAgentGetter, channel, chatID, peerKind string) string {
|
||||
if cfg == nil {
|
||||
return config.DefaultAgentID
|
||||
}
|
||||
for _, binding := range cfg.Bindings {
|
||||
if bindingMatchesInbound(binding, channel, chatID, peerKind) {
|
||||
return config.NormalizeAgentID(binding.AgentID)
|
||||
}
|
||||
}
|
||||
if agentStore != nil {
|
||||
if ag, err := agentStore.GetDefault(ctx); err == nil && ag != nil && ag.AgentKey != "" {
|
||||
return ag.AgentKey
|
||||
}
|
||||
}
|
||||
return cfg.ResolveDefaultAgentID()
|
||||
}
|
||||
|
||||
func bindingMatchesInbound(binding config.AgentBinding, channel, chatID, peerKind string) bool {
|
||||
match := binding.Match
|
||||
if match.Channel != channel {
|
||||
return false
|
||||
}
|
||||
|
||||
// Peer-level match (most specific)
|
||||
if match.Peer != nil {
|
||||
return match.Peer.Kind == peerKind && match.Peer.ID == chatID
|
||||
}
|
||||
|
||||
// Channel-level match (least specific, no peer constraint)
|
||||
return true
|
||||
}
|
||||
|
||||
func inboundMessageTenantContext(ctx context.Context, msg bus.InboundMessage) context.Context {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if msg.TenantID != uuid.Nil {
|
||||
return store.WithTenantID(ctx, msg.TenantID)
|
||||
}
|
||||
return store.WithTenantID(ctx, store.MasterTenantID)
|
||||
}
|
||||
|
||||
// overrideSessionKeyFromLocalKey extracts topic/thread ID from the composite
|
||||
// local_key and returns the correct session key for forum topics or DM threads.
|
||||
// If localKey is empty or has no suffix, the original sessionKey is returned unchanged.
|
||||
|
||||
@@ -40,12 +40,25 @@ func processNormalMessage(
|
||||
// Determine target agent via bindings or explicit AgentID
|
||||
agentID := msg.AgentID
|
||||
if agentID == "" {
|
||||
agentID = resolveAgentRoute(deps.Cfg, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
agentID = resolveAgentRouteForInbound(ctx, deps.Cfg, deps.AgentStore, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
}
|
||||
|
||||
agentLoop, err := deps.Agents.Get(ctx, agentID)
|
||||
if err != nil {
|
||||
slog.Warn("inbound: agent not found", "agent", agentID, "channel", msg.Channel)
|
||||
slog.Warn("inbound: agent not found", "agent", agentID, "channel", msg.Channel, "error", err)
|
||||
errContent := formatAgentError(err)
|
||||
if deps.ChannelMgr != nil {
|
||||
if ct := deps.ChannelMgr.ChannelTypeForName(msg.Channel); isExternalChannel(ct) {
|
||||
errContent = ""
|
||||
}
|
||||
}
|
||||
deps.MsgBus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: msg.Channel,
|
||||
ChatID: msg.ChatID,
|
||||
Content: errContent,
|
||||
Metadata: msg.Metadata,
|
||||
TenantID: msg.TenantID,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -1,12 +1,18 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/agent"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/bitrix24"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/sessions"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
// TestIsSafeBitrixEntityToken pins the validation contract for webhook-sourced
|
||||
@@ -151,3 +157,112 @@ func TestResolveSenderNameTruncatesLongMetadata(t *testing.T) {
|
||||
t.Fatalf("resolveSenderName() length = %d, want 100", len([]rune(got)))
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveAgentRouteForInbound_FallsBackToDBDefaultAgent(t *testing.T) {
|
||||
cfg := &config.Config{}
|
||||
got := resolveAgentRouteForInbound(context.Background(), cfg, defaultAgentGetterStub{
|
||||
agent: &store.AgentData{AgentKey: "co-assistant"},
|
||||
}, "co-assistant-2-0", "channel-1", string(sessions.PeerDirect))
|
||||
if got != "co-assistant" {
|
||||
t.Fatalf("resolveAgentRouteForInbound() = %q, want DB default agent key", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveAgentRouteForInbound_BindingWinsOverDBDefault(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Bindings: []config.AgentBinding{{
|
||||
AgentID: "bound-agent",
|
||||
Match: config.BindingMatch{
|
||||
Channel: "co-assistant-2-0",
|
||||
},
|
||||
}},
|
||||
}
|
||||
got := resolveAgentRouteForInbound(context.Background(), cfg, defaultAgentGetterStub{
|
||||
agent: &store.AgentData{AgentKey: "co-assistant"},
|
||||
}, "co-assistant-2-0", "channel-1", string(sessions.PeerDirect))
|
||||
if got != "bound-agent" {
|
||||
t.Fatalf("resolveAgentRouteForInbound() = %q, want binding agent", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessNormalMessage_AgentLookupFailurePublishesExternalCleanup(t *testing.T) {
|
||||
msgBus := bus.New()
|
||||
channelMgr := channels.NewManager(msgBus)
|
||||
channelMgr.RegisterChannel("discord-prod", consumerTestChannel{
|
||||
name: "discord-prod",
|
||||
channelType: channels.TypeDiscord,
|
||||
running: true,
|
||||
})
|
||||
|
||||
metadata := map[string]string{
|
||||
"message_id": "discord-message-1",
|
||||
"placeholder_key": "discord-message-1",
|
||||
}
|
||||
|
||||
processNormalMessage(context.Background(), bus.InboundMessage{
|
||||
Channel: "discord-prod",
|
||||
SenderID: "user-1",
|
||||
ChatID: "channel-1",
|
||||
Content: "hello",
|
||||
PeerKind: string(sessions.PeerDirect),
|
||||
AgentID: "missing-agent",
|
||||
Metadata: metadata,
|
||||
}, &ConsumerDeps{
|
||||
Cfg: &config.Config{},
|
||||
Agents: agent.NewRouter(),
|
||||
ChannelMgr: channelMgr,
|
||||
MsgBus: msgBus,
|
||||
})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
got, ok := msgBus.SubscribeOutbound(ctx)
|
||||
if !ok {
|
||||
t.Fatal("expected outbound cleanup message")
|
||||
}
|
||||
if got.Content != "" {
|
||||
t.Fatalf("outbound content = %q, want empty cleanup for external Discord channel", got.Content)
|
||||
}
|
||||
if got.Channel != "discord-prod" || got.ChatID != "channel-1" {
|
||||
t.Fatalf("outbound route = %s/%s, want discord-prod/channel-1", got.Channel, got.ChatID)
|
||||
}
|
||||
if got.Metadata["placeholder_key"] != "discord-message-1" {
|
||||
t.Fatalf("placeholder_key = %q, want metadata preserved", got.Metadata["placeholder_key"])
|
||||
}
|
||||
}
|
||||
|
||||
type consumerTestChannel struct {
|
||||
name string
|
||||
channelType string
|
||||
running bool
|
||||
}
|
||||
|
||||
func (c consumerTestChannel) Name() string { return c.name }
|
||||
func (c consumerTestChannel) Type() string { return c.channelType }
|
||||
func (c consumerTestChannel) Start(context.Context) error {
|
||||
return nil
|
||||
}
|
||||
func (c consumerTestChannel) Stop(context.Context) error {
|
||||
return nil
|
||||
}
|
||||
func (c consumerTestChannel) Send(context.Context, bus.OutboundMessage) error {
|
||||
return nil
|
||||
}
|
||||
func (c consumerTestChannel) IsRunning() bool {
|
||||
return c.running
|
||||
}
|
||||
func (c consumerTestChannel) IsAllowed(string) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
type defaultAgentGetterStub struct {
|
||||
agent *store.AgentData
|
||||
err error
|
||||
}
|
||||
|
||||
func (s defaultAgentGetterStub) GetDefault(context.Context) (*store.AgentData, error) {
|
||||
if s.err != nil {
|
||||
return nil, s.err
|
||||
}
|
||||
return s.agent, nil
|
||||
}
|
||||
@@ -304,30 +304,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
|
||||
}
|
||||
}
|
||||
|
||||
// PATCHED: Clear AgentID so the gateway consumer's resolveAgentRoute
|
||||
// (cmd/gateway_consumer_normal.go:40-43) matches via cfg.Bindings.
|
||||
// The consumer checks: if msg.AgentID == "" → resolveAgentRoute(cfg, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
// Bindings in config.json match by channel name + peer.kind + peer.id.
|
||||
// If no binding matches, resolveAgentRoute falls back to cfg.ResolveDefaultAgentID().
|
||||
targetAgentID := ""
|
||||
slog.Info("discord: binding routing enabled",
|
||||
"channel_id", channelID,
|
||||
"channel_name", c.Name(),
|
||||
"peer_kind", peerKind,
|
||||
)
|
||||
|
||||
// Voice agent routing
|
||||
if c.config.VoiceAgentID != "" {
|
||||
for _, mi := range mediaList {
|
||||
if mi.Type == media.TypeAudio || mi.Type == media.TypeVoice {
|
||||
targetAgentID = c.config.VoiceAgentID
|
||||
slog.Debug("discord: routing voice inbound to speaking agent",
|
||||
"agent_id", targetAgentID, "media_type", mi.Type,
|
||||
)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
targetAgentID := c.targetAgentID(mediaList)
|
||||
|
||||
// Collect contact for processed messages (DM + group-mentioned).
|
||||
if cc := c.ContactCollector(); cc != nil {
|
||||
@@ -357,6 +334,22 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Channel) targetAgentID(mediaList []media.MediaInfo) string {
|
||||
targetAgentID := c.AgentID()
|
||||
if c.config.VoiceAgentID == "" {
|
||||
return targetAgentID
|
||||
}
|
||||
for _, mi := range mediaList {
|
||||
if mi.Type == media.TypeAudio || mi.Type == media.TypeVoice {
|
||||
slog.Debug("discord: routing voice inbound to speaking agent",
|
||||
"agent_id", c.config.VoiceAgentID, "media_type", mi.Type,
|
||||
)
|
||||
return c.config.VoiceAgentID
|
||||
}
|
||||
}
|
||||
return targetAgentID
|
||||
}
|
||||
|
||||
// checkGroupPolicy evaluates the group policy for a sender, with pairing support.
|
||||
// When RequireMention is enabled, pairing replies only fire if the bot was
|
||||
// explicitly addressed — otherwise the bot stays silent in the channel.
|
||||
|
||||
@@ -11,6 +11,8 @@ import (
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/media"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
)
|
||||
|
||||
// --- resolveDisplayName ---
|
||||
@@ -71,6 +73,30 @@ func TestResolveCachedChannelTitle(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestTargetAgentIDUsesConfiguredChannelAgent(t *testing.T) {
|
||||
ch := &Channel{BaseChannel: channels.NewBaseChannel(channels.TypeDiscord, nil, nil)}
|
||||
ch.SetAgentID("co-assistant")
|
||||
|
||||
if got := ch.targetAgentID(nil); got != "co-assistant" {
|
||||
t.Fatalf("targetAgentID() = %q, want configured channel agent", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTargetAgentIDVoiceOverrideOnlyForAudio(t *testing.T) {
|
||||
ch := &Channel{
|
||||
BaseChannel: channels.NewBaseChannel(channels.TypeDiscord, nil, nil),
|
||||
config: config.DiscordConfig{VoiceAgentID: "voice-agent"},
|
||||
}
|
||||
ch.SetAgentID("co-assistant")
|
||||
|
||||
if got := ch.targetAgentID([]media.MediaInfo{{Type: media.TypeImage}}); got != "co-assistant" {
|
||||
t.Fatalf("image targetAgentID() = %q, want channel agent", got)
|
||||
}
|
||||
if got := ch.targetAgentID([]media.MediaInfo{{Type: media.TypeVoice}}); got != "voice-agent" {
|
||||
t.Fatalf("voice targetAgentID() = %q, want voice override", got)
|
||||
}
|
||||
}
|
||||
|
||||
// --- tryHandleCommand: routing only (no session calls) ---
|
||||
|
||||
func TestTryHandleCommandRoutingNonCommand(t *testing.T) {
|
||||
|
||||
Reference in new issue
Block a user