From 93081159c80067ac186482ccc84576c9fbbcef1d Mon Sep 17 00:00:00 2001 From: SebastianBoehler <27767932+SebastianBoehler@users.noreply.github.com> Date: Sun, 15 Mar 2026 13:36:46 +0100 Subject: [PATCH] refactor(telegram): persist silent group messages outside agent loop --- pkg/agent/instance.go | 4 - pkg/agent/loop.go | 85 ++++---- pkg/agent/loop_test.go | 189 +----------------- pkg/channels/base.go | 114 +++++++---- pkg/channels/manager.go | 45 +++-- pkg/channels/telegram/telegram.go | 62 +++--- .../telegram/telegram_dispatch_test.go | 74 ++++++- .../telegram_group_command_filter_test.go | 13 +- pkg/config/config.go | 20 +- pkg/config/config_test.go | 17 +- 10 files changed, 287 insertions(+), 336 deletions(-) diff --git a/pkg/agent/instance.go b/pkg/agent/instance.go index e1df1dc06..0c7baa1ee 100644 --- a/pkg/agent/instance.go +++ b/pkg/agent/instance.go @@ -37,7 +37,6 @@ type AgentInstance struct { ContextBuilder *ContextBuilder Tools *tools.ToolRegistry Subagents *config.SubagentsConfig - GroupChat *config.AgentGroupChatConfig SkillsFilter []string Candidates []providers.FallbackCandidate @@ -109,14 +108,12 @@ 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 } @@ -235,7 +232,6 @@ 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 0015ffde4..392ea6924 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -69,8 +69,6 @@ const ( metadataKeyAccountID = "account_id" metadataKeyGuildID = "guild_id" metadataKeyTeamID = "team_id" - metadataKeyIsGroup = "is_group" - metadataKeyIsMentioned = "is_mentioned" metadataKeyParentPeerKind = "parent_peer_kind" metadataKeyParentPeerID = "parent_peer_id" ) @@ -446,6 +444,9 @@ func (al *AgentLoop) RegisterTool(tool tools.Tool) { func (al *AgentLoop) SetChannelManager(cm *channels.Manager) { al.channelManager = cm + if cm != nil { + cm.SetPassiveInboundRecorder(&passiveInboundRecorder{registry: al.registry}) + } } // SetMediaStore injects a MediaStore for media lifecycle management. @@ -713,11 +714,6 @@ 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, @@ -738,33 +734,6 @@ 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, @@ -1930,6 +1899,37 @@ func mapCommandError(result commands.ExecuteResult) string { return fmt.Sprintf("Failed to execute /%s: %v", result.Command, result.Err) } +type passiveInboundRecorder struct { + registry *AgentRegistry +} + +func (r *passiveInboundRecorder) RecordPassiveInbound(ctx context.Context, msg bus.InboundMessage) error { + if r == nil || r.registry == nil { + return fmt.Errorf("passive inbound recorder not configured") + } + + route := r.registry.ResolveRoute(routing.RouteInput{ + Channel: msg.Channel, + AccountID: inboundMetadata(msg, metadataKeyAccountID), + Peer: extractPeer(msg), + ParentPeer: extractParentPeer(msg), + GuildID: inboundMetadata(msg, metadataKeyGuildID), + TeamID: inboundMetadata(msg, metadataKeyTeamID), + }) + + agent, ok := r.registry.GetAgent(route.AgentID) + if !ok { + agent = r.registry.GetDefaultAgent() + } + if agent == nil { + return fmt.Errorf("no agent available for passive inbound route (agent_id=%s)", route.AgentID) + } + + sessionKey := resolveScopeKey(route, msg.SessionKey) + agent.Sessions.AddMessage(sessionKey, "user", msg.Content) + return agent.Sessions.Save(sessionKey) +} + // extractPeer extracts the routing peer from the inbound message's structured Peer field. func extractPeer(msg bus.InboundMessage) *routing.RoutePeer { if msg.Peer.Kind == "" { @@ -1953,23 +1953,6 @@ 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 25eb99e65..13a5bfd7b 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -439,36 +439,11 @@ 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} +func TestPassiveInboundRecorder_PersistsObservedMessage(t *testing.T) { + al, _, _, _, cleanup := newTestAgentLoop(t) + defer cleanup() + recorder := &passiveInboundRecorder{registry: al.registry} msg := bus.InboundMessage{ Channel: "telegram", SenderID: "user1", @@ -478,18 +453,10 @@ func TestProcessMessage_GroupReplyRequiresMention_ObservesWithoutReply(t *testin 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) + if err := recorder.RecordPassiveInbound(context.Background(), msg); err != nil { + t.Fatalf("RecordPassiveInbound error: %v", err) } route := al.registry.ResolveRoute(routing.RouteInput{ @@ -510,150 +477,6 @@ func TestProcessMessage_GroupReplyRequiresMention_ObservesWithoutReply(t *testin } } -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/base.go b/pkg/channels/base.go index edb5b6f08..a400804c5 100644 --- a/pkg/channels/base.go +++ b/pkg/channels/base.go @@ -82,6 +82,10 @@ type MessageLengthProvider interface { MaxMessageLength() int } +type PassiveInboundRecorder interface { + RecordPassiveInbound(ctx context.Context, msg bus.InboundMessage) error +} + type BaseChannel struct { config any bus *bus.MessageBus @@ -94,6 +98,7 @@ type BaseChannel struct { placeholderRecorder PlaceholderRecorder owner Channel // the concrete channel that embeds this BaseChannel reasoningChannelID string + passiveRecorder PassiveInboundRecorder } func NewBaseChannel( @@ -121,6 +126,10 @@ func (c *BaseChannel) MaxMessageLength() int { return c.maxMessageLength } +func (c *BaseChannel) GroupTrigger() config.GroupTriggerConfig { + return c.groupTrigger +} + // ShouldRespondInGroup determines whether the bot should respond in a group chat. // Each channel is responsible for: // 1. Detecting isMentioned (platform-specific) @@ -237,40 +246,9 @@ func (c *BaseChannel) HandleMessage( metadata map[string]string, senderOpts ...bus.SenderInfo, ) { - // Use SenderInfo-based allow check when available, else fall back to string - var sender bus.SenderInfo - if len(senderOpts) > 0 { - sender = senderOpts[0] - } - if sender.CanonicalID != "" || sender.PlatformID != "" { - if !c.IsAllowedSender(sender) { - return - } - } else { - if !c.IsAllowed(senderID) { - return - } - } - - // Set SenderID to canonical if available, otherwise keep the raw senderID - resolvedSenderID := senderID - if sender.CanonicalID != "" { - resolvedSenderID = sender.CanonicalID - } - - scope := BuildMediaScope(c.name, chatID, messageID) - - msg := bus.InboundMessage{ - Channel: c.name, - SenderID: resolvedSenderID, - Sender: sender, - ChatID: chatID, - Content: content, - Media: media, - Peer: peer, - MessageID: messageID, - MediaScope: scope, - Metadata: metadata, + msg, ok := c.buildInboundMessage(peer, messageID, senderID, chatID, content, media, metadata, senderOpts...) + if !ok { + return } // Auto-trigger typing indicator, message reaction, and placeholder before publishing. @@ -310,6 +288,70 @@ func (c *BaseChannel) HandleMessage( } } +func (c *BaseChannel) PersistMessage( + ctx context.Context, + peer bus.Peer, + messageID, senderID, chatID, content string, + media []string, + metadata map[string]string, + senderOpts ...bus.SenderInfo, +) { + msg, ok := c.buildInboundMessage(peer, messageID, senderID, chatID, content, media, metadata, senderOpts...) + if !ok || c.passiveRecorder == nil { + return + } + if err := c.passiveRecorder.RecordPassiveInbound(ctx, msg); err != nil { + logger.ErrorCF("channels", "Failed to persist passive inbound message", map[string]any{ + "channel": c.name, + "chat_id": chatID, + "error": err.Error(), + }) + } +} + +func (c *BaseChannel) buildInboundMessage( + peer bus.Peer, + messageID, senderID, chatID, content string, + media []string, + metadata map[string]string, + senderOpts ...bus.SenderInfo, +) (bus.InboundMessage, bool) { + var sender bus.SenderInfo + if len(senderOpts) > 0 { + sender = senderOpts[0] + } + if sender.CanonicalID != "" || sender.PlatformID != "" { + if !c.IsAllowedSender(sender) { + return bus.InboundMessage{}, false + } + } else { + if !c.IsAllowed(senderID) { + return bus.InboundMessage{}, false + } + } + + // Set SenderID to canonical if available, otherwise keep the raw senderID + resolvedSenderID := senderID + if sender.CanonicalID != "" { + resolvedSenderID = sender.CanonicalID + } + + scope := BuildMediaScope(c.name, chatID, messageID) + + return bus.InboundMessage{ + Channel: c.name, + SenderID: resolvedSenderID, + Sender: sender, + ChatID: chatID, + Content: content, + Media: media, + Peer: peer, + MessageID: messageID, + MediaScope: scope, + Metadata: metadata, + }, true +} + func (c *BaseChannel) SetRunning(running bool) { c.running.Store(running) } @@ -336,6 +378,10 @@ func (c *BaseChannel) SetOwner(ch Channel) { c.owner = ch } +func (c *BaseChannel) SetPassiveInboundRecorder(r PassiveInboundRecorder) { + c.passiveRecorder = r +} + // BuildMediaScope constructs a scope key for media lifecycle tracking. func BuildMediaScope(channel, chatID, messageID string) string { id := messageID diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 472895a7a..3a2054c4f 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -77,18 +77,19 @@ type channelWorker struct { } type Manager struct { - channels map[string]Channel - workers map[string]*channelWorker - bus *bus.MessageBus - config *config.Config - mediaStore media.MediaStore - dispatchTask *asyncTask - mux *http.ServeMux - httpServer *http.Server - mu sync.RWMutex - placeholders sync.Map // "channel:chatID" → placeholderID (string) - typingStops sync.Map // "channel:chatID" → func() - reactionUndos sync.Map // "channel:chatID" → reactionEntry + channels map[string]Channel + workers map[string]*channelWorker + bus *bus.MessageBus + config *config.Config + mediaStore media.MediaStore + passiveRecorder PassiveInboundRecorder + dispatchTask *asyncTask + mux *http.ServeMux + httpServer *http.Server + mu sync.RWMutex + placeholders sync.Map // "channel:chatID" → placeholderID (string) + typingStops sync.Map // "channel:chatID" → func() + reactionUndos sync.Map // "channel:chatID" → reactionEntry } type asyncTask struct { @@ -220,6 +221,13 @@ func (m *Manager) initChannel(name, displayName string) { if setter, ok := ch.(interface{ SetOwner(ch Channel) }); ok { setter.SetOwner(ch) } + if m.passiveRecorder != nil { + if setter, ok := ch.(interface { + SetPassiveInboundRecorder(r PassiveInboundRecorder) + }); ok { + setter.SetPassiveInboundRecorder(m.passiveRecorder) + } + } m.channels[name] = ch logger.InfoCF("channels", "Channel enabled successfully", map[string]any{ "channel": displayName, @@ -227,6 +235,19 @@ func (m *Manager) initChannel(name, displayName string) { } } +func (m *Manager) SetPassiveInboundRecorder(r PassiveInboundRecorder) { + m.mu.Lock() + defer m.mu.Unlock() + m.passiveRecorder = r + for _, ch := range m.channels { + if setter, ok := ch.(interface { + SetPassiveInboundRecorder(r PassiveInboundRecorder) + }); ok { + setter.SetPassiveInboundRecorder(r) + } + } +} + func (m *Manager) initChannels() error { logger.InfoC("channels", "Initializing channel manager") diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index aeb0f4a7d..013557015 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -524,20 +524,6 @@ 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) - if isMentioned { - content = c.stripBotMention(content) - } - respond, cleaned := c.ShouldRespondInGroup(isMentioned, content) - if !respond { - return nil - } - content = cleaned - } - // For forum topics, embed the thread ID as "chatID/threadID" so replies // route to the correct topic and each topic gets its own session. // Only forum groups (IsForum) are handled; regular group reply threads @@ -548,13 +534,6 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID) } - logger.DebugCF("telegram", "Received message", map[string]any{ - "sender_id": sender.CanonicalID, - "chat_id": compositeChatID, - "thread_id": threadID, - "preview": utils.Truncate(content, 50), - }) - peerKind := "direct" peerID := fmt.Sprintf("%d", user.ID) if message.Chat.Type != "private" { @@ -566,11 +545,9 @@ 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"), - "is_mentioned": fmt.Sprintf("%t", isMentioned), + "user_id": fmt.Sprintf("%d", user.ID), + "username": user.Username, + "first_name": user.FirstName, } // Set parent_peer metadata for per-topic agent binding. @@ -579,6 +556,39 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes metadata["parent_peer_id"] = fmt.Sprintf("%d", threadID) } + isMentioned := false + // In group chats, apply unified group trigger filtering + if message.Chat.Type != "private" { + isMentioned = c.isBotMentioned(message) + if isMentioned { + content = c.stripBotMention(content) + } + respond, cleaned := c.ShouldRespondInGroup(isMentioned, content) + if !respond { + if c.GroupTrigger().ObserveOnly { + c.PersistMessage(c.ctx, + peer, + messageID, + platformID, + compositeChatID, + content, + mediaPaths, + metadata, + sender, + ) + } + return nil + } + content = cleaned + } + + logger.DebugCF("telegram", "Received message", map[string]any{ + "sender_id": sender.CanonicalID, + "chat_id": compositeChatID, + "thread_id": threadID, + "preview": utils.Truncate(content, 50), + }) + c.HandleMessage(c.ctx, peer, messageID, diff --git a/pkg/channels/telegram/telegram_dispatch_test.go b/pkg/channels/telegram/telegram_dispatch_test.go index 5045ddb90..58e90d370 100644 --- a/pkg/channels/telegram/telegram_dispatch_test.go +++ b/pkg/channels/telegram/telegram_dispatch_test.go @@ -9,8 +9,18 @@ import ( "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" ) +type passiveRecorderStub struct { + msgs []bus.InboundMessage +} + +func (r *passiveRecorderStub) RecordPassiveInbound(_ context.Context, msg bus.InboundMessage) error { + r.msgs = append(r.msgs, msg) + return nil +} + func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) { messageBus := bus.NewMessageBus() ch := &TelegramChannel{ @@ -85,8 +95,60 @@ func TestHandleMessage_IgnoresBotAuthoredMessages(t *testing.T) { } } -func TestHandleMessage_ForwardsMentionMetadataForGroups(t *testing.T) { - ch, messageBus := newGroupMentionOnlyChannel(t, "testbot") +func TestHandleMessage_GroupObserveOnly_PersistsUnmentionedMessages(t *testing.T) { + ch, messageBus := newGroupChannelWithTrigger(t, "testbot", config.GroupTriggerConfig{ + MentionOnly: true, + ObserveOnly: true, + }) + recorder := &passiveRecorderStub{} + ch.SetPassiveInboundRecorder(recorder) + + msg := &telego.Message{ + Text: "hello group", + 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(), 150*time.Millisecond) + defer cancel() + + if _, ok := messageBus.ConsumeInbound(ctx); ok { + t.Fatal("expected unmentioned group message to stay out of the agent pipeline") + } + if len(recorder.msgs) != 1 { + t.Fatalf("passive recorder calls = %d, want 1", len(recorder.msgs)) + } + + recorded := recorder.msgs[0] + if recorded.Content != "hello group" { + t.Fatalf("content=%q", recorded.Content) + } + if recorded.ChatID != "-100123" { + t.Fatalf("chat_id=%q", recorded.ChatID) + } + if recorded.Peer.Kind != "group" || recorded.Peer.ID != "-100123" { + t.Fatalf("peer=%+v", recorded.Peer) + } +} + +func TestHandleMessage_GroupObserveOnly_ForwardsMentionedMessages(t *testing.T) { + ch, messageBus := newGroupChannelWithTrigger(t, "testbot", config.GroupTriggerConfig{ + MentionOnly: true, + ObserveOnly: true, + }) + recorder := &passiveRecorderStub{} + ch.SetPassiveInboundRecorder(recorder) msg := &telego.Message{ Text: "@testbot hello", @@ -117,10 +179,10 @@ func TestHandleMessage_ForwardsMentionMetadataForGroups(t *testing.T) { 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.Content != "hello" { + t.Fatalf("content=%q", inbound.Content) } - if inbound.Metadata["is_mentioned"] != "true" { - t.Fatalf("is_mentioned=%q", inbound.Metadata["is_mentioned"]) + if len(recorder.msgs) != 0 { + t.Fatalf("passive recorder calls = %d, want 0", len(recorder.msgs)) } } diff --git a/pkg/channels/telegram/telegram_group_command_filter_test.go b/pkg/channels/telegram/telegram_group_command_filter_test.go index 0d5b985fe..3ef57ed61 100644 --- a/pkg/channels/telegram/telegram_group_command_filter_test.go +++ b/pkg/channels/telegram/telegram_group_command_filter_test.go @@ -41,13 +41,17 @@ func newTestTelegramBot(t *testing.T, username string) *telego.Bot { return bot } -func newGroupMentionOnlyChannel(t *testing.T, botUsername string) (*TelegramChannel, *bus.MessageBus) { +func newGroupChannelWithTrigger( + t *testing.T, + botUsername string, + trigger config.GroupTriggerConfig, +) (*TelegramChannel, *bus.MessageBus) { t.Helper() messageBus := bus.NewMessageBus() ch := &TelegramChannel{ BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil, - channels.WithGroupTrigger(config.GroupTriggerConfig{MentionOnly: true}), + channels.WithGroupTrigger(trigger), ), bot: newTestTelegramBot(t, botUsername), chatIDs: make(map[string]int64), @@ -56,6 +60,11 @@ func newGroupMentionOnlyChannel(t *testing.T, botUsername string) (*TelegramChan return ch, messageBus } +func newGroupMentionOnlyChannel(t *testing.T, botUsername string) (*TelegramChannel, *bus.MessageBus) { + t.Helper() + return newGroupChannelWithTrigger(t, botUsername, config.GroupTriggerConfig{MentionOnly: true}) +} + func TestHandleMessage_GroupMentionOnly_BotCommandEntity(t *testing.T) { tests := []struct { name string diff --git a/pkg/config/config.go b/pkg/config/config.go index f202f9bfb..d4c01e016 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -142,18 +142,13 @@ 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"` - GroupChat *AgentGroupChatConfig `json:"group_chat,omitempty"` - Subagents *SubagentsConfig `json:"subagents,omitempty"` -} - -type AgentGroupChatConfig struct { - ReplyRequiresMention bool `json:"reply_requires_mention,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"` + Subagents *SubagentsConfig `json:"subagents,omitempty"` } type SubagentsConfig struct { @@ -255,6 +250,7 @@ type ChannelsConfig struct { // GroupTriggerConfig controls when the bot responds in group chats. type GroupTriggerConfig struct { MentionOnly bool `json:"mention_only,omitempty"` + ObserveOnly bool `json:"observe_only,omitempty"` Prefixes []string `json:"prefixes,omitempty"` } diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index e88136e0e..c2d11e84e 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -86,15 +86,20 @@ func TestAgentConfig_FullParse(t *testing.T) { "primary": "claude-opus", "fallbacks": ["haiku"] }, - "group_chat": { - "reply_requires_mention": true - }, "subagents": { "allow_agents": ["sales"] } } ] }, + "channels": { + "telegram": { + "group_trigger": { + "mention_only": true, + "observe_only": true + } + } + }, "bindings": [ { "agent_id": "support", @@ -137,15 +142,15 @@ 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) } if support.Subagents == nil || len(support.Subagents.AllowAgents) != 1 { t.Errorf("support.Subagents = %+v", support.Subagents) } + if !cfg.Channels.Telegram.GroupTrigger.MentionOnly || !cfg.Channels.Telegram.GroupTrigger.ObserveOnly { + t.Errorf("cfg.Channels.Telegram.GroupTrigger = %+v", cfg.Channels.Telegram.GroupTrigger) + } if len(cfg.Bindings) != 1 { t.Fatalf("bindings len = %d, want 1", len(cfg.Bindings))