fix: bugs #1-3-5 from #bugs-synapbus + live SSE notifications

- Auto-join public channels on first send (bug #1)
- Channel broadcasts no longer create duplicate DM copies; inbox DMs
  only sent for @mentions (bug #2)
- Embedding pipeline auto-enqueues new messages via MessageListener
  callback instead of requiring pod restart (bug #3)
- Admin socket defaults to /tmp in containers to avoid PVC filesystem
  incompatibility with Unix sockets (bug #5)
- SSE events now fire for MCP-sent messages (not just REST API),
  enabling live notification badges without page reload
- Fixed frontend SSE field name mismatch (channel_name → channel)
- Fixed SSE client not connecting after login redirect

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

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"]

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)
 }

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")
 	}
 }
 

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)

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 }}

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)
+	}
+}

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)
 }

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) {

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,

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")
 			}
 		}
 	})

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,

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")
 	}
 }
 

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();
 		}
 	});