From 0f2737ce53bcecdef09173de135a1f73c066b301 Mon Sep 17 00:00:00 2001 From: viettranx Date: Sun, 8 Mar 2026 14:00:34 +0700 Subject: [PATCH] feat(media): persistent media storage, read_document tool, and pipeline refactor - Add persistent media storage (internal/media/) replacing temp file deletion - Add MediaRef type for lightweight media references in session messages - Refactor media pipeline to use bus.MediaFile{Path, MimeType} across all channels - Add read_document builtin tool for PDF/DOCX/XLSX analysis via Gemini native API - Move image sanitization from Telegram to shared agent/media layer - Add media reload for multi-turn conversations (images from last 5 messages) - Add reply-to-message media resolution for Telegram (re-download on reply) - Add media inventory to compaction summary to preserve awareness after truncation - Fix coreToolSummaries for read_image, read_document, create_image tools - Add real-time trace update events via WebSocket broadcast - Improve trace detail UI with media refs and tool result display --- cmd/gateway.go | 12 +- cmd/gateway_builtin_tools.go | 4 + cmd/gateway_consumer.go | 17 +- cmd/gateway_managed.go | 33 ++- internal/agent/loop.go | 100 +++++++-- internal/agent/loop_history.go | 21 ++ internal/agent/loop_types.go | 12 +- internal/agent/media.go | 133 +++++++++++- .../media_sanitize.go} | 41 ++-- internal/agent/media_tool_routing.go | 80 +++++++ internal/agent/resolver.go | 5 + internal/agent/systemprompt.go | 3 + internal/bus/inbound_debounce.go | 2 +- internal/bus/types.go | 9 +- internal/channels/channel.go | 8 +- internal/channels/telegram/handlers.go | 24 ++- internal/channels/telegram/media.go | 77 ++++--- internal/gateway/methods/chat.go | 25 ++- internal/media/store.go | 139 ++++++++++++ internal/providers/types.go | 12 +- internal/store/pg/helpers.go | 21 ++ internal/store/pg/sessions.go | 3 + internal/store/pg/sessions_ops.go | 5 + internal/tools/announce_queue.go | 4 +- internal/tools/create_image.go | 3 +- internal/tools/delegate.go | 23 +- internal/tools/delegate_async.go | 8 +- internal/tools/delegate_prep.go | 36 +++- internal/tools/delegate_sync.go | 2 +- internal/tools/filesystem_write.go | 5 +- internal/tools/read_document.go | 158 ++++++++++++++ internal/tools/read_document_gemini.go | 120 +++++++++++ internal/tools/read_document_resolve.go | 199 ++++++++++++++++++ internal/tools/result.go | 9 +- internal/tools/subagent.go | 2 +- internal/tools/subagent_exec.go | 8 +- internal/tools/subagent_spawn_tool.go | 8 +- internal/tracing/collector.go | 13 ++ pkg/protocol/events.go | 3 + ui/web/src/api/protocol.ts | 3 + ui/web/src/hooks/use-query-invalidation.ts | 14 +- .../builtin-tool-settings-dialog.tsx | 2 +- .../src/pages/traces/trace-detail-dialog.tsx | 42 +++- 43 files changed, 1313 insertions(+), 135 deletions(-) rename internal/{channels/telegram/image_sanitize.go => agent/media_sanitize.go} (59%) create mode 100644 internal/agent/media_tool_routing.go create mode 100644 internal/media/store.go create mode 100644 internal/tools/read_document.go create mode 100644 internal/tools/read_document_gemini.go create mode 100644 internal/tools/read_document_resolve.go diff --git a/cmd/gateway.go b/cmd/gateway.go index d4af2a42..4726bd20 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -246,7 +246,7 @@ func runGateway() { batchMeta["origin_session_key"] = meta.OriginSessionKey } // Collect media from all items in the batch. - var batchMedia []string + var batchMedia []bus.MediaFile for _, item := range items { batchMedia = append(batchMedia, item.Media...) } @@ -346,6 +346,16 @@ func runGateway() { } if pgStores.Tracing != nil { traceCollector = tracing.NewCollector(pgStores.Tracing) + traceCollector.OnFlush = func(traceIDs []uuid.UUID) { + ids := make([]string, len(traceIDs)) + for i, id := range traceIDs { + ids[i] = id.String() + } + msgBus.Broadcast(bus.Event{ + Name: protocol.EventTraceUpdated, + Payload: map[string]any{"trace_ids": ids}, + }) + } traceCollector.Start() slog.Info("LLM tracing enabled") } diff --git a/cmd/gateway_builtin_tools.go b/cmd/gateway_builtin_tools.go index ec1893c2..2f5922b0 100644 --- a/cmd/gateway_builtin_tools.go +++ b/cmd/gateway_builtin_tools.go @@ -43,6 +43,10 @@ func builtinToolSeedData() []store.BuiltinToolDef { Settings: json.RawMessage(`{"provider":"openrouter","model":"google/gemini-2.5-flash-image"}`), Requires: []string{"vision_provider"}, }, + {Name: "read_document", DisplayName: "Read Document", Description: "Analyze documents (PDF, Word, Excel, PowerPoint, CSV, etc.) using a document-capable LLM provider", Category: "media", Enabled: true, + Settings: json.RawMessage(`{"provider":"gemini","model":"gemini-2.5-flash"}`), + Requires: []string{"document_provider"}, + }, {Name: "create_image", DisplayName: "Create Image", Description: "Generate images from text prompts using an image generation provider", Category: "media", Enabled: true, Settings: json.RawMessage(`{"provider":"openrouter","model":"google/gemini-2.5-flash-image"}`), Requires: []string{"image_gen_provider"}, diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index 5ce7b512..13cb9118 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -225,7 +225,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents // Delegation announces carry media as ForwardMedia (not deleted, forwarded to output). // User-uploaded media goes in Media (loaded as images for LLM, then deleted). - var reqMedia, fwdMedia []string + var reqMedia, fwdMedia []bus.MediaFile if msg.Metadata["delegation_id"] != "" || msg.Metadata["subagent_id"] != "" { fwdMedia = msg.Media } else { @@ -933,18 +933,21 @@ func mediaToMarkdown(media []agent.MediaResult, cfg *config.Config) string { // mediaToMarkdownFromPaths is like mediaToMarkdown but accepts raw file paths // ([]string from bus.InboundMessage.Media) instead of []agent.MediaResult. -func mediaToMarkdownFromPaths(paths []string, cfg *config.Config) string { - if len(paths) == 0 { +func mediaToMarkdownFromPaths(files []bus.MediaFile, cfg *config.Config) string { + if len(files) == 0 { return "" } - media := make([]agent.MediaResult, 0, len(paths)) - for _, p := range paths { - ct := mime.TypeByExtension(filepath.Ext(p)) + media := make([]agent.MediaResult, 0, len(files)) + for _, f := range files { + ct := f.MimeType + if ct == "" { + ct = mime.TypeByExtension(filepath.Ext(f.Path)) + } if ct == "" { ct = "application/octet-stream" } media = append(media, agent.MediaResult{ - Path: p, + Path: f.Path, ContentType: ct, }) } diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index fc582ef9..22e52172 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -3,6 +3,7 @@ package cmd import ( "context" "log/slog" + "path/filepath" "github.com/google/uuid" @@ -12,6 +13,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/hooks" httpapi "github.com/nextlevelbuilder/goclaw/internal/http" mcpbridge "github.com/nextlevelbuilder/goclaw/internal/mcp" + "github.com/nextlevelbuilder/goclaw/internal/media" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/sandbox" "github.com/nextlevelbuilder/goclaw/internal/skills" @@ -62,6 +64,24 @@ func wireExtras( groupWriterCache = store.NewGroupWriterCache(stores.Agents, gwCache) } + // 1c. Persistent media storage for cross-turn image/document access + mediaStore, err := media.NewStore(filepath.Join(workspace, ".media")) + if err != nil { + slog.Warn("media store creation failed, images will not persist across turns", "error", err) + } + + // Wire media cleanup on session delete. + if mediaStore != nil { + if pgSess, ok := sessStore.(*pg.PGSessionStore); ok { + pgSess.OnDelete = func(sessionKey string) { + _ = mediaStore.DeleteSession(sessionKey) + } + } + // Register document analysis tool (needs mediaStore for file access). + toolsReg.Register(tools.NewReadDocumentTool(providerReg, mediaStore)) + slog.Info("read_document tool registered") + } + // 2. User seeding callback: seeds per-user context files on first chat var ensureUserFiles agent.EnsureUserFilesFunc if stores.Agents != nil { @@ -127,6 +147,7 @@ func wireExtras( MCPStore: stores.MCP, MCPPool: mcpPool, GroupWriterCache: groupWriterCache, + MediaStore: mediaStore, OnEvent: func(event agent.AgentEvent) { msgBus.Broadcast(bus.Event{ Name: protocol.EventAgent, @@ -340,6 +361,7 @@ func wireExtras( result, err := loop.Run(ctx, agent.RunRequest{ SessionKey: req.SessionKey, Message: req.Message, + Media: req.Media, UserID: req.UserID, Channel: req.Channel, ChatID: req.ChatID, @@ -357,13 +379,15 @@ func wireExtras( if err != nil { return nil, err } + var drMedia []bus.MediaFile + for _, m := range result.Media { + drMedia = append(drMedia, bus.MediaFile{Path: m.Path, MimeType: m.ContentType}) + } dr := &tools.DelegateRunResult{ Content: result.Content, Iterations: result.Iterations, Deliverables: result.Deliverables, - } - for _, m := range result.Media { - dr.MediaPaths = append(dr.MediaPaths, m.Path) + Media: drMedia, } return dr, nil } @@ -372,6 +396,9 @@ func wireExtras( delegateMgr.SetTeamStore(stores.Teams) } delegateMgr.SetSessionStore(stores.Sessions) + if mediaStore != nil { + delegateMgr.SetMediaLoader(mediaStore) + } // Hook engine (quality gates) hookEngine := hooks.NewEngine() diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 1acfac35..f9f1390c 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -14,6 +14,7 @@ import ( "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -292,20 +293,66 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // (hadBootstrap) — no extra DB roundtrip needed for bootstrap detection. messages, hadBootstrap := l.buildMessages(ctx, history, summary, req.Message, req.ExtraSystemPrompt, req.SessionKey, req.Channel, req.PeerKind, req.UserID, req.HistoryLimit, req.SkillFilter) - // 2. Attach vision images to the current user message (last in messages slice). - // Images are only attached to the live request, NOT persisted in session history. + // 1b. Determine image routing strategy. + // If read_image tool has a dedicated vision provider, images are NOT attached inline + // to the main LLM — the agent calls read_image tool instead. This avoids sending + // images to providers that don't support vision or have strict content filters. + deferToReadImageTool := l.hasReadImageProvider() + + if !deferToReadImageTool { + // Inline mode: reload historical images directly into messages for main provider. + l.reloadMediaForMessages(messages, maxMediaReloadMessages) + } + + // 2. Process media: sanitize images, persist to media store. + var mediaRefs []providers.MediaRef if len(req.Media) > 0 { - if images := loadImages(req.Media); len(images) > 0 { - messages[len(messages)-1].Images = images - ctx = tools.WithMediaImages(ctx, images) // make images available to read_image tool - slog.Info("vision: attached images to user message", "count", len(images), "agent", l.id, "session", req.SessionKey) - } - // Clean up temp media files — they're now base64-encoded in memory. - for _, p := range req.Media { - if err := os.Remove(p); err != nil { - slog.Debug("vision: failed to clean temp media file", "path", p, "error", err) + mediaRefs = l.persistMedia(req.SessionKey, req.Media) + // Load current-turn images from persisted refs. + var imageFiles []bus.MediaFile + for _, ref := range mediaRefs { + if ref.Kind == "image" { + if p, err := l.mediaStore.LoadPath(ref.ID); err == nil { + imageFiles = append(imageFiles, bus.MediaFile{Path: p, MimeType: ref.MimeType}) + } } } + if images := loadImages(imageFiles); len(images) > 0 { + if deferToReadImageTool { + // Tool mode: store in context only — agent calls read_image tool. + ctx = tools.WithMediaImages(ctx, images) + slog.Info("vision: deferring to read_image tool", "count", len(images), "agent", l.id) + } else { + // Inline mode: attach to message + context. + messages[len(messages)-1].Images = images + ctx = tools.WithMediaImages(ctx, images) + slog.Info("vision: attached images inline to main provider", "count", len(images), "agent", l.id) + } + } + } + + // 2a. Tool mode: also load historical images into context for read_image tool. + // Without this, read_image can only see current-turn images, not previous turns. + if deferToReadImageTool && l.mediaStore != nil { + ctx = l.loadHistoricalImagesForTool(ctx, mediaRefs, messages) + } + + // 2b. Collect document MediaRefs (current + historical) for read_document tool. + var docRefs []providers.MediaRef + for _, ref := range mediaRefs { + if ref.Kind == "document" { + docRefs = append(docRefs, ref) + } + } + for i := len(messages) - 1; i >= 0; i-- { + for _, ref := range messages[i].MediaRefs { + if ref.Kind == "document" { + docRefs = append(docRefs, ref) + } + } + } + if len(docRefs) > 0 { + ctx = tools.WithMediaDocRefs(ctx, docRefs) } // 2b. Cross-session recovery: notify team leads about orphaned pending tasks @@ -353,12 +400,13 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // 3. Buffer new messages — write to session only AFTER the run completes. // This prevents concurrent runs from seeing each other's in-progress messages. - // NOTE: pendingMsgs stores TEXT ONLY (no images) to avoid bloating session storage. + // NOTE: pendingMsgs stores text + lightweight MediaRefs (not base64 images). var pendingMsgs []providers.Message if !req.HideInput { pendingMsgs = append(pendingMsgs, providers.Message{ - Role: "user", - Content: req.Message, + Role: "user", + Content: req.Message, + MediaRefs: mediaRefs, }) } @@ -690,8 +738,12 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // Collect MEDIA: paths from tool results. // Prefer result.Media (explicit) over ForLLM MEDIA: prefix (legacy) to avoid duplicates. if len(result.Media) > 0 { - for _, p := range result.Media { - mediaResults = append(mediaResults, MediaResult{Path: p, ContentType: mimeFromExt(filepath.Ext(p))}) + for _, mf := range result.Media { + ct := mf.MimeType + if ct == "" { + ct = mimeFromExt(filepath.Ext(mf.Path)) + } + mediaResults = append(mediaResults, MediaResult{Path: mf.Path, ContentType: ct}) } } else if mr := parseMediaResult(result.ForLLM); mr != nil { mediaResults = append(mediaResults, *mr) @@ -825,8 +877,12 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // Collect MEDIA: paths from tool results. // Prefer result.Media (explicit) over ForLLM MEDIA: prefix (legacy) to avoid duplicates. if len(r.result.Media) > 0 { - for _, p := range r.result.Media { - mediaResults = append(mediaResults, MediaResult{Path: p, ContentType: mimeFromExt(filepath.Ext(p))}) + for _, mf := range r.result.Media { + ct := mf.MimeType + if ct == "" { + ct = mimeFromExt(filepath.Ext(mf.Path)) + } + mediaResults = append(mediaResults, MediaResult{Path: mf.Path, ContentType: ct}) } } else if mr := parseMediaResult(r.result.ForLLM); mr != nil { mediaResults = append(mediaResults, *mr) @@ -942,8 +998,12 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) l.maybeSummarize(ctx, req.SessionKey) // Include forwarded media from delegation results (not cleaned up like req.Media) - for _, p := range req.ForwardMedia { - mediaResults = append(mediaResults, MediaResult{Path: p, ContentType: mimeFromExt(filepath.Ext(p))}) + for _, mf := range req.ForwardMedia { + ct := mf.MimeType + if ct == "" { + ct = mimeFromExt(filepath.Ext(mf.Path)) + } + mediaResults = append(mediaResults, MediaResult{Path: mf.Path, ContentType: ct}) } return &RunResult{ diff --git a/internal/agent/loop_history.go b/internal/agent/loop_history.go index 32594575..35d33624 100644 --- a/internal/agent/loop_history.go +++ b/internal/agent/loop_history.go @@ -354,15 +354,36 @@ func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) { toSummarize := history[:len(history)-keepLast] var sb string + var mediaKinds []string for _, m := range toSummarize { if m.Role == "user" { sb += fmt.Sprintf("user: %s\n", m.Content) } else if m.Role == "assistant" { sb += fmt.Sprintf("assistant: %s\n", SanitizeAssistantContent(m.Content)) } + for _, ref := range m.MediaRefs { + mediaKinds = append(mediaKinds, ref.Kind) + } } prompt := "Provide a concise summary of this conversation, preserving key context:\n" + if len(mediaKinds) > 0 { + // Deduplicate and count media types for a compact note. + counts := make(map[string]int) + for _, k := range mediaKinds { + counts[k]++ + } + prompt += "\nNote: user shared media files (" + first := true + for k, n := range counts { + if !first { + prompt += ", " + } + prompt += fmt.Sprintf("%d %s(s)", n, k) + first = false + } + prompt += ") which are no longer in context. Mention briefly if relevant.\n" + } if summary != "" { prompt += "Existing context: " + summary + "\n" } diff --git a/internal/agent/loop_types.go b/internal/agent/loop_types.go index b7b9300f..53a3bd0e 100644 --- a/internal/agent/loop_types.go +++ b/internal/agent/loop_types.go @@ -10,6 +10,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bootstrap" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/media" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/skills" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -102,6 +103,9 @@ type Loop struct { // Team store for cross-session pending task detection teamStore store.TeamStore + + // Persistent media storage for cross-turn image/document access + mediaStore *media.Store } // AgentEvent is emitted during agent execution for WS broadcasting. @@ -186,6 +190,9 @@ type LoopConfig struct { // Team store for cross-session pending task detection TeamStore store.TeamStore + + // Persistent media storage for cross-turn image/document access + MediaStore *media.Store } func NewLoop(cfg LoopConfig) *Loop { @@ -248,6 +255,7 @@ func NewLoop(cfg LoopConfig) *Loop { thinkingLevel: cfg.ThinkingLevel, groupWriterCache: cfg.GroupWriterCache, teamStore: cfg.TeamStore, + mediaStore: cfg.MediaStore, } } @@ -255,8 +263,8 @@ func NewLoop(cfg LoopConfig) *Loop { type RunRequest struct { SessionKey string // composite key: agent:{agentId}:{channel}:{peerKind}:{chatId} Message string // user message - Media []string // local file paths to images (already sanitized) - ForwardMedia []string // media paths to forward to output (not deleted, from delegation results) + Media []bus.MediaFile // local media files with MIME types + ForwardMedia []bus.MediaFile // media files to forward to output (from delegation results) Channel string // source channel ChatID string // source chat ID PeerKind string // "direct" or "group" (for session key building and tool context) diff --git a/internal/agent/media.go b/internal/agent/media.go index 43c3f9d3..9abd82bb 100644 --- a/internal/agent/media.go +++ b/internal/agent/media.go @@ -7,6 +7,7 @@ import ( "path/filepath" "strings" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/providers" ) @@ -15,25 +16,28 @@ const maxImageBytes = 10 * 1024 * 1024 // loadImages reads local image files and returns base64-encoded ImageContent slices. // Non-image files and files that fail to read are skipped with a warning log. -func loadImages(paths []string) []providers.ImageContent { - if len(paths) == 0 { +func loadImages(files []bus.MediaFile) []providers.ImageContent { + if len(files) == 0 { return nil } var images []providers.ImageContent - for _, p := range paths { - mime := inferImageMime(p) + for _, f := range files { + mime := f.MimeType if mime == "" { + mime = inferImageMime(f.Path) + } + if !strings.HasPrefix(mime, "image/") { continue } - data, err := os.ReadFile(p) + data, err := os.ReadFile(f.Path) if err != nil { - slog.Warn("vision: failed to read image file", "path", p, "error", err) + slog.Warn("vision: failed to read image file", "path", f.Path, "error", err) continue } if len(data) > maxImageBytes { - slog.Warn("vision: image file too large, skipping", "path", p, "size", len(data)) + slog.Warn("vision: image file too large, skipping", "path", f.Path, "size", len(data)) continue } @@ -45,6 +49,121 @@ func loadImages(paths []string) []providers.ImageContent { return images } +// persistMedia sanitizes images, saves all media files to persistent storage, +// and returns lightweight MediaRefs. Non-image files are saved as-is. +// If mediaStore is nil, falls back to legacy behavior (no persistence, delete temp files). +func (l *Loop) persistMedia(sessionKey string, files []bus.MediaFile) []providers.MediaRef { + if l.mediaStore == nil { + // Fallback: no persistent storage configured — delete temp files after loading. + slog.Warn("media: no media store configured, temp files will be deleted", "agent", l.id) + defer func() { + for _, f := range files { + _ = os.Remove(f.Path) + } + }() + return nil + } + + var refs []providers.MediaRef + for _, f := range files { + mime := f.MimeType + if mime == "" { + mime = mimeFromExt(filepath.Ext(f.Path)) // fallback for legacy callers + } + kind := mediaKindFromMime(mime) + + // Sanitize images before persistent storage. + srcPath := f.Path + if kind == "image" { + sanitized, err := SanitizeImage(f.Path) + if err != nil { + slog.Warn("media: sanitize image failed, using original", "path", f.Path, "error", err) + } else { + srcPath = sanitized + mime = "image/jpeg" // sanitized output is always JPEG + // Clean up original if sanitized to a different file. + if sanitized != f.Path { + _ = os.Remove(f.Path) + } + } + } + + id, _, err := l.mediaStore.SaveFile(sessionKey, srcPath, mime) + if err != nil { + slog.Warn("media: failed to persist file", "path", f.Path, "error", err) + _ = os.Remove(srcPath) + continue + } + + refs = append(refs, providers.MediaRef{ + ID: id, + MimeType: mime, + Kind: kind, + }) + slog.Debug("media: persisted file", "id", id, "kind", kind, "mime", mime, "agent", l.id) + } + return refs +} + +// mediaKindFromMime returns the media kind ("image", "video", "audio", "document") +// based on MIME type prefix. +func mediaKindFromMime(mime string) string { + switch { + case strings.HasPrefix(mime, "image/"): + return "image" + case strings.HasPrefix(mime, "video/"): + return "video" + case strings.HasPrefix(mime, "audio/"): + return "audio" + default: + return "document" + } +} + +// maxMediaReloadMessages is the default number of recent messages with image MediaRefs +// to reload for LLM vision context. +const maxMediaReloadMessages = 5 + +// reloadMediaForMessages populates Images on historical messages that have image MediaRefs. +// Only reloads the last maxMessages messages with image refs (newest first) to limit context usage. +func (l *Loop) reloadMediaForMessages(msgs []providers.Message, maxMessages int) { + if l.mediaStore == nil || maxMessages <= 0 { + return + } + + count := 0 + for i := len(msgs) - 1; i >= 0 && count < maxMessages; i-- { + if len(msgs[i].MediaRefs) == 0 || len(msgs[i].Images) > 0 { + continue // skip if no refs or already loaded + } + + hasImageRef := false + var imageFiles []bus.MediaFile + for _, ref := range msgs[i].MediaRefs { + if ref.Kind != "image" { + continue + } + hasImageRef = true + p, err := l.mediaStore.LoadPath(ref.ID) + if err != nil { + slog.Debug("media: reload skip missing file", "id", ref.ID, "error", err) + continue + } + imageFiles = append(imageFiles, bus.MediaFile{Path: p, MimeType: ref.MimeType}) + } + + if !hasImageRef { + continue + } + count++ + + if images := loadImages(imageFiles); len(images) > 0 { + msgs[i].Images = images + slog.Debug("media: reloaded images for historical message", "index", i, "count", len(images)) + } + } +} + // inferImageMime returns the MIME type for supported image extensions, or "" if not an image. func inferImageMime(path string) string { switch strings.ToLower(filepath.Ext(path)) { diff --git a/internal/channels/telegram/image_sanitize.go b/internal/agent/media_sanitize.go similarity index 59% rename from internal/channels/telegram/image_sanitize.go rename to internal/agent/media_sanitize.go index c558da72..a06ead50 100644 --- a/internal/channels/telegram/image_sanitize.go +++ b/internal/agent/media_sanitize.go @@ -1,4 +1,4 @@ -package telegram +package agent import ( "bytes" @@ -12,25 +12,27 @@ import ( "github.com/disintegration/imaging" ) +// Image sanitization constants (moved from telegram/image_sanitize.go). const ( - // imageMaxSide is the maximum pixels per side before resize (matching TS imageMaxSide: 1200). + // imageMaxSide is the maximum pixels per side before resize. imageMaxSide = 1200 - - // imageMaxBytes is the max file size after compression (matching TS: 5MB). - imageMaxBytes = 5 * 1024 * 1024 + // imageSanitizeMaxBytes is the max file size after compression (5MB, Anthropic API limit). + imageSanitizeMaxBytes = 5 * 1024 * 1024 ) -// jpegQualities is the grid of quality levels to try (matching TS jpegQualities). +// jpegQualities is the grid of quality levels to try during sanitization. var jpegQualities = []int{85, 75, 65, 55, 45, 35} -// sanitizeImage resizes and compresses an image for LLM vision input. -// Returns the path to the sanitized image (JPEG), or the original path if no processing needed. -// Pipeline (port from TS resizeToJpeg + sanitizeImageForVision): -// 1. Decode image (JPEG/PNG) -// 2. Auto-orient via EXIF -// 3. Resize if larger than imageMaxSide -// 4. Encode as JPEG, iterate quality until under imageMaxBytes -func sanitizeImage(inputPath string) (string, error) { +// Ensure standard image decoders are registered. +func init() { + image.RegisterFormat("jpeg", "\xff\xd8", jpeg.Decode, jpeg.DecodeConfig) + image.RegisterFormat("png", "\x89PNG", png.Decode, png.DecodeConfig) +} + +// SanitizeImage resizes and compresses an image for LLM vision input. +// Applied uniformly to all channels at the agent loop level. +// Pipeline: decode → auto-orient EXIF → resize if >1200px → JPEG compress until <5MB. +func SanitizeImage(inputPath string) (string, error) { img, err := imaging.Open(inputPath, imaging.AutoOrientation(true)) if err != nil { return "", fmt.Errorf("open image: %w", err) @@ -39,19 +41,16 @@ func sanitizeImage(inputPath string) (string, error) { bounds := img.Bounds() w, h := bounds.Dx(), bounds.Dy() - // Resize if either dimension exceeds max if w > imageMaxSide || h > imageMaxSide { img = imaging.Fit(img, imageMaxSide, imageMaxSide, imaging.Lanczos) } - // Try encoding at decreasing quality until under size limit for _, quality := range jpegQualities { var buf bytes.Buffer if err := jpeg.Encode(&buf, img, &jpeg.Options{Quality: quality}); err != nil { return "", fmt.Errorf("encode jpeg (q=%d): %w", quality, err) } - - if buf.Len() <= imageMaxBytes { + if buf.Len() <= imageSanitizeMaxBytes { outPath := filepath.Join(os.TempDir(), fmt.Sprintf("goclaw_sanitized_%d.jpg", os.Getpid())) if err := os.WriteFile(outPath, buf.Bytes(), 0644); err != nil { return "", fmt.Errorf("write sanitized image: %w", err) @@ -62,9 +61,3 @@ func sanitizeImage(inputPath string) (string, error) { return "", fmt.Errorf("image too large even at lowest quality (dimensions: %dx%d)", w, h) } - -// Ensure standard image decoders are registered. -func init() { - image.RegisterFormat("jpeg", "\xff\xd8", jpeg.Decode, jpeg.DecodeConfig) - image.RegisterFormat("png", "\x89PNG", png.Decode, png.DecodeConfig) -} diff --git a/internal/agent/media_tool_routing.go b/internal/agent/media_tool_routing.go new file mode 100644 index 00000000..ada0d093 --- /dev/null +++ b/internal/agent/media_tool_routing.go @@ -0,0 +1,80 @@ +package agent + +import ( + "context" + "encoding/json" + "log/slog" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/tools" +) + +// 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. +func (l *Loop) hasReadImageProvider() bool { + if l.builtinToolSettings == nil { + return false + } + raw, ok := l.builtinToolSettings["read_image"] + if !ok || len(raw) == 0 { + return false + } + // Check if provider field is set (non-empty JSON with provider key). + var cfg struct { + Provider string `json:"provider"` + } + if err := json.Unmarshal(raw, &cfg); err != nil || cfg.Provider == "" { + return false + } + return true +} + +// loadHistoricalImagesForTool collects image MediaRefs from historical messages +// and loads them into context for the read_image tool. Merges with any images already +// in context (from current turn). Limited to last maxMediaReloadMessages messages with image refs. +func (l *Loop) loadHistoricalImagesForTool(ctx context.Context, currentRefs []providers.MediaRef, messages []providers.Message) context.Context { + // Start with current-turn images already in context. + existing := tools.MediaImagesFromCtx(ctx) + + // Collect image paths from historical messages (scan backwards, limit count). + var histPaths []bus.MediaFile + count := 0 + for i := len(messages) - 1; i >= 0 && count < maxMediaReloadMessages; i-- { + if len(messages[i].MediaRefs) == 0 { + continue + } + hasImage := false + for _, ref := range messages[i].MediaRefs { + if ref.Kind != "image" { + continue + } + hasImage = true + p, err := l.mediaStore.LoadPath(ref.ID) + if err != nil { + continue + } + histPaths = append(histPaths, bus.MediaFile{Path: p, MimeType: ref.MimeType}) + } + if hasImage { + count++ + } + } + + if len(histPaths) == 0 { + return ctx + } + + histImages := loadImages(histPaths) + if len(histImages) == 0 { + return ctx + } + + // Merge: existing (current turn) + historical. + merged := make([]providers.ImageContent, 0, len(existing)+len(histImages)) + merged = append(merged, existing...) + merged = append(merged, histImages...) + slog.Debug("vision: loaded historical images for read_image tool", "current", len(existing), "historical", len(histImages)) + return tools.WithMediaImages(ctx, merged) +} diff --git a/internal/agent/resolver.go b/internal/agent/resolver.go index c5d8689c..7f94a336 100644 --- a/internal/agent/resolver.go +++ b/internal/agent/resolver.go @@ -13,6 +13,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/config" mcpbridge "github.com/nextlevelbuilder/goclaw/internal/mcp" + "github.com/nextlevelbuilder/goclaw/internal/media" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/skills" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -72,6 +73,9 @@ type ResolverDeps struct { // Group file writer cache GroupWriterCache *store.GroupWriterCache + + // Persistent media storage for cross-turn image/document access + MediaStore *media.Store } // NewManagedResolver creates a ResolverFunc that builds Loops from DB agent data. @@ -369,6 +373,7 @@ func NewManagedResolver(deps ResolverDeps) ResolverFunc { ThinkingLevel: ag.ParseThinkingLevel(), GroupWriterCache: deps.GroupWriterCache, TeamStore: deps.TeamStore, + MediaStore: deps.MediaStore, }) slog.Info("resolved agent from DB", "agent", agentKey, "model", ag.Model, "provider", ag.Provider) diff --git a/internal/agent/systemprompt.go b/internal/agent/systemprompt.go index d9a18ca5..687f12c3 100644 --- a/internal/agent/systemprompt.go +++ b/internal/agent/systemprompt.go @@ -67,6 +67,9 @@ var coreToolSummaries = map[string]string{ "session_status": "Show session status (model, tokens, compaction count)", "sessions_history": "Fetch message history for a session", "sessions_send": "Send a message into another session", + "read_image": "Analyze images attached to the conversation. MUST call this when you see tags", + "read_document": "Analyze documents (PDF, DOCX, etc.) attached to the conversation. MUST call this when you see tags", + "create_image": "Generate images from text descriptions using AI", } // BuildSystemPrompt constructs the full system prompt with all sections. diff --git a/internal/bus/inbound_debounce.go b/internal/bus/inbound_debounce.go index 36eac3db..6506ce7e 100644 --- a/internal/bus/inbound_debounce.go +++ b/internal/bus/inbound_debounce.go @@ -152,7 +152,7 @@ func mergeInboundMessages(msgs []InboundMessage) InboundMessage { last.Content = strings.Join(parts, "\n") // Merge media from all messages. - var allMedia []string + var allMedia []MediaFile for _, m := range msgs { allMedia = append(allMedia, m.Media...) } diff --git a/internal/bus/types.go b/internal/bus/types.go index 90c79f83..f7207b18 100644 --- a/internal/bus/types.go +++ b/internal/bus/types.go @@ -2,13 +2,20 @@ package bus import "context" +// MediaFile represents an inbound media file with its MIME type. +// Used throughout the media pipeline to preserve content type from channel download to storage. +type MediaFile struct { + Path string `json:"path"` + MimeType string `json:"mime_type,omitempty"` // e.g. "application/pdf", "image/jpeg" +} + // InboundMessage represents a message received from a channel (Telegram, Discord, etc.) type InboundMessage struct { Channel string `json:"channel"` SenderID string `json:"sender_id"` ChatID string `json:"chat_id"` Content string `json:"content"` - Media []string `json:"media,omitempty"` + Media []MediaFile `json:"media,omitempty"` SessionKey string `json:"session_key"` // deprecated: gateway builds canonical key PeerKind string `json:"peer_kind,omitempty"` // "direct" or "group" (used for session key) AgentID string `json:"agent_id,omitempty"` // target agent (for multi-agent routing) diff --git a/internal/channels/channel.go b/internal/channels/channel.go index a2526466..7c429b1e 100644 --- a/internal/channels/channel.go +++ b/internal/channels/channel.go @@ -255,12 +255,18 @@ func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []st userID = senderID[:idx] } + // Convert string paths to MediaFile (for channels that haven't been updated yet). + var mediaFiles []bus.MediaFile + for _, p := range media { + mediaFiles = append(mediaFiles, bus.MediaFile{Path: p}) + } + msg := bus.InboundMessage{ Channel: c.name, SenderID: senderID, ChatID: chatID, Content: content, - Media: media, + Media: mediaFiles, PeerKind: peerKind, UserID: userID, Metadata: metadata, diff --git a/internal/channels/telegram/handlers.go b/internal/channels/telegram/handlers.go index 3fb2ae06..414b7081 100644 --- a/internal/channels/telegram/handlers.go +++ b/internal/channels/telegram/handlers.go @@ -193,7 +193,22 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { // Process media (photos, audio, voice, documents) mediaList := c.resolveMedia(ctx, message) - var mediaPaths []string + + // 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. @@ -236,7 +251,10 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { } if m.FilePath != "" { - mediaPaths = append(mediaPaths, m.FilePath) + mediaFiles = append(mediaFiles, bus.MediaFile{ + Path: m.FilePath, + MimeType: m.ContentType, + }) } } @@ -439,7 +457,7 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { SenderID: senderID, ChatID: chatIDStr, Content: finalContent, - Media: mediaPaths, + Media: mediaFiles, PeerKind: peerKind, UserID: userID, AgentID: targetAgentID, diff --git a/internal/channels/telegram/media.go b/internal/channels/telegram/media.go index 3526153c..8db03309 100644 --- a/internal/channels/telegram/media.go +++ b/internal/channels/telegram/media.go @@ -54,15 +54,10 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media if err != nil { slog.Warn("failed to download photo", "file_id", photo.FileID, "error", err) } else { - // Sanitize image for LLM vision - sanitized, sanitizeErr := sanitizeImage(filePath) - if sanitizeErr != nil { - slog.Warn("failed to sanitize image, using original", "error", sanitizeErr) - sanitized = filePath - } + // Pass raw file to agent loop — sanitization now happens at loop level. results = append(results, MediaInfo{ Type: "image", - FilePath: sanitized, + FilePath: filePath, FileID: photo.FileID, ContentType: "image/jpeg", FileSize: int64(photo.FileSize), @@ -72,34 +67,52 @@ func (c *Channel) resolveMedia(ctx context.Context, msg *telego.Message) []Media // Video if msg.Video != nil { - results = append(results, MediaInfo{ - Type: "video", - FileID: msg.Video.FileID, - ContentType: msg.Video.MimeType, - FileName: msg.Video.FileName, - FileSize: int64(msg.Video.FileSize), - }) + 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) + } else { + results = append(results, MediaInfo{ + Type: "video", + FilePath: filePath, + FileID: msg.Video.FileID, + ContentType: msg.Video.MimeType, + FileName: msg.Video.FileName, + FileSize: int64(msg.Video.FileSize), + }) + } } // Video Note (round video) if msg.VideoNote != nil { - results = append(results, MediaInfo{ - Type: "video", - FileID: msg.VideoNote.FileID, - ContentType: "video/mp4", - FileSize: int64(msg.VideoNote.FileSize), - }) + 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) + } else { + results = append(results, MediaInfo{ + Type: "video", + FilePath: filePath, + FileID: msg.VideoNote.FileID, + ContentType: "video/mp4", + FileSize: int64(msg.VideoNote.FileSize), + }) + } } // Animation (GIF) if msg.Animation != nil { - results = append(results, MediaInfo{ - Type: "animation", - FileID: msg.Animation.FileID, - ContentType: msg.Animation.MimeType, - FileName: msg.Animation.FileName, - FileSize: int64(msg.Animation.FileSize), - }) + 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) + } else { + results = append(results, MediaInfo{ + Type: "animation", + FilePath: filePath, + FileID: msg.Animation.FileID, + ContentType: msg.Animation.MimeType, + FileName: msg.Animation.FileName, + FileSize: int64(msg.Animation.FileSize), + }) + } } // Audio @@ -251,7 +264,11 @@ func buildMediaTags(mediaList []MediaInfo) string { tags = append(tags, "") } case "document": - tags = append(tags, "") + if m.FileName != "" { + tags = append(tags, fmt.Sprintf("", m.FileName)) + } else { + tags = append(tags, "") + } } } return strings.Join(tags, "\n") @@ -303,7 +320,9 @@ func extractDocumentContent(filePath, fileName string) (string, error) { ext := strings.ToLower(filepath.Ext(fileName)) mime, isText := textExtensions[ext] if !isText { - return fmt.Sprintf("[File: %s — binary format not supported, only text files can be processed]", fileName), nil + // Binary files (PDF, DOCX, etc.) are persisted via MediaRef and analyzed + // by the read_document tool. Return a hint instead of "not supported" placeholder. + return fmt.Sprintf("[File: %s — use read_document tool to analyze this file]", fileName), nil } data, err := os.ReadFile(filePath) diff --git a/internal/gateway/methods/chat.go b/internal/gateway/methods/chat.go index 41e233fa..e047f104 100644 --- a/internal/gateway/methods/chat.go +++ b/internal/gateway/methods/chat.go @@ -7,6 +7,7 @@ import ( "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/agent" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/sessions" @@ -34,10 +35,11 @@ func (m *ChatMethods) Register(router *gateway.MethodRouter) { } type chatSendParams struct { - Message string `json:"message"` - AgentID string `json:"agentId"` - SessionKey string `json:"sessionKey"` - Stream bool `json:"stream"` + Message string `json:"message"` + AgentID string `json:"agentId"` + SessionKey string `json:"sessionKey"` + Stream bool `json:"stream"` + Media []string `json:"media,omitempty"` // local file paths from upload endpoint } func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { @@ -105,9 +107,16 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re defer m.agents.UnregisterRun(runID) defer cancel() + // Convert string paths to bus.MediaFile for the unified media pipeline. + var mediaFiles []bus.MediaFile + for _, p := range params.Media { + mediaFiles = append(mediaFiles, bus.MediaFile{Path: p}) + } + result, err := loop.Run(runCtx, agent.RunRequest{ SessionKey: sessionKey, Message: params.Message, + Media: mediaFiles, Channel: "ws", ChatID: client.ID(), RunID: runID, @@ -124,11 +133,15 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re return } - client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ + resp := map[string]interface{}{ "runId": result.RunID, "content": result.Content, "usage": result.Usage, - })) + } + if len(result.Media) > 0 { + resp["media"] = result.Media + } + client.SendResponse(protocol.NewOKResponse(req.ID, resp)) }() } diff --git a/internal/media/store.go b/internal/media/store.go new file mode 100644 index 00000000..b6505985 --- /dev/null +++ b/internal/media/store.go @@ -0,0 +1,139 @@ +package media + +import ( + "crypto/sha256" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "strings" + + "github.com/google/uuid" +) + +// Store provides persistent media file storage scoped by session. +// Files are organized as: {baseDir}/{sessionHash}/{uuid}.{ext} +type Store struct { + baseDir string +} + +// NewStore creates a media store rooted at baseDir. +// The directory is created if it doesn't exist. +func NewStore(baseDir string) (*Store, error) { + if err := os.MkdirAll(baseDir, 0755); err != nil { + return nil, fmt.Errorf("media: create base dir: %w", err) + } + return &Store{baseDir: baseDir}, nil +} + +// SaveFile moves or copies a file to persistent storage. +// Returns the unique media ID and the destination path. +func (s *Store) SaveFile(sessionKey, srcPath, mime string) (id string, dstPath string, err error) { + dir := s.sessionDir(sessionKey) + if err := os.MkdirAll(dir, 0755); err != nil { + return "", "", fmt.Errorf("media: create session dir: %w", err) + } + + mediaID := uuid.New().String() + ext := extFromMime(mime) + if ext == "" { + ext = filepath.Ext(srcPath) + } + dstPath = filepath.Join(dir, mediaID+ext) + + // Try rename first (fast, same filesystem). + if err := os.Rename(srcPath, dstPath); err == nil { + return mediaID, dstPath, nil + } + + // Fallback: copy + remove source. + if err := copyFile(srcPath, dstPath); err != nil { + return "", "", fmt.Errorf("media: copy file: %w", err) + } + _ = os.Remove(srcPath) // best-effort cleanup of source + return mediaID, dstPath, nil +} + +// LoadPath returns the filesystem path for a media ID. +// Returns an error if the file doesn't exist. +func (s *Store) LoadPath(id string) (string, error) { + // Media files are stored as {sessionHash}/{id}.{ext}. + // Since we don't know the session hash, glob for the ID across all session dirs. + matches, err := filepath.Glob(filepath.Join(s.baseDir, "*", id+".*")) + if err != nil { + return "", fmt.Errorf("media: glob for %s: %w", id, err) + } + if len(matches) == 0 { + return "", fmt.Errorf("media: file not found: %s", id) + } + return matches[0], nil +} + +// DeleteSession removes all media files for a session. +func (s *Store) DeleteSession(sessionKey string) error { + dir := s.sessionDir(sessionKey) + if err := os.RemoveAll(dir); err != nil { + slog.Warn("media: failed to delete session dir", "dir", dir, "error", err) + return err + } + return nil +} + +// sessionDir returns the directory path for a session's media files. +// Uses first 12 chars of SHA-256 hash of sessionKey for filesystem safety. +func (s *Store) sessionDir(sessionKey string) string { + h := sha256.Sum256([]byte(sessionKey)) + hash := fmt.Sprintf("%x", h[:6]) // 12 hex chars + return filepath.Join(s.baseDir, hash) +} + +// extFromMime returns a file extension (with dot) for a MIME type. +func extFromMime(mime string) string { + switch { + case strings.HasPrefix(mime, "image/jpeg"): + return ".jpg" + case strings.HasPrefix(mime, "image/png"): + return ".png" + case strings.HasPrefix(mime, "image/gif"): + return ".gif" + case strings.HasPrefix(mime, "image/webp"): + return ".webp" + case strings.HasPrefix(mime, "video/mp4"): + return ".mp4" + case strings.HasPrefix(mime, "audio/ogg"), strings.HasPrefix(mime, "audio/opus"): + return ".ogg" + case strings.HasPrefix(mime, "audio/mpeg"): + return ".mp3" + case strings.HasPrefix(mime, "audio/wav"): + return ".wav" + case strings.HasPrefix(mime, "application/pdf"): + return ".pdf" + case mime == "application/vnd.openxmlformats-officedocument.wordprocessingml.document": + return ".docx" + case mime == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": + return ".xlsx" + default: + return "" + } +} + +// copyFile copies src to dst using buffered I/O. +func copyFile(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + + if _, err := io.Copy(out, in); err != nil { + return err + } + return out.Close() +} diff --git a/internal/providers/types.go b/internal/providers/types.go index 3d8a0c5b..054bccaa 100644 --- a/internal/providers/types.go +++ b/internal/providers/types.go @@ -81,12 +81,22 @@ type ImageContent struct { Data string `json:"data"` // base64-encoded image bytes } +// MediaRef is a lightweight reference to a persistently stored media file. +// Stored in session JSONB (~60 bytes each) instead of megabytes for base64. +// On reload, MediaRefs are resolved to file paths and loaded into Images (for images). +type MediaRef struct { + ID string `json:"id"` // unique media ID (uuid) + MimeType string `json:"mime_type"` // e.g. "image/jpeg", "application/pdf" + Kind string `json:"kind"` // "image", "video", "audio", "document" +} + // Message represents a conversation message. type Message struct { Role string `json:"role"` // "system", "user", "assistant", "tool" Content string `json:"content"` Thinking string `json:"thinking,omitempty"` // reasoning_content for thinking models (Kimi, DeepSeek, etc.) - Images []ImageContent `json:"images,omitempty"` // vision: base64 images + Images []ImageContent `json:"images,omitempty"` // vision: base64 images (runtime only, not persisted) + MediaRefs []MediaRef `json:"media_refs,omitempty"` // persistent media file references ToolCalls []ToolCall `json:"tool_calls,omitempty"` ToolCallID string `json:"tool_call_id,omitempty"` // for role="tool" responses diff --git a/internal/store/pg/helpers.go b/internal/store/pg/helpers.go index d68bb72e..d57988c2 100644 --- a/internal/store/pg/helpers.go +++ b/internal/store/pg/helpers.go @@ -117,8 +117,29 @@ func execMapUpdate(ctx context.Context, db *sql.DB, table string, id uuid.UUID, args = append(args, val) i++ } + // Auto-set updated_at for tables that have the column, unless caller already included it. + if _, ok := updates["updated_at"]; !ok && tableHasUpdatedAt(table) { + setClauses = append(setClauses, fmt.Sprintf("updated_at = $%d", i)) + args = append(args, time.Now().UTC()) + i++ + } args = append(args, id) q := fmt.Sprintf("UPDATE %s SET %s WHERE id = $%d", table, strings.Join(setClauses, ", "), i) _, err := db.ExecContext(ctx, q, args...) return err } + +// tablesWithUpdatedAt lists tables that have an updated_at column. +var tablesWithUpdatedAt = map[string]bool{ + "agents": true, "llm_providers": true, "sessions": true, + "channel_instances": true, "cron_jobs": true, "custom_tools": true, + "skills": true, "mcp_servers": true, "agent_links": true, + "agent_teams": true, "team_tasks": true, "builtin_tools": true, + "agent_context_files": true, "user_context_files": true, + "user_agent_overrides": true, "config_secrets": true, + "memory_documents": true, "memory_chunks": true, "embedding_cache": true, +} + +func tableHasUpdatedAt(table string) bool { + return tablesWithUpdatedAt[table] +} diff --git a/internal/store/pg/sessions.go b/internal/store/pg/sessions.go index 21f900bf..48321cad 100644 --- a/internal/store/pg/sessions.go +++ b/internal/store/pg/sessions.go @@ -18,6 +18,9 @@ type PGSessionStore struct { mu sync.RWMutex // In-memory cache for hot sessions (reduces DB reads during tool loops) cache map[string]*store.SessionData + // OnDelete is called with the session key when a session is deleted. + // Used for media file cleanup. + OnDelete func(sessionKey string) } func NewPGSessionStore(db *sql.DB) *PGSessionStore { diff --git a/internal/store/pg/sessions_ops.go b/internal/store/pg/sessions_ops.go index 9fb0c27d..95dee2e1 100644 --- a/internal/store/pg/sessions_ops.go +++ b/internal/store/pg/sessions_ops.go @@ -34,6 +34,11 @@ func (s *PGSessionStore) Delete(key string) error { delete(s.cache, key) s.mu.Unlock() + // Clean up associated media files before deleting from DB. + if s.OnDelete != nil { + s.OnDelete(key) + } + _, err := s.db.Exec("DELETE FROM sessions WHERE session_key = $1", key) return err } diff --git a/internal/tools/announce_queue.go b/internal/tools/announce_queue.go index 0ec262ea..240cb724 100644 --- a/internal/tools/announce_queue.go +++ b/internal/tools/announce_queue.go @@ -5,6 +5,8 @@ import ( "strings" "sync" "time" + + "github.com/nextlevelbuilder/goclaw/internal/bus" ) // AnnounceQueueItem represents a single subagent result waiting to be announced. @@ -13,7 +15,7 @@ type AnnounceQueueItem struct { Label string Status string // "completed", "failed", "cancelled" Result string - Media []string // media file paths from tool results + Media []bus.MediaFile // media files from tool results Runtime time.Duration Iterations int } diff --git a/internal/tools/create_image.go b/internal/tools/create_image.go index d4dd07c4..182364f4 100644 --- a/internal/tools/create_image.go +++ b/internal/tools/create_image.go @@ -14,6 +14,7 @@ import ( "strings" "time" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/providers" ) @@ -131,7 +132,7 @@ func (t *CreateImageTool) Execute(ctx context.Context, args map[string]interface } result := &Result{ForLLM: fmt.Sprintf("MEDIA:%s", imagePath)} - result.Media = []string{imagePath} + result.Media = []bus.MediaFile{{Path: imagePath, MimeType: "image/png"}} result.Deliverable = fmt.Sprintf("[Generated image: %s]\nPrompt: %s", filepath.Base(imagePath), prompt) result.Provider = providerName result.Model = model diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go index 2bfca7f4..eb2db85a 100644 --- a/internal/tools/delegate.go +++ b/internal/tools/delegate.go @@ -83,6 +83,9 @@ type DelegateRunRequest struct { ExtraSystemPrompt string MaxIterations int // per-delegation override (0 = use agent default) + // Media propagation: parent's media files for delegate's vision context. + Media []bus.MediaFile + // Delegation context (bridged to agent.RunRequest for event enrichment) DelegationID string TeamID string @@ -94,15 +97,15 @@ type DelegateRunRequest struct { type DelegateRunResult struct { Content string Iterations int - MediaPaths []string // media file paths from tool results (e.g. generated images) - Deliverables []string // actual content from tool outputs (e.g. written file text, image prompt) + Media []bus.MediaFile // media files from tool results (e.g. generated images) + Deliverables []string // actual content from tool outputs (e.g. written file text, image prompt) } // DelegateArtifacts holds forwarded artifacts from delegation results. // Used to accumulate artifacts from intermediate completions until the final // announce fires. New artifact types (files, voice, etc.) should be added here. type DelegateArtifacts struct { - Media []string // file paths to forward (images, documents, audio, etc.) + Media []bus.MediaFile // media files to forward (images, documents, audio, etc.) Results []DelegateResultSummary // result summaries from completed delegations CompletedTaskIDs []string // team task IDs auto-completed by delegations (for announce context) } @@ -127,7 +130,7 @@ type DelegateResult struct { Iterations int DelegationID string // for async: the delegation ID to track/cancel TeamTaskID string // auto-created or provided team task ID (for tracing) - MediaPaths []string // media file paths from delegation result + Media []bus.MediaFile // media files from delegation result } // linkSettings holds per-user restriction rules from agent_links.settings JSONB. @@ -138,6 +141,12 @@ type linkSettings struct { UserDeny []string `json:"user_deny"` } +// MediaPathLoader resolves a media ID to a local file path. +// Used to propagate parent images to delegates without importing the media package. +type MediaPathLoader interface { + LoadPath(id string) (string, error) +} + // DelegateManager manages inter-agent delegation lifecycle. // Similar to SubagentManager but delegates to fully-configured named agents. type DelegateManager struct { @@ -146,6 +155,7 @@ type DelegateManager struct { agentStore store.AgentStore teamStore store.TeamStore // optional: enables auto-complete of team tasks sessionStore store.SessionStore // optional: enables session cleanup + mediaLoader MediaPathLoader // optional: enables image propagation to delegates msgBus *bus.MessageBus // for event broadcast + async announce (PublishInbound) hookEngine *hooks.Engine // optional: quality gate evaluation @@ -187,6 +197,11 @@ func (dm *DelegateManager) SetHookEngine(engine *hooks.Engine) { dm.hookEngine = engine } +// SetMediaLoader enables image propagation from parent to delegate agents. +func (dm *DelegateManager) SetMediaLoader(ml MediaPathLoader) { + dm.mediaLoader = ml +} + // SetProgressEnabled toggles "Your team is working on it..." chat notifications. func (dm *DelegateManager) SetProgressEnabled(enabled bool) { dm.progressEnabled = enabled diff --git a/internal/tools/delegate_async.go b/internal/tools/delegate_async.go index 2f4c4ca7..db82fae4 100644 --- a/internal/tools/delegate_async.go +++ b/internal/tools/delegate_async.go @@ -89,12 +89,12 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) // need to call team_tasks to aggregate. arts := &DelegateArtifacts{} if result != nil { - arts.Media = result.MediaPaths + arts.Media = result.Media arts.Results = []DelegateResultSummary{{ AgentKey: task.TargetAgentKey, DisplayName: task.TargetDisplayName, Content: result.Content, - HasMedia: len(result.MediaPaths) > 0, + HasMedia: len(result.Media) > 0, Deliverables: result.Deliverables, }} } else if runErr != nil { @@ -139,12 +139,12 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) // Last completion: collect all accumulated artifacts + own result artifacts := dm.collectArtifacts(oKey) if result != nil { - artifacts.Media = append(artifacts.Media, result.MediaPaths...) + artifacts.Media = append(artifacts.Media, result.Media...) artifacts.Results = append(artifacts.Results, DelegateResultSummary{ AgentKey: task.TargetAgentKey, DisplayName: task.TargetDisplayName, Content: result.Content, - HasMedia: len(result.MediaPaths) > 0, + HasMedia: len(result.Media) > 0, Deliverables: result.Deliverables, }) } diff --git a/internal/tools/delegate_prep.go b/internal/tools/delegate_prep.go index ff4cf6c2..a60b3338 100644 --- a/internal/tools/delegate_prep.go +++ b/internal/tools/delegate_prep.go @@ -319,7 +319,7 @@ func buildDelegateMessage(opts DelegateOpts) string { } func (dm *DelegateManager) buildRunRequest(task *DelegationTask, message string) DelegateRunRequest { - return DelegateRunRequest{ + req := DelegateRunRequest{ SessionKey: task.SessionKey, Message: message, UserID: task.UserID, @@ -340,4 +340,38 @@ func (dm *DelegateManager) buildRunRequest(task *DelegationTask, message string) TeamTaskID: func() string { if task.TeamTaskID != uuid.Nil { return task.TeamTaskID.String() }; return "" }(), ParentAgentID: task.SourceAgentKey, } + + // Propagate parent's recent image media to delegate for vision context. + if dm.mediaLoader != nil && dm.sessionStore != nil && task.OriginSessionKey != "" { + req.Media = dm.resolveParentMedia(task.OriginSessionKey) + } + + return req +} + +// resolveParentMedia loads image media files from the parent session's recent MediaRefs. +func (dm *DelegateManager) resolveParentMedia(parentSessionKey string) []bus.MediaFile { + history := dm.sessionStore.GetHistory(parentSessionKey) + if len(history) == 0 { + return nil + } + + // Scan last 5 messages for image MediaRefs. + var files []bus.MediaFile + count := 0 + for i := len(history) - 1; i >= 0 && count < 5; i-- { + if len(history[i].MediaRefs) == 0 { + continue + } + for _, ref := range history[i].MediaRefs { + if ref.Kind != "image" { + continue + } + if p, err := dm.mediaLoader.LoadPath(ref.ID); err == nil { + files = append(files, bus.MediaFile{Path: p, MimeType: ref.MimeType}) + } + } + count++ + } + return files } diff --git a/internal/tools/delegate_sync.go b/internal/tools/delegate_sync.go index 6ba9438f..133306b8 100644 --- a/internal/tools/delegate_sync.go +++ b/internal/tools/delegate_sync.go @@ -66,5 +66,5 @@ func (dm *DelegateManager) Delegate(ctx context.Context, opts DelegateOpts) (*De dm.saveDelegationHistory(task, result.Content, nil, duration) slog.Info("delegation completed", "id", task.ID, "target", opts.TargetAgentKey, "iterations", result.Iterations) - return &DelegateResult{Content: result.Content, Iterations: result.Iterations, DelegationID: task.ID, TeamTaskID: task.TeamTaskID.String(), MediaPaths: result.MediaPaths}, nil + return &DelegateResult{Content: result.Content, Iterations: result.Iterations, DelegationID: task.ID, TeamTaskID: task.TeamTaskID.String(), Media: result.Media}, nil } diff --git a/internal/tools/filesystem_write.go b/internal/tools/filesystem_write.go index 20d49d1a..ce70c83f 100644 --- a/internal/tools/filesystem_write.go +++ b/internal/tools/filesystem_write.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/sandbox" "github.com/nextlevelbuilder/goclaw/internal/store" ) @@ -140,7 +141,7 @@ func (t *WriteFileTool) Execute(ctx context.Context, args map[string]interface{} result := SilentResult(fmt.Sprintf("File written: %s (%d bytes)", path, len(content))) result.Deliverable = content if deliver { - result.Media = []string{resolved} + result.Media = []bus.MediaFile{{Path: resolved}} } return result } @@ -164,7 +165,7 @@ func (t *WriteFileTool) executeInSandbox(ctx context.Context, path, content, san workspace = t.workspace } hostPath := filepath.Join(workspace, path) - result.Media = []string{hostPath} + result.Media = []bus.MediaFile{{Path: hostPath}} } return result } diff --git a/internal/tools/read_document.go b/internal/tools/read_document.go new file mode 100644 index 00000000..980819a9 --- /dev/null +++ b/internal/tools/read_document.go @@ -0,0 +1,158 @@ +package tools + +import ( + "context" + "fmt" + "log/slog" + "os" + "strings" + + "github.com/nextlevelbuilder/goclaw/internal/providers" +) + +// textReadableMIMEs are MIME types whose content can be returned directly without LLM analysis. +var textReadableMIMEs = map[string]bool{ + "application/json": true, + "text/csv": true, + "text/plain": true, + "text/html": true, + "text/xml": true, + "application/xml": true, + "text/markdown": true, + "application/javascript": true, + "text/css": true, + "application/yaml": true, + "text/yaml": true, +} + +// documentMaxTextBytes is the max size for direct text return (500KB). +const documentMaxTextBytes = 500 * 1024 + +// --- Context helpers for media documents --- + +const ctxMediaDocRefs toolContextKey = "tool_media_doc_refs" + +// WithMediaDocRefs stores document MediaRefs in context for read_document tool access. +func WithMediaDocRefs(ctx context.Context, refs []providers.MediaRef) context.Context { + return context.WithValue(ctx, ctxMediaDocRefs, refs) +} + +// MediaDocRefsFromCtx retrieves stored document MediaRefs from context. +func MediaDocRefsFromCtx(ctx context.Context) []providers.MediaRef { + v, _ := ctx.Value(ctxMediaDocRefs).([]providers.MediaRef) + return v +} + +// --- ReadDocumentTool --- + +// documentMaxBytes is the max file size for document analysis (20MB). +const documentMaxBytes = 20 * 1024 * 1024 + +// ReadDocumentTool uses a document-capable provider to analyze files +// attached to the current conversation. Follows same pattern as ReadImageTool. +type ReadDocumentTool struct { + registry *providers.Registry + mediaLoader MediaPathLoader +} + +func NewReadDocumentTool(registry *providers.Registry, mediaLoader MediaPathLoader) *ReadDocumentTool { + return &ReadDocumentTool{registry: registry, mediaLoader: mediaLoader} +} + +func (t *ReadDocumentTool) Name() string { return "read_document" } + +func (t *ReadDocumentTool) Description() string { + return "Analyze documents (PDF, DOCX, images of documents, etc.) attached to the conversation. " + + "Use when you see tags and need to extract or analyze document content. " + + "Specify what you want to extract or analyze." +} + +func (t *ReadDocumentTool) Parameters() map[string]interface{} { + return map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "prompt": map[string]interface{}{ + "type": "string", + "description": "What to analyze. E.g. 'Extract all tables', 'Summarize key findings', 'What does page 3 say?'", + }, + "media_id": map[string]interface{}{ + "type": "string", + "description": "Optional: specific media_id from tag. If omitted, uses most recent document.", + }, + }, + "required": []string{"prompt"}, + } +} + +func (t *ReadDocumentTool) Execute(ctx context.Context, args map[string]interface{}) *Result { + prompt, _ := args["prompt"].(string) + if prompt == "" { + prompt = "Analyze this document and describe its contents." + } + mediaID, _ := args["media_id"].(string) + + // Resolve document file path from MediaRefs in context. + docPath, docMime, err := t.resolveDocumentFile(ctx, mediaID) + if err != nil { + return ErrorResult(err.Error()) + } + + slog.Info("read_document: resolved file", "path", docPath, "mime", docMime, "media_id", mediaID) + + // Read document file. + data, err := os.ReadFile(docPath) + if err != nil { + return ErrorResult(fmt.Sprintf("Failed to read document file: %v", err)) + } + slog.Info("read_document: file loaded", "size_bytes", len(data)) + if len(data) > documentMaxBytes { + return ErrorResult(fmt.Sprintf("Document too large: %d bytes (max %d)", len(data), documentMaxBytes)) + } + + // Fast path: text-readable files — return content directly without LLM. + if textReadableMIMEs[docMime] || strings.HasPrefix(docMime, "text/") { + content := string(data) + if len(data) > documentMaxTextBytes { + content = content[:documentMaxTextBytes] + "\n\n[... truncated at 500KB ...]" + } + slog.Info("read_document: returning text content directly", "mime", docMime, "size", len(data)) + return NewResult(content) + } + + // Find a document-capable provider. + provider, model, err := t.resolveDocumentProviderWithConfig(ctx) + if err != nil { + return ErrorResult(err.Error()) + } + + // Try primary provider, fallback to next available on error. + resp, usedProvider, usedModel := t.callDocumentProvider(ctx, provider, model, prompt, data, docMime) + if resp == nil { + // Primary failed — try fallback providers from priority list. + slog.Warn("read_document: primary provider failed, trying fallback", "primary", provider.Name()) + for _, fbName := range documentProviderPriority { + if fbName == provider.Name() { + continue // skip the one that already failed + } + fbProvider, fbModel, fbErr := t.resolveDocumentProviderByName(fbName) + if fbErr != nil { + continue + } + resp, usedProvider, usedModel = t.callDocumentProvider(ctx, fbProvider, fbModel, prompt, data, docMime) + if resp != nil { + slog.Info("read_document: fallback succeeded", "provider", usedProvider) + break + } + } + } + if resp == nil { + return ErrorResult("Document analysis failed: all providers returned errors") + } + + result := NewResult(resp.Content) + result.Usage = resp.Usage + result.Provider = usedProvider + result.Model = usedModel + return result +} + diff --git a/internal/tools/read_document_gemini.go b/internal/tools/read_document_gemini.go new file mode 100644 index 00000000..ee554379 --- /dev/null +++ b/internal/tools/read_document_gemini.go @@ -0,0 +1,120 @@ +package tools + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/providers" +) + +// geminiNativeDocumentCall sends a document to Gemini's native generateContent API. +// Uses inlineData with the document's MIME type (e.g. application/pdf) which Gemini +// supports natively, unlike the OpenAI-compat endpoint that only handles images. +func geminiNativeDocumentCall(ctx context.Context, apiKey, model, prompt string, docData []byte, docMime string) (*providers.ChatResponse, error) { + b64 := base64.StdEncoding.EncodeToString(docData) + + // Gemini generateContent request body. + body := map[string]interface{}{ + "contents": []map[string]interface{}{ + { + "parts": []map[string]interface{}{ + { + "inline_data": map[string]interface{}{ + "mime_type": docMime, + "data": b64, + }, + }, + { + "text": prompt, + }, + }, + }, + }, + "generationConfig": map[string]interface{}{ + "maxOutputTokens": 16384, + "temperature": 0.2, + }, + } + + bodyJSON, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("marshal request: %w", err) + } + + url := fmt.Sprintf("https://generativelanguage.googleapis.com/v1beta/models/%s:generateContent?key=%s", model, apiKey) + + httpReq, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(bodyJSON)) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + httpReq.Header.Set("Content-Type", "application/json") + + client := &http.Client{Timeout: 120 * time.Second} + httpResp, err := client.Do(httpReq) + if err != nil { + return nil, fmt.Errorf("HTTP request: %w", err) + } + defer httpResp.Body.Close() + + respBody, err := io.ReadAll(httpResp.Body) + if err != nil { + return nil, fmt.Errorf("read response: %w", err) + } + + if httpResp.StatusCode != 200 { + return nil, fmt.Errorf("HTTP %d: %s", httpResp.StatusCode, truncateStr(string(respBody), 500)) + } + + // Parse Gemini native response format. + var geminiResp struct { + Candidates []struct { + Content struct { + Parts []struct { + Text string `json:"text"` + } `json:"parts"` + } `json:"content"` + } `json:"candidates"` + UsageMetadata struct { + PromptTokenCount int `json:"promptTokenCount"` + CandidatesTokenCount int `json:"candidatesTokenCount"` + TotalTokenCount int `json:"totalTokenCount"` + } `json:"usageMetadata"` + } + + if err := json.Unmarshal(respBody, &geminiResp); err != nil { + return nil, fmt.Errorf("parse response: %w", err) + } + + // Extract text from response. + var content string + if len(geminiResp.Candidates) > 0 { + for _, part := range geminiResp.Candidates[0].Content.Parts { + if part.Text != "" { + if content != "" { + content += "\n" + } + content += part.Text + } + } + } + if content == "" { + return nil, fmt.Errorf("empty response from Gemini") + } + + return &providers.ChatResponse{ + Content: content, + FinishReason: "stop", + Usage: &providers.Usage{ + PromptTokens: geminiResp.UsageMetadata.PromptTokenCount, + CompletionTokens: geminiResp.UsageMetadata.CandidatesTokenCount, + TotalTokens: geminiResp.UsageMetadata.TotalTokenCount, + }, + }, nil +} + diff --git a/internal/tools/read_document_resolve.go b/internal/tools/read_document_resolve.go new file mode 100644 index 00000000..53dab51e --- /dev/null +++ b/internal/tools/read_document_resolve.go @@ -0,0 +1,199 @@ +package tools + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "log/slog" + "path/filepath" + "strings" + + "github.com/nextlevelbuilder/goclaw/internal/providers" +) + +// documentProviderPriority is the order in which providers are tried for document analysis. +// Gemini has best native PDF support (50MB, 258 tokens/page). +var documentProviderPriority = []string{"gemini", "anthropic", "openrouter", "dashscope"} + +// documentModelOverrides maps provider names to preferred document-capable models. +var documentModelOverrides = map[string]string{ + "gemini": "gemini-2.5-flash", + "openrouter": "google/gemini-2.5-flash", + "dashscope": "qwen-vl-max", +} + +// resolveDocumentFile finds the document file path from context MediaRefs. +func (t *ReadDocumentTool) resolveDocumentFile(ctx context.Context, mediaID string) (path, mime string, err error) { + if t.mediaLoader == nil { + return "", "", fmt.Errorf("no media storage configured — cannot access document files") + } + + refs := MediaDocRefsFromCtx(ctx) + if len(refs) == 0 { + return "", "", fmt.Errorf("no documents available in this conversation. The user may not have sent a document.") + } + + // Find specific media_id or use most recent document. + var ref *providers.MediaRef + if mediaID != "" { + for i := range refs { + if refs[i].ID == mediaID { + ref = &refs[i] + break + } + } + if ref == nil { + return "", "", fmt.Errorf("document with media_id %q not found in conversation", mediaID) + } + } else { + // Use the last (most recent) document ref. + ref = &refs[len(refs)-1] + } + + p, err := t.mediaLoader.LoadPath(ref.ID) + if err != nil { + return "", "", fmt.Errorf("document file not found: %v", err) + } + + // Determine MIME type: prefer ref's stored MIME, fall back to extension. + mime = ref.MimeType + if mime == "" || mime == "application/octet-stream" { + mime = mimeFromDocExt(filepath.Ext(p)) + } + + return p, mime, nil +} + +// resolveDocumentProviderWithConfig checks per-agent config, global settings, then hardcoded priority. +func (t *ReadDocumentTool) resolveDocumentProviderWithConfig(ctx context.Context) (providers.Provider, string, error) { + // 1. Global builtin_tools.settings + if p, model, ok := t.resolveFromBuiltinSettings(ctx); ok { + return p, model, nil + } + // 2. Hardcoded defaults + return t.resolveDocumentProvider() +} + +// resolveFromBuiltinSettings checks global builtin tool settings for provider/model config. +func (t *ReadDocumentTool) resolveFromBuiltinSettings(ctx context.Context) (providers.Provider, string, bool) { + settings := BuiltinToolSettingsFromCtx(ctx) + if settings == nil { + return nil, "", false + } + raw, ok := settings["read_document"] + if !ok || len(raw) == 0 { + return nil, "", false + } + var cfg struct { + Provider string `json:"provider"` + Model string `json:"model"` + } + if err := json.Unmarshal(raw, &cfg); err != nil || cfg.Provider == "" { + return nil, "", false + } + p, err := t.registry.Get(cfg.Provider) + if err != nil { + return nil, "", false + } + model := cfg.Model + if model == "" { + model = p.DefaultModel() + } + return p, model, true +} + +// resolveDocumentProvider finds the first available document-capable provider. +func (t *ReadDocumentTool) resolveDocumentProvider() (providers.Provider, string, error) { + for _, name := range documentProviderPriority { + p, err := t.registry.Get(name) + if err != nil { + continue + } + model := p.DefaultModel() + if override, ok := documentModelOverrides[name]; ok { + model = override + } + return p, model, nil + } + return nil, "", fmt.Errorf("no document-capable provider available (need one of: %v)", documentProviderPriority) +} + +// callDocumentProvider sends a document to a provider for analysis. +// For Gemini providers, uses native generateContent API (supports PDF natively). +// For others, falls back to OpenAI-compat chat with base64 document. +func (t *ReadDocumentTool) callDocumentProvider(ctx context.Context, provider providers.Provider, model, prompt string, data []byte, mime string) (*providers.ChatResponse, string, string) { + provName := provider.Name() + + // Gemini: use native API (OpenAI-compat endpoint doesn't support non-image MIME types). + if strings.HasPrefix(provName, "gemini") { + oaiProv, ok := provider.(*providers.OpenAIProvider) + if !ok { + slog.Warn("read_document: gemini provider is not OpenAIProvider", "provider", provName) + return nil, "", "" + } + apiKey := oaiProv.APIKey() + slog.Info("read_document: using gemini native API", + "provider", provName, "model", model, + "doc_size", len(data), "mime", mime) + resp, err := geminiNativeDocumentCall(ctx, apiKey, model, prompt, data, mime) + if err != nil { + slog.Warn("read_document: gemini native call failed", "error", err) + return nil, "", "" + } + return resp, provName, model + } + + // Other providers: use standard Chat API with document as base64 image_url. + slog.Info("read_document: using chat API", "provider", provName, "model", model, "doc_size", len(data)) + resp, err := provider.Chat(ctx, providers.ChatRequest{ + Messages: []providers.Message{ + { + Role: "user", + Content: prompt, + Images: []providers.ImageContent{{MimeType: mime, Data: base64.StdEncoding.EncodeToString(data)}}, + }, + }, + Model: model, + Options: map[string]interface{}{ + "max_tokens": 16384, + "temperature": 0.2, + }, + }) + if err != nil { + slog.Warn("read_document: chat call failed", "provider", provName, "error", err) + return nil, "", "" + } + return resp, provName, model +} + +// resolveDocumentProviderByName gets a specific provider by name and applies model override. +func (t *ReadDocumentTool) resolveDocumentProviderByName(name string) (providers.Provider, string, error) { + p, err := t.registry.Get(name) + if err != nil { + return nil, "", err + } + model := p.DefaultModel() + if override, ok := documentModelOverrides[name]; ok { + model = override + } + return p, model, nil +} + +// mimeFromDocExt returns MIME type for document file extensions. +func mimeFromDocExt(ext string) string { + switch strings.ToLower(ext) { + case ".pdf": + return "application/pdf" + case ".doc", ".docx": + return "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + case ".xls", ".xlsx": + return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + case ".ppt", ".pptx": + return "application/vnd.openxmlformats-officedocument.presentationml.presentation" + case ".csv": + return "text/csv" + default: + return "application/octet-stream" + } +} diff --git a/internal/tools/result.go b/internal/tools/result.go index 8de2695d..08d6564d 100644 --- a/internal/tools/result.go +++ b/internal/tools/result.go @@ -1,6 +1,9 @@ package tools -import "github.com/nextlevelbuilder/goclaw/internal/providers" +import ( + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/providers" +) // Result is the unified return type from tool execution. type Result struct { @@ -11,8 +14,8 @@ type Result struct { Async bool `json:"async"` // running asynchronously Err error `json:"-"` // internal error (not serialized) - // Media holds file paths to forward as output (e.g. images from delegation). - Media []string `json:"-"` + // Media holds media files to forward as output (e.g. images from delegation). + Media []bus.MediaFile `json:"-"` // Deliverable holds the primary work output from this tool execution. // Used to capture actual content (e.g. written file text, image prompt) for team diff --git a/internal/tools/subagent.go b/internal/tools/subagent.go index 1b0e707b..562326a5 100644 --- a/internal/tools/subagent.go +++ b/internal/tools/subagent.go @@ -59,7 +59,7 @@ type SubagentTask struct { OriginSessionKey string `json:"originSessionKey,omitempty"` // exact parent session key for announce routing (WS uses non-standard format) CreatedAt int64 `json:"createdAt"` CompletedAt int64 `json:"completedAt,omitempty"` - Media []string `json:"-"` // media file paths from tool results + Media []bus.MediaFile `json:"-"` // media files from tool results OriginTraceID uuid.UUID `json:"-"` // parent trace for announce linking OriginRootSpanID uuid.UUID `json:"-"` // parent agent's root span ID cancelFunc context.CancelFunc `json:"-"` // per-task context cancel diff --git a/internal/tools/subagent_exec.go b/internal/tools/subagent_exec.go index 0ebf6b56..f5842b23 100644 --- a/internal/tools/subagent_exec.go +++ b/internal/tools/subagent_exec.go @@ -166,7 +166,7 @@ func (sm *SubagentManager) executeTask(ctx context.Context, task *SubagentTask) } // Run LLM iteration loop (similar to agent loop but simplified) - var mediaPaths []string + var mediaFiles []bus.MediaFile maxIterations := 20 for iteration < maxIterations { @@ -229,7 +229,7 @@ func (sm *SubagentManager) executeTask(ctx context.Context, task *SubagentTask) // Capture media file paths from tool results (e.g. image generation). if len(result.Media) > 0 { - mediaPaths = append(mediaPaths, result.Media...) + mediaFiles = append(mediaFiles, result.Media...) } else if strings.HasPrefix(strings.TrimSpace(result.ForLLM), "MEDIA:") { // Fallback: parse MEDIA: prefix from ForLLM (same as agent loop's parseMediaResult) p := strings.TrimSpace(strings.TrimSpace(result.ForLLM)[6:]) @@ -237,7 +237,7 @@ func (sm *SubagentManager) executeTask(ctx context.Context, task *SubagentTask) p = strings.TrimSpace(p[:nl]) } if p != "" { - mediaPaths = append(mediaPaths, p) + mediaFiles = append(mediaFiles, bus.MediaFile{Path: p}) } } @@ -255,7 +255,7 @@ func (sm *SubagentManager) executeTask(ctx context.Context, task *SubagentTask) } task.Status = TaskStatusCompleted task.Result = finalContent - task.Media = mediaPaths + task.Media = mediaFiles sm.mu.Unlock() slog.Info("subagent completed", "id", task.ID, "iterations", iteration) diff --git a/internal/tools/subagent_spawn_tool.go b/internal/tools/subagent_spawn_tool.go index 254fe0a3..0ca9ad1a 100644 --- a/internal/tools/subagent_spawn_tool.go +++ b/internal/tools/subagent_spawn_tool.go @@ -260,9 +260,9 @@ Briefly tell the user what you've delegated and to whom. Be friendly and natural } mediaNote := "" - if len(result.MediaPaths) > 0 { + if len(result.Media) > 0 { mediaNote = fmt.Sprintf("\n\n[%d media file(s) attached — will be delivered automatically. Do NOT recreate or call create_image.]", - len(result.MediaPaths)) + len(result.Media)) } forLLM := fmt.Sprintf( @@ -273,8 +273,8 @@ Briefly tell the user what you've delegated and to whom. Be friendly and natural agentKey, result.Iterations, result.Content, mediaNote) toolResult := NewResult(forLLM) - if len(result.MediaPaths) > 0 { - toolResult.Media = result.MediaPaths + if len(result.Media) > 0 { + toolResult.Media = result.Media } return toolResult } diff --git a/internal/tracing/collector.go b/internal/tracing/collector.go index d1031ef7..4b014f58 100644 --- a/internal/tracing/collector.go +++ b/internal/tracing/collector.go @@ -48,6 +48,10 @@ type Collector struct { verbose bool // when true, LLM spans include full input messages exporter SpanExporter // optional external exporter (nil = disabled) + + // OnFlush is called after each flush cycle with the trace IDs that had + // their aggregates updated. Used to broadcast realtime trace events. + OnFlush func(traceIDs []uuid.UUID) } // NewCollector creates a new tracing collector backed by the given store. @@ -215,6 +219,15 @@ done: slog.Warn("tracing: aggregate update failed", "trace_id", traceID, "error", err) } } + + // Notify listeners about updated traces (realtime WS push). + if c.OnFlush != nil { + ids := make([]uuid.UUID, 0, len(dirty)) + for id := range dirty { + ids = append(ids, id) + } + c.OnFlush(ids) + } } } diff --git a/pkg/protocol/events.go b/pkg/protocol/events.go index 8c80535e..ed1710c1 100644 --- a/pkg/protocol/events.go +++ b/pkg/protocol/events.go @@ -56,6 +56,9 @@ const ( EventAgentLinkUpdated = "agent_link.updated" EventAgentLinkDeleted = "agent_link.deleted" + // Trace lifecycle events (realtime trace/span updates). + EventTraceUpdated = "trace.updated" + // Cache invalidation events (internal, not forwarded to WS clients). EventCacheInvalidate = "cache.invalidate" diff --git a/ui/web/src/api/protocol.ts b/ui/web/src/api/protocol.ts index c55ff0b7..f6d1e1b2 100644 --- a/ui/web/src/api/protocol.ts +++ b/ui/web/src/api/protocol.ts @@ -196,6 +196,9 @@ export const Events = { AGENT_LINK_CREATED: "agent_link.created", AGENT_LINK_UPDATED: "agent_link.updated", AGENT_LINK_DELETED: "agent_link.deleted", + + // Trace lifecycle + TRACE_UPDATED: "trace.updated", } as const; /** All event names relevant to team debug view */ diff --git a/ui/web/src/hooks/use-query-invalidation.ts b/ui/web/src/hooks/use-query-invalidation.ts index cec676dd..81e34fbc 100644 --- a/ui/web/src/hooks/use-query-invalidation.ts +++ b/ui/web/src/hooks/use-query-invalidation.ts @@ -12,11 +12,14 @@ import type { AgentEventPayload } from "@/types/chat"; export function useWsQueryInvalidation() { const queryClient = useQueryClient(); - // When an agent run completes/fails → refresh sessions + traces + usage + // When an agent run starts/completes/fails → refresh sessions + traces + usage const handleAgentEvent = useCallback( (payload: unknown) => { const event = payload as AgentEventPayload; if (!event) return; + if (event.type === "run.started") { + queryClient.invalidateQueries({ queryKey: queryKeys.traces.all }); + } if (event.type === "run.completed" || event.type === "run.failed") { queryClient.invalidateQueries({ queryKey: queryKeys.sessions.all }); queryClient.invalidateQueries({ queryKey: queryKeys.traces.all }); @@ -26,6 +29,14 @@ export function useWsQueryInvalidation() { [queryClient], ); + // Trace aggregate updates (spans flushed) → refresh traces list + const handleTraceUpdated = useCallback( + () => { + queryClient.invalidateQueries({ queryKey: queryKeys.traces.all }); + }, + [queryClient], + ); + // Cron events → refresh cron jobs list const handleCronEvent = useCallback( () => { @@ -43,6 +54,7 @@ export function useWsQueryInvalidation() { ); useWsEvent(Events.AGENT, handleAgentEvent); + useWsEvent(Events.TRACE_UPDATED, handleTraceUpdated); useWsEvent(Events.CRON, handleCronEvent); useWsEvent(Events.HEALTH, handleHealthEvent); } diff --git a/ui/web/src/pages/builtin-tools/builtin-tool-settings-dialog.tsx b/ui/web/src/pages/builtin-tools/builtin-tool-settings-dialog.tsx index a5f257f1..b07b1830 100644 --- a/ui/web/src/pages/builtin-tools/builtin-tool-settings-dialog.tsx +++ b/ui/web/src/pages/builtin-tools/builtin-tool-settings-dialog.tsx @@ -30,7 +30,7 @@ interface Props { onSave: (name: string, settings: Record) => Promise; } -const MEDIA_TOOLS = new Set(["read_image", "create_image"]); +const MEDIA_TOOLS = new Set(["read_image", "create_image", "read_document"]); export function BuiltinToolSettingsDialog({ tool, open, onOpenChange, onSave }: Props) { const isMedia = tool ? MEDIA_TOOLS.has(tool.name) : false; diff --git a/ui/web/src/pages/traces/trace-detail-dialog.tsx b/ui/web/src/pages/traces/trace-detail-dialog.tsx index e5b9b11c..6b956d53 100644 --- a/ui/web/src/pages/traces/trace-detail-dialog.tsx +++ b/ui/web/src/pages/traces/trace-detail-dialog.tsx @@ -1,4 +1,4 @@ -import { useState, useEffect, useMemo } from "react"; +import { useState, useEffect, useCallback, useMemo } from "react"; import { Dialog, DialogContent, @@ -8,8 +8,11 @@ import { import { Badge } from "@/components/ui/badge"; import { ChevronRight, ChevronDown, Copy, Check } from "lucide-react"; import { useClipboard } from "@/hooks/use-clipboard"; +import { useWsEvent } from "@/hooks/use-ws-event"; +import { Events } from "@/api/protocol"; import { formatDate, formatDuration, formatTokens, computeDurationMs } from "@/lib/format"; import type { TraceData, SpanData } from "./hooks/use-traces"; +import type { AgentEventPayload } from "@/types/chat"; interface SpanNode { span: SpanData; @@ -51,6 +54,15 @@ export function TraceDetailDialog({ traceId, onClose, getTrace, onNavigateTrace const [loading, setLoading] = useState(true); const { copied, copy } = useClipboard(); + const fetchTrace = useCallback(() => { + getTrace(traceId).then((result) => { + if (result) { + setTrace(result.trace); + setSpans(result.spans ?? []); + } + }); + }, [traceId, getTrace]); + useEffect(() => { setLoading(true); getTrace(traceId) @@ -63,6 +75,34 @@ export function TraceDetailDialog({ traceId, onClose, getTrace, onNavigateTrace .finally(() => setLoading(false)); }, [traceId, getTrace]); + // Auto-refetch when trace aggregates update (spans flushed every ~5s) + useWsEvent( + Events.TRACE_UPDATED, + useCallback( + (payload: unknown) => { + const data = payload as { trace_ids?: string[] }; + if (data?.trace_ids?.includes(traceId)) { + fetchTrace(); + } + }, + [traceId, fetchTrace], + ), + ); + + // Also refetch on run completion (final status + output) + useWsEvent( + Events.AGENT, + useCallback( + (payload: unknown) => { + const event = payload as AgentEventPayload; + if (event?.type === "run.completed" || event?.type === "run.failed") { + fetchTrace(); + } + }, + [fetchTrace], + ), + ); + const spanTree = useMemo(() => buildSpanTree(spans), [spans]); return (