diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 9a54f5077..f731d4e5a 100644 --- a/pkg/agent/loop.go +++ b/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 - Channel string // Target channel for tool execution - ChatID string // Target chat ID for tool execution - UserMessage string // User message content (may include prefix) - Media []string // media:// refs from inbound message - DefaultResponse string // Response when LLM returns empty - EnableSummary bool // Whether to trigger summarization - SendResponse bool // Whether to send response via bus - NoHistory bool // If true, don't load session history (for heartbeat) + SessionKey string // Session identifier for history/context + Channel string // Target channel for tool execution + ChatID string // Target chat ID for tool execution + UserMessage string // User message content (may include prefix) + Media []string // media:// refs from inbound message + DefaultResponse string // Response when LLM returns empty + EnableSummary bool // Whether to trigger summarization + SendResponse bool // Whether to send response via bus + 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, - ChatID: msg.ChatID, - Content: response, + Channel: msg.Channel, + ChatID: msg.ChatID, + 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, - Channel: msg.Channel, - ChatID: msg.ChatID, - UserMessage: msg.Content, - Media: msg.Media, - DefaultResponse: defaultResponse, - EnableSummary: true, - SendResponse: false, + SessionKey: sessionKey, + Channel: msg.Channel, + ChatID: msg.ChatID, + UserMessage: msg.Content, + Media: msg.Media, + DefaultResponse: defaultResponse, + EnableSummary: true, + 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, - ChatID: opts.ChatID, - Content: finalContent, + Channel: opts.Channel, + ChatID: opts.ChatID, + 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, - ChatID: opts.ChatID, - Content: "Context window exceeded. Compressing history and retrying...", + Channel: opts.Channel, + ChatID: opts.ChatID, + 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, - ChatID: opts.ChatID, - Content: r.result.ForUser, + Channel: opts.Channel, + ChatID: opts.ChatID, + Content: r.result.ForUser, + ThreadID: opts.ThreadID, }) logger.DebugCF("agent", "Sent tool result to user", map[string]any{ diff --git a/pkg/bus/types.go b/pkg/bus/types.go index 7ad8f0417..2fad31356 100644 --- a/pkg/bus/types.go +++ b/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"` - ChatID string `json:"chat_id"` - Content string `json:"content"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + 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. diff --git a/pkg/channels/base.go b/pkg/channels/base.go index 063a66523..360e3d70e 100644 --- a/pkg/channels/base.go +++ b/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 { - c.placeholderRecorder.RecordTypingStop(c.name, chatID, stop) + if stop, err := tc.StartTyping(ctx, chatID, threadID); err == nil { + 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 != "" { - c.placeholderRecorder.RecordPlaceholder(c.name, chatID, phID) + if phID, err := pc.SendPlaceholder(ctx, chatID, threadID, messageID); err == nil && phID != "" { + c.placeholderRecorder.RecordPlaceholder(c.name, chatID, threadID, phID) } } } diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index c3bcbff8d..ce8a13baf 100644 --- a/pkg/channels/discord/discord.go +++ b/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 } diff --git a/pkg/channels/feishu/feishu_32.go b/pkg/channels/feishu/feishu_32.go index f5e3aa224..27c93c5b1 100644 --- a/pkg/channels/feishu/feishu_32.go +++ b/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 } diff --git a/pkg/channels/feishu/feishu_64.go b/pkg/channels/feishu/feishu_64.go index 00f73064d..399f09b41 100644 --- a/pkg/channels/feishu/feishu_64.go +++ b/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, diff --git a/pkg/channels/interfaces.go b/pkg/channels/interfaces.go index b3a493761..0aa4a89d0 100644 --- a/pkg/channels/interfaces.go +++ b/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) - RecordTypingStop(channel, chatID string, stop func()) - RecordReactionUndo(channel, chatID string, undo func()) + RecordPlaceholder(channel, chatID, threadID, placeholderID string) + RecordTypingStop(channel, chatID, threadID string, stop func()) + RecordReactionUndo(channel, chatID, threadID string, undo func()) } // CommandRegistrarCapable is implemented by channels that can register diff --git a/pkg/channels/irc/irc.go b/pkg/channels/irc/irc.go index 28c59b540..b613cb52b 100644 --- a/pkg/channels/irc/irc.go +++ b/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 { diff --git a/pkg/channels/line/line.go b/pkg/channels/line/line.go index b36350a06..a53a79241 100644 --- a/pkg/channels/line/line.go +++ b/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 } diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 2b1cf8e84..50dd24d8f 100644 --- a/pkg/channels/manager.go +++ b/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) { - key := channel + ":" + chatID +func (m *Manager) RecordPlaceholder(channel, chatID, threadID, placeholderID string) { + 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()) { - key := channel + ":" + chatID +func (m *Manager) RecordTypingStop(channel, chatID, threadID string, stop func()) { + 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()) { - key := channel + ":" + chatID +func (m *Manager) RecordReactionUndo(channel, chatID, threadID string, undo func()) { + 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 { diff --git a/pkg/channels/manager_test.go b/pkg/channels/manager_test.go index f09ecfe2f..201b6b00a 100644 --- a/pkg/channels/manager_test.go +++ b/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) diff --git a/pkg/channels/pico/pico.go b/pkg/channels/pico/pico.go index 8d8b62a67..d5069989b 100644 --- a/pkg/channels/pico/pico.go +++ b/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 } diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 0a36247a6..ba2edfbfe 100644 --- a/pkg/channels/telegram/telegram.go +++ b/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,