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
This commit is contained in:
viettranx committed 2026-03-18 08:12:10 +07:00
1 parent 796a7ecd22
commit 120fc2d09c
8 files changed
+161 -33

No files matched your search

+23 -5
View File
@@ -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
+14
View File
@@ -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{
+36
View File
@@ -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)
}
}
}
}
+53 -27
View File
@@ -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 <media:*> 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 <media:*> 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
@@ -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 {
+7 -1
View File
@@ -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"}}
+8
View File
@@ -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)
+8
View File
@@ -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)