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

Merge pull request #4 from dimonb/codex/telegram-forum-topics

Browse source 
Add Telegram forum topic routing support
This commit is contained in:
DimonB 2026-03-09 12:32:52 +02:00 • committed by GitHub
commit 13ae906206
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
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"
}
}
}
]
}
```

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

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

27
pkg/config/config.go
View file

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

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