feat: group chats observe context, respond on mention only by default (issue #12)
- Add ObserveOnly flag to bus.InboundMessage - Add BaseChannel.ObserveGroupMessage (publishes without typing/placeholder) - Update all 10 channels to call ObserveGroupMessage instead of silently dropping non-mention group messages (Telegram, Discord, Slack, Matrix, LINE, IRC, OneBot, DingTalk, Feishu, WhatsApp) - Add agent.observeMessage to store observed messages in session history with sender attribution format [DisplayName]: content - Default group_trigger.mention_only=true for Telegram, Discord, Slack, IRC, OneBot, DingTalk, Feishu, QQ (Matrix/LINE already defaulted to true) - Add GroupTrigger support to WhatsApp native (default MentionOnly=false to preserve existing respond-to-all behavior; users can opt-in) - Add tests for ObserveGroupMessage and observeMessage in loop_test.go Closes #12 https://claude.ai/code/session_01BEyzHEY8t2ghPnfr5WsSRB
This commit is contained in:
parent
e6f8793348
commit
316ce8041e
17 changed files with 449 additions and 35 deletions
|
|
@ -1372,6 +1372,11 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
|
||||||
return al.processSystemMessage(ctx, msg)
|
return al.processSystemMessage(ctx, msg)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Observe-only messages are stored in session history but do not trigger a response.
|
||||||
|
if msg.ObserveOnly {
|
||||||
|
return al.observeMessage(ctx, msg)
|
||||||
|
}
|
||||||
|
|
||||||
route, agent, routeErr := al.resolveMessageRoute(msg)
|
route, agent, routeErr := al.resolveMessageRoute(msg)
|
||||||
if routeErr != nil {
|
if routeErr != nil {
|
||||||
return "", routeErr
|
return "", routeErr
|
||||||
|
|
@ -1432,6 +1437,57 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
|
||||||
return al.runAgentLoop(ctx, agent, opts)
|
return al.runAgentLoop(ctx, agent, opts)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// observeMessage records an observe-only group message into session history without calling the LLM.
|
||||||
|
// It is invoked when a channel publishes a message with ObserveOnly=true (i.e. the bot is not
|
||||||
|
// mentioned in mention_only mode). Context is retained so the bot can reference the full
|
||||||
|
// conversation when it is later @mentioned.
|
||||||
|
func (al *AgentLoop) observeMessage(ctx context.Context, msg bus.InboundMessage) (string, error) {
|
||||||
|
route, agent, err := al.resolveMessageRoute(msg)
|
||||||
|
if err != nil {
|
||||||
|
logger.DebugCF("agent", "observeMessage: failed to resolve route, skipping", map[string]any{
|
||||||
|
"channel": msg.Channel,
|
||||||
|
"error": err.Error(),
|
||||||
|
})
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
sessionKey := resolveScopeKey(route, msg.SessionKey)
|
||||||
|
|
||||||
|
// Attribute the message to its sender so the agent has full context when later mentioned.
|
||||||
|
senderName := msg.Sender.DisplayName
|
||||||
|
if senderName == "" {
|
||||||
|
senderName = msg.SenderID
|
||||||
|
}
|
||||||
|
content := msg.Content
|
||||||
|
if content == "" && len(msg.Media) > 0 {
|
||||||
|
content = "[media]"
|
||||||
|
}
|
||||||
|
if senderName != "" && content != "" {
|
||||||
|
content = fmt.Sprintf("[%s]: %s", senderName, content)
|
||||||
|
}
|
||||||
|
|
||||||
|
if content == "" {
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
agent.Sessions.AddMessage(sessionKey, "user", content)
|
||||||
|
if saveErr := agent.Sessions.Save(sessionKey); saveErr != nil {
|
||||||
|
logger.DebugCF("agent", "observeMessage: failed to save session", map[string]any{
|
||||||
|
"session_key": sessionKey,
|
||||||
|
"error": saveErr.Error(),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.DebugCF("agent", "Observed group message (no response)", map[string]any{
|
||||||
|
"channel": msg.Channel,
|
||||||
|
"session_key": sessionKey,
|
||||||
|
"sender": senderName,
|
||||||
|
"preview": utils.Truncate(content, 60),
|
||||||
|
})
|
||||||
|
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
|
|
||||||
func (al *AgentLoop) resolveMessageRoute(msg bus.InboundMessage) (routing.ResolvedRoute, *AgentInstance, error) {
|
func (al *AgentLoop) resolveMessageRoute(msg bus.InboundMessage) (routing.ResolvedRoute, *AgentInstance, error) {
|
||||||
registry := al.GetRegistry()
|
registry := al.GetRegistry()
|
||||||
route := registry.ResolveRoute(routing.RouteInput{
|
route := registry.ResolveRoute(routing.RouteInput{
|
||||||
|
|
|
||||||
|
|
@ -3365,3 +3365,131 @@ func TestProcessMessage_ContextOverflow_AnthropicStyle(t *testing.T) {
|
||||||
t.Fatalf("expected 2 calls for retry, got %d", provider.calls)
|
t.Fatalf("expected 2 calls for retry, got %d", provider.calls)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestObserveOnlyMessage_AddsToHistoryWithoutLLMCall(t *testing.T) {
|
||||||
|
tmpDir := t.TempDir()
|
||||||
|
cfg := &config.Config{
|
||||||
|
Agents: config.AgentsConfig{
|
||||||
|
Defaults: config.AgentDefaults{
|
||||||
|
Workspace: tmpDir,
|
||||||
|
ModelName: "test-model",
|
||||||
|
MaxTokens: 4096,
|
||||||
|
MaxToolIterations: 10,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
Session: config.SessionConfig{
|
||||||
|
DMScope: "per-channel-peer",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
msgBus := bus.NewMessageBus()
|
||||||
|
provider := &countingMockProvider{response: "LLM reply"}
|
||||||
|
al := NewAgentLoop(cfg, msgBus, provider)
|
||||||
|
|
||||||
|
observeMsg := bus.InboundMessage{
|
||||||
|
Channel: "telegram",
|
||||||
|
SenderID: "telegram:user1",
|
||||||
|
Sender: bus.SenderInfo{
|
||||||
|
Platform: "telegram",
|
||||||
|
PlatformID: "user1",
|
||||||
|
CanonicalID: "telegram:user1",
|
||||||
|
DisplayName: "Alice",
|
||||||
|
},
|
||||||
|
ChatID: "group42",
|
||||||
|
Content: "we should meet at 3pm",
|
||||||
|
Peer: bus.Peer{Kind: "group", ID: "group42"},
|
||||||
|
ObserveOnly: true,
|
||||||
|
}
|
||||||
|
|
||||||
|
// Compute session key before processing
|
||||||
|
route := al.registry.ResolveRoute(routing.RouteInput{
|
||||||
|
Channel: observeMsg.Channel,
|
||||||
|
Peer: extractPeer(observeMsg),
|
||||||
|
})
|
||||||
|
sessionKey := route.SessionKey
|
||||||
|
|
||||||
|
_, err := al.processMessage(context.Background(), observeMsg)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("processMessage() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// LLM must NOT have been called for an observe-only message
|
||||||
|
if provider.calls != 0 {
|
||||||
|
t.Fatalf("expected 0 LLM calls for observe-only message, got %d", provider.calls)
|
||||||
|
}
|
||||||
|
|
||||||
|
// No outbound response should have been sent
|
||||||
|
select {
|
||||||
|
case msg := <-msgBus.OutboundChan():
|
||||||
|
t.Fatalf("expected no outbound message for observe-only, got %+v", msg)
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
|
// The message should have been added to session history with sender attribution
|
||||||
|
defaultAgent := al.registry.GetDefaultAgent()
|
||||||
|
if defaultAgent == nil {
|
||||||
|
t.Fatal("expected default agent")
|
||||||
|
}
|
||||||
|
history := defaultAgent.Sessions.GetHistory(sessionKey)
|
||||||
|
if len(history) == 0 {
|
||||||
|
t.Fatal("expected observe-only message to be stored in session history")
|
||||||
|
}
|
||||||
|
last := history[len(history)-1]
|
||||||
|
if last.Role != "user" {
|
||||||
|
t.Fatalf("expected role 'user', got %q", last.Role)
|
||||||
|
}
|
||||||
|
if !strings.Contains(last.Content, "Alice") || !strings.Contains(last.Content, "we should meet at 3pm") {
|
||||||
|
t.Fatalf("expected sender attribution and content in history, got %q", last.Content)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestObserveOnlyMessage_DoesNotRespondEvenWithContinuation(t *testing.T) {
|
||||||
|
tmpDir := t.TempDir()
|
||||||
|
cfg := &config.Config{
|
||||||
|
Agents: config.AgentsConfig{
|
||||||
|
Defaults: config.AgentDefaults{
|
||||||
|
Workspace: tmpDir,
|
||||||
|
ModelName: "test-model",
|
||||||
|
MaxTokens: 4096,
|
||||||
|
MaxToolIterations: 10,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
msgBus := bus.NewMessageBus()
|
||||||
|
provider := &countingMockProvider{response: "LLM reply"}
|
||||||
|
al := NewAgentLoop(cfg, msgBus, provider)
|
||||||
|
|
||||||
|
// Send multiple observe-only messages
|
||||||
|
for _, content := range []string{"msg1", "msg2", "msg3"} {
|
||||||
|
msg := bus.InboundMessage{
|
||||||
|
Channel: "slack",
|
||||||
|
SenderID: "slack:userA",
|
||||||
|
ChatID: "channel1",
|
||||||
|
Content: content,
|
||||||
|
Peer: bus.Peer{Kind: "channel", ID: "channel1"},
|
||||||
|
ObserveOnly: true,
|
||||||
|
}
|
||||||
|
if _, err := al.processMessage(context.Background(), msg); err != nil {
|
||||||
|
t.Fatalf("processMessage(%q) error = %v", content, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if provider.calls != 0 {
|
||||||
|
t.Fatalf("expected 0 LLM calls for observe-only messages, got %d", provider.calls)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Confirm history accumulated
|
||||||
|
route := al.registry.ResolveRoute(routing.RouteInput{
|
||||||
|
Channel: "slack",
|
||||||
|
Peer: bus.Peer{Kind: "channel", ID: "channel1"},
|
||||||
|
})
|
||||||
|
defaultAgent := al.registry.GetDefaultAgent()
|
||||||
|
if defaultAgent == nil {
|
||||||
|
t.Fatal("expected default agent")
|
||||||
|
}
|
||||||
|
history := defaultAgent.Sessions.GetHistory(route.SessionKey)
|
||||||
|
if len(history) != 3 {
|
||||||
|
t.Fatalf("expected 3 observed messages in history, got %d", len(history))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -27,6 +27,7 @@ type InboundMessage struct {
|
||||||
MediaScope string `json:"media_scope,omitempty"` // media lifecycle scope
|
MediaScope string `json:"media_scope,omitempty"` // media lifecycle scope
|
||||||
SessionKey string `json:"session_key"`
|
SessionKey string `json:"session_key"`
|
||||||
Metadata map[string]string `json:"metadata,omitempty"`
|
Metadata map[string]string `json:"metadata,omitempty"`
|
||||||
|
ObserveOnly bool `json:"observe_only,omitempty"` // store in history but do not respond
|
||||||
}
|
}
|
||||||
|
|
||||||
type OutboundMessage struct {
|
type OutboundMessage struct {
|
||||||
|
|
|
||||||
|
|
@ -329,6 +329,62 @@ func (c *BaseChannel) HandleMessage(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ObserveGroupMessage records a group message into conversation history without triggering a response.
|
||||||
|
// It is called when ShouldRespondInGroup returns false but the message should still be retained
|
||||||
|
// as context (e.g. mention_only mode). Unlike HandleMessage, it does not start typing indicators,
|
||||||
|
// reactions, or placeholder messages.
|
||||||
|
func (c *BaseChannel) ObserveGroupMessage(
|
||||||
|
ctx context.Context,
|
||||||
|
peer bus.Peer,
|
||||||
|
messageID, senderID, chatID, content string,
|
||||||
|
media []string,
|
||||||
|
metadata map[string]string,
|
||||||
|
senderOpts ...bus.SenderInfo,
|
||||||
|
) {
|
||||||
|
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
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
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,
|
||||||
|
ObserveOnly: true,
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := c.bus.PublishInbound(ctx, msg); err != nil {
|
||||||
|
logger.DebugCF("channels", "Failed to publish observed group message", map[string]any{
|
||||||
|
"channel": c.name,
|
||||||
|
"chat_id": chatID,
|
||||||
|
"error": err.Error(),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (c *BaseChannel) SetRunning(running bool) {
|
func (c *BaseChannel) SetRunning(running bool) {
|
||||||
c.running.Store(running)
|
c.running.Store(running)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,7 +1,9 @@
|
||||||
package channels
|
package channels
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/sipeed/picoclaw/pkg/bus"
|
"github.com/sipeed/picoclaw/pkg/bus"
|
||||||
"github.com/sipeed/picoclaw/pkg/config"
|
"github.com/sipeed/picoclaw/pkg/config"
|
||||||
|
|
@ -263,3 +265,72 @@ func TestIsAllowedSender(t *testing.T) {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestObserveGroupMessage_PublishesObserveOnly(t *testing.T) {
|
||||||
|
mb := bus.NewMessageBus()
|
||||||
|
defer mb.Close()
|
||||||
|
|
||||||
|
ch := NewBaseChannel("test", nil, mb, nil)
|
||||||
|
|
||||||
|
peer := bus.Peer{Kind: "group", ID: "group123"}
|
||||||
|
sender := bus.SenderInfo{
|
||||||
|
Platform: "test",
|
||||||
|
PlatformID: "user1",
|
||||||
|
CanonicalID: "test:user1",
|
||||||
|
DisplayName: "Alice",
|
||||||
|
}
|
||||||
|
|
||||||
|
go ch.ObserveGroupMessage(
|
||||||
|
context.Background(),
|
||||||
|
peer,
|
||||||
|
"msg1", "user1", "group123", "hello world",
|
||||||
|
nil, nil, sender,
|
||||||
|
)
|
||||||
|
|
||||||
|
select {
|
||||||
|
case msg := <-mb.InboundChan():
|
||||||
|
if !msg.ObserveOnly {
|
||||||
|
t.Fatalf("expected ObserveOnly=true, got false")
|
||||||
|
}
|
||||||
|
if msg.Channel != "test" {
|
||||||
|
t.Fatalf("expected channel 'test', got %q", msg.Channel)
|
||||||
|
}
|
||||||
|
if msg.Content != "hello world" {
|
||||||
|
t.Fatalf("expected content 'hello world', got %q", msg.Content)
|
||||||
|
}
|
||||||
|
if msg.Peer.Kind != "group" {
|
||||||
|
t.Fatalf("expected peer kind 'group', got %q", msg.Peer.Kind)
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("timed out waiting for observe-only message on bus")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestObserveGroupMessage_SkipsTypingAndPlaceholder(t *testing.T) {
|
||||||
|
// ObserveGroupMessage must not trigger placeholder/typing even if the channel
|
||||||
|
// has those capabilities (the owner is nil so they can't fire, but we verify
|
||||||
|
// the channel field ObserveOnly is set and no extra messages are sent).
|
||||||
|
mb := bus.NewMessageBus()
|
||||||
|
defer mb.Close()
|
||||||
|
|
||||||
|
ch := NewBaseChannel("test", nil, mb, nil)
|
||||||
|
|
||||||
|
peer := bus.Peer{Kind: "group", ID: "g1"}
|
||||||
|
go ch.ObserveGroupMessage(context.Background(), peer, "", "s1", "g1", "hi", nil, nil)
|
||||||
|
|
||||||
|
select {
|
||||||
|
case msg := <-mb.InboundChan():
|
||||||
|
if !msg.ObserveOnly {
|
||||||
|
t.Fatalf("expected ObserveOnly=true")
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("timed out")
|
||||||
|
}
|
||||||
|
|
||||||
|
// No second message (no placeholder / typing broadcast)
|
||||||
|
select {
|
||||||
|
case extra := <-mb.InboundChan():
|
||||||
|
t.Fatalf("unexpected extra message: %+v", extra)
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -197,6 +197,13 @@ func (c *DingTalkChannel) onChatBotMessageReceived(
|
||||||
// In group chats, apply unified group trigger filtering
|
// In group chats, apply unified group trigger filtering
|
||||||
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
||||||
if !respond {
|
if !respond {
|
||||||
|
observeSender := bus.SenderInfo{
|
||||||
|
Platform: "dingtalk",
|
||||||
|
PlatformID: senderID,
|
||||||
|
CanonicalID: identity.BuildCanonicalID("dingtalk", senderID),
|
||||||
|
DisplayName: senderNick,
|
||||||
|
}
|
||||||
|
c.ObserveGroupMessage(ctx, peer, "", senderID, chatID, content, nil, metadata, observeSender)
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
content = cleaned
|
content = cleaned
|
||||||
|
|
|
||||||
|
|
@ -416,6 +416,13 @@ func (c *DiscordChannel) handleMessage(s *discordgo.Session, m *discordgo.Messag
|
||||||
logger.DebugCF("discord", "Group message ignored by group trigger", map[string]any{
|
logger.DebugCF("discord", "Group message ignored by group trigger", map[string]any{
|
||||||
"user_id": m.Author.ID,
|
"user_id": m.Author.ID,
|
||||||
})
|
})
|
||||||
|
observePeer := bus.Peer{Kind: "channel", ID: m.ChannelID}
|
||||||
|
observeMeta := map[string]string{
|
||||||
|
"guild_id": m.GuildID,
|
||||||
|
"user_id": m.Author.ID,
|
||||||
|
"platform": "discord",
|
||||||
|
}
|
||||||
|
c.ObserveGroupMessage(c.ctx, observePeer, m.ID, m.Author.ID, m.ChannelID, content, nil, observeMeta, sender)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
content = cleaned
|
content = cleaned
|
||||||
|
|
|
||||||
|
|
@ -463,6 +463,7 @@ func (c *FeishuChannel) handleMessageReceive(ctx context.Context, event *larkim.
|
||||||
// In group chats, apply unified group trigger filtering
|
// In group chats, apply unified group trigger filtering
|
||||||
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
||||||
if !respond {
|
if !respond {
|
||||||
|
c.ObserveGroupMessage(ctx, peer, messageID, senderID, chatID, content, mediaRefs, metadata, senderInfo)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
content = cleaned
|
content = cleaned
|
||||||
|
|
|
||||||
|
|
@ -81,6 +81,13 @@ func (c *IRCChannel) onPrivmsg(conn *ircevent.Connection, e ircmsg.Message) {
|
||||||
}
|
}
|
||||||
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
||||||
if !respond {
|
if !respond {
|
||||||
|
observeID := fmt.Sprintf("%s-%d", nick, time.Now().UnixNano())
|
||||||
|
observeMeta := map[string]string{
|
||||||
|
"platform": "irc",
|
||||||
|
"server": c.config.Server,
|
||||||
|
"channel": target,
|
||||||
|
}
|
||||||
|
c.ObserveGroupMessage(c.ctx, peer, observeID, nick, chatID, content, nil, observeMeta, sender)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
content = cleaned
|
content = cleaned
|
||||||
|
|
|
||||||
|
|
@ -357,6 +357,17 @@ func (c *LINEChannel) processEvent(event lineEvent) {
|
||||||
logger.DebugCF("line", "Ignoring group message by group trigger", map[string]any{
|
logger.DebugCF("line", "Ignoring group message by group trigger", map[string]any{
|
||||||
"chat_id": chatID,
|
"chat_id": chatID,
|
||||||
})
|
})
|
||||||
|
observeSender := bus.SenderInfo{
|
||||||
|
Platform: "line",
|
||||||
|
PlatformID: senderID,
|
||||||
|
CanonicalID: identity.BuildCanonicalID("line", senderID),
|
||||||
|
}
|
||||||
|
observePeer := bus.Peer{Kind: "group", ID: chatID}
|
||||||
|
observeMeta := map[string]string{
|
||||||
|
"platform": "line",
|
||||||
|
"source_type": event.Source.Type,
|
||||||
|
}
|
||||||
|
c.ObserveGroupMessage(c.ctx, observePeer, msg.ID, senderID, chatID, content, mediaPaths, observeMeta, observeSender)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
content = cleaned
|
content = cleaned
|
||||||
|
|
|
||||||
|
|
@ -723,6 +723,12 @@ func (c *MatrixChannel) handleMessageEvent(ctx context.Context, evt *event.Event
|
||||||
"mention_only": c.config.GroupTrigger.MentionOnly,
|
"mention_only": c.config.GroupTrigger.MentionOnly,
|
||||||
"prefixes": c.config.GroupTrigger.Prefixes,
|
"prefixes": c.config.GroupTrigger.Prefixes,
|
||||||
})
|
})
|
||||||
|
observePeer := bus.Peer{Kind: "group", ID: roomID}
|
||||||
|
observeMeta := map[string]string{
|
||||||
|
"room_id": roomID,
|
||||||
|
"platform": "matrix",
|
||||||
|
}
|
||||||
|
c.ObserveGroupMessage(ctx, observePeer, evt.ID.String(), senderID, roomID, content, mediaPaths, observeMeta, sender)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
content = cleaned
|
content = cleaned
|
||||||
|
|
|
||||||
|
|
@ -1030,6 +1030,13 @@ func (c *OneBotChannel) handleMessage(raw *oneBotRawEvent) {
|
||||||
"is_mentioned": isBotMentioned,
|
"is_mentioned": isBotMentioned,
|
||||||
"content": truncate(content, 100),
|
"content": truncate(content, 100),
|
||||||
})
|
})
|
||||||
|
observeSenderInfo := bus.SenderInfo{
|
||||||
|
Platform: "onebot",
|
||||||
|
PlatformID: senderID,
|
||||||
|
CanonicalID: identity.BuildCanonicalID("onebot", senderID),
|
||||||
|
DisplayName: sender.Nickname,
|
||||||
|
}
|
||||||
|
c.ObserveGroupMessage(c.ctx, peer, messageID, senderID, chatID, content, nil, metadata, observeSenderInfo)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
content = strippedContent
|
content = strippedContent
|
||||||
|
|
|
||||||
|
|
@ -316,6 +316,15 @@ func (c *SlackChannel) handleMessageEvent(ev *slackevents.MessageEvent) {
|
||||||
if !strings.HasPrefix(channelID, "D") {
|
if !strings.HasPrefix(channelID, "D") {
|
||||||
respond, cleaned := c.ShouldRespondInGroup(false, content)
|
respond, cleaned := c.ShouldRespondInGroup(false, content)
|
||||||
if !respond {
|
if !respond {
|
||||||
|
observePeer := bus.Peer{Kind: "channel", ID: channelID}
|
||||||
|
observeMeta := map[string]string{
|
||||||
|
"message_ts": messageTS,
|
||||||
|
"channel_id": channelID,
|
||||||
|
"thread_ts": threadTS,
|
||||||
|
"platform": "slack",
|
||||||
|
"team_id": c.teamID,
|
||||||
|
}
|
||||||
|
c.ObserveGroupMessage(c.ctx, observePeer, messageTS, senderID, chatID, content, nil, observeMeta, sender)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
content = cleaned
|
content = cleaned
|
||||||
|
|
|
||||||
|
|
@ -690,6 +690,16 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
|
||||||
content = "[media only]"
|
content = "[media only]"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 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
|
||||||
|
// must share one session per group.
|
||||||
|
compositeChatID := fmt.Sprintf("%d", chatID)
|
||||||
|
threadID := message.MessageThreadID
|
||||||
|
if message.Chat.IsForum && threadID != 0 {
|
||||||
|
compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID)
|
||||||
|
}
|
||||||
|
|
||||||
// In group chats, apply unified group trigger filtering
|
// In group chats, apply unified group trigger filtering
|
||||||
if message.Chat.Type != "private" {
|
if message.Chat.Type != "private" {
|
||||||
isMentioned := c.isBotMentioned(message)
|
isMentioned := c.isBotMentioned(message)
|
||||||
|
|
@ -698,6 +708,14 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
|
||||||
}
|
}
|
||||||
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
||||||
if !respond {
|
if !respond {
|
||||||
|
observePeer := bus.Peer{Kind: "group", ID: compositeChatID}
|
||||||
|
observeMeta := map[string]string{
|
||||||
|
"user_id": platformID,
|
||||||
|
"username": user.Username,
|
||||||
|
"first_name": user.FirstName,
|
||||||
|
"is_group": "true",
|
||||||
|
}
|
||||||
|
c.ObserveGroupMessage(ctx, observePeer, messageIDStr, platformID, compositeChatID, content, mediaPaths, observeMeta, sender)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
content = cleaned
|
content = cleaned
|
||||||
|
|
@ -720,16 +738,6 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
|
||||||
content = c.prependTelegramQuotedReply(content, message.ReplyToMessage)
|
content = c.prependTelegramQuotedReply(content, message.ReplyToMessage)
|
||||||
}
|
}
|
||||||
|
|
||||||
// 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
|
|
||||||
// must share one session per group.
|
|
||||||
compositeChatID := fmt.Sprintf("%d", chatID)
|
|
||||||
threadID := message.MessageThreadID
|
|
||||||
if message.Chat.IsForum && threadID != 0 {
|
|
||||||
compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID)
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.DebugCF("telegram", "Received message", map[string]any{
|
logger.DebugCF("telegram", "Received message", map[string]any{
|
||||||
"sender_id": sender.CanonicalID,
|
"sender_id": sender.CanonicalID,
|
||||||
"chat_id": compositeChatID,
|
"chat_id": compositeChatID,
|
||||||
|
|
|
||||||
|
|
@ -35,6 +35,7 @@ func NewWhatsAppChannel(cfg config.WhatsAppConfig, bus *bus.MessageBus) (*WhatsA
|
||||||
bus,
|
bus,
|
||||||
cfg.AllowFrom,
|
cfg.AllowFrom,
|
||||||
channels.WithMaxMessageLength(65536),
|
channels.WithMaxMessageLength(65536),
|
||||||
|
channels.WithGroupTrigger(cfg.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(cfg.ReasoningChannelID),
|
channels.WithReasoningChannelID(cfg.ReasoningChannelID),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -248,5 +249,29 @@ func (c *WhatsAppChannel) handleIncomingMessage(msg map[string]any) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// In group chats, apply unified group trigger filtering.
|
||||||
|
// Mention detection: check if the "mentions" field in the message payload
|
||||||
|
// contains the sender's own JID (proxy for being mentioned by others),
|
||||||
|
// or use false when no mention info is available.
|
||||||
|
if peer.Kind == "group" {
|
||||||
|
isMentioned := false
|
||||||
|
if mentionList, ok := msg["mentions"].([]any); ok {
|
||||||
|
for _, m := range mentionList {
|
||||||
|
if jid, ok := m.(string); ok && jid != "" {
|
||||||
|
// Any mention in the payload counts as the bot being addressed
|
||||||
|
_ = jid
|
||||||
|
isMentioned = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
|
||||||
|
if !respond {
|
||||||
|
c.ObserveGroupMessage(c.ctx, peer, messageID, senderID, chatID, content, mediaPaths, metadata, sender)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
content = cleaned
|
||||||
|
}
|
||||||
|
|
||||||
c.HandleMessage(c.ctx, peer, messageID, senderID, chatID, content, mediaPaths, metadata, sender)
|
c.HandleMessage(c.ctx, peer, messageID, senderID, chatID, content, mediaPaths, metadata, sender)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -358,6 +358,7 @@ type WhatsAppConfig struct {
|
||||||
UseNative bool `json:"use_native" yaml:"-" env:"PICOCLAW_CHANNELS_WHATSAPP_USE_NATIVE"`
|
UseNative bool `json:"use_native" yaml:"-" env:"PICOCLAW_CHANNELS_WHATSAPP_USE_NATIVE"`
|
||||||
SessionStorePath string `json:"session_store_path" yaml:"-" env:"PICOCLAW_CHANNELS_WHATSAPP_SESSION_STORE_PATH"`
|
SessionStorePath string `json:"session_store_path" yaml:"-" env:"PICOCLAW_CHANNELS_WHATSAPP_SESSION_STORE_PATH"`
|
||||||
AllowFrom FlexibleStringSlice `json:"allow_from" yaml:"-" env:"PICOCLAW_CHANNELS_WHATSAPP_ALLOW_FROM"`
|
AllowFrom FlexibleStringSlice `json:"allow_from" yaml:"-" env:"PICOCLAW_CHANNELS_WHATSAPP_ALLOW_FROM"`
|
||||||
|
GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty" yaml:"-"`
|
||||||
ReasoningChannelID string `json:"reasoning_channel_id" yaml:"-" env:"PICOCLAW_CHANNELS_WHATSAPP_REASONING_CHANNEL_ID"`
|
ReasoningChannelID string `json:"reasoning_channel_id" yaml:"-" env:"PICOCLAW_CHANNELS_WHATSAPP_REASONING_CHANNEL_ID"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -51,10 +51,12 @@ func DefaultConfig() *Config {
|
||||||
UseNative: false,
|
UseNative: false,
|
||||||
SessionStorePath: "",
|
SessionStorePath: "",
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: false},
|
||||||
},
|
},
|
||||||
Telegram: TelegramConfig{
|
Telegram: TelegramConfig{
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: true},
|
||||||
Typing: TypingConfig{Enabled: true},
|
Typing: TypingConfig{Enabled: true},
|
||||||
Placeholder: PlaceholderConfig{
|
Placeholder: PlaceholderConfig{
|
||||||
Enabled: true,
|
Enabled: true,
|
||||||
|
|
@ -67,11 +69,13 @@ func DefaultConfig() *Config {
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
AppID: "",
|
AppID: "",
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: true},
|
||||||
},
|
},
|
||||||
Discord: DiscordConfig{
|
Discord: DiscordConfig{
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
MentionOnly: false,
|
MentionOnly: false,
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: true},
|
||||||
},
|
},
|
||||||
MaixCam: MaixCamConfig{
|
MaixCam: MaixCamConfig{
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
|
|
@ -83,6 +87,7 @@ func DefaultConfig() *Config {
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
AppID: "",
|
AppID: "",
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: true},
|
||||||
MaxMessageLength: 2000,
|
MaxMessageLength: 2000,
|
||||||
MaxBase64FileSizeMiB: 0,
|
MaxBase64FileSizeMiB: 0,
|
||||||
},
|
},
|
||||||
|
|
@ -90,10 +95,12 @@ func DefaultConfig() *Config {
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
ClientID: "",
|
ClientID: "",
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: true},
|
||||||
},
|
},
|
||||||
Slack: SlackConfig{
|
Slack: SlackConfig{
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: true},
|
||||||
},
|
},
|
||||||
Matrix: MatrixConfig{
|
Matrix: MatrixConfig{
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
|
|
@ -125,6 +132,7 @@ func DefaultConfig() *Config {
|
||||||
WSUrl: "ws://127.0.0.1:3001",
|
WSUrl: "ws://127.0.0.1:3001",
|
||||||
ReconnectInterval: 5,
|
ReconnectInterval: 5,
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: true},
|
||||||
},
|
},
|
||||||
WeCom: WeComConfig{
|
WeCom: WeComConfig{
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
|
|
@ -148,6 +156,11 @@ func DefaultConfig() *Config {
|
||||||
MaxConnections: 100,
|
MaxConnections: 100,
|
||||||
AllowFrom: FlexibleStringSlice{},
|
AllowFrom: FlexibleStringSlice{},
|
||||||
},
|
},
|
||||||
|
IRC: IRCConfig{
|
||||||
|
Enabled: false,
|
||||||
|
AllowFrom: FlexibleStringSlice{},
|
||||||
|
GroupTrigger: GroupTriggerConfig{MentionOnly: true},
|
||||||
|
},
|
||||||
},
|
},
|
||||||
Hooks: HooksConfig{
|
Hooks: HooksConfig{
|
||||||
Enabled: true,
|
Enabled: true,
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue