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

feat(telegram): support thread-aware routing and draft delivery

Browse source
This commit is contained in:
dj-oyu 2026-03-04 04:52:54 +09:00
parent 8144a1d197
commit fa5a97ca47
6 changed files with 227 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
pkg/agent/loop.go
View file

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

32
pkg/agent/loop_thread_test.go Normal file
View file

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

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

@ -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,18 +315,22 @@ 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,
MessageThreadID: tid,
DraftID: draftID,
Text: htmlContent,
ParseMode: telego.ModeHTML,
@ -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)
}
@ -361,6 +383,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
case "image":
params := &telego.SendPhotoParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Photo: telego.InputFile{File: file},
Caption: part.Caption,
}
@ -368,6 +391,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
case "audio":
params := &telego.SendAudioParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Audio: telego.InputFile{File: file},
Caption: part.Caption,
}
@ -375,6 +399,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
case "video":
params := &telego.SendVideoParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Video: telego.InputFile{File: file},
Caption: part.Caption,
}
@ -382,6 +407,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
default: // "file" or unknown types
params := &telego.SendDocumentParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Document: telego.InputFile{File: file},
Caption: part.Caption,
}
@ -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)
@ -530,6 +557,8 @@ 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),
"thread_id": threadID,
"chat_route": chatIDStr,
"preview": utils.Truncate(content, 50),
})
@ -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 {

50
pkg/channels/telegram/telegram_test.go Normal file
View file

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

2
pkg/config/config.go
View file

@ -248,6 +248,8 @@ type TelegramConfig struct {
GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"`
Typing TypingConfig `json:"typing,omitempty"`
Placeholder PlaceholderConfig `json:"placeholder,omitempty"`
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"`
}

2
pkg/config/defaults.go
View file

@ -37,6 +37,8 @@ func DefaultConfig() *Config {
Token: "",
AllowFrom: FlexibleStringSlice{},
Typing: TypingConfig{Enabled: true},
SubagentThreadID: 0,
HeartbeatThreadID: 0,
Placeholder: PlaceholderConfig{
Enabled: true,
Text: "Thinking... 💭",