fix(channels): defer media download, fix compaction status & context cancel

- Defer Telegram media download until after mention gate — pending
  history now uses lightweight tags (no download) saving bandwidth
  and avoiding errors on large files
- Fix compaction status query: has_summary now false when new messages
  arrive after last compaction, re-enabling the compact button
- Fix HTTP compact endpoint: detach context from request so LLM
  summarization isn't cancelled when browser closes connection
This commit is contained in:
viettranx committed 2026-03-10 17:12:13 +07:00
1 parent f1953203c4
commit 9181eebcea
4 files changed
+129 -78

No files matched your search

+74 -75
View File
@@ -191,81 +191,14 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
content += message.Caption
}
// Process media (photos, audio, voice, documents)
mediaList := c.resolveMedia(ctx, message)
// When user replies to a message with media, also resolve media from the replied message.
// This re-downloads the file from Telegram so the agent can analyze it even if the
// original turn was compacted out of session history.
if message.ReplyToMessage != nil && len(mediaList) == 0 {
replyMedia := c.resolveMedia(ctx, message.ReplyToMessage)
if len(replyMedia) > 0 {
mediaList = append(mediaList, replyMedia...)
slog.Debug("telegram: resolved media from replied message",
"reply_msg_id", message.ReplyToMessage.MessageID,
"media_count", len(replyMedia),
)
}
}
var mediaFiles []bus.MediaFile
if len(mediaList) > 0 {
// First pass: process each media item.
// For audio/voice: attempt STT transcription so that buildMediaTags can embed the transcript.
// For documents: extract text content to append after the media tags.
// Note: buildMediaTags is called AFTER this loop so it picks up populated Transcript fields.
var extraContent string
for i := range mediaList {
m := &mediaList[i]
switch m.Type {
case "audio", "voice":
transcript, sttErr := c.transcribeAudio(ctx, m.FilePath)
if sttErr != nil {
slog.Warn("telegram: STT transcription failed, falling back to media placeholder",
"type", m.Type, "error", sttErr,
)
} else {
m.Transcript = transcript
}
case "document":
// Extract text content from documents
if m.FileName != "" && m.FilePath != "" {
docContent, err := extractDocumentContent(m.FilePath, m.FileName)
if err != nil {
slog.Warn("document extraction failed", "file", m.FileName, "error", err)
} else if docContent != "" {
extraContent += "\n\n" + docContent
}
}
case "video", "animation":
// Video files are handled by the read_video tool via MediaRef pipeline.
}
if m.FilePath != "" {
mediaFiles = append(mediaFiles, bus.MediaFile{
Path: m.FilePath,
MimeType: m.ContentType,
})
}
}
// Build media tags AFTER the processing loop so transcript fields are populated.
mediaTags := buildMediaTags(mediaList)
if mediaTags != "" {
if content != "" {
content = mediaTags + "\n\n" + content
} else {
content = mediaTags
}
}
// Append any extra content accumulated during processing (doc text, video note, etc.)
if extraContent != "" {
content += extraContent
// Build lightweight media tags from message metadata (no download).
// Used for pending history recording and bot command handling.
// Actual media download + processing is deferred until after mention gating.
if tags := lightweightMediaTags(message); tags != "" {
if content != "" {
content = tags + "\n\n" + content
} else {
content = tags
}
}
@@ -338,6 +271,72 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
}
}
// --- Media download (only when bot will process the message) ---
// Deferred until after mention + pairing gates to avoid downloading
// media for messages that only get recorded in pending history.
mediaList := c.resolveMedia(ctx, message)
if message.ReplyToMessage != nil && len(mediaList) == 0 {
replyMedia := c.resolveMedia(ctx, message.ReplyToMessage)
if len(replyMedia) > 0 {
mediaList = append(mediaList, replyMedia...)
slog.Debug("telegram: resolved media from replied message",
"reply_msg_id", message.ReplyToMessage.MessageID,
"media_count", len(replyMedia),
)
}
}
var mediaFiles []bus.MediaFile
if len(mediaList) > 0 {
var extraContent string
for i := range mediaList {
m := &mediaList[i]
switch m.Type {
case "audio", "voice":
transcript, sttErr := c.transcribeAudio(ctx, m.FilePath)
if sttErr != nil {
slog.Warn("telegram: STT transcription failed",
"type", m.Type, "error", sttErr)
} else {
m.Transcript = transcript
}
case "document":
if m.FileName != "" && m.FilePath != "" {
docContent, err := extractDocumentContent(m.FilePath, m.FileName)
if err != nil {
slog.Warn("document extraction failed", "file", m.FileName, "error", err)
} else if docContent != "" {
extraContent += "\n\n" + docContent
}
}
case "video", "animation":
// Handled by read_video tool via MediaRef pipeline.
}
if m.FilePath != "" {
mediaFiles = append(mediaFiles, bus.MediaFile{
Path: m.FilePath,
MimeType: m.ContentType,
})
}
}
// Replace lightweight media tags with full tags (includes transcripts).
fullTags := buildMediaTags(mediaList)
lightTags := lightweightMediaTags(message)
if lightTags != "" && fullTags != "" {
content = strings.Replace(content, lightTags, fullTags, 1)
} else if fullTags != "" {
if content != "" {
content = fullTags + "\n\n" + content
} else {
content = fullTags
}
}
if extraContent != "" {
content += extraContent
}
}
slog.Debug("telegram message received",
"sender_id", senderID,
"chat_id", fmt.Sprintf("%d", chatID),
+37
View File
@@ -8,6 +8,7 @@ import (
"net/http"
"os"
"path/filepath"
"strings"
"time"
"github.com/mymmrac/telego"
@@ -239,3 +240,39 @@ func buildMediaTags(mediaList []MediaInfo) string {
func extractDocumentContent(filePath, fileName string) (string, error) {
return media.ExtractDocumentContent(filePath, fileName)
}
// lightweightMediaTags builds media placeholder tags from Telegram message metadata
// without downloading any files. Used for pending history recording when bot is not mentioned.
func lightweightMediaTags(msg *telego.Message) string {
var tags []string
if msg.Photo != nil && len(msg.Photo) > 0 {
tags = append(tags, "<media:image>")
}
if msg.Video != nil {
tags = append(tags, "<media:video>")
}
if msg.VideoNote != nil {
tags = append(tags, "<media:video>")
}
if msg.Animation != nil {
tags = append(tags, "<media:video>")
}
if msg.Audio != nil {
tags = append(tags, "<media:audio>")
}
if msg.Voice != nil {
tags = append(tags, "<media:voice>")
}
if msg.Document != nil {
name := msg.Document.FileName
if name != "" {
tags = append(tags, fmt.Sprintf("<media:document name=%q>", name))
} else {
tags = append(tags, "<media:document>")
}
}
if len(tags) == 0 {
return ""
}
return strings.Join(tags, "\n")
}
+3 -1
View File
@@ -151,7 +151,9 @@ func (h *PendingMessagesHandler) handleCompact(w http.ResponseWriter, r *http.Re
return
}
ctx, cancel := context.WithTimeout(r.Context(), 180*time.Second)
// Detach from HTTP request context so LLM summarization isn't
// cancelled when the browser closes the connection.
ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), 180*time.Second)
defer cancel()
keepRecent := h.keepRecent
+15 -2
View File
@@ -150,9 +150,22 @@ func (s *PGPendingMessageStore) ListGroups(ctx context.Context) ([]store.Pending
rows, err := s.db.QueryContext(ctx,
`SELECT channel_name, history_key,
COUNT(*) AS message_count,
BOOL_OR(is_summary) AS has_summary,
BOOL_OR(is_summary)
AND NOT EXISTS (
SELECT 1 FROM channel_pending_messages n
WHERE n.channel_name = m.channel_name
AND n.history_key = m.history_key
AND NOT n.is_summary
AND n.created_at > (
SELECT MAX(s.created_at)
FROM channel_pending_messages s
WHERE s.channel_name = m.channel_name
AND s.history_key = m.history_key
AND s.is_summary
)
) AS has_summary,
MAX(created_at) AS last_activity
FROM channel_pending_messages
FROM channel_pending_messages m
GROUP BY channel_name, history_key
ORDER BY last_activity DESC`,
)