From 641028cf15142d5d819b3eb40236f7165fa46bae Mon Sep 17 00:00:00 2001 From: SebastianBoehler <27767932+SebastianBoehler@users.noreply.github.com> Date: Sun, 15 Mar 2026 11:40:39 +0100 Subject: [PATCH] feat(agent): observe group chats but reply only on mention --- pkg/agent/instance.go | 4 + pkg/agent/loop.go | 58 ++++- pkg/agent/loop_test.go | 215 ++++++++++++++++++ pkg/channels/telegram/telegram.go | 18 +- .../telegram/telegram_dispatch_test.go | 74 ++++++ pkg/config/config.go | 19 +- pkg/config/config_test.go | 6 + 7 files changed, 381 insertions(+), 13 deletions(-) diff --git a/pkg/agent/instance.go b/pkg/agent/instance.go index 0c7baa1ee..e1df1dc06 100644 --- a/pkg/agent/instance.go +++ b/pkg/agent/instance.go @@ -37,6 +37,7 @@ type AgentInstance struct { ContextBuilder *ContextBuilder Tools *tools.ToolRegistry Subagents *config.SubagentsConfig + GroupChat *config.AgentGroupChatConfig SkillsFilter []string Candidates []providers.FallbackCandidate @@ -108,12 +109,14 @@ func NewAgentInstance( agentID := routing.DefaultAgentID agentName := "" var subagents *config.SubagentsConfig + var groupChat *config.AgentGroupChatConfig var skillsFilter []string if agentCfg != nil { agentID = routing.NormalizeAgentID(agentCfg.ID) agentName = agentCfg.Name subagents = agentCfg.Subagents + groupChat = agentCfg.GroupChat skillsFilter = agentCfg.Skills } @@ -232,6 +235,7 @@ func NewAgentInstance( ContextBuilder: contextBuilder, Tools: toolsRegistry, Subagents: subagents, + GroupChat: groupChat, SkillsFilter: skillsFilter, Candidates: candidates, Router: router, diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index bee8d91a7..0015ffde4 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -69,6 +69,8 @@ const ( metadataKeyAccountID = "account_id" metadataKeyGuildID = "guild_id" metadataKeyTeamID = "team_id" + metadataKeyIsGroup = "is_group" + metadataKeyIsMentioned = "is_mentioned" metadataKeyParentPeerKind = "parent_peer_kind" metadataKeyParentPeerID = "parent_peer_id" ) @@ -711,6 +713,11 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) "route_channel": route.Channel, }) + if shouldObserveGroupMessage(msg, agent) { + al.observeGroupMessage(agent, sessionKey, msg) + return "", nil + } + opts := processOptions{ SessionKey: sessionKey, Channel: msg.Channel, @@ -731,6 +738,33 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) return al.runAgentLoop(ctx, agent, opts) } +func shouldObserveGroupMessage(msg bus.InboundMessage, agent *AgentInstance) bool { + if agent == nil || agent.GroupChat == nil || !agent.GroupChat.ReplyRequiresMention { + return false + } + if !inboundMetadataBoolValue(msg, metadataKeyIsGroup) { + return false + } + isMentioned, ok := inboundMetadataBool(msg, metadataKeyIsMentioned) + if !ok { + return false + } + return !isMentioned +} + +func (al *AgentLoop) observeGroupMessage(agent *AgentInstance, sessionKey string, msg bus.InboundMessage) { + agent.Sessions.AddMessage(sessionKey, "user", msg.Content) + agent.Sessions.Save(sessionKey) + al.maybeSummarize(agent, sessionKey, msg.Channel, msg.ChatID) + logger.InfoCF("agent", "Observed group message without replying", + map[string]any{ + "agent_id": agent.ID, + "session_key": sessionKey, + "channel": msg.Channel, + "chat_id": msg.ChatID, + }) +} + func (al *AgentLoop) resolveMessageRoute(msg bus.InboundMessage) (routing.ResolvedRoute, *AgentInstance, error) { route := al.registry.ResolveRoute(routing.RouteInput{ Channel: msg.Channel, @@ -1044,7 +1078,12 @@ func (al *AgentLoop) runLLMIteration( ctx, activeCandidates, func(ctx context.Context, provider, model string) (*providers.LLMResponse, error) { - return agent.Provider.Chat(ctx, messages, providerToolDefs, model, llmOpts) + modelRef := provider + "/" + model + candidateProvider, candidateModel, err := providers.CreateProviderForModelRef(al.cfg, modelRef) + if err != nil { + return nil, err + } + return candidateProvider.Chat(ctx, messages, providerToolDefs, candidateModel, llmOpts) }, ) if fbErr != nil { @@ -1914,6 +1953,23 @@ func inboundMetadata(msg bus.InboundMessage, key string) string { return msg.Metadata[key] } +func inboundMetadataBool(msg bus.InboundMessage, key string) (bool, bool) { + value := strings.TrimSpace(strings.ToLower(inboundMetadata(msg, key))) + switch value { + case "true", "1", "yes": + return true, true + case "false", "0", "no": + return false, true + default: + return false, false + } +} + +func inboundMetadataBoolValue(msg bus.InboundMessage, key string) bool { + value, ok := inboundMetadataBool(msg, key) + return ok && value +} + // extractParentPeer extracts the parent peer (reply-to) from inbound message metadata. func extractParentPeer(msg bus.InboundMessage) *routing.RoutePeer { parentKind := inboundMetadata(msg, metadataKeyParentPeerKind) diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 2e456fa60..25eb99e65 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -439,6 +439,221 @@ func TestProcessMessage_UsesRouteSessionKey(t *testing.T) { } } +func TestProcessMessage_GroupReplyRequiresMention_ObservesWithoutReply(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + List: []config.AgentConfig{{ + ID: "main", + Default: true, + GroupChat: &config.AgentGroupChatConfig{ + ReplyRequiresMention: true, + }, + }}, + }, + } + + msgBus := bus.NewMessageBus() + provider := &countingMockProvider{response: "group reply"} + al := NewAgentLoop(cfg, msgBus, provider) + helper := testHelper{al: al} + + msg := bus.InboundMessage{ + Channel: "telegram", + SenderID: "user1", + ChatID: "chat1", + Content: "hello everyone", + Peer: bus.Peer{ + Kind: "group", + ID: "chat1", + }, + Metadata: map[string]string{ + "is_group": "true", + "is_mentioned": "false", + }, + } + + response := helper.executeAndGetResponse(t, context.Background(), msg) + if response != "" { + t.Fatalf("expected no reply, got %q", response) + } + if provider.calls != 0 { + t.Fatalf("LLM should not be called for observed-only message, calls=%d", provider.calls) + } + + route := al.registry.ResolveRoute(routing.RouteInput{ + Channel: msg.Channel, + Peer: extractPeer(msg), + }) + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("No default agent found") + } + + history := defaultAgent.Sessions.GetHistory(route.SessionKey) + if len(history) != 1 { + t.Fatalf("expected observed history len=1, got %d", len(history)) + } + if history[0].Role != "user" || history[0].Content != "hello everyone" { + t.Fatalf("unexpected observed message: %+v", history[0]) + } +} + +func TestProcessMessage_GroupReplyRequiresMention_RepliesWhenMentioned(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + List: []config.AgentConfig{{ + ID: "main", + Default: true, + GroupChat: &config.AgentGroupChatConfig{ + ReplyRequiresMention: true, + }, + }}, + }, + } + + msgBus := bus.NewMessageBus() + provider := &countingMockProvider{response: "group reply"} + al := NewAgentLoop(cfg, msgBus, provider) + helper := testHelper{al: al} + + baseMsg := bus.InboundMessage{ + Channel: "telegram", + SenderID: "user1", + ChatID: "chat1", + Peer: bus.Peer{ + Kind: "group", + ID: "chat1", + }, + Metadata: map[string]string{ + "is_group": "true", + }, + } + + _ = helper.executeAndGetResponse(t, context.Background(), bus.InboundMessage{ + Channel: baseMsg.Channel, + SenderID: baseMsg.SenderID, + ChatID: baseMsg.ChatID, + Content: "keep this in context", + Peer: baseMsg.Peer, + Metadata: map[string]string{ + "is_group": "true", + "is_mentioned": "false", + }, + }) + + response := helper.executeAndGetResponse(t, context.Background(), bus.InboundMessage{ + Channel: baseMsg.Channel, + SenderID: baseMsg.SenderID, + ChatID: baseMsg.ChatID, + Content: "@bot answer now", + Peer: baseMsg.Peer, + Metadata: map[string]string{ + "is_group": "true", + "is_mentioned": "true", + }, + }) + + if response != "group reply" { + t.Fatalf("unexpected reply: %q", response) + } + if provider.calls != 1 { + t.Fatalf("LLM should be called once for mentioned message, calls=%d", provider.calls) + } + + route := al.registry.ResolveRoute(routing.RouteInput{ + Channel: baseMsg.Channel, + Peer: extractPeer(baseMsg), + }) + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("No default agent found") + } + + history := defaultAgent.Sessions.GetHistory(route.SessionKey) + if len(history) != 3 { + t.Fatalf("expected history len=3, got %d", len(history)) + } + if history[0].Content != "keep this in context" || history[1].Content != "@bot answer now" || history[2].Content != "group reply" { + t.Fatalf("unexpected history: %+v", history) + } +} + +func TestProcessMessage_GroupReplyRequiresMention_SkipsSuppressionWithoutMentionMetadata(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + List: []config.AgentConfig{{ + ID: "main", + Default: true, + GroupChat: &config.AgentGroupChatConfig{ + ReplyRequiresMention: true, + }, + }}, + }, + } + + msgBus := bus.NewMessageBus() + provider := &countingMockProvider{response: "other channel reply"} + al := NewAgentLoop(cfg, msgBus, provider) + helper := testHelper{al: al} + + response := helper.executeAndGetResponse(t, context.Background(), bus.InboundMessage{ + Channel: "whatsapp", + SenderID: "user1", + ChatID: "chat1", + Content: "hello group", + Peer: bus.Peer{ + Kind: "group", + ID: "chat1", + }, + Metadata: map[string]string{ + "is_group": "true", + }, + }) + + if response != "other channel reply" { + t.Fatalf("unexpected reply: %q", response) + } + if provider.calls != 1 { + t.Fatalf("LLM should still run when mention metadata is unavailable, calls=%d", provider.calls) + } +} + func TestProcessMessage_CommandOutcomes(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") if err != nil { diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 4a8d34a9f..aeb0f4a7d 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -415,6 +415,12 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes if user == nil { return fmt.Errorf("message sender (user) is nil") } + if user.IsBot { + logger.DebugCF("telegram", "Ignoring bot-authored message", map[string]any{ + "user_id": user.ID, + }) + return nil + } platformID := fmt.Sprintf("%d", user.ID) sender := bus.SenderInfo{ @@ -518,9 +524,10 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes content = "[empty message]" } + isMentioned := false // In group chats, apply unified group trigger filtering if message.Chat.Type != "private" { - isMentioned := c.isBotMentioned(message) + isMentioned = c.isBotMentioned(message) if isMentioned { content = c.stripBotMention(content) } @@ -559,10 +566,11 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes messageID := fmt.Sprintf("%d", message.MessageID) metadata := map[string]string{ - "user_id": fmt.Sprintf("%d", user.ID), - "username": user.Username, - "first_name": user.FirstName, - "is_group": fmt.Sprintf("%t", message.Chat.Type != "private"), + "user_id": fmt.Sprintf("%d", user.ID), + "username": user.Username, + "first_name": user.FirstName, + "is_group": fmt.Sprintf("%t", message.Chat.Type != "private"), + "is_mentioned": fmt.Sprintf("%t", isMentioned), } // Set parent_peer metadata for per-topic agent binding. diff --git a/pkg/channels/telegram/telegram_dispatch_test.go b/pkg/channels/telegram/telegram_dispatch_test.go index 1ea4a4824..5045ddb90 100644 --- a/pkg/channels/telegram/telegram_dispatch_test.go +++ b/pkg/channels/telegram/telegram_dispatch_test.go @@ -50,3 +50,77 @@ func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) { t.Fatalf("content=%q", inbound.Content) } } + +func TestHandleMessage_IgnoresBotAuthoredMessages(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &TelegramChannel{ + BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), + chatIDs: make(map[string]int64), + ctx: context.Background(), + } + + msg := &telego.Message{ + Text: "hello from another bot", + MessageID: 10, + Chat: telego.Chat{ + ID: -100123, + Type: "group", + }, + From: &telego.User{ + ID: 777, + FirstName: "Felix", + IsBot: true, + }, + } + + if err := ch.handleMessage(context.Background(), msg); err != nil { + t.Fatalf("handleMessage error: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + + if _, ok := messageBus.ConsumeInbound(ctx); ok { + t.Fatal("expected bot-authored message to be ignored") + } +} + +func TestHandleMessage_ForwardsMentionMetadataForGroups(t *testing.T) { + ch, messageBus := newGroupMentionOnlyChannel(t, "testbot") + + msg := &telego.Message{ + Text: "@testbot hello", + Entities: []telego.MessageEntity{{ + Type: telego.EntityTypeMention, + Offset: 0, + Length: len("@testbot"), + }}, + MessageID: 11, + Chat: telego.Chat{ + ID: -100123, + Type: "group", + }, + From: &telego.User{ + ID: 42, + FirstName: "Alice", + }, + } + + if err := ch.handleMessage(context.Background(), msg); err != nil { + t.Fatalf("handleMessage error: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + inbound, ok := messageBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected inbound message to be forwarded") + } + if inbound.Metadata["is_group"] != "true" { + t.Fatalf("is_group=%q", inbound.Metadata["is_group"]) + } + if inbound.Metadata["is_mentioned"] != "true" { + t.Fatalf("is_mentioned=%q", inbound.Metadata["is_mentioned"]) + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 161290c64..f202f9bfb 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -142,13 +142,18 @@ func (m AgentModelConfig) MarshalJSON() ([]byte, error) { } type AgentConfig struct { - ID string `json:"id"` - Default bool `json:"default,omitempty"` - Name string `json:"name,omitempty"` - Workspace string `json:"workspace,omitempty"` - Model *AgentModelConfig `json:"model,omitempty"` - Skills []string `json:"skills,omitempty"` - Subagents *SubagentsConfig `json:"subagents,omitempty"` + ID string `json:"id"` + Default bool `json:"default,omitempty"` + Name string `json:"name,omitempty"` + Workspace string `json:"workspace,omitempty"` + Model *AgentModelConfig `json:"model,omitempty"` + Skills []string `json:"skills,omitempty"` + GroupChat *AgentGroupChatConfig `json:"group_chat,omitempty"` + Subagents *SubagentsConfig `json:"subagents,omitempty"` +} + +type AgentGroupChatConfig struct { + ReplyRequiresMention bool `json:"reply_requires_mention,omitempty"` } type SubagentsConfig struct { diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 7ec0e90ba..e88136e0e 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -86,6 +86,9 @@ func TestAgentConfig_FullParse(t *testing.T) { "primary": "claude-opus", "fallbacks": ["haiku"] }, + "group_chat": { + "reply_requires_mention": true + }, "subagents": { "allow_agents": ["sales"] } @@ -134,6 +137,9 @@ func TestAgentConfig_FullParse(t *testing.T) { if support.Model == nil || support.Model.Primary != "claude-opus" { t.Errorf("support.Model = %+v", support.Model) } + if support.GroupChat == nil || !support.GroupChat.ReplyRequiresMention { + t.Errorf("support.GroupChat = %+v", support.GroupChat) + } if len(support.Model.Fallbacks) != 1 || support.Model.Fallbacks[0] != "haiku" { t.Errorf("support.Model.Fallbacks = %v", support.Model.Fallbacks) }