diff --git a/README.md b/README.md index a072a002b..85da92626 100644 --- a/README.md +++ b/README.md @@ -348,6 +348,36 @@ Talk to your picoclaw through Telegram, Discord, WhatsApp, Matrix, QQ, DingTalk, > Get your user ID from `@userinfobot` on Telegram. +**Optional: Telegram forum topics** + +PicoClaw now isolates Telegram forum topics automatically. Each topic in a forum-enabled supergroup gets its own conversation/session, and replies stay in the same topic thread. + +```json +{ + "channels": { + "telegram": { + "enabled": true, + "token": "YOUR_BOT_TOKEN", + "allow_from": ["YOUR_USER_ID"], + "groups": { + "-1001234567890": { + "topics": { + "1": { "agent_id": "main" }, + "42": { "agent_id": "coder" } + } + } + } + } + } +} +``` + +Notes: +- Topic IDs are Telegram `message_thread_id` values. +- The General topic uses thread ID `1`. +- Internally, topic targets use the form `-1001234567890:topic:42`. +- Non-forum groups do not create separate sessions for reply threads. + **3. Run** ```bash @@ -1283,7 +1313,16 @@ picoclaw agent -m "Hello" "telegram": { "enabled": true, "token": "123456:ABC...", - "allow_from": ["123456789"] + "allow_from": ["123456789"], + "groups": { + "-1001234567890": { + "topics": { + "42": { + "agent_id": "coder" + } + } + } + } }, "discord": { "enabled": true, diff --git a/docs/channels/telegram/README.zh.md b/docs/channels/telegram/README.zh.md index d453c68fa..0789d236c 100644 --- a/docs/channels/telegram/README.zh.md +++ b/docs/channels/telegram/README.zh.md @@ -11,7 +11,15 @@ Telegram Channel 通过 Telegram 机器人 API 使用长轮询实现基于机器 "enabled": true, "token": "123456789:ABCdefGHIjklMNOpqrsTUVwxyz", "allow_from": ["123456789"], - "proxy": "" + "proxy": "", + "groups": { + "-1001234567890": { + "topics": { + "1": { "agent_id": "main" }, + "42": { "agent_id": "coder" } + } + } + } } } } @@ -23,6 +31,7 @@ Telegram Channel 通过 Telegram 机器人 API 使用长轮询实现基于机器 | token | string | 是 | Telegram 机器人 API Token | | allow_from | array | 否 | 用户ID白名单，空表示允许所有用户 | | proxy | string | 否 | 连接 Telegram API 的代理 URL (例如 http://127.0.0.1:7890) | +| groups | object | 否 | 群组级配置；可用于 Telegram forum topics 的每 topic agent 路由 | ## 设置流程 @@ -31,3 +40,68 @@ Telegram Channel 通过 Telegram 机器人 API 使用长轮询实现基于机器 3. 获取 HTTP API Token 4. 将 Token 填入配置文件中 5. (可选) 配置 `allow_from` 以限制允许互动的用户 ID (可通过 `@userinfobot` 获取 ID) + +## Forum Topics / 话题线程 + +PicoClaw 现在支持 Telegram forum supergroup 的 topics（话题线程）： + +- 每个 topic 会使用独立会话，上下文不会再和同群其他 topic 混在一起 +- 回复、占位消息、typing 指示器都会回到原 topic +- topic 的内部目标格式是 `-1001234567890:topic:42` +- `General` 话题的 thread id 固定为 `1` +- 普通群聊中的 reply thread 不会被当成独立 session + +### 按 topic 指定 agent + +配置路径： + +`channels.telegram.groups..topics..agent_id` + +示例： + +```json +{ + "channels": { + "telegram": { + "enabled": true, + "token": "123456789:ABCdefGHIjklMNOpqrsTUVwxyz", + "groups": { + "-1001234567890": { + "topics": { + "1": { "agent_id": "main" }, + "42": { "agent_id": "coder" }, + "77": { "agent_id": "support" } + } + } + } + } + } +} +``` + +上面配置表示： + +- General 话题走 `main` +- thread `42` 走 `coder` +- thread `77` 走 `support` + +### 进阶：使用 bindings 绑定指定 topic + +如果你更想走通用 routing 机制，也可以直接把 topic 当作 group peer： + +```json +{ + "bindings": [ + { + "agent_id": "coder", + "match": { + "channel": "telegram", + "peer": { + "kind": "group", + "id": "-1001234567890:topic:42" + } + } + } + ] +} +``` diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 1ff50fcc2..44326fec9 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -73,6 +73,8 @@ const ( metadataKeyTeamID = "team_id" metadataKeyParentPeerKind = "parent_peer_kind" metadataKeyParentPeerID = "parent_peer_id" + metadataKeyRouteAgentID = "route_agent_id" + metadataKeyRouteMatchedBy = "route_matched_by" ) func NewAgentLoop( @@ -663,12 +665,14 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) func (al *AgentLoop) resolveMessageRoute(msg bus.InboundMessage) (routing.ResolvedRoute, *AgentInstance, error) { route := al.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), + Channel: msg.Channel, + AccountID: inboundMetadata(msg, metadataKeyAccountID), + Peer: extractPeer(msg), + ParentPeer: extractParentPeer(msg), + GuildID: inboundMetadata(msg, metadataKeyGuildID), + TeamID: inboundMetadata(msg, metadataKeyTeamID), + OverrideAgentID: inboundMetadata(msg, metadataKeyRouteAgentID), + OverrideMatchedBy: inboundMetadata(msg, metadataKeyRouteMatchedBy), }) agent, ok := al.registry.GetAgent(route.AgentID) diff --git a/pkg/channels/telegram/target.go b/pkg/channels/telegram/target.go new file mode 100644 index 000000000..22b98533a --- /dev/null +++ b/pkg/channels/telegram/target.go @@ -0,0 +1,84 @@ +package telegram + +import ( + "fmt" + "strconv" + "strings" +) + +const ( + telegramTopicSeparator = ":topic:" + telegramGeneralTopicID = 1 +) + +type telegramTarget struct { + ChatID int64 + MessageThreadID int +} + +func parseTelegramTarget(raw string) (telegramTarget, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return telegramTarget{}, fmt.Errorf("empty telegram chat ID") + } + + base, topic, hasTopic := strings.Cut(raw, telegramTopicSeparator) + chatID, err := parseChatID(base) + if err != nil { + return telegramTarget{}, err + } + + target := telegramTarget{ChatID: chatID} + if !hasTopic { + return target, nil + } + + threadID, err := strconv.Atoi(strings.TrimSpace(topic)) + if err != nil { + return telegramTarget{}, fmt.Errorf("invalid telegram topic ID %q: %w", topic, err) + } + if threadID <= 0 { + return telegramTarget{}, fmt.Errorf("invalid telegram topic ID %d", threadID) + } + target.MessageThreadID = threadID + return target, nil +} + +func buildTelegramTopicChatID(chatID int64, threadID int) string { + if threadID <= 0 { + return strconv.FormatInt(chatID, 10) + } + return fmt.Sprintf("%d%s%d", chatID, telegramTopicSeparator, threadID) +} + +func resolveTelegramForumThreadID(isForum bool, messageThreadID int) (int, bool) { + if !isForum { + return 0, false + } + if messageThreadID <= 0 { + return telegramGeneralTopicID, true + } + return messageThreadID, true +} + +func (t telegramTarget) chatIDString() string { + return strconv.FormatInt(t.ChatID, 10) +} + +func (t telegramTarget) topicChatID() string { + return buildTelegramTopicChatID(t.ChatID, t.MessageThreadID) +} + +func (t telegramTarget) messageThreadIDForSend() (int, bool) { + if t.MessageThreadID <= 0 || t.MessageThreadID == telegramGeneralTopicID { + return 0, false + } + return t.MessageThreadID, true +} + +func (t telegramTarget) messageThreadIDForTyping() (int, bool) { + if t.MessageThreadID <= 0 { + return 0, false + } + return t.MessageThreadID, true +} diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 73f0cd2db..cbc1664d2 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -174,7 +174,7 @@ func (c *TelegramChannel) SendMessageWithID(ctx context.Context, msg bus.Outboun return "", channels.ErrNotRunning } - cid, err := parseChatID(msg.ChatID) + target, err := parseTelegramTarget(msg.ChatID) if err != nil { return "", fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) } @@ -186,7 +186,7 @@ func (c *TelegramChannel) SendMessageWithID(ctx context.Context, msg bus.Outboun chunks := telegramMessageChunks(msg.Content) ids := make([]string, 0, len(chunks)) for _, chunk := range chunks { - msgID, err := c.sendHTMLChunk(ctx, cid, chunk.HTML, chunk.Markdown) + msgID, err := c.sendHTMLChunk(ctx, target, chunk.HTML, chunk.Markdown) if err != nil { return "", err } @@ -201,9 +201,16 @@ func (c *TelegramChannel) SendMessageWithID(ctx context.Context, msg bus.Outboun // sendHTMLChunk sends a single HTML message, falling back to the original // markdown as plain text on parse failure so users never see raw HTML tags. -func (c *TelegramChannel) sendHTMLChunk(ctx context.Context, chatID int64, htmlContent, mdFallback string) (int, error) { - tgMsg := tu.Message(tu.ID(chatID), htmlContent) +func (c *TelegramChannel) sendHTMLChunk( + ctx context.Context, + target telegramTarget, + htmlContent, mdFallback string, +) (int, error) { + tgMsg := tu.Message(tu.ID(target.ChatID), htmlContent) tgMsg.ParseMode = telego.ModeHTML + if threadID, ok := target.messageThreadIDForSend(); ok { + tgMsg.MessageThreadID = threadID + } msg, err := c.bot.SendMessage(ctx, tgMsg) if err != nil { @@ -225,13 +232,24 @@ func (c *TelegramChannel) sendHTMLChunk(ctx context.Context, chatID int64, htmlC // (Telegram's typing indicator expires after ~5s) in a background goroutine. // The returned stop function is idempotent and cancels the goroutine. func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func(), error) { - cid, err := parseChatID(chatID) + target, err := parseTelegramTarget(chatID) if err != nil { return func() {}, err } + sendTyping := func(callCtx context.Context) { + params := &telego.SendChatActionParams{ + ChatID: tu.ID(target.ChatID), + Action: telego.ChatActionTyping, + } + if threadID, ok := target.messageThreadIDForTyping(); ok { + params.MessageThreadID = threadID + } + _ = c.bot.SendChatAction(callCtx, params) + } + // Send the first typing action immediately - _ = c.bot.SendChatAction(ctx, tu.ChatAction(tu.ID(cid), telego.ChatActionTyping)) + sendTyping(ctx) typingCtx, cancel := context.WithCancel(ctx) go func() { @@ -242,7 +260,7 @@ func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func( case <-typingCtx.Done(): return case <-ticker.C: - _ = c.bot.SendChatAction(typingCtx, tu.ChatAction(tu.ID(cid), telego.ChatActionTyping)) + sendTyping(typingCtx) } } }() @@ -252,7 +270,7 @@ func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func( // EditMessage implements channels.MessageEditor. func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error { - cid, err := parseChatID(chatID) + target, err := parseTelegramTarget(chatID) if err != nil { return err } @@ -266,7 +284,7 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag } for i, mid := range messageIDs { - if err := c.editHTMLChunk(ctx, cid, mid, chunks[i].HTML, chunks[i].Markdown); err != nil { + if err := c.editHTMLChunk(ctx, target.ChatID, mid, chunks[i].HTML, chunks[i].Markdown); err != nil { return err } } @@ -278,6 +296,9 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag // It sends a placeholder message (e.g. "Thinking... 💭") that will later be // edited to the actual response via EditMessage (channels.MessageEditor). func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) { + if c.config == nil { + return "", nil + } phCfg := c.config.Channels.Telegram.Placeholder if !phCfg.Enabled { return "", nil @@ -288,12 +309,16 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (s text = "Thinking... 💭" } - cid, err := parseChatID(chatID) + target, err := parseTelegramTarget(chatID) if err != nil { return "", err } - pMsg, err := c.bot.SendMessage(ctx, tu.Message(tu.ID(cid), text)) + params := tu.Message(tu.ID(target.ChatID), text) + if threadID, ok := target.messageThreadIDForSend(); ok { + params.MessageThreadID = threadID + } + pMsg, err := c.bot.SendMessage(ctx, params) if err != nil { return "", err } @@ -307,7 +332,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe return channels.ErrNotRunning } - chatID, err := parseChatID(msg.ChatID) + target, err := parseTelegramTarget(msg.ChatID) if err != nil { return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) } @@ -339,31 +364,43 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe switch part.Type { case "image": params := &telego.SendPhotoParams{ - ChatID: tu.ID(chatID), + ChatID: tu.ID(target.ChatID), Photo: telego.InputFile{File: file}, Caption: part.Caption, } + if threadID, ok := target.messageThreadIDForSend(); ok { + params.MessageThreadID = threadID + } _, err = c.bot.SendPhoto(ctx, params) case "audio": params := &telego.SendAudioParams{ - ChatID: tu.ID(chatID), + ChatID: tu.ID(target.ChatID), Audio: telego.InputFile{File: file}, Caption: part.Caption, } + if threadID, ok := target.messageThreadIDForSend(); ok { + params.MessageThreadID = threadID + } _, err = c.bot.SendAudio(ctx, params) case "video": params := &telego.SendVideoParams{ - ChatID: tu.ID(chatID), + ChatID: tu.ID(target.ChatID), Video: telego.InputFile{File: file}, Caption: part.Caption, } + if threadID, ok := target.messageThreadIDForSend(); ok { + params.MessageThreadID = threadID + } _, err = c.bot.SendVideo(ctx, params) default: // "file" or unknown types params := &telego.SendDocumentParams{ - ChatID: tu.ID(chatID), + ChatID: tu.ID(target.ChatID), Document: telego.InputFile{File: file}, Caption: part.Caption, } + if threadID, ok := target.messageThreadIDForSend(); ok { + params.MessageThreadID = threadID + } _, err = c.bot.SendDocument(ctx, params) } @@ -410,11 +447,15 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes chatID := message.Chat.ID c.chatIDs[platformID] = chatID + threadID, hasTopic := resolveTelegramForumThreadID(message.Chat.IsForum, message.MessageThreadID) content := "" mediaPaths := []string{} - chatIDStr := fmt.Sprintf("%d", chatID) + chatIDStr := buildTelegramTopicChatID(chatID, threadID) + if !hasTopic { + chatIDStr = fmt.Sprintf("%d", chatID) + } messageIDStr := fmt.Sprintf("%d", message.MessageID) scope := channels.BuildMediaScope("telegram", chatIDStr, messageIDStr) @@ -508,8 +549,10 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes logger.DebugCF("telegram", "Received message", map[string]any{ "sender_id": sender.CanonicalID, - "chat_id": fmt.Sprintf("%d", chatID), + "chat_id": chatIDStr, "preview": utils.Truncate(content, 50), + "is_forum": message.Chat.IsForum, + "thread_id": threadID, }) // Placeholder is now auto-triggered by BaseChannel.HandleMessage via PlaceholderCapable @@ -518,7 +561,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes peerID := fmt.Sprintf("%d", user.ID) if message.Chat.Type != "private" { peerKind = "group" - peerID = fmt.Sprintf("%d", chatID) + peerID = chatIDStr } peer := bus.Peer{Kind: peerKind, ID: peerID} @@ -529,13 +572,24 @@ 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_forum": fmt.Sprintf("%t", message.Chat.IsForum), + "chat_id": fmt.Sprintf("%d", chatID), + } + if hasTopic { + metadata["thread_id"] = strconv.Itoa(threadID) + metadata["parent_peer_kind"] = "group" + metadata["parent_peer_id"] = fmt.Sprintf("%d", chatID) + if agentID := c.topicAgentID(chatID, threadID); agentID != "" { + metadata["route_agent_id"] = agentID + metadata["route_matched_by"] = "telegram.topic" + } } c.HandleMessage(c.ctx, peer, messageID, platformID, - fmt.Sprintf("%d", chatID), + chatIDStr, content, mediaPaths, metadata, @@ -584,9 +638,25 @@ func (c *TelegramChannel) downloadFile(ctx context.Context, fileID, ext string) } func parseChatID(chatIDStr string) (int64, error) { - var id int64 - _, err := fmt.Sscanf(chatIDStr, "%d", &id) - return id, err + return strconv.ParseInt(strings.TrimSpace(chatIDStr), 10, 64) +} + +func (c *TelegramChannel) topicAgentID(chatID int64, threadID int) string { + if c.config == nil || threadID <= 0 { + return "" + } + + groupCfg, ok := c.config.Channels.Telegram.Groups[strconv.FormatInt(chatID, 10)] + if !ok || len(groupCfg.Topics) == 0 { + return "" + } + + topicCfg, ok := groupCfg.Topics[strconv.Itoa(threadID)] + if !ok { + return "" + } + + return strings.TrimSpace(topicCfg.AgentID) } type telegramMessageChunk struct { diff --git a/pkg/channels/telegram/telegram_dispatch_test.go b/pkg/channels/telegram/telegram_dispatch_test.go index 1ea4a4824..963eb146b 100644 --- a/pkg/channels/telegram/telegram_dispatch_test.go +++ b/pkg/channels/telegram/telegram_dispatch_test.go @@ -9,6 +9,7 @@ import ( "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" ) func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) { @@ -50,3 +51,114 @@ func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) { t.Fatalf("content=%q", inbound.Content) } } + +func TestHandleMessage_ForumTopic_IsolatesChatAndAddsRoutingMetadata(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &TelegramChannel{ + BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), + chatIDs: make(map[string]int64), + ctx: context.Background(), + config: &config.Config{ + Channels: config.ChannelsConfig{ + Telegram: config.TelegramConfig{ + Groups: map[string]config.TelegramGroupConfig{ + "-1001234567890": { + Topics: map[string]config.TelegramTopicConfig{ + "42": {AgentID: "coder"}, + }, + }, + }, + }, + }, + }, + } + + msg := &telego.Message{ + Text: "hello topic", + MessageID: 10, + MessageThreadID: 42, + Chat: telego.Chat{ + ID: -1001234567890, + Type: "supergroup", + IsForum: true, + }, + 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.ChatID != "-1001234567890:topic:42" { + t.Fatalf("chat_id=%q", inbound.ChatID) + } + if inbound.Peer.ID != "-1001234567890:topic:42" { + t.Fatalf("peer_id=%q", inbound.Peer.ID) + } + if inbound.Metadata["parent_peer_kind"] != "group" { + t.Fatalf("parent_peer_kind=%q", inbound.Metadata["parent_peer_kind"]) + } + if inbound.Metadata["parent_peer_id"] != "-1001234567890" { + t.Fatalf("parent_peer_id=%q", inbound.Metadata["parent_peer_id"]) + } + if inbound.Metadata["route_agent_id"] != "coder" { + t.Fatalf("route_agent_id=%q", inbound.Metadata["route_agent_id"]) + } + if inbound.Metadata["route_matched_by"] != "telegram.topic" { + t.Fatalf("route_matched_by=%q", inbound.Metadata["route_matched_by"]) + } +} + +func TestHandleMessage_NonForumGroup_IgnoresThreadID(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 group", + MessageID: 11, + MessageThreadID: 42, + Chat: telego.Chat{ + ID: -1001234567890, + Type: "supergroup", + }, + 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.ChatID != "-1001234567890" { + t.Fatalf("chat_id=%q", inbound.ChatID) + } + if inbound.Peer.ID != "-1001234567890" { + t.Fatalf("peer_id=%q", inbound.Peer.ID) + } + if _, exists := inbound.Metadata["thread_id"]; exists { + t.Fatalf("unexpected thread_id metadata=%q", inbound.Metadata["thread_id"]) + } +} diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index ffb9bbad0..e65c8eba7 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -4,6 +4,8 @@ import ( "context" "encoding/json" "errors" + "fmt" + "os" "strings" "testing" @@ -14,6 +16,8 @@ import ( "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/media" ) const testToken = "1234567890:aaaabbbbaaaabbbbaaaabbbbaaaabbbbccc" @@ -38,14 +42,28 @@ func (s *stubCaller) Call(ctx context.Context, url string, data *ta.RequestData) type stubConstructor struct{} func (s *stubConstructor) JSONRequest(parameters any) (*ta.RequestData, error) { - return &ta.RequestData{}, nil + body, err := json.Marshal(parameters) + if err != nil { + return nil, err + } + return &ta.RequestData{ + ContentType: ta.ContentTypeJSON, + BodyRaw: body, + }, nil } func (s *stubConstructor) MultipartRequest( parameters map[string]string, files map[string]ta.NamedReader, ) (*ta.RequestData, error) { - return &ta.RequestData{}, nil + body, err := json.Marshal(parameters) + if err != nil { + return nil, err + } + return &ta.RequestData{ + ContentType: ta.ContentTypeJSON, + BodyRaw: body, + }, nil } // successResponse returns a ta.Response that telego will treat as a successful SendMessage. @@ -84,6 +102,14 @@ func newTestChannel(t *testing.T, caller *stubCaller) *TelegramChannel { } } +func decodeCallBody(t *testing.T, call stubCall) map[string]any { + t.Helper() + + var body map[string]any + require.NoError(t, json.Unmarshal(call.Data.BodyRaw, &body)) + return body +} + func TestSend_Wrapper(t *testing.T) { caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { @@ -132,6 +158,47 @@ func TestSendMessageWithID_ShortMessage_SingleCall(t *testing.T) { assert.Len(t, caller.calls, 1, "short message should result in exactly one SendMessage call") } +func TestSendMessageWithID_ForumTopic_UsesThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + msgID, err := ch.SendMessageWithID(context.Background(), bus.OutboundMessage{ + ChatID: "-1001234567890:topic:42", + Content: "Hello, topic!", + }) + + assert.NoError(t, err) + assert.Equal(t, "1", msgID) + require.Len(t, caller.calls, 1) + body := decodeCallBody(t, caller.calls[0]) + assert.Equal(t, float64(42), body["message_thread_id"]) +} + +func TestSendMessageWithID_GeneralTopic_OmitsThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + msgID, err := ch.SendMessageWithID(context.Background(), bus.OutboundMessage{ + ChatID: "-1001234567890:topic:1", + Content: "Hello, general!", + }) + + assert.NoError(t, err) + assert.Equal(t, "1", msgID) + require.Len(t, caller.calls, 1) + body := decodeCallBody(t, caller.calls[0]) + _, hasThreadID := body["message_thread_id"] + assert.False(t, hasThreadID) +} + func TestSendMessageWithID_LongMessage_SingleCall(t *testing.T) { caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { @@ -238,6 +305,92 @@ func TestEditMessage_MultipleChunkIDs(t *testing.T) { assert.Len(t, caller.calls, 2, "multi-part edit should update every tracked message") } +func TestStartTyping_ForumTopic_UsesThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + stop, err := ch.StartTyping(context.Background(), "-1001234567890:topic:42") + require.NoError(t, err) + stop() + + require.NotEmpty(t, caller.calls) + body := decodeCallBody(t, caller.calls[0]) + assert.Equal(t, float64(42), body["message_thread_id"]) +} + +func TestStartTyping_GeneralTopic_KeepsThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + stop, err := ch.StartTyping(context.Background(), "-1001234567890:topic:1") + require.NoError(t, err) + stop() + + require.NotEmpty(t, caller.calls) + body := decodeCallBody(t, caller.calls[0]) + assert.Equal(t, float64(1), body["message_thread_id"]) +} + +func TestSendPlaceholder_ForumTopic_UsesThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + ch.config = config.DefaultConfig() + ch.config.Channels.Telegram.Placeholder.Enabled = true + ch.config.Channels.Telegram.Placeholder.Text = "Thinking" + + msgID, err := ch.SendPlaceholder(context.Background(), "-1001234567890:topic:42") + require.NoError(t, err) + assert.Equal(t, "1", msgID) + require.Len(t, caller.calls, 1) + body := decodeCallBody(t, caller.calls[0]) + assert.Equal(t, float64(42), body["message_thread_id"]) +} + +func TestSendMedia_ForumTopic_UsesThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + store := media.NewFileMediaStore() + ch.SetMediaStore(store) + + tmpFile, err := os.CreateTemp(t.TempDir(), "telegram-media-*.jpg") + require.NoError(t, err) + _, err = tmpFile.WriteString("hello") + require.NoError(t, err) + require.NoError(t, tmpFile.Close()) + + ref, err := store.Store(tmpFile.Name(), media.MediaMeta{Filename: "photo.jpg"}, "test-scope") + require.NoError(t, err) + + err = ch.SendMedia(context.Background(), bus.OutboundMediaMessage{ + ChatID: "-1001234567890:topic:42", + Parts: []bus.MediaPart{{ + Type: "image", + Ref: ref, + }}, + }) + require.NoError(t, err) + require.Len(t, caller.calls, 1) + body := decodeCallBody(t, caller.calls[0]) + assert.Equal(t, "42", fmt.Sprint(body["message_thread_id"])) +} + func TestSendMessageWithID_NotRunning(t *testing.T) { caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { @@ -271,3 +424,37 @@ func TestSendMessageWithID_InvalidChatID(t *testing.T) { assert.True(t, errors.Is(err, channels.ErrSendFailed), "error should wrap ErrSendFailed") assert.Empty(t, caller.calls) } + +func TestParseTelegramTarget(t *testing.T) { + tests := []struct { + name string + input string + wantChatID int64 + wantThread int + wantErr bool + }{ + {name: "base chat", input: "12345", wantChatID: 12345}, + {name: "forum topic", input: "-100123:topic:42", wantChatID: -100123, wantThread: 42}, + {name: "invalid topic", input: "-100123:topic:abc", wantErr: true}, + {name: "invalid chat", input: "abc", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + target, err := parseTelegramTarget(tt.input) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.wantChatID, target.ChatID) + assert.Equal(t, tt.wantThread, target.MessageThreadID) + }) + } +} + +func TestResolveTelegramForumThreadID_GeneralTopicDefaultsToOne(t *testing.T) { + threadID, ok := resolveTelegramForumThreadID(true, 0) + require.True(t, ok) + assert.Equal(t, telegramGeneralTopicID, threadID) +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 693889bbf..f37f47125 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -262,15 +262,24 @@ type WhatsAppConfig struct { } type TelegramConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_TELEGRAM_ENABLED"` - Token string `json:"token" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"` - BaseURL string `json:"base_url" env:"PICOCLAW_CHANNELS_TELEGRAM_BASE_URL"` - Proxy string `json:"proxy" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_TELEGRAM_ALLOW_FROM"` - GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"` - Typing TypingConfig `json:"typing,omitempty"` - Placeholder PlaceholderConfig `json:"placeholder,omitempty"` - ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_TELEGRAM_REASONING_CHANNEL_ID"` + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_TELEGRAM_ENABLED"` + Token string `json:"token" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"` + BaseURL string `json:"base_url" env:"PICOCLAW_CHANNELS_TELEGRAM_BASE_URL"` + Proxy string `json:"proxy" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_TELEGRAM_ALLOW_FROM"` + Groups map[string]TelegramGroupConfig `json:"groups,omitempty"` + GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"` + Typing TypingConfig `json:"typing,omitempty"` + Placeholder PlaceholderConfig `json:"placeholder,omitempty"` + ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_TELEGRAM_REASONING_CHANNEL_ID"` +} + +type TelegramTopicConfig struct { + AgentID string `json:"agent_id,omitempty"` +} + +type TelegramGroupConfig struct { + Topics map[string]TelegramTopicConfig `json:"topics,omitempty"` } type FeishuConfig struct { diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 47f79c6f0..d1258d57d 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -189,6 +189,41 @@ func TestConfig_BackwardCompat_NoAgentsList(t *testing.T) { } } +func TestTelegramTopicAgentConfig_Parse(t *testing.T) { + jsonData := `{ + "channels": { + "telegram": { + "groups": { + "-1001234567890": { + "topics": { + "42": { + "agent_id": "coder" + } + } + } + } + } + } + }` + + cfg := DefaultConfig() + if err := json.Unmarshal([]byte(jsonData), cfg); err != nil { + t.Fatalf("unmarshal: %v", err) + } + + groupCfg, ok := cfg.Channels.Telegram.Groups["-1001234567890"] + if !ok { + t.Fatal("expected telegram group config to be parsed") + } + topicCfg, ok := groupCfg.Topics["42"] + if !ok { + t.Fatal("expected telegram topic config to be parsed") + } + if topicCfg.AgentID != "coder" { + t.Errorf("topicCfg.AgentID = %q, want %q", topicCfg.AgentID, "coder") + } +} + // TestDefaultConfig_HeartbeatEnabled verifies heartbeat is enabled by default func TestDefaultConfig_HeartbeatEnabled(t *testing.T) { cfg := DefaultConfig() diff --git a/pkg/routing/route.go b/pkg/routing/route.go index 9eb060c53..f17c754c3 100644 --- a/pkg/routing/route.go +++ b/pkg/routing/route.go @@ -14,6 +14,10 @@ type RouteInput struct { ParentPeer *RoutePeer GuildID string TeamID string + // OverrideAgentID forces routing to a specific agent before bindings/defaults. + // Intended for channel-level explicit overrides such as Telegram topic config. + OverrideAgentID string + OverrideMatchedBy string } // ResolvedRoute is the result of agent routing. @@ -23,7 +27,7 @@ type ResolvedRoute struct { AccountID string SessionKey string MainSessionKey string - MatchedBy string // "binding.peer", "binding.peer.parent", "binding.guild", "binding.team", "binding.account", "binding.channel", "default" + MatchedBy string // e.g. "binding.peer", "binding.peer.parent", "binding.guild", "binding.team", "binding.account", "binding.channel", "default", "override" } // RouteResolver determines which agent handles a message based on config bindings. @@ -37,8 +41,8 @@ func NewRouteResolver(cfg *config.Config) *RouteResolver { } // ResolveRoute determines which agent handles the message and constructs session keys. -// Implements the 7-level priority cascade: -// peer > parent_peer > guild > team > account > channel_wildcard > default +// Implements the 8-level priority cascade: +// override > peer > parent_peer > guild > team > account > channel_wildcard > default func (r *RouteResolver) ResolveRoute(input RouteInput) ResolvedRoute { channel := strings.ToLower(strings.TrimSpace(input.Channel)) accountID := NormalizeAccountID(input.AccountID) @@ -73,6 +77,15 @@ func (r *RouteResolver) ResolveRoute(input RouteInput) ResolvedRoute { } } + // Priority 0: Explicit override + if override := strings.TrimSpace(input.OverrideAgentID); override != "" { + matchedBy := strings.TrimSpace(input.OverrideMatchedBy) + if matchedBy == "" { + matchedBy = "override" + } + return choose(override, matchedBy) + } + // Priority 1: Peer binding if peer != nil && strings.TrimSpace(peer.ID) != "" { if match := r.findPeerMatch(bindings, peer); match != nil { diff --git a/pkg/routing/route_test.go b/pkg/routing/route_test.go index 8255db5f9..8e64ba625 100644 --- a/pkg/routing/route_test.go +++ b/pkg/routing/route_test.go @@ -261,6 +261,79 @@ func TestResolveRoute_InvalidAgentFallsToDefault(t *testing.T) { } } +func TestResolveRoute_ParentPeerFallback(t *testing.T) { + agents := []config.AgentConfig{ + {ID: "main", Default: true}, + {ID: "ops"}, + } + bindings := []config.AgentBinding{ + { + AgentID: "ops", + Match: config.BindingMatch{ + Channel: "telegram", + AccountID: "*", + Peer: &config.PeerMatch{Kind: "group", ID: "-1001234567890"}, + }, + }, + } + cfg := testConfig(agents, bindings) + r := NewRouteResolver(cfg) + + route := r.ResolveRoute(RouteInput{ + Channel: "telegram", + Peer: &RoutePeer{Kind: "group", ID: "-1001234567890:topic:42"}, + ParentPeer: &RoutePeer{Kind: "group", ID: "-1001234567890"}, + }) + + if route.AgentID != "ops" { + t.Errorf("AgentID = %q, want 'ops'", route.AgentID) + } + if route.MatchedBy != "binding.peer.parent" { + t.Errorf("MatchedBy = %q, want 'binding.peer.parent'", route.MatchedBy) + } + if route.SessionKey != "agent:ops:telegram:group:-1001234567890:topic:42" { + t.Errorf("SessionKey = %q", route.SessionKey) + } +} + +func TestResolveRoute_OverrideBeatsBindings(t *testing.T) { + agents := []config.AgentConfig{ + {ID: "main", Default: true}, + {ID: "coder"}, + {ID: "support"}, + } + bindings := []config.AgentBinding{ + { + AgentID: "support", + Match: config.BindingMatch{ + Channel: "telegram", + AccountID: "*", + Peer: &config.PeerMatch{Kind: "group", ID: "-1001234567890"}, + }, + }, + } + cfg := testConfig(agents, bindings) + r := NewRouteResolver(cfg) + + route := r.ResolveRoute(RouteInput{ + Channel: "telegram", + Peer: &RoutePeer{Kind: "group", ID: "-1001234567890:topic:42"}, + ParentPeer: &RoutePeer{Kind: "group", ID: "-1001234567890"}, + OverrideAgentID: "coder", + OverrideMatchedBy: "telegram.topic", + }) + + if route.AgentID != "coder" { + t.Errorf("AgentID = %q, want 'coder'", route.AgentID) + } + if route.MatchedBy != "telegram.topic" { + t.Errorf("MatchedBy = %q, want 'telegram.topic'", route.MatchedBy) + } + if route.SessionKey != "agent:coder:telegram:group:-1001234567890:topic:42" { + t.Errorf("SessionKey = %q", route.SessionKey) + } +} + func TestResolveRoute_DefaultAgentSelection(t *testing.T) { agents := []config.AgentConfig{ {ID: "alpha"}, diff --git a/pkg/routing/session_key_test.go b/pkg/routing/session_key_test.go index ad7a1ca02..0ec52af5b 100644 --- a/pkg/routing/session_key_test.go +++ b/pkg/routing/session_key_test.go @@ -84,6 +84,19 @@ func TestBuildAgentPeerSessionKey_GroupPeer(t *testing.T) { } } +func TestBuildAgentPeerSessionKey_GroupTopicPeer(t *testing.T) { + got := BuildAgentPeerSessionKey(SessionKeyParams{ + AgentID: "main", + Channel: "telegram", + Peer: &RoutePeer{Kind: "group", ID: "-1001234567890:topic:42"}, + DMScope: DMScopePerPeer, + }) + want := "agent:main:telegram:group:-1001234567890:topic:42" + if got != want { + t.Errorf("GroupTopicPeer = %q, want %q", got, want) + } +} + func TestBuildAgentPeerSessionKey_NilPeer(t *testing.T) { got := BuildAgentPeerSessionKey(SessionKeyParams{ AgentID: "main",