feat(slack): read the earlier messages of a thread

The Slack add-on declares history.thread (manifest api 2) and serves the
op history with conversations.replies. It pages through the thread, keeps
the newest messages, marks the bot's own, names the authors from a small
cache and gives file names only. A thread of more than 1000 messages is
refused instead of shown from its middle. The fake Slack serves
conversations.replies.

Assisted-by: Claude Code:claude-sonnet-5-5
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

addons/slack/manifest.json

@@ -6,7 +6,7 @@
   "os": "linux",
   "arch": "amd64",
   "host": { "min": 1, "max": 1 },
-  "api": { "min": 1, "max": 1 },
+  "api": { "min": 2, "max": 2 },
   "protocol": { "min": 1, "max": 1 },
   "dataFormat": { "min": 1, "max": 1 },
   "capabilities": {
@@ -18,7 +18,8 @@
           "description": "Talk to one of your saved agents from Slack, in direct messages and in channel threads.",
           "senderHint": "A user ID (U...), a channel ID (C...) or an @name",
           "delivery": "pull",
-          "page": "apps"
+          "page": "apps",
+          "history": { "thread": true, "maxMessages": 30 }
         }
       ]
     },

addons/slack/slacktest/fake.go

@@ -58,6 +58,8 @@ type Server struct {
 	refuse   map[string]string
 	limit    map[string]int
 	calls    map[string]int
+	thread   map[string][]map[string]any // by channel and thread: the messages conversations.replies returns
+	failing  map[string]string           // method -> error code, once
 	seq      int
 	links    int
 	upgrader websocket.Upgrader
@@ -93,7 +95,7 @@ func (c *conn) send(v any) error {
 // New starts a fake Slack. Close it when the test ends.
 func New() *Server {
 	s := &Server{apps: map[string]*App{}, tickets: map[string]string{}, conns: map[string][]*conn{}, pending: map[string][]any{},
-		users: map[string]map[string]any{}, files: map[string]file{}, uploads: map[string]*Upload{}, refuse: map[string]string{}, limit: map[string]int{}, calls: map[string]int{}}
+		users: map[string]map[string]any{}, files: map[string]file{}, uploads: map[string]*Upload{}, refuse: map[string]string{}, limit: map[string]int{}, calls: map[string]int{}, thread: map[string][]map[string]any{}, failing: map[string]string{}}
 	mux := http.NewServeMux()
 	mux.HandleFunc("/api/", s.api)
 	mux.HandleFunc("/link", s.link)
@@ -145,6 +147,27 @@ func (s *Server) AddUser(id, name, real string) {
 	s.users[id] = map[string]any{"id": id, "name": name, "profile": map[string]any{"real_name": real, "display_name": ""}}
 }
 
+// AddThreadMessage appends a message to a thread, as conversations.replies
+// returns it. The first message of a thread is its parent: ts equals thread.
+// fields are the keys of a Slack message, such as user, text, files or bot_id;
+// ts and thread_ts are set here.
+func (s *Server) AddThreadMessage(channel, thread, ts string, fields map[string]any) {
+	m := map[string]any{"type": "message", "ts": ts, "thread_ts": thread}
+	for k, v := range fields {
+		m[k] = v
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.thread[channel+"\x00"+thread] = append(s.thread[channel+"\x00"+thread], m)
+}
+
+// FailOnce makes the next call of a method answer ok:false with code.
+func (s *Server) FailOnce(method, code string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.failing[method] = code
+}
+
 // AddFile registers a file that a message can carry.
 func (s *Server) AddFile(id, name, mime string, data []byte) {
 	s.mu.Lock()
@@ -313,6 +336,10 @@ func (s *Server) api(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	code := s.refuse[token]
+	if once, ok := s.failing[method]; ok {
+		delete(s.failing, method)
+		code = once
+	}
 	app := s.apps[token]
 	s.mu.Unlock()
 	switch {
@@ -365,6 +392,8 @@ func (s *Server) api(w http.ResponseWriter, r *http.Request) {
 		s.posts = append(s.posts, Post{Token: token, Channel: channel, Thread: thread, Text: text, TS: ts, Edited: method == "chat.update"})
 		s.mu.Unlock()
 		s.reply(w, map[string]any{"ok": true, "channel": channel, "ts": ts})
+	case "conversations.replies":
+		s.replies(w, r)
 	case "users.info":
 		s.mu.Lock()
 		u, ok := s.users[r.Form.Get("user")]
@@ -420,6 +449,41 @@ func (s *Server) api(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// replies answers conversations.replies: the messages of a thread, oldest
+// first, those before latest when it is given, a page of limit messages from
+// the cursor on.
+func (s *Server) replies(w http.ResponseWriter, r *http.Request) {
+	channel, thread := r.Form.Get("channel"), r.Form.Get("ts")
+	s.mu.Lock()
+	all := append([]map[string]any(nil), s.thread[channel+"\x00"+thread]...)
+	s.mu.Unlock()
+	if len(all) == 0 {
+		s.reply(w, map[string]any{"ok": false, "error": "thread_not_found"})
+		return
+	}
+	if latest, err := strconv.ParseFloat(r.Form.Get("latest"), 64); err == nil {
+		kept := all[:0:0]
+		for _, m := range all {
+			if t, _ := strconv.ParseFloat(m["ts"].(string), 64); t < latest {
+				kept = append(kept, m)
+			}
+		}
+		all = kept
+	}
+	limit, _ := strconv.Atoi(r.Form.Get("limit"))
+	if limit < 1 || limit > 1000 {
+		limit = 100
+	}
+	from, _ := strconv.Atoi(r.Form.Get("cursor"))
+	from = min(max(from, 0), len(all))
+	to := min(from+limit, len(all))
+	out := map[string]any{"ok": true, "messages": all[from:to], "has_more": to < len(all)}
+	if to < len(all) {
+		out["response_metadata"] = map[string]any{"next_cursor": strconv.Itoa(to)}
+	}
+	s.reply(w, out)
+}
+
 func (s *Server) download(w http.ResponseWriter, r *http.Request) {
 	s.mu.Lock()
 	app := s.apps[bearer(r)]

addons/slack/transport/api.go

@@ -205,6 +205,43 @@ func (a *API) UserInfo(ctx context.Context, bot, id string) (User, error) {
 	return r.User, err
 }
 
+// Reply is a message of a thread as conversations.replies returns it.
+type Reply struct {
+	User     string `json:"user"`
+	BotID    string `json:"bot_id"`
+	Username string `json:"username"`
+	Subtype  string `json:"subtype"`
+	Text     string `json:"text"`
+	TS       string `json:"ts"`
+	Files    []struct {
+		Name string `json:"name"`
+		ID   string `json:"id"`
+	} `json:"files"`
+}
+
+// Replies returns one page of the messages of a thread, oldest first, the
+// parent message included. Only messages before latest are returned, when it
+// is set. It needs the scopes channels:history, groups:history, im:history and
+// mpim:history, by the kind of the conversation. The cursor is empty for the
+// first page; the next one is empty on the last.
+func (a *API) Replies(ctx context.Context, bot, channel, thread, latest, cursor string, limit int) ([]Reply, string, error) {
+	form := url.Values{"channel": {channel}, "ts": {thread}, "limit": {strconv.Itoa(limit)}}
+	if latest != "" {
+		form.Set("latest", latest)
+	}
+	if cursor != "" {
+		form.Set("cursor", cursor)
+	}
+	var r struct {
+		Messages []Reply `json:"messages"`
+		Meta     struct {
+			Next string `json:"next_cursor"`
+		} `json:"response_metadata"`
+	}
+	err := a.call(ctx, bot, "conversations.replies", form, &r)
+	return r.Messages, r.Meta.Next, err
+}
+
 // File is a file that a message carries.
 type File struct {
 	ID       string `json:"id"`

addons/slack/transport/history_test.go

@@ -0,0 +1,104 @@
+// SPDX-License-Identifier: MIT
+package transport_test
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/mudler/LocalAI/addons/slack/slacktest"
+	"github.com/mudler/LocalAI/pkg/addons/contract"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("the history of a thread", func() {
+	ctx := context.Background()
+	history := func(f *slacktest.Server, limit int, before string) contract.ConnectorResponse {
+		r := runtimeOf(f)
+		out, err := r.Connector(ctx, contract.ConnectorRequest{Op: "history", Kind: "slack", Token: secret, Chat: "C1", Thread: "100.000001", Limit: limit, Before: before})
+		Expect(err).NotTo(HaveOccurred())
+		return out
+	}
+	add := func(f *slacktest.Server, ts string, fields map[string]any) {
+		f.AddThreadMessage("C1", "100.000001", ts, fields)
+	}
+
+	It("returns the messages oldest first with names, the bot's own marked, and files by name", func() {
+		f := fake()
+		f.AddUser("U2", "bo", "Bo")
+		add(f, "100.000001", map[string]any{"user": "U1", "text": "the deploy <https://x.example|log> failed"})
+		add(f, "100.000002", map[string]any{"user": "U2", "text": "retrying", "files": []any{map[string]any{"id": "F1", "name": "out.txt"}}})
+		add(f, "100.000003", map[string]any{"user": "UBOT", "text": "Done."})
+		add(f, "100.000004", map[string]any{"user": "U1", "text": "<@UBOT> and now?"})
+		out := history(f, 30, "100.000004")
+		Expect(out.Error).To(BeNil())
+		Expect(out.Truncated).To(BeFalse())
+		Expect(out.Messages).To(HaveLen(3), "the message named by before is not returned")
+		m := out.Messages
+		Expect([]string{m[0].SenderName, m[0].SenderUsername, m[0].Text}).To(Equal([]string{"Ada Lovelace", "ada", "the deploy log (https://x.example) failed"}))
+		Expect(m[1].SenderName).To(Equal("Bo"))
+		Expect([]any{m[1].Media, m[1].MediaNames}).To(Equal([]any{1, []string{"out.txt"}}))
+		Expect(m[2].Own).To(BeTrue())
+		Expect(m[0].Own).To(BeFalse())
+		Expect(m[0].TS.Unix()).To(Equal(int64(100)))
+	})
+
+	It("keeps the newest messages within the limit and says that older ones exist", func() {
+		f := fake()
+		for i := 1; i <= 10; i++ {
+			add(f, fmt.Sprintf("100.%06d", i), map[string]any{"user": "U1", "text": fmt.Sprintf("m%d", i)})
+		}
+		out := history(f, 4, "")
+		Expect(out.Truncated).To(BeTrue())
+		Expect(out.Messages).To(HaveLen(4))
+		Expect(out.Messages[0].Text).To(Equal("m7"))
+		Expect(out.Messages[3].Text).To(Equal("m10"))
+	})
+
+	It("reads a long thread page by page and looks each person up once", func() {
+		f := fake()
+		for i := 1; i <= 450; i++ {
+			add(f, fmt.Sprintf("100.%06d", i), map[string]any{"user": "U1", "text": fmt.Sprintf("m%d", i)})
+		}
+		out := history(f, 3, "")
+		Expect(out.Messages[2].Text).To(Equal("m450"))
+		Expect(f.Calls("conversations.replies")).To(Equal(3))
+		Expect(f.Calls("users.info")).To(Equal(1))
+	})
+
+	It("refuses a thread that is longer than one history reads, rather than show its middle", func() {
+		f := fake()
+		for i := 1; i <= 1100; i++ {
+			add(f, fmt.Sprintf("100.%06d", i), map[string]any{"user": "U1", "text": "x"})
+		}
+		out := history(f, 3, "")
+		Expect(out.Error).NotTo(BeNil())
+		Expect(out.Error.Code).To(Equal(contract.ConnectorFailed))
+		Expect(f.Calls("conversations.replies")).To(Equal(5))
+	})
+
+	It("falls back to the user ID when the name cannot be read, and skips joins", func() {
+		f := fake()
+		add(f, "100.000001", map[string]any{"user": "U9", "text": "hello"})
+		add(f, "100.000002", map[string]any{"user": "U1", "subtype": "channel_join", "text": "joined"})
+		out := history(f, 30, "")
+		Expect(out.Messages).To(HaveLen(1))
+		Expect([]string{out.Messages[0].SenderID, out.Messages[0].SenderName}).To(Equal([]string{"U9", ""}))
+	})
+
+	It("types a missing scope, a missing thread and a rate limit", func() {
+		f := fake()
+		add(f, "100.000001", map[string]any{"user": "U1", "text": "hello"})
+		f.FailOnce("conversations.replies", "missing_scope")
+		out := history(f, 30, "")
+		Expect(out.Error.Code).To(Equal(contract.ConnectorFailed))
+		Expect(out.Error.Message).To(ContainSubstring("missing_scope"))
+		f.RateLimitOnce("conversations.replies", 7)
+		out = history(f, 30, "")
+		Expect(out.Error.Code).To(Equal(contract.ConnectorRateLimited))
+		Expect(out.Error.RetryAfter).To(Equal(7))
+		Expect(f.Calls("conversations.replies")).To(Equal(2), "a rate limit is not waited out")
+		Expect(strings.Count(out.Error.Message, "xoxb")).To(BeZero())
+	})
+})

addons/slack/transport/runtime.go

@@ -8,6 +8,8 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
+	"slices"
+	"strconv"
 	"sync"
 	"time"
 
@@ -68,10 +70,11 @@ func (r *Runtime) Close() { r.Socket.Close() }
 type platform struct{ r *Runtime }
 
 var (
-	_ connector.Formatter   = platform{}
-	_ connector.Splitter    = platform{}
-	_ connector.MediaReader = platform{}
-	_ connector.FileSender  = platform{}
+	_ connector.Formatter     = platform{}
+	_ connector.Splitter      = platform{}
+	_ connector.MediaReader   = platform{}
+	_ connector.FileSender    = platform{}
+	_ connector.HistoryReader = platform{}
 )
 
 func (platform) Kind() string                  { return Kind }
@@ -223,3 +226,77 @@ func (r *Runtime) take(req contract.ConnectorRequest) (string, bool) {
 	delete(r.working, key)
 	return ph.ts, ok && time.Since(ph.at) < placeholderLife
 }
+
+// Limits of the history of a thread.
+const (
+	// repliesPage is the size of one page of conversations.replies.
+	repliesPage = 200
+	// repliesPages is how many pages one history reads. Slack returns a thread
+	// oldest first, so the newest messages are at its end: a thread longer
+	// than repliesPages pages is refused, not shown from its middle.
+	repliesPages = 5
+	// historyNames bounds the people whose names one history looks up.
+	historyNames = 30
+)
+
+// errThreadTooLong means the thread has more messages than one history reads.
+var errThreadTooLong = errors.New("slack: the thread is too long to read")
+
+// History returns the earlier messages of a thread: the newest req.Limit
+// before req.Before, oldest first, with the names of their authors. A message
+// of the bot is marked Own. Messages that no person or app wrote (joins,
+// deletions) are left out. The file bytes are never read, only their names.
+func (p platform) History(ctx context.Context, req contract.ConnectorRequest) ([]contract.ConnectorHistoryMessage, bool, error) {
+	b, err := bot(req.Token)
+	if err != nil {
+		return nil, false, err
+	}
+	var all []Reply
+	cursor := ""
+	for page := 0; ; page++ {
+		if page == repliesPages {
+			return nil, false, errThreadTooLong
+		}
+		got, next, err := p.r.API.Replies(ctx, b, req.Chat, req.Thread, req.Before, cursor, repliesPage)
+		if err != nil {
+			return nil, false, err
+		}
+		all = append(all, got...)
+		if next == "" {
+			break
+		}
+		cursor = next
+	}
+	all = slices.DeleteFunc(all, func(m Reply) bool {
+		return m.TS == req.Before || (m.Subtype != "" && !personal[m.Subtype] && m.Subtype != "bot_message")
+	})
+	more := len(all) > req.Limit
+	if more {
+		all = all[len(all)-req.Limit:]
+	}
+	self := p.r.Socket.botUser(ctx, req.Token)
+	users := p.r.Socket.namer(ParseTokens(req.Token))
+	out := make([]contract.ConnectorHistoryMessage, 0, len(all))
+	for _, m := range all {
+		h := contract.ConnectorHistoryMessage{MessageID: m.TS, SenderID: m.User, Text: Plain(m.Text, ""), Own: self != "" && m.User == self}
+		if sec, err := strconv.ParseFloat(m.TS, 64); err == nil {
+			h.TS = time.UnixMicro(int64(sec * 1e6)).UTC()
+		}
+		switch {
+		case m.User != "":
+			if u, ok := users(ctx, m.User); ok {
+				h.SenderName, h.SenderUsername = u.Display(), u.Name
+			}
+		case m.BotID != "":
+			h.SenderID, h.SenderName = m.BotID, m.Username
+		}
+		for _, f := range m.Files {
+			h.Media++
+			if len(h.MediaNames) < contract.MaxHistoryMedia && f.Name != "" {
+				h.MediaNames = append(h.MediaNames, f.Name)
+			}
+		}
+		out = append(out, h)
+	}
+	return out, more, nil
+}

addons/slack/transport/socket.go

@@ -414,3 +414,49 @@ func (s *Socket) user(ctx context.Context, ss *session, id string) (User, bool)
 	ss.mu.Unlock()
 	return u, true
 }
+
+// botUser is the Slack user ID of the bot of a secret: from the open link when
+// there is one, else from the Web API. It is empty when it cannot be told.
+func (s *Socket) botUser(ctx context.Context, secret string) string {
+	t := ParseTokens(secret)
+	s.mu.Lock()
+	ss := s.sessions[keyOf(t)]
+	s.mu.Unlock()
+	if ss != nil {
+		ss.mu.Lock()
+		id := ss.bot.UserID
+		ss.mu.Unlock()
+		if id != "" {
+			return id
+		}
+	}
+	id, err := s.API.AuthTest(ctx, t.Bot)
+	if err != nil {
+		return ""
+	}
+	return id.UserID
+}
+
+// namer returns a function that looks up the name of a person. It shares the
+// cache of the link of the tokens, when one is open; else it keeps a cache
+// for the one call. It looks up at most historyNames people, then gives IDs.
+func (s *Socket) namer(t Tokens) func(ctx context.Context, id string) (User, bool) {
+	s.mu.Lock()
+	ss := s.sessions[keyOf(t)]
+	s.mu.Unlock()
+	if ss == nil {
+		ss = &session{tokens: t, users: map[string]cachedUser{}}
+	}
+	asked := 0
+	return func(ctx context.Context, id string) (User, bool) {
+		ss.mu.Lock()
+		c, ok := ss.users[id]
+		ss.mu.Unlock()
+		if !ok || time.Since(c.at) >= userLife {
+			if asked++; asked > historyNames {
+				return User{}, false
+			}
+		}
+		return s.user(ctx, ss, id)
+	}
+}

tests/studiobot/transport.go

@@ -38,7 +38,7 @@ func TelegramPaced(api *transport.API) Addon {
 
 // Slack is the Slack add-on over api. The caller closes the runtime.
 func Slack(rt *slack.Runtime) Addon {
-	return Addon{ID: "slack", Kind: contract.ConnectorKind{ID: slack.Kind, Title: "Slack", Delivery: contract.DeliveryPull}, Runtime: rt}
+	return Addon{ID: "slack", Kind: contract.ConnectorKind{ID: slack.Kind, Title: "Slack", Delivery: contract.DeliveryPull, History: &contract.ConnectorHistory{Thread: true, MaxMessages: 30}}, Runtime: rt}
 }
 
 // BotTransport is a studio.ConnectorTransport that serves the real connector