diff --git a/internal/channels/telegram/handlers.go b/internal/channels/telegram/handlers.go index c8b62d9d..3b55d2d9 100644 --- a/internal/channels/telegram/handlers.go +++ b/internal/channels/telegram/handlers.go @@ -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), diff --git a/internal/channels/telegram/media.go b/internal/channels/telegram/media.go index 1e8cfd26..18f41434 100644 --- a/internal/channels/telegram/media.go +++ b/internal/channels/telegram/media.go @@ -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, "") + } + if msg.Video != nil { + tags = append(tags, "") + } + if msg.VideoNote != nil { + tags = append(tags, "") + } + if msg.Animation != nil { + tags = append(tags, "") + } + if msg.Audio != nil { + tags = append(tags, "") + } + if msg.Voice != nil { + tags = append(tags, "") + } + if msg.Document != nil { + name := msg.Document.FileName + if name != "" { + tags = append(tags, fmt.Sprintf("", name)) + } else { + tags = append(tags, "") + } + } + if len(tags) == 0 { + return "" + } + return strings.Join(tags, "\n") +} diff --git a/internal/http/pending_messages.go b/internal/http/pending_messages.go index 3b15b76d..ca88b12f 100644 --- a/internal/http/pending_messages.go +++ b/internal/http/pending_messages.go @@ -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 diff --git a/internal/store/pg/pending_message_store.go b/internal/store/pg/pending_message_store.go index 1a1543cc..17889fa0 100644 --- a/internal/store/pg/pending_message_store.go +++ b/internal/store/pg/pending_message_store.go @@ -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`, )