diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index f731d4e5a..755197be8 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -1166,9 +1166,11 @@ func (al *AgentLoop) runLLMIteration( outCtx, outCancel := context.WithTimeout(context.Background(), 5*time.Second) defer outCancel() _ = al.bus.PublishOutbound(outCtx, bus.OutboundMessage{ - Channel: opts.Channel, - ChatID: opts.ChatID, - Content: result.ForUser, + Channel: opts.Channel, + ChatID: opts.ChatID, + Content: result.ForUser, + ThreadID: opts.ThreadID, + ReplyToMessageID: opts.ReplyToMessageID, }) } diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 50dd24d8f..0ae0f40ba 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -84,9 +84,9 @@ type Manager struct { mux *http.ServeMux httpServer *http.Server mu sync.RWMutex - placeholders sync.Map // "channel:chatID" → placeholderID (string) - typingStops sync.Map // "channel:chatID" → func() - reactionUndos sync.Map // "channel:chatID" → reactionEntry + placeholders sync.Map // "channel:chatID:threadID" → placeholderEntry + typingStops sync.Map // "channel:chatID:threadID" → typingEntry + reactionUndos sync.Map // "channel:chatID:threadID" → reactionEntry } type asyncTask struct { diff --git a/pkg/channels/manager_test.go b/pkg/channels/manager_test.go index 201b6b00a..79d5f45b4 100644 --- a/pkg/channels/manager_test.go +++ b/pkg/channels/manager_test.go @@ -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) { m := newTestManager() diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index ba2edfbfe..c7456e002 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -213,15 +213,9 @@ 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 { 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} - } + tgMsg.MessageThreadID = parseOptionalInt(threadID) + if mid := parseOptionalInt(replyToMsgID); mid != 0 { + tgMsg.ReplyParameters = &telego.ReplyParameters{MessageID: mid} } if _, err := c.bot.SendMessage(ctx, tgMsg); err != nil { @@ -248,14 +242,9 @@ func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string, thread 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 + params.MessageThreadID = parseOptionalInt(threadID) _ = c.bot.SendChatAction(sctx, params) } @@ -318,15 +307,9 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string, th } 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} - } + params.MessageThreadID = parseOptionalInt(threadID) + if mid := parseOptionalInt(replyToMsgID); mid != 0 { + params.ReplyParameters = &telego.ReplyParameters{MessageID: mid} } pMsg, err := c.bot.SendMessage(ctx, params) @@ -628,6 +611,15 @@ func parseChatID(chatIDStr string) (int64, error) { 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 { if text == "" { return "" diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index 3a2f1aa66..e27c640d6 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -271,3 +271,21 @@ func TestSend_InvalidChatID(t *testing.T) { assert.True(t, errors.Is(err, channels.ErrSendFailed), "error should wrap ErrSendFailed") 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) + } +}