diff --git a/CLAUDE.md b/CLAUDE.md index cb6f258..57ecad7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -103,6 +103,8 @@ make lint # Run linters - Go 1.25+ (per go.mod) + go-chi/chi (HTTP), mark3labs/mcp-go (MCP), ory/fosite (OAuth), spf13/cobra (CLI), modernc.org/sqlite (storage), TFMV/hnsw (vectors). NEW: coreos/go-oidc/v3 (OIDC), golang.org/x/oauth2 (OAuth client) (007-platform-features-bundle) - Go 1.25+ (backend), SvelteKit 2 + Svelte 5 (frontend), SvelteKit (website) + go-chi/chi (HTTP), mark3labs/mcp-go (MCP), modernc.org/sqlite (storage), SherClockHolmes/webpush-go (push notifications — NEW) (008-webui-pwa-analytics) - SQLite (existing DB, 1 new migration for push_subscriptions), localStorage (font size) (008-webui-pwa-analytics) +- Go 1.25+ (backend), Svelte 5 + Tailwind (frontend) + go-chi/chi (HTTP), mark3labs/mcp-go (MCP), modernc.org/sqlite (storage), spf13/cobra (CLI) (009-attachments-threads) +- SQLite (modernc.org/sqlite, pure Go) + content-addressable filesystem (SHA-256) (009-attachments-threads) ## Recent Changes - 002-mcp-auth-ux-polish: Added Go 1.23+ + ory/fosite (OAuth 2.1), mark3labs/mcp-go (MCP server), go-chi/chi (HTTP), Svelte 5 + Tailwind (Web UI) diff --git a/autonomous_summary.md b/autonomous_summary.md index 5ed8ae7..2b941b0 100644 --- a/autonomous_summary.md +++ b/autonomous_summary.md @@ -1,62 +1,91 @@ -# Autonomous Execution Summary: SynapBus v0.7.0 +# Autonomous Implementation Summary: Attachments & Threads Enhancement +**Branch**: `009-attachments-threads` **Date**: 2026-03-17 -**Branch**: `008-webui-pwa-analytics` -**Status**: Complete — all tests pass, binary builds, endpoints verified +**Status**: Complete -## Features Implemented +## What Was Built -### 1. Analytics Dashboard (P1) -- 4 new REST endpoints (`/api/analytics/timeline`, `/top-agents`, `/top-channels`, `/summary`) -- SVG bar chart (`AnalyticsChart.svelte`), ranked list (`TopList.svelte`), redesigned dashboard with stat cards and time span selector (1h, 4h, 24h, 7d, 1month) +### Attachments (Web UI + MCP + Admin CLI) -### 2. PWA Conversion (P1) -- PWA manifest (`manifest.json`), service worker (`sw.js`), SVG icon -- Cache-first for static assets, network-only for API, push notification handling +1. **Web UI file upload**: Users can attach files (images, PDFs, text files) when composing messages via a paperclip button in the compose form. Uploaded files show as chips with remove buttons. -### 3. Push Notifications (P1) -- `internal/push/` package — VAPID key generation, Web Push sending, SQLite subscription store -- API: `POST/DELETE /api/push/subscribe`, `GET /api/push/vapid-key` -- Push toggle in Settings, migration `012_push_subscriptions.sql` +2. **Attachment display**: Image attachments render as thumbnails (200x200px max) inline with messages. Clicking opens a fullscreen overlay with close (Escape, click-outside, X button) and download. Non-image files show as file icons with filename, size, and download link. -### 4. Auto-Resizing Textarea (P1) -- ComposeForm textarea auto-grows 3→12 lines, then scrollbar. Resets on send. +3. **MCP agent support**: Agents can attach files by uploading via `upload_attachment` action and passing hashes in the new `attachments` parameter on `send_message`. Tool descriptions updated to document the workflow. -### 5. Smart Mention/Channel Highlighting (P2) -- Entities store caches agents/channels. MessageBody validates @mentions and #channels: - existing → link, deleted → "inactive" badge, unknown → plain text. Handles email/issue number edge cases. +4. **File type validation**: Upload restricted to images (jpg, png, gif, webp, svg), PDFs, and text files (txt, md, csv, json, xml, yaml, log). Invalid types rejected with clear error. -### 6. Editable Agent Display Name (P2) -- Inline edit on agent detail page (click → edit, Enter → save, Escape → cancel) +5. **Admin backup/restore**: `synapbus attachments backup --output path.tar.gz` and `synapbus attachments restore --input path.tar.gz` commands. Restore is dedup-safe (skips existing files). -### 7. Editable Human Display Name (P2) -- `PUT /api/auth/profile` endpoint, `UpdateDisplayName` in UserStore, Settings page field +### Threads -### 8. Font Size Preference (P3) -- fontSize store (12–24px, 2px steps), -/+ controls in Settings, persisted in localStorage +1. **Thread visibility**: Messages with replies show an always-visible "N replies" badge with a chat bubble icon. Clicking opens the thread panel. -### 9. Version Display (P3) -- `GET /api/version` endpoint, version footer in layout linked to GitHub repo +2. **Reply count in API**: All message API responses now include `reply_count` (batch-loaded via efficient GROUP BY query). -### 10. MCP Prompts (P2) -- 4 prompts: daily-digest, agent-health-check, channel-overview, debug-agent -- `internal/mcp/prompts.go` registered in server.go +3. **MCP threading**: `reply_to` parameter description updated to clearly guide agents on threading behavior. Thread context visible in MCP responses. -### 11. Website Update (P3) -- Updated hero/features messaging at ~/repos/synapbus-website/ for individual/small-team positioning +4. **Attachment display in threads**: Thread panel also renders attachment thumbnails/file icons. + +## Files Modified + +### Backend (Go) +| File | Changes | +|------|---------| +| `internal/messaging/types.go` | Added `AttachmentInfo` struct, `ReplyCount`, `Attachments` fields to `Message` | +| `internal/messaging/options.go` | Added `Attachments []string` to `SendOptions` | +| `internal/messaging/store.go` | Added `GetReplyCounts` batch query method + interface | +| `internal/messaging/service.go` | Added `AttachmentLinker` interface, `EnrichMessages`, attachment linking in `SendMessage` | +| `internal/attachments/model.go` | Added `ErrUnsupportedType` error | +| `internal/attachments/mime.go` | Added `IsAllowedType` function | +| `internal/attachments/service.go` | Added file type validation in `Upload` | +| `internal/api/messages_handler.go` | Accept `attachments[]` in send, `EnrichMessages` in all handlers | +| `internal/mcp/tools_hybrid.go` | Added `attachments` param to `send_message`, updated descriptions, enrich responses | +| `internal/mcp/bridge.go` | Handle attachments in `callSendMessage`, `callSendChannelMessage` | +| `internal/channels/service.go` | Added `attachments` param to `BroadcastMessage` | +| `internal/actions/registry.go` | Updated `upload_attachment` description | +| `cmd/synapbus/main.go` | Added `attachmentLinkerAdapter`, wired into messaging service | +| `cmd/synapbus/admin.go` | Added `backup` and `restore` subcommands | + +### Frontend (Svelte) +| File | Changes | +|------|---------| +| `web/src/lib/api/client.ts` | Added `attachments.upload()`, `attachments` param in `messages.send()` | +| `web/src/lib/components/ComposeForm.svelte` | Attachment upload button, file picker, preview chips | +| `web/src/lib/components/AttachmentPreview.svelte` | **NEW** — thumbnail + fullscreen overlay component | +| `web/src/lib/components/MessageList.svelte` | Attachment display, thread reply count badges | +| `web/src/lib/components/ThreadPanel.svelte` | Attachment display in thread messages | + +### Tests +| File | New Tests | +|------|-----------| +| `internal/messaging/store_test.go` | `TestSQLiteMessageStore_GetReplyCounts` (4 subtests) | +| `internal/messaging/service_test.go` | `TestMessagingService_EnrichMessages` (3 subtests) | +| `internal/attachments/mime_test.go` | `TestIsAllowedType` (12 cases) | +| `internal/attachments/service_test.go` | `TestService_Upload_FileTypeValidation` (5 cases) | +| `internal/channels/service_test.go` | Updated 13 call sites for new `BroadcastMessage` signature | ## Test Results -All 24 Go packages PASS. All API endpoints verified via curl. Web UI builds successfully. Binary compiles with CGO_ENABLED=0. +- **Go tests**: 25 packages, all pass, 0 failures +- **Integration tests**: 9 E2E tests, all pass +- **New tests**: 24 test cases added, all pass +- **Web build**: Svelte SPA builds successfully +- **Binary build**: 90MB arm64 binary compiles cleanly -## New Files +## Architecture Decisions -- `internal/api/analytics_handler.go` + test -- `internal/api/version_handler.go` + test -- `internal/api/push_handler.go` -- `internal/push/service.go` + test, `store.go` + test -- `internal/mcp/prompts.go` + test -- `schema/012_push_subscriptions.sql` -- `web/src/lib/components/AnalyticsChart.svelte`, `TopList.svelte` -- `web/src/lib/stores/fontSize.ts`, `entities.ts` -- `web/static/manifest.json`, `sw.js`, `icons/icon.svg` +1. **No circular dependencies**: Used `AttachmentLinker` interface + adapter pattern to avoid messaging->attachments import +2. **Batch loading**: Reply counts loaded via single GROUP BY query; attachments loaded per-message (acceptable for LAN scale) +3. **Client-side thumbnails**: CSS-only resizing (no server-side image processing, preserves zero-CGO constraint) +4. **Zero new migrations**: Leveraged existing `reply_to` column (migration 007) and `attachments` table (migration 001) +5. **Zero new dependencies**: All using Go stdlib + existing libraries + +## Constitution Compliance + +All 10 principles satisfied: +- I. Local-First: No external dependencies added +- II. MCP-Native: Agent features use MCP tools exclusively +- III. Pure Go, Zero CGO: stdlib archive/tar + compress/gzip for backup +- X. Web UI First-Class: Full attachment and thread UI experience diff --git a/cmd/synapbus/admin.go b/cmd/synapbus/admin.go index d1e0207..1626dda 100644 --- a/cmd/synapbus/admin.go +++ b/cmd/synapbus/admin.go @@ -1,11 +1,15 @@ package main import ( + "archive/tar" "bufio" + "compress/gzip" "encoding/json" "fmt" + "io" "net" "os" + "path/filepath" "strings" "text/tabwriter" @@ -987,7 +991,48 @@ func addAdminCommands(rootCmd *cobra.Command) { }, } - attachmentsCmd.AddCommand(attachmentsGCCmd) + var attachmentsBackupOutput string + var attachmentsBackupDataDir string + attachmentsBackupCmd := &cobra.Command{ + Use: "backup", + Short: "Create a tar.gz backup of all attachments (no server required)", + RunE: func(cmd *cobra.Command, args []string) error { + attachDir := filepath.Join(attachmentsBackupDataDir, "attachments") + if _, err := os.Stat(attachDir); os.IsNotExist(err) { + return fmt.Errorf("attachments directory does not exist: %s", attachDir) + } + fileCount, totalSize, err := backupAttachments(attachDir, attachmentsBackupOutput) + if err != nil { + return fmt.Errorf("backup failed: %w", err) + } + fmt.Printf("Backup complete: %d files, %s total, written to %s\n", fileCount, formatBytes(totalSize), attachmentsBackupOutput) + return nil + }, + } + attachmentsBackupCmd.Flags().StringVar(&attachmentsBackupOutput, "output", "", "Output path for the tar.gz archive") + attachmentsBackupCmd.Flags().StringVar(&attachmentsBackupDataDir, "data", "./data", "Data directory") + attachmentsBackupCmd.MarkFlagRequired("output") + + var attachmentsRestoreInput string + var attachmentsRestoreDataDir string + attachmentsRestoreCmd := &cobra.Command{ + Use: "restore", + Short: "Restore attachments from a tar.gz backup (no server required)", + RunE: func(cmd *cobra.Command, args []string) error { + attachDir := filepath.Join(attachmentsRestoreDataDir, "attachments") + restored, skipped, err := restoreAttachments(attachDir, attachmentsRestoreInput) + if err != nil { + return fmt.Errorf("restore failed: %w", err) + } + fmt.Printf("Restore complete: %d files restored, %d files skipped (already exist)\n", restored, skipped) + return nil + }, + } + attachmentsRestoreCmd.Flags().StringVar(&attachmentsRestoreInput, "input", "", "Input path for the tar.gz archive") + attachmentsRestoreCmd.Flags().StringVar(&attachmentsRestoreDataDir, "data", "./data", "Data directory") + attachmentsRestoreCmd.MarkFlagRequired("input") + + attachmentsCmd.AddCommand(attachmentsGCCmd, attachmentsBackupCmd, attachmentsRestoreCmd) // ----- add persistent flag and commands to root ----- rootCmd.PersistentFlags().StringVar(&adminSocket, "socket", "/tmp/synapbus.sock", "Path to admin Unix socket") @@ -1012,3 +1057,146 @@ func toTableRows(data []map[string]string, headerMap map[string]string) []map[st } return rows } + +// backupAttachments creates a tar.gz archive of the attachments directory. +// Returns the number of files archived and total bytes of file content. +func backupAttachments(attachmentsDir, outputPath string) (int, int64, error) { + outFile, err := os.Create(outputPath) + if err != nil { + return 0, 0, fmt.Errorf("create output file: %w", err) + } + defer outFile.Close() + + gzw := gzip.NewWriter(outFile) + defer gzw.Close() + + tw := tar.NewWriter(gzw) + defer tw.Close() + + var fileCount int + var totalSize int64 + + err = filepath.Walk(attachmentsDir, func(path string, info os.FileInfo, err error) error { + if err != nil { + return err + } + // Skip directories — tar entries for files include the path. + if info.IsDir() { + return nil + } + + relPath, err := filepath.Rel(attachmentsDir, path) + if err != nil { + return fmt.Errorf("relative path: %w", err) + } + + header, err := tar.FileInfoHeader(info, "") + if err != nil { + return fmt.Errorf("file info header: %w", err) + } + header.Name = relPath + + if err := tw.WriteHeader(header); err != nil { + return fmt.Errorf("write header: %w", err) + } + + f, err := os.Open(path) + if err != nil { + return fmt.Errorf("open file: %w", err) + } + defer f.Close() + + if _, err := io.Copy(tw, f); err != nil { + return fmt.Errorf("copy file: %w", err) + } + + fileCount++ + totalSize += info.Size() + return nil + }) + + return fileCount, totalSize, err +} + +// restoreAttachments extracts a tar.gz archive into the attachments directory. +// Files that already exist on disk are skipped. Returns (restored, skipped) counts. +func restoreAttachments(attachmentsDir, inputPath string) (int, int, error) { + inFile, err := os.Open(inputPath) + if err != nil { + return 0, 0, fmt.Errorf("open input file: %w", err) + } + defer inFile.Close() + + gzr, err := gzip.NewReader(inFile) + if err != nil { + return 0, 0, fmt.Errorf("gzip reader: %w", err) + } + defer gzr.Close() + + tr := tar.NewReader(gzr) + + var restored, skipped int + + for { + header, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + return restored, skipped, fmt.Errorf("read tar entry: %w", err) + } + + // Only handle regular files. + if header.Typeflag != tar.TypeReg { + continue + } + + // Sanitize: reject absolute paths and path traversal. + cleanName := filepath.Clean(header.Name) + if filepath.IsAbs(cleanName) || strings.HasPrefix(cleanName, "..") { + return restored, skipped, fmt.Errorf("invalid path in archive: %s", header.Name) + } + + destPath := filepath.Join(attachmentsDir, cleanName) + + // Skip if already exists (content-addressable, so same hash = same content). + if _, err := os.Stat(destPath); err == nil { + skipped++ + continue + } + + // Ensure parent directory exists. + if err := os.MkdirAll(filepath.Dir(destPath), 0o755); err != nil { + return restored, skipped, fmt.Errorf("create directory: %w", err) + } + + outFile, err := os.Create(destPath) + if err != nil { + return restored, skipped, fmt.Errorf("create file: %w", err) + } + + if _, err := io.Copy(outFile, tr); err != nil { + outFile.Close() + return restored, skipped, fmt.Errorf("write file: %w", err) + } + outFile.Close() + + restored++ + } + + return restored, skipped, nil +} + +// formatBytes returns a human-readable byte count string. +func formatBytes(b int64) string { + const unit = 1024 + if b < unit { + return fmt.Sprintf("%d B", b) + } + div, exp := int64(unit), 0 + for n := b / unit; n >= unit; n /= unit { + div *= unit + exp++ + } + return fmt.Sprintf("%.1f %ciB", float64(b)/float64(div), "KMGTPE"[exp]) +} diff --git a/cmd/synapbus/main.go b/cmd/synapbus/main.go index be65872..5e90fcf 100644 --- a/cmd/synapbus/main.go +++ b/cmd/synapbus/main.go @@ -281,6 +281,7 @@ func runServe(cmd *cobra.Command, args []string) error { } attachmentStore := attachments.NewSQLiteStore(db.DB, slog.Default()) attachmentService := attachments.NewService(attachmentStore, cas, slog.Default()) + msgService.SetAttachmentLinker(&attachmentLinkerAdapter{svc: attachmentService}) slog.Info("attachment service initialized", "dir", attachmentsDir) // Initialize auth subsystem @@ -836,6 +837,33 @@ func (a *a2aAgentListerAdapter) ListAllActiveAgents(ctx context.Context) ([]a2a. return result, nil } +// attachmentLinkerAdapter adapts attachments.Service to messaging.AttachmentLinker. +type attachmentLinkerAdapter struct { + svc *attachments.Service +} + +func (a *attachmentLinkerAdapter) AttachToMessage(ctx context.Context, hash string, messageID int64) error { + return a.svc.AttachToMessage(ctx, hash, messageID) +} + +func (a *attachmentLinkerAdapter) GetByMessageID(ctx context.Context, messageID int64) ([]messaging.AttachmentInfo, error) { + atts, err := a.svc.GetByMessageID(ctx, messageID) + if err != nil { + return nil, err + } + results := make([]messaging.AttachmentInfo, len(atts)) + for i, att := range atts { + results[i] = messaging.AttachmentInfo{ + Hash: att.Hash, + OriginalFilename: att.OriginalFilename, + Size: att.Size, + MIMEType: att.MIMEType, + IsImage: attachments.IsImageType(att.MIMEType), + } + } + return results, nil +} + // agentListerAdapter adapts agents.AgentService to auth.AgentLister. type agentListerAdapter struct { agentService *agents.AgentService diff --git a/internal/actions/registry.go b/internal/actions/registry.go index b390e80..118597a 100644 --- a/internal/actions/registry.go +++ b/internal/actions/registry.go @@ -425,7 +425,7 @@ func allActions() []Action { { Name: "upload_attachment", Category: "attachments", - Description: "Upload a file attachment. Content must be base64-encoded. Returns the SHA-256 hash for later retrieval. Max file size: 50MB.", + Description: "Upload a file attachment. Content must be base64-encoded. Returns the SHA-256 hash for later retrieval. Upload first, then use the returned hash in send_message's attachments parameter to link it to a message. Max file size: 50MB.", Params: []Param{ {Name: "content", Type: "string", Description: "Base64-encoded file content", Required: true}, {Name: "filename", Type: "string", Description: "Original filename (optional, used for MIME detection and display)"}, diff --git a/internal/api/attachments_handler.go b/internal/api/attachments_handler.go index a383740..fafe798 100644 --- a/internal/api/attachments_handler.go +++ b/internal/api/attachments_handler.go @@ -137,6 +137,8 @@ func (h *AttachmentsHandler) Upload(w http.ResponseWriter, r *http.Request) { http.Error(w, `{"error":"empty file not allowed"}`, http.StatusBadRequest) case attachments.ErrFileTooLarge: http.Error(w, `{"error":"file exceeds maximum size of 50MB"}`, http.StatusRequestEntityTooLarge) + case attachments.ErrUnsupportedType: + http.Error(w, `{"error":"unsupported file type: only images, PDFs, and text files are allowed"}`, http.StatusBadRequest) default: h.logger.Error("upload attachment failed", "error", err) http.Error(w, `{"error":"internal server error"}`, http.StatusInternalServerError) diff --git a/internal/api/channels_handler.go b/internal/api/channels_handler.go index 8bd5c10..78346e7 100644 --- a/internal/api/channels_handler.go +++ b/internal/api/channels_handler.go @@ -228,6 +228,9 @@ func (h *ChannelsHandler) ChannelMessages(w http.ResponseWriter, r *http.Request return } + // Enrich messages with reply counts and attachments. + h.msgService.EnrichMessages(r.Context(), paginated.Messages) + // Compute last_read_message_id across owned agents var lastReadMessageID int64 ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) diff --git a/internal/api/messages_handler.go b/internal/api/messages_handler.go index 2b559f4..a822b7f 100644 --- a/internal/api/messages_handler.go +++ b/internal/api/messages_handler.go @@ -86,6 +86,8 @@ func (h *MessagesHandler) ListMessages(w http.ResponseWriter, r *http.Request) { allMessages = []*messaging.Message{} } + h.msgService.EnrichMessages(r.Context(), allMessages) + sortMessagesByTime(allMessages) if len(allMessages) > limit { allMessages = allMessages[:limit] @@ -122,6 +124,8 @@ func (h *MessagesHandler) GetMessage(w http.ResponseWriter, r *http.Request) { return } + h.msgService.EnrichMessages(r.Context(), []*messaging.Message{msg}) + writeJSON(w, http.StatusOK, msg) } @@ -230,6 +234,8 @@ func (h *MessagesHandler) GetConversation(w http.ResponseWriter, r *http.Request return } + h.msgService.EnrichMessages(r.Context(), messages) + writeJSON(w, http.StatusOK, map[string]any{ "conversation": conv, "messages": messages, @@ -245,14 +251,15 @@ func (h *MessagesHandler) SendMessage(w http.ResponseWriter, r *http.Request) { } var req struct { - From string `json:"from"` - To string `json:"to"` - Body string `json:"body"` - Priority int `json:"priority"` - ChannelID *int64 `json:"channel_id,omitempty"` - ConversationID *int64 `json:"conversation_id,omitempty"` - Subject string `json:"subject,omitempty"` - ReplyTo *int64 `json:"reply_to,omitempty"` + From string `json:"from"` + To string `json:"to"` + Body string `json:"body"` + Priority int `json:"priority"` + ChannelID *int64 `json:"channel_id,omitempty"` + ConversationID *int64 `json:"conversation_id,omitempty"` + Subject string `json:"subject,omitempty"` + ReplyTo *int64 `json:"reply_to,omitempty"` + Attachments []string `json:"attachments,omitempty"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -307,6 +314,7 @@ func (h *MessagesHandler) SendMessage(w http.ResponseWriter, r *http.Request) { ConversationID: req.ConversationID, Subject: req.Subject, ReplyTo: req.ReplyTo, + Attachments: req.Attachments, } msg, err := h.msgService.SendMessage(r.Context(), req.From, req.To, req.Body, opts) @@ -319,6 +327,8 @@ func (h *MessagesHandler) SendMessage(w http.ResponseWriter, r *http.Request) { // SSE broadcast is handled by the MessageListener on the messaging // service, so it fires for both REST and MCP message paths. + h.msgService.EnrichMessages(r.Context(), []*messaging.Message{msg}) + writeJSON(w, http.StatusCreated, msg) } @@ -465,6 +475,8 @@ func (h *MessagesHandler) SearchMessages(w http.ResponseWriter, r *http.Request) allMessages = []*messaging.Message{} } + h.msgService.EnrichMessages(r.Context(), allMessages) + writeJSON(w, http.StatusOK, map[string]any{ "messages": allMessages, "query": query, @@ -505,6 +517,8 @@ func (h *MessagesHandler) GetReplies(w http.ResponseWriter, r *http.Request) { return } + h.msgService.EnrichMessages(r.Context(), replies) + writeJSON(w, http.StatusOK, map[string]any{ "replies": replies, "total": len(replies), @@ -550,6 +564,8 @@ func (h *MessagesHandler) DMMessages(w http.ResponseWriter, r *http.Request) { return } + h.msgService.EnrichMessages(r.Context(), msgs) + // Include last_read_message_id for the human agent's DM with the peer lastRead, _ := h.msgService.GetLastReadForDM(r.Context(), agentNames, peerAgent) diff --git a/internal/api/notifications_handler_test.go b/internal/api/notifications_handler_test.go index e958815..f51367d 100644 --- a/internal/api/notifications_handler_test.go +++ b/internal/api/notifications_handler_test.go @@ -333,7 +333,7 @@ func TestChannelMessages_IncludesLastRead(t *testing.T) { } // Broadcast messages - msgs, err := channelService.BroadcastMessage(ctx, ch.ID, "human-agent", "Hello channel", 5, "", nil) + msgs, err := channelService.BroadcastMessage(ctx, ch.ID, "human-agent", "Hello channel", 5, "", nil, nil) if err != nil { t.Fatalf("broadcast: %v", err) } diff --git a/internal/attachments/mime.go b/internal/attachments/mime.go index 039d2ce..90f7539 100644 --- a/internal/attachments/mime.go +++ b/internal/attachments/mime.go @@ -118,3 +118,27 @@ func DefaultFilename(mimeType string) string { func IsImageType(mimeType string) bool { return imageTypes[mimeType] } + +// IsAllowedType returns true if the MIME type is allowed for upload. +// Allowed: image/*, application/pdf, text/*. +func IsAllowedType(mimeType string) bool { + // Normalize: strip parameters like "; charset=utf-8". + base := mimeType + if idx := strings.Index(mimeType, ";"); idx >= 0 { + base = strings.TrimSpace(mimeType[:idx]) + } + if strings.HasPrefix(base, "image/") { + return true + } + if base == "application/pdf" { + return true + } + if strings.HasPrefix(base, "text/") { + return true + } + // Also allow JSON and XML which may be detected as application/* + if base == "application/json" || base == "application/xml" { + return true + } + return false +} diff --git a/internal/attachments/mime_test.go b/internal/attachments/mime_test.go index 6df64ef..142e674 100644 --- a/internal/attachments/mime_test.go +++ b/internal/attachments/mime_test.go @@ -154,6 +154,35 @@ func TestIsImageType(t *testing.T) { } } +func TestIsAllowedType(t *testing.T) { + tests := []struct { + mimeType string + want bool + }{ + {"image/png", true}, + {"image/jpeg", true}, + {"image/gif", true}, + {"application/pdf", true}, + {"text/plain", true}, + {"text/csv", true}, + {"text/plain; charset=utf-8", true}, + {"application/json", true}, + {"application/octet-stream", false}, + {"application/zip", false}, + {"application/x-executable", false}, + {"video/mp4", false}, + } + + for _, tt := range tests { + t.Run(tt.mimeType, func(t *testing.T) { + got := IsAllowedType(tt.mimeType) + if got != tt.want { + t.Errorf("IsAllowedType(%q) = %v, want %v", tt.mimeType, got, tt.want) + } + }) + } +} + func min(a, b int) int { if a < b { return a diff --git a/internal/attachments/model.go b/internal/attachments/model.go index 93e693e..40799c9 100644 --- a/internal/attachments/model.go +++ b/internal/attachments/model.go @@ -11,10 +11,11 @@ const MaxFileSize = 50 * 1024 * 1024 // 50 MB // Sentinel errors. var ( - ErrNotFound = errors.New("attachment not found") - ErrFileTooLarge = errors.New("file exceeds maximum size of 50MB") - ErrEmptyFile = errors.New("empty file not allowed") - ErrFileMissing = errors.New("attachment file missing from disk") + ErrNotFound = errors.New("attachment not found") + ErrFileTooLarge = errors.New("file exceeds maximum size of 50MB") + ErrEmptyFile = errors.New("empty file not allowed") + ErrFileMissing = errors.New("attachment file missing from disk") + ErrUnsupportedType = errors.New("unsupported file type: only images (jpg, png, gif, webp, svg), PDFs, and text files are allowed") ) // Attachment represents the metadata for a stored file. diff --git a/internal/attachments/service.go b/internal/attachments/service.go index 6866f86..1d64773 100644 --- a/internal/attachments/service.go +++ b/internal/attachments/service.go @@ -60,6 +60,11 @@ func (s *Service) Upload(ctx context.Context, req UploadRequest) (*UploadResult, mimeType = DetectMIMEType(sniffBuf, req.Filename) } + // Validate file type against allowlist. + if !IsAllowedType(mimeType) { + return nil, ErrUnsupportedType + } + // Assign default filename if missing. filename := req.Filename if filename == "" { diff --git a/internal/attachments/service_test.go b/internal/attachments/service_test.go index 5092dcf..b5dd1df 100644 --- a/internal/attachments/service_test.go +++ b/internal/attachments/service_test.go @@ -202,6 +202,80 @@ func TestService_Dedup(t *testing.T) { } } +func TestService_Upload_FileTypeValidation(t *testing.T) { + // PNG magic bytes. + pngContent := []byte{0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0x00} + // PDF magic bytes. + pdfContent := []byte("%PDF-1.4 some pdf content here") + // Plain text content. + textContent := []byte("just some plain text content") + + tests := []struct { + name string + content []byte + filename string + mimeType string + wantErr error + }{ + { + name: "valid image upload", + content: pngContent, + filename: "photo.png", + wantErr: nil, + }, + { + name: "valid PDF upload", + content: pdfContent, + filename: "report.pdf", + wantErr: nil, + }, + { + name: "valid text file upload", + content: textContent, + filename: "notes.txt", + wantErr: nil, + }, + { + name: "invalid type zip rejected", + content: []byte("not real zip content"), + filename: "archive.zip", + mimeType: "application/zip", + wantErr: ErrUnsupportedType, + }, + { + name: "invalid type executable rejected", + content: []byte{0x7f, 0x45, 0x4c, 0x46}, + filename: "program.exe", + mimeType: "application/x-executable", + wantErr: ErrUnsupportedType, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(tt.content), + Filename: tt.filename, + MIMEType: tt.mimeType, + UploadedBy: "agent-a", + }) + + if tt.wantErr != nil { + if err != tt.wantErr { + t.Errorf("expected error %v, got %v", tt.wantErr, err) + } + return + } + if err != nil { + t.Fatalf("Upload: %v", err) + } + }) + } +} + func TestService_GarbageCollect(t *testing.T) { svc, db := newTestService(t) ctx := context.Background() diff --git a/internal/channels/service.go b/internal/channels/service.go index 893ff64..0804fc6 100644 --- a/internal/channels/service.go +++ b/internal/channels/service.go @@ -465,7 +465,7 @@ func (s *Service) UpdateChannel(ctx context.Context, channelID int64, req Update // If the message body contains @mentions, mentioned members receive a // "mention":true flag in their inbox notification metadata, and the channel // message metadata includes "mentioned_agents". -func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAgent, body string, priority int, metadata string, replyTo *int64) ([]*messaging.Message, error) { +func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAgent, body string, priority int, metadata string, replyTo *int64, attachments []string) ([]*messaging.Message, error) { ch, err := s.store.GetChannel(ctx, channelID) if err != nil { return nil, err @@ -522,11 +522,12 @@ func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAge channelMetaBytes, _ := json.Marshal(channelMetaObj) channelMsg, err := s.msgService.SendMessage(ctx, fromAgent, "", body, messaging.SendOptions{ - Subject: fmt.Sprintf("channel:%s", ch.Name), - Priority: priority, - Metadata: string(channelMetaBytes), - ChannelID: &channelID, - ReplyTo: replyTo, + Subject: fmt.Sprintf("channel:%s", ch.Name), + Priority: priority, + Metadata: string(channelMetaBytes), + ChannelID: &channelID, + ReplyTo: replyTo, + Attachments: attachments, }) if err != nil { return nil, fmt.Errorf("create channel message: %w", err) diff --git a/internal/channels/service_test.go b/internal/channels/service_test.go index 24e70de..65d0243 100644 --- a/internal/channels/service_test.go +++ b/internal/channels/service_test.go @@ -533,7 +533,7 @@ func TestService_BroadcastMessage(t *testing.T) { ch, _ := svc.CreateChannel(ctx, CreateChannelRequest{Name: "alerts", Type: TypeStandard, CreatedBy: "agent-a"}) t.Run("broadcast creates channel message", func(t *testing.T) { - msgs, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hello", 5, "", nil) + msgs, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hello", 5, "", nil, nil) if err != nil { t.Fatalf("BroadcastMessage: %v", err) } @@ -568,7 +568,7 @@ func TestService_BroadcastMessage(t *testing.T) { t.Run("broadcast without mentions sends no DMs", func(t *testing.T) { svc.JoinChannel(ctx, ch.ID, "agent-b") svc.JoinChannel(ctx, ch.ID, "agent-c") - _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no-dm-test", 5, "", nil) + _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no-dm-test", 5, "", nil, nil) if err != nil { t.Fatalf("BroadcastMessage: %v", err) } @@ -583,7 +583,7 @@ func TestService_BroadcastMessage(t *testing.T) { }) t.Run("sender does not receive own message", func(t *testing.T) { - svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no self-message", 5, "", nil) + svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no self-message", 5, "", nil, nil) inboxResult, _ := svc.msgService.ReadInbox(ctx, "agent-a", messaging.ReadOptions{IncludeRead: true}) for _, m := range inboxResult.Messages { if m.Body == "no self-message" { @@ -594,7 +594,7 @@ func TestService_BroadcastMessage(t *testing.T) { t.Run("non-member auto-joins public channel on broadcast", func(t *testing.T) { seedAgent(t, svc.store.(*SQLiteChannelStore).db, "outsider") - _, err := svc.BroadcastMessage(ctx, ch.ID, "outsider", "auto-joined", 5, "", nil) + _, err := svc.BroadcastMessage(ctx, ch.ID, "outsider", "auto-joined", 5, "", nil, nil) if err != nil { t.Fatalf("expected auto-join for public channel, got %v", err) } @@ -605,10 +605,10 @@ func TestService_BroadcastMessage(t *testing.T) { }) t.Run("broadcast with reply_to", func(t *testing.T) { - msgs, _ := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "original", 5, "", nil) + msgs, _ := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "original", 5, "", nil, nil) original := msgs[0] - replies, err := svc.BroadcastMessage(ctx, ch.ID, "agent-b", "reply to original", 5, "", &original.ID) + replies, err := svc.BroadcastMessage(ctx, ch.ID, "agent-b", "reply to original", 5, "", &original.ID, nil) if err != nil { t.Fatalf("BroadcastMessage with reply_to: %v", err) } @@ -625,7 +625,7 @@ func TestService_BroadcastMessage(t *testing.T) { t.Fatalf("create private channel: %v", err) } seedAgent(t, svc.store.(*SQLiteChannelStore).db, "outsider2") - _, err = svc.BroadcastMessage(ctx, privCh.ID, "outsider2", "unauthorized", 5, "", nil) + _, err = svc.BroadcastMessage(ctx, privCh.ID, "outsider2", "unauthorized", 5, "", nil, nil) if !errors.Is(err, ErrNotChannelMember) { t.Errorf("expected ErrNotChannelMember for private channel, got %v", err) } @@ -644,7 +644,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) { svc.JoinChannel(ctx, ch.ID, "agent-c") t.Run("mentioned member gets mention flag in inbox", func(t *testing.T) { - _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hey @agent-b check this", 5, "", nil) + _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hey @agent-b check this", 5, "", nil, nil) if err != nil { t.Fatalf("BroadcastMessage: %v", err) } @@ -677,7 +677,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) { }) t.Run("channel message metadata includes mentioned_agents", func(t *testing.T) { - _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "cc @agent-b and @agent-c", 5, "", nil) + _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "cc @agent-b and @agent-c", 5, "", nil, nil) if err != nil { t.Fatalf("BroadcastMessage: %v", err) } @@ -707,7 +707,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) { }) t.Run("self-mention is excluded", func(t *testing.T) { - _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "I am @agent-a and cc @agent-b", 5, "", nil) + _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "I am @agent-a and cc @agent-b", 5, "", nil, nil) if err != nil { t.Fatalf("BroadcastMessage: %v", err) } @@ -732,7 +732,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) { }) t.Run("no mentions produces no mention metadata", func(t *testing.T) { - _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "just a normal message", 5, "", nil) + _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "just a normal message", 5, "", nil, nil) if err != nil { t.Fatalf("BroadcastMessage: %v", err) } @@ -752,7 +752,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) { t.Run("non-member mention is ignored", func(t *testing.T) { seedAgent(t, svc.store.(*SQLiteChannelStore).db, "outsider") - _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hey @outsider and @agent-b", 5, "", nil) + _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hey @outsider and @agent-b", 5, "", nil, nil) if err != nil { t.Fatalf("BroadcastMessage: %v", err) } diff --git a/internal/mcp/bridge.go b/internal/mcp/bridge.go index 650ae64..6a8580c 100644 --- a/internal/mcp/bridge.go +++ b/internal/mcp/bridge.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "io" + "strings" "time" "github.com/synapbus/synapbus/internal/agents" @@ -132,12 +133,34 @@ func (b *ServiceBridge) callSendMessage(ctx context.Context, args map[string]any replyTo = &v } + var attachmentHashes []string + if attVal, ok := args["attachments"]; ok { + switch v := attVal.(type) { + case string: + for _, h := range strings.Split(v, ",") { + h = strings.TrimSpace(h) + if h != "" { + attachmentHashes = append(attachmentHashes, h) + } + } + case []any: + for _, item := range v { + if s, ok := item.(string); ok && s != "" { + attachmentHashes = append(attachmentHashes, s) + } + } + case []string: + attachmentHashes = v + } + } + opts := messaging.SendOptions{ - Subject: getString(args, "subject", ""), - Priority: getInt(args, "priority", 5), - Metadata: getString(args, "metadata", ""), - ChannelID: channelID, - ReplyTo: replyTo, + Subject: getString(args, "subject", ""), + Priority: getInt(args, "priority", 5), + Metadata: getString(args, "metadata", ""), + ChannelID: channelID, + ReplyTo: replyTo, + Attachments: attachmentHashes, } msg, err := b.msgService.SendMessage(ctx, b.agentName, to, body, opts) @@ -554,7 +577,28 @@ func (b *ServiceBridge) callSendChannelMessage(ctx context.Context, args map[str } } - messages, err := b.channelService.BroadcastMessage(ctx, channelID, b.agentName, body, priority, metadata, replyTo) + var attachmentHashes []string + if attVal, ok := args["attachments"]; ok { + switch v := attVal.(type) { + case string: + for _, h := range strings.Split(v, ",") { + h = strings.TrimSpace(h) + if h != "" { + attachmentHashes = append(attachmentHashes, h) + } + } + case []any: + for _, item := range v { + if s, ok := item.(string); ok && s != "" { + attachmentHashes = append(attachmentHashes, s) + } + } + case []string: + attachmentHashes = v + } + } + + messages, err := b.channelService.BroadcastMessage(ctx, channelID, b.agentName, body, priority, metadata, replyTo, attachmentHashes) if err != nil { return nil, err } diff --git a/internal/mcp/tools_hybrid.go b/internal/mcp/tools_hybrid.go index 5464933..8728195 100644 --- a/internal/mcp/tools_hybrid.go +++ b/internal/mcp/tools_hybrid.go @@ -84,14 +84,15 @@ func (h *HybridToolRegistrar) myStatusTool() mcplib.Tool { func (h *HybridToolRegistrar) sendMessageTool() mcplib.Tool { return mcplib.NewTool("send_message", - mcplib.WithDescription("Send a message to another agent (DM) or to a channel. Specify exactly one of 'to' (agent name for DM) or 'channel' (channel name or numeric ID)."), + mcplib.WithDescription("Send a message to another agent (DM) or to a channel. Supports attachments — upload files first via the execute tool, then pass the returned hashes here. Specify exactly one of 'to' (agent name for DM) or 'channel' (channel name or numeric ID)."), mcplib.WithString("to", mcplib.Description("Recipient agent name for direct messages")), mcplib.WithString("channel", mcplib.Description("Channel name or numeric ID for channel messages")), mcplib.WithString("body", mcplib.Description("Message body text"), mcplib.Required()), mcplib.WithString("subject", mcplib.Description("Conversation subject (optional)")), mcplib.WithNumber("priority", mcplib.Description("Message priority (1-10, default 5)"), mcplib.Min(1), mcplib.Max(10)), mcplib.WithString("metadata", mcplib.Description("JSON metadata object (optional)")), - mcplib.WithNumber("reply_to", mcplib.Description("ID of the message to reply to (optional, for threading)")), + mcplib.WithNumber("reply_to", mcplib.Description("ID of the parent message to reply to. Creates a threaded reply. Always use reply_to when responding to a message that is itself a thread reply, to keep conversations organized.")), + mcplib.WithString("attachments", mcplib.Description("Comma-separated list of attachment hashes to link to this message. Upload attachments first using the upload_attachment action via the execute tool.")), ) } @@ -323,6 +324,16 @@ func (h *HybridToolRegistrar) handleSendMessage(ctx context.Context, req mcplib. replyTo = &v } + var attachmentHashes []string + if attStr := req.GetString("attachments", ""); attStr != "" { + for _, h := range strings.Split(attStr, ",") { + h = strings.TrimSpace(h) + if h != "" { + attachmentHashes = append(attachmentHashes, h) + } + } + } + // Channel message path. if channel != "" { if h.channelService == nil { @@ -335,7 +346,7 @@ func (h *HybridToolRegistrar) handleSendMessage(ctx context.Context, req mcplib. return mcplib.NewToolResultError(fmt.Sprintf("send_message to channel failed: %s", err)), nil } - messages, err := h.channelService.BroadcastMessage(ctx, channelID, agentName, body, priority, metadataStr, replyTo) + messages, err := h.channelService.BroadcastMessage(ctx, channelID, agentName, body, priority, metadataStr, replyTo, attachmentHashes) if err != nil { return mcplib.NewToolResultError(fmt.Sprintf("send_message to channel failed: %s", err)), nil } @@ -345,19 +356,30 @@ func (h *HybridToolRegistrar) handleSendMessage(ctx context.Context, req mcplib. messageID = messages[0].ID } - return resultJSON(map[string]any{ + result := map[string]any{ "channel_id": channelID, "message_id": messageID, "status": "sent", - }) + } + + // Enrich channel messages with attachment info. + if len(messages) > 0 && len(attachmentHashes) > 0 { + h.msgService.EnrichMessages(ctx, messages) + if len(messages[0].Attachments) > 0 { + result["attachments"] = messages[0].Attachments + } + } + + return resultJSON(result) } // DM path. opts := messaging.SendOptions{ - Subject: subject, - Priority: priority, - Metadata: metadataStr, - ReplyTo: replyTo, + Subject: subject, + Priority: priority, + Metadata: metadataStr, + ReplyTo: replyTo, + Attachments: attachmentHashes, } msg, err := h.msgService.SendMessage(ctx, agentName, to, body, opts) @@ -365,11 +387,21 @@ func (h *HybridToolRegistrar) handleSendMessage(ctx context.Context, req mcplib. return mcplib.NewToolResultError(fmt.Sprintf("send_message failed: %s", err)), nil } - return resultJSON(map[string]any{ + result := map[string]any{ "message_id": msg.ID, "conversation_id": msg.ConversationID, "status": msg.Status, - }) + } + + // Enrich message with attachment info. + if len(attachmentHashes) > 0 { + h.msgService.EnrichMessages(ctx, []*messaging.Message{msg}) + if len(msg.Attachments) > 0 { + result["attachments"] = msg.Attachments + } + } + + return resultJSON(result) } func (h *HybridToolRegistrar) handleSearch(ctx context.Context, req mcplib.CallToolRequest) (*mcplib.CallToolResult, error) { diff --git a/internal/messaging/options.go b/internal/messaging/options.go index 2ec494b..07cacd3 100644 --- a/internal/messaging/options.go +++ b/internal/messaging/options.go @@ -2,12 +2,13 @@ package messaging // SendOptions configures message sending behavior. type SendOptions struct { - Subject string `json:"subject,omitempty"` - Priority int `json:"priority,omitempty"` - Metadata string `json:"metadata,omitempty"` - ChannelID *int64 `json:"channel_id,omitempty"` - ConversationID *int64 `json:"conversation_id,omitempty"` - ReplyTo *int64 `json:"reply_to,omitempty"` + Subject string `json:"subject,omitempty"` + Priority int `json:"priority,omitempty"` + Metadata string `json:"metadata,omitempty"` + ChannelID *int64 `json:"channel_id,omitempty"` + ConversationID *int64 `json:"conversation_id,omitempty"` + ReplyTo *int64 `json:"reply_to,omitempty"` + Attachments []string `json:"attachments,omitempty"` // attachment hashes to link } // ReadOptions configures inbox reading behavior. diff --git a/internal/messaging/service.go b/internal/messaging/service.go index 65f8cf2..7d4c858 100644 --- a/internal/messaging/service.go +++ b/internal/messaging/service.go @@ -24,12 +24,20 @@ type MessageListener interface { OnMessageSent(ctx context.Context, msg *Message) } +// AttachmentLinker links attachment hashes to message IDs. This avoids +// importing the attachments package directly. Set via SetAttachmentLinker. +type AttachmentLinker interface { + AttachToMessage(ctx context.Context, hash string, messageID int64) error + GetByMessageID(ctx context.Context, messageID int64) ([]AttachmentInfo, error) +} + // MessagingService provides business logic for messaging operations. type MessagingService struct { store MessageStore tracer *trace.Tracer dispatcher dispatcher.EventDispatcher embeddings EmbeddingNotifier + attLinker AttachmentLinker listeners []MessageListener logger *slog.Logger } @@ -53,6 +61,11 @@ func (s *MessagingService) SetEmbeddingNotifier(n EmbeddingNotifier) { s.embeddings = n } +// SetAttachmentLinker sets the attachment linker for message-attachment binding. +func (s *MessagingService) SetAttachmentLinker(l AttachmentLinker) { + s.attLinker = l +} + // AddMessageListener registers a listener that is notified after message creation. func (s *MessagingService) AddMessageListener(l MessageListener) { s.listeners = append(s.listeners, l) @@ -145,6 +158,19 @@ func (s *MessagingService) SendMessage(ctx context.Context, from, to, body strin return nil, fmt.Errorf("insert message: %w", err) } + // Link attachments if provided. + if s.attLinker != nil && len(opts.Attachments) > 0 { + for _, hash := range opts.Attachments { + if err := s.attLinker.AttachToMessage(ctx, hash, msg.ID); err != nil { + s.logger.Error("failed to link attachment", + "hash", hash, + "message_id", msg.ID, + "error", err, + ) + } + } + } + // Enqueue for embedding (async, best-effort) if s.embeddings != nil { s.embeddings.OnMessageCreated(ctx, msg.ID, msg.Body) @@ -523,6 +549,44 @@ func (s *MessagingService) GetConversationIDsForDM(ctx context.Context, agentNam return s.store.GetConversationIDsForDM(ctx, agentNames, peerAgent, lastMessageID) } +// EnrichMessages populates ReplyCount and Attachments on a slice of messages. +func (s *MessagingService) EnrichMessages(ctx context.Context, msgs []*Message) { + if len(msgs) == 0 { + return + } + + ids := make([]int64, len(msgs)) + for i, m := range msgs { + ids[i] = m.ID + } + + // Batch-load reply counts. + counts, err := s.store.GetReplyCounts(ctx, ids) + if err != nil { + s.logger.Error("failed to load reply counts", "error", err) + } else { + for _, m := range msgs { + if c, ok := counts[m.ID]; ok { + m.ReplyCount = c + } + } + } + + // Batch-load attachments. + if s.attLinker != nil { + for _, m := range msgs { + atts, err := s.attLinker.GetByMessageID(ctx, m.ID) + if err != nil { + s.logger.Error("failed to load attachments", "message_id", m.ID, "error", err) + continue + } + if len(atts) > 0 { + m.Attachments = atts + } + } + } +} + // GetConversation returns a conversation and its messages. func (s *MessagingService) GetConversation(ctx context.Context, id int64) (*Conversation, []*Message, error) { conv, err := s.store.GetConversation(ctx, id) diff --git a/internal/messaging/service_test.go b/internal/messaging/service_test.go index cb49600..fa801a2 100644 --- a/internal/messaging/service_test.go +++ b/internal/messaging/service_test.go @@ -728,5 +728,96 @@ func TestMessagingService_ReadInbox_DateFiltering(t *testing.T) { }) } +// mockAttachmentLinker is a test double for the AttachmentLinker interface. +type mockAttachmentLinker struct { + attachments map[int64][]AttachmentInfo +} + +func (m *mockAttachmentLinker) AttachToMessage(_ context.Context, _ string, _ int64) error { + return nil +} + +func (m *mockAttachmentLinker) GetByMessageID(_ context.Context, messageID int64) ([]AttachmentInfo, error) { + return m.attachments[messageID], nil +} + +func TestMessagingService_EnrichMessages(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + // Send a parent message and replies to it. + parent, err := svc.SendMessage(ctx, "sender", "receiver", "parent message", SendOptions{Subject: "Enrich Test"}) + if err != nil { + t.Fatalf("SendMessage (parent): %v", err) + } + + replyTo := parent.ID + for i := 0; i < 3; i++ { + _, err := svc.SendMessage(ctx, "receiver", "sender", "reply", SendOptions{ + Subject: "Enrich Test", + ReplyTo: &replyTo, + }) + if err != nil { + t.Fatalf("SendMessage (reply %d): %v", i, err) + } + } + + // Send a message with no replies. + noReplies, err := svc.SendMessage(ctx, "sender", "receiver", "standalone", SendOptions{Subject: "Enrich Standalone"}) + if err != nil { + t.Fatalf("SendMessage (standalone): %v", err) + } + + t.Run("reply counts populated", func(t *testing.T) { + msgs := []*Message{parent, noReplies} + svc.EnrichMessages(ctx, msgs) + + if parent.ReplyCount != 3 { + t.Errorf("parent ReplyCount = %d, want 3", parent.ReplyCount) + } + if noReplies.ReplyCount != 0 { + t.Errorf("noReplies ReplyCount = %d, want 0", noReplies.ReplyCount) + } + }) + + t.Run("attachments populated when linker set", func(t *testing.T) { + linker := &mockAttachmentLinker{ + attachments: map[int64][]AttachmentInfo{ + parent.ID: { + {Hash: "abc123", OriginalFilename: "photo.png", Size: 1024, MIMEType: "image/png", IsImage: true}, + }, + }, + } + svc.SetAttachmentLinker(linker) + + // Reset enrichment state. + parent.ReplyCount = 0 + parent.Attachments = nil + noReplies.ReplyCount = 0 + noReplies.Attachments = nil + + msgs := []*Message{parent, noReplies} + svc.EnrichMessages(ctx, msgs) + + if parent.ReplyCount != 3 { + t.Errorf("parent ReplyCount = %d, want 3", parent.ReplyCount) + } + if len(parent.Attachments) != 1 { + t.Fatalf("parent Attachments count = %d, want 1", len(parent.Attachments)) + } + if parent.Attachments[0].Hash != "abc123" { + t.Errorf("attachment hash = %s, want abc123", parent.Attachments[0].Hash) + } + if noReplies.Attachments != nil { + t.Errorf("noReplies Attachments should be nil, got %v", noReplies.Attachments) + } + }) + + t.Run("empty slice is a no-op", func(t *testing.T) { + svc.EnrichMessages(ctx, []*Message{}) + // Should not panic or error. + }) +} + // suppress unused import warning for storage package var _ = storage.RunMigrations diff --git a/internal/messaging/store.go b/internal/messaging/store.go index 3bdb0ed..48a4c21 100644 --- a/internal/messaging/store.go +++ b/internal/messaging/store.go @@ -39,6 +39,7 @@ type MessageStore interface { GetLastReadForDM(ctx context.Context, agentNames []string, peerAgent string) (int64, error) GetConversationIDsForChannel(ctx context.Context, channelID int64, lastMessageID int64) ([]int64, error) GetConversationIDsForDM(ctx context.Context, agentNames []string, peerAgent string, lastMessageID int64) ([]int64, error) + GetReplyCounts(ctx context.Context, messageIDs []int64) (map[int64]int, error) } // SQLiteMessageStore implements MessageStore using SQLite. @@ -1007,6 +1008,44 @@ func (s *SQLiteMessageStore) GetConversationIDsForDM(ctx context.Context, agentN return ids, rows.Err() } +// GetReplyCounts returns a map of message ID → reply count for the given IDs. +func (s *SQLiteMessageStore) GetReplyCounts(ctx context.Context, messageIDs []int64) (map[int64]int, error) { + if len(messageIDs) == 0 { + return map[int64]int{}, nil + } + + placeholders := make([]string, len(messageIDs)) + args := make([]any, len(messageIDs)) + for i, id := range messageIDs { + placeholders[i] = "?" + args[i] = id + } + + query := fmt.Sprintf( + `SELECT reply_to, COUNT(*) FROM messages + WHERE reply_to IN (%s) + GROUP BY reply_to`, + strings.Join(placeholders, ","), + ) + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("get reply counts: %w", err) + } + defer rows.Close() + + counts := make(map[int64]int) + for rows.Next() { + var replyTo int64 + var count int + if err := rows.Scan(&replyTo, &count); err != nil { + return nil, fmt.Errorf("scan reply count: %w", err) + } + counts[replyTo] = count + } + return counts, rows.Err() +} + // scanMessage scans a single message from sql.Row. func scanMessage(row *sql.Row) (*Message, error) { var msg Message diff --git a/internal/messaging/store_test.go b/internal/messaging/store_test.go index 1a29d3a..b9d481e 100644 --- a/internal/messaging/store_test.go +++ b/internal/messaging/store_test.go @@ -1027,6 +1027,132 @@ func TestSQLiteMessageStore_GetChannelMessages_Offset(t *testing.T) { }) } +func TestSQLiteMessageStore_GetReplyCounts(t *testing.T) { + db := newTestDB(t) + store := NewSQLiteMessageStore(db) + ctx := context.Background() + + seedAgent(t, db, "sender") + seedAgent(t, db, "replier") + + conv := &Conversation{Subject: "reply counts", CreatedBy: "sender"} + if err := store.InsertConversation(ctx, conv); err != nil { + t.Fatalf("InsertConversation: %v", err) + } + + // Insert parent message + parent := &Message{ + ConversationID: conv.ID, + FromAgent: "sender", + ToAgent: "replier", + Body: "parent message", + Priority: 5, + Status: StatusPending, + } + if err := store.InsertMessage(ctx, parent); err != nil { + t.Fatalf("InsertMessage (parent): %v", err) + } + + // Insert 3 replies to parent + for i := 0; i < 3; i++ { + reply := &Message{ + ConversationID: conv.ID, + FromAgent: "replier", + ToAgent: "sender", + ReplyTo: &parent.ID, + Body: fmt.Sprintf("reply %d", i), + Priority: 5, + Status: StatusPending, + } + if err := store.InsertMessage(ctx, reply); err != nil { + t.Fatalf("InsertMessage (reply %d): %v", i, err) + } + } + + t.Run("parent with 3 replies", func(t *testing.T) { + counts, err := store.GetReplyCounts(ctx, []int64{parent.ID}) + if err != nil { + t.Fatalf("GetReplyCounts: %v", err) + } + if counts[parent.ID] != 3 { + t.Errorf("reply count for parent = %d, want 3", counts[parent.ID]) + } + }) + + t.Run("message with no replies returns 0", func(t *testing.T) { + // Insert a message with no replies + noReply := &Message{ + ConversationID: conv.ID, + FromAgent: "sender", + ToAgent: "replier", + Body: "no replies here", + Priority: 5, + Status: StatusPending, + } + if err := store.InsertMessage(ctx, noReply); err != nil { + t.Fatalf("InsertMessage (noReply): %v", err) + } + + counts, err := store.GetReplyCounts(ctx, []int64{noReply.ID}) + if err != nil { + t.Fatalf("GetReplyCounts: %v", err) + } + if counts[noReply.ID] != 0 { + t.Errorf("reply count for noReply = %d, want 0", counts[noReply.ID]) + } + }) + + t.Run("multiple parents", func(t *testing.T) { + // Insert a second parent with 2 replies + parent2 := &Message{ + ConversationID: conv.ID, + FromAgent: "sender", + ToAgent: "replier", + Body: "second parent", + Priority: 5, + Status: StatusPending, + } + if err := store.InsertMessage(ctx, parent2); err != nil { + t.Fatalf("InsertMessage (parent2): %v", err) + } + for i := 0; i < 2; i++ { + reply := &Message{ + ConversationID: conv.ID, + FromAgent: "replier", + ToAgent: "sender", + ReplyTo: &parent2.ID, + Body: fmt.Sprintf("reply to parent2 %d", i), + Priority: 5, + Status: StatusPending, + } + if err := store.InsertMessage(ctx, reply); err != nil { + t.Fatalf("InsertMessage (parent2 reply %d): %v", i, err) + } + } + + counts, err := store.GetReplyCounts(ctx, []int64{parent.ID, parent2.ID}) + if err != nil { + t.Fatalf("GetReplyCounts: %v", err) + } + if counts[parent.ID] != 3 { + t.Errorf("reply count for parent = %d, want 3", counts[parent.ID]) + } + if counts[parent2.ID] != 2 { + t.Errorf("reply count for parent2 = %d, want 2", counts[parent2.ID]) + } + }) + + t.Run("empty slice returns empty map", func(t *testing.T) { + counts, err := store.GetReplyCounts(ctx, []int64{}) + if err != nil { + t.Fatalf("GetReplyCounts: %v", err) + } + if len(counts) != 0 { + t.Errorf("expected empty map, got %v", counts) + } + }) +} + func TestSQLiteMessageStore_CombinedFiltersAndPagination(t *testing.T) { db := newTestDB(t) store := NewSQLiteMessageStore(db) diff --git a/internal/messaging/types.go b/internal/messaging/types.go index f7b6461..f59fe80 100644 --- a/internal/messaging/types.go +++ b/internal/messaging/types.go @@ -14,6 +14,16 @@ const ( StatusFailed = "failed" ) +// AttachmentInfo is a lightweight attachment summary included in message responses. +// It avoids importing the attachments package into the messaging package. +type AttachmentInfo struct { + Hash string `json:"hash"` + OriginalFilename string `json:"original_filename"` + Size int64 `json:"size"` + MIMEType string `json:"mime_type"` + IsImage bool `json:"is_image"` +} + // Message represents a single message in the system. type Message struct { ID int64 `json:"id"` @@ -30,6 +40,8 @@ type Message struct { ClaimedAt *time.Time `json:"claimed_at,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` + ReplyCount int `json:"reply_count"` + Attachments []AttachmentInfo `json:"attachments,omitempty"` } // Conversation groups related messages into a thread. diff --git a/internal/web/dist/index.html b/internal/web/dist/index.html index f8207d5..f4c9dcf 100644 --- a/internal/web/dist/index.html +++ b/internal/web/dist/index.html @@ -11,30 +11,30 @@ - - + + - - - - - - + + + + + +
+ + + +{#if attachment.is_image} + + + + + {#if showOverlay} + + {/if} +{:else} + +
+ + + + +
+ + {attachment.original_filename} + + {formatSize(attachment.size)} +
+
+{/if} diff --git a/web/src/lib/components/ComposeForm.svelte b/web/src/lib/components/ComposeForm.svelte index 4534554..45b3671 100644 --- a/web/src/lib/components/ComposeForm.svelte +++ b/web/src/lib/components/ComposeForm.svelte @@ -1,5 +1,5 @@