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

Merge pull request #1 from stolyarchuk/telegram-business-mode

Browse source 
Telegram business mode
This commit is contained in:
Roman Stolyarchuk 2026-05-10 09:11:19 +03:00 • committed by GitHub
commit f3507b375b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 516 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -43,6 +43,31 @@ var (
reInlineCode = regexp.MustCompile("`([^`]+)`")
)
func telegramAllowedUpdates(businessMode bool) []string {
updates := []string{
telego.MessageUpdates,
telego.EditedMessageUpdates,
telego.ChannelPostUpdates,
telego.EditedChannelPostUpdates,
telego.CallbackQueryUpdates,
telego.InlineQueryUpdates,
telego.ChosenInlineResultUpdates,
telego.ShippingQueryUpdates,
telego.PreCheckoutQueryUpdates,
telego.PollUpdates,
telego.PollAnswerUpdates,
}
if businessMode {
updates = append(updates,
telego.BusinessConnectionUpdates,
telego.BusinessMessageUpdates,
telego.EditedBusinessMessageUpdates,
telego.DeletedBusinessMessagesUpdates,
)
}
return updates
}
type TelegramChannel struct {
*channels.BaseChannel
bot *telego.Bot
@ -123,7 +148,8 @@ func (c *TelegramChannel) Start(ctx context.Context) error {
c.ctx, c.cancel = context.WithCancel(ctx)
updates, err := c.bot.UpdatesViaLongPolling(c.ctx, &telego.GetUpdatesParams{
Timeout: 30,
Timeout: 30,
AllowedUpdates: telegramAllowedUpdates(c.businessModeEnabled()),
})
if err != nil {
c.cancel()
@ -140,6 +166,20 @@ func (c *TelegramChannel) Start(ctx context.Context) error {
bh.HandleMessage(func(ctx *th.Context, message telego.Message) error {
return c.handleMessage(ctx, &message)
}, th.AnyMessage())
if c.businessModeEnabled() {
bh.HandleBusinessMessage(func(ctx *th.Context, message telego.Message) error {
return c.handleBusinessMessage(ctx, &message)
}, th.AnyBusinessMessage())
bh.HandleEditedBusinessMessage(func(ctx *th.Context, message telego.Message) error {
return c.handleBusinessMessage(ctx, &message)
}, th.AnyEditedBusinessMessage())
bh.HandleBusinessConnection(func(ctx *th.Context, connection telego.BusinessConnection) error {
return c.handleBusinessConnection(ctx, connection)
}, th.AnyBusinessConnection())
bh.HandleDeletedBusinessMessages(func(ctx *th.Context, deleted telego.BusinessMessagesDeleted) error {
return c.handleDeletedBusinessMessages(ctx, deleted)
}, th.AnyDeletedBusinessMessages())
}
c.SetRunning(true)
logger.InfoCF("telegram", "Telegram bot connected", map[string]any{
@ -189,7 +229,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]
useMarkdownV2 := c.tgCfg.UseMarkdownV2
chatID, threadID, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context)
target, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context)
if err != nil {
return nil, fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed)
}
@ -253,8 +293,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]
if smallerLen <= 0 {
msgID, err := c.sendChunk(ctx, sendChunkParams{
chatID: chatID,
threadID: threadID,
target: target,
content: content,
replyToID: replyToID,
mdFallback: chunk,
@ -293,8 +332,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]
}
msgID, err := c.sendChunk(ctx, sendChunkParams{
chatID: chatID,
threadID: threadID,
target: target,
content: content,
replyToID: replyToID,
mdFallback: chunk,
@ -318,8 +356,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]
}
type sendChunkParams struct {
chatID int64
threadID int
target telegramOutboundTarget
content string
replyToID string
mdFallback string
@ -332,8 +369,9 @@ func (c *TelegramChannel) sendChunk(
ctx context.Context,
params sendChunkParams,
) (string, error) {
tgMsg := tu.Message(tu.ID(params.chatID), params.content)
tgMsg.MessageThreadID = params.threadID
tgMsg := tu.Message(tu.ID(params.target.chatID), params.content)
tgMsg.MessageThreadID = params.target.threadID
tgMsg.BusinessConnectionID = params.target.businessConnectionID
if params.useMarkdownV2 {
tgMsg.WithParseMode(telego.ModeMarkdownV2)
} else {
@ -375,13 +413,14 @@ const maxTypingDuration = 5 * time.Minute
// The goroutine also exits automatically after maxTypingDuration if cancel is
// never called (e.g. when the LLM fails or times out without publishing).
func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
cid, threadID, err := parseTelegramChatID(chatID)
target, err := resolveTelegramOutboundTarget(chatID, nil)
if err != nil {
return func() {}, err
}
action := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping)
action.MessageThreadID = threadID
action := tu.ChatAction(tu.ID(target.chatID), telego.ChatActionTyping)
action.MessageThreadID = target.threadID
action.BusinessConnectionID = target.businessConnectionID
// Send the first typing action immediately
_ = c.bot.SendChatAction(ctx, action)
@ -398,8 +437,9 @@ func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func(
case <-maxCtx.Done():
return
case <-ticker.C:
a := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping)
a.MessageThreadID = threadID
a := tu.ChatAction(tu.ID(target.chatID), telego.ChatActionTyping)
a.MessageThreadID = target.threadID
a.BusinessConnectionID = target.businessConnectionID
_ = c.bot.SendChatAction(typingCtx, a)
}
}
@ -411,7 +451,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 {
useMarkdownV2 := c.tgCfg.UseMarkdownV2
cid, _, err := parseTelegramChatID(chatID)
target, err := resolveTelegramOutboundTarget(chatID, nil)
if err != nil {
return err
}
@ -420,7 +460,8 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag
return err
}
parsedContent := parseContent(content, useMarkdownV2)
editMsg := tu.EditMessageText(tu.ID(cid), mid, parsedContent)
editMsg := tu.EditMessageText(tu.ID(target.chatID), mid, parsedContent)
editMsg.BusinessConnectionID = target.businessConnectionID
if useMarkdownV2 {
editMsg.WithParseMode(telego.ModeMarkdownV2)
} else {
@ -439,7 +480,9 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag
// Network errors or timeouts should NOT trigger a retry with different content.
if strings.Contains(err.Error(), "Bad Request") {
logParseFailed(err, useMarkdownV2)
_, err = c.bot.EditMessageText(ctx, tu.EditMessageText(tu.ID(cid), mid, content))
plainEdit := tu.EditMessageText(tu.ID(target.chatID), mid, content)
plainEdit.BusinessConnectionID = target.businessConnectionID
_, err = c.bot.EditMessageText(ctx, plainEdit)
}
}
@ -467,7 +510,7 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag
// DeleteMessage implements channels.MessageDeleter.
func (c *TelegramChannel) DeleteMessage(ctx context.Context, chatID string, messageID string) error {
cid, _, err := parseTelegramChatID(chatID)
target, err := resolveTelegramOutboundTarget(chatID, nil)
if err != nil {
return err
}
@ -475,8 +518,14 @@ func (c *TelegramChannel) DeleteMessage(ctx context.Context, chatID string, mess
if err != nil {
return err
}
if target.businessConnectionID != "" {
return c.bot.DeleteBusinessMessages(ctx, &telego.DeleteBusinessMessagesParams{
BusinessConnectionID: target.businessConnectionID,
MessageIDs: []int{mid},
})
}
return c.bot.DeleteMessage(ctx, &telego.DeleteMessageParams{
ChatID: tu.ID(cid),
ChatID: tu.ID(target.chatID),
MessageID: mid,
})
}
@ -575,13 +624,14 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (s
text := phCfg.GetRandomText()
cid, threadID, err := parseTelegramChatID(chatID)
target, err := resolveTelegramOutboundTarget(chatID, nil)
if err != nil {
return "", err
}
phMsg := tu.Message(tu.ID(cid), text)
phMsg.MessageThreadID = threadID
phMsg := tu.Message(tu.ID(target.chatID), text)
phMsg.MessageThreadID = target.threadID
phMsg.BusinessConnectionID = target.businessConnectionID
pMsg, err := c.bot.SendMessage(ctx, phMsg)
if err != nil {
return "", err
@ -598,7 +648,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
trackedChatID := telegramToolFeedbackChatKey(msg.ChatID, &msg.Context)
trackedMsgID, hasTrackedMsg := c.currentToolFeedbackMessage(trackedChatID)
chatID, threadID, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context)
target, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context)
if err != nil {
return nil, fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed)
}
@ -632,10 +682,11 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
switch part.Type {
case "image":
params := &telego.SendPhotoParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Photo: telego.InputFile{File: file},
Caption: part.Caption,
BusinessConnectionID: target.businessConnectionID,
ChatID: tu.ID(target.chatID),
MessageThreadID: target.threadID,
Photo: telego.InputFile{File: file},
Caption: part.Caption,
}
tgResult, err = c.bot.SendPhoto(ctx, params)
if err != nil && strings.Contains(err.Error(), "PHOTO_INVALID_DIMENSIONS") {
@ -645,10 +696,11 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
}
docParams := &telego.SendDocumentParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Document: telego.InputFile{File: file},
Caption: part.Caption,
BusinessConnectionID: target.businessConnectionID,
ChatID: tu.ID(target.chatID),
MessageThreadID: target.threadID,
Document: telego.InputFile{File: file},
Caption: part.Caption,
}
tgResult, err = c.bot.SendDocument(ctx, docParams)
}
@ -658,35 +710,39 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
fn := strings.ToLower(part.Filename)
if strings.Contains(fn, "voice") && (strings.HasSuffix(fn, ".ogg") || strings.HasSuffix(fn, ".oga")) {
vparams := &telego.SendVoiceParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Voice: telego.InputFile{File: file},
Caption: part.Caption,
BusinessConnectionID: target.businessConnectionID,
ChatID: tu.ID(target.chatID),
MessageThreadID: target.threadID,
Voice: telego.InputFile{File: file},
Caption: part.Caption,
}
tgResult, err = c.bot.SendVoice(ctx, vparams)
} else {
params := &telego.SendAudioParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Audio: telego.InputFile{File: file},
Caption: part.Caption,
BusinessConnectionID: target.businessConnectionID,
ChatID: tu.ID(target.chatID),
MessageThreadID: target.threadID,
Audio: telego.InputFile{File: file},
Caption: part.Caption,
}
tgResult, err = c.bot.SendAudio(ctx, params)
}
case "video":
params := &telego.SendVideoParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Video: telego.InputFile{File: file},
Caption: part.Caption,
BusinessConnectionID: target.businessConnectionID,
ChatID: tu.ID(target.chatID),
MessageThreadID: target.threadID,
Video: telego.InputFile{File: file},
Caption: part.Caption,
}
tgResult, err = c.bot.SendVideo(ctx, params)
default: // "file" or unknown types
params := &telego.SendDocumentParams{
ChatID: tu.ID(chatID),
MessageThreadID: threadID,
Document: telego.InputFile{File: file},
Caption: part.Caption,
BusinessConnectionID: target.businessConnectionID,
ChatID: tu.ID(target.chatID),
MessageThreadID: target.threadID,
Document: telego.InputFile{File: file},
Caption: part.Caption,
}
tgResult, err = c.bot.SendDocument(ctx, params)
}
@ -713,6 +769,76 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
}
func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Message) error {
return c.handleTelegramMessage(ctx, message, "")
}
func (c *TelegramChannel) businessModeEnabled() bool {
return c != nil && c.tgCfg != nil && c.tgCfg.BusinessMode
}
func (c *TelegramChannel) handleBusinessMessage(ctx context.Context, message *telego.Message) error {
if !c.businessModeEnabled() {
return nil
}
if message == nil {
return fmt.Errorf("message is nil")
}
businessConnectionID := strings.TrimSpace(message.BusinessConnectionID)
if businessConnectionID == "" {
return fmt.Errorf("business message missing business_connection_id")
}
c.markBusinessMessageRead(ctx, businessConnectionID, message.Chat.ID, message.MessageID)
return c.handleTelegramMessage(ctx, message, businessConnectionID)
}
func (c *TelegramChannel) markBusinessMessageRead(
ctx context.Context,
businessConnectionID string,
chatID int64,
messageID int,
) {
if c == nil || c.bot == nil || strings.TrimSpace(businessConnectionID) == "" || chatID == 0 || messageID == 0 {
return
}
if err := c.bot.ReadBusinessMessage(ctx, &telego.ReadBusinessMessageParams{
BusinessConnectionID: businessConnectionID,
ChatID: chatID,
MessageID: messageID,
}); err != nil {
logger.DebugCF("telegram", "Failed to mark business message as read", map[string]any{
"business_connection_id": businessConnectionID,
"chat_id": chatID,
"message_id": messageID,
"error": err.Error(),
})
}
}
func (c *TelegramChannel) handleBusinessConnection(_ context.Context, connection telego.BusinessConnection) error {
if !c.businessModeEnabled() {
return nil
}
logger.InfoCF("telegram", "Business connection updated", map[string]any{
"business_connection_id": connection.ID,
"user_chat_id": connection.UserChatID,
"is_enabled": connection.IsEnabled,
})
return nil
}
func (c *TelegramChannel) handleDeletedBusinessMessages(_ context.Context, deleted telego.BusinessMessagesDeleted) error {
if !c.businessModeEnabled() {
return nil
}
logger.DebugCF("telegram", "Business messages deleted", map[string]any{
"business_connection_id": deleted.BusinessConnectionID,
"chat_id": deleted.Chat.ID,
"message_count": len(deleted.MessageIDs),
})
return nil
}
func (c *TelegramChannel) handleTelegramMessage(ctx context.Context, message *telego.Message, businessConnectionID string) error {
if message == nil {
return fmt.Errorf("message is nil")
}
@ -745,7 +871,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
content := ""
mediaPaths := []string{}
chatIDStr := fmt.Sprintf("%d", chatID)
chatIDStr := formatTelegramDeliveryChatID(chatID, businessConnectionID, 0)
messageIDStr := fmt.Sprintf("%d", message.MessageID)
scope := channels.BuildMediaScope("telegram", chatIDStr, messageIDStr)
@ -864,10 +990,10 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
// 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)
compositeChatID := formatTelegramDeliveryChatID(chatID, businessConnectionID, 0)
threadID := message.MessageThreadID
if message.Chat.IsForum && threadID != 0 {
compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID)
compositeChatID = formatTelegramDeliveryChatID(chatID, businessConnectionID, threadID)
}
logger.DebugCF("telegram", "Received message", map[string]any{
@ -889,16 +1015,22 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
"first_name": user.FirstName,
"is_group": fmt.Sprintf("%t", message.Chat.Type != "private"),
}
if businessConnectionID != "" {
metadata["business_connection_id"] = businessConnectionID
}
inboundCtx := bus.InboundContext{
Channel: c.Name(),
ChatID: fmt.Sprintf("%d", chatID),
ChatID: formatTelegramDeliveryChatID(chatID, businessConnectionID, 0),
ChatType: peerKind,
SenderID: platformID,
MessageID: messageID,
Mentioned: isMentioned,
Raw: metadata,
}
if businessConnectionID != "" {
inboundCtx.Account = businessConnectionID
}
if message.Chat.IsForum && threadID != 0 {
inboundCtx.TopicID = fmt.Sprintf("%d", threadID)
}
@ -1123,11 +1255,11 @@ func (c *TelegramChannel) PrepareToolFeedbackMessageContent(content string) stri
}
func telegramToolFeedbackChatKey(chatID string, outboundCtx *bus.InboundContext) string {
resolvedChatID, threadID, err := resolveTelegramOutboundTarget(chatID, outboundCtx)
if err != nil || threadID == 0 {
target, err := resolveTelegramOutboundTarget(chatID, outboundCtx)
if err != nil {
return strings.TrimSpace(chatID)
}
return fmt.Sprintf("%d/%d", resolvedChatID, threadID)
return target.String()
}
func (c *TelegramChannel) ToolFeedbackMessageChatID(chatID string, outboundCtx *bus.InboundContext) string {
@ -1153,26 +1285,93 @@ func parseTelegramChatID(chatID string) (int64, int, error) {
return cid, tid, nil
}
func resolveTelegramOutboundTarget(chatID string, outboundCtx *bus.InboundContext) (int64, int, error) {
type telegramOutboundTarget struct {
businessConnectionID string
chatID int64
threadID int
}
func (t telegramOutboundTarget) String() string {
return formatTelegramDeliveryChatID(t.chatID, t.businessConnectionID, t.threadID)
}
func formatTelegramDeliveryChatID(chatID int64, businessConnectionID string, threadID int) string {
base := fmt.Sprintf("%d", chatID)
if strings.TrimSpace(businessConnectionID) != "" {
base = fmt.Sprintf("business:%s:%d", url.QueryEscape(strings.TrimSpace(businessConnectionID)), chatID)
}
if threadID != 0 {
return fmt.Sprintf("%s/%d", base, threadID)
}
return base
}
func parseTelegramOutboundTarget(chatID string) (telegramOutboundTarget, error) {
targetChatID := strings.TrimSpace(chatID)
if strings.HasPrefix(targetChatID, "business:") {
rest := strings.TrimPrefix(targetChatID, "business:")
parts := strings.SplitN(rest, ":", 2)
if len(parts) != 2 || strings.TrimSpace(parts[0]) == "" || strings.TrimSpace(parts[1]) == "" {
return telegramOutboundTarget{}, fmt.Errorf("invalid business chat ID %q", chatID)
}
businessConnectionID, unescapeErr := url.QueryUnescape(strings.TrimSpace(parts[0]))
if unescapeErr != nil {
return telegramOutboundTarget{}, fmt.Errorf("invalid business connection ID in chat ID %q: %w", chatID, unescapeErr)
}
cid, threadID, err := parseTelegramChatID(parts[1])
if err != nil {
return telegramOutboundTarget{}, err
}
return telegramOutboundTarget{
businessConnectionID: businessConnectionID,
chatID: cid,
threadID: threadID,
}, nil
}
cid, threadID, err := parseTelegramChatID(targetChatID)
if err != nil {
return telegramOutboundTarget{}, err
}
return telegramOutboundTarget{chatID: cid, threadID: threadID}, nil
}
func businessConnectionIDFromContext(outboundCtx *bus.InboundContext) string {
if outboundCtx == nil {
return ""
}
if outboundCtx.Raw != nil {
if businessConnectionID := strings.TrimSpace(outboundCtx.Raw["business_connection_id"]); businessConnectionID != "" {
return businessConnectionID
}
}
return strings.TrimSpace(outboundCtx.Account)
}
func resolveTelegramOutboundTarget(chatID string, outboundCtx *bus.InboundContext) (telegramOutboundTarget, error) {
targetChatID := strings.TrimSpace(chatID)
if targetChatID == "" && outboundCtx != nil {
targetChatID = strings.TrimSpace(outboundCtx.ChatID)
}
resolvedChatID, resolvedThreadID, err := parseTelegramChatID(targetChatID)
resolved, err := parseTelegramOutboundTarget(targetChatID)
if err != nil {
return 0, 0, err
return telegramOutboundTarget{}, err
}
if resolvedThreadID != 0 || outboundCtx == nil {
return resolvedChatID, resolvedThreadID, nil
if resolved.businessConnectionID == "" {
resolved.businessConnectionID = businessConnectionIDFromContext(outboundCtx)
}
if resolved.threadID != 0 || outboundCtx == nil {
return resolved, nil
}
topicID := strings.TrimSpace(outboundCtx.TopicID)
if topicID == "" {
return resolvedChatID, resolvedThreadID, nil
return resolved, nil
}
if threadID, convErr := strconv.Atoi(topicID); convErr == nil {
return resolvedChatID, threadID, nil
resolved.threadID = threadID
return resolved, nil
}
return resolvedChatID, resolvedThreadID, nil
return resolved, nil
}
func logParseFailed(err error, useMarkdownV2 bool) {
@ -1284,16 +1483,19 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann
return nil, fmt.Errorf("streaming disabled in config")
}
cid, threadID, err := parseTelegramChatID(chatID)
target, err := resolveTelegramOutboundTarget(chatID, nil)
if err != nil {
return nil, err
}
if target.businessConnectionID != "" {
return nil, fmt.Errorf("streaming is not supported for telegram business chats")
}
streamCfg := c.tgCfg.Streaming
return &telegramStreamer{
bot: c.bot,
chatID: cid,
threadID: threadID,
chatID: target.chatID,
threadID: target.threadID,
draftID: cryptoRandInt(),
throttleInterval: time.Duration(streamCfg.ThrottleSeconds) * time.Second,
minGrowth: streamCfg.MinGrowthChars,

97
pkg/channels/telegram/telegram_dispatch_test.go
View file

@ -2,12 +2,14 @@ package telegram
import (
"context"
"strings"
"testing"
"github.com/mymmrac/telego"
"github.com/sipeed/picoclaw/pkg/bus"
"github.com/sipeed/picoclaw/pkg/channels"
"github.com/sipeed/picoclaw/pkg/config"
)
func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) {
@ -16,6 +18,7 @@ func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) {
BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil),
chatIDs: make(map[string]int64),
ctx: context.Background(),
tgCfg: &config.TelegramSettings{BusinessMode: true},
}
msg := &telego.Message{
@ -46,3 +49,97 @@ func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) {
t.Fatalf("content=%q", inbound.Content)
}
}
func TestHandleBusinessMessage_DisabledBusinessModeIgnoresMessage(t *testing.T) {
messageBus := bus.NewMessageBus()
ch := &TelegramChannel{
BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil),
chatIDs: make(map[string]int64),
ctx: context.Background(),
tgCfg: &config.TelegramSettings{BusinessMode: false},
}
msg := &telego.Message{
Text: "ignored business message",
MessageID: 18,
BusinessConnectionID: "biz-conn-1",
Chat: telego.Chat{
ID: 777,
Type: "private",
},
From: &telego.User{
ID: 42,
FirstName: "Alice",
},
}
if err := ch.handleBusinessMessage(context.Background(), msg); err != nil {
t.Fatalf("handleBusinessMessage error: %v", err)
}
select {
case inbound := <-messageBus.InboundChan():
t.Fatalf("expected disabled business mode to ignore message, got %#v", inbound)
default:
}
}
func TestTelegramAllowedUpdates_BusinessMode(t *testing.T) {
disabled := strings.Join(telegramAllowedUpdates(false), ",")
if strings.Contains(disabled, telego.BusinessMessageUpdates) {
t.Fatalf("disabled updates include business messages: %s", disabled)
}
enabled := strings.Join(telegramAllowedUpdates(true), ",")
if !strings.Contains(enabled, telego.BusinessMessageUpdates) {
t.Fatalf("enabled updates do not include business messages: %s", enabled)
}
if !strings.Contains(enabled, telego.DeletedBusinessMessagesUpdates) {
t.Fatalf("enabled updates do not include deleted business messages: %s", enabled)
}
}
func TestHandleBusinessMessage_ForwardsWithBusinessContext(t *testing.T) {
messageBus := bus.NewMessageBus()
ch := &TelegramChannel{
BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil),
chatIDs: make(map[string]int64),
ctx: context.Background(),
tgCfg: &config.TelegramSettings{BusinessMode: true},
}
msg := &telego.Message{
Text: "hello from business",
MessageID: 17,
BusinessConnectionID: "biz-conn-1",
Chat: telego.Chat{
ID: 777,
Type: "private",
},
From: &telego.User{
ID: 42,
FirstName: "Alice",
},
}
if err := ch.handleBusinessMessage(context.Background(), msg); err != nil {
t.Fatalf("handleBusinessMessage error: %v", err)
}
inbound, ok := <-messageBus.InboundChan()
if !ok {
t.Fatal("expected inbound message to be forwarded")
}
if inbound.ChatID != "business:biz-conn-1:777" {
t.Fatalf("chat_id=%q", inbound.ChatID)
}
if inbound.Context.Account != "biz-conn-1" {
t.Fatalf("account=%q", inbound.Context.Account)
}
if inbound.Context.Raw["business_connection_id"] != "biz-conn-1" {
t.Fatalf("business_connection_id=%q", inbound.Context.Raw["business_connection_id"])
}
if inbound.Content != "hello from business" {
t.Fatalf("content=%q", inbound.Content)
}
}

130
pkg/channels/telegram/telegram_test.go
View file

@ -150,6 +150,90 @@ func newTestChannelWithConstructor(
}
}
func TestHandleBusinessMessage_MarksMessageRead(t *testing.T) {
messageBus := bus.NewMessageBus()
caller := &stubCaller{
callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) {
if !strings.Contains(url, "readBusinessMessage") {
t.Fatalf("unexpected API call: %s", url)
}
b, err := json.Marshal(true)
require.NoError(t, err)
return &ta.Response{Ok: true, Result: b}, nil
},
}
ch := newTestChannel(t, caller)
ch.BaseChannel = channels.NewBaseChannel("telegram", nil, messageBus, nil)
ch.BaseChannel.SetOwner(ch)
ch.ctx = context.Background()
ch.tgCfg.BusinessMode = true
msg := &telego.Message{
Text: "hello from business",
MessageID: 17,
BusinessConnectionID: "biz-conn-1",
Chat: telego.Chat{
ID: 777,
Type: "private",
},
From: &telego.User{
ID: 42,
FirstName: "Alice",
},
}
require.NoError(t, ch.handleBusinessMessage(context.Background(), msg))
require.Len(t, caller.calls, 1)
assert.Contains(t, caller.calls[0].URL, "readBusinessMessage")
var params struct {
BusinessConnectionID string `json:"business_connection_id"`
ChatID int64 `json:"chat_id"`
MessageID int `json:"message_id"`
}
require.NoError(t, json.Unmarshal(caller.calls[0].Data.BodyRaw, &params))
assert.Equal(t, "biz-conn-1", params.BusinessConnectionID)
assert.Equal(t, int64(777), params.ChatID)
assert.Equal(t, 17, params.MessageID)
inbound := <-messageBus.InboundChan()
assert.Equal(t, "hello from business", inbound.Content)
}
func TestHandleBusinessMessage_ReadFailureStillForwardsMessage(t *testing.T) {
messageBus := bus.NewMessageBus()
caller := &stubCaller{
callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) {
return nil, errors.New("missing can_read_messages")
},
}
ch := newTestChannel(t, caller)
ch.BaseChannel = channels.NewBaseChannel("telegram", nil, messageBus, nil)
ch.BaseChannel.SetOwner(ch)
ch.ctx = context.Background()
ch.tgCfg.BusinessMode = true
msg := &telego.Message{
Text: "still forward me",
MessageID: 18,
BusinessConnectionID: "biz-conn-1",
Chat: telego.Chat{
ID: 777,
Type: "private",
},
From: &telego.User{
ID: 42,
FirstName: "Alice",
},
}
require.NoError(t, ch.handleBusinessMessage(context.Background(), msg))
require.Len(t, caller.calls, 1)
inbound := <-messageBus.InboundChan()
assert.Equal(t, "still forward me", inbound.Content)
}
func TestSendMedia_ImageFallbacksToDocumentOnInvalidDimensions(t *testing.T) {
constructor := &multipartRecordingConstructor{}
caller := &stubCaller{
@ -271,6 +355,41 @@ func TestSend_ShortMessage_SingleCall(t *testing.T) {
assert.Len(t, caller.calls, 1, "short message should result in exactly one SendMessage call")
}
func TestSend_BusinessMessageIncludesBusinessConnectionID(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: "business:biz-conn-1:777",
Content: "hello business",
Context: bus.InboundContext{
Channel: "telegram",
ChatID: "business:biz-conn-1:777",
Account: "biz-conn-1",
Raw: map[string]string{
"business_connection_id": "biz-conn-1",
},
},
})
require.NoError(t, err)
require.Len(t, caller.calls, 1)
var params struct {
ChatID int64 `json:"chat_id"`
BusinessConnectionID string `json:"business_connection_id"`
Text string `json:"text"`
}
require.NoError(t, json.Unmarshal(caller.calls[0].Data.BodyRaw, &params))
assert.Equal(t, int64(777), params.ChatID)
assert.Equal(t, "biz-conn-1", params.BusinessConnectionID)
assert.Equal(t, "hello business", params.Text)
}
func TestSend_NonToolFeedbackDeletesTrackedProgressMessage(t *testing.T) {
caller := &stubCaller{
callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) {
@ -686,6 +805,17 @@ func TestParseTelegramChatID_InvalidThreadID(t *testing.T) {
assert.Contains(t, err.Error(), "invalid thread ID")
}
func TestParseTelegramOutboundTarget_BusinessConnectionIDWithColon(t *testing.T) {
formatted := formatTelegramDeliveryChatID(777, "biz:conn:1", 42)
target, err := parseTelegramOutboundTarget(formatted)
require.NoError(t, err)
assert.Equal(t, int64(777), target.chatID)
assert.Equal(t, 42, target.threadID)
assert.Equal(t, "biz:conn:1", target.businessConnectionID)
}
func TestSend_WithForumThreadID(t *testing.T) {
caller := &stubCaller{
callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) {

1
pkg/config/config.go
View file

@ -363,6 +363,7 @@ type TelegramSettings struct {
BaseURL string `json:"base_url" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_BASE_URL"`
Proxy string `json:"proxy" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"`
Streaming StreamingConfig `json:"streaming,omitempty" yaml:"-"`
BusinessMode bool `json:"business_mode" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_BUSINESS_MODE"`
UseMarkdownV2 bool `json:"use_markdown_v2" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_USE_MARKDOWN_V2"`
}

3
pkg/config/config_channel_test.go
View file

@ -17,6 +17,7 @@ import (
type testTelegramConfig struct {
BaseURL string `json:"base_url" yaml:"-"`
Proxy string `json:"proxy" yaml:"-"`
BusinessMode bool `json:"business_mode" yaml:"-"`
UseMarkdownV2 bool `json:"use_markdown_v2" yaml:"-"`
Streaming StreamingConfig `json:"streaming,omitempty" yaml:"-"`
Token SecureString `json:"token,omitzero" yaml:"token,omitempty"`
@ -107,6 +108,7 @@ func TestChannel_JSON_Unmarshal(t *testing.T) {
"reasoning_channel_id": "-100xxx",
"settings": {
"base_url": "https://custom-api.example.com",
"business_mode": true,
"use_markdown_v2": true,
"streaming": {"enabled": true, "throttle_seconds": 2},
"token": "[NOT_HERE]"
@ -126,6 +128,7 @@ func TestChannel_JSON_Unmarshal(t *testing.T) {
var cfg testTelegramConfig
require.NoError(t, ch.Decode(&cfg))
assert.Equal(t, "https://custom-api.example.com", cfg.BaseURL)
assert.True(t, cfg.BusinessMode)
assert.True(t, cfg.UseMarkdownV2)
assert.True(t, cfg.Streaming.Enabled)
assert.Equal(t, 2, cfg.Streaming.ThrottleSeconds)

1
pkg/config/defaults.go
View file

@ -497,6 +497,7 @@ func defaultChannels() ChannelsConfig {
"placeholder": map[string]any{"enabled": true, "text": []string{"Thinking... 💭"}},
"settings": map[string]any{
"streaming": map[string]any{"enabled": true, "throttle_seconds": 3, "min_growth_chars": 200},
"business_mode": false,
"use_markdown_v2": false,
},
},

10
web/frontend/src/components/channels/channel-forms/telegram-form.tsx
View file

@ -86,6 +86,16 @@ export function TelegramForm({
placeholder="https://api.telegram.org"
/>
</Field>
<div>
<SwitchCardField
label={t("channels.field.businessMode")}
hint={t("channels.form.desc.businessMode")}
checked={asBool(config.business_mode)}
onCheckedChange={(checked) => onChange("business_mode", checked)}
ariaLabel={t("channels.field.businessMode")}
/>
</div>
</CardContent>
</Card>

2
web/frontend/src/i18n/locales/en.json
View file

@ -375,6 +375,7 @@
"verificationToken": "Verification Token",
"encryptKey": "Encrypt Key",
"baseUrl": "API Base URL",
"businessMode": "Business Mode",
"proxy": "HTTP Proxy",
"mentionOnly": "Mention Only",
"typingEnabled": "Typing Indicator",
@ -414,6 +415,7 @@
"verificationToken": "Verification token for event callbacks.",
"encryptKey": "Encryption key used to decrypt callback payloads.",
"baseUrl": "Platform API base URL. Official endpoint is used by default.",
"businessMode": "Receive and reply to Telegram Business messages for connected business accounts.",
"proxy": "HTTP proxy address for outbound network access.",
"mentionOnly": "Only respond when the bot is explicitly mentioned in group chats.",
"typingEnabled": "Display typing status while the assistant is generating a response.",

2
web/frontend/src/i18n/locales/pt-br.json
View file

@ -367,6 +367,7 @@
"verificationToken": "Token de Verificação",
"encryptKey": "Chave de Criptografia",
"baseUrl": "URL Base da API",
"businessMode": "Modo Business",
"proxy": "Proxy HTTP",
"mentionOnly": "Apenas com Menção",
"typingEnabled": "Indicador de Digitação",
@ -406,6 +407,7 @@
"verificationToken": "Token de verificação para callbacks de eventos.",
"encryptKey": "Chave de criptografia usada para descriptografar payloads de callback.",
"baseUrl": "URL base da API da plataforma. O endpoint oficial é usado por padrão.",
"businessMode": "Receber e responder mensagens do Telegram Business para contas comerciais conectadas.",
"proxy": "Endereço de proxy HTTP para acesso de rede de saída.",
"mentionOnly": "Responder apenas quando o bot for explicitamente mencionado em chats em grupo.",
"typingEnabled": "Exibir status de digitação enquanto o assistente está gerando uma resposta.",

2
web/frontend/src/i18n/locales/zh.json
View file

@ -375,6 +375,7 @@
"verificationToken": "Verification Token",
"encryptKey": "Encrypt Key",
"baseUrl": "API Base URL",
"businessMode": "Business Mode",
"proxy": "HTTP 代理",
"mentionOnly": "仅提及时响应",
"typingEnabled": "输入中提示",
@ -414,6 +415,7 @@
"verificationToken": "事件回调验证令牌",
"encryptKey": "消息加密密钥，用于解密回调内容",
"baseUrl": "平台 API 地址，默认使用官方地址",
"businessMode": "接收并回复已连接商业账号的 Telegram Business 消息",
"proxy": "HTTP 代理地址，用于网络访问",
"mentionOnly": "在群聊中仅当明确提及时才响应",
"typingEnabled": "在生成回复时显示“正在输入”状态",