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
This commit is contained in:
viettranx committed 2026-03-08 14:00:34 +07:00
1 parent d9cdbb8e6e
commit 0f2737ce53
43 files changed
+1313 -135

No files matched your search

+11 -1
View File
@@ -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")
}
+4
View File
@@ -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"},
+10 -7
View File
@@ -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,
})
}
+30 -3
View File
@@ -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()
+80 -20
View File
@@ -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{
+21
View File
@@ -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"
}
+10 -2
View File
@@ -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)
+126 -7
View File
@@ -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)) {
@@ -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)
}
+80
View File
@@ -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)
}
+5
View File
@@ -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)
+3
View File
@@ -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 <media:image> tags",
"read_document": "Analyze documents (PDF, DOCX, etc.) attached to the conversation. MUST call this when you see <media:document> tags",
"create_image": "Generate images from text descriptions using AI",
}
// BuildSystemPrompt constructs the full system prompt with all sections.
+1 -1
View File
@@ -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...)
}
+8 -1
View File
@@ -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)
+7 -1
View File
@@ -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,
+21 -3
View File
@@ -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,
+48 -29
View File
@@ -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, "<media:voice>")
}
case "document":
tags = append(tags, "<media:document>")
if m.FileName != "" {
tags = append(tags, fmt.Sprintf("<media:document name=%q>", m.FileName))
} else {
tags = append(tags, "<media:document>")
}
}
}
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)
+19 -6
View File
@@ -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))
}()
}
+139
View File
@@ -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()
}
+11 -1
View File
@@ -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
+21
View File
@@ -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]
}
+3
View File
@@ -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 {
+5
View File
@@ -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
}
+3 -1
View File
@@ -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
}
+2 -1
View File
@@ -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
+19 -4
View File
@@ -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
+4 -4
View File
@@ -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,
})
}
+35 -1
View File
@@ -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
}
+1 -1
View File
@@ -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
}
+3 -2
View File
@@ -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
}
+158
View File
@@ -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 <media:document> 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 <media:document> 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
}
+120
View File
@@ -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
}
+199
View File
@@ -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"
}
}
+6 -3
View File
@@ -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
+1 -1
View File
@@ -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
+4 -4
View File
@@ -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)
+4 -4
View File
@@ -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
}
+13
View File
@@ -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)
}
}
}
+3
View File
@@ -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"
+3
View File
@@ -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 */
+13 -1
View File
@@ -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);
}
@@ -30,7 +30,7 @@ interface Props {
onSave: (name: string, settings: Record<string, unknown>) => Promise<void>;
}
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;
@@ -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 (