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

feat(agent): observe group chats but reply only on mention

Browse source
This commit is contained in:
SebastianBoehler 2026-03-15 11:40:39 +01:00
parent ac9aea43c8
commit 641028cf15
7 changed files with 381 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
pkg/agent/instance.go
View file

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

58
pkg/agent/loop.go
View file

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

215
pkg/agent/loop_test.go
View file

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

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

@ -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)
}
@ -563,6 +570,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
"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.

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

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

5
pkg/config/config.go
View file

@ -148,9 +148,14 @@ type AgentConfig struct {
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 {
AllowAgents []string `json:"allow_agents,omitempty"`
Model *AgentModelConfig `json:"model,omitempty"`

6
pkg/config/config_test.go
View file

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