From 120fc2d09c0dec1948b67720557997e8ca370bc0 Mon Sep 17 00:00:00 2001 From: viettranx Date: Wed, 18 Mar 2026 08:12:10 +0700 Subject: [PATCH] fix(media): chain provider format, post-write verification, group media history (#206) Cherry-picked valuable changes from PR #206: - hasReadImageProvider supports chain format {"providers":[...]} config - create_image/video/audio verify file persistence after write with diagnostic logging - HistoryEntry gains Media field + CollectMedia() for group media context on @mention - Zalo extractContentAndMedia refactored: all media types via DetectMIMEType/BuildMediaTags, 20MB limit - Discord/Zalo pass media paths to Record() and collect historical media on @mention - Zalo send_helpers logs directory contents when checkFileSize stat fails --- internal/agent/media_tool_routing.go | 28 +++++-- internal/channels/discord/handler.go | 14 ++++ internal/channels/history.go | 36 +++++++++ internal/channels/zalo/personal/handlers.go | 80 ++++++++++++------- .../zalo/personal/protocol/send_helpers.go | 12 +++ internal/tools/create_audio.go | 8 +- internal/tools/create_image.go | 8 ++ internal/tools/create_video.go | 8 ++ 8 files changed, 161 insertions(+), 33 deletions(-) diff --git a/internal/agent/media_tool_routing.go b/internal/agent/media_tool_routing.go index ada0d093..e2231837 100644 --- a/internal/agent/media_tool_routing.go +++ b/internal/agent/media_tool_routing.go @@ -13,6 +13,7 @@ import ( // hasReadImageProvider checks if the read_image builtin tool has a dedicated provider configured. // When true, images should NOT be attached inline to the main LLM — instead the agent // uses the read_image tool which routes to the configured vision provider. +// Supports both legacy flat format {"provider":"X"} and chain format {"providers":[...]}. func (l *Loop) hasReadImageProvider() bool { if l.builtinToolSettings == nil { return false @@ -21,14 +22,31 @@ func (l *Loop) hasReadImageProvider() bool { if !ok || len(raw) == 0 { return false } - // Check if provider field is set (non-empty JSON with provider key). - var cfg struct { + + // Try chain format first: {"providers":[{"provider":"X",...}]} + var chain struct { + Providers []struct { + Provider string `json:"provider"` + Enabled *bool `json:"enabled,omitempty"` + } `json:"providers"` + } + if json.Unmarshal(raw, &chain) == nil && len(chain.Providers) > 0 { + for _, p := range chain.Providers { + if p.Provider != "" && (p.Enabled == nil || *p.Enabled) { + return true + } + } + } + + // Fallback: legacy flat format {"provider":"X"} + var legacy struct { Provider string `json:"provider"` } - if err := json.Unmarshal(raw, &cfg); err != nil || cfg.Provider == "" { - return false + if json.Unmarshal(raw, &legacy) == nil && legacy.Provider != "" { + return true } - return true + + return false } // loadHistoricalImagesForTool collects image MediaRefs from historical messages diff --git a/internal/channels/discord/handler.go b/internal/channels/discord/handler.go index cba40d34..bc602a38 100644 --- a/internal/channels/discord/handler.go +++ b/internal/channels/discord/handler.go @@ -139,10 +139,18 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate } } if !mentioned { + // Collect media file paths for group history context. + var mediaPaths []string + for _, mf := range mediaFiles { + if mf.Path != "" { + mediaPaths = append(mediaPaths, mf.Path) + } + } c.groupHistory.Record(channelID, channels.HistoryEntry{ Sender: senderName, SenderID: senderID, Body: content, + Media: mediaPaths, Timestamp: m.Timestamp, MessageID: m.ID, }, c.historyLimit) @@ -206,6 +214,12 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate } else { finalContent = annotated } + // Collect media from pending history entries (sent before this @mention). + if histMediaPaths := c.groupHistory.CollectMedia(channelID); len(histMediaPaths) > 0 { + for _, p := range histMediaPaths { + mediaFiles = append(mediaFiles, bus.MediaFile{Path: p}) + } + } } metadata := map[string]string{ diff --git a/internal/channels/history.go b/internal/channels/history.go index 2dfd76b8..1206bc40 100644 --- a/internal/channels/history.go +++ b/internal/channels/history.go @@ -11,6 +11,7 @@ import ( "context" "fmt" "log/slog" + "os" "strings" "sync" "time" @@ -34,6 +35,7 @@ type HistoryEntry struct { Sender string SenderID string Body string + Media []string // temp file paths for images/attachments (RAM-only, not persisted to DB) Timestamp time.Time MessageID string } @@ -122,6 +124,8 @@ func (ph *PendingHistory) Record(historyKey string, entry HistoryEntry, limit in existing = append(existing, entry) count = len(existing) // capture pre-trim count so MaybeCompact sees threshold exceeded if len(existing) > limit { + trimmed := existing[:len(existing)-limit] + go cleanupMedia(trimmed) existing = existing[len(existing)-limit:] } ph.entries[historyKey] = existing @@ -254,10 +258,14 @@ func (ph *PendingHistory) Clear(historyKey string) { } ph.mu.Lock() + toClean := ph.entries[historyKey] delete(ph.entries, historyKey) ph.removeFromOrder(historyKey) ph.mu.Unlock() + // Clean up any remaining media temp files (after CollectMedia took what it needed). + go cleanupMedia(toClean) + if ph.store != nil { // Remove pending flushes for this key ph.removeFromFlushBuf(historyKey) @@ -285,6 +293,34 @@ func (ph *PendingHistory) evictOldKeys() { for len(ph.order) > maxHistoryKeys { oldest := ph.order[0] ph.order = ph.order[1:] + evicted := ph.entries[oldest] delete(ph.entries, oldest) + go cleanupMedia(evicted) + } +} + +// CollectMedia returns all media file paths from pending entries for a history key +// and removes them from the entries to prevent double-cleanup by Clear(). +func (ph *PendingHistory) CollectMedia(historyKey string) []string { + ph.mu.Lock() + defer ph.mu.Unlock() + + entries := ph.entries[historyKey] + var paths []string + for i := range entries { + paths = append(paths, entries[i].Media...) + entries[i].Media = nil // prevent double-cleanup + } + return paths +} + +// cleanupMedia removes temp files from history entries. Best-effort, logs warnings. +func cleanupMedia(entries []HistoryEntry) { + for _, e := range entries { + for _, path := range e.Media { + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + slog.Warn("pending_history: media cleanup failed", "path", path, "error", err) + } + } } } diff --git a/internal/channels/zalo/personal/handlers.go b/internal/channels/zalo/personal/handlers.go index 910d3f39..f068d764 100644 --- a/internal/channels/zalo/personal/handlers.go +++ b/internal/channels/zalo/personal/handlers.go @@ -12,6 +12,7 @@ import ( "time" "github.com/nextlevelbuilder/goclaw/internal/channels" + "github.com/nextlevelbuilder/goclaw/internal/channels/media" "github.com/nextlevelbuilder/goclaw/internal/channels/typing" "github.com/nextlevelbuilder/goclaw/internal/channels/zalo/personal/protocol" "github.com/nextlevelbuilder/goclaw/internal/tools" @@ -92,6 +93,7 @@ func (c *Channel) handleGroupMessage(msg protocol.GroupMessage) { Sender: senderName, SenderID: senderID, Body: content, + Media: media, Timestamp: time.Now(), MessageID: msg.Data.MsgID, }, c.historyLimit) @@ -124,12 +126,17 @@ func (c *Channel) handleGroupMessage(msg protocol.GroupMessage) { c.startTyping(threadID, protocol.ThreadTypeGroup) + // Collect media from pending history entries (images sent before this @mention). + // Must come after BuildContext — CollectMedia nulls out Media fields to prevent double-cleanup. + histMedia := c.groupHistory.CollectMedia(threadID) + allMedia := append(histMedia, media...) + metadata := map[string]string{ "message_id": msg.Data.MsgID, "platform": channels.TypeZaloPersonal, "group_id": threadID, } - c.HandleMessage(senderID, threadID, finalContent, media, metadata, "group") + c.HandleMessage(senderID, threadID, finalContent, allMedia, metadata, "group") // Clear pending history after sending to agent (matches Telegram/Discord/Slack/Feishu pattern). c.groupHistory.Clear(threadID) @@ -155,42 +162,61 @@ func (c *Channel) startTyping(threadID string, threadType protocol.ThreadType) { ctrl.Start() } -// extractContentAndMedia returns text content and optional local media paths from a message. -// For text messages, media is nil. For image attachments, the image is downloaded to a temp file. +// extractContentAndMedia returns text content (with tags) and optional local media +// file paths from a Zalo message. For text messages, media is nil. For attachments, the file +// is downloaded and classified by MIME type, matching the pattern used by Telegram/Discord/Feishu. func extractContentAndMedia(content protocol.Content) (string, []string) { if text := content.Text(); text != "" { return text, nil } att := content.ParseAttachment() - if att == nil { + if att == nil || att.Href == "" { return "", nil } - text := content.AttachmentText() - if text == "" { - return "", nil - } - var media []string - if att.IsImage() { - if path, err := downloadFile(context.Background(), att.Href); err != nil { - slog.Warn("zalo_personal: failed to download image", "url", att.Href, "error", err) - } else { - media = []string{path} + + // Download the attachment file. + filePath, err := downloadFile(context.Background(), att.Href) + if err != nil { + slog.Warn("zalo_personal: failed to download attachment", "url", att.Href, "error", err) + // Return human-readable fallback so the message isn't silently dropped. + if text := content.AttachmentText(); text != "" { + return text, nil } + return "", nil } - return text, media + + // Classify by MIME type (image, video, audio, document) — same as Discord/Feishu. + mimeType := media.DetectMIMEType(filePath) + mediaKind := media.MediaKindFromMime(mimeType) + + // For images, also check via Zalo CDN path patterns (e.g. /jpg/, /png/) since + // temp files lose the original extension context. + if mediaKind != media.TypeImage && att.IsImage() { + mediaKind = media.TypeImage + } + + // Build the tag that the agent loop's enrichImageIDs/enrichMediaIDs expects. + tag := media.BuildMediaTags([]media.MediaInfo{{ + Type: mediaKind, + FilePath: filePath, + ContentType: mimeType, + FileName: att.Title, + }}) + + return tag, []string{filePath} } -const maxImageBytes = 10 * 1024 * 1024 // 10MB +const maxMediaBytes = 20 * 1024 * 1024 // 20MB (matches Telegram default) -// downloadFile downloads an image URL to a temp file and returns the local path. -// Validates against SSRF and enforces HTTPS-only, timeout, and size limits. -func downloadFile(ctx context.Context, imageURL string) (string, error) { - if err := tools.CheckSSRF(imageURL); err != nil { +// downloadFile downloads a URL to a temp file and returns the local path. +// Validates against SSRF and enforces timeout and size limits. +func downloadFile(ctx context.Context, fileURL string) (string, error) { + if err := tools.CheckSSRF(fileURL); err != nil { return "", fmt.Errorf("ssrf check: %w", err) } client := &http.Client{Timeout: 30 * time.Second} - req, err := http.NewRequestWithContext(ctx, http.MethodGet, imageURL, nil) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, fileURL, nil) if err != nil { return "", fmt.Errorf("download: %w", err) } @@ -204,14 +230,14 @@ func downloadFile(ctx context.Context, imageURL string) (string, error) { return "", fmt.Errorf("download status %d", resp.StatusCode) } - // Strip query params before extracting extension - path := imageURL + // Strip query params before extracting extension. + path := fileURL if i := strings.IndexByte(path, '?'); i >= 0 { path = path[:i] } ext := filepath.Ext(path) if ext == "" || len(ext) > 5 { - ext = ".jpg" + ext = ".bin" } tmpFile, err := os.CreateTemp("", "goclaw_zca_*"+ext) @@ -220,14 +246,14 @@ func downloadFile(ctx context.Context, imageURL string) (string, error) { } defer tmpFile.Close() - written, err := io.Copy(tmpFile, io.LimitReader(resp.Body, maxImageBytes+1)) + written, err := io.Copy(tmpFile, io.LimitReader(resp.Body, maxMediaBytes+1)) if err != nil { os.Remove(tmpFile.Name()) return "", fmt.Errorf("save: %w", err) } - if written > maxImageBytes { + if written > maxMediaBytes { os.Remove(tmpFile.Name()) - return "", fmt.Errorf("image too large: %d bytes", written) + return "", fmt.Errorf("file too large: %d bytes (max %d)", written, maxMediaBytes) } return tmpFile.Name(), nil diff --git a/internal/channels/zalo/personal/protocol/send_helpers.go b/internal/channels/zalo/personal/protocol/send_helpers.go index f50d1994..0a2925e4 100644 --- a/internal/channels/zalo/personal/protocol/send_helpers.go +++ b/internal/channels/zalo/personal/protocol/send_helpers.go @@ -6,8 +6,10 @@ import ( "encoding/hex" "fmt" "io" + "log/slog" "mime/multipart" "os" + "path/filepath" ) // maxUploadSize is the maximum file size for Zalo uploads (25 MB). @@ -17,6 +19,16 @@ const maxUploadSize = 25 * 1024 * 1024 func checkFileSize(filePath string) error { fi, err := os.Stat(filePath) if err != nil { + // Log directory contents for diagnostics when file is missing. + dir := filepath.Dir(filePath) + if entries, dirErr := os.ReadDir(dir); dirErr == nil { + names := make([]string, 0, len(entries)) + for _, e := range entries { + names = append(names, e.Name()) + } + slog.Warn("zalo_personal: file stat failed, dir listing", + "path", filePath, "dir", dir, "contents", names, "error", err) + } return fmt.Errorf("zalo_personal: stat file: %w", err) } if fi.Size() > maxUploadSize { diff --git a/internal/tools/create_audio.go b/internal/tools/create_audio.go index 61494e47..90e8d335 100644 --- a/internal/tools/create_audio.go +++ b/internal/tools/create_audio.go @@ -158,7 +158,13 @@ func (t *CreateAudioTool) Execute(ctx context.Context, args map[string]any) *Res return ErrorResult(fmt.Sprintf("failed to save generated audio: %v", err)) } - slog.Info("create_audio: audio saved", "path", audioPath, "provider", providerName, "type", audioType) + // Verify file was persisted. + if fi, err := os.Stat(audioPath); err != nil { + slog.Warn("create_audio: file missing immediately after write", "path", audioPath, "error", err) + return ErrorResult(fmt.Sprintf("generated audio file missing after write: %v", err)) + } else { + slog.Info("create_audio: file saved", "path", audioPath, "size", fi.Size(), "data_len", len(audioBytes), "provider", providerName, "type", audioType) + } result := &Result{ForLLM: fmt.Sprintf("MEDIA:%s", audioPath)} result.Media = []bus.MediaFile{{Path: audioPath, MimeType: "audio/mpeg"}} diff --git a/internal/tools/create_image.go b/internal/tools/create_image.go index 4b79dfab..21ab2571 100644 --- a/internal/tools/create_image.go +++ b/internal/tools/create_image.go @@ -109,6 +109,14 @@ func (t *CreateImageTool) Execute(ctx context.Context, args map[string]any) *Res return ErrorResult(fmt.Sprintf("failed to save generated image: %v", err)) } + // Verify file was persisted (diagnostic for disappearing files). + if fi, err := os.Stat(imagePath); err != nil { + slog.Warn("create_image: file missing immediately after write", "path", imagePath, "error", err) + return ErrorResult(fmt.Sprintf("generated image file missing after write: %v", err)) + } else { + slog.Info("create_image: file saved", "path", imagePath, "size", fi.Size(), "data_len", len(chainResult.Data)) + } + result := &Result{ForLLM: fmt.Sprintf("MEDIA:%s", imagePath)} result.Media = []bus.MediaFile{{Path: imagePath, MimeType: "image/png"}} result.Deliverable = fmt.Sprintf("[Generated image: %s]\nPrompt: %s", filepath.Base(imagePath), prompt) diff --git a/internal/tools/create_video.go b/internal/tools/create_video.go index 7a057b56..40e34dba 100644 --- a/internal/tools/create_video.go +++ b/internal/tools/create_video.go @@ -127,6 +127,14 @@ func (t *CreateVideoTool) Execute(ctx context.Context, args map[string]any) *Res return ErrorResult(fmt.Sprintf("failed to save generated video: %v", err)) } + // Verify file was persisted. + if fi, err := os.Stat(videoPath); err != nil { + slog.Warn("create_video: file missing immediately after write", "path", videoPath, "error", err) + return ErrorResult(fmt.Sprintf("generated video file missing after write: %v", err)) + } else { + slog.Info("create_video: file saved", "path", videoPath, "size", fi.Size(), "data_len", len(chainResult.Data)) + } + result := &Result{ForLLM: fmt.Sprintf("MEDIA:%s", videoPath)} result.Media = []bus.MediaFile{{Path: videoPath, MimeType: "video/mp4"}} result.Deliverable = fmt.Sprintf("[Generated video: %s]\nPrompt: %s", filepath.Base(videoPath), prompt)