fix(telegram): bounded concurrent handlers, stall-aware downloads, media error feedback

- Replace sequential polling dispatch with semaphore-bounded goroutines
  (max 20 concurrent handlers) to prevent polling hangs from slow media
  downloads without risking unbounded goroutine growth
- Add sync.WaitGroup for graceful shutdown — Stop() waits for in-flight
  handlers with a 15s timeout
- Add progressReader that aborts downloads stalled for 60s (no data)
- Return MediaError from resolveMedia so callers know what failed and why
- Annotate model content with error context when media is skipped
  (e.g. "[sent media (image) — skipped: exceeds 20 MB limit]")
- Send Telegram reply to user when their file is too large or download fails
- Use errMediaTooLarge sentinel in all size-check paths for consistent
  error classification

internal/channels/telegram/channel.go

@@ -43,6 +43,8 @@ type Channel struct {
 	requireMention   bool
 	pollCancel       context.CancelFunc // cancels the long polling context
 	pollDone         chan struct{}       // closed when polling goroutine exits
+	handlerWg        sync.WaitGroup     // tracks in-flight handler goroutines for graceful shutdown
+	handlerSem       chan struct{}       // bounded semaphore for concurrent handler goroutines
 }
 
 type thinkingCancel struct {
@@ -152,6 +154,7 @@ func (c *Channel) Start(ctx context.Context) error {
 
 	c.SetRunning(true)
 	c.groupHistory.StartFlusher()
+	c.handlerSem = make(chan struct{}, 20) // limit concurrent message handlers
 	slog.Info("telegram bot connected", "username", c.bot.Username())
 
 	// Register bot menu commands with retry.
@@ -189,9 +192,29 @@ func (c *Channel) Start(ctx context.Context) error {
 					return
 				}
 				if update.Message != nil {
-					c.handleMessage(pollCtx, update)
+					select {
+					case c.handlerSem <- struct{}{}:
+						c.handlerWg.Add(1)
+						go func(u telego.Update) {
+							defer c.handlerWg.Done()
+							defer func() { <-c.handlerSem }()
+							c.handleMessage(pollCtx, u)
+						}(update)
+					case <-pollCtx.Done():
+						return
+					}
 				} else if update.CallbackQuery != nil {
-					c.handleCallbackQuery(pollCtx, update.CallbackQuery)
+					select {
+					case c.handlerSem <- struct{}{}:
+						c.handlerWg.Add(1)
+						go func(q *telego.CallbackQuery) {
+							defer c.handlerWg.Done()
+							defer func() { <-c.handlerSem }()
+							c.handleCallbackQuery(pollCtx, q)
+						}(update.CallbackQuery)
+					case <-pollCtx.Done():
+						return
+					}
 				} else {
 					// Log non-message updates for delivery diagnostics
 					updateType := "unknown"
@@ -270,11 +293,24 @@ func (c *Channel) Stop(_ context.Context) error {
 	if c.pollDone != nil {
 		select {
 		case <-c.pollDone:
-			slog.Info("telegram bot stopped")
+			slog.Info("telegram polling goroutine stopped")
 		case <-time.After(10 * time.Second):
 			slog.Warn("telegram polling goroutine did not exit within timeout")
 		}
 	}
+
+	// Wait for in-flight handler goroutines to finish processing.
+	handlerDone := make(chan struct{})
+	go func() {
+		c.handlerWg.Wait()
+		close(handlerDone)
+	}()
+	select {
+	case <-handlerDone:
+		slog.Info("telegram bot stopped")
+	case <-time.After(15 * time.Second):
+		slog.Warn("telegram handler goroutines did not drain within timeout")
+	}
 	return nil
 }
 

internal/channels/telegram/handlers.go

@@ -313,9 +313,9 @@ 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)
+	mediaList, mediaErrors := c.resolveMedia(ctx, message)
 	if message.ReplyToMessage != nil && len(mediaList) == 0 {
-		replyMedia := c.resolveMedia(ctx, message.ReplyToMessage)
+		replyMedia, replyErrors := c.resolveMedia(ctx, message.ReplyToMessage)
 		if len(replyMedia) > 0 {
 			mediaList = append(mediaList, replyMedia...)
 			slog.Debug("telegram: resolved media from replied message",
@@ -323,6 +323,7 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
 				"media_count", len(replyMedia),
 			)
 		}
+		mediaErrors = append(mediaErrors, replyErrors...)
 	}
 
 	var mediaFiles []bus.MediaFile
@@ -376,6 +377,31 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
 		}
 	}
 
+	// Annotate content + notify user for any media download failures.
+	// Replace the per-type lightweight tag with an error-annotated version so the
+	// model knows the specific media was skipped and why.
+	if len(mediaErrors) > 0 {
+		for _, me := range mediaErrors {
+			errTag := fmt.Sprintf("[sent media (%s) — skipped: %s]", me.Type, me.Reason)
+			if lightTag := lightweightTagForType(me.Type, message); lightTag != "" {
+				content = strings.Replace(content, lightTag, errTag, 1)
+			} else {
+				content = errTag + "\n" + content
+			}
+		}
+
+		// Send a short reply so the user knows their file was skipped.
+		for _, me := range mediaErrors {
+			var errText string
+			if me.MaxBytes > 0 {
+				errText = fmt.Sprintf("⚠️ File too large (max %d MB). Skipped.", me.MaxBytes/(1024*1024))
+			} else {
+				errText = "⚠️ Failed to download the attached file. Skipped."
+			}
+			_ = c.sendHTML(ctx, chatID, errText, 0, messageThreadID)
+		}
+	}
+
 	slog.Debug("telegram message received",
 		"sender_id", senderID,
 		"chat_id", fmt.Sprintf("%d", chatID),

internal/channels/telegram/media.go

@@ -2,6 +2,7 @@ package telegram
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
@@ -28,15 +29,29 @@ const (
 
 	// downloadMaxRetries is the number of download retry attempts.
 	downloadMaxRetries = 3
+
+	// stallTimeout is how long a download can receive zero bytes before being aborted.
+	stallTimeout = 60 * time.Second
 )
 
+// errMediaTooLarge indicates a file exceeded the configured max download size.
+var errMediaTooLarge = errors.New("file exceeds max size")
+
+// MediaError records a media download failure with enough context for user/model feedback.
+type MediaError struct {
+	Type     string // "image", "video", "audio", "voice", "document", "animation"
+	Reason   string // human-readable reason
+	MaxBytes int64  // configured limit (0 if not a size error)
+}
+
 // MediaInfo is an alias for the shared media.MediaInfo type.
 type MediaInfo = media.MediaInfo
 
 // resolveMedia extracts and downloads media from a Telegram message.
-// Returns a list of MediaInfo for each media item found.
-func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []MediaInfo {
+// Returns successfully downloaded media and any download errors for feedback.
+func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) ([]MediaInfo, []MediaError) {
 	var results []MediaInfo
+	var mediaErrors []MediaError
 
 	maxBytes := c.config.MediaMaxBytes
 	if maxBytes == 0 {
@@ -53,6 +68,7 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media
 		filePath, err := c.downloadMedia(ctx, photo.FileID, maxBytes)
 		if err != nil {
 			slog.Warn("failed to download photo", "file_id", photo.FileID, "error", err)
+			mediaErrors = append(mediaErrors, newMediaError("image", err, maxBytes))
 		} else {
 			// Pass raw file to agent loop — sanitization now happens at loop level.
 			results = append(results, MediaInfo{
@@ -70,6 +86,7 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media
 		filePath, err := c.downloadMedia(ctx, msg.Video.FileID, maxBytes)
 		if err != nil {
 			slog.Warn("failed to download video", "file_id", msg.Video.FileID, "error", err)
+			mediaErrors = append(mediaErrors, newMediaError("video", err, maxBytes))
 		} else {
 			results = append(results, MediaInfo{
 				Type:        "video",
@@ -87,6 +104,7 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media
 		filePath, err := c.downloadMedia(ctx, msg.VideoNote.FileID, maxBytes)
 		if err != nil {
 			slog.Warn("failed to download video note", "file_id", msg.VideoNote.FileID, "error", err)
+			mediaErrors = append(mediaErrors, newMediaError("video", err, maxBytes))
 		} else {
 			results = append(results, MediaInfo{
 				Type:        "video",
@@ -103,6 +121,7 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media
 		filePath, err := c.downloadMedia(ctx, msg.Animation.FileID, maxBytes)
 		if err != nil {
 			slog.Warn("failed to download animation", "file_id", msg.Animation.FileID, "error", err)
+			mediaErrors = append(mediaErrors, newMediaError("animation", err, maxBytes))
 		} else {
 			results = append(results, MediaInfo{
 				Type:        "animation",
@@ -120,6 +139,7 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media
 		filePath, err := c.downloadMedia(ctx, msg.Audio.FileID, maxBytes)
 		if err != nil {
 			slog.Warn("failed to download audio", "file_id", msg.Audio.FileID, "error", err)
+			mediaErrors = append(mediaErrors, newMediaError("audio", err, maxBytes))
 		} else {
 			results = append(results, MediaInfo{
 				Type:        "audio",
@@ -137,6 +157,7 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media
 		filePath, err := c.downloadMedia(ctx, msg.Voice.FileID, maxBytes)
 		if err != nil {
 			slog.Warn("failed to download voice", "file_id", msg.Voice.FileID, "error", err)
+			mediaErrors = append(mediaErrors, newMediaError("voice", err, maxBytes))
 		} else {
 			results = append(results, MediaInfo{
 				Type:        "voice",
@@ -153,6 +174,7 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media
 		filePath, err := c.downloadMedia(ctx, msg.Document.FileID, maxBytes)
 		if err != nil {
 			slog.Warn("failed to download document", "file_id", msg.Document.FileID, "error", err)
+			mediaErrors = append(mediaErrors, newMediaError("document", err, maxBytes))
 		} else {
 			results = append(results, MediaInfo{
 				Type:        "document",
@@ -165,7 +187,7 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media
 		}
 	}
 
-	return results
+	return results, mediaErrors
 }
 
 // downloadMedia downloads a file from Telegram by file_id with retry logic.
@@ -203,7 +225,7 @@ func (c *Channel) downloadMedia(ctx context.Context, fileID string, maxBytes int
 
 	// Check file size before downloading (FileSize may be 0 for large files on local Bot API).
 	if file.FileSize > 0 && int64(file.FileSize) > maxBytes {
-		return "", fmt.Errorf("file too large: %d bytes (max %d)", file.FileSize, maxBytes)
+		return "", fmt.Errorf("%w: %d bytes (limit %d)", errMediaTooLarge, file.FileSize, maxBytes)
 	}
 
 	// Local Bot API (--local mode) returns absolute filesystem paths and does NOT
@@ -275,15 +297,19 @@ func (c *Channel) downloadMedia(ctx context.Context, fileID string, maxBytes int
 	}
 	defer tmpFile.Close()
 
+	// Wrap the response body with stall detection: abort if no data received for 60s.
+	progressBody := newProgressReader(resp.Body, dlCancel, stallTimeout)
+	defer progressBody.Stop()
+
 	// Copy with size limit
-	written, err := io.Copy(tmpFile, io.LimitReader(resp.Body, maxBytes+1))
+	written, err := io.Copy(tmpFile, io.LimitReader(progressBody, maxBytes+1))
 	if err != nil {
 		os.Remove(tmpFile.Name())
 		return "", fmt.Errorf("save file: %w", err)
 	}
 	if written > maxBytes {
 		os.Remove(tmpFile.Name())
-		return "", fmt.Errorf("file exceeds max size during download: %d bytes", written)
+		return "", fmt.Errorf("%w: %d bytes (limit %d)", errMediaTooLarge, written, maxBytes)
 	}
 
 	return tmpFile.Name(), nil
@@ -302,7 +328,7 @@ func copyLocalFile(srcPath string, maxBytes int64) (string, error) {
 		return "", fmt.Errorf("stat local file: %w", err)
 	}
 	if info.Size() > maxBytes {
-		return "", fmt.Errorf("file too large: %d bytes (max %d)", info.Size(), maxBytes)
+		return "", fmt.Errorf("%w: %d bytes (limit %d)", errMediaTooLarge, info.Size(), maxBytes)
 	}
 
 	ext := filepath.Ext(srcPath)
@@ -371,3 +397,86 @@ func lightweightMediaTags(msg *telego.Message) string {
 	}
 	return strings.Join(tags, "\n")
 }
+
+// lightweightTagForType returns the single lightweight tag that matches a given media type
+// within a Telegram message. Used for targeted replacement when a specific media fails.
+func lightweightTagForType(mediaType string, msg *telego.Message) string {
+	switch mediaType {
+	case "image":
+		if msg.Photo != nil && len(msg.Photo) > 0 {
+			return "[sent an image]"
+		}
+	case "video":
+		if msg.Video != nil || msg.VideoNote != nil {
+			return "[sent a video]"
+		}
+	case "animation":
+		if msg.Animation != nil {
+			return "[sent a video]"
+		}
+	case "audio":
+		if msg.Audio != nil {
+			return "[sent audio]"
+		}
+	case "voice":
+		if msg.Voice != nil {
+			return "[sent a voice message]"
+		}
+	case "document":
+		if msg.Document != nil {
+			if msg.Document.FileName != "" {
+				return fmt.Sprintf("[sent a file: %s]", msg.Document.FileName)
+			}
+			return "[sent a file]"
+		}
+	}
+	return ""
+}
+
+// newMediaError builds a MediaError from a download error, detecting size-limit failures.
+func newMediaError(mediaType string, err error, maxBytes int64) MediaError {
+	me := MediaError{Type: mediaType}
+	if errors.Is(err, errMediaTooLarge) {
+		me.Reason = fmt.Sprintf("exceeds %d MB limit", maxBytes/(1024*1024))
+		me.MaxBytes = maxBytes
+	} else {
+		me.Reason = "download failed"
+	}
+	return me
+}
+
+// progressReader wraps an io.Reader and cancels a context if no data is
+// received within a specified timeout. Used to detect mid-stream stalls.
+type progressReader struct {
+	io.Reader
+	cancel context.CancelFunc
+	timer  *time.Timer
+	d      time.Duration
+}
+
+func newProgressReader(r io.Reader, cancel context.CancelFunc, d time.Duration) *progressReader {
+	pr := &progressReader{
+		Reader: r,
+		cancel: cancel,
+		d:      d,
+	}
+	pr.timer = time.AfterFunc(d, func() {
+		slog.Warn("telegram media: download stalled, aborting", "timeout", d)
+		cancel()
+	})
+	return pr
+}
+
+func (pr *progressReader) Read(p []byte) (n int, err error) {
+	n, err = pr.Reader.Read(p)
+	if n > 0 {
+		pr.timer.Reset(pr.d)
+	}
+	return n, err
+}
+
+func (pr *progressReader) Stop() {
+	if pr.timer != nil {
+		pr.timer.Stop()
+	}
+}