opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(telegram): persist silent group messages outside agent loop

Browse source
This commit is contained in:
SebastianBoehler 2026-03-15 13:36:46 +01:00
parent 641028cf15
commit 93081159c8
10 changed files with 287 additions and 336 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
pkg/agent/instance.go
View file

@ -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,

85
pkg/agent/loop.go
View file

@ -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)

189
pkg/agent/loop_test.go
View file

@ -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 {

114
pkg/channels/base.go
View file

@ -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

45
pkg/channels/manager.go
View file

@ -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")

62
pkg/channels/telegram/telegram.go
View file

@ -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,

74
pkg/channels/telegram/telegram_dispatch_test.go
View file

@ -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))
}
}

13
pkg/channels/telegram/telegram_group_command_filter_test.go
View file

@ -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

20
pkg/config/config.go
View file

@ -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"`
}

17
pkg/config/config_test.go
View file

@ -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))