diff --git a/Dockerfile b/Dockerfile index 9b09a9b..9c8760a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -19,7 +19,7 @@ RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w -X main.version=${VERSION} # Stage 3: Runtime FROM alpine:3.19 -RUN apk add --no-cache ca-certificates tzdata +RUN apk add --no-cache ca-certificates tzdata && touch /.dockerenv COPY --from=go-builder /synapbus /synapbus EXPOSE 8080 VOLUME ["/data"] diff --git a/cmd/synapbus/admin.go b/cmd/synapbus/admin.go index c191cea..cec53d8 100644 --- a/cmd/synapbus/admin.go +++ b/cmd/synapbus/admin.go @@ -17,7 +17,7 @@ var adminSocket string // adminRequest sends a command over the Unix socket and returns the parsed response. func adminRequest(command string, args interface{}) (map[string]interface{}, error) { socket := adminSocket - if s := os.Getenv("SYNAPBUS_SOCKET"); s != "" && socket == "/data/synapbus.sock" { + if s := os.Getenv("SYNAPBUS_SOCKET"); s != "" && socket == "/tmp/synapbus.sock" { socket = s } @@ -966,7 +966,7 @@ func addAdminCommands(rootCmd *cobra.Command) { attachmentsCmd.AddCommand(attachmentsGCCmd) // ----- add persistent flag and commands to root ----- - rootCmd.PersistentFlags().StringVar(&adminSocket, "socket", "/data/synapbus.sock", "Path to admin Unix socket") + rootCmd.PersistentFlags().StringVar(&adminSocket, "socket", "/tmp/synapbus.sock", "Path to admin Unix socket") rootCmd.AddCommand(userCmd, agentCmd, auditCmd, backupCmd, messagesCmd, channelsCmd, conversationsCmd, embeddingsCmd, dbCmd, retentionCmd, webhookCmd, k8sCmd, attachmentsCmd) } diff --git a/cmd/synapbus/admin_test.go b/cmd/synapbus/admin_test.go index ecd5f27..3873d9e 100644 --- a/cmd/synapbus/admin_test.go +++ b/cmd/synapbus/admin_test.go @@ -270,8 +270,8 @@ func TestDefaultSocketPath(t *testing.T) { if f == nil { t.Fatal("--socket persistent flag not found") } - if f.DefValue != "/data/synapbus.sock" { - t.Errorf("default socket path = %q, want %q", f.DefValue, "/data/synapbus.sock") + if f.DefValue != "/tmp/synapbus.sock" { + t.Errorf("default socket path = %q, want %q", f.DefValue, "/tmp/synapbus.sock") } } diff --git a/cmd/synapbus/main.go b/cmd/synapbus/main.go index 061d41a..8bda60c 100644 --- a/cmd/synapbus/main.go +++ b/cmd/synapbus/main.go @@ -161,7 +161,13 @@ func runServe(cmd *cobra.Command, args []string) error { messageRetention = mr } if adminSocketPath == "" { - adminSocketPath = filepath.Join(dataDir, "synapbus.sock") + // Default to /tmp in containers — PVC-backed filesystems (NFS, Ceph, + // EBS CSI) often don't support Unix domain sockets. + if _, err := os.Stat("/.dockerenv"); err == nil { + adminSocketPath = "/tmp/synapbus.sock" + } else { + adminSocketPath = filepath.Join(dataDir, "synapbus.sock") + } } // Configure slog with JSON handler writing to stderr (stdout is for console output) @@ -389,6 +395,9 @@ func runServe(cmd *cobra.Command, args []string) error { embPipeline = search.NewPipeline(embProvider, embStore, vectorIndex, searchCfg) embPipeline.Start(ctx) + // Wire pipeline into messaging so new messages auto-enqueue + msgService.SetEmbeddingNotifier(embPipeline) + // Create search service with semantic support searchService = search.NewService(db.DB, embProvider, vectorIndex, msgService) slog.Info("semantic search enabled", @@ -529,8 +538,13 @@ func runServe(cmd *cobra.Command, args []string) error { r.Mount("/mcp", mcpSrv.Handler()) }) - // Create SSE hub for real-time events + // Create SSE hub and broadcaster for real-time events sseHub := api.NewSSEHub() + sseBroadcaster := api.NewSSEBroadcaster(sseHub, agentService, channelService) + + // Register broadcaster as a message listener so SSE events fire + // for messages sent via MCP (agents) as well as the REST API. + msgService.AddMessageListener(sseBroadcaster) // Mount API routes (traces, export, stats, metrics, attachments, messages, agents, channels, SSE) sessionMiddleware := api.SessionToOwnerMiddleware(userStore, sessionStore) @@ -544,6 +558,7 @@ func runServe(cmd *cobra.Command, args []string) error { APIKeyService: apiKeyService, DeadLetterStore: deadLetterStore, SSEHub: sseHub, + Broadcaster: sseBroadcaster, SessionMiddleware: sessionMiddleware, }) r.Mount("/", apiRouter) diff --git a/deploy/helm/synapbus/templates/deployment.yaml b/deploy/helm/synapbus/templates/deployment.yaml index 8042cc2..2965a71 100644 --- a/deploy/helm/synapbus/templates/deployment.yaml +++ b/deploy/helm/synapbus/templates/deployment.yaml @@ -60,6 +60,8 @@ spec: volumeMounts: - name: data mountPath: /data + - name: run + mountPath: /tmp volumes: - name: data {{- if .Values.persistence.enabled }} @@ -68,6 +70,10 @@ spec: {{- else }} emptyDir: {} {{- end }} + - name: run + emptyDir: + medium: Memory + sizeLimit: 1Mi {{- with .Values.nodeSelector }} nodeSelector: {{- toYaml . | nindent 8 }} diff --git a/internal/api/broadcaster.go b/internal/api/broadcaster.go index 754e0b9..683015f 100644 --- a/internal/api/broadcaster.go +++ b/internal/api/broadcaster.go @@ -6,6 +6,7 @@ import ( "github.com/synapbus/synapbus/internal/agents" "github.com/synapbus/synapbus/internal/channels" + "github.com/synapbus/synapbus/internal/messaging" ) // NewMessageEvent is broadcast when a new message is sent. @@ -107,3 +108,24 @@ func (b *SSEBroadcaster) BroadcastChannelMessage(ctx context.Context, channelID } } } + +// OnMessageSent implements messaging.MessageListener so the SSEBroadcaster +// can be wired directly into the messaging service. This ensures SSE events +// fire for messages sent via MCP (agents) as well as the REST API. +func (b *SSEBroadcaster) OnMessageSent(ctx context.Context, msg *messaging.Message) { + event := NewMessageEvent{ + MessageID: msg.ID, + FromAgent: msg.FromAgent, + ToAgent: msg.ToAgent, + } + + if msg.ChannelID != nil { + ch, err := b.channelService.GetChannel(ctx, *msg.ChannelID) + if err == nil { + event.Channel = ch.Name + } + b.BroadcastChannelMessage(ctx, *msg.ChannelID, event) + } else { + b.BroadcastDM(ctx, event) + } +} diff --git a/internal/api/messages_handler.go b/internal/api/messages_handler.go index 96172f1..4418311 100644 --- a/internal/api/messages_handler.go +++ b/internal/api/messages_handler.go @@ -315,23 +315,8 @@ func (h *MessagesHandler) SendMessage(w http.ResponseWriter, r *http.Request) { return } - // Broadcast real-time event to connected SSE clients - if h.broadcaster != nil { - event := NewMessageEvent{ - MessageID: msg.ID, - FromAgent: msg.FromAgent, - ToAgent: msg.ToAgent, - } - if msg.ChannelID != nil && h.broadcaster.channelService != nil { - ch, chErr := h.broadcaster.channelService.GetChannel(r.Context(), *msg.ChannelID) - if chErr == nil { - event.Channel = ch.Name - } - h.broadcaster.BroadcastChannelMessage(r.Context(), *msg.ChannelID, event) - } else { - h.broadcaster.BroadcastDM(r.Context(), event) - } - } + // SSE broadcast is handled by the MessageListener on the messaging + // service, so it fires for both REST and MCP message paths. writeJSON(w, http.StatusCreated, msg) } diff --git a/internal/api/router.go b/internal/api/router.go index d574d13..32c0d9e 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -31,6 +31,7 @@ type RouterConfig struct { K8sService *k8s.K8sService K8sStore k8s.K8sStore SSEHub *SSEHub + Broadcaster *SSEBroadcaster SessionMiddleware func(http.Handler) http.Handler } @@ -88,9 +89,10 @@ func NewRouterWithConfig(cfg RouterConfig) chi.Router { notificationsHandler := NewNotificationsHandler(cfg.MsgService, cfg.AgentService, cfg.ChannelService) // Wire up SSE broadcaster for real-time events - if cfg.SSEHub != nil { - broadcaster := NewSSEBroadcaster(cfg.SSEHub, cfg.AgentService, cfg.ChannelService) - messagesHandler.SetBroadcaster(broadcaster) + if cfg.Broadcaster != nil { + messagesHandler.SetBroadcaster(cfg.Broadcaster) + } else if cfg.SSEHub != nil { + messagesHandler.SetBroadcaster(NewSSEBroadcaster(cfg.SSEHub, cfg.AgentService, cfg.ChannelService)) } r.Group(func(r chi.Router) { diff --git a/internal/channels/service.go b/internal/channels/service.go index 967bb75..74ab763 100644 --- a/internal/channels/service.go +++ b/internal/channels/service.go @@ -471,13 +471,18 @@ func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAge return nil, err } - // Verify sender is a member + // Verify sender is a member; auto-join public channels on first send. isMember, err := s.store.IsMember(ctx, channelID, fromAgent) if err != nil { return nil, fmt.Errorf("check membership: %w", err) } if !isMember { - return nil, ErrNotChannelMember + if ch.IsPrivate { + return nil, ErrNotChannelMember + } + if err := s.JoinChannel(ctx, channelID, fromAgent); err != nil { + return nil, fmt.Errorf("auto-join public channel: %w", err) + } } // Get members for mentions and inbox notifications @@ -526,10 +531,10 @@ func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAge return nil, fmt.Errorf("create channel message: %w", err) } - // 2. Deliver inbox notifications to other members. + // 2. Deliver inbox notifications only to @mentioned members. recipientCount := 0 for _, m := range members { - if m.AgentName == fromAgent { + if m.AgentName == fromAgent || !mentionedMembers[m.AgentName] { continue } @@ -537,12 +542,7 @@ func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAge "channel_id": channelID, "channel_name": ch.Name, "channel_message_id": channelMsg.ID, - } - if len(mentionedAgentsList) > 0 { - inboxMetaObj["mentioned_agents"] = mentionedAgentsList - } - if mentionedMembers[m.AgentName] { - inboxMetaObj["mention"] = true + "mention": true, } inboxMetaBytes, _ := json.Marshal(inboxMetaObj) @@ -552,7 +552,7 @@ func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAge Metadata: string(inboxMetaBytes), }) if err != nil { - s.logger.Error("failed to send channel notification", + s.logger.Error("failed to send mention notification", "channel_id", channelID, "from", fromAgent, "to", m.AgentName, diff --git a/internal/channels/service_test.go b/internal/channels/service_test.go index 9c80bb0..ca1571e 100644 --- a/internal/channels/service_test.go +++ b/internal/channels/service_test.go @@ -565,26 +565,21 @@ func TestService_BroadcastMessage(t *testing.T) { } }) - t.Run("broadcast delivers inbox notifications", func(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", "multi-member test", 5, "") + _, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no-dm-test", 5, "") if err != nil { t.Fatalf("BroadcastMessage: %v", err) } - // agent-b should have an inbox notification + // agent-b should NOT have an inbox notification (no @mention) inboxResult, _ := svc.msgService.ReadInbox(ctx, "agent-b", messaging.ReadOptions{IncludeRead: true}) - found := false for _, m := range inboxResult.Messages { - if m.Body == "multi-member test" { - found = true - break + if m.Body == "no-dm-test" { + t.Error("agent-b should not receive inbox DM for non-mention broadcast") } } - if !found { - t.Error("agent-b did not receive inbox notification") - } }) t.Run("sender does not receive own message", func(t *testing.T) { @@ -597,12 +592,29 @@ func TestService_BroadcastMessage(t *testing.T) { } }) - t.Run("non-member cannot broadcast", func(t *testing.T) { - // agent-c is a member but let's test someone who isn'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", "unauthorized", 5, "") + _, err := svc.BroadcastMessage(ctx, ch.ID, "outsider", "auto-joined", 5, "") + if err != nil { + t.Fatalf("expected auto-join for public channel, got %v", err) + } + isMember, _ := svc.IsMember(ctx, ch.ID, "outsider") + if !isMember { + t.Error("outsider should be a member after auto-join") + } + }) + + t.Run("non-member cannot broadcast to private channel", func(t *testing.T) { + privCh, err := svc.CreateChannel(ctx, CreateChannelRequest{ + Name: "private-test", Type: TypeStandard, IsPrivate: true, CreatedBy: "agent-a", + }) + if err != nil { + t.Fatalf("create private channel: %v", err) + } + seedAgent(t, svc.store.(*SQLiteChannelStore).db, "outsider2") + _, err = svc.BroadcastMessage(ctx, privCh.ID, "outsider2", "unauthorized", 5, "") if !errors.Is(err, ErrNotChannelMember) { - t.Errorf("expected ErrNotChannelMember, got %v", err) + t.Errorf("expected ErrNotChannelMember for private channel, got %v", err) } }) @@ -642,20 +654,11 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) { t.Error("agent-b did not receive inbox notification") } - // agent-c was NOT mentioned — inbox notification should NOT have mention:true + // agent-c was NOT mentioned — should NOT receive an inbox DM at all inboxResult, _ = svc.msgService.ReadInbox(ctx, "agent-c", messaging.ReadOptions{IncludeRead: true}) for _, m := range inboxResult.Messages { if m.Body == "hey @agent-b check this" { - var meta map[string]any - json.Unmarshal(m.Metadata, &meta) - if meta["mention"] == true { - t.Error("agent-c should NOT have mention flag") - } - // But should still have mentioned_agents list - if _, ok := meta["mentioned_agents"]; !ok { - t.Error("agent-c metadata should have mentioned_agents list") - } - break + t.Error("agent-c should not receive inbox DM when not @mentioned") } } }) diff --git a/internal/messaging/service.go b/internal/messaging/service.go index 3de44d6..65f8cf2 100644 --- a/internal/messaging/service.go +++ b/internal/messaging/service.go @@ -12,11 +12,25 @@ import ( "github.com/synapbus/synapbus/internal/trace" ) +// EmbeddingNotifier is called when messages are created or deleted so the +// embedding pipeline can enqueue them without a direct import dependency. +type EmbeddingNotifier interface { + OnMessageCreated(ctx context.Context, messageID int64, body string) +} + +// MessageListener is notified after every message is persisted. +// Implementations must not block — use goroutines for slow work. +type MessageListener interface { + OnMessageSent(ctx context.Context, msg *Message) +} + // MessagingService provides business logic for messaging operations. type MessagingService struct { store MessageStore tracer *trace.Tracer dispatcher dispatcher.EventDispatcher + embeddings EmbeddingNotifier + listeners []MessageListener logger *slog.Logger } @@ -34,6 +48,16 @@ func (s *MessagingService) SetDispatcher(d dispatcher.EventDispatcher) { s.dispatcher = d } +// SetEmbeddingNotifier sets the embedding pipeline callback for new messages. +func (s *MessagingService) SetEmbeddingNotifier(n EmbeddingNotifier) { + s.embeddings = n +} + +// AddMessageListener registers a listener that is notified after message creation. +func (s *MessagingService) AddMessageListener(l MessageListener) { + s.listeners = append(s.listeners, l) +} + // SendMessage creates a message, auto-creating conversations as needed. func (s *MessagingService) SendMessage(ctx context.Context, from, to, body string, opts SendOptions) (*Message, error) { // Validate inputs @@ -121,6 +145,16 @@ func (s *MessagingService) SendMessage(ctx context.Context, from, to, body strin return nil, fmt.Errorf("insert message: %w", err) } + // Enqueue for embedding (async, best-effort) + if s.embeddings != nil { + s.embeddings.OnMessageCreated(ctx, msg.ID, msg.Body) + } + + // Notify listeners (SSE, etc.) + for _, l := range s.listeners { + l.OnMessageSent(ctx, msg) + } + s.logger.Info("message sent", "from", from, "to", to, diff --git a/tests/integration/mcp_e2e_test.go b/tests/integration/mcp_e2e_test.go index 3e8c793..1f044b0 100644 --- a/tests/integration/mcp_e2e_test.go +++ b/tests/integration/mcp_e2e_test.go @@ -480,17 +480,17 @@ func TestE2E_ChannelMessaging(t *testing.T) { t.Errorf("send status = %v, want sent", sendResult["status"]) } - // Bob reads his inbox and should see the channel message. - inbox := bobClient.CallTool("execute", map[string]any{ - "code": `call("read_inbox", { include_read: true })`, + // Bob reads channel messages (DMs only sent for @mentions). + chMsgs := bobClient.CallTool("execute", map[string]any{ + "code": `call("get_channel_messages", { channel_name: "project-x", limit: 10 })`, }) - inboxData := unwrapCallResult(t, inbox) - count := inboxData["count"].(float64) + chData := unwrapCallResult(t, chMsgs) + count := chData["count"].(float64) if count < 1 { - t.Fatalf("Bob's inbox count = %v, want >= 1", count) + t.Fatalf("channel message count = %v, want >= 1", count) } - messages := inboxData["messages"].([]any) + messages := chData["messages"].([]any) found := false for _, m := range messages { msg := m.(map[string]any) @@ -500,7 +500,7 @@ func TestE2E_ChannelMessaging(t *testing.T) { } } if !found { - t.Error("Bob did not receive the channel message") + t.Error("channel message not found via get_channel_messages") } } diff --git a/web/src/routes/+layout.svelte b/web/src/routes/+layout.svelte index 2653bbf..e913e13 100644 --- a/web/src/routes/+layout.svelte +++ b/web/src/routes/+layout.svelte @@ -21,8 +21,8 @@ return client.onEvent((event) => { if (event.type === 'new_message') { const d = event.data; - if (d.channel_name) { - notifications.incrementUnread('channel', d.channel_name); + if (d.channel) { + notifications.incrementUnread('channel', d.channel); } else if (d.from_agent) { notifications.incrementUnread('dm', d.from_agent); } @@ -50,7 +50,11 @@ }); $effect(() => { - if ($user && sseClient && !sseClient.connected) { + if ($user && !sseClient) { + sseClient = new SSEClient(); + sseClient.connect(); + sseUnsubscribe = setupNotifications(sseClient); + } else if ($user && sseClient && !sseClient.connected) { sseClient.connect(); } });