From b985578c7a9c297a02f8245f0a5db7bdeaf78322 Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Wed, 4 Mar 2026 04:52:54 +0900 Subject: [PATCH] feat(telegram): support thread-aware routing and draft delivery --- pkg/agent/loop.go | 30 +++++- pkg/agent/loop_thread_test.go | 32 ++++++ pkg/channels/telegram/telegram.go | 139 ++++++++++++++++++------- pkg/channels/telegram/telegram_test.go | 50 +++++++++ pkg/config/config.go | 14 +-- pkg/config/defaults.go | 10 +- 6 files changed, 227 insertions(+), 48 deletions(-) create mode 100644 pkg/agent/loop_thread_test.go create mode 100644 pkg/channels/telegram/telegram_test.go diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index dada8987e..06f0ad8c3 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -639,10 +639,15 @@ func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, cha if agent == nil { return "", fmt.Errorf("no default agent for heartbeat") } + heartbeatThreadID := 0 + if al.cfg != nil { + heartbeatThreadID = al.cfg.Channels.Telegram.HeartbeatThreadID + } + heartbeatChatID := al.withTelegramThread(channel, chatID, heartbeatThreadID) return al.runAgentLoop(ctx, agent, processOptions{ SessionKey: "heartbeat", Channel: channel, - ChatID: chatID, + ChatID: heartbeatChatID, UserMessage: content, DefaultResponse: defaultResponse, EnableSummary: false, @@ -848,9 +853,14 @@ func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMe label = label[idx+1:] } notification := formatSubagentCompletion(label, msg.Metadata) + subagentThreadID := 0 + if al.cfg != nil { + subagentThreadID = al.cfg.Channels.Telegram.SubagentThreadID + } + notifyChatID := al.withTelegramThread(originChannel, originChatID, subagentThreadID) _ = al.bus.PublishOutbound(ctx, bus.OutboundMessage{ Channel: originChannel, - ChatID: originChatID, + ChatID: notifyChatID, Content: notification, SkipPlaceholder: true, }) @@ -915,6 +925,22 @@ func formatDurationMs(ms int64) string { return fmt.Sprintf("%dm%ds", mins, sec) } +func (al *AgentLoop) withTelegramThread(channel, chatID string, threadID int) string { + if channel != "telegram" || threadID <= 0 || chatID == "" { + return chatID + } + + baseChatID := chatID + if slash := strings.Index(baseChatID, "/"); slash >= 0 { + baseChatID = baseChatID[:slash] + } + if baseChatID == "" { + return chatID + } + + return fmt.Sprintf("%s/%d", baseChatID, threadID) +} + // acquireSessionLock gets or creates a per-session semaphore and acquires it. // Returns false if the context is canceled before the lock is acquired. func (al *AgentLoop) acquireSessionLock(ctx context.Context, sessionKey string) bool { diff --git a/pkg/agent/loop_thread_test.go b/pkg/agent/loop_thread_test.go new file mode 100644 index 000000000..2e96fdb40 --- /dev/null +++ b/pkg/agent/loop_thread_test.go @@ -0,0 +1,32 @@ +package agent + +import "testing" + +func TestWithTelegramThread(t *testing.T) { + al := &AgentLoop{} + + tests := []struct { + name string + channel string + chatID string + threadID int + want string + }{ + {name: "non telegram unchanged", channel: "discord", chatID: "123", threadID: 7, want: "123"}, + {name: "zero thread unchanged", channel: "telegram", chatID: "123", threadID: 0, want: "123"}, + {name: "negative thread unchanged", channel: "telegram", chatID: "123", threadID: -1, want: "123"}, + {name: "append thread", channel: "telegram", chatID: "123", threadID: 7, want: "123/7"}, + {name: "replace thread", channel: "telegram", chatID: "123/5", threadID: 7, want: "123/7"}, + {name: "group id", channel: "telegram", chatID: "-100123", threadID: 42, want: "-100123/42"}, + {name: "empty chat unchanged", channel: "telegram", chatID: "", threadID: 9, want: ""}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := al.withTelegramThread(tc.channel, tc.chatID, tc.threadID) + if got != tc.want { + t.Fatalf("withTelegramThread(%q, %q, %d) = %q, want %q", tc.channel, tc.chatID, tc.threadID, got, tc.want) + } + }) + } +} diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 9035a082f..44078eeae 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -168,7 +168,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err return channels.ErrNotRunning } - chatID, err := parseChatID(msg.ChatID) + chatID, threadID, err := parseChatID(msg.ChatID) if err != nil { return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) } @@ -178,6 +178,9 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err // Typing/placeholder handled by Manager.preSend — just send the message tgMsg := tu.Message(tu.ID(chatID), htmlContent) tgMsg.ParseMode = telego.ModeHTML + if threadID != 0 { + tgMsg.MessageThreadID = threadID + } if _, err = c.bot.SendMessage(ctx, tgMsg); err != nil { logger.ErrorCF("telegram", "HTML parse failed, falling back to plain text", map[string]any{ @@ -199,7 +202,7 @@ func (c *TelegramChannel) SendWithID(ctx context.Context, chatID string, content return "", channels.ErrNotRunning } - cid, err := parseChatID(chatID) + cid, tid, err := parseChatID(chatID) if err != nil { return "", fmt.Errorf("invalid chat ID %s: %w", chatID, channels.ErrSendFailed) } @@ -207,6 +210,9 @@ func (c *TelegramChannel) SendWithID(ctx context.Context, chatID string, content htmlContent := markdownToTelegramHTML(content) tgMsg := tu.Message(tu.ID(cid), htmlContent) tgMsg.ParseMode = telego.ModeHTML + if tid != 0 { + tgMsg.MessageThreadID = tid + } sent, err := c.bot.SendMessage(ctx, tgMsg) if err != nil { @@ -226,13 +232,17 @@ func (c *TelegramChannel) SendWithID(ctx context.Context, chatID string, content // (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) + cid, tid, err := parseChatID(chatID) if err != nil { return func() {}, err } // Send the first typing action immediately - _ = c.bot.SendChatAction(ctx, tu.ChatAction(tu.ID(cid), telego.ChatActionTyping)) + firstAction := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) + if tid != 0 { + firstAction.MessageThreadID = tid + } + _ = c.bot.SendChatAction(ctx, firstAction) typingCtx, cancel := context.WithCancel(ctx) go func() { @@ -243,7 +253,11 @@ 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)) + action := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) + if tid != 0 { + action.MessageThreadID = tid + } + _ = c.bot.SendChatAction(typingCtx, action) } } }() @@ -253,7 +267,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) + cid, _, err := parseChatID(chatID) if err != nil { return err } @@ -282,12 +296,16 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (s text = "Thinking... 💭" } - cid, err := parseChatID(chatID) + cid, tid, err := parseChatID(chatID) if err != nil { return "", err } - pMsg, err := c.bot.SendMessage(ctx, tu.Message(tu.ID(cid), text)) + params := tu.Message(tu.ID(cid), text) + if tid != 0 { + params.MessageThreadID = tid + } + pMsg, err := c.bot.SendMessage(ctx, params) if err != nil { return "", err } @@ -297,21 +315,25 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (s // SendDraft implements channels.DraftSender. // It uses Telegram Bot API's sendMessageDraft for progressive message streaming -// without the "edited" indicator. Only works in private chats. +// without the "edited" indicator. In groups, draft is used for dedicated topics only. func (c *TelegramChannel) SendDraft(ctx context.Context, chatID string, draftID int, content string) error { if !c.IsRunning() { return channels.ErrNotRunning } - cid, err := parseChatID(chatID) + cid, tid, err := parseChatID(chatID) if err != nil { return fmt.Errorf("invalid chat ID %s: %w", chatID, channels.ErrSendFailed) } + if !isLikelyPrivateChatID(cid) && tid == 0 { + return fmt.Errorf("telegram draft unsupported for non-threaded group chat: %w", channels.ErrSendFailed) + } htmlContent := markdownToTelegramHTML(content) params := &telego.SendMessageDraftParams{ - ChatID: cid, - DraftID: draftID, - Text: htmlContent, - ParseMode: telego.ModeHTML, + ChatID: cid, + MessageThreadID: tid, + DraftID: draftID, + Text: htmlContent, + ParseMode: telego.ModeHTML, } if err = c.bot.SendMessageDraft(ctx, params); err != nil { // HTML parse failure — retry as plain text @@ -328,7 +350,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe return channels.ErrNotRunning } - chatID, err := parseChatID(msg.ChatID) + chatID, threadID, err := parseChatID(msg.ChatID) if err != nil { return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) } @@ -360,30 +382,34 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe switch part.Type { case "image": params := &telego.SendPhotoParams{ - ChatID: tu.ID(chatID), - Photo: telego.InputFile{File: file}, - Caption: part.Caption, + ChatID: tu.ID(chatID), + MessageThreadID: threadID, + Photo: telego.InputFile{File: file}, + Caption: part.Caption, } _, err = c.bot.SendPhoto(ctx, params) case "audio": params := &telego.SendAudioParams{ - ChatID: tu.ID(chatID), - Audio: telego.InputFile{File: file}, - Caption: part.Caption, + ChatID: tu.ID(chatID), + MessageThreadID: threadID, + Audio: telego.InputFile{File: file}, + Caption: part.Caption, } _, err = c.bot.SendAudio(ctx, params) case "video": params := &telego.SendVideoParams{ - ChatID: tu.ID(chatID), - Video: telego.InputFile{File: file}, - Caption: part.Caption, + ChatID: tu.ID(chatID), + MessageThreadID: threadID, + Video: telego.InputFile{File: file}, + Caption: part.Caption, } _, err = c.bot.SendVideo(ctx, params) default: // "file" or unknown types params := &telego.SendDocumentParams{ - ChatID: tu.ID(chatID), - Document: telego.InputFile{File: file}, - Caption: part.Caption, + ChatID: tu.ID(chatID), + MessageThreadID: threadID, + Document: telego.InputFile{File: file}, + Caption: part.Caption, } _, err = c.bot.SendDocument(ctx, params) } @@ -431,11 +457,12 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes chatID := message.Chat.ID c.chatIDs[platformID] = chatID + threadID := message.MessageThreadID content := "" mediaPaths := []string{} - chatIDStr := fmt.Sprintf("%d", chatID) + chatIDStr := formatChatID(chatID, threadID) messageIDStr := fmt.Sprintf("%d", message.MessageID) scope := channels.BuildMediaScope("telegram", chatIDStr, messageIDStr) @@ -528,9 +555,11 @@ 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), - "preview": utils.Truncate(content, 50), + "sender_id": sender.CanonicalID, + "chat_id": fmt.Sprintf("%d", chatID), + "thread_id": threadID, + "chat_route": chatIDStr, + "preview": utils.Truncate(content, 50), }) // Placeholder is now auto-triggered by BaseChannel.HandleMessage via PlaceholderCapable @@ -556,7 +585,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes peer, messageID, platformID, - fmt.Sprintf("%d", chatID), + chatIDStr, content, mediaPaths, metadata, @@ -604,10 +633,48 @@ func (c *TelegramChannel) downloadFile(ctx context.Context, fileID, ext string) return c.downloadFileWithInfo(file, ext) } -func parseChatID(chatIDStr string) (int64, error) { - var id int64 - _, err := fmt.Sscanf(chatIDStr, "%d", &id) - return id, err +func parseChatID(chatIDStr string) (int64, int, error) { + trimmed := strings.TrimSpace(chatIDStr) + if trimmed == "" { + return 0, 0, fmt.Errorf("empty chat ID") + } + + parts := strings.Split(trimmed, "/") + if len(parts) > 2 { + return 0, 0, fmt.Errorf("invalid chat ID format: %q", chatIDStr) + } + + cid, err := strconv.ParseInt(parts[0], 10, 64) + if err != nil { + return 0, 0, fmt.Errorf("invalid chat ID %q: %w", parts[0], err) + } + + tid := 0 + if len(parts) == 2 { + if parts[1] == "" { + return 0, 0, fmt.Errorf("invalid thread ID in %q", chatIDStr) + } + tid, err = strconv.Atoi(parts[1]) + if err != nil { + return 0, 0, fmt.Errorf("invalid thread ID %q: %w", parts[1], err) + } + if tid < 0 { + return 0, 0, fmt.Errorf("thread ID must be non-negative: %d", tid) + } + } + + return cid, tid, nil +} + +func formatChatID(chatID int64, threadID int) string { + if threadID != 0 { + return fmt.Sprintf("%d/%d", chatID, threadID) + } + return fmt.Sprintf("%d", chatID) +} + +func isLikelyPrivateChatID(chatID int64) bool { + return chatID > 0 } func markdownToTelegramHTML(text string) string { diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go new file mode 100644 index 000000000..83588f926 --- /dev/null +++ b/pkg/channels/telegram/telegram_test.go @@ -0,0 +1,50 @@ +package telegram + +import "testing" + +func TestParseChatID(t *testing.T) { + tests := []struct { + name string + input string + wantCID int64 + wantTID int + wantErr bool + }{ + {name: "plain private", input: "12345", wantCID: 12345, wantTID: 0}, + {name: "group topic", input: "-100123/45", wantCID: -100123, wantTID: 45}, + {name: "trim spaces", input: " -100200/7 ", wantCID: -100200, wantTID: 7}, + {name: "topic zero", input: "-100/0", wantCID: -100, wantTID: 0}, + {name: "empty", input: "", wantErr: true}, + {name: "bad chat", input: "abc/def", wantErr: true}, + {name: "missing topic", input: "-100/", wantErr: true}, + {name: "too many parts", input: "-100/1/2", wantErr: true}, + {name: "negative topic", input: "-100/-1", wantErr: true}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + gotCID, gotTID, err := parseChatID(tc.input) + if tc.wantErr { + if err == nil { + t.Fatalf("parseChatID(%q) expected error, got nil", tc.input) + } + return + } + if err != nil { + t.Fatalf("parseChatID(%q) unexpected error: %v", tc.input, err) + } + if gotCID != tc.wantCID || gotTID != tc.wantTID { + t.Fatalf("parseChatID(%q) = (%d, %d), want (%d, %d)", tc.input, gotCID, gotTID, tc.wantCID, tc.wantTID) + } + }) + } +} + +func TestFormatChatID(t *testing.T) { + if got := formatChatID(-100, 42); got != "-100/42" { + t.Fatalf("formatChatID(-100, 42) = %q, want %q", got, "-100/42") + } + if got := formatChatID(12345, 0); got != "12345" { + t.Fatalf("formatChatID(12345, 0) = %q, want %q", got, "12345") + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 8ac57f16c..3dea1d069 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -240,15 +240,17 @@ type WhatsAppConfig struct { } type TelegramConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_TELEGRAM_ENABLED"` - Token string `json:"token" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"` - Proxy string `json:"proxy" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"` - WebAppURL string `json:"web_app_url" env:"PICOCLAW_CHANNELS_TELEGRAM_WEB_APP_URL"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_TELEGRAM_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_TELEGRAM_ENABLED"` + Token string `json:"token" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"` + Proxy string `json:"proxy" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"` + WebAppURL string `json:"web_app_url" env:"PICOCLAW_CHANNELS_TELEGRAM_WEB_APP_URL"` + 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"` + SubagentThreadID int `json:"subagent_thread_id,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_SUBAGENT_THREAD_ID"` + HeartbeatThreadID int `json:"heartbeat_thread_id,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_HEARTBEAT_THREAD_ID"` + ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_TELEGRAM_REASONING_CHANNEL_ID"` } type FeishuConfig struct { diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 22cc2a822..d95f09ed0 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -33,10 +33,12 @@ func DefaultConfig() *Config { AllowFrom: FlexibleStringSlice{}, }, Telegram: TelegramConfig{ - Enabled: false, - Token: "", - AllowFrom: FlexibleStringSlice{}, - Typing: TypingConfig{Enabled: true}, + Enabled: false, + Token: "", + AllowFrom: FlexibleStringSlice{}, + Typing: TypingConfig{Enabled: true}, + SubagentThreadID: 0, + HeartbeatThreadID: 0, Placeholder: PlaceholderConfig{ Enabled: true, Text: "Thinking... 💭",