feat(telegram): add reply-to-message threading support

Bot responses are now sent as replies to the user's original message,
creating a natural thread in Telegram chats.

Changes:
- bus/types.go: Add ThreadID to InboundMessage, add ThreadID and
  ReplyToMessageID to OutboundMessage
- channels/interfaces.go: Extend TypingCapable.StartTyping and
  PlaceholderCapable.SendPlaceholder with threadID parameter;
  extend PlaceholderRecorder methods with threadID for scoped tracking
- channels/base.go: Resolve threadID from metadata and pass to
  typing/placeholder/reaction pipelines; populate InboundMessage.ThreadID
- channels/manager.go: Add placeholderKey helper for thread-scoped
  storage keys (channel:chatID:threadID)
- channels/telegram: Set ReplyParameters and MessageThreadID on Send,
  SendPlaceholder, and StartTyping; pass thread_id in metadata
- channels/discord, feishu, pico, irc, line: Accept new parameters
  (unused — no threading support on these platforms)
- agent/loop.go: Thread ReplyToMessageID and ThreadID through
  processOptions and all PublishOutbound calls

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

pkg/agent/loop.go

@@ -52,15 +52,17 @@ type AgentLoop struct {
 
 // processOptions configures how a message is processed
 type processOptions struct {
-	SessionKey      string   // Session identifier for history/context
+	SessionKey       string   // Session identifier for history/context
-	Channel         string   // Target channel for tool execution
+	Channel          string   // Target channel for tool execution
-	ChatID          string   // Target chat ID for tool execution
+	ChatID           string   // Target chat ID for tool execution
-	UserMessage     string   // User message content (may include prefix)
+	UserMessage      string   // User message content (may include prefix)
-	Media           []string // media:// refs from inbound message
+	Media            []string // media:// refs from inbound message
-	DefaultResponse string   // Response when LLM returns empty
+	DefaultResponse  string   // Response when LLM returns empty
-	EnableSummary   bool     // Whether to trigger summarization
+	EnableSummary    bool     // Whether to trigger summarization
-	SendResponse    bool     // Whether to send response via bus
+	SendResponse     bool     // Whether to send response via bus
-	NoHistory       bool     // If true, don't load session history (for heartbeat)
+	NoHistory        bool     // If true, don't load session history (for heartbeat)
+	ReplyToMessageID string   // Platform message ID to reply to (threaded response)
+	ThreadID         string   // Forum topic / thread ID for Telegram forum routing
 }
 
 const (
@@ -351,9 +353,11 @@ func (al *AgentLoop) Run(ctx context.Context) error {
 
 					if !alreadySent {
 						al.bus.PublishOutbound(ctx, bus.OutboundMessage{
-							Channel: msg.Channel,
+							Channel:          msg.Channel,
-							ChatID:  msg.ChatID,
+							ChatID:           msg.ChatID,
-							Content: response,
+							Content:          response,
+							ReplyToMessageID: msg.MessageID,
+							ThreadID:         msg.ThreadID,
 						})
 						logger.InfoCF("agent", "Published outbound response",
 							map[string]any{
@@ -616,14 +620,16 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
 		})
 
 	return al.runAgentLoop(ctx, agent, processOptions{
-		SessionKey:      sessionKey,
+		SessionKey:       sessionKey,
-		Channel:         msg.Channel,
+		Channel:          msg.Channel,
-		ChatID:          msg.ChatID,
+		ChatID:           msg.ChatID,
-		UserMessage:     msg.Content,
+		UserMessage:      msg.Content,
-		Media:           msg.Media,
+		Media:            msg.Media,
-		DefaultResponse: defaultResponse,
+		DefaultResponse:  defaultResponse,
-		EnableSummary:   true,
+		EnableSummary:    true,
-		SendResponse:    false,
+		SendResponse:     false,
+		ThreadID:         msg.ThreadID,
+		ReplyToMessageID: msg.MessageID,
 	})
 }
 
@@ -789,9 +795,11 @@ func (al *AgentLoop) runAgentLoop(
 	// 7. Optional: send response via bus
 	if opts.SendResponse {
 		al.bus.PublishOutbound(ctx, bus.OutboundMessage{
-			Channel: opts.Channel,
+			Channel:          opts.Channel,
-			ChatID:  opts.ChatID,
+			ChatID:           opts.ChatID,
-			Content: finalContent,
+			Content:          finalContent,
+			ReplyToMessageID: opts.ReplyToMessageID,
+			ThreadID:         opts.ThreadID,
 		})
 	}
 
@@ -1010,9 +1018,10 @@ func (al *AgentLoop) runLLMIteration(
 
 				if retry == 0 && !constants.IsInternalChannel(opts.Channel) {
 					al.bus.PublishOutbound(ctx, bus.OutboundMessage{
-						Channel: opts.Channel,
+						Channel:  opts.Channel,
-						ChatID:  opts.ChatID,
+						ChatID:   opts.ChatID,
-						Content: "Context window exceeded. Compressing history and retrying...",
+						Content:  "Context window exceeded. Compressing history and retrying...",
+						ThreadID: opts.ThreadID,
 					})
 				}
 
@@ -1207,9 +1216,10 @@ func (al *AgentLoop) runLLMIteration(
 			// Send ForUser content to user immediately if not Silent
 			if !r.result.Silent && r.result.ForUser != "" && opts.SendResponse {
 				al.bus.PublishOutbound(ctx, bus.OutboundMessage{
-					Channel: opts.Channel,
+					Channel:  opts.Channel,
-					ChatID:  opts.ChatID,
+					ChatID:   opts.ChatID,
-					Content: r.result.ForUser,
+					Content:  r.result.ForUser,
+					ThreadID: opts.ThreadID,
 				})
 				logger.DebugCF("agent", "Sent tool result to user",
 					map[string]any{

pkg/bus/types.go

@@ -24,15 +24,18 @@ type InboundMessage struct {
 	Media      []string          `json:"media,omitempty"`
 	Peer       Peer              `json:"peer"`                  // routing peer
 	MessageID  string            `json:"message_id,omitempty"`  // platform message ID
+	ThreadID   string            `json:"thread_id,omitempty"`   // forum topic / thread ID
 	MediaScope string            `json:"media_scope,omitempty"` // media lifecycle scope
 	SessionKey string            `json:"session_key"`
 	Metadata   map[string]string `json:"metadata,omitempty"`
 }
 
 type OutboundMessage struct {
-	Channel string `json:"channel"`
+	Channel          string `json:"channel"`
-	ChatID  string `json:"chat_id"`
+	ChatID           string `json:"chat_id"`
-	Content string `json:"content"`
+	Content          string `json:"content"`
+	ThreadID         string `json:"thread_id,omitempty"`           // forum topic / message thread ID
+	ReplyToMessageID string `json:"reply_to_message_id,omitempty"` // platform message ID to reply to
 }
 
 // MediaPart describes a single media attachment to send.

pkg/channels/base.go

@@ -256,6 +256,12 @@ func (c *BaseChannel) HandleMessage(
 
 	scope := BuildMediaScope(c.name, chatID, messageID)
 
+	// Resolve thread ID from metadata (set by channels that support threading).
+	threadID := ""
+	if metadata != nil {
+		threadID = metadata["thread_id"]
+	}
+
 	msg := bus.InboundMessage{
 		Channel:    c.name,
 		SenderID:   resolvedSenderID,
@@ -265,6 +271,7 @@ func (c *BaseChannel) HandleMessage(
 		Media:      media,
 		Peer:       peer,
 		MessageID:  messageID,
+		ThreadID:   threadID,
 		MediaScope: scope,
 		Metadata:   metadata,
 	}
@@ -274,20 +281,20 @@ func (c *BaseChannel) HandleMessage(
 	if c.owner != nil && c.placeholderRecorder != nil {
 		// Typing — independent pipeline
 		if tc, ok := c.owner.(TypingCapable); ok {
-			if stop, err := tc.StartTyping(ctx, chatID); err == nil {
+			if stop, err := tc.StartTyping(ctx, chatID, threadID); err == nil {
-				c.placeholderRecorder.RecordTypingStop(c.name, chatID, stop)
+				c.placeholderRecorder.RecordTypingStop(c.name, chatID, threadID, stop)
 			}
 		}
 		// Reaction — independent pipeline
 		if rc, ok := c.owner.(ReactionCapable); ok && messageID != "" {
 			if undo, err := rc.ReactToMessage(ctx, chatID, messageID); err == nil {
-				c.placeholderRecorder.RecordReactionUndo(c.name, chatID, undo)
+				c.placeholderRecorder.RecordReactionUndo(c.name, chatID, threadID, undo)
 			}
 		}
 		// Placeholder — independent pipeline
 		if pc, ok := c.owner.(PlaceholderCapable); ok {
-			if phID, err := pc.SendPlaceholder(ctx, chatID); err == nil && phID != "" {
+			if phID, err := pc.SendPlaceholder(ctx, chatID, threadID, messageID); err == nil && phID != "" {
-				c.placeholderRecorder.RecordPlaceholder(c.name, chatID, phID)
+				c.placeholderRecorder.RecordPlaceholder(c.name, chatID, threadID, phID)
 			}
 		}
 	}

pkg/channels/discord/discord.go

@@ -241,7 +241,7 @@ func (c *DiscordChannel) EditMessage(ctx context.Context, chatID string, message
 // SendPlaceholder implements channels.PlaceholderCapable.
 // It sends a placeholder message that will later be edited to the actual
 // response via EditMessage (channels.MessageEditor).
-func (c *DiscordChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) {
+func (c *DiscordChannel) SendPlaceholder(ctx context.Context, chatID string, _ string, _ string) (string, error) {
 	if !c.config.Placeholder.Enabled {
 		return "", nil
 	}
@@ -488,7 +488,7 @@ func (c *DiscordChannel) stopTyping(chatID string) {
 
 // StartTyping implements channels.TypingCapable.
 // It starts a continuous typing indicator and returns an idempotent stop function.
-func (c *DiscordChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
+func (c *DiscordChannel) StartTyping(ctx context.Context, chatID string, _ string) (func(), error) {
 	c.startTyping(chatID)
 	return func() { c.stopTyping(chatID) }, nil
 }

pkg/channels/feishu/feishu_32.go

@@ -46,7 +46,7 @@ func (c *FeishuChannel) EditMessage(ctx context.Context, chatID, messageID, cont
 }
 
 // SendPlaceholder is a stub method to satisfy PlaceholderCapable
-func (c *FeishuChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) {
+func (c *FeishuChannel) SendPlaceholder(ctx context.Context, chatID string, _ string, _ string) (string, error) {
 	return "", errUnsupported
 }
 

pkg/channels/feishu/feishu_64.go

@@ -153,7 +153,7 @@ func (c *FeishuChannel) EditMessage(ctx context.Context, chatID, messageID, cont
 
 // SendPlaceholder implements channels.PlaceholderCapable.
 // Sends an interactive card with placeholder text and returns its message ID.
-func (c *FeishuChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) {
+func (c *FeishuChannel) SendPlaceholder(ctx context.Context, chatID string, _ string, _ string) (string, error) {
 	if !c.config.Placeholder.Enabled {
 		logger.DebugCF("feishu", "Placeholder disabled, skipping", map[string]any{
 			"chat_id": chatID,

pkg/channels/interfaces.go

@@ -9,8 +9,9 @@ import (
 // TypingCapable — channels that can show a typing/thinking indicator.
 // StartTyping begins the indicator and returns a stop function.
 // The stop function MUST be idempotent and safe to call multiple times.
+// When threadID is non-empty it routes the typing action to the given forum topic.
 type TypingCapable interface {
-	StartTyping(ctx context.Context, chatID string) (stop func(), err error)
+	StartTyping(ctx context.Context, chatID string, threadID string) (stop func(), err error)
 }
 
 // MessageEditor — channels that can edit an existing message.
@@ -32,16 +33,16 @@ type ReactionCapable interface {
 // SendPlaceholder returns the platform message ID of the placeholder so that
 // Manager.preSend can later edit it via MessageEditor.EditMessage.
 type PlaceholderCapable interface {
-	SendPlaceholder(ctx context.Context, chatID string) (messageID string, err error)
+	SendPlaceholder(ctx context.Context, chatID string, threadID string, replyToMsgID string) (messageID string, err error)
 }
 
 // PlaceholderRecorder is injected into channels by Manager.
 // Channels call these methods on inbound to register typing/placeholder state.
 // Manager uses the registered state on outbound to stop typing and edit placeholders.
 type PlaceholderRecorder interface {
-	RecordPlaceholder(channel, chatID, placeholderID string)
+	RecordPlaceholder(channel, chatID, threadID, placeholderID string)
-	RecordTypingStop(channel, chatID string, stop func())
+	RecordTypingStop(channel, chatID, threadID string, stop func())
-	RecordReactionUndo(channel, chatID string, undo func())
+	RecordReactionUndo(channel, chatID, threadID string, undo func())
 }
 
 // CommandRegistrarCapable is implemented by channels that can register

pkg/channels/irc/irc.go

@@ -163,7 +163,7 @@ func (c *IRCChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
 
 // StartTyping implements channels.TypingCapable using IRCv3 +typing client tag.
 // Requires typing.enabled in config and server support for message-tags capability.
-func (c *IRCChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
+func (c *IRCChannel) StartTyping(ctx context.Context, chatID string, _ string) (func(), error) {
 	noop := func() {}
 
 	if !c.config.Typing.Enabled || !c.IsRunning() || c.conn == nil {

pkg/channels/line/line.go

@@ -583,7 +583,7 @@ func (c *LINEChannel) sendPush(ctx context.Context, to, content, quoteToken stri
 // NOTE: The LINE loading animation API only works for 1:1 chats.
 // Group/room chat IDs (starting with "C" or "R") are detected automatically;
 // for these, a no-op stop function is returned without calling the API.
-func (c *LINEChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
+func (c *LINEChannel) StartTyping(ctx context.Context, chatID string, _ string) (func(), error) {
 	if chatID == "" {
 		return func() {}, nil
 	}

pkg/channels/manager.go

@@ -93,31 +93,35 @@ type asyncTask struct {
 	cancel context.CancelFunc
 }
 
+func placeholderKey(channel, chatID, threadID string) string {
+	return channel + ":" + chatID + ":" + threadID
+}
+
 // RecordPlaceholder registers a placeholder message for later editing.
 // Implements PlaceholderRecorder.
-func (m *Manager) RecordPlaceholder(channel, chatID, placeholderID string) {
+func (m *Manager) RecordPlaceholder(channel, chatID, threadID, placeholderID string) {
-	key := channel + ":" + chatID
+	key := placeholderKey(channel, chatID, threadID)
 	m.placeholders.Store(key, placeholderEntry{id: placeholderID, createdAt: time.Now()})
 }
 
 // RecordTypingStop registers a typing stop function for later invocation.
 // Implements PlaceholderRecorder.
-func (m *Manager) RecordTypingStop(channel, chatID string, stop func()) {
+func (m *Manager) RecordTypingStop(channel, chatID, threadID string, stop func()) {
-	key := channel + ":" + chatID
+	key := placeholderKey(channel, chatID, threadID)
 	m.typingStops.Store(key, typingEntry{stop: stop, createdAt: time.Now()})
 }
 
 // RecordReactionUndo registers a reaction undo function for later invocation.
 // Implements PlaceholderRecorder.
-func (m *Manager) RecordReactionUndo(channel, chatID string, undo func()) {
+func (m *Manager) RecordReactionUndo(channel, chatID, threadID string, undo func()) {
-	key := channel + ":" + chatID
+	key := placeholderKey(channel, chatID, threadID)
 	m.reactionUndos.Store(key, reactionEntry{undo: undo, createdAt: time.Now()})
 }
 
 // preSend handles typing stop, reaction undo, and placeholder editing before sending a message.
 // Returns true if the message was edited into a placeholder (skip Send).
 func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMessage, ch Channel) bool {
-	key := name + ":" + msg.ChatID
+	key := placeholderKey(name, msg.ChatID, msg.ThreadID)
 
 	// 1. Stop typing
 	if v, loaded := m.typingStops.LoadAndDelete(key); loaded {

pkg/channels/manager_test.go

@@ -455,7 +455,7 @@ func TestPreSend_PlaceholderEditSuccess(t *testing.T) {
 	}
 
 	// Register placeholder
-	m.RecordPlaceholder("test", "123", "456")
+	m.RecordPlaceholder("test", "123", "", "456")
 
 	msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello"}
 	edited := m.preSend(context.Background(), "test", msg, ch)
@@ -485,7 +485,7 @@ func TestPreSend_PlaceholderEditFails_FallsThrough(t *testing.T) {
 		},
 	}
 
-	m.RecordPlaceholder("test", "123", "456")
+	m.RecordPlaceholder("test", "123", "", "456")
 
 	msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello"}
 	edited := m.preSend(context.Background(), "test", msg, ch)
@@ -505,7 +505,7 @@ func TestPreSend_TypingStopCalled(t *testing.T) {
 		},
 	}
 
-	m.RecordTypingStop("test", "123", func() {
+	m.RecordTypingStop("test", "123", "", func() {
 		stopCalled = true
 	})
 
@@ -551,10 +551,10 @@ func TestPreSend_TypingAndPlaceholder(t *testing.T) {
 		},
 	}
 
-	m.RecordTypingStop("test", "123", func() {
+	m.RecordTypingStop("test", "123", "", func() {
 		stopCalled = true
 	})
-	m.RecordPlaceholder("test", "123", "456")
+	m.RecordPlaceholder("test", "123", "", "456")
 
 	msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello"}
 	edited := m.preSend(context.Background(), "test", msg, ch)
@@ -579,7 +579,7 @@ func TestRecordPlaceholder_ConcurrentSafe(t *testing.T) {
 		go func(i int) {
 			defer wg.Done()
 			chatID := fmt.Sprintf("chat_%d", i%10)
-			m.RecordPlaceholder("test", chatID, fmt.Sprintf("msg_%d", i))
+			m.RecordPlaceholder("test", chatID, "", fmt.Sprintf("msg_%d", i))
 		}(i)
 	}
 	wg.Wait()
@@ -594,7 +594,7 @@ func TestRecordTypingStop_ConcurrentSafe(t *testing.T) {
 		go func(i int) {
 			defer wg.Done()
 			chatID := fmt.Sprintf("chat_%d", i%10)
-			m.RecordTypingStop("test", chatID, func() {})
+			m.RecordTypingStop("test", chatID, "", func() {})
 		}(i)
 	}
 	wg.Wait()
@@ -616,7 +616,7 @@ func TestSendWithRetry_PreSendEditsPlaceholder(t *testing.T) {
 		},
 	}
 
-	m.RecordPlaceholder("test", "123", "456")
+	m.RecordPlaceholder("test", "123", "", "456")
 
 	w := &channelWorker{
 		ch:      ch,
@@ -781,10 +781,10 @@ func TestPreSendStillWorksWithWrappedTypes(t *testing.T) {
 	}
 
 	// Use the new wrapped types via the public API
-	m.RecordTypingStop("test", "chat1", func() {
+	m.RecordTypingStop("test", "chat1", "", func() {
 		stopCalled = true
 	})
-	m.RecordPlaceholder("test", "chat1", "ph_id")
+	m.RecordPlaceholder("test", "chat1", "", "ph_id")
 
 	msg := bus.OutboundMessage{Channel: "test", ChatID: "chat1", Content: "response"}
 	edited := m.preSend(context.Background(), "test", msg, ch)

pkg/channels/pico/pico.go

@@ -160,7 +160,7 @@ func (c *PicoChannel) EditMessage(ctx context.Context, chatID string, messageID
 }
 
 // StartTyping implements channels.TypingCapable.
-func (c *PicoChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
+func (c *PicoChannel) StartTyping(ctx context.Context, chatID string, _ string) (func(), error) {
 	startMsg := newMessage(TypeTypingStart, nil)
 	if err := c.broadcastToSession(chatID, startMsg); err != nil {
 		return func() {}, err
@@ -174,7 +174,7 @@ func (c *PicoChannel) StartTyping(ctx context.Context, chatID string) (func(), e
 // SendPlaceholder implements channels.PlaceholderCapable.
 // It sends a placeholder message via the Pico Protocol that will later be
 // edited to the actual response via EditMessage (channels.MessageEditor).
-func (c *PicoChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) {
+func (c *PicoChannel) SendPlaceholder(ctx context.Context, chatID string, _ string, _ string) (string, error) {
 	if !c.config.Placeholder.Enabled {
 		return "", nil
 	}

pkg/channels/telegram/telegram.go

@@ -200,7 +200,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
 			continue
 		}
 
-		if err := c.sendHTMLChunk(ctx, chatID, htmlContent, chunk); err != nil {
+		if err := c.sendHTMLChunk(ctx, chatID, htmlContent, chunk, msg.ThreadID, msg.ReplyToMessageID); err != nil {
 			return err
 		}
 	}
@@ -210,9 +210,19 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
 
 // 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) error {
+func (c *TelegramChannel) sendHTMLChunk(ctx context.Context, chatID int64, htmlContent, mdFallback, threadID, replyToMsgID string) error {
 	tgMsg := tu.Message(tu.ID(chatID), htmlContent)
 	tgMsg.ParseMode = telego.ModeHTML
+	if threadID != "" {
+		if tid, err2 := strconv.Atoi(threadID); err2 == nil {
+			tgMsg.MessageThreadID = tid
+		}
+	}
+	if replyToMsgID != "" {
+		if mid, err2 := strconv.Atoi(replyToMsgID); err2 == nil {
+			tgMsg.ReplyParameters = &telego.ReplyParameters{MessageID: mid}
+		}
+	}
 
 	if _, err := c.bot.SendMessage(ctx, tgMsg); err != nil {
 		logger.ErrorCF("telegram", "HTML parse failed, falling back to plain text", map[string]any{
@@ -231,14 +241,26 @@ func (c *TelegramChannel) sendHTMLChunk(ctx context.Context, chatID int64, htmlC
 // It sends ChatAction(typing) immediately and then repeats every 4 seconds
 // (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) {
+// When threadID is non-empty it routes the action to the given forum topic.
+func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string, threadID string) (func(), error) {
 	cid, err := parseChatID(chatID)
 	if err != nil {
 		return func() {}, err
 	}
 
+	tid := 0
+	if threadID != "" {
+		tid, _ = strconv.Atoi(threadID)
+	}
+
+	sendTyping := func(sctx context.Context) {
+		params := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping)
+		params.MessageThreadID = tid
+		_ = c.bot.SendChatAction(sctx, 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() {
@@ -249,7 +271,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)
 			}
 		}
 	}()
@@ -277,7 +299,9 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag
 // SendPlaceholder implements channels.PlaceholderCapable.
 // 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) {
+// When threadID is non-empty the placeholder is sent into the given forum topic.
+// When replyToMsgID is non-empty the placeholder is sent as a reply to that message.
+func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string, threadID string, replyToMsgID string) (string, error) {
 	phCfg := c.config.Channels.Telegram.Placeholder
 	if !phCfg.Enabled {
 		return "", nil
@@ -293,7 +317,19 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (s
 		return "", err
 	}
 
-	pMsg, err := c.bot.SendMessage(ctx, tu.Message(tu.ID(cid), text))
+	params := tu.Message(tu.ID(cid), text)
+	if threadID != "" {
+		if tid, err2 := strconv.Atoi(threadID); err2 == nil {
+			params.MessageThreadID = tid
+		}
+	}
+	if replyToMsgID != "" {
+		if mid, err2 := strconv.Atoi(replyToMsgID); err2 == nil {
+			params.ReplyParameters = &telego.ReplyParameters{MessageID: mid}
+		}
+	}
+
+	pMsg, err := c.bot.SendMessage(ctx, params)
 	if err != nil {
 		return "", err
 	}
@@ -530,6 +566,9 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
 		"first_name": user.FirstName,
 		"is_group":   fmt.Sprintf("%t", message.Chat.Type != "private"),
 	}
+	if message.MessageThreadID != 0 {
+		metadata["thread_id"] = fmt.Sprintf("%d", message.MessageThreadID)
+	}
 
 	c.HandleMessage(c.ctx,
 		peer,