fix(telegram-threading): improve thread isolation and reduce duplication

- Fix async tool callback missing ThreadID/ReplyToMessageID in PublishOutbound
- Fix stale comments on Manager sync.Map fields (now "channel:chatID:threadID")
- Extract parseOptionalInt helper to eliminate repeated strconv.Atoi pattern
- Add TestPreSend_ThreadIsolation to verify thread-scoped placeholder isolation
- Add TestParseOptionalInt for the new helper

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Denys Vitali 2026-03-08 09:44:20 +00:00
parent 7e1b04e5f5
commit 30e8f2d49e
5 changed files with 80 additions and 30 deletions

View file

@ -1169,6 +1169,8 @@ func (al *AgentLoop) runLLMIteration(
Channel: opts.Channel, Channel: opts.Channel,
ChatID: opts.ChatID, ChatID: opts.ChatID,
Content: result.ForUser, Content: result.ForUser,
ThreadID: opts.ThreadID,
ReplyToMessageID: opts.ReplyToMessageID,
}) })
} }

View file

@ -84,9 +84,9 @@ type Manager struct {
mux *http.ServeMux mux *http.ServeMux
httpServer *http.Server httpServer *http.Server
mu sync.RWMutex mu sync.RWMutex
placeholders sync.Map // "channel:chatID" → placeholderID (string) placeholders sync.Map // "channel:chatID:threadID" → placeholderEntry
typingStops sync.Map // "channel:chatID" → func() typingStops sync.Map // "channel:chatID:threadID" → typingEntry
reactionUndos sync.Map // "channel:chatID" → reactionEntry reactionUndos sync.Map // "channel:chatID:threadID" → reactionEntry
} }
type asyncTask struct { type asyncTask struct {

View file

@ -570,6 +570,44 @@ func TestPreSend_TypingAndPlaceholder(t *testing.T) {
} }
} }
func TestPreSend_ThreadIsolation(t *testing.T) {
m := newTestManager()
ch := &mockMessageEditor{
mockChannel: mockChannel{
sendFn: func(_ context.Context, _ bus.OutboundMessage) error {
return nil
},
},
editFn: func(_ context.Context, _, _, _ string) error {
return nil
},
}
// Register placeholder for thread "100"
m.RecordPlaceholder("test", "123", "100", "ph-100")
// Sending to the same chat but different thread should NOT find the placeholder
msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello", ThreadID: "200"}
edited := m.preSend(context.Background(), "test", msg, ch)
if edited {
t.Fatal("expected preSend to return false for a different threadID")
}
// Sending to the correct thread should find it
msg.ThreadID = "100"
edited = m.preSend(context.Background(), "test", msg, ch)
if !edited {
t.Fatal("expected preSend to return true for matching threadID")
}
// Second call to same thread should not find it (already consumed)
edited = m.preSend(context.Background(), "test", msg, ch)
if edited {
t.Fatal("expected preSend to return false after placeholder was consumed")
}
}
func TestRecordPlaceholder_ConcurrentSafe(t *testing.T) { func TestRecordPlaceholder_ConcurrentSafe(t *testing.T) {
m := newTestManager() m := newTestManager()

View file

@ -213,16 +213,10 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
func (c *TelegramChannel) sendHTMLChunk(ctx context.Context, chatID int64, htmlContent, mdFallback, threadID, replyToMsgID 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 := tu.Message(tu.ID(chatID), htmlContent)
tgMsg.ParseMode = telego.ModeHTML tgMsg.ParseMode = telego.ModeHTML
if threadID != "" { tgMsg.MessageThreadID = parseOptionalInt(threadID)
if tid, err2 := strconv.Atoi(threadID); err2 == nil { if mid := parseOptionalInt(replyToMsgID); mid != 0 {
tgMsg.MessageThreadID = tid
}
}
if replyToMsgID != "" {
if mid, err2 := strconv.Atoi(replyToMsgID); err2 == nil {
tgMsg.ReplyParameters = &telego.ReplyParameters{MessageID: mid} tgMsg.ReplyParameters = &telego.ReplyParameters{MessageID: mid}
} }
}
if _, err := c.bot.SendMessage(ctx, tgMsg); err != nil { if _, err := c.bot.SendMessage(ctx, tgMsg); err != nil {
logger.ErrorCF("telegram", "HTML parse failed, falling back to plain text", map[string]any{ logger.ErrorCF("telegram", "HTML parse failed, falling back to plain text", map[string]any{
@ -248,14 +242,9 @@ func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string, thread
return func() {}, err return func() {}, err
} }
tid := 0
if threadID != "" {
tid, _ = strconv.Atoi(threadID)
}
sendTyping := func(sctx context.Context) { sendTyping := func(sctx context.Context) {
params := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) params := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping)
params.MessageThreadID = tid params.MessageThreadID = parseOptionalInt(threadID)
_ = c.bot.SendChatAction(sctx, params) _ = c.bot.SendChatAction(sctx, params)
} }
@ -318,16 +307,10 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string, th
} }
params := tu.Message(tu.ID(cid), text) params := tu.Message(tu.ID(cid), text)
if threadID != "" { params.MessageThreadID = parseOptionalInt(threadID)
if tid, err2 := strconv.Atoi(threadID); err2 == nil { if mid := parseOptionalInt(replyToMsgID); mid != 0 {
params.MessageThreadID = tid
}
}
if replyToMsgID != "" {
if mid, err2 := strconv.Atoi(replyToMsgID); err2 == nil {
params.ReplyParameters = &telego.ReplyParameters{MessageID: mid} params.ReplyParameters = &telego.ReplyParameters{MessageID: mid}
} }
}
pMsg, err := c.bot.SendMessage(ctx, params) pMsg, err := c.bot.SendMessage(ctx, params)
if err != nil { if err != nil {
@ -628,6 +611,15 @@ func parseChatID(chatIDStr string) (int64, error) {
return id, err return id, err
} }
// parseOptionalInt converts a string to int, returning 0 if the string is empty or invalid.
func parseOptionalInt(s string) int {
if s == "" {
return 0
}
n, _ := strconv.Atoi(s)
return n
}
func markdownToTelegramHTML(text string) string { func markdownToTelegramHTML(text string) string {
if text == "" { if text == "" {
return "" return ""

View file

@ -271,3 +271,21 @@ func TestSend_InvalidChatID(t *testing.T) {
assert.True(t, errors.Is(err, channels.ErrSendFailed), "error should wrap ErrSendFailed") assert.True(t, errors.Is(err, channels.ErrSendFailed), "error should wrap ErrSendFailed")
assert.Empty(t, caller.calls) assert.Empty(t, caller.calls)
} }
func TestParseOptionalInt(t *testing.T) {
tests := []struct {
input string
want int
}{
{"", 0},
{"0", 0},
{"42", 42},
{"-1", -1},
{"abc", 0},
{"123456", 123456},
}
for _, tt := range tests {
got := parseOptionalInt(tt.input)
assert.Equal(t, tt.want, got, "parseOptionalInt(%q)", tt.input)
}
}