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

fix(telegram): resolve streaming redundant drafts and routing

Browse source
This commit is contained in:
Badgerbees 2026-03-27 20:37:49 +07:00
parent 76cd7f8ad5
commit cadf42b1b1
1 changed files with 30 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -888,7 +888,7 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann
return nil, fmt.Errorf("streaming disabled in config") return nil, fmt.Errorf("streaming disabled in config")
} }
cid, _, err := parseTelegramChatID(chatID) cid, tid, err := parseTelegramChatID(chatID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -897,6 +897,7 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann
return &telegramStreamer{ return &telegramStreamer{
bot: c.bot, bot: c.bot,
chatID: cid, chatID: cid,
threadID: tid,
draftID: cryptoRandInt(), draftID: cryptoRandInt(),
throttleInterval: time.Duration(streamCfg.ThrottleSeconds) * time.Second, throttleInterval: time.Duration(streamCfg.ThrottleSeconds) * time.Second,
minGrowth: streamCfg.MinGrowthChars, minGrowth: streamCfg.MinGrowthChars,
@ -909,6 +910,7 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann
type telegramStreamer struct { type telegramStreamer struct {
bot *telego.Bot bot *telego.Bot
chatID int64 chatID int64
threadID int
draftID int draftID int
throttleInterval time.Duration throttleInterval time.Duration
minGrowth int minGrowth int
@ -937,6 +939,7 @@ func (s *telegramStreamer) Update(ctx context.Context, content string) error {
err := s.bot.SendMessageDraft(ctx, &telego.SendMessageDraftParams{ err := s.bot.SendMessageDraft(ctx, &telego.SendMessageDraftParams{
ChatID: s.chatID, ChatID: s.chatID,
MessageThreadID: s.threadID,
DraftID: s.draftID, DraftID: s.draftID,
Text: htmlContent, Text: htmlContent,
ParseMode: telego.ModeHTML, ParseMode: telego.ModeHTML,
@ -958,6 +961,9 @@ func (s *telegramStreamer) Update(ctx context.Context, content string) error {
func (s *telegramStreamer) Finalize(ctx context.Context, content string) error { func (s *telegramStreamer) Finalize(ctx context.Context, content string) error {
htmlContent := markdownToTelegramHTML(content) htmlContent := markdownToTelegramHTML(content)
tgMsg := tu.Message(tu.ID(s.chatID), htmlContent) tgMsg := tu.Message(tu.ID(s.chatID), htmlContent)
if s.threadID != 0 {
tgMsg.MessageThreadID = s.threadID
}
tgMsg.ParseMode = telego.ModeHTML tgMsg.ParseMode = telego.ModeHTML
if _, err := s.bot.SendMessage(ctx, tgMsg); err != nil { if _, err := s.bot.SendMessage(ctx, tgMsg); err != nil {
@ -972,11 +978,29 @@ func (s *telegramStreamer) Finalize(ctx context.Context, content string) error {
return fmt.Errorf("telegram finalize: %w", err) return fmt.Errorf("telegram finalize: %w", err)
} }
} }
// Always clear the partial draft after final delivery to avoid UI ghosts (best-effort)
if !s.failed {
_ = s.bot.SendMessageDraft(ctx, &telego.SendMessageDraftParams{
ChatID: s.chatID,
MessageThreadID: s.threadID,
DraftID: s.draftID,
Text: "",
})
}
return nil return nil
} }
func (s *telegramStreamer) Cancel(ctx context.Context) { func (s *telegramStreamer) Cancel(ctx context.Context) {
// Draft auto-expires on Telegram's side; nothing to clean up. if !s.failed {
_ = s.bot.SendMessageDraft(ctx, &telego.SendMessageDraftParams{
ChatID: s.chatID,
MessageThreadID: s.threadID,
DraftID: s.draftID,
Text: "",
})
}
} }
// cryptoRandInt returns a non-zero random int using crypto/rand. // cryptoRandInt returns a non-zero random int using crypto/rand.