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

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 {

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

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,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) {
+func parseChatID(chatIDStr string) (int64, int, error) {
-	var id int64
+	trimmed := strings.TrimSpace(chatIDStr)
-	_, err := fmt.Sscanf(chatIDStr, "%d", &id)
+	if trimmed == "" {
-	return id, err
+		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 {

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

pkg/config/config.go

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

pkg/config/defaults.go

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