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

Add Telegram forum topic routing support

Browse source
This commit is contained in:
Dmitrii Balabanov 2026-03-09 12:31:53 +02:00
parent 91502fa15d
commit bdf0fd6c2e
12 changed files with 758 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
README.md
View file

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

76
docs/channels/telegram/README.zh.md
View file

@ -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.<chat_id>.topics.<thread_id>.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"
}
}
}
]
}
```

4
pkg/agent/loop.go
View file

@ -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(
@ -669,6 +671,8 @@ func (al *AgentLoop) resolveMessageRoute(msg bus.InboundMessage) (routing.Resolv
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)

84
pkg/channels/telegram/target.go Normal file
View file

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

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

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

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

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

191
pkg/channels/telegram/telegram_test.go
View file

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

9
pkg/config/config.go
View file

@ -267,12 +267,21 @@ type TelegramConfig struct {
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 {
Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_FEISHU_ENABLED"`
AppID string `json:"app_id" env:"PICOCLAW_CHANNELS_FEISHU_APP_ID"`

35
pkg/config/config_test.go
View file

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

19
pkg/routing/route.go
View file

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

73
pkg/routing/route_test.go
View file

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

13
pkg/routing/session_key_test.go
View file

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