From 2a082f4edff9f717f40fc1580216edf88aadf55f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=83=8A=E3=83=A0?= Date: Tue, 7 Jul 2026 10:53:02 +0800 Subject: [PATCH] fix(discord): preserve channel agent routing (#1380) Co-authored-by: Collective Developer --- cmd/gateway_consumer.go | 2 +- cmd/gateway_consumer_debounce.go | 5 +- cmd/gateway_consumer_handlers.go | 6 +- cmd/gateway_consumer_helpers.go | 66 ++++++++++--- cmd/gateway_consumer_normal.go | 17 +++- cmd/gateway_consumer_normal_test.go | 115 ++++++++++++++++++++++ internal/channels/discord/handler.go | 41 ++++---- internal/channels/discord/handler_test.go | 26 +++++ 8 files changed, 233 insertions(+), 45 deletions(-) diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index c5651cb4..b86f898f 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -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) } } diff --git a/cmd/gateway_consumer_debounce.go b/cmd/gateway_consumer_debounce.go index e3aa9f11..febf9751 100644 --- a/cmd/gateway_consumer_debounce.go +++ b/cmd/gateway_consumer_debounce.go @@ -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 { diff --git a/cmd/gateway_consumer_handlers.go b/cmd/gateway_consumer_handlers.go index 36baef0b..d245ed85 100644 --- a/cmd/gateway_consumer_handlers.go +++ b/cmd/gateway_consumer_handlers.go @@ -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 == "" { diff --git a/cmd/gateway_consumer_helpers.go b/cmd/gateway_consumer_helpers.go index 38df22f6..07d5ae9b 100644 --- a/cmd/gateway_consumer_helpers.go +++ b/cmd/gateway_consumer_helpers.go @@ -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. diff --git a/cmd/gateway_consumer_normal.go b/cmd/gateway_consumer_normal.go index 2fc24fbf..d63e896f 100644 --- a/cmd/gateway_consumer_normal.go +++ b/cmd/gateway_consumer_normal.go @@ -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 } diff --git a/cmd/gateway_consumer_normal_test.go b/cmd/gateway_consumer_normal_test.go index 64017f43..c6465fc8 100644 --- a/cmd/gateway_consumer_normal_test.go +++ b/cmd/gateway_consumer_normal_test.go @@ -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 +} diff --git a/internal/channels/discord/handler.go b/internal/channels/discord/handler.go index bfdb242c..9f37c5d9 100644 --- a/internal/channels/discord/handler.go +++ b/internal/channels/discord/handler.go @@ -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. diff --git a/internal/channels/discord/handler_test.go b/internal/channels/discord/handler_test.go index ef3ec83c..208bcd9b 100644 --- a/internal/channels/discord/handler_test.go +++ b/internal/channels/discord/handler_test.go @@ -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) {