diff --git a/CLAUDE.md b/CLAUDE.md index 57ecad7..74210bc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -105,6 +105,7 @@ make lint # Run linters - 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) +- SQLite (modernc.org/sqlite, pure Go) — new migration 013_reactions.sql (010-reactions-workflows) ## 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 2b941b0..d699991 100644 --- a/autonomous_summary.md +++ b/autonomous_summary.md @@ -1,91 +1,101 @@ -# Autonomous Implementation Summary: Attachments & Threads Enhancement +# Autonomous Implementation Summary: Message Reactions & Workflow States -**Branch**: `009-attachments-threads` -**Date**: 2026-03-17 -**Status**: Complete +**Branch**: `010-reactions-workflows` +**Date**: 2026-03-18 +**Status**: Complete (StalemateWorker extension deferred) ## What Was Built -### Attachments (Web UI + MCP + Admin CLI) +### Message Reactions +- **Toggle semantics**: Add a reaction → added. Add same reaction again → removed. One per type per agent per message. +- **5 reaction types**: approve, reject, in_progress, done, published +- **Metadata support**: JSON metadata on reactions (e.g., `{"url": "https://..."}` for published) +- **100-reaction limit** per message (safety) -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. +### Workflow State Derivation +- State computed from reactions: published > done > rejected > in_progress > approved > proposed +- No denormalization — state derived on read from reaction list +- Channel messages with no reactions → "proposed" state +- Terminal states (rejected, done, published) don't trigger stalemate checks -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. +### Channel Workflow Settings +- `auto_approve` — skip proposed state for new messages +- `stalemate_remind_after` — duration before reminder DM (default 24h) +- `stalemate_escalate_after` — duration before escalation to #approvals (default 72h) -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. +### REST API +- `POST /api/messages/{id}/reactions` — toggle reaction (add or remove) +- `GET /api/messages/{id}/reactions` — get reactions + workflow state +- `DELETE /api/messages/{id}/reactions/{reaction}` — remove reaction +- `PUT /api/channels/{name}/settings` — update workflow settings +- `GET /api/channels/{name}/messages/by-state?state=X` — list messages by state -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. +### MCP Tools (via execute bridge) +- `react` — add/toggle reaction on a message +- `unreact` — remove a reaction +- `get_reactions` — query reactions and workflow state +- `list_by_state` — list messages by workflow state in a channel -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). +### Web UI +- **WorkflowBadge** component: colored pills (yellow/green/blue/red/gray/cyan) per state +- **ReactionPills** component: grouped reaction pills with count, agent names on hover, click-to-toggle +- Published reactions with URL show clickable link icon +- Integrated into channel message view -### Threads +### Admin CLI +- `synapbus channels update --name X --auto-approve=true --stalemate-remind-after=12h --stalemate-escalate-after=48h` -1. **Thread visibility**: Messages with replies show an always-visible "N replies" badge with a chat bubble icon. Clicking opens the thread panel. +## Files Created/Modified -2. **Reply count in API**: All message API responses now include `reply_count` (batch-loaded via efficient GROUP BY query). +### New Files +| File | Description | +|------|-------------| +| `internal/storage/schema/013_reactions.sql` | Migration: message_reactions table + channel columns | +| `internal/reactions/model.go` | Reaction types, state derivation, constants | +| `internal/reactions/store.go` | SQLite CRUD for reactions | +| `internal/reactions/service.go` | Business logic: toggle, remove, get, list by state | +| `internal/reactions/model_test.go` | 23 test cases for model functions | +| `internal/reactions/store_test.go` | 6 test functions for store operations | +| `internal/api/reactions_handler.go` | REST API handlers for reactions | +| `web/src/lib/components/WorkflowBadge.svelte` | Colored state badge component | +| `web/src/lib/components/ReactionPills.svelte` | Reaction toggle pills component | -3. **MCP threading**: `reply_to` parameter description updated to clearly guide agents on threading behavior. Thread context visible in MCP responses. - -4. **Attachment display in threads**: Thread panel also renders attachment thumbnails/file icons. - -## Files Modified - -### Backend (Go) +### Modified Files | 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 | +| `internal/messaging/types.go` | Added WorkflowState, Reactions, ReactionInfo to Message | +| `internal/messaging/service.go` | Added ReactionEnricher interface, enrichment in EnrichMessages | +| `internal/channels/types.go` | Added AutoApprove, StalemateRemindAfter, StalemateEscalateAfter, ChannelSettings | +| `internal/channels/store.go` | Updated SELECT queries for new columns, added UpdateChannelSettings | +| `internal/channels/service.go` | Added UpdateChannelSettings method | +| `internal/api/router.go` | Registered reaction and channel settings routes | +| `internal/api/channels_handler.go` | Added UpdateSettings, ListByState handlers | +| `internal/mcp/bridge.go` | Added react/unreact/get_reactions/list_by_state bridge methods | +| `internal/mcp/tools_hybrid.go` | Added reactionService to registrar | +| `internal/mcp/server.go` | Added reactionService parameter | +| `internal/actions/registry.go` | Registered 4 new reaction actions | +| `cmd/synapbus/main.go` | Wired reaction service, adapter, passed to router+MCP | +| `cmd/synapbus/admin.go` | Added channels update CLI command | +| `internal/admin/socket.go` | Added channels.update_settings handler | +| `web/src/lib/api/client.ts` | Added reactions.toggle/get methods | +| `web/src/routes/channels/[name]/+page.svelte` | Integrated WorkflowBadge + ReactionPills | ## Test Results -- **Go tests**: 25 packages, all pass, 0 failures -- **Integration tests**: 9 E2E tests, all pass -- **New tests**: 24 test cases added, all pass +- **25 Go test packages**: all pass, 0 failures +- **New tests**: 29+ test cases (model: 23, store: 6) +- **Integration tests**: 9 E2E tests pass - **Web build**: Svelte SPA builds successfully -- **Binary build**: 90MB arm64 binary compiles cleanly +- **Binary build**: Compiles cleanly + +## Deferred + +- **StalemateWorker extension** (T023-T025): The data model, channel settings, and query infrastructure are in place. The worker just needs a scan loop added to detect stale messages and send DMs/escalations. This is a straightforward follow-up task. ## Architecture Decisions -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 +1. **Separate reactions package**: Clean domain separation from messaging +2. **Toggle semantics**: INSERT if absent, DELETE if present — simple, atomic, idempotent +3. **Derived workflow state**: No denormalization; state computed from reactions on read +4. **Bridge actions (not hybrid tools)**: Consistent with attachments pattern — 4 hybrid tools are stable surface area +5. **ReactionEnricher adapter**: Avoids circular dependency between reactions and messaging packages diff --git a/cmd/synapbus/admin.go b/cmd/synapbus/admin.go index 1626dda..c82d3bb 100644 --- a/cmd/synapbus/admin.go +++ b/cmd/synapbus/admin.go @@ -636,7 +636,46 @@ func addAdminCommands(rootCmd *cobra.Command) { channelsJoinCmd.MarkFlagRequired("channel") channelsJoinCmd.MarkFlagRequired("agent") - channelsCmd.AddCommand(channelsListCmd, channelsShowCmd, channelsCreateCmd, channelsJoinCmd) + var ( + channelsUpdateName string + channelsUpdateAutoApprove string + channelsUpdateStalemateRemind string + channelsUpdateStalemateEscalate string + ) + channelsUpdateCmd := &cobra.Command{ + Use: "update", + Short: "Update channel settings (auto-approve, stalemate timers)", + RunE: func(cmd *cobra.Command, args []string) error { + if channelsUpdateName == "" { + return fmt.Errorf("--name is required") + } + reqArgs := map[string]interface{}{ + "name": channelsUpdateName, + } + if cmd.Flags().Changed("auto-approve") { + reqArgs["auto_approve"] = channelsUpdateAutoApprove == "true" + } + if cmd.Flags().Changed("stalemate-remind-after") { + reqArgs["stalemate_remind_after"] = channelsUpdateStalemateRemind + } + if cmd.Flags().Changed("stalemate-escalate-after") { + reqArgs["stalemate_escalate_after"] = channelsUpdateStalemateEscalate + } + resp, err := adminRequest("channels.update_settings", reqArgs) + if err != nil { + return err + } + printJSON(resp["data"]) + return nil + }, + } + channelsUpdateCmd.Flags().StringVar(&channelsUpdateName, "name", "", "Channel name") + channelsUpdateCmd.Flags().StringVar(&channelsUpdateAutoApprove, "auto-approve", "", "Auto-approve messages (true|false)") + channelsUpdateCmd.Flags().StringVar(&channelsUpdateStalemateRemind, "stalemate-remind-after", "", "Stalemate reminder duration (e.g. 24h)") + channelsUpdateCmd.Flags().StringVar(&channelsUpdateStalemateEscalate, "stalemate-escalate-after", "", "Stalemate escalation duration (e.g. 72h)") + channelsUpdateCmd.MarkFlagRequired("name") + + channelsCmd.AddCommand(channelsListCmd, channelsShowCmd, channelsCreateCmd, channelsJoinCmd, channelsUpdateCmd) // ----- conversations commands ----- conversationsCmd := &cobra.Command{ diff --git a/cmd/synapbus/main.go b/cmd/synapbus/main.go index 5e90fcf..9635a0b 100644 --- a/cmd/synapbus/main.go +++ b/cmd/synapbus/main.go @@ -41,6 +41,7 @@ import ( mcpserver "github.com/synapbus/synapbus/internal/mcp" "github.com/synapbus/synapbus/internal/messaging" prommetrics "github.com/synapbus/synapbus/internal/metrics" + "github.com/synapbus/synapbus/internal/reactions" "github.com/synapbus/synapbus/internal/search" "github.com/synapbus/synapbus/internal/search/embedding" "github.com/synapbus/synapbus/internal/storage" @@ -284,6 +285,12 @@ func runServe(cmd *cobra.Command, args []string) error { msgService.SetAttachmentLinker(&attachmentLinkerAdapter{svc: attachmentService}) slog.Info("attachment service initialized", "dir", attachmentsDir) + // Create reaction service + reactionStore := reactions.NewSQLiteStore(db.DB) + reactionService := reactions.NewService(reactionStore, slog.Default()) + msgService.SetReactionEnricher(&reactionEnricherAdapter{svc: reactionService}) + slog.Info("reaction service initialized") + // Initialize auth subsystem authSecret := make([]byte, 32) if _, err := rand.Read(authSecret); err != nil { @@ -466,7 +473,7 @@ func runServe(cmd *cobra.Command, args []string) error { actionIndex := actions.NewIndex(actionRegistry.List()) // Create MCP server (4 hybrid tools: my_status, send_message, search, execute) - mcpSrv := mcpserver.NewMCPServer(msgService, agentService, channelService, swarmService, attachmentService, searchService, con, jsPool, actionRegistry, actionIndex, db.DB) + mcpSrv := mcpserver.NewMCPServer(msgService, agentService, channelService, swarmService, attachmentService, searchService, reactionService, con, jsPool, actionRegistry, actionIndex, db.DB) startTime := time.Now() // Start task expiry worker @@ -620,6 +627,7 @@ func runServe(cmd *cobra.Command, args []string) error { ChannelService: channelService, APIKeyService: apiKeyService, DeadLetterStore: deadLetterStore, + ReactionService: reactionService, SSEHub: sseHub, Broadcaster: sseBroadcaster, SessionMiddleware: sessionMiddleware, @@ -864,6 +872,32 @@ func (a *attachmentLinkerAdapter) GetByMessageID(ctx context.Context, messageID return results, nil } +// reactionEnricherAdapter adapts reactions.Service to messaging.ReactionEnricher. +type reactionEnricherAdapter struct { + svc *reactions.Service +} + +func (a *reactionEnricherAdapter) GetByMessageIDs(ctx context.Context, messageIDs []int64) (map[int64][]messaging.ReactionInfo, error) { + rxMap, err := a.svc.GetReactionsByMessageIDs(ctx, messageIDs) + if err != nil { + return nil, err + } + result := make(map[int64][]messaging.ReactionInfo, len(rxMap)) + for msgID, rxs := range rxMap { + infos := make([]messaging.ReactionInfo, len(rxs)) + for i, rx := range rxs { + infos[i] = messaging.ReactionInfo{ + AgentName: rx.AgentName, + Reaction: rx.Reaction, + Metadata: rx.Metadata, + CreatedAt: rx.CreatedAt, + } + } + result[msgID] = infos + } + return result, 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 118597a..b9e04ed 100644 --- a/internal/actions/registry.go +++ b/internal/actions/registry.go @@ -6,10 +6,10 @@ type Registry struct { ordered []Action // maintains insertion order } -// NewRegistry creates a registry pre-populated with all 23 agent-callable actions. +// NewRegistry creates a registry pre-populated with all 27 agent-callable actions. func NewRegistry() *Registry { r := &Registry{ - actions: make(map[string]Action, 23), + actions: make(map[string]Action, 27), } for _, a := range allActions() { r.actions[a.Name] = a @@ -42,7 +42,7 @@ func (r *Registry) ListByCategory(category string) []Action { return out } -// allActions returns the canonical list of all 23 agent-callable actions. +// allActions returns the canonical list of all 27 agent-callable actions. func allActions() []Action { return []Action{ // ── Messaging (7 actions) ────────────────────────────────────── @@ -455,5 +455,75 @@ func allActions() []Action { }, }, }, + + // ── Reactions (4 actions) ──────────────────────────────────── + { + Name: "react", + Category: "reactions", + Description: "Add or toggle a reaction on a message. Valid reactions: approve, reject, in_progress, done, published. Adding the same reaction again removes it (toggle).", + Params: []Param{ + {Name: "message_id", Type: "number", Description: "ID of the message to react to", Required: true}, + {Name: "reaction", Type: "string", Description: "Reaction type: approve, reject, in_progress, done, published", Required: true}, + {Name: "metadata", Type: "string", Description: "JSON metadata object (optional)"}, + }, + Returns: "JSON with action ('added' or 'removed') and reaction details", + Examples: []Example{ + { + Description: "Approve a message", + Code: `call("react", {"message_id": 42, "reaction": "approve"})`, + }, + { + Description: "Toggle a reaction off (call same reaction again)", + Code: `call("react", {"message_id": 42, "reaction": "approve"})`, + }, + }, + }, + { + Name: "unreact", + Category: "reactions", + Description: "Remove a specific reaction from a message.", + Params: []Param{ + {Name: "message_id", Type: "number", Description: "ID of the message to remove reaction from", Required: true}, + {Name: "reaction", Type: "string", Description: "Reaction type to remove: approve, reject, in_progress, done, published", Required: true}, + }, + Returns: "JSON with message_id, reaction, and status 'removed'", + Examples: []Example{ + { + Description: "Remove an approval reaction", + Code: `call("unreact", {"message_id": 42, "reaction": "approve"})`, + }, + }, + }, + { + Name: "get_reactions", + Category: "reactions", + Description: "Get all reactions on a message and its derived workflow state.", + Params: []Param{ + {Name: "message_id", Type: "number", Description: "ID of the message to get reactions for", Required: true}, + }, + Returns: "JSON with reactions array and workflow_state", + Examples: []Example{ + { + Description: "Get reactions and workflow state for a message", + Code: `call("get_reactions", {"message_id": 42})`, + }, + }, + }, + { + Name: "list_by_state", + Category: "reactions", + Description: "List messages in a channel filtered by workflow state. Valid states: proposed, approved, in_progress, rejected, done, published.", + Params: []Param{ + {Name: "channel", Type: "string", Description: "Channel name", Required: true}, + {Name: "state", Type: "string", Description: "Workflow state to filter by: proposed, approved, in_progress, rejected, done, published", Required: true}, + }, + Returns: "JSON with message_ids array and count", + Examples: []Example{ + { + Description: "List approved messages in a channel", + Code: `call("list_by_state", {"channel": "approvals", "state": "approved"})`, + }, + }, + }, } } diff --git a/internal/actions/registry_test.go b/internal/actions/registry_test.go index 4a33389..a1ec619 100644 --- a/internal/actions/registry_test.go +++ b/internal/actions/registry_test.go @@ -4,11 +4,11 @@ import ( "testing" ) -func TestRegistryHas23Actions(t *testing.T) { +func TestRegistryHas27Actions(t *testing.T) { r := NewRegistry() got := len(r.List()) - if got != 23 { - t.Errorf("expected 23 actions, got %d", got) + if got != 27 { + t.Errorf("expected 27 actions, got %d", got) } } @@ -23,6 +23,7 @@ func TestRegistryCategories(t *testing.T) { {"channels", 9}, {"swarm", 5}, {"attachments", 2}, + {"reactions", 4}, } for _, tt := range tests { @@ -49,6 +50,8 @@ func TestRegistryGetByName(t *testing.T) { "post_task", "bid_task", "accept_bid", "complete_task", "list_tasks", // attachments "upload_attachment", "download_attachment", + // reactions + "react", "unreact", "get_reactions", "list_by_state", } for _, name := range allNames { diff --git a/internal/admin/socket.go b/internal/admin/socket.go index eb71af0..fd0904e 100644 --- a/internal/admin/socket.go +++ b/internal/admin/socket.go @@ -169,6 +169,8 @@ func (s *AdminServer) dispatch(req Request) Response { return s.handleChannelsCreate(ctx, req.Args) case "channels.join": return s.handleChannelsJoin(ctx, req.Args) + case "channels.update_settings": + return s.handleChannelsUpdateSettings(ctx, req.Args) // --- conversations --- case "conversations.list": @@ -995,6 +997,64 @@ func (s *AdminServer) handleChannelsJoin(ctx context.Context, args json.RawMessa }} } +func (s *AdminServer) handleChannelsUpdateSettings(ctx context.Context, args json.RawMessage) Response { + var p struct { + Name string `json:"name"` + AutoApprove *bool `json:"auto_approve,omitempty"` + StalemateRemindAfter string `json:"stalemate_remind_after,omitempty"` + StalemateEscalateAfter string `json:"stalemate_escalate_after,omitempty"` + } + if err := json.Unmarshal(args, &p); err != nil { + return Response{OK: false, Error: "invalid args: " + err.Error()} + } + if p.Name == "" { + return Response{OK: false, Error: "name is required"} + } + + // Build the SET clause dynamically based on provided fields + var setClauses []string + var setArgs []interface{} + + if p.AutoApprove != nil { + autoApproveVal := 0 + if *p.AutoApprove { + autoApproveVal = 1 + } + setClauses = append(setClauses, "auto_approve = ?") + setArgs = append(setArgs, autoApproveVal) + } + if p.StalemateRemindAfter != "" { + setClauses = append(setClauses, "stalemate_remind_after = ?") + setArgs = append(setArgs, p.StalemateRemindAfter) + } + if p.StalemateEscalateAfter != "" { + setClauses = append(setClauses, "stalemate_escalate_after = ?") + setArgs = append(setArgs, p.StalemateEscalateAfter) + } + + if len(setClauses) == 0 { + return Response{OK: false, Error: "at least one setting must be provided (auto_approve, stalemate_remind_after, stalemate_escalate_after)"} + } + + query := fmt.Sprintf("UPDATE channels SET %s WHERE LOWER(name) = LOWER(?)", strings.Join(setClauses, ", ")) + setArgs = append(setArgs, p.Name) + + result, err := s.db.ExecContext(ctx, query, setArgs...) + if err != nil { + return Response{OK: false, Error: "update channel settings: " + err.Error()} + } + + rowsAffected, _ := result.RowsAffected() + if rowsAffected == 0 { + return Response{OK: false, Error: fmt.Sprintf("channel not found: %s", p.Name)} + } + + return Response{OK: true, Data: map[string]interface{}{ + "channel": p.Name, + "updated": true, + }} +} + // ---------- conversations handlers ---------- func (s *AdminServer) handleConversationsList(ctx context.Context, args json.RawMessage) Response { diff --git a/internal/api/channels_handler.go b/internal/api/channels_handler.go index 78346e7..3bbfc48 100644 --- a/internal/api/channels_handler.go +++ b/internal/api/channels_handler.go @@ -1,6 +1,7 @@ package api import ( + "context" "encoding/json" "log/slog" "net/http" @@ -15,10 +16,16 @@ import ( // ChannelsHandler handles REST API requests for channels. type ChannelsHandler struct { - channelService *channels.Service - agentService *agents.AgentService - msgService *messaging.MessagingService - logger *slog.Logger + channelService *channels.Service + agentService *agents.AgentService + msgService *messaging.MessagingService + reactionService ChannelReactionService + logger *slog.Logger +} + +// ChannelReactionService is the subset of reactions.Service needed by ChannelsHandler. +type ChannelReactionService interface { + ListByState(ctx context.Context, channelID int64, state string) ([]int64, error) } // NewChannelsHandler creates a new channels handler. @@ -31,6 +38,11 @@ func NewChannelsHandler(channelService *channels.Service, agentService *agents.A } } +// SetReactionService sets the reaction service for workflow state queries. +func (h *ChannelsHandler) SetReactionService(svc ChannelReactionService) { + h.reactionService = svc +} + // ListChannels handles GET /api/channels. func (h *ChannelsHandler) ListChannels(w http.ResponseWriter, r *http.Request) { ownerID, ok := OwnerIDFromContext(r.Context()) @@ -298,3 +310,112 @@ func (h *ChannelsHandler) LeaveChannel(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]string{"status": "left"}) } + +// UpdateSettings handles PUT /api/channels/{name}/settings. +func (h *ChannelsHandler) UpdateSettings(w http.ResponseWriter, r *http.Request) { + _, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + name := chi.URLParam(r, "name") + ch, err := h.channelService.GetChannelByName(r.Context(), name) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Channel not found")) + return + } + + var req struct { + AutoApprove *bool `json:"auto_approve"` + StalemateRemindAfter *string `json:"stalemate_remind_after"` + StalemateEscalateAfter *string `json:"stalemate_escalate_after"` + } + + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_request", "Invalid JSON body")) + return + } + + settings := channels.ChannelSettings{ + AutoApprove: ch.AutoApprove, + StalemateRemindAfter: ch.StalemateRemindAfter, + StalemateEscalateAfter: ch.StalemateEscalateAfter, + } + + if req.AutoApprove != nil { + settings.AutoApprove = *req.AutoApprove + } + if req.StalemateRemindAfter != nil { + settings.StalemateRemindAfter = *req.StalemateRemindAfter + } + if req.StalemateEscalateAfter != nil { + settings.StalemateEscalateAfter = *req.StalemateEscalateAfter + } + + updated, err := h.channelService.UpdateChannelSettings(r.Context(), ch.ID, settings) + if err != nil { + h.logger.Error("update channel settings failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to update channel settings")) + return + } + + writeJSON(w, http.StatusOK, map[string]any{"channel": updated}) +} + +// ListByState handles GET /api/channels/{name}/messages/by-state?state=X. +func (h *ChannelsHandler) ListByState(w http.ResponseWriter, r *http.Request) { + _, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + if h.reactionService == nil { + writeJSON(w, http.StatusServiceUnavailable, errorBody("unavailable", "Reactions service not configured")) + return + } + + name := chi.URLParam(r, "name") + ch, err := h.channelService.GetChannelByName(r.Context(), name) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Channel not found")) + return + } + + state := r.URL.Query().Get("state") + if state == "" { + writeJSON(w, http.StatusBadRequest, errorBody("missing_state", "Query parameter 'state' is required")) + return + } + + ids, err := h.reactionService.ListByState(r.Context(), ch.ID, state) + if err != nil { + h.logger.Error("list by state failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("invalid_state", err.Error())) + return + } + + // Load messages by IDs + var messages []*messaging.Message + for _, id := range ids { + msg, err := h.msgService.GetMessageByID(r.Context(), id) + if err != nil { + continue + } + messages = append(messages, msg) + } + + if messages == nil { + messages = []*messaging.Message{} + } + + // Enrich messages with reactions, reply counts, attachments + h.msgService.EnrichMessages(r.Context(), messages) + + writeJSON(w, http.StatusOK, map[string]any{ + "messages": messages, + "state": state, + "total": len(messages), + }) +} diff --git a/internal/api/reactions_handler.go b/internal/api/reactions_handler.go new file mode 100644 index 0000000..af739c6 --- /dev/null +++ b/internal/api/reactions_handler.go @@ -0,0 +1,233 @@ +package api + +import ( + "encoding/json" + "fmt" + "log/slog" + "net/http" + "strconv" + + "github.com/go-chi/chi/v5" + + "github.com/synapbus/synapbus/internal/agents" + "github.com/synapbus/synapbus/internal/auth" + "github.com/synapbus/synapbus/internal/messaging" + "github.com/synapbus/synapbus/internal/reactions" +) + +// ReactionsHandler handles REST API requests for message reactions. +type ReactionsHandler struct { + reactionService *reactions.Service + msgService *messaging.MessagingService + agentService *agents.AgentService + logger *slog.Logger +} + +// NewReactionsHandler creates a new reactions handler. +func NewReactionsHandler(reactionService *reactions.Service, msgService *messaging.MessagingService, agentService *agents.AgentService) *ReactionsHandler { + return &ReactionsHandler{ + reactionService: reactionService, + msgService: msgService, + agentService: agentService, + logger: slog.Default().With("component", "api.reactions"), + } +} + +// Toggle handles POST /api/messages/{id}/reactions. +func (h *ReactionsHandler) Toggle(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_id", "Invalid message ID")) + return + } + + var req struct { + Reaction string `json:"reaction"` + Metadata json.RawMessage `json:"metadata,omitempty"` + } + + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_request", "Invalid JSON body")) + return + } + + if req.Reaction == "" { + writeJSON(w, http.StatusBadRequest, errorBody("validation_error", "Reaction type is required")) + return + } + + // Verify the message exists + msg, err := h.msgService.GetMessageByID(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Message not found")) + return + } + + // Determine the acting agent name from the session + agentName, err := h.resolveAgentName(r, ownerID, msg) + if err != nil { + h.logger.Error("resolve agent name failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("no_agent", err.Error())) + return + } + + result, err := h.reactionService.Toggle(r.Context(), id, agentName, req.Reaction, req.Metadata) + if err != nil { + if err == reactions.ErrInvalidReaction { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_reaction", err.Error())) + return + } + if err == reactions.ErrReactionLimit { + writeJSON(w, http.StatusBadRequest, errorBody("reaction_limit", err.Error())) + return + } + h.logger.Error("toggle reaction failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to toggle reaction")) + return + } + + // Reload reactions and workflow state for the response + rxs, state, err := h.reactionService.GetReactions(r.Context(), id) + if err != nil { + h.logger.Error("get reactions failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to get reactions")) + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "action": result.Action, + "reaction": result.Reaction, + "reactions": rxs, + "workflow_state": state, + }) +} + +// GetReactions handles GET /api/messages/{id}/reactions. +func (h *ReactionsHandler) GetReactions(w http.ResponseWriter, r *http.Request) { + _, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_id", "Invalid message ID")) + return + } + + // Verify the message exists + _, err = h.msgService.GetMessageByID(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Message not found")) + return + } + + rxs, state, err := h.reactionService.GetReactions(r.Context(), id) + if err != nil { + h.logger.Error("get reactions failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to get reactions")) + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "reactions": rxs, + "workflow_state": state, + }) +} + +// Remove handles DELETE /api/messages/{id}/reactions/{reaction}. +func (h *ReactionsHandler) Remove(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_id", "Invalid message ID")) + return + } + + reactionType := chi.URLParam(r, "reaction") + if reactionType == "" { + writeJSON(w, http.StatusBadRequest, errorBody("validation_error", "Reaction type is required")) + return + } + + // Verify the message exists + msg, err := h.msgService.GetMessageByID(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Message not found")) + return + } + + // Determine the acting agent name + agentName, err := h.resolveAgentName(r, ownerID, msg) + if err != nil { + h.logger.Error("resolve agent name failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("no_agent", err.Error())) + return + } + + if err := h.reactionService.Remove(r.Context(), id, agentName, reactionType); err != nil { + if err == reactions.ErrInvalidReaction { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_reaction", err.Error())) + return + } + h.logger.Error("remove reaction failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to remove reaction")) + return + } + + // Reload reactions and workflow state for the response + rxs, state, err := h.reactionService.GetReactions(r.Context(), id) + if err != nil { + h.logger.Error("get reactions failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to get reactions")) + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "status": "removed", + "reactions": rxs, + "workflow_state": state, + }) +} + +// resolveAgentName determines the agent name for the current session user. +// For session-authenticated users (Web UI), it returns the human agent. +// For API key / OAuth, it falls back to the first owned agent. +func (h *ReactionsHandler) resolveAgentName(r *http.Request, ownerID int64, msg *messaging.Message) (string, error) { + if _, isSession := auth.SessionIDFromContext(r.Context()); isSession { + humanAgent, err := h.agentService.GetHumanAgentForUser(r.Context(), ownerID) + if err != nil { + return "", err + } + if humanAgent == nil { + return "", fmt.Errorf("no human agent found for this user") + } + return humanAgent.Name, nil + } + + // Non-session: find an owned agent + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil || len(ownedAgents) == 0 { + return "", fmt.Errorf("no agents registered") + } + + // Prefer human-type agent + for _, a := range ownedAgents { + if a.Type == "human" { + return a.Name, nil + } + } + return ownedAgents[0].Name, nil +} diff --git a/internal/api/router.go b/internal/api/router.go index 01c28a7..c64f8e6 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -13,6 +13,7 @@ import ( "github.com/synapbus/synapbus/internal/k8s" "github.com/synapbus/synapbus/internal/messaging" "github.com/synapbus/synapbus/internal/push" + "github.com/synapbus/synapbus/internal/reactions" "github.com/synapbus/synapbus/internal/trace" "github.com/synapbus/synapbus/internal/webhooks" ) @@ -32,6 +33,7 @@ type RouterConfig struct { WebhookStore webhooks.WebhookStore K8sService *k8s.K8sService K8sStore k8s.K8sStore + ReactionService *reactions.Service PushService *push.Service SSEHub *SSEHub Broadcaster *SSEBroadcaster @@ -142,9 +144,24 @@ func NewRouterWithConfig(cfg RouterConfig) chi.Router { }) } + // Reactions + if cfg.ReactionService != nil { + reactionsHandler := NewReactionsHandler(cfg.ReactionService, cfg.MsgService, cfg.AgentService) + r.Group(func(r chi.Router) { + r.Use(authMiddleware) + + r.Post("/api/messages/{id}/reactions", reactionsHandler.Toggle) + r.Get("/api/messages/{id}/reactions", reactionsHandler.GetReactions) + r.Delete("/api/messages/{id}/reactions/{reaction}", reactionsHandler.Remove) + }) + } + // Channels if cfg.ChannelService != nil { channelsHandler := NewChannelsHandler(cfg.ChannelService, cfg.AgentService, cfg.MsgService) + if cfg.ReactionService != nil { + channelsHandler.SetReactionService(cfg.ReactionService) + } r.Group(func(r chi.Router) { r.Use(authMiddleware) @@ -152,6 +169,8 @@ func NewRouterWithConfig(cfg RouterConfig) chi.Router { r.Get("/api/channels/{name}", channelsHandler.GetChannel) r.Post("/api/channels", channelsHandler.CreateChannel) r.Get("/api/channels/{name}/messages", channelsHandler.ChannelMessages) + r.Get("/api/channels/{name}/messages/by-state", channelsHandler.ListByState) + r.Put("/api/channels/{name}/settings", channelsHandler.UpdateSettings) r.Post("/api/channels/{name}/join", channelsHandler.JoinChannel) r.Post("/api/channels/{name}/leave", channelsHandler.LeaveChannel) }) diff --git a/internal/channels/service.go b/internal/channels/service.go index 0804fc6..fe51884 100644 --- a/internal/channels/service.go +++ b/internal/channels/service.go @@ -459,6 +459,31 @@ func (s *Service) UpdateChannel(ctx context.Context, channelID int64, req Update return ch, nil } +// UpdateChannelSettings updates the workflow-related settings for a channel. +func (s *Service) UpdateChannelSettings(ctx context.Context, channelID int64, settings ChannelSettings) (*Channel, error) { + store, ok := s.store.(*SQLiteChannelStore) + if !ok { + return nil, fmt.Errorf("channel store does not support settings update") + } + + if err := store.UpdateChannelSettings(ctx, channelID, settings); err != nil { + return nil, err + } + + // Reload channel to return updated state + ch, err := s.store.GetChannel(ctx, channelID) + if err != nil { + return nil, err + } + + s.logger.Info("channel settings updated", + "channel_id", channelID, + "auto_approve", settings.AutoApprove, + ) + + return ch, nil +} + // BroadcastMessage sends a message to a channel. It creates a single channel // message (visible in the channel timeline via GetChannelMessages) and also // delivers individual DM notifications to each member's inbox. diff --git a/internal/channels/store.go b/internal/channels/store.go index 85a0f84..4132de1 100644 --- a/internal/channels/store.go +++ b/internal/channels/store.go @@ -83,9 +83,9 @@ func (s *SQLiteChannelStore) GetChannel(ctx context.Context, id int64) (*Channel var ch Channel var isPrivate, isSystem int err := s.db.QueryRowContext(ctx, - `SELECT id, name, description, topic, type, is_private, is_system, created_by, created_at, updated_at + `SELECT id, name, description, topic, type, is_private, is_system, created_by, auto_approve, stalemate_remind_after, stalemate_escalate_after, created_at, updated_at FROM channels WHERE id = ?`, id, - ).Scan(&ch.ID, &ch.Name, &ch.Description, &ch.Topic, &ch.Type, &isPrivate, &isSystem, &ch.CreatedBy, &ch.CreatedAt, &ch.UpdatedAt) + ).Scan(&ch.ID, &ch.Name, &ch.Description, &ch.Topic, &ch.Type, &isPrivate, &isSystem, &ch.CreatedBy, &ch.AutoApprove, &ch.StalemateRemindAfter, &ch.StalemateEscalateAfter, &ch.CreatedAt, &ch.UpdatedAt) if err != nil { if err == sql.ErrNoRows { return nil, ErrChannelNotFound @@ -102,9 +102,9 @@ func (s *SQLiteChannelStore) GetChannelByName(ctx context.Context, name string) var ch Channel var isPrivate, isSystem int err := s.db.QueryRowContext(ctx, - `SELECT id, name, description, topic, type, is_private, is_system, created_by, created_at, updated_at + `SELECT id, name, description, topic, type, is_private, is_system, created_by, auto_approve, stalemate_remind_after, stalemate_escalate_after, created_at, updated_at FROM channels WHERE LOWER(name) = LOWER(?)`, name, - ).Scan(&ch.ID, &ch.Name, &ch.Description, &ch.Topic, &ch.Type, &isPrivate, &isSystem, &ch.CreatedBy, &ch.CreatedAt, &ch.UpdatedAt) + ).Scan(&ch.ID, &ch.Name, &ch.Description, &ch.Topic, &ch.Type, &isPrivate, &isSystem, &ch.CreatedBy, &ch.AutoApprove, &ch.StalemateRemindAfter, &ch.StalemateEscalateAfter, &ch.CreatedAt, &ch.UpdatedAt) if err != nil { if err == sql.ErrNoRows { return nil, ErrChannelNotFound @@ -120,7 +120,7 @@ func (s *SQLiteChannelStore) GetChannelByName(ctx context.Context, name string) // is a member or has a pending invite. func (s *SQLiteChannelStore) ListChannels(ctx context.Context, agentName string) ([]*Channel, error) { rows, err := s.db.QueryContext(ctx, - `SELECT DISTINCT c.id, c.name, c.description, c.topic, c.type, c.is_private, c.is_system, c.created_by, c.created_at, c.updated_at + `SELECT DISTINCT c.id, c.name, c.description, c.topic, c.type, c.is_private, c.is_system, c.created_by, c.auto_approve, c.stalemate_remind_after, c.stalemate_escalate_after, c.created_at, c.updated_at FROM channels c WHERE c.is_private = 0 OR EXISTS (SELECT 1 FROM channel_members cm WHERE cm.channel_id = c.id AND cm.agent_name = ?) @@ -137,7 +137,7 @@ func (s *SQLiteChannelStore) ListChannels(ctx context.Context, agentName string) for rows.Next() { var ch Channel var isPrivate, isSystem int - if err := rows.Scan(&ch.ID, &ch.Name, &ch.Description, &ch.Topic, &ch.Type, &isPrivate, &isSystem, &ch.CreatedBy, &ch.CreatedAt, &ch.UpdatedAt); err != nil { + if err := rows.Scan(&ch.ID, &ch.Name, &ch.Description, &ch.Topic, &ch.Type, &isPrivate, &isSystem, &ch.CreatedBy, &ch.AutoApprove, &ch.StalemateRemindAfter, &ch.StalemateEscalateAfter, &ch.CreatedAt, &ch.UpdatedAt); err != nil { return nil, fmt.Errorf("scan channel: %w", err) } ch.IsPrivate = isPrivate != 0 @@ -415,6 +415,23 @@ func (s *SQLiteChannelStore) GetChannelSummaries(ctx context.Context, agentName return summaries, rows.Err() } +// UpdateChannelSettings updates the workflow-related settings for a channel. +func (s *SQLiteChannelStore) UpdateChannelSettings(ctx context.Context, id int64, settings ChannelSettings) error { + result, err := s.db.ExecContext(ctx, + `UPDATE channels SET auto_approve = ?, stalemate_remind_after = ?, stalemate_escalate_after = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?`, + settings.AutoApprove, settings.StalemateRemindAfter, settings.StalemateEscalateAfter, id, + ) + if err != nil { + return fmt.Errorf("update channel settings: %w", err) + } + rowsAffected, _ := result.RowsAffected() + if rowsAffected == 0 { + return ErrChannelNotFound + } + s.logger.Info("channel settings updated", "id", id) + return nil +} + // isUniqueConstraintError checks if an error is a SQLite unique constraint violation. func isUniqueConstraintError(err error) bool { return strings.Contains(err.Error(), "UNIQUE constraint failed") diff --git a/internal/channels/types.go b/internal/channels/types.go index 23b66f2..76fe0d1 100644 --- a/internal/channels/types.go +++ b/internal/channels/types.go @@ -25,16 +25,19 @@ const ( // Channel represents a named group communication space. type Channel struct { - ID int64 `json:"id"` - Name string `json:"name"` - Description string `json:"description"` - Topic string `json:"topic"` - Type string `json:"type"` - IsPrivate bool `json:"is_private"` - IsSystem bool `json:"is_system"` - CreatedBy string `json:"created_by"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID int64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Topic string `json:"topic"` + Type string `json:"type"` + IsPrivate bool `json:"is_private"` + IsSystem bool `json:"is_system"` + CreatedBy string `json:"created_by"` + AutoApprove bool `json:"auto_approve"` + StalemateRemindAfter string `json:"stalemate_remind_after"` + StalemateEscalateAfter string `json:"stalemate_escalate_after"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // ChannelWithCount embeds Channel and adds a member count. @@ -107,6 +110,13 @@ type JoinChannelRequest struct { AgentName string `json:"agent_name"` } +// ChannelSettings holds workflow-related settings for a channel. +type ChannelSettings struct { + AutoApprove bool `json:"auto_approve"` + StalemateRemindAfter string `json:"stalemate_remind_after"` + StalemateEscalateAfter string `json:"stalemate_escalate_after"` +} + // InviteRequest is the input for inviting an agent to a channel. type InviteRequest struct { ChannelID int64 `json:"channel_id"` diff --git a/internal/mcp/bridge.go b/internal/mcp/bridge.go index 6a8580c..a8e66d8 100644 --- a/internal/mcp/bridge.go +++ b/internal/mcp/bridge.go @@ -14,6 +14,7 @@ import ( "github.com/synapbus/synapbus/internal/attachments" "github.com/synapbus/synapbus/internal/channels" "github.com/synapbus/synapbus/internal/messaging" + "github.com/synapbus/synapbus/internal/reactions" "github.com/synapbus/synapbus/internal/search" ) @@ -26,6 +27,7 @@ type ServiceBridge struct { swarmService *channels.SwarmService attachmentService *attachments.Service searchService *search.Service + reactionService *reactions.Service agentName string } @@ -37,6 +39,7 @@ func NewServiceBridge( swarmService *channels.SwarmService, attachmentService *attachments.Service, searchService *search.Service, + reactionService *reactions.Service, agentName string, ) *ServiceBridge { return &ServiceBridge{ @@ -46,6 +49,7 @@ func NewServiceBridge( swarmService: swarmService, attachmentService: attachmentService, searchService: searchService, + reactionService: reactionService, agentName: agentName, } } @@ -103,6 +107,16 @@ func (b *ServiceBridge) Call(ctx context.Context, actionName string, args map[st case "download_attachment": return b.callDownloadAttachment(ctx, args) + // --- Reactions --- + case "react": + return b.callReact(ctx, args) + case "unreact": + return b.callUnreact(ctx, args) + case "get_reactions": + return b.callGetReactions(ctx, args) + case "list_by_state": + return b.callListByState(ctx, args) + // --- DM send (also accessible via bridge for execute tool) --- case "send_message": return b.callSendMessage(ctx, args) @@ -920,6 +934,136 @@ func (b *ServiceBridge) callDownloadAttachment(ctx context.Context, args map[str }, nil } +// --- Reaction implementations --- + +func (b *ServiceBridge) callReact(ctx context.Context, args map[string]any) (any, error) { + if b.reactionService == nil { + return nil, fmt.Errorf("reaction service not available") + } + + messageID := getInt(args, "message_id", 0) + if messageID == 0 { + return nil, fmt.Errorf("'message_id' parameter is required") + } + + reaction := getString(args, "reaction", "") + if reaction == "" { + return nil, fmt.Errorf("'reaction' parameter is required") + } + + var metadata json.RawMessage + if metaStr := getString(args, "metadata", ""); metaStr != "" { + if !json.Valid([]byte(metaStr)) { + return nil, fmt.Errorf("metadata must be valid JSON") + } + metadata = json.RawMessage(metaStr) + } + + result, err := b.reactionService.Toggle(ctx, int64(messageID), b.agentName, reaction, metadata) + if err != nil { + return nil, err + } + + resp := map[string]any{ + "action": result.Action, + "message_id": messageID, + "reaction": reaction, + } + if result.Reaction != nil { + resp["id"] = result.Reaction.ID + resp["created_at"] = result.Reaction.CreatedAt + } + return resp, nil +} + +func (b *ServiceBridge) callUnreact(ctx context.Context, args map[string]any) (any, error) { + if b.reactionService == nil { + return nil, fmt.Errorf("reaction service not available") + } + + messageID := getInt(args, "message_id", 0) + if messageID == 0 { + return nil, fmt.Errorf("'message_id' parameter is required") + } + + reaction := getString(args, "reaction", "") + if reaction == "" { + return nil, fmt.Errorf("'reaction' parameter is required") + } + + if err := b.reactionService.Remove(ctx, int64(messageID), b.agentName, reaction); err != nil { + return nil, err + } + + return map[string]any{ + "message_id": messageID, + "reaction": reaction, + "status": "removed", + }, nil +} + +func (b *ServiceBridge) callGetReactions(ctx context.Context, args map[string]any) (any, error) { + if b.reactionService == nil { + return nil, fmt.Errorf("reaction service not available") + } + + messageID := getInt(args, "message_id", 0) + if messageID == 0 { + return nil, fmt.Errorf("'message_id' parameter is required") + } + + rxns, state, err := b.reactionService.GetReactions(ctx, int64(messageID)) + if err != nil { + return nil, err + } + + return map[string]any{ + "reactions": rxns, + "workflow_state": state, + "count": len(rxns), + }, nil +} + +func (b *ServiceBridge) callListByState(ctx context.Context, args map[string]any) (any, error) { + if b.reactionService == nil { + return nil, fmt.Errorf("reaction service not available") + } + if b.channelService == nil { + return nil, fmt.Errorf("channel service not available") + } + + channelName := getString(args, "channel", "") + if channelName == "" { + return nil, fmt.Errorf("'channel' parameter is required") + } + + state := getString(args, "state", "") + if state == "" { + return nil, fmt.Errorf("'state' parameter is required") + } + + ch, err := b.channelService.GetChannelByName(ctx, channelName) + if err != nil { + return nil, err + } + + messageIDs, err := b.reactionService.ListByState(ctx, ch.ID, state) + if err != nil { + return nil, err + } + + if messageIDs == nil { + messageIDs = []int64{} + } + + return map[string]any{ + "message_ids": messageIDs, + "count": len(messageIDs), + "channel": channelName, + "state": state, + }, nil +} + // --- Helpers --- // resolveChannelID resolves a channel ID from either channel_id or channel_name in args. diff --git a/internal/mcp/bridge_test.go b/internal/mcp/bridge_test.go index 077b899..bb3abdd 100644 --- a/internal/mcp/bridge_test.go +++ b/internal/mcp/bridge_test.go @@ -43,6 +43,7 @@ func newTestBridge(t *testing.T) (*ServiceBridge, *messaging.MessagingService, * swarmService, nil, // attachmentService nil, // searchService + nil, // reactionService "agent-a", ) return bridge, msgService, agentService, channelService @@ -185,7 +186,7 @@ func TestBridge_JoinChannel(t *testing.T) { bridge.agentService, bridge.channelService, bridge.swarmService, - nil, nil, + nil, nil, nil, "agent-b", ) diff --git a/internal/mcp/channel_tools_test.go b/internal/mcp/channel_tools_test.go index 69c531b..541dc48 100644 --- a/internal/mcp/channel_tools_test.go +++ b/internal/mcp/channel_tools_test.go @@ -50,6 +50,7 @@ func newTestHybridWithChannels(t *testing.T) (*HybridToolRegistrar, *channels.Se nil, // swarmService nil, // attachmentService nil, // searchService + nil, // reactionService jsPool, actionRegistry, actionIndex, diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 38f887a..da93860 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -18,6 +18,7 @@ import ( "github.com/synapbus/synapbus/internal/console" "github.com/synapbus/synapbus/internal/jsruntime" "github.com/synapbus/synapbus/internal/messaging" + "github.com/synapbus/synapbus/internal/reactions" "github.com/synapbus/synapbus/internal/search" "github.com/synapbus/synapbus/internal/trace" ) @@ -40,6 +41,7 @@ func NewMCPServer( swarmService *channels.SwarmService, attachmentService *attachments.Service, searchService *search.Service, + reactionService *reactions.Service, consolePrinter *console.Printer, jsPool *jsruntime.Pool, actionRegistry *actions.Registry, @@ -153,6 +155,7 @@ func NewMCPServer( swarmService, attachmentService, searchService, + reactionService, jsPool, actionRegistry, actionIndex, diff --git a/internal/mcp/server_test.go b/internal/mcp/server_test.go index 5e82225..30362aa 100644 --- a/internal/mcp/server_test.go +++ b/internal/mcp/server_test.go @@ -38,7 +38,7 @@ func newTestMCPServer(t *testing.T, con *console.Printer) (*MCPServer, *messagin actionRegistry := actions.NewRegistry() actionIndex := actions.NewIndex(actionRegistry.List()) - srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, con, jsPool, actionRegistry, actionIndex, db) + srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, con, jsPool, actionRegistry, actionIndex, db) return srv, msgService, agentService } @@ -133,7 +133,7 @@ func TestMCPToolCall_WithValidAPIKey(t *testing.T) { actionIndex := actions.NewIndex(actionRegistry.List()) // Create MCP server - srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, jsPool, actionRegistry, actionIndex, db) + srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, nil, jsPool, actionRegistry, actionIndex, db) // Mount with auth middleware, just like main.go does mux := http.NewServeMux() @@ -188,7 +188,7 @@ func TestMCPToolCall_InvalidAPIKeyReturns401(t *testing.T) { actionRegistry := actions.NewRegistry() actionIndex := actions.NewIndex(actionRegistry.List()) - srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, jsPool, actionRegistry, actionIndex, db) + srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, nil, jsPool, actionRegistry, actionIndex, db) mux := http.NewServeMux() handler := agents.OptionalAuthMiddlewareWithAPIKeys(agentService, apiKeyService)(srv.Handler()) diff --git a/internal/mcp/tools_hybrid.go b/internal/mcp/tools_hybrid.go index 8728195..cab6be5 100644 --- a/internal/mcp/tools_hybrid.go +++ b/internal/mcp/tools_hybrid.go @@ -18,6 +18,7 @@ import ( "github.com/synapbus/synapbus/internal/channels" "github.com/synapbus/synapbus/internal/jsruntime" "github.com/synapbus/synapbus/internal/messaging" + "github.com/synapbus/synapbus/internal/reactions" "github.com/synapbus/synapbus/internal/search" ) @@ -29,6 +30,7 @@ type HybridToolRegistrar struct { swarmService *channels.SwarmService attachmentService *attachments.Service searchService *search.Service + reactionService *reactions.Service jsPool *jsruntime.Pool actionRegistry *actions.Registry actionIndex *actions.Index @@ -44,6 +46,7 @@ func NewHybridToolRegistrar( swarmService *channels.SwarmService, attachmentService *attachments.Service, searchService *search.Service, + reactionService *reactions.Service, jsPool *jsruntime.Pool, actionRegistry *actions.Registry, actionIndex *actions.Index, @@ -56,6 +59,7 @@ func NewHybridToolRegistrar( swarmService: swarmService, attachmentService: attachmentService, searchService: searchService, + reactionService: reactionService, jsPool: jsPool, actionRegistry: actionRegistry, actionIndex: actionIndex, @@ -475,6 +479,7 @@ func (h *HybridToolRegistrar) handleExecute(ctx context.Context, req mcplib.Call h.swarmService, h.attachmentService, h.searchService, + h.reactionService, agentName, ) diff --git a/internal/mcp/tools_test.go b/internal/mcp/tools_test.go index d669641..50d86d3 100644 --- a/internal/mcp/tools_test.go +++ b/internal/mcp/tools_test.go @@ -68,6 +68,7 @@ func newTestHybridRegistrar(t *testing.T) (*HybridToolRegistrar, *messaging.Mess nil, // swarmService nil, // attachmentService nil, // searchService + nil, // reactionService jsPool, actionRegistry, actionIndex, diff --git a/internal/messaging/service.go b/internal/messaging/service.go index 7d4c858..baeccd4 100644 --- a/internal/messaging/service.go +++ b/internal/messaging/service.go @@ -31,6 +31,12 @@ type AttachmentLinker interface { GetByMessageID(ctx context.Context, messageID int64) ([]AttachmentInfo, error) } +// ReactionEnricher loads reactions for messages. This avoids importing the +// reactions package directly. Set via SetReactionEnricher. +type ReactionEnricher interface { + GetByMessageIDs(ctx context.Context, messageIDs []int64) (map[int64][]ReactionInfo, error) +} + // MessagingService provides business logic for messaging operations. type MessagingService struct { store MessageStore @@ -38,6 +44,7 @@ type MessagingService struct { dispatcher dispatcher.EventDispatcher embeddings EmbeddingNotifier attLinker AttachmentLinker + rxEnricher ReactionEnricher listeners []MessageListener logger *slog.Logger } @@ -66,6 +73,11 @@ func (s *MessagingService) SetAttachmentLinker(l AttachmentLinker) { s.attLinker = l } +// SetReactionEnricher sets the reaction enricher for message reaction loading. +func (s *MessagingService) SetReactionEnricher(e ReactionEnricher) { + s.rxEnricher = e +} + // AddMessageListener registers a listener that is notified after message creation. func (s *MessagingService) AddMessageListener(l MessageListener) { s.listeners = append(s.listeners, l) @@ -585,6 +597,38 @@ func (s *MessagingService) EnrichMessages(ctx context.Context, msgs []*Message) } } } + + // Batch-load reactions and derive workflow state. + if s.rxEnricher != nil { + rxMap, err := s.rxEnricher.GetByMessageIDs(ctx, ids) + if err != nil { + s.logger.Error("failed to load reactions", "error", err) + } else { + for _, m := range msgs { + if rxs, ok := rxMap[m.ID]; ok && len(rxs) > 0 { + m.Reactions = rxs + // Derive workflow state from reactions + highestPriority := 0 + priorities := map[string]int{ + "approve": 2, "in_progress": 3, "reject": 4, "done": 5, "published": 6, + } + states := map[string]string{ + "approve": "approved", "in_progress": "in_progress", "reject": "rejected", "done": "done", "published": "published", + } + for _, rx := range rxs { + if p, ok := priorities[rx.Reaction]; ok && p > highestPriority { + highestPriority = p + m.WorkflowState = states[rx.Reaction] + } + } + } + // Default to "proposed" for channel messages with no reactions + if m.WorkflowState == "" && m.ChannelID != nil { + m.WorkflowState = "proposed" + } + } + } + } } // GetConversation returns a conversation and its messages. diff --git a/internal/messaging/types.go b/internal/messaging/types.go index f59fe80..036a697 100644 --- a/internal/messaging/types.go +++ b/internal/messaging/types.go @@ -42,6 +42,16 @@ type Message struct { UpdatedAt time.Time `json:"updated_at"` ReplyCount int `json:"reply_count"` Attachments []AttachmentInfo `json:"attachments,omitempty"` + WorkflowState string `json:"workflow_state,omitempty"` + Reactions []ReactionInfo `json:"reactions,omitempty"` +} + +// ReactionInfo is a lightweight reaction summary included in message responses. +type ReactionInfo struct { + AgentName string `json:"agent_name"` + Reaction string `json:"reaction"` + Metadata json.RawMessage `json:"metadata,omitempty"` + CreatedAt time.Time `json:"created_at"` } // Conversation groups related messages into a thread. diff --git a/internal/reactions/model.go b/internal/reactions/model.go new file mode 100644 index 0000000..f8ec348 --- /dev/null +++ b/internal/reactions/model.go @@ -0,0 +1,113 @@ +// Package reactions provides message reaction types, storage, and workflow state logic. +package reactions + +import ( + "encoding/json" + "errors" + "time" +) + +// Valid reaction types. +const ( + ReactionApprove = "approve" + ReactionReject = "reject" + ReactionInProgress = "in_progress" + ReactionDone = "done" + ReactionPublished = "published" +) + +// Workflow states (derived from reactions). +const ( + StateProposed = "proposed" + StateApproved = "approved" + StateInProgress = "in_progress" + StateRejected = "rejected" + StateDone = "done" + StatePublished = "published" +) + +// reactionPriority maps reaction types to their priority for state derivation. +// Higher number = higher priority = wins for badge display. +var reactionPriority = map[string]int{ + ReactionApprove: 2, + ReactionInProgress: 3, + ReactionReject: 4, + ReactionDone: 5, + ReactionPublished: 6, +} + +// reactionToState maps reaction types to workflow states. +var reactionToState = map[string]string{ + ReactionApprove: StateApproved, + ReactionReject: StateRejected, + ReactionInProgress: StateInProgress, + ReactionDone: StateDone, + ReactionPublished: StatePublished, +} + +// TerminalStates are states that should not trigger stalemate checks. +var TerminalStates = map[string]bool{ + StateRejected: true, + StateDone: true, + StatePublished: true, +} + +// MaxReactionsPerMessage is the safety limit. +const MaxReactionsPerMessage = 100 + +// Sentinel errors. +var ( + ErrInvalidReaction = errors.New("invalid reaction type: must be one of approve, reject, in_progress, done, published") + ErrReactionLimit = errors.New("maximum reactions per message (100) reached") + ErrNotMember = errors.New("only channel members can react to messages") +) + +// Reaction represents a single reaction on a message. +type Reaction struct { + ID int64 `json:"id"` + MessageID int64 `json:"message_id"` + AgentName string `json:"agent_name"` + Reaction string `json:"reaction"` + Metadata json.RawMessage `json:"metadata"` + CreatedAt time.Time `json:"created_at"` +} + +// ValidReactions is the set of allowed reaction types. +var ValidReactions = map[string]bool{ + ReactionApprove: true, + ReactionReject: true, + ReactionInProgress: true, + ReactionDone: true, + ReactionPublished: true, +} + +// IsValidReaction returns true if the reaction type is valid. +func IsValidReaction(r string) bool { + return ValidReactions[r] +} + +// ComputeWorkflowState derives the workflow state from a list of reactions. +// Returns "proposed" if no reactions exist. +func ComputeWorkflowState(reactions []*Reaction) string { + if len(reactions) == 0 { + return StateProposed + } + + highestPriority := 0 + highestState := StateProposed + + for _, r := range reactions { + p, ok := reactionPriority[r.Reaction] + if ok && p > highestPriority { + highestPriority = p + highestState = reactionToState[r.Reaction] + } + } + + return highestState +} + +// IsTerminalState returns true if the state should not trigger stalemate checks. +func IsTerminalState(state string) bool { + return TerminalStates[state] +} diff --git a/internal/reactions/model_test.go b/internal/reactions/model_test.go new file mode 100644 index 0000000..caaf56b --- /dev/null +++ b/internal/reactions/model_test.go @@ -0,0 +1,150 @@ +package reactions + +import ( + "testing" +) + +func TestIsValidReaction(t *testing.T) { + tests := []struct { + name string + reaction string + want bool + }{ + {"approve is valid", ReactionApprove, true}, + {"reject is valid", ReactionReject, true}, + {"in_progress is valid", ReactionInProgress, true}, + {"done is valid", ReactionDone, true}, + {"published is valid", ReactionPublished, true}, + {"empty string is invalid", "", false}, + {"thumbs_up is invalid", "thumbs_up", false}, + {"like is invalid", "like", false}, + {"APPROVE uppercase is invalid", "APPROVE", false}, + {"random text is invalid", "foobar", false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := IsValidReaction(tt.reaction) + if got != tt.want { + t.Errorf("IsValidReaction(%q) = %v, want %v", tt.reaction, got, tt.want) + } + }) + } +} + +func TestComputeWorkflowState(t *testing.T) { + tests := []struct { + name string + reactions []*Reaction + want string + }{ + { + name: "empty reactions returns proposed", + reactions: []*Reaction{}, + want: StateProposed, + }, + { + name: "nil reactions returns proposed", + reactions: nil, + want: StateProposed, + }, + { + name: "single approve returns approved", + reactions: []*Reaction{ + {Reaction: ReactionApprove, AgentName: "agent-a"}, + }, + want: StateApproved, + }, + { + name: "approve + in_progress returns in_progress (higher priority wins)", + reactions: []*Reaction{ + {Reaction: ReactionApprove, AgentName: "agent-a"}, + {Reaction: ReactionInProgress, AgentName: "agent-b"}, + }, + want: StateInProgress, + }, + { + name: "single reject returns rejected", + reactions: []*Reaction{ + {Reaction: ReactionReject, AgentName: "agent-a"}, + }, + want: StateRejected, + }, + { + name: "single done returns done", + reactions: []*Reaction{ + {Reaction: ReactionDone, AgentName: "agent-a"}, + }, + want: StateDone, + }, + { + name: "single published returns published", + reactions: []*Reaction{ + {Reaction: ReactionPublished, AgentName: "agent-a"}, + }, + want: StatePublished, + }, + { + name: "all five types - published wins", + reactions: []*Reaction{ + {Reaction: ReactionApprove, AgentName: "agent-a"}, + {Reaction: ReactionInProgress, AgentName: "agent-b"}, + {Reaction: ReactionReject, AgentName: "agent-c"}, + {Reaction: ReactionDone, AgentName: "agent-d"}, + {Reaction: ReactionPublished, AgentName: "agent-e"}, + }, + want: StatePublished, + }, + { + name: "published wins over everything", + reactions: []*Reaction{ + {Reaction: ReactionDone, AgentName: "agent-a"}, + {Reaction: ReactionReject, AgentName: "agent-b"}, + {Reaction: ReactionPublished, AgentName: "agent-c"}, + }, + want: StatePublished, + }, + { + name: "reject beats in_progress", + reactions: []*Reaction{ + {Reaction: ReactionInProgress, AgentName: "agent-a"}, + {Reaction: ReactionReject, AgentName: "agent-b"}, + }, + want: StateRejected, + }, + { + name: "done beats reject", + reactions: []*Reaction{ + {Reaction: ReactionReject, AgentName: "agent-a"}, + {Reaction: ReactionDone, AgentName: "agent-b"}, + }, + want: StateDone, + }, + { + name: "in_progress beats approve", + reactions: []*Reaction{ + {Reaction: ReactionInProgress, AgentName: "agent-a"}, + {Reaction: ReactionApprove, AgentName: "agent-b"}, + }, + want: StateInProgress, + }, + { + name: "multiple approves still returns approved", + reactions: []*Reaction{ + {Reaction: ReactionApprove, AgentName: "agent-a"}, + {Reaction: ReactionApprove, AgentName: "agent-b"}, + {Reaction: ReactionApprove, AgentName: "agent-c"}, + }, + want: StateApproved, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := ComputeWorkflowState(tt.reactions) + if got != tt.want { + t.Errorf("ComputeWorkflowState() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/internal/reactions/service.go b/internal/reactions/service.go new file mode 100644 index 0000000..319bd30 --- /dev/null +++ b/internal/reactions/service.go @@ -0,0 +1,127 @@ +package reactions + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" +) + +// Service provides business logic for message reactions. +type Service struct { + store Store + logger *slog.Logger +} + +// NewService creates a new reaction service. +func NewService(store Store, logger *slog.Logger) *Service { + return &Service{ + store: store, + logger: logger.With("component", "reactions"), + } +} + +// ToggleResult describes what happened after a toggle operation. +type ToggleResult struct { + Action string `json:"action"` // "added" or "removed" + Reaction *Reaction `json:"reaction,omitempty"` +} + +// Toggle adds a reaction if it doesn't exist, or removes it if it does. +// Returns the action taken and the reaction (if added). +func (s *Service) Toggle(ctx context.Context, messageID int64, agentName, reactionType string, metadata json.RawMessage) (*ToggleResult, error) { + if !IsValidReaction(reactionType) { + return nil, ErrInvalidReaction + } + + // Check if reaction already exists + exists, err := s.store.Exists(ctx, messageID, agentName, reactionType) + if err != nil { + return nil, fmt.Errorf("check existing reaction: %w", err) + } + + if exists { + // Toggle off — remove it + if err := s.store.Delete(ctx, messageID, agentName, reactionType); err != nil { + return nil, fmt.Errorf("remove reaction: %w", err) + } + s.logger.Info("reaction removed", + "message_id", messageID, + "agent", agentName, + "reaction", reactionType, + ) + return &ToggleResult{Action: "removed"}, nil + } + + // Check reaction count limit + count, err := s.store.CountByMessage(ctx, messageID) + if err != nil { + return nil, fmt.Errorf("count reactions: %w", err) + } + if count >= MaxReactionsPerMessage { + return nil, ErrReactionLimit + } + + // Toggle on — add it + if metadata == nil { + metadata = json.RawMessage("{}") + } + + r := &Reaction{ + MessageID: messageID, + AgentName: agentName, + Reaction: reactionType, + Metadata: metadata, + } + + if err := s.store.Insert(ctx, r); err != nil { + return nil, fmt.Errorf("add reaction: %w", err) + } + + s.logger.Info("reaction added", + "message_id", messageID, + "agent", agentName, + "reaction", reactionType, + ) + + return &ToggleResult{Action: "added", Reaction: r}, nil +} + +// Remove explicitly removes a reaction. +func (s *Service) Remove(ctx context.Context, messageID int64, agentName, reactionType string) error { + if !IsValidReaction(reactionType) { + return ErrInvalidReaction + } + + if err := s.store.Delete(ctx, messageID, agentName, reactionType); err != nil { + return fmt.Errorf("remove reaction: %w", err) + } + + s.logger.Info("reaction removed", + "message_id", messageID, + "agent", agentName, + "reaction", reactionType, + ) + return nil +} + +// GetReactions returns all reactions for a message and the computed workflow state. +func (s *Service) GetReactions(ctx context.Context, messageID int64) ([]*Reaction, string, error) { + reactions, err := s.store.GetByMessageID(ctx, messageID) + if err != nil { + return nil, "", fmt.Errorf("get reactions: %w", err) + } + + state := ComputeWorkflowState(reactions) + return reactions, state, nil +} + +// GetReactionsByMessageIDs returns reactions grouped by message ID. +func (s *Service) GetReactionsByMessageIDs(ctx context.Context, messageIDs []int64) (map[int64][]*Reaction, error) { + return s.store.GetByMessageIDs(ctx, messageIDs) +} + +// ListByState returns message IDs in a channel that have the given workflow state. +func (s *Service) ListByState(ctx context.Context, channelID int64, state string) ([]int64, error) { + return s.store.GetMessageIDsByState(ctx, channelID, state) +} diff --git a/internal/reactions/store.go b/internal/reactions/store.go new file mode 100644 index 0000000..8a5be88 --- /dev/null +++ b/internal/reactions/store.go @@ -0,0 +1,209 @@ +package reactions + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "strings" +) + +// Store defines the storage interface for reactions. +type Store interface { + Insert(ctx context.Context, r *Reaction) error + Delete(ctx context.Context, messageID int64, agentName, reaction string) error + GetByMessageID(ctx context.Context, messageID int64) ([]*Reaction, error) + GetByMessageIDs(ctx context.Context, messageIDs []int64) (map[int64][]*Reaction, error) + Exists(ctx context.Context, messageID int64, agentName, reaction string) (bool, error) + CountByMessage(ctx context.Context, messageID int64) (int, error) + // GetMessageIDsByState returns message IDs in a channel that have the given workflow state. + GetMessageIDsByState(ctx context.Context, channelID int64, state string) ([]int64, error) +} + +// SQLiteStore implements Store using SQLite. +type SQLiteStore struct { + db *sql.DB +} + +// NewSQLiteStore creates a new SQLite-backed reaction store. +func NewSQLiteStore(db *sql.DB) *SQLiteStore { + return &SQLiteStore{db: db} +} + +func (s *SQLiteStore) Insert(ctx context.Context, r *Reaction) error { + metadata := r.Metadata + if metadata == nil { + metadata = json.RawMessage("{}") + } + + result, err := s.db.ExecContext(ctx, + `INSERT INTO message_reactions (message_id, agent_name, reaction, metadata, created_at) + VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)`, + r.MessageID, r.AgentName, r.Reaction, string(metadata), + ) + if err != nil { + return fmt.Errorf("insert reaction: %w", err) + } + id, err := result.LastInsertId() + if err != nil { + return fmt.Errorf("get reaction id: %w", err) + } + r.ID = id + return nil +} + +func (s *SQLiteStore) Delete(ctx context.Context, messageID int64, agentName, reaction string) error { + result, err := s.db.ExecContext(ctx, + `DELETE FROM message_reactions WHERE message_id = ? AND agent_name = ? AND reaction = ?`, + messageID, agentName, reaction, + ) + if err != nil { + return fmt.Errorf("delete reaction: %w", err) + } + n, _ := result.RowsAffected() + if n == 0 { + return fmt.Errorf("reaction not found") + } + return nil +} + +func (s *SQLiteStore) GetByMessageID(ctx context.Context, messageID int64) ([]*Reaction, error) { + rows, err := s.db.QueryContext(ctx, + `SELECT id, message_id, agent_name, reaction, metadata, created_at + FROM message_reactions WHERE message_id = ? + ORDER BY created_at ASC`, messageID, + ) + if err != nil { + return nil, fmt.Errorf("get reactions: %w", err) + } + defer rows.Close() + return scanReactions(rows) +} + +func (s *SQLiteStore) GetByMessageIDs(ctx context.Context, messageIDs []int64) (map[int64][]*Reaction, error) { + if len(messageIDs) == 0 { + return map[int64][]*Reaction{}, 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 id, message_id, agent_name, reaction, metadata, created_at + FROM message_reactions WHERE message_id IN (%s) + ORDER BY created_at ASC`, + strings.Join(placeholders, ","), + ) + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("get reactions by ids: %w", err) + } + defer rows.Close() + + all, err := scanReactions(rows) + if err != nil { + return nil, err + } + + result := make(map[int64][]*Reaction) + for _, r := range all { + result[r.MessageID] = append(result[r.MessageID], r) + } + return result, nil +} + +func (s *SQLiteStore) Exists(ctx context.Context, messageID int64, agentName, reaction string) (bool, error) { + var count int + err := s.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM message_reactions WHERE message_id = ? AND agent_name = ? AND reaction = ?`, + messageID, agentName, reaction, + ).Scan(&count) + if err != nil { + return false, fmt.Errorf("check reaction exists: %w", err) + } + return count > 0, nil +} + +func (s *SQLiteStore) CountByMessage(ctx context.Context, messageID int64) (int, error) { + var count int + err := s.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM message_reactions WHERE message_id = ?`, messageID, + ).Scan(&count) + if err != nil { + return 0, fmt.Errorf("count reactions: %w", err) + } + return count, nil +} + +func (s *SQLiteStore) GetMessageIDsByState(ctx context.Context, channelID int64, state string) ([]int64, error) { + var query string + var args []any + + if state == StateProposed { + // Messages with no reactions + query = `SELECT m.id FROM messages m + WHERE m.channel_id = ? + AND NOT EXISTS (SELECT 1 FROM message_reactions r WHERE r.message_id = m.id) + ORDER BY m.created_at DESC` + args = []any{channelID} + } else { + // Find the reaction type for this state + var reactionType string + for rt, st := range reactionToState { + if st == state { + reactionType = rt + break + } + } + if reactionType == "" { + return nil, fmt.Errorf("unknown workflow state: %s", state) + } + + // Messages where the highest-priority reaction maps to this state + // We get all messages with this reaction type and filter in app layer + query = `SELECT DISTINCT r.message_id FROM message_reactions r + JOIN messages m ON m.id = r.message_id + WHERE m.channel_id = ? AND r.reaction = ? + ORDER BY m.created_at DESC` + args = []any{channelID, reactionType} + } + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("get message ids by state: %w", err) + } + defer rows.Close() + + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("scan message id: %w", err) + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +func scanReactions(rows *sql.Rows) ([]*Reaction, error) { + var reactions []*Reaction + for rows.Next() { + var r Reaction + var metadata string + err := rows.Scan(&r.ID, &r.MessageID, &r.AgentName, &r.Reaction, &metadata, &r.CreatedAt) + if err != nil { + return nil, fmt.Errorf("scan reaction: %w", err) + } + r.Metadata = json.RawMessage(metadata) + reactions = append(reactions, &r) + } + if reactions == nil { + reactions = []*Reaction{} + } + return reactions, rows.Err() +} diff --git a/internal/reactions/store_test.go b/internal/reactions/store_test.go new file mode 100644 index 0000000..1b204af --- /dev/null +++ b/internal/reactions/store_test.go @@ -0,0 +1,332 @@ +package reactions + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "testing" + + _ "modernc.org/sqlite" + + "github.com/synapbus/synapbus/internal/storage" +) + +func newTestDB(t *testing.T) *sql.DB { + t.Helper() + dsn := fmt.Sprintf("file:%s?mode=memory&cache=shared", t.Name()) + db, err := sql.Open("sqlite", dsn) + if err != nil { + t.Fatalf("open database: %v", err) + } + t.Cleanup(func() { db.Close() }) + + if _, err := db.Exec("PRAGMA foreign_keys=ON"); err != nil { + t.Fatalf("enable foreign keys: %v", err) + } + + ctx := context.Background() + if err := storage.RunMigrations(ctx, db); err != nil { + t.Fatalf("run migrations: %v", err) + } + + return db +} + +// seedTestMessage creates a test user, agent, conversation, and message, +// returning the message ID. +func seedTestMessage(t *testing.T, db *sql.DB, agentName string) int64 { + t.Helper() + + // Ensure user exists + db.Exec(`INSERT OR IGNORE INTO users (id, username, password_hash, display_name) VALUES (1, 'testowner', 'hash', 'Test Owner')`) + + // Ensure agent exists + _, err := db.Exec( + `INSERT OR IGNORE INTO agents (name, display_name, type, capabilities, owner_id, api_key_hash, status) VALUES (?, ?, 'ai', '{}', 1, 'testhash', 'active')`, + agentName, agentName, + ) + if err != nil { + t.Fatalf("seed agent %s: %v", agentName, err) + } + + // Create conversation + result, err := db.Exec( + `INSERT INTO conversations (subject, created_by, created_at, updated_at) VALUES ('test', ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`, + agentName, + ) + if err != nil { + t.Fatalf("create conversation: %v", err) + } + convID, _ := result.LastInsertId() + + // Create message + result, err = db.Exec( + `INSERT INTO messages (conversation_id, from_agent, body, priority, status, created_at) VALUES (?, ?, 'test body', 5, 'pending', CURRENT_TIMESTAMP)`, + convID, agentName, + ) + if err != nil { + t.Fatalf("create message: %v", err) + } + msgID, _ := result.LastInsertId() + return msgID +} + +func TestSQLiteStore_InsertAndGet(t *testing.T) { + db := newTestDB(t) + store := NewSQLiteStore(db) + ctx := context.Background() + + msgID := seedTestMessage(t, db, "agent-a") + + r := &Reaction{ + MessageID: msgID, + AgentName: "agent-a", + Reaction: ReactionApprove, + Metadata: json.RawMessage(`{"comment":"looks good"}`), + } + + if err := store.Insert(ctx, r); err != nil { + t.Fatalf("Insert: %v", err) + } + + if r.ID == 0 { + t.Error("reaction ID should not be 0 after insert") + } + + // Verify it exists + exists, err := store.Exists(ctx, msgID, "agent-a", ReactionApprove) + if err != nil { + t.Fatalf("Exists: %v", err) + } + if !exists { + t.Error("expected reaction to exist after insert") + } + + // GetByMessageID + reactions, err := store.GetByMessageID(ctx, msgID) + if err != nil { + t.Fatalf("GetByMessageID: %v", err) + } + if len(reactions) != 1 { + t.Fatalf("got %d reactions, want 1", len(reactions)) + } + if reactions[0].AgentName != "agent-a" { + t.Errorf("AgentName = %q, want %q", reactions[0].AgentName, "agent-a") + } + if reactions[0].Reaction != ReactionApprove { + t.Errorf("Reaction = %q, want %q", reactions[0].Reaction, ReactionApprove) + } + if string(reactions[0].Metadata) != `{"comment":"looks good"}` { + t.Errorf("Metadata = %s, want %s", reactions[0].Metadata, `{"comment":"looks good"}`) + } +} + +func TestSQLiteStore_UniqueConstraint(t *testing.T) { + db := newTestDB(t) + store := NewSQLiteStore(db) + ctx := context.Background() + + msgID := seedTestMessage(t, db, "agent-a") + + r := &Reaction{ + MessageID: msgID, + AgentName: "agent-a", + Reaction: ReactionApprove, + } + + if err := store.Insert(ctx, r); err != nil { + t.Fatalf("Insert first: %v", err) + } + + // Inserting the same reaction again should fail with UNIQUE constraint + r2 := &Reaction{ + MessageID: msgID, + AgentName: "agent-a", + Reaction: ReactionApprove, + } + err := store.Insert(ctx, r2) + if err == nil { + t.Error("expected error on duplicate insert, got nil") + } +} + +func TestSQLiteStore_Delete(t *testing.T) { + db := newTestDB(t) + store := NewSQLiteStore(db) + ctx := context.Background() + + msgID := seedTestMessage(t, db, "agent-a") + + r := &Reaction{ + MessageID: msgID, + AgentName: "agent-a", + Reaction: ReactionApprove, + } + if err := store.Insert(ctx, r); err != nil { + t.Fatalf("Insert: %v", err) + } + + // Delete the reaction + if err := store.Delete(ctx, msgID, "agent-a", ReactionApprove); err != nil { + t.Fatalf("Delete: %v", err) + } + + // Verify it's gone + exists, err := store.Exists(ctx, msgID, "agent-a", ReactionApprove) + if err != nil { + t.Fatalf("Exists: %v", err) + } + if exists { + t.Error("expected reaction to not exist after delete") + } + + // Deleting a non-existent reaction should return an error + err = store.Delete(ctx, msgID, "agent-a", ReactionApprove) + if err == nil { + t.Error("expected error when deleting non-existent reaction, got nil") + } +} + +func TestSQLiteStore_GetByMessageID(t *testing.T) { + db := newTestDB(t) + store := NewSQLiteStore(db) + ctx := context.Background() + + msgID := seedTestMessage(t, db, "agent-a") + // Seed a second agent + db.Exec(`INSERT OR IGNORE INTO agents (name, display_name, type, capabilities, owner_id, api_key_hash, status) VALUES ('agent-b', 'agent-b', 'ai', '{}', 1, 'testhash2', 'active')`) + + // Insert multiple reactions from different agents + reactions := []*Reaction{ + {MessageID: msgID, AgentName: "agent-a", Reaction: ReactionApprove}, + {MessageID: msgID, AgentName: "agent-b", Reaction: ReactionInProgress}, + {MessageID: msgID, AgentName: "agent-a", Reaction: ReactionDone}, + } + + for _, r := range reactions { + if err := store.Insert(ctx, r); err != nil { + t.Fatalf("Insert: %v", err) + } + } + + got, err := store.GetByMessageID(ctx, msgID) + if err != nil { + t.Fatalf("GetByMessageID: %v", err) + } + + if len(got) != 3 { + t.Fatalf("got %d reactions, want 3", len(got)) + } + + // Verify results are ordered by created_at ASC + for i, r := range got { + if r.ID == 0 { + t.Errorf("reaction[%d] ID should not be 0", i) + } + if r.MessageID != msgID { + t.Errorf("reaction[%d] MessageID = %d, want %d", i, r.MessageID, msgID) + } + } + + // Test with a message that has no reactions + msgID2 := seedTestMessage(t, db, "agent-a") + got2, err := store.GetByMessageID(ctx, msgID2) + if err != nil { + t.Fatalf("GetByMessageID (empty): %v", err) + } + if len(got2) != 0 { + t.Errorf("got %d reactions for empty message, want 0", len(got2)) + } +} + +func TestSQLiteStore_CountByMessage(t *testing.T) { + db := newTestDB(t) + store := NewSQLiteStore(db) + ctx := context.Background() + + msgID := seedTestMessage(t, db, "agent-a") + db.Exec(`INSERT OR IGNORE INTO agents (name, display_name, type, capabilities, owner_id, api_key_hash, status) VALUES ('agent-b', 'agent-b', 'ai', '{}', 1, 'testhash2', 'active')`) + + // Count should be 0 initially + count, err := store.CountByMessage(ctx, msgID) + if err != nil { + t.Fatalf("CountByMessage: %v", err) + } + if count != 0 { + t.Errorf("initial count = %d, want 0", count) + } + + // Insert some reactions + for _, r := range []*Reaction{ + {MessageID: msgID, AgentName: "agent-a", Reaction: ReactionApprove}, + {MessageID: msgID, AgentName: "agent-b", Reaction: ReactionDone}, + } { + if err := store.Insert(ctx, r); err != nil { + t.Fatalf("Insert: %v", err) + } + } + + count, err = store.CountByMessage(ctx, msgID) + if err != nil { + t.Fatalf("CountByMessage: %v", err) + } + if count != 2 { + t.Errorf("count = %d, want 2", count) + } + + // Delete one and verify count decreases + if err := store.Delete(ctx, msgID, "agent-a", ReactionApprove); err != nil { + t.Fatalf("Delete: %v", err) + } + count, err = store.CountByMessage(ctx, msgID) + if err != nil { + t.Fatalf("CountByMessage after delete: %v", err) + } + if count != 1 { + t.Errorf("count after delete = %d, want 1", count) + } +} + +func TestSQLiteStore_GetByMessageIDs(t *testing.T) { + db := newTestDB(t) + store := NewSQLiteStore(db) + ctx := context.Background() + + msgID1 := seedTestMessage(t, db, "agent-a") + msgID2 := seedTestMessage(t, db, "agent-a") + + // Add reactions to msg1 + if err := store.Insert(ctx, &Reaction{MessageID: msgID1, AgentName: "agent-a", Reaction: ReactionApprove}); err != nil { + t.Fatalf("Insert: %v", err) + } + if err := store.Insert(ctx, &Reaction{MessageID: msgID1, AgentName: "agent-a", Reaction: ReactionDone}); err != nil { + t.Fatalf("Insert: %v", err) + } + + // Add reaction to msg2 + if err := store.Insert(ctx, &Reaction{MessageID: msgID2, AgentName: "agent-a", Reaction: ReactionReject}); err != nil { + t.Fatalf("Insert: %v", err) + } + + result, err := store.GetByMessageIDs(ctx, []int64{msgID1, msgID2}) + if err != nil { + t.Fatalf("GetByMessageIDs: %v", err) + } + + if len(result[msgID1]) != 2 { + t.Errorf("msg1 reactions = %d, want 2", len(result[msgID1])) + } + if len(result[msgID2]) != 1 { + t.Errorf("msg2 reactions = %d, want 1", len(result[msgID2])) + } + + // Empty slice returns empty map + result, err = store.GetByMessageIDs(ctx, []int64{}) + if err != nil { + t.Fatalf("GetByMessageIDs (empty): %v", err) + } + if len(result) != 0 { + t.Errorf("expected empty map, got %v", result) + } +} diff --git a/internal/storage/schema/013_reactions.sql b/internal/storage/schema/013_reactions.sql new file mode 100644 index 0000000..19f234d --- /dev/null +++ b/internal/storage/schema/013_reactions.sql @@ -0,0 +1,21 @@ +-- Message reactions for workflow state tracking +-- Supports: approve, reject, in_progress, done, published + +CREATE TABLE IF NOT EXISTS message_reactions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE, + agent_name TEXT NOT NULL, + reaction TEXT NOT NULL CHECK(reaction IN ('approve', 'reject', 'in_progress', 'done', 'published')), + metadata TEXT NOT NULL DEFAULT '{}', + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(message_id, agent_name, reaction) +); + +CREATE INDEX idx_reactions_message ON message_reactions(message_id); +CREATE INDEX idx_reactions_agent ON message_reactions(agent_name); +CREATE INDEX idx_reactions_type ON message_reactions(reaction); + +-- Channel workflow settings +ALTER TABLE channels ADD COLUMN auto_approve BOOLEAN NOT NULL DEFAULT 0; +ALTER TABLE channels ADD COLUMN stalemate_remind_after TEXT NOT NULL DEFAULT '24h'; +ALTER TABLE channels ADD COLUMN stalemate_escalate_after TEXT NOT NULL DEFAULT '72h'; diff --git a/internal/web/dist/index.html b/internal/web/dist/index.html index f4c9dcf..37dad4b 100644 --- a/internal/web/dist/index.html +++ b/internal/web/dist/index.html @@ -11,30 +11,30 @@ - - + + - - - - - - + + + + + +
+ +{#if grouped.length > 0} +
+ {#each grouped as group} + + {/each} +
+{/if} diff --git a/web/src/lib/components/WorkflowBadge.svelte b/web/src/lib/components/WorkflowBadge.svelte new file mode 100644 index 0000000..285ae24 --- /dev/null +++ b/web/src/lib/components/WorkflowBadge.svelte @@ -0,0 +1,26 @@ + + + + {state} + diff --git a/web/src/routes/channels/[name]/+page.svelte b/web/src/routes/channels/[name]/+page.svelte index ab2125b..fde4d16 100644 --- a/web/src/routes/channels/[name]/+page.svelte +++ b/web/src/routes/channels/[name]/+page.svelte @@ -5,6 +5,8 @@ import { notifications } from '$lib/stores/notifications'; import MessageBody from '$lib/components/MessageBody.svelte'; import AttachmentPreview from '$lib/components/AttachmentPreview.svelte'; + import WorkflowBadge from '$lib/components/WorkflowBadge.svelte'; + import ReactionPills from '$lib/components/ReactionPills.svelte'; let channel = $state(null); let members = $state([]); @@ -306,6 +308,10 @@ {formatTime(msg.created_at)}
+ {#if msg.workflow_state} + + {/if} + {#if msg.attachments?.length > 0}
{#each msg.attachments as att}