From ab0848293e6979ffba0a1487fa4c66f4c276ad8e Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 3 Nov 2026 15:54:39 +0000 Subject: [PATCH] feat(slack,telegram): stream files by handle Both add-ons implement the streaming side of the connector route: they read a file from the platform and put it to the host chunk by chunk, and they pull the file to send from the host and stream it into the platform upload. Neither holds more than one chunk in memory. Slack uploads with the external upload flow and tells Slack the length first. Telegram sends a multipart body of known length, and a picture over 10 MB goes as a document, because sendPhoto refuses more. The client of a streamed file has no total timeout; the context carries the bound. The manifests declare the limits that the platforms state, and files.transfer. Telegram moves to manifest API 2 for it. The limits are: Slack: 1 GB per file, as in the help center article https://slack.com/help/articles/201330736-Add-files-to-Slack The upload API names no per-file cap https://docs.slack.dev/reference/methods/files.getUploadURLExternal Telegram, public Bot API: bots download up to 20 MB with getFile and send up to 50 MB (10 MB for photos) https://core.telegram.org/bots/faq A local Bot API server lifts both (no download limit, 2000 MB upload) https://core.telegram.org/bots/api#using-a-local-bot-api-server The add-on has no setting for another server, so it declares the limits of the public one. Slack gets the host operation prefix connector. in the official host policy, as Telegram has, so it can call the two file operations. Assisted-by: Claude:claude-sonnet-5-5 go test --- addons/slack/cmd/slack/main.go | 4 ++ addons/slack/manifest.json | 5 ++ addons/slack/transport/api.go | 66 +++++++++++++++++- addons/slack/transport/runtime.go | 46 ++++++++++++- addons/telegram/cmd/telegram/main.go | 9 ++- addons/telegram/manifest.json | 7 +- addons/telegram/transport/botapi.go | 54 ++++++++++++++- addons/telegram/transport/media.go | 86 ++++++++++++++++++++++-- addons/telegram/transport/media_test.go | 6 +- addons/telegram/transport/runtime.go | 21 +++++- deploy/addons/host-policy.json | 3 +- pkg/addons/packaging/host_policy_test.go | 2 +- 12 files changed, 293 insertions(+), 16 deletions(-) diff --git a/addons/slack/cmd/slack/main.go b/addons/slack/cmd/slack/main.go index bc08822461..ca16e75348 100644 --- a/addons/slack/cmd/slack/main.go +++ b/addons/slack/cmd/slack/main.go @@ -10,6 +10,7 @@ package main import ( "os" + "github.com/mudler/LocalAI/addons/internal/hostclient" "github.com/mudler/LocalAI/addons/internal/protocol" "github.com/mudler/LocalAI/addons/slack/transport" "github.com/mudler/LocalAI/pkg/addons/contract" @@ -19,6 +20,9 @@ func main() { os.Exit(protocol.Run(os.Args, func(s protocol.Startup) contract.Runtime { rt := transport.New(&transport.API{}) rt.Socket.StateDir = s.DataDir + if h := hostclient.New(s); h != nil { + rt.Host = h + } return rt })) } diff --git a/addons/slack/manifest.json b/addons/slack/manifest.json index 44e653d9ad..82b1911891 100644 --- a/addons/slack/manifest.json +++ b/addons/slack/manifest.json @@ -18,6 +18,11 @@ "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", + "files": { + "receive": { "maxBytes": 1073741824 }, + "send": { "maxBytes": 1073741824 }, + "transfer": true + }, "page": "apps", "history": { "thread": true, "maxMessages": 30 }, "options": [ diff --git a/addons/slack/transport/api.go b/addons/slack/transport/api.go index 864ba1c77c..7fde77ea0d 100644 --- a/addons/slack/transport/api.go +++ b/addons/slack/transport/api.go @@ -7,6 +7,7 @@ package transport import ( + "bufio" "bytes" "context" "encoding/json" @@ -20,6 +21,7 @@ import ( "time" "github.com/mudler/LocalAI/addons/internal/connector" + "github.com/mudler/LocalAI/pkg/addons/contract" "github.com/mudler/LocalAI/pkg/httpclient" ) @@ -288,11 +290,18 @@ func (a *API) onFilesHost(address string) (*url.URL, error) { // upload: an upload address, the bytes, and the completion that posts the // file. It needs the scope files:write. func (a *API) Upload(ctx context.Context, bot, channel, thread, name string, data []byte) (string, error) { + return a.UploadStream(ctx, bot, channel, thread, name, int64(len(data)), bytes.NewReader(data)) +} + +// UploadStream is Upload for a file that is read as it is sent: body holds +// size bytes, which Slack is told first (the length of the upload). The body +// is read once; a failed upload starts again with a new body. +func (a *API) UploadStream(ctx context.Context, bot, channel, thread, name string, size int64, body io.Reader) (string, error) { var r struct { URL string `json:"upload_url"` ID string `json:"file_id"` } - if err := a.call(ctx, bot, "files.getUploadURLExternal", url.Values{"filename": {name}, "length": {strconv.Itoa(len(data))}}, &r); err != nil { + if err := a.call(ctx, bot, "files.getUploadURLExternal", url.Values{"filename": {name}, "length": {strconv.FormatInt(size, 10)}}, &r); err != nil { return "", err } // The address is signed for this one upload, so the bot token does not @@ -301,12 +310,13 @@ func (a *API) Upload(ctx context.Context, bot, channel, thread, name string, dat if err != nil || r.ID == "" { return "", errors.New("slack: the upload address is not on the files host of Slack") } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), bytes.NewReader(data)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), io.NopCloser(body)) if err != nil { return "", errors.New("slack: could not build an upload request") } + req.ContentLength = size req.Header.Set("Content-Type", "application/octet-stream") - resp, err := a.client().Do(req) + resp, err := a.streamClient().Do(req) if err != nil { if ctx.Err() != nil { return "", ctx.Err() @@ -332,6 +342,56 @@ func (a *API) Upload(ctx context.Context, bot, channel, thread, name string, dat return r.ID, nil } +// streamClient is the client of a file that is streamed. It has no overall +// timeout, because a large file takes minutes; the context carries the bound. +func (a *API) streamClient() *http.Client { + if a.Client != nil { + return a.Client + } + return httpclient.NewWithTimeout(contract.ConnectorTransferTimeout) +} + +// DownloadStream opens a file of Slack for reading. The caller closes it and +// holds no more than it reads. The bot token goes with the request, so the +// address must be on the files host (see onFilesHost). +func (a *API) DownloadStream(ctx context.Context, bot, address string) (io.ReadCloser, error) { + u, err := a.onFilesHost(address) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) + if err != nil { + return nil, errors.New("slack: could not build a download request") + } + req.Header.Set("Authorization", "Bearer "+bot) + resp, err := a.streamClient().Do(req) + if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, errors.New("slack: the file could not be downloaded") + } + switch { + case resp.StatusCode == http.StatusTooManyRequests: + _ = resp.Body.Close() + return nil, &connector.RetryAfter{Seconds: retryAfter(resp.Header.Get("Retry-After"))} + case resp.StatusCode != http.StatusOK: + _ = resp.Body.Close() + return nil, fmt.Errorf("slack: the download answered %d", resp.StatusCode) + } + br := bufio.NewReader(resp.Body) + if head, _ := br.Peek(64); bytes.HasPrefix(bytes.TrimSpace(head), []byte(" max { + return connector.MediaStream{}, ErrTooLarge + } + body, err := p.r.API.DownloadStream(ctx, b, f.Download) + if err != nil { + return connector.MediaStream{}, err + } + name := f.Name + if name == "" { + name = f.ID + } + return connector.MediaStream{Name: name, MIME: f.Mimetype, Size: f.Size, Body: body}, nil +} + +// SendStream uploads a file that the host hands over by handle. +func (p platform) SendStream(ctx context.Context, req contract.ConnectorRequest, body io.Reader) (string, error) { + b, err := bot(req.Token) + if err != nil { + return "", err + } + return p.r.API.UploadStream(ctx, b, req.Chat, req.Thread, req.Transfer.Name, req.Transfer.Size, body) +} + // mark is the key of the "working" message of a conversation. It holds a hash // of the token, never the token. func (r *Runtime) mark(req contract.ConnectorRequest) string { diff --git a/addons/telegram/cmd/telegram/main.go b/addons/telegram/cmd/telegram/main.go index d84eb7e36f..5b820ce417 100644 --- a/addons/telegram/cmd/telegram/main.go +++ b/addons/telegram/cmd/telegram/main.go @@ -8,11 +8,18 @@ package main import ( "os" + "github.com/mudler/LocalAI/addons/internal/hostclient" "github.com/mudler/LocalAI/addons/internal/protocol" "github.com/mudler/LocalAI/addons/telegram/transport" "github.com/mudler/LocalAI/pkg/addons/contract" ) func main() { - os.Exit(protocol.Run(os.Args, func(protocol.Startup) contract.Runtime { return transport.Runtime{API: &transport.API{}} })) + os.Exit(protocol.Run(os.Args, func(s protocol.Startup) contract.Runtime { + rt := transport.Runtime{API: &transport.API{}} + if h := hostclient.New(s); h != nil { + rt.Host = h + } + return rt + })) } diff --git a/addons/telegram/manifest.json b/addons/telegram/manifest.json index 7a9ac532af..adc9667643 100644 --- a/addons/telegram/manifest.json +++ b/addons/telegram/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,6 +18,11 @@ "description": "Talk to one of your saved agents from a Telegram bot.", "senderHint": "A chat ID, a user ID or an @username", "delivery": "pull", + "files": { + "receive": { "maxBytes": 20971520 }, + "send": { "maxBytes": 52428800 }, + "transfer": true + }, "page": "bots", "drafts": true, "previewMs": 400 diff --git a/addons/telegram/transport/botapi.go b/addons/telegram/transport/botapi.go index f17b877f21..39f0f72d08 100644 --- a/addons/telegram/transport/botapi.go +++ b/addons/telegram/transport/botapi.go @@ -128,14 +128,20 @@ func (t *API) post(ctx context.Context, token, method, contentType string, body } func (t *API) do(ctx context.Context, token, method, contentType string, body []byte, wait time.Duration, out any) error { + return t.doReader(ctx, token, method, contentType, bytes.NewReader(body), int64(len(body)), wait, out) +} + +// doReader is do for a body that is read as it is sent: size bytes long. +func (t *API) doReader(ctx context.Context, token, method, contentType string, body io.Reader, size int64, wait time.Duration, out any) error { base := t.BaseURL if base == "" { base = DefaultBase } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimSuffix(base, "/")+"/bot"+url.PathEscape(token)+"/"+method, bytes.NewReader(body)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimSuffix(base, "/")+"/bot"+url.PathEscape(token)+"/"+method, io.NopCloser(body)) if err != nil { return errors.New("telegram: could not build a Telegram request") } + req.ContentLength = size req.Header.Set("Content-Type", contentType) resp, err := t.client(wait).Do(req) if err != nil { @@ -494,6 +500,52 @@ func (t *API) SendFile(ctx context.Context, token string, chatID, thread int64, return messageID(raw), nil } +// photoMaxBytes is the largest photo that sendPhoto takes. A larger picture +// goes as a document, which Telegram does not recompress. +const photoMaxBytes = 10 << 20 + +// SendFileStream is SendFile for a file that is read as it is sent: body holds +// size bytes. The multipart head and tail are small and built here, so the +// request has a length and the file is never held in memory. Telegram answers +// 413 for a file over the limit of the Bot API server, which is ErrTooLarge. +func (t *API) SendFileStream(ctx context.Context, token string, chatID, thread int64, name, mimeType string, size int64, body io.Reader) (int64, error) { + method, field := "sendDocument", "document" + if photoTypes[mimeType] && size <= photoMaxBytes { + method, field = "sendPhoto", "photo" + } + var buf bytes.Buffer + w := multipart.NewWriter(&buf) + _ = w.WriteField("chat_id", strconv.FormatInt(chatID, 10)) + if thread != 0 { + _ = w.WriteField("message_thread_id", strconv.FormatInt(thread, 10)) + } + h := textproto.MIMEHeader{} + h.Set("Content-Disposition", mime.FormatMediaType("form-data", map[string]string{"name": field, "filename": name})) + if mimeType == "" { + mimeType = "application/octet-stream" + } + h.Set("Content-Type", mimeType) + if _, err := w.CreatePart(h); err != nil { + return 0, errors.New("telegram: could not build the upload") + } + head := buf.Len() + if err := w.Close(); err != nil { + return 0, errors.New("telegram: could not build the upload") + } + raw := buf.Bytes() + var out json.RawMessage + r := io.MultiReader(bytes.NewReader(raw[:head]), io.LimitReader(body, size), bytes.NewReader(raw[head:])) + err := redact(t.doReader(ctx, token, method, w.FormDataContentType(), r, int64(len(raw))+size, contract.ConnectorTransferTimeout, &out), token) + var ce *callError + if errors.As(err, &ce) && ce.Code == http.StatusRequestEntityTooLarge { + return 0, fmt.Errorf("telegram: the file has %d bytes: %w", size, ErrTooLarge) + } + if err != nil { + return 0, err + } + return messageID(out), nil +} + // Edit replaces the text of a message the bot sent with plain text. func (t *API) Edit(ctx context.Context, token string, chatID, messageID int64, text string) error { return t.edit(ctx, token, chatID, messageID, text, "") diff --git a/addons/telegram/transport/media.go b/addons/telegram/transport/media.go index 52a47d3b12..f88a168564 100644 --- a/addons/telegram/transport/media.go +++ b/addons/telegram/transport/media.go @@ -17,8 +17,9 @@ import ( ) // MaxDownloadBytes is the largest file that Telegram lets a bot download -// with getFile. The add-on hands over less: contract.MaxConnectorFileBytes, -// so that the file fits the envelope of the connector route. +// with getFile on the public Bot API (20 MB), as the Bots FAQ states: +// https://core.telegram.org/bots/faq. Without a handle the add-on hands over +// less: contract.MaxConnectorFileBytes, so that the file fits the envelope. const MaxDownloadBytes = 20 << 20 // ErrTooLarge means a file of a message is larger than the add-on hands @@ -90,13 +91,14 @@ func mediaRefs(m tgMessage) []string { return nil } -// bestPhoto is the largest size of a photo that the add-on can hand over. +// bestPhoto is the largest size of a photo that the add-on can hand over +// (at most MaxDownloadBytes; the host's limit is checked when it is read). // Telegram sends each photo in several sizes; when none says that it fits, // the smallest one is tried. func bestPhoto(sizes []tgPhoto) tgPhoto { best, found := sizes[0], false for _, p := range sizes { - if p.Size <= 0 || p.Size > contract.MaxConnectorFileBytes { + if p.Size <= 0 || p.Size > MaxDownloadBytes { continue } if !found || p.Width*p.Height > best.Width*best.Height { @@ -188,6 +190,82 @@ func (t *API) Media(ctx context.Context, token, ref string) (contract.ConnectorF return contract.ConnectorFile{Name: contract.SafeFileName(strings.ReplaceAll(name, "_", "-") + ext), MIME: mimeType, Data: data}, nil } +// MediaStream opens the file of a reference for streaming to the host. The +// size that getFile states is checked against max before a byte is read, and +// the type is decided from the first bytes, which stay in the stream. The name +// is made here, as in Media. +func (t *API) MediaStream(ctx context.Context, token, ref string, max int64) (connector.MediaStream, error) { + kind, id, ok := strings.Cut(ref, ":") + if !ok || !refKinds[kind] || !fileID.MatchString(id) { + return connector.MediaStream{}, errors.New("telegram: not a file reference") + } + var f struct { + UniqueID string `json:"file_unique_id"` + Size int64 `json:"file_size"` + Path string `json:"file_path"` + } + err := t.call(ctx, token, "getFile", map[string]any{"file_id": id}, 0, &f) + switch { + case refusedBecause(err, "file is too big"): + return connector.MediaStream{}, fmt.Errorf("telegram: Telegram lets a bot download %d MB at most: %w", MaxDownloadBytes>>20, ErrTooLarge) + case err != nil: + return connector.MediaStream{}, err + case f.Size > max: + return connector.MediaStream{}, fmt.Errorf("telegram: the file has %d bytes: %w", f.Size, ErrTooLarge) + case !filePath.MatchString(f.Path): + return connector.MediaStream{}, errors.New("telegram: Telegram named no file to download") + } + body, err := t.getStream(ctx, token, f.Path) + if err != nil { + return connector.MediaStream{}, redact(err, token) + } + rd, head := connector.Sniff(body, 512) + mimeType, ext := fileType(kind, head) + if mimeType == "" { + _ = body.Close() + return connector.MediaStream{}, fmt.Errorf("telegram: the %s is not of a type the add-on reads", strings.ReplaceAll(kind, "_", " ")) + } + name := kind + if fileID.MatchString(f.UniqueID) { + name += "-" + f.UniqueID + } + return connector.MediaStream{Name: contract.SafeFileName(strings.ReplaceAll(name, "_", "-") + ext), MIME: mimeType, Size: f.Size, + Body: struct { + io.Reader + io.Closer + }{rd, body}}, nil +} + +// getStream opens the download of a file of the bot. The context carries the +// bound of the transfer; the client has no total timeout. +func (t *API) getStream(ctx context.Context, token, path string) (io.ReadCloser, error) { + base := t.BaseURL + if base == "" { + base = DefaultBase + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimSuffix(base, "/")+"/file/bot"+url.PathEscape(token)+"/"+path, nil) + if err != nil { + return nil, errors.New("telegram: could not build a download request") + } + resp, err := t.client(contract.ConnectorTransferTimeout).Do(req) + if err != nil { + var ue *url.Error + if errors.As(err, &ue) { + err = ue.Err + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, fmt.Errorf("telegram: the download failed: %w", err) + } + if resp.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)) + _ = resp.Body.Close() + return nil, fmt.Errorf("telegram: the download answered %d", resp.StatusCode) + } + return resp.Body, nil +} + // download reads a file of the bot, at most contract.MaxConnectorFileBytes. // The token is in the path, as for the methods, and is taken out of every // error. diff --git a/addons/telegram/transport/media_test.go b/addons/telegram/transport/media_test.go index 91d905e73a..70d5d1533d 100644 --- a/addons/telegram/transport/media_test.go +++ b/addons/telegram/transport/media_test.go @@ -55,7 +55,7 @@ var _ = Describe("the files of messages", func() { ups, err := f.client().Updates(ctx, secretToken, 1, 0) Expect(err).NotTo(HaveOccurred()) Expect(ups).To(HaveLen(7)) - Expect(ups[0].Media).To(Equal([]string{"photo:big"}), "the largest size that fits") + Expect(ups[0].Media).To(Equal([]string{"photo:huge"}), "the largest size that a bot may download") Expect(ups[0].Text).To(Equal("what is this?")) Expect(ups[0].Unsupported).To(BeFalse()) Expect(ups[1].Media).To(Equal([]string{"voice:v1"})) @@ -71,7 +71,9 @@ var _ = Describe("the files of messages", func() { It("finds a mention of the bot in the caption of a photo in a group", func() { f := newFakeBot() - f.reply["getMe"] = func(map[string]any) (int, string) { return 200, `{"ok":true,"result":{"id":123456,"username":"helper_bot"}}` } + f.reply["getMe"] = func(map[string]any) (int, string) { + return 200, `{"ok":true,"result":{"id":123456,"username":"helper_bot"}}` + } f.reply["getUpdates"] = func(map[string]any) (int, string) { return 200, updates(`{"update_id":1,"message":{"chat":{"id":-300,"type":"supergroup"},"from":{"id":7},"caption":"@helper_bot read this","caption_entities":[{"type":"mention","offset":0,"length":11}],"photo":[{"file_id":"p1","file_size":10,"width":1,"height":1}]}}`) } diff --git a/addons/telegram/transport/runtime.go b/addons/telegram/transport/runtime.go index e8545e2751..360972cd21 100644 --- a/addons/telegram/transport/runtime.go +++ b/addons/telegram/transport/runtime.go @@ -2,6 +2,7 @@ package transport import ( "context" + "io" "strconv" "github.com/mudler/LocalAI/addons/internal/connector" @@ -21,6 +22,9 @@ const Kind = "telegram" type Runtime struct { idle.Runtime API *API + // Host is the host operation channel, with which a file moves by handle. + // Nil: the add-on keeps the inline path. + Host connector.Host } var _ contract.ConnectorRuntime = Runtime{} @@ -32,7 +36,7 @@ func (r Runtime) Connector(ctx context.Context, req contract.ConnectorRequest) ( if api == nil { api = &API{} } - return connector.Serve(ctx, platform{api}, req) + return connector.ServeWith(ctx, platform{api}, r.Host, req) } // platform is the Bot API as a connector.Platform. Telegram names chats and @@ -100,6 +104,21 @@ func (p platform) SendFile(ctx context.Context, req contract.ConnectorRequest) ( return messageRef(id), err } +// MediaStream streams a file of a message to the host by handle. +func (p platform) MediaStream(ctx context.Context, token, ref string, max int64) (connector.MediaStream, error) { + return p.api.MediaStream(ctx, token, ref, max) +} + +// SendStream sends a file that the host hands over by handle. +func (p platform) SendStream(ctx context.Context, req contract.ConnectorRequest, body io.Reader) (string, error) { + chat, thread, err := where(req) + if err != nil { + return "", err + } + id, err := p.api.SendFileStream(ctx, req.Token, chat, thread, req.Transfer.Name, req.Transfer.MIME, req.Transfer.Size, body) + return messageRef(id), err +} + // SendRich sends one part of an answer as MarkdownV2, or as plain text when // Telegram refuses the markup. func (p platform) SendRich(ctx context.Context, req contract.ConnectorRequest) (string, error) { diff --git a/deploy/addons/host-policy.json b/deploy/addons/host-policy.json index 152459252f..354075796a 100644 --- a/deploy/addons/host-policy.json +++ b/deploy/addons/host-policy.json @@ -6,6 +6,7 @@ "sandbox": ["workspace.", "egress."], "office": ["workspace."], "web": ["workspace."], - "telegram": ["connector."] + "telegram": ["connector."], + "slack": ["connector."] } } diff --git a/pkg/addons/packaging/host_policy_test.go b/pkg/addons/packaging/host_policy_test.go index 13994fcd18..d4b8fb179a 100644 --- a/pkg/addons/packaging/host_policy_test.go +++ b/pkg/addons/packaging/host_policy_test.go @@ -18,7 +18,7 @@ var _ = Describe("official host policy deployment", func() { grants, err := config.LoadAddonHostPolicy(filepath.Join(root, "deploy", "addons", "host-policy.json")) Expect(err).NotTo(HaveOccurred()) // Deployment-owned exception: historical identity grants, not core defaults. - legacy := "fine-tuning=finetune.,quantization=quantize.,sandbox=workspace.,sandbox=egress.,office=workspace.,web=workspace.,telegram=connector." + legacy := "fine-tuning=finetune.,quantization=quantize.,sandbox=workspace.,sandbox=egress.,office=workspace.,web=workspace.,telegram=connector.,slack=connector." want, err := config.ParseAddonHostOps(legacy) Expect(err).NotTo(HaveOccurred()) Expect(grants).To(Equal(want))