From 2ee39c84837a7f26ee39c9150288a8014f77a3fb Mon Sep 17 00:00:00 2001 From: viettranx Date: Fri, 27 Feb 2026 19:59:51 +0700 Subject: [PATCH] feat: Add pairing enforcement, group history, and display name resolution for Discord & WhatsApp channels. - Implement DM pairing flow for Discord and WhatsApp (checkDMPolicy, sendPairingReply with debounce) - Add "pairing" case to BaseChannel.CheckPolicy() to reject instead of falling through to "open" - Add group pending history tracking to Discord (record when not mentioned, prepend context when mentioned) - Resolve Discord display names with priority: server nickname > global name > username - Fix Gemini collapse format to prevent model from imitating tool call patterns - Fix formatTokens crash on null/undefined input --- cmd/gateway.go | 4 +- internal/channels/channel.go | 4 + internal/channels/discord/discord.go | 200 +++++++++++++++---- internal/channels/discord/factory.go | 6 +- internal/channels/whatsapp/factory.go | 4 +- internal/channels/whatsapp/whatsapp.go | 117 +++++++++-- internal/config/config_channels.go | 1 + internal/providers/openai_gemini.go | 11 +- ui/web/src/lib/format.ts | 3 +- ui/web/src/pages/channels/channel-schemas.ts | 1 + 10 files changed, 288 insertions(+), 63 deletions(-) diff --git a/cmd/gateway.go b/cmd/gateway.go index ffe92dfb..3076035a 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -694,7 +694,7 @@ func runGateway() { } if cfg.Channels.Discord.Enabled && cfg.Channels.Discord.Token != "" && instanceLoader == nil { - dc, err := discord.New(cfg.Channels.Discord, msgBus) + dc, err := discord.New(cfg.Channels.Discord, msgBus, nil) if err != nil { slog.Error("failed to initialize discord channel", "error", err) } else { @@ -704,7 +704,7 @@ func runGateway() { } if cfg.Channels.WhatsApp.Enabled && cfg.Channels.WhatsApp.BridgeURL != "" && instanceLoader == nil { - wa, err := whatsapp.New(cfg.Channels.WhatsApp, msgBus) + wa, err := whatsapp.New(cfg.Channels.WhatsApp, msgBus, nil) if err != nil { slog.Error("failed to initialize whatsapp channel", "error", err) } else { diff --git a/internal/channels/channel.go b/internal/channels/channel.go index 233a6ab8..2730116a 100644 --- a/internal/channels/channel.go +++ b/internal/channels/channel.go @@ -194,6 +194,10 @@ func (c *BaseChannel) CheckPolicy(peerKind, dmPolicy, groupPolicy, senderID stri return false case "allowlist": return c.IsAllowed(senderID) + case "pairing": + // Channels with pairing handle this before CheckPolicy. + // If we reach here, no pairing service → still allow if in allowlist. + return c.IsAllowed(senderID) default: // "open" return true } diff --git a/internal/channels/discord/discord.go b/internal/channels/discord/discord.go index f86d7884..390a7808 100644 --- a/internal/channels/discord/discord.go +++ b/internal/channels/discord/discord.go @@ -13,21 +13,28 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/channels/typing" "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/store" ) +const pairingDebounceTime = 60 * time.Second + // Channel connects to Discord via the Bot API using gateway events. type Channel struct { *channels.BaseChannel - session *discordgo.Session - config config.DiscordConfig - botUserID string // populated on start - requireMention bool // require @bot mention in groups (default true) - placeholders sync.Map // placeholderKey string → messageID string - typingCtrls sync.Map // channelID string → *typing.Controller + session *discordgo.Session + config config.DiscordConfig + botUserID string // populated on start + requireMention bool // require @bot mention in groups (default true) + placeholders sync.Map // placeholderKey string → messageID string + typingCtrls sync.Map // channelID string → *typing.Controller + pairingService store.PairingStore + pairingDebounce sync.Map // senderID → time.Time + groupHistory *channels.PendingHistory + historyLimit int } // New creates a new Discord channel from config. -func New(cfg config.DiscordConfig, msgBus *bus.MessageBus) (*Channel, error) { +func New(cfg config.DiscordConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore) (*Channel, error) { session, err := discordgo.New("Bot " + cfg.Token) if err != nil { return nil, fmt.Errorf("create discord session: %w", err) @@ -45,11 +52,19 @@ func New(cfg config.DiscordConfig, msgBus *bus.MessageBus) (*Channel, error) { requireMention = *cfg.RequireMention } + historyLimit := cfg.HistoryLimit + if historyLimit == 0 { + historyLimit = channels.DefaultGroupHistoryLimit + } + return &Channel{ BaseChannel: base, session: session, config: cfg, requireMention: requireMention, + pairingService: pairingSvc, + groupHistory: channels.NewPendingHistory(), + historyLimit: historyLimit, }, nil } @@ -194,23 +209,29 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate } senderID := m.Author.ID - senderName := m.Author.Username + senderName := resolveDisplayName(m) channelID := m.ChannelID isDM := m.GuildID == "" - // DM/Group policy check (matching TS channel policy pattern) + // DM/Group policy check peerKind := "group" if isDM { peerKind = "direct" } - if !c.CheckPolicy(peerKind, c.config.DMPolicy, c.config.GroupPolicy, senderID) { - slog.Debug("discord message rejected by policy", - "user_id", senderID, - "username", senderName, - "peer_kind", peerKind, - ) - return + + if isDM { + if !c.checkDMPolicy(senderID, channelID) { + return + } + } else { + if !c.CheckPolicy("group", "", c.config.GroupPolicy, senderID) { + slog.Debug("discord group message rejected by policy", + "user_id", senderID, + "username", senderName, + ) + return + } } // Check allowlist (for "open" policy, still apply allowlist if configured) @@ -222,25 +243,6 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate return } - // Mention gating: in groups, only respond when bot is @mentioned (default true). - if peerKind == "group" && c.requireMention { - mentioned := false - for _, u := range m.Mentions { - if u.ID == c.botUserID { - mentioned = true - break - } - } - if !mentioned { - slog.Debug("discord group message skipped: bot not mentioned", - "channel_id", channelID, - "user_id", senderID, - "username", senderName, - ) - return - } - } - // Build content content := m.Content @@ -256,6 +258,33 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate content = "[empty message]" } + // Mention gating: in groups, only respond when bot is @mentioned (default true). + // When not mentioned, record message to pending history for later context. + if peerKind == "group" && c.requireMention { + mentioned := false + for _, u := range m.Mentions { + if u.ID == c.botUserID { + mentioned = true + break + } + } + if !mentioned { + c.groupHistory.Record(channelID, channels.HistoryEntry{ + Sender: senderName, + Body: content, + Timestamp: m.Timestamp, + MessageID: m.ID, + }, c.historyLimit) + + slog.Debug("discord group message recorded (no mention)", + "channel_id", channelID, + "user_id", senderID, + "username", senderName, + ) + return + } + } + slog.Debug("discord message received", "sender_id", senderID, "channel_id", channelID, @@ -288,22 +317,113 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate c.placeholders.Store(m.ID, placeholder.ID) } - // Annotate current message with sender name so LLM knows who is talking in groups. - if peerKind == "group" && senderName != "" { - content = fmt.Sprintf("[From: %s]\n%s", senderName, content) + // Build final content with group context. + finalContent := content + if peerKind == "group" { + annotated := fmt.Sprintf("[From: %s]\n%s", senderName, content) + if c.historyLimit > 0 { + finalContent = c.groupHistory.BuildContext(channelID, annotated, c.historyLimit) + } else { + finalContent = annotated + } } metadata := map[string]string{ "message_id": m.ID, "user_id": senderID, - "username": senderName, + "username": m.Author.Username, + "display_name": senderName, "guild_id": m.GuildID, "channel_id": channelID, "is_dm": fmt.Sprintf("%t", isDM), "placeholder_key": m.ID, // keyed by inbound message ID for placeholder lookup } - c.HandleMessage(senderID, channelID, content, nil, metadata, peerKind) + c.HandleMessage(senderID, channelID, finalContent, nil, metadata, peerKind) + + // Clear pending history after sending to agent. + if peerKind == "group" { + c.groupHistory.Clear(channelID) + } +} + +// checkDMPolicy evaluates the DM policy for a sender, handling pairing flow. +func (c *Channel) checkDMPolicy(senderID, channelID string) bool { + dmPolicy := c.config.DMPolicy + if dmPolicy == "" { + dmPolicy = "pairing" + } + + switch dmPolicy { + case "disabled": + slog.Debug("discord DM rejected: disabled", "sender_id", senderID) + return false + case "open": + return true + case "allowlist": + if !c.IsAllowed(senderID) { + slog.Debug("discord DM rejected by allowlist", "sender_id", senderID) + return false + } + return true + default: // "pairing" + paired := false + if c.pairingService != nil { + paired = c.pairingService.IsPaired(senderID, c.Name()) + } + inAllowList := c.HasAllowList() && c.IsAllowed(senderID) + + if paired || inAllowList { + return true + } + + c.sendPairingReply(senderID, channelID) + return false + } +} + +// sendPairingReply sends a pairing code to the user via DM. +func (c *Channel) sendPairingReply(senderID, channelID string) { + if c.pairingService == nil { + return + } + + // Debounce + if lastSent, ok := c.pairingDebounce.Load(senderID); ok { + if time.Since(lastSent.(time.Time)) < pairingDebounceTime { + return + } + } + + code, err := c.pairingService.RequestPairing(senderID, c.Name(), channelID, "default") + if err != nil { + slog.Debug("discord pairing request failed", "sender_id", senderID, "error", err) + return + } + + replyText := fmt.Sprintf( + "GoClaw: access not configured.\n\nYour Discord user ID: %s\n\nPairing code: %s\n\nAsk the bot owner to approve with:\n goclaw pairing approve %s", + senderID, code, code, + ) + + if _, err := c.session.ChannelMessageSend(channelID, replyText); err != nil { + slog.Warn("failed to send discord pairing reply", "error", err) + } else { + c.pairingDebounce.Store(senderID, time.Now()) + slog.Info("discord pairing reply sent", "sender_id", senderID, "code", code) + } +} + +// resolveDisplayName returns the best available display name for a Discord message author. +// Priority: server nickname > global display name > username. +func resolveDisplayName(m *discordgo.MessageCreate) string { + if m.Member != nil && m.Member.Nick != "" { + return m.Member.Nick + } + if m.Author.GlobalName != "" { + return m.Author.GlobalName + } + return m.Author.Username } // lastIndexByte returns the last index of byte c in s, or -1. diff --git a/internal/channels/discord/factory.go b/internal/channels/discord/factory.go index 36d48624..a19edd7a 100644 --- a/internal/channels/discord/factory.go +++ b/internal/channels/discord/factory.go @@ -21,11 +21,12 @@ type discordInstanceConfig struct { GroupPolicy string `json:"group_policy,omitempty"` AllowFrom []string `json:"allow_from,omitempty"` RequireMention *bool `json:"require_mention,omitempty"` + HistoryLimit int `json:"history_limit,omitempty"` } // Factory creates a Discord channel from DB instance data. func Factory(name string, creds json.RawMessage, cfg json.RawMessage, - msgBus *bus.MessageBus, _ store.PairingStore) (channels.Channel, error) { + msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) { var c discordCreds if len(creds) > 0 { @@ -51,6 +52,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage, DMPolicy: ic.DMPolicy, GroupPolicy: ic.GroupPolicy, RequireMention: ic.RequireMention, + HistoryLimit: ic.HistoryLimit, } // DB instances default to "pairing" for groups (secure by default). @@ -58,7 +60,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage, dcCfg.GroupPolicy = "pairing" } - ch, err := New(dcCfg, msgBus) + ch, err := New(dcCfg, msgBus, pairingSvc) if err != nil { return nil, err } diff --git a/internal/channels/whatsapp/factory.go b/internal/channels/whatsapp/factory.go index b4c2da40..e9016f6d 100644 --- a/internal/channels/whatsapp/factory.go +++ b/internal/channels/whatsapp/factory.go @@ -24,7 +24,7 @@ type whatsappInstanceConfig struct { // Factory creates a WhatsApp channel from DB instance data. func Factory(name string, creds json.RawMessage, cfg json.RawMessage, - msgBus *bus.MessageBus, _ store.PairingStore) (channels.Channel, error) { + msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) { var c whatsappCreds if len(creds) > 0 { @@ -56,7 +56,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage, waCfg.GroupPolicy = "pairing" } - ch, err := New(waCfg, msgBus) + ch, err := New(waCfg, msgBus, pairingSvc) if err != nil { return nil, err } diff --git a/internal/channels/whatsapp/whatsapp.go b/internal/channels/whatsapp/whatsapp.go index 6650adda..53ad2024 100644 --- a/internal/channels/whatsapp/whatsapp.go +++ b/internal/channels/whatsapp/whatsapp.go @@ -14,23 +14,28 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/store" ) +const pairingDebounceTime = 60 * time.Second + // Channel connects to a WhatsApp bridge via WebSocket. // The bridge (e.g. whatsapp-web.js based) handles the actual WhatsApp // protocol; this channel just sends/receives JSON messages over WS. type Channel struct { *channels.BaseChannel - conn *websocket.Conn - config config.WhatsAppConfig - mu sync.Mutex - connected bool - ctx context.Context - cancel context.CancelFunc + conn *websocket.Conn + config config.WhatsAppConfig + mu sync.Mutex + connected bool + ctx context.Context + cancel context.CancelFunc + pairingService store.PairingStore + pairingDebounce sync.Map // senderID → time.Time } // New creates a new WhatsApp channel from config. -func New(cfg config.WhatsAppConfig, msgBus *bus.MessageBus) (*Channel, error) { +func New(cfg config.WhatsAppConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore) (*Channel, error) { if cfg.BridgeURL == "" { return nil, fmt.Errorf("whatsapp bridge_url is required") } @@ -38,8 +43,9 @@ func New(cfg config.WhatsAppConfig, msgBus *bus.MessageBus) (*Channel, error) { base := channels.NewBaseChannel("whatsapp", msgBus, cfg.AllowFrom) return &Channel{ - BaseChannel: base, - config: cfg, + BaseChannel: base, + config: cfg, + pairingService: pairingSvc, }, nil } @@ -210,9 +216,15 @@ func (c *Channel) handleIncomingMessage(msg map[string]interface{}) { } // DM/Group policy check - if !c.CheckPolicy(peerKind, c.config.DMPolicy, c.config.GroupPolicy, senderID) { - slog.Debug("whatsapp message rejected by policy", "sender_id", senderID, "peer_kind", peerKind) - return + if peerKind == "direct" { + if !c.checkDMPolicy(senderID, chatID) { + return + } + } else { + if !c.CheckPolicy("group", "", c.config.GroupPolicy, senderID) { + slog.Debug("whatsapp group message rejected by policy", "sender_id", senderID) + return + } } // Allowlist check @@ -252,3 +264,84 @@ func (c *Channel) handleIncomingMessage(msg map[string]interface{}) { c.HandleMessage(senderID, chatID, content, media, metadata, peerKind) } + +// checkDMPolicy evaluates the DM policy for a sender, handling pairing flow. +func (c *Channel) checkDMPolicy(senderID, chatID string) bool { + dmPolicy := c.config.DMPolicy + if dmPolicy == "" { + dmPolicy = "pairing" + } + + switch dmPolicy { + case "disabled": + slog.Debug("whatsapp DM rejected: disabled", "sender_id", senderID) + return false + case "open": + return true + case "allowlist": + if !c.IsAllowed(senderID) { + slog.Debug("whatsapp DM rejected by allowlist", "sender_id", senderID) + return false + } + return true + default: // "pairing" + paired := false + if c.pairingService != nil { + paired = c.pairingService.IsPaired(senderID, c.Name()) + } + inAllowList := c.HasAllowList() && c.IsAllowed(senderID) + + if paired || inAllowList { + return true + } + + c.sendPairingReply(senderID, chatID) + return false + } +} + +// sendPairingReply sends a pairing code to the user via the WS bridge. +func (c *Channel) sendPairingReply(senderID, chatID string) { + if c.pairingService == nil { + return + } + + // Debounce + if lastSent, ok := c.pairingDebounce.Load(senderID); ok { + if time.Since(lastSent.(time.Time)) < pairingDebounceTime { + return + } + } + + code, err := c.pairingService.RequestPairing(senderID, c.Name(), chatID, "default") + if err != nil { + slog.Debug("whatsapp pairing request failed", "sender_id", senderID, "error", err) + return + } + + replyText := fmt.Sprintf( + "GoClaw: access not configured.\n\nYour WhatsApp ID: %s\n\nPairing code: %s\n\nAsk the bot owner to approve with:\n goclaw pairing approve %s", + senderID, code, code, + ) + + c.mu.Lock() + defer c.mu.Unlock() + + if c.conn == nil { + slog.Warn("whatsapp bridge not connected, cannot send pairing reply") + return + } + + payload, _ := json.Marshal(map[string]interface{}{ + "type": "message", + "to": chatID, + "content": replyText, + }) + + if err := c.conn.WriteMessage(websocket.TextMessage, payload); err != nil { + slog.Warn("failed to send whatsapp pairing reply", "error", err) + } else { + c.pairingDebounce.Store(senderID, time.Now()) + slog.Info("whatsapp pairing reply sent", "sender_id", senderID, "code", code) + } +} diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index aef4720d..54927092 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -32,6 +32,7 @@ type DiscordConfig struct { DMPolicy string `json:"dm_policy,omitempty"` // "open" (default), "allowlist", "disabled" GroupPolicy string `json:"group_policy,omitempty"` // "open" (default), "allowlist", "disabled" RequireMention *bool `json:"require_mention,omitempty"` // require @bot mention in groups (default true) + HistoryLimit int `json:"history_limit,omitempty"` // max pending group messages for context (default 50, 0=disabled) } type SlackConfig struct { diff --git a/internal/providers/openai_gemini.go b/internal/providers/openai_gemini.go index b2569694..20388e7e 100644 --- a/internal/providers/openai_gemini.go +++ b/internal/providers/openai_gemini.go @@ -10,8 +10,11 @@ import ( // thought_signature (required by Gemini 2.5+) into plain text. This handles old // session history stored before the thought_signature capture fix. // Tool call cycles (assistant + tool results) are collapsed into: -// - assistant: original content + "[Called tool_name({args})]" -// - user: "[tool_name result]: content" +// - assistant: original content + summary +// - user: tool result content +// +// The collapsed format deliberately avoids looking like a tool call invocation +// to prevent Gemini from imitating the pattern instead of using structured tool_calls. func collapseToolCallsWithoutSig(msgs []Message) []Message { // Collect tool_call IDs that need collapsing. collapseIDs := make(map[string]bool) @@ -46,7 +49,7 @@ func collapseToolCallsWithoutSig(msgs []Message) []Message { } for _, tc := range m.ToolCalls { argsJSON, _ := json.Marshal(tc.Arguments) - fmt.Fprintf(&sb, "[Called %s(%s)]\n", tc.Name, string(argsJSON)) + fmt.Fprintf(&sb, "(Used tool \"%s\" with %s)\n", tc.Name, string(argsJSON)) } result = append(result, Message{ Role: "assistant", @@ -68,7 +71,7 @@ func collapseToolCallsWithoutSig(msgs []Message) []Message { if toolBuf.Len() > 0 { toolBuf.WriteString("\n\n") } - fmt.Fprintf(&toolBuf, "[%s result]: %s", toolName, toolMsg.Content) + fmt.Fprintf(&toolBuf, "Result from %s: %s", toolName, toolMsg.Content) } if toolBuf.Len() > 0 { result = append(result, Message{ diff --git a/ui/web/src/lib/format.ts b/ui/web/src/lib/format.ts index fb2c10b1..2cf7c873 100644 --- a/ui/web/src/lib/format.ts +++ b/ui/web/src/lib/format.ts @@ -24,7 +24,8 @@ export function formatRelativeTime(date: string | Date): string { return formatDate(d); } -export function formatTokens(count: number): string { +export function formatTokens(count: number | null | undefined): string { + if (count == null) return "0"; if (count >= 1_000_000) return `${(count / 1_000_000).toFixed(1)}M`; if (count >= 1_000) return `${(count / 1_000).toFixed(1)}K`; return count.toString(); diff --git a/ui/web/src/pages/channels/channel-schemas.ts b/ui/web/src/pages/channels/channel-schemas.ts index 2963a10d..ff61a142 100644 --- a/ui/web/src/pages/channels/channel-schemas.ts +++ b/ui/web/src/pages/channels/channel-schemas.ts @@ -71,6 +71,7 @@ export const configSchema: Record = { { key: "dm_policy", label: "DM Policy", type: "select", options: dmPolicyOptions, defaultValue: "open" }, { key: "group_policy", label: "Group Policy", type: "select", options: groupPolicyOptions, defaultValue: "open" }, { key: "require_mention", label: "Require @mention in groups", type: "boolean", defaultValue: true }, + { key: "history_limit", label: "Group History Limit", type: "number", defaultValue: 50, help: "Max pending group messages for context (0 = disabled)" }, { key: "allow_from", label: "Allowed Users", type: "tags", help: "Discord user IDs" }, ], feishu: [