From 19d91f51cefb068bb6e06cdec2e5042b0bfd88ad Mon Sep 17 00:00:00 2001 From: Aleksey Genus Date: Mon, 3 Aug 2026 22:36:41 +0200 Subject: [PATCH] fix(telegram): support topics in private bot chats --- pkg/channels/telegram/telegram.go | 9 +-- pkg/channels/telegram/telegram_test.go | 91 ++++++++++++++++++++++++++ pkg/session/allocator_test.go | 33 ++++++++++ 3 files changed, 127 insertions(+), 6 deletions(-) diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index d7265fb3a1..6e88d5a2d4 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -1153,13 +1153,10 @@ func (c *TelegramChannel) handleMessages(ctx context.Context, messages []*telego content = c.prependTelegramQuotedReply(content, message.ReplyToMessage) } - // For forum topics, embed the thread ID as "chatID/threadID" so replies - // route to the correct topic and each topic gets its own session. - // Only forum groups (IsForum) are handled; regular group reply threads - // must share one session per group. compositeChatID := fmt.Sprintf("%d", chatID) threadID := message.MessageThreadID - if message.Chat.IsForum && threadID != 0 { + isTopic := threadID != 0 && (message.Chat.IsForum || message.IsTopicMessage) + if isTopic { compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID) } @@ -1192,7 +1189,7 @@ func (c *TelegramChannel) handleMessages(ctx context.Context, messages []*telego Mentioned: isMentioned, Raw: metadata, } - if message.Chat.IsForum && threadID != 0 { + if isTopic { inboundCtx.TopicID = fmt.Sprintf("%d", threadID) } if message.ReplyToMessage != nil { diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index 19b181cf27..0070b86bf6 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -1014,6 +1014,33 @@ func TestSend_WithForumThreadID(t *testing.T) { assert.Len(t, caller.calls, 1) } +func TestSend_WithPrivateTopicThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + _, err := ch.Send(context.Background(), bus.OutboundMessage{ + ChatID: "123456789/42", + Content: "Hello from private topic", + }) + + require.NoError(t, err) + require.Len(t, caller.calls, 1) + + var params struct { + ChatID int64 `json:"chat_id"` + MessageThreadID int `json:"message_thread_id"` + Text string `json:"text"` + } + require.NoError(t, json.Unmarshal(caller.calls[0].Data.BodyRaw, ¶ms)) + assert.Equal(t, int64(123456789), params.ChatID) + assert.Equal(t, 42, params.MessageThreadID) + assert.Equal(t, "Hello from private topic", params.Text) +} + func TestSend_UsesContextTopicIDWhenChatIDDoesNotIncludeThread(t *testing.T) { caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { @@ -1248,6 +1275,70 @@ func TestHandleMessage_ForumTopic_SetsMetadata(t *testing.T) { assert.Equal(t, "42", inbound.Context.TopicID) } +func TestHandleMessage_PrivateTopic_SetsMetadata(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &TelegramChannel{ + BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), + chatIDs: make(map[string]int64), + ctx: context.Background(), + } + + msg := &telego.Message{ + Text: "hello from private topic", + MessageID: 12, + MessageThreadID: 42, + IsTopicMessage: true, + Chat: telego.Chat{ + ID: 123456789, + Type: "private", + }, + From: &telego.User{ + ID: 7, + FirstName: "Alice", + }, + } + + err := ch.handleMessage(context.Background(), msg) + require.NoError(t, err) + + inbound, ok := <-messageBus.InboundChan() + require.True(t, ok) + assert.Equal(t, "123456789/42", inbound.ChatID) + assert.Equal(t, "direct", inbound.Context.ChatType) + assert.Equal(t, "42", inbound.Context.TopicID) +} + +func TestHandleMessage_PrivateChat_NoTopicMetadata(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &TelegramChannel{ + BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), + chatIDs: make(map[string]int64), + ctx: context.Background(), + } + + msg := &telego.Message{ + Text: "hello from private chat", + MessageID: 13, + Chat: telego.Chat{ + ID: 123456789, + Type: "private", + }, + From: &telego.User{ + ID: 7, + FirstName: "Alice", + }, + } + + err := ch.handleMessage(context.Background(), msg) + require.NoError(t, err) + + inbound, ok := <-messageBus.InboundChan() + require.True(t, ok) + assert.Equal(t, "123456789", inbound.ChatID) + assert.Equal(t, "direct", inbound.Context.ChatType) + assert.Empty(t, inbound.Context.TopicID) +} + func TestHandleMessage_NoForum_NoThreadMetadata(t *testing.T) { messageBus := bus.NewMessageBus() ch := &TelegramChannel{ diff --git a/pkg/session/allocator_test.go b/pkg/session/allocator_test.go index 9750ffc39c..590c25a6ad 100644 --- a/pkg/session/allocator_test.go +++ b/pkg/session/allocator_test.go @@ -119,6 +119,39 @@ func TestAllocateRouteSession_TelegramForumTopicsRemainIsolatedByDefault(t *test } } +func TestAllocateRouteSession_TelegramPrivateTopicsRemainIsolatedByDefault(t *testing.T) { + first := AllocateRouteSession(AllocationInput{ + AgentID: "main", + Context: bus.InboundContext{ + Channel: "telegram", + ChatID: "123456789/42", + ChatType: "direct", + TopicID: "42", + SenderID: "7", + }, + SessionPolicy: routing.SessionPolicy{ + Dimensions: []string{"chat"}, + }, + }) + second := AllocateRouteSession(AllocationInput{ + AgentID: "main", + Context: bus.InboundContext{ + Channel: "telegram", + ChatID: "123456789/99", + ChatType: "direct", + TopicID: "99", + SenderID: "7", + }, + SessionPolicy: routing.SessionPolicy{ + Dimensions: []string{"chat"}, + }, + }) + + if first.SessionKey == second.SessionKey { + t.Fatalf("private topics should not share default session key: %q", first.SessionKey) + } +} + func TestAllocateRouteSession_PicoDirectAliasesIncludeLegacyChatKey(t *testing.T) { allocation := AllocateRouteSession(AllocationInput{ AgentID: "main",