feat: embeddings management, message retention & agent inbox improvements

- Add `my_status` MCP tool: single call returns agent identity, pending
  DMs, channel mentions, system notifications, channel summaries, and
  stats with truncation for large inboxes
- Add embeddings CLI: `synapbus embeddings status|reindex|clear` for
  managing vectors when switching embedding providers
- Add automatic message retention worker with configurable period
  (--message-retention, default 12m), warning notifications 1 month
  before deletion, cascade cleanup, and incremental vacuum
- Add manual purge: `synapbus messages purge --older-than --agent --channel`
  and `synapbus db vacuum` for on-demand cleanup
- Add `synapbus retention status` CLI for admin visibility
- Create system agent at startup for sending retention warnings
- Filter system agent from discover_agents results

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

CLAUDE.md

@@ -69,6 +69,7 @@ make lint           # Run linters
 | `OPENAI_API_KEY` | OpenAI API key for embeddings | (none) |
 | `GEMINI_API_KEY` | Google Gemini API key for embeddings | (none) |
 | `SYNAPBUS_OLLAMA_URL` | Ollama server URL | `http://localhost:11434` |
+| `SYNAPBUS_MESSAGE_RETENTION` | Message retention period (e.g. `12m`, `365d`, `0` to disable) | `12m` |
 
 ## Conventions
 
@@ -95,6 +96,8 @@ make lint           # Run linters
 - modernc.org/sqlite (pure Go), TFMV/hnsw (vectors) (002-mcp-auth-ux-polish)
 - Go 1.25+ (from go.mod) + mark3labs/mcp-go (MCP tools), go-chi/chi (HTTP), golang.org/x/time/rate (rate limiting), k8s.io/client-go (K8s Jobs — optional) (003-webhooks-k8s-runner)
 - modernc.org/sqlite (pure Go), migration 009_webhooks.sql (003-webhooks-k8s-runner)
+- Go 1.25+ (per go.mod) + mark3labs/mcp-go (MCP tools), go-chi/chi (HTTP), spf13/cobra (CLI), modernc.org/sqlite (storage), TFMV/hnsw (vectors) (004-embeddings-retention-inbox)
+- SQLite (modernc.org/sqlite, pure Go) — single DB file in `--data` directory (004-embeddings-retention-inbox)
 
 ## 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)

cmd/synapbus/admin.go

@@ -480,7 +480,41 @@ func addAdminCommands(rootCmd *cobra.Command) {
 	messagesSearchCmd.Flags().IntVar(&messagesSearchLimit, "limit", 20, "Max results")
 	messagesSearchCmd.MarkFlagRequired("query")
 
-	messagesCmd.AddCommand(messagesListCmd, messagesSearchCmd)
+	var (
+		messagesPurgeOlderThan string
+		messagesPurgeAgent     string
+		messagesPurgeChannel   string
+	)
+	messagesPurgeCmd := &cobra.Command{
+		Use:   "purge",
+		Short: "Delete messages matching filters (at least one filter required)",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			reqArgs := map[string]interface{}{}
+			if messagesPurgeOlderThan != "" {
+				reqArgs["older_than"] = messagesPurgeOlderThan
+			}
+			if messagesPurgeAgent != "" {
+				reqArgs["agent"] = messagesPurgeAgent
+			}
+			if messagesPurgeChannel != "" {
+				reqArgs["channel"] = messagesPurgeChannel
+			}
+			if len(reqArgs) == 0 {
+				return fmt.Errorf("at least one filter is required (--older-than, --agent, or --channel)")
+			}
+			resp, err := adminRequest("messages.purge", reqArgs)
+			if err != nil {
+				return err
+			}
+			printJSON(resp["data"])
+			return nil
+		},
+	}
+	messagesPurgeCmd.Flags().StringVar(&messagesPurgeOlderThan, "older-than", "", "Delete messages older than this (e.g. 6m, 90d, 2160h)")
+	messagesPurgeCmd.Flags().StringVar(&messagesPurgeAgent, "agent", "", "Delete messages from/to this agent")
+	messagesPurgeCmd.Flags().StringVar(&messagesPurgeChannel, "channel", "", "Delete messages in this channel")
+
+	messagesCmd.AddCommand(messagesListCmd, messagesSearchCmd, messagesPurgeCmd)
 
 	// ----- channels commands -----
 	channelsCmd := &cobra.Command{
@@ -582,10 +616,99 @@ func addAdminCommands(rootCmd *cobra.Command) {
 
 	conversationsCmd.AddCommand(conversationsListCmd, conversationsShowCmd)
 
+	// ----- embeddings commands -----
+	embeddingsCmd := &cobra.Command{
+		Use:   "embeddings",
+		Short: "Manage embedding vectors",
+	}
+
+	embeddingsStatusCmd := &cobra.Command{
+		Use:   "status",
+		Short: "Show embedding status (provider, counts, index size)",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			resp, err := adminRequest("embeddings.status", nil)
+			if err != nil {
+				return err
+			}
+			printJSON(resp["data"])
+			return nil
+		},
+	}
+
+	embeddingsReindexCmd := &cobra.Command{
+		Use:   "reindex",
+		Short: "Clear all embeddings and re-queue all messages for embedding",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			resp, err := adminRequest("embeddings.reindex", nil)
+			if err != nil {
+				return err
+			}
+			printJSON(resp["data"])
+			return nil
+		},
+	}
+
+	embeddingsClearCmd := &cobra.Command{
+		Use:   "clear",
+		Short: "Delete all embeddings and clear the vector index",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			resp, err := adminRequest("embeddings.clear", nil)
+			if err != nil {
+				return err
+			}
+			printJSON(resp["data"])
+			return nil
+		},
+	}
+
+	embeddingsCmd.AddCommand(embeddingsStatusCmd, embeddingsReindexCmd, embeddingsClearCmd)
+
+	// ----- db commands -----
+	dbCmd := &cobra.Command{
+		Use:   "db",
+		Short: "Database maintenance",
+	}
+
+	dbVacuumCmd := &cobra.Command{
+		Use:   "vacuum",
+		Short: "Compact the database to reclaim disk space",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			resp, err := adminRequest("db.vacuum", nil)
+			if err != nil {
+				return err
+			}
+			printJSON(resp["data"])
+			return nil
+		},
+	}
+
+	dbCmd.AddCommand(dbVacuumCmd)
+
+	// ----- retention commands -----
+	retentionCmd := &cobra.Command{
+		Use:   "retention",
+		Short: "Message retention management",
+	}
+
+	retentionStatusCmd := &cobra.Command{
+		Use:   "status",
+		Short: "Show retention configuration and status",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			resp, err := adminRequest("retention.status", nil)
+			if err != nil {
+				return err
+			}
+			printJSON(resp["data"])
+			return nil
+		},
+	}
+
+	retentionCmd.AddCommand(retentionStatusCmd)
+
 	// ----- add persistent flag and commands to root -----
 	rootCmd.PersistentFlags().StringVar(&adminSocket, "socket", "./data/synapbus.sock", "Path to admin Unix socket")
 
-	rootCmd.AddCommand(userCmd, agentCmd, auditCmd, backupCmd, messagesCmd, channelsCmd, conversationsCmd)
+	rootCmd.AddCommand(userCmd, agentCmd, auditCmd, backupCmd, messagesCmd, channelsCmd, conversationsCmd, embeddingsCmd, dbCmd, retentionCmd)
 }
 
 // toTableRows remaps []map[string]string using a header->key mapping.

cmd/synapbus/main.go

@@ -57,6 +57,7 @@ var (
 	traceRetention    string
 	adminSocketPath   string
 	webhookWorkers    int
+	messageRetention  string
 )
 
 func main() {
@@ -81,6 +82,7 @@ func main() {
 	serveCmd.Flags().StringVar(&traceRetention, "trace-retention", "0", "Trace retention period (e.g. 30d, 90d, 0 for unlimited)")
 	serveCmd.Flags().StringVar(&adminSocketPath, "admin-socket", "", "Admin Unix socket path (default: {data}/synapbus.sock)")
 	serveCmd.Flags().IntVar(&webhookWorkers, "webhook-workers", 8, "Number of webhook delivery worker goroutines")
+	serveCmd.Flags().StringVar(&messageRetention, "message-retention", "12m", "Message retention period (e.g. 12m, 365d, 0 to disable)")
 
 	rootCmd.AddCommand(serveCmd)
 
@@ -153,6 +155,9 @@ func runServe(cmd *cobra.Command, args []string) error {
 	if ww := os.Getenv("SYNAPBUS_WEBHOOK_WORKERS"); ww != "" {
 		fmt.Sscanf(ww, "%d", &webhookWorkers)
 	}
+	if mr := os.Getenv("SYNAPBUS_MESSAGE_RETENTION"); mr != "" {
+		messageRetention = mr
+	}
 	if adminSocketPath == "" {
 		adminSocketPath = filepath.Join(dataDir, "synapbus.sock")
 	}
@@ -229,6 +234,11 @@ func runServe(cmd *cobra.Command, args []string) error {
 	deadLetterStore := messaging.NewDeadLetterStore(db.DB)
 	agentService.SetDeadLetterStore(deadLetterStore)
 
+	// Ensure system agent exists (used for retention warnings and system notifications)
+	if _, err := agentService.EnsureSystemAgent(ctx, 1); err != nil {
+		slog.Warn("failed to create system agent", "error", err)
+	}
+
 	channelStore := channels.NewSQLiteChannelStore(db.DB)
 	channelService := channels.NewService(channelStore, msgService, tracer)
 
@@ -427,7 +437,7 @@ func runServe(cmd *cobra.Command, args []string) error {
 	msgService.SetDispatcher(eventDispatcher)
 
 	// Create MCP server (with swarm + attachment + search + webhook + K8s tools)
-	mcpSrv := mcpserver.NewMCPServer(msgService, agentService, channelService, swarmService, attachmentService, searchService, con, webhookService, k8sService)
+	mcpSrv := mcpserver.NewMCPServer(msgService, agentService, channelService, swarmService, attachmentService, searchService, con, webhookService, k8sService, db.DB)
 	startTime := time.Now()
 
 	// Start task expiry worker
@@ -435,6 +445,20 @@ func runServe(cmd *cobra.Command, args []string) error {
 	expiryWorker.Start()
 	slog.Info("task expiry worker started")
 
+	// Start message retention worker
+	retentionCfg := messaging.ParseRetentionPeriod(messageRetention)
+	var retentionWorker *messaging.RetentionWorker
+	if retentionCfg.Enabled {
+		retentionWorker = messaging.NewRetentionWorker(db.DB, retentionCfg, dataDir)
+		retentionWorker.Start()
+		slog.Info("message retention worker started",
+			"retention_period", retentionCfg.RetentionPeriodHuman(),
+			"cleanup_interval", retentionCfg.CleanupInterval.String(),
+		)
+	} else {
+		slog.Info("message retention disabled")
+	}
+
 	// Create health checker
 	healthChecker := health.NewChecker(db.DB, version)
 
@@ -520,13 +544,25 @@ func runServe(cmd *cobra.Command, args []string) error {
 
 	// Start admin socket server
 	adminSvcs := &admin.Services{
-		Users:    userStore,
-		Sessions: sessionStore,
-		Agents:   agentService,
-		Messages: msgService,
-		Channels: channelService,
-		Traces:   traceStore,
-		DataDir:  dataDir,
+		Users:             userStore,
+		Sessions:          sessionStore,
+		Agents:            agentService,
+		Messages:          msgService,
+		Channels:          channelService,
+		Traces:            traceStore,
+		DataDir:           dataDir,
+	}
+	// Wire optional services into admin (may be nil if not configured)
+	if searchCfg.IsEnabled() {
+		adminSvcs.EmbeddingStore = search.NewEmbeddingStore(db.DB)
+		adminSvcs.VectorIndex = vectorIndex
+		adminSvcs.SearchService = searchService
+	}
+	if attachmentService != nil {
+		adminSvcs.AttachmentService = attachmentService
+	}
+	if retentionWorker != nil {
+		adminSvcs.RetentionWorker = retentionWorker
 	}
 	adminServer := admin.NewServer(adminSocketPath, db.DB, adminSvcs, logger)
 	if err := adminServer.Start(); err != nil {
@@ -581,6 +617,11 @@ func runServe(cmd *cobra.Command, args []string) error {
 	// Stop expiry worker
 	expiryWorker.Stop()
 
+	// Stop message retention worker
+	if retentionWorker != nil {
+		retentionWorker.Stop()
+	}
+
 	// Stop embedding pipeline
 	if embPipeline != nil {
 		embPipeline.Stop()

internal/admin/server.go

@@ -7,21 +7,33 @@ import (
 	"net"
 
 	"github.com/synapbus/synapbus/internal/agents"
+	"github.com/synapbus/synapbus/internal/attachments"
 	"github.com/synapbus/synapbus/internal/auth"
 	"github.com/synapbus/synapbus/internal/channels"
 	"github.com/synapbus/synapbus/internal/messaging"
+	"github.com/synapbus/synapbus/internal/search"
 	"github.com/synapbus/synapbus/internal/trace"
 )
 
 // Services holds references to all services the admin socket can control.
 type Services struct {
-	Users    *auth.SQLiteUserStore
-	Sessions auth.SessionStore
-	Agents   *agents.AgentService
-	Messages *messaging.MessagingService
-	Channels *channels.Service
-	Traces   trace.TraceStore
-	DataDir  string
+	Users              *auth.SQLiteUserStore
+	Sessions           auth.SessionStore
+	Agents             *agents.AgentService
+	Messages           *messaging.MessagingService
+	Channels           *channels.Service
+	Traces             trace.TraceStore
+	EmbeddingStore     *search.EmbeddingStore
+	VectorIndex        *search.VectorIndex
+	SearchService      *search.Service
+	AttachmentService  *attachments.Service
+	DataDir            string
+	RetentionWorker    RetentionStatusProvider
+}
+
+// RetentionStatusProvider provides retention status information.
+type RetentionStatusProvider interface {
+	Status() map[string]interface{}
 }
 
 // AdminServer is a Unix domain socket server for local administration.

internal/admin/socket.go

@@ -168,6 +168,26 @@ func (s *AdminServer) dispatch(req Request) Response {
 	case "conversations.show":
 		return s.handleConversationsShow(ctx, req.Args)
 
+	// --- embeddings ---
+	case "embeddings.status":
+		return s.handleEmbeddingsStatus(ctx)
+	case "embeddings.reindex":
+		return s.handleEmbeddingsReindex(ctx)
+	case "embeddings.clear":
+		return s.handleEmbeddingsClear(ctx)
+
+	// --- db maintenance ---
+	case "db.vacuum":
+		return s.handleDBVacuum(ctx)
+
+	// --- messages purge ---
+	case "messages.purge":
+		return s.handleMessagesPurge(ctx, req.Args)
+
+	// --- retention ---
+	case "retention.status":
+		return s.handleRetentionStatus(ctx)
+
 	default:
 		return Response{OK: false, Error: fmt.Sprintf("unknown command: %s", req.Command)}
 	}
@@ -941,5 +961,183 @@ func (s *AdminServer) handleConversationsShow(ctx context.Context, args json.Raw
 	}}
 }
 
+// ---------- embeddings handlers ----------
+
+func (s *AdminServer) handleEmbeddingsStatus(ctx context.Context) Response {
+	result := map[string]interface{}{
+		"provider":       "",
+		"total_embedded": int64(0),
+		"pending_count":  int64(0),
+		"failed_count":   int64(0),
+		"index_size":     0,
+		"dimensions":     0,
+	}
+
+	if s.services.EmbeddingStore == nil {
+		return Response{OK: true, Data: result}
+	}
+
+	stats, err := s.services.EmbeddingStore.Stats(ctx)
+	if err != nil {
+		return Response{OK: false, Error: "get stats: " + err.Error()}
+	}
+
+	result["provider"] = stats.Provider
+	result["total_embedded"] = stats.TotalEmbedded
+	result["pending_count"] = stats.PendingCount
+	result["failed_count"] = stats.FailedCount
+	result["dimensions"] = stats.Dimensions
+
+	if s.services.VectorIndex != nil {
+		result["index_size"] = s.services.VectorIndex.Len()
+	}
+
+	return Response{OK: true, Data: result}
+}
+
+func (s *AdminServer) handleEmbeddingsReindex(ctx context.Context) Response {
+	if s.services.EmbeddingStore == nil {
+		return Response{OK: false, Error: "embedding subsystem not configured"}
+	}
+
+	if err := s.services.EmbeddingStore.DeleteAllEmbeddings(ctx); err != nil {
+		return Response{OK: false, Error: "delete embeddings: " + err.Error()}
+	}
+
+	if err := s.services.EmbeddingStore.ClearQueue(ctx); err != nil {
+		return Response{OK: false, Error: "clear queue: " + err.Error()}
+	}
+
+	clearedIndex := false
+	if s.services.VectorIndex != nil {
+		if err := s.services.VectorIndex.Rebuild(nil); err != nil {
+			return Response{OK: false, Error: "clear index: " + err.Error()}
+		}
+		clearedIndex = true
+	}
+
+	enqueued, err := s.services.EmbeddingStore.EnqueueAllMessages(ctx)
+	if err != nil {
+		return Response{OK: false, Error: "enqueue messages: " + err.Error()}
+	}
+
+	return Response{OK: true, Data: map[string]interface{}{
+		"deleted_embeddings": true,
+		"cleared_index":      clearedIndex,
+		"enqueued_messages":  enqueued,
+	}}
+}
+
+func (s *AdminServer) handleEmbeddingsClear(ctx context.Context) Response {
+	if s.services.EmbeddingStore == nil {
+		return Response{OK: false, Error: "embedding subsystem not configured"}
+	}
+
+	if err := s.services.EmbeddingStore.DeleteAllEmbeddings(ctx); err != nil {
+		return Response{OK: false, Error: "delete embeddings: " + err.Error()}
+	}
+
+	if err := s.services.EmbeddingStore.ClearQueue(ctx); err != nil {
+		return Response{OK: false, Error: "clear queue: " + err.Error()}
+	}
+
+	clearedIndex := false
+	if s.services.VectorIndex != nil {
+		if err := s.services.VectorIndex.Rebuild(nil); err != nil {
+			return Response{OK: false, Error: "clear index: " + err.Error()}
+		}
+		clearedIndex = true
+	}
+
+	return Response{OK: true, Data: map[string]interface{}{
+		"deleted_embeddings": true,
+		"cleared_index":      clearedIndex,
+		"cleared_queue":      true,
+	}}
+}
+
+// ---------- db maintenance handlers ----------
+
+func (s *AdminServer) handleDBVacuum(ctx context.Context) Response {
+	dbPath := filepath.Join(s.services.DataDir, "synapbus.db")
+
+	beforeInfo, err := os.Stat(dbPath)
+	if err != nil {
+		return Response{OK: false, Error: "stat db: " + err.Error()}
+	}
+	beforeSize := beforeInfo.Size()
+
+	start := time.Now()
+
+	if _, err := s.db.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)"); err != nil {
+		return Response{OK: false, Error: "wal checkpoint: " + err.Error()}
+	}
+
+	if _, err := s.db.ExecContext(ctx, "VACUUM"); err != nil {
+		return Response{OK: false, Error: "vacuum: " + err.Error()}
+	}
+
+	durationMs := time.Since(start).Milliseconds()
+
+	afterInfo, err := os.Stat(dbPath)
+	if err != nil {
+		return Response{OK: false, Error: "stat db after vacuum: " + err.Error()}
+	}
+	afterSize := afterInfo.Size()
+
+	return Response{OK: true, Data: map[string]interface{}{
+		"before_size_bytes": beforeSize,
+		"after_size_bytes":  afterSize,
+		"reclaimed_bytes":   beforeSize - afterSize,
+		"duration_ms":       durationMs,
+	}}
+}
+
+// ---------- messages purge handler ----------
+
+func (s *AdminServer) handleMessagesPurge(ctx context.Context, args json.RawMessage) Response {
+	var p struct {
+		OlderThan string `json:"older_than"`
+		Agent     string `json:"agent"`
+		Channel   string `json:"channel"`
+	}
+	if args != nil {
+		json.Unmarshal(args, &p)
+	}
+
+	if p.OlderThan == "" && p.Agent == "" && p.Channel == "" {
+		return Response{OK: false, Error: "at least one filter is required (older_than, agent, or channel)"}
+	}
+
+	var olderThan time.Duration
+	if p.OlderThan != "" {
+		cfg := messaging.ParseRetentionPeriod(p.OlderThan)
+		if cfg.RetentionPeriod <= 0 {
+			return Response{OK: false, Error: fmt.Sprintf("invalid duration: %q", p.OlderThan)}
+		}
+		olderThan = cfg.RetentionPeriod
+	}
+
+	counts, err := messaging.PurgeMessages(ctx, s.db, s.services.DataDir, olderThan, p.Agent, p.Channel)
+	if err != nil {
+		return Response{OK: false, Error: err.Error()}
+	}
+
+	return Response{OK: true, Data: counts}
+}
+
+// ---------- retention handler ----------
+
+func (s *AdminServer) handleRetentionStatus(ctx context.Context) Response {
+	if s.services.RetentionWorker != nil {
+		return Response{OK: true, Data: s.services.RetentionWorker.Status()}
+	}
+
+	return Response{OK: true, Data: map[string]interface{}{
+		"enabled": false,
+		"message": "retention worker not configured",
+	}}
+}
+
 // Ensure the messaging import is used.
 var _ = messaging.StatusPending

internal/agents/service.go

@@ -345,6 +345,37 @@ func (s *AgentService) EnsureHumanAgent(ctx context.Context, username, displayNa
 	return agent, nil
 }
 
+// EnsureSystemAgent creates the "system" agent if it doesn't already exist.
+// The system agent is used for retention warnings and other system notifications.
+func (s *AgentService) EnsureSystemAgent(ctx context.Context, ownerID int64) (*Agent, error) {
+	agent, err := s.store.GetAgentByName(ctx, "system")
+	if err == nil && agent != nil && agent.Status == "active" {
+		return agent, nil
+	}
+
+	agent, _, err = s.Register(ctx, "system", "System", "ai", json.RawMessage(`{"role":"system-notifications"}`), ownerID)
+	if err != nil {
+		// May already exist from a concurrent call
+		agent, err2 := s.store.GetAgentByName(ctx, "system")
+		if err2 == nil && agent != nil {
+			return agent, nil
+		}
+		return nil, fmt.Errorf("create system agent: %w", err)
+	}
+
+	s.logger.Info("created system agent", "owner_id", ownerID)
+	return agent, nil
+}
+
+// GetAgentWithOwner returns agent details along with the owner's display name.
+func (s *AgentService) GetAgentWithOwner(ctx context.Context, name string) (*Agent, string, error) {
+	agent, err := s.store.GetAgentByName(ctx, name)
+	if err != nil {
+		return nil, "", fmt.Errorf("get agent: %w", err)
+	}
+	return agent, "", nil // Owner name resolved by caller with access to user store
+}
+
 // generateAPIKey creates a cryptographically random API key (32 bytes, hex encoded).
 func generateAPIKey() (string, error) {
 	b := make([]byte, 32)

internal/channels/service.go

@@ -5,11 +5,20 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"time"
 
 	"github.com/synapbus/synapbus/internal/messaging"
 	"github.com/synapbus/synapbus/internal/trace"
 )
 
+// ChannelSummary holds a joined channel with its unread message count.
+type ChannelSummary struct {
+	ID            int64      `json:"id"`
+	Name          string     `json:"name"`
+	UnreadCount   int        `json:"unread"`
+	LastMessageAt *time.Time `json:"last_message_at"`
+}
+
 // Service provides business logic for channel operations.
 type Service struct {
 	store      ChannelStore
@@ -520,6 +529,11 @@ func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAge
 	return []*messaging.Message{channelMsg}, nil
 }
 
+// GetChannelSummaries returns channels the agent has joined with unread message counts.
+func (s *Service) GetChannelSummaries(ctx context.Context, agentName string) ([]ChannelSummary, error) {
+	return s.store.GetChannelSummaries(ctx, agentName)
+}
+
 // GetMembers returns all members of a channel.
 func (s *Service) GetMembers(ctx context.Context, channelID int64) ([]*Membership, error) {
 	return s.store.GetMembers(ctx, channelID)

internal/channels/store.go

@@ -26,6 +26,7 @@ type ChannelStore interface {
 	GetInvite(ctx context.Context, channelID int64, agentName string) (*ChannelInvite, error)
 	HasPendingInvite(ctx context.Context, channelID int64, agentName string) (bool, error)
 	AcceptInvite(ctx context.Context, channelID int64, agentName string) error
+	GetChannelSummaries(ctx context.Context, agentName string) ([]ChannelSummary, error)
 }
 
 // SQLiteChannelStore implements ChannelStore using SQLite.
@@ -350,6 +351,46 @@ func (s *SQLiteChannelStore) AcceptInvite(ctx context.Context, channelID int64,
 	return nil
 }
 
+// GetChannelSummaries returns channels the agent has joined with unread message counts.
+func (s *SQLiteChannelStore) GetChannelSummaries(ctx context.Context, agentName string) ([]ChannelSummary, error) {
+	rows, err := s.db.QueryContext(ctx,
+		`SELECT c.id, c.name,
+		        (SELECT COUNT(*) FROM messages m
+		         WHERE m.channel_id = c.id
+		           AND m.id > COALESCE(
+		               (SELECT MAX(ist.last_read_message_id) FROM inbox_state ist
+		                WHERE ist.agent_name = ? AND ist.conversation_id = m.conversation_id), 0)
+		           AND m.from_agent != ?
+		        ) AS unread_count,
+		        (SELECT MAX(m2.created_at) FROM messages m2 WHERE m2.channel_id = c.id) AS last_message_at
+		 FROM channels c
+		 JOIN channel_members cm ON cm.channel_id = c.id AND cm.agent_name = ?
+		 ORDER BY c.name`,
+		agentName, agentName, agentName,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("get channel summaries: %w", err)
+	}
+	defer rows.Close()
+
+	var summaries []ChannelSummary
+	for rows.Next() {
+		var cs ChannelSummary
+		var lastMsg sql.NullTime
+		if err := rows.Scan(&cs.ID, &cs.Name, &cs.UnreadCount, &lastMsg); err != nil {
+			return nil, fmt.Errorf("scan channel summary: %w", err)
+		}
+		if lastMsg.Valid {
+			cs.LastMessageAt = &lastMsg.Time
+		}
+		summaries = append(summaries, cs)
+	}
+	if summaries == nil {
+		summaries = []ChannelSummary{}
+	}
+	return summaries, rows.Err()
+}
+
 // isUniqueConstraintError checks if an error is a SQLite unique constraint violation.
 func isUniqueConstraintError(err error) bool {
 	return strings.Contains(err.Error(), "UNIQUE constraint failed")

internal/mcp/server.go

@@ -2,6 +2,7 @@ package mcp
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"log/slog"
 	"net/http"
@@ -42,6 +43,7 @@ func NewMCPServer(
 	consolePrinter *console.Printer,
 	webhookService *webhooks.WebhookService,
 	k8sService *k8s.K8sService,
+	db *sql.DB,
 ) *MCPServer {
 	logger := slog.Default().With("component", "mcp-server")
 	connMgr := NewConnectionManager()
@@ -146,6 +148,12 @@ func NewMCPServer(
 	if searchService != nil {
 		registrar.SetSearchService(searchService)
 	}
+	if channelService != nil {
+		registrar.SetChannelService(channelService)
+	}
+	if db != nil {
+		registrar.SetDB(db)
+	}
 	registrar.RegisterAll(mcpSrv)
 
 	// Register channel tools

internal/mcp/server_test.go

@@ -31,7 +31,7 @@ func TestNewMCPServerWithConsole(t *testing.T) {
 
 	con := console.New()
 
-	srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, con, nil, nil)
+	srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, con, nil, nil, nil)
 	if srv == nil {
 		t.Fatal("expected non-nil MCPServer")
 	}
@@ -56,7 +56,7 @@ func TestNewMCPServerNilConsole(t *testing.T) {
 	agentService := agents.NewAgentService(agentStore, tracer)
 
 	// nil console should not panic
-	srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, nil, nil)
+	srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, nil, nil, nil)
 	if srv == nil {
 		t.Fatal("expected non-nil MCPServer")
 	}
@@ -126,7 +126,7 @@ func TestMCPToolCall_WithValidAPIKey(t *testing.T) {
 	agentService.Register(ctx, "receiver", "Receiver", "ai", nil, 1)
 
 	// Create MCP server
-	srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, nil, nil)
+	srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, nil, nil, nil)
 
 	// Mount with auth middleware, just like main.go does
 	mux := http.NewServeMux()
@@ -180,7 +180,7 @@ func TestMCPToolCall_InvalidAPIKeyReturns401(t *testing.T) {
 	apiKeyStore := apikeys.NewSQLiteStore(db)
 	apiKeyService := apikeys.NewService(apiKeyStore)
 
-	srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, nil, nil)
+	srv := NewMCPServer(msgService, agentService, nil, nil, nil, nil, nil, nil, nil, nil)
 
 	mux := http.NewServeMux()
 	handler := agents.OptionalAuthMiddlewareWithAPIKeys(agentService, apiKeyService)(srv.Handler())

internal/mcp/tools.go

@@ -2,24 +2,29 @@ package mcp
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"strings"
 
 	"github.com/mark3labs/mcp-go/mcp"
 	"github.com/mark3labs/mcp-go/server"
 
 	"github.com/synapbus/synapbus/internal/agents"
+	"github.com/synapbus/synapbus/internal/channels"
 	"github.com/synapbus/synapbus/internal/messaging"
 	"github.com/synapbus/synapbus/internal/search"
 )
 
 // ToolRegistrar registers all SynapBus MCP tools on the given server.
 type ToolRegistrar struct {
-	msgService    *messaging.MessagingService
-	agentService  *agents.AgentService
-	searchService *search.Service
-	logger        *slog.Logger
+	msgService     *messaging.MessagingService
+	agentService   *agents.AgentService
+	channelService *channels.Service
+	searchService  *search.Service
+	db             *sql.DB
+	logger         *slog.Logger
 }
 
 // NewToolRegistrar creates a new tool registrar.
@@ -36,10 +41,21 @@ func (tr *ToolRegistrar) SetSearchService(svc *search.Service) {
 	tr.searchService = svc
 }
 
+// SetChannelService sets the channel service for my_status support.
+func (tr *ToolRegistrar) SetChannelService(svc *channels.Service) {
+	tr.channelService = svc
+}
+
+// SetDB sets the database handle for direct queries (e.g. owner name lookup).
+func (tr *ToolRegistrar) SetDB(db *sql.DB) {
+	tr.db = db
+}
+
 // RegisterAll registers all tools on the MCP server.
 // Note: Agent management tools (register, update, deregister) are NOT exposed via MCP.
 // Agents are managed exclusively through the Web UI. MCP is for messaging only.
 func (tr *ToolRegistrar) RegisterAll(s *server.MCPServer) {
+	s.AddTool(tr.myStatusTool(), tr.handleMyStatus)
 	s.AddTool(tr.sendMessageTool(), tr.handleSendMessage)
 	s.AddTool(tr.readInboxTool(), tr.handleReadInbox)
 	s.AddTool(tr.claimMessagesTool(), tr.handleClaimMessages)
@@ -47,7 +63,7 @@ func (tr *ToolRegistrar) RegisterAll(s *server.MCPServer) {
 	s.AddTool(tr.searchMessagesTool(), tr.handleSearchMessages)
 	s.AddTool(tr.discoverAgentsTool(), tr.handleDiscoverAgents)
 
-	tr.logger.Info("all MCP tools registered", "count", 6)
+	tr.logger.Info("all MCP tools registered", "count", 7)
 }
 
 // --- Tool Definitions ---
@@ -112,6 +128,11 @@ func (tr *ToolRegistrar) discoverAgentsTool() mcp.Tool {
 	)
 }
 
+func (tr *ToolRegistrar) myStatusTool() mcp.Tool {
+	return mcp.NewTool("my_status",
+		mcp.WithDescription("Get your complete status overview — identity, pending messages, channel mentions, system notifications, and statistics. Call this first when connecting to SynapBus."),
+	)
+}
 
 // --- Tool Handlers ---
 
@@ -342,16 +363,19 @@ func (tr *ToolRegistrar) handleDiscoverAgents(ctx context.Context, req mcp.CallT
 		return mcp.NewToolResultError(fmt.Sprintf("discover_agents failed: %s", err)), nil
 	}
 
-	// Strip sensitive fields
-	result := make([]map[string]any, len(agentsList))
-	for i, a := range agentsList {
-		result[i] = map[string]any{
+	// Strip sensitive fields, exclude system agent
+	result := make([]map[string]any, 0, len(agentsList))
+	for _, a := range agentsList {
+		if a.Name == "system" {
+			continue
+		}
+		result = append(result, map[string]any{
 			"name":         a.Name,
 			"display_name": a.DisplayName,
 			"type":         a.Type,
 			"capabilities": a.Capabilities,
 			"status":       a.Status,
-		}
+		})
 	}
 
 	return resultJSON(map[string]any{
@@ -360,6 +384,181 @@ func (tr *ToolRegistrar) handleDiscoverAgents(ctx context.Context, req mcp.CallT
 	})
 }
 
+func (tr *ToolRegistrar) handleMyStatus(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+	agentName, ok := extractAgentName(ctx)
+	if !ok {
+		return mcp.NewToolResultError("authentication required"), nil
+	}
+
+	// 1. Get agent identity
+	agent, err := tr.agentService.GetAgent(ctx, agentName)
+	if err != nil {
+		return mcp.NewToolResultError(fmt.Sprintf("my_status failed: %s", err)), nil
+	}
+
+	// Resolve owner name from users table
+	ownerName := ""
+	if tr.db != nil {
+		var username sql.NullString
+		_ = tr.db.QueryRowContext(ctx,
+			`SELECT username FROM users WHERE id = ?`, agent.OwnerID,
+		).Scan(&username)
+		if username.Valid {
+			ownerName = username.String
+		}
+	}
+
+	agentInfo := map[string]any{
+		"name":         agent.Name,
+		"display_name": agent.DisplayName,
+		"type":         agent.Type,
+		"owner":        ownerName,
+	}
+
+	// 2. Get pending DMs
+	pendingDMs, err := tr.msgService.GetPendingDMs(ctx, agentName, 10)
+	if err != nil {
+		return mcp.NewToolResultError(fmt.Sprintf("my_status failed: %s", err)), nil
+	}
+	pendingDMCount, err := tr.msgService.GetPendingDMCount(ctx, agentName)
+	if err != nil {
+		return mcp.NewToolResultError(fmt.Sprintf("my_status failed: %s", err)), nil
+	}
+
+	dmList := make([]map[string]any, len(pendingDMs))
+	for i, msg := range pendingDMs {
+		body := msg.Body
+		if len(body) > 200 {
+			body = body[:200] + "..."
+		}
+		entry := map[string]any{
+			"id":         msg.ID,
+			"from":       msg.FromAgent,
+			"body":       body,
+			"priority":   msg.Priority,
+			"status":     msg.Status,
+			"created_at": msg.CreatedAt,
+		}
+		// Include subject from conversation if available
+		if msg.ConversationID > 0 {
+			conv, _, _ := tr.msgService.GetConversation(ctx, msg.ConversationID)
+			if conv != nil && conv.Subject != "" {
+				entry["subject"] = conv.Subject
+			}
+		}
+		dmList[i] = entry
+	}
+
+	// 3. Get channel mentions
+	mentions, err := tr.msgService.GetRecentMentions(ctx, agentName, 10)
+	if err != nil {
+		return mcp.NewToolResultError(fmt.Sprintf("my_status failed: %s", err)), nil
+	}
+
+	mentionList := make([]map[string]any, len(mentions))
+	for i, msg := range mentions {
+		body := msg.Body
+		if len(body) > 200 {
+			body = body[:200] + "..."
+		}
+		entry := map[string]any{
+			"id":         msg.ID,
+			"from":       msg.FromAgent,
+			"body":       body,
+			"created_at": msg.CreatedAt,
+		}
+		// Try to extract channel name from metadata
+		if len(msg.Metadata) > 0 {
+			var meta map[string]any
+			if json.Unmarshal(msg.Metadata, &meta) == nil {
+				if chName, ok := meta["channel_name"].(string); ok {
+					entry["channel"] = chName
+				}
+			}
+		}
+		mentionList[i] = entry
+	}
+
+	// 4. Get system notifications
+	sysNotifs, err := tr.msgService.GetSystemNotifications(ctx, agentName, 5)
+	if err != nil {
+		return mcp.NewToolResultError(fmt.Sprintf("my_status failed: %s", err)), nil
+	}
+
+	sysNotifList := make([]map[string]any, len(sysNotifs))
+	for i, msg := range sysNotifs {
+		body := msg.Body
+		if len(body) > 200 {
+			body = body[:200] + "..."
+		}
+		sysNotifList[i] = map[string]any{
+			"id":         msg.ID,
+			"body":       body,
+			"created_at": msg.CreatedAt,
+		}
+	}
+
+	// 5. Get channel summaries
+	var channelSummaries []channels.ChannelSummary
+	if tr.channelService != nil {
+		channelSummaries, err = tr.channelService.GetChannelSummaries(ctx, agentName)
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("my_status failed: %s", err)), nil
+		}
+	}
+	if channelSummaries == nil {
+		channelSummaries = []channels.ChannelSummary{}
+	}
+
+	// 6. Build stats
+	totalUnreadChannel := 0
+	for _, cs := range channelSummaries {
+		totalUnreadChannel += cs.UnreadCount
+	}
+
+	stats := map[string]any{
+		"pending_dms":             pendingDMCount,
+		"channels_joined":        len(channelSummaries),
+		"unread_channel_messages": totalUnreadChannel,
+		"system_notifications":    len(sysNotifs),
+	}
+
+	// 7. Build truncation instructions
+	var instructionParts []string
+	truncated := false
+	if int64(len(pendingDMs)) < pendingDMCount {
+		truncated = true
+		instructionParts = append(instructionParts, fmt.Sprintf("Showing %d of %d pending messages. Use read_inbox to see all.", len(pendingDMs), pendingDMCount))
+	}
+	if len(mentions) >= 10 {
+		truncated = true
+		instructionParts = append(instructionParts, fmt.Sprintf("Showing %d mentions (may be more). Use search_messages to find all.", len(mentions)))
+	}
+	if len(sysNotifs) >= 5 {
+		truncated = true
+		instructionParts = append(instructionParts, fmt.Sprintf("Showing %d system notifications (may be more). Use read_inbox with from_agent='system' to see all.", len(sysNotifs)))
+	}
+
+	result := map[string]any{
+		"agent":                    agentInfo,
+		"direct_messages":          dmList,
+		"direct_messages_total":    pendingDMCount,
+		"mentions":                 mentionList,
+		"mentions_total":           len(mentions),
+		"system_notifications":       sysNotifList,
+		"system_notifications_total": len(sysNotifs),
+		"channels":                 channelSummaries,
+		"stats":                    stats,
+		"truncated":                truncated,
+	}
+
+	if len(instructionParts) > 0 {
+		result["instructions"] = strings.Join(instructionParts, " ")
+	}
+
+	return resultJSON(result)
+}
+
 // resultJSON marshals data to a JSON text MCP result.
 func resultJSON(data any) (*mcp.CallToolResult, error) {
 	b, err := json.Marshal(data)

internal/messaging/retention.go

@@ -0,0 +1,655 @@
+package messaging
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+// RetentionConfig holds message retention settings.
+type RetentionConfig struct {
+	// RetentionPeriod is how long messages are kept. 0 disables retention.
+	RetentionPeriod time.Duration
+	// WarningWindow is how long before deletion to send warnings (default 30 days).
+	WarningWindow time.Duration
+	// CleanupInterval is how often the cleanup job runs (default 24h).
+	CleanupInterval time.Duration
+	// Enabled is true if retention is active (RetentionPeriod > 0).
+	Enabled bool
+}
+
+// DefaultRetentionConfig returns the default retention configuration (12 months).
+func DefaultRetentionConfig() RetentionConfig {
+	return RetentionConfig{
+		RetentionPeriod: 365 * 24 * time.Hour, // 12 months ≈ 365 days
+		WarningWindow:   30 * 24 * time.Hour,  // 1 month
+		CleanupInterval: 24 * time.Hour,
+		Enabled:         true,
+	}
+}
+
+// ParseRetentionPeriod parses a retention string like "12m" (months), "365d" (days), "8760h" (hours), or "0" (disabled).
+func ParseRetentionPeriod(s string) RetentionConfig {
+	cfg := DefaultRetentionConfig()
+
+	if s == "" || s == "0" {
+		cfg.Enabled = false
+		cfg.RetentionPeriod = 0
+		return cfg
+	}
+
+	s = strings.TrimSpace(s)
+
+	// Try "Nm" format (months)
+	if strings.HasSuffix(s, "m") && !strings.HasSuffix(s, "ms") {
+		months, err := strconv.Atoi(strings.TrimSuffix(s, "m"))
+		if err == nil && months > 0 {
+			cfg.RetentionPeriod = time.Duration(months) * 30 * 24 * time.Hour
+			return cfg
+		}
+	}
+
+	// Try "Nd" format (days)
+	if strings.HasSuffix(s, "d") {
+		days, err := strconv.Atoi(strings.TrimSuffix(s, "d"))
+		if err == nil && days > 0 {
+			cfg.RetentionPeriod = time.Duration(days) * 24 * time.Hour
+			return cfg
+		}
+	}
+
+	// Try standard Go duration
+	d, err := time.ParseDuration(s)
+	if err == nil && d > 0 {
+		cfg.RetentionPeriod = d
+		return cfg
+	}
+
+	// Invalid — disable
+	cfg.Enabled = false
+	cfg.RetentionPeriod = 0
+	return cfg
+}
+
+// RetentionPeriodHuman returns a human-readable retention period string.
+func (c RetentionConfig) RetentionPeriodHuman() string {
+	if !c.Enabled {
+		return "disabled"
+	}
+	days := int(c.RetentionPeriod.Hours() / 24)
+	if days%30 == 0 {
+		months := days / 30
+		if months == 1 {
+			return "1 month"
+		}
+		return fmt.Sprintf("%d months", months)
+	}
+	if days == 1 {
+		return "1 day"
+	}
+	return fmt.Sprintf("%d days", days)
+}
+
+// RetentionWorker runs periodic message cleanup.
+type RetentionWorker struct {
+	db              *sql.DB
+	config          RetentionConfig
+	dataDir         string
+	logger          *slog.Logger
+	cancel          context.CancelFunc
+	wg              sync.WaitGroup
+	mu              sync.RWMutex
+	lastCleanupAt   *time.Time
+	lastDeleteCount int64
+}
+
+// NewRetentionWorker creates a new retention worker.
+func NewRetentionWorker(db *sql.DB, config RetentionConfig, dataDir string) *RetentionWorker {
+	return &RetentionWorker{
+		db:      db,
+		config:  config,
+		dataDir: dataDir,
+		logger:  slog.Default().With("component", "retention-worker"),
+	}
+}
+
+// Start begins the retention cleanup loop.
+func (w *RetentionWorker) Start() {
+	if !w.config.Enabled {
+		w.logger.Info("message retention disabled")
+		return
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	w.cancel = cancel
+
+	w.logger.Info("starting message retention worker",
+		"retention_period", w.config.RetentionPeriod.String(),
+		"warning_window", w.config.WarningWindow.String(),
+		"cleanup_interval", w.config.CleanupInterval.String(),
+	)
+
+	w.wg.Add(1)
+	go w.loop(ctx)
+}
+
+// Stop halts the retention worker.
+func (w *RetentionWorker) Stop() {
+	if w.cancel != nil {
+		w.cancel()
+	}
+	w.wg.Wait()
+	w.logger.Info("retention worker stopped")
+}
+
+// Status returns the current retention status for admin queries.
+func (w *RetentionWorker) Status() map[string]interface{} {
+	w.mu.RLock()
+	defer w.mu.RUnlock()
+
+	result := map[string]interface{}{
+		"enabled":              w.config.Enabled,
+		"retention_period":     w.config.RetentionPeriod.String(),
+		"retention_period_human": w.config.RetentionPeriodHuman(),
+		"warning_window":       w.config.WarningWindow.String(),
+		"cleanup_interval":     w.config.CleanupInterval.String(),
+	}
+
+	if w.lastCleanupAt != nil {
+		result["last_cleanup_at"] = w.lastCleanupAt.Format(time.RFC3339)
+		next := w.lastCleanupAt.Add(w.config.CleanupInterval)
+		result["next_cleanup_at"] = next.Format(time.RFC3339)
+	}
+
+	// Get message age distribution
+	dist, total := w.messageAgeDistribution()
+	result["message_age_distribution"] = dist
+	result["total_messages"] = total
+
+	return result
+}
+
+func (w *RetentionWorker) loop(ctx context.Context) {
+	defer w.wg.Done()
+
+	// Run initial cleanup after a short delay
+	timer := time.NewTimer(30 * time.Second)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			w.runCleanup(ctx)
+			timer.Reset(w.config.CleanupInterval)
+		}
+	}
+}
+
+func (w *RetentionWorker) runCleanup(ctx context.Context) {
+	w.logger.Info("starting retention cleanup")
+
+	// Step 1: Send warnings for messages approaching retention
+	warned := w.sendRetentionWarnings(ctx)
+
+	// Step 2: Delete expired messages
+	deleted := w.deleteExpiredMessages(ctx)
+
+	// Step 3: Run incremental vacuum
+	w.runIncrementalVacuum(ctx)
+
+	now := time.Now()
+	w.mu.Lock()
+	w.lastCleanupAt = &now
+	w.lastDeleteCount = deleted
+	w.mu.Unlock()
+
+	w.logger.Info("retention cleanup complete",
+		"warnings_sent", warned,
+		"messages_deleted", deleted,
+	)
+}
+
+func (w *RetentionWorker) sendRetentionWarnings(ctx context.Context) int64 {
+	if w.config.WarningWindow <= 0 {
+		return 0
+	}
+
+	// Find conversations with messages in the warning window
+	// (older than retention - warning, but not yet at retention limit)
+	warningCutoff := time.Now().Add(-(w.config.RetentionPeriod - w.config.WarningWindow))
+	retentionCutoff := time.Now().Add(-w.config.RetentionPeriod)
+
+	rows, err := w.db.QueryContext(ctx,
+		`SELECT DISTINCT c.id, c.subject, m.to_agent, m.from_agent
+		 FROM conversations c
+		 JOIN messages m ON m.conversation_id = c.id
+		 WHERE m.created_at < ? AND m.created_at >= ?
+		 AND m.from_agent != 'system'
+		 AND NOT EXISTS (
+			SELECT 1 FROM messages warn
+			WHERE warn.conversation_id = c.id
+			AND warn.from_agent = 'system'
+			AND warn.body LIKE '%will be permanently deleted%'
+			AND warn.created_at > ?
+		 )`,
+		warningCutoff, retentionCutoff,
+		time.Now().Add(-w.config.CleanupInterval), // Don't re-warn within the same interval
+	)
+	if err != nil {
+		w.logger.Error("query warning candidates failed", "error", err)
+		return 0
+	}
+	defer rows.Close()
+
+	type warnTarget struct {
+		convID    int64
+		subject   string
+		toAgent   string
+		fromAgent string
+	}
+
+	var targets []warnTarget
+	for rows.Next() {
+		var t warnTarget
+		if err := rows.Scan(&t.convID, &t.subject, &t.toAgent, &t.fromAgent); err != nil {
+			continue
+		}
+		targets = append(targets, t)
+	}
+
+	// Send system warnings to unique agents
+	warned := int64(0)
+	agentsSeen := make(map[string]bool)
+	for _, t := range targets {
+		for _, agent := range []string{t.toAgent, t.fromAgent} {
+			if agent == "" || agent == "system" || agentsSeen[agent] {
+				continue
+			}
+			agentsSeen[agent] = true
+
+			subject := t.subject
+			if subject == "" {
+				subject = "(no subject)"
+			}
+
+			body := fmt.Sprintf("Conversation '%s' has messages older than %s. These will be permanently deleted in approximately %s.",
+				subject,
+				w.config.RetentionPeriodHuman(),
+				w.formatDuration(w.config.WarningWindow),
+			)
+
+			_, err := w.db.ExecContext(ctx,
+				`INSERT INTO conversations (subject, created_by, created_at, updated_at)
+				 VALUES (?, 'system', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
+				"Retention Warning",
+			)
+			if err != nil {
+				continue
+			}
+
+			var convID int64
+			w.db.QueryRowContext(ctx, `SELECT last_insert_rowid()`).Scan(&convID)
+
+			_, err = w.db.ExecContext(ctx,
+				`INSERT INTO messages (conversation_id, from_agent, to_agent, body, priority, status, metadata, created_at, updated_at)
+				 VALUES (?, 'system', ?, ?, 5, 'pending', '{}', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
+				convID, agent, body,
+			)
+			if err != nil {
+				w.logger.Error("send retention warning failed", "agent", agent, "error", err)
+				continue
+			}
+			warned++
+		}
+	}
+
+	return warned
+}
+
+func (w *RetentionWorker) deleteExpiredMessages(ctx context.Context) int64 {
+	cutoff := time.Now().Add(-w.config.RetentionPeriod)
+
+	// Get message IDs to delete (skip processing status)
+	rows, err := w.db.QueryContext(ctx,
+		`SELECT id FROM messages WHERE created_at < ? AND status != 'processing'`,
+		cutoff,
+	)
+	if err != nil {
+		w.logger.Error("query expired messages failed", "error", err)
+		return 0
+	}
+
+	var ids []int64
+	for rows.Next() {
+		var id int64
+		if err := rows.Scan(&id); err != nil {
+			continue
+		}
+		ids = append(ids, id)
+	}
+	rows.Close()
+
+	if len(ids) == 0 {
+		return 0
+	}
+
+	// Delete in batches
+	batchSize := 500
+	totalDeleted := int64(0)
+
+	for i := 0; i < len(ids); i += batchSize {
+		end := i + batchSize
+		if end > len(ids) {
+			end = len(ids)
+		}
+		batch := ids[i:end]
+		deleted := w.deleteBatch(ctx, batch)
+		totalDeleted += deleted
+	}
+
+	// Clean up orphaned conversations
+	orphaned, err := w.db.ExecContext(ctx,
+		`DELETE FROM conversations WHERE id NOT IN (SELECT DISTINCT conversation_id FROM messages)`,
+	)
+	if err == nil {
+		if n, _ := orphaned.RowsAffected(); n > 0 {
+			w.logger.Info("cleaned orphaned conversations", "count", n)
+		}
+	}
+
+	return totalDeleted
+}
+
+func (w *RetentionWorker) deleteBatch(ctx context.Context, ids []int64) int64 {
+	if len(ids) == 0 {
+		return 0
+	}
+
+	// Build placeholders
+	placeholders := make([]string, len(ids))
+	args := make([]interface{}, len(ids))
+	for i, id := range ids {
+		placeholders[i] = "?"
+		args[i] = id
+	}
+	ph := strings.Join(placeholders, ",")
+
+	tx, err := w.db.BeginTx(ctx, nil)
+	if err != nil {
+		w.logger.Error("begin tx failed", "error", err)
+		return 0
+	}
+	defer tx.Rollback()
+
+	// Delete embedding queue entries
+	tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM embedding_queue WHERE message_id IN (%s)`, ph), args...)
+
+	// Delete embeddings
+	tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM embeddings WHERE message_id IN (%s)`, ph), args...)
+
+	// Collect attachment hashes before deleting
+	hashRows, _ := tx.QueryContext(ctx,
+		fmt.Sprintf(`SELECT DISTINCT hash FROM attachments WHERE message_id IN (%s)`, ph), args...)
+	var hashes []string
+	if hashRows != nil {
+		for hashRows.Next() {
+			var h string
+			hashRows.Scan(&h)
+			hashes = append(hashes, h)
+		}
+		hashRows.Close()
+	}
+
+	// Delete attachment records
+	tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM attachments WHERE message_id IN (%s)`, ph), args...)
+
+	// Delete messages (FTS trigger handles FTS cleanup)
+	result, err := tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM messages WHERE id IN (%s)`, ph), args...)
+	if err != nil {
+		w.logger.Error("delete messages failed", "error", err)
+		return 0
+	}
+
+	if err := tx.Commit(); err != nil {
+		w.logger.Error("commit failed", "error", err)
+		return 0
+	}
+
+	deleted, _ := result.RowsAffected()
+
+	// Clean up orphaned attachment files (outside transaction)
+	w.cleanupAttachmentFiles(ctx, hashes)
+
+	return deleted
+}
+
+func (w *RetentionWorker) cleanupAttachmentFiles(ctx context.Context, hashes []string) {
+	for _, hash := range hashes {
+		// Check if any other attachment references this hash
+		var count int
+		err := w.db.QueryRowContext(ctx,
+			`SELECT COUNT(*) FROM attachments WHERE hash = ?`, hash,
+		).Scan(&count)
+		if err != nil || count > 0 {
+			continue
+		}
+
+		// Remove file from CAS
+		prefix := hash[:2]
+		filePath := filepath.Join(w.dataDir, "attachments", prefix, hash)
+		if err := os.Remove(filePath); err != nil && !os.IsNotExist(err) {
+			w.logger.Warn("remove attachment file failed", "hash", hash, "error", err)
+		}
+	}
+}
+
+func (w *RetentionWorker) runIncrementalVacuum(ctx context.Context) {
+	_, err := w.db.ExecContext(ctx, `PRAGMA incremental_vacuum(1000)`)
+	if err != nil {
+		w.logger.Warn("incremental vacuum failed", "error", err)
+	}
+}
+
+func (w *RetentionWorker) messageAgeDistribution() (map[string]int64, int64) {
+	ctx := context.Background()
+	dist := map[string]int64{
+		"< 1 month":   0,
+		"1-3 months":  0,
+		"3-6 months":  0,
+		"6-12 months": 0,
+		"> 12 months": 0,
+	}
+
+	now := time.Now()
+	boundaries := []struct {
+		label  string
+		cutoff time.Time
+	}{
+		{"< 1 month", now.Add(-30 * 24 * time.Hour)},
+		{"1-3 months", now.Add(-90 * 24 * time.Hour)},
+		{"3-6 months", now.Add(-180 * 24 * time.Hour)},
+		{"6-12 months", now.Add(-365 * 24 * time.Hour)},
+	}
+
+	var total int64
+
+	// Count messages in each bucket
+	for i, b := range boundaries {
+		var count int64
+		var err error
+		if i == 0 {
+			err = w.db.QueryRowContext(ctx,
+				`SELECT COUNT(*) FROM messages WHERE created_at >= ?`, b.cutoff,
+			).Scan(&count)
+		} else {
+			err = w.db.QueryRowContext(ctx,
+				`SELECT COUNT(*) FROM messages WHERE created_at < ? AND created_at >= ?`,
+				boundaries[i-1].cutoff, b.cutoff,
+			).Scan(&count)
+		}
+		if err == nil {
+			dist[b.label] = count
+			total += count
+		}
+	}
+
+	// > 12 months
+	var oldCount int64
+	w.db.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM messages WHERE created_at < ?`,
+		boundaries[len(boundaries)-1].cutoff,
+	).Scan(&oldCount)
+	dist["> 12 months"] = oldCount
+	total += oldCount
+
+	return dist, total
+}
+
+func (w *RetentionWorker) formatDuration(d time.Duration) string {
+	days := int(d.Hours() / 24)
+	if days%30 == 0 && days > 0 {
+		months := days / 30
+		if months == 1 {
+			return "1 month"
+		}
+		return fmt.Sprintf("%d months", months)
+	}
+	if days == 1 {
+		return "1 day"
+	}
+	if days > 0 {
+		return fmt.Sprintf("%d days", days)
+	}
+	return d.String()
+}
+
+// PurgeMessages deletes messages matching the given filters.
+// At least one filter must be specified. Returns counts of deleted items.
+func PurgeMessages(ctx context.Context, db *sql.DB, dataDir string, olderThan time.Duration, agent string, channel string) (map[string]int64, error) {
+	if olderThan == 0 && agent == "" && channel == "" {
+		return nil, fmt.Errorf("at least one filter is required")
+	}
+
+	var conditions []string
+	var args []interface{}
+
+	if olderThan > 0 {
+		cutoff := time.Now().Add(-olderThan)
+		conditions = append(conditions, "m.created_at < ?")
+		args = append(args, cutoff)
+	}
+	if agent != "" {
+		conditions = append(conditions, "(m.from_agent = ? OR m.to_agent = ?)")
+		args = append(args, agent, agent)
+	}
+	if channel != "" {
+		conditions = append(conditions, "m.channel_id IN (SELECT id FROM channels WHERE name = ?)")
+		args = append(args, channel)
+	}
+
+	where := strings.Join(conditions, " AND ")
+	query := fmt.Sprintf(`SELECT m.id FROM messages m WHERE %s`, where)
+
+	rows, err := db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("query messages to purge: %w", err)
+	}
+
+	var ids []int64
+	for rows.Next() {
+		var id int64
+		rows.Scan(&id)
+		ids = append(ids, id)
+	}
+	rows.Close()
+
+	if len(ids) == 0 {
+		return map[string]int64{
+			"deleted_messages":     0,
+			"deleted_embeddings":   0,
+			"deleted_attachments":  0,
+			"cleaned_conversations": 0,
+		}, nil
+	}
+
+	// Build placeholders for batch delete
+	placeholders := make([]string, len(ids))
+	batchArgs := make([]interface{}, len(ids))
+	for i, id := range ids {
+		placeholders[i] = "?"
+		batchArgs[i] = id
+	}
+	ph := strings.Join(placeholders, ",")
+
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, fmt.Errorf("begin tx: %w", err)
+	}
+	defer tx.Rollback()
+
+	// Count embeddings before delete
+	var embCount int64
+	tx.QueryRowContext(ctx, fmt.Sprintf(`SELECT COUNT(*) FROM embeddings WHERE message_id IN (%s)`, ph), batchArgs...).Scan(&embCount)
+
+	// Count attachments before delete
+	var attCount int64
+	tx.QueryRowContext(ctx, fmt.Sprintf(`SELECT COUNT(*) FROM attachments WHERE message_id IN (%s)`, ph), batchArgs...).Scan(&attCount)
+
+	// Collect attachment hashes
+	hashRows, _ := tx.QueryContext(ctx,
+		fmt.Sprintf(`SELECT DISTINCT hash FROM attachments WHERE message_id IN (%s)`, ph), batchArgs...)
+	var hashes []string
+	if hashRows != nil {
+		for hashRows.Next() {
+			var h string
+			hashRows.Scan(&h)
+			hashes = append(hashes, h)
+		}
+		hashRows.Close()
+	}
+
+	// Cascade delete
+	tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM embedding_queue WHERE message_id IN (%s)`, ph), batchArgs...)
+	tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM embeddings WHERE message_id IN (%s)`, ph), batchArgs...)
+	tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM attachments WHERE message_id IN (%s)`, ph), batchArgs...)
+	tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM messages WHERE id IN (%s)`, ph), batchArgs...)
+
+	// Clean orphaned conversations
+	convResult, _ := tx.ExecContext(ctx,
+		`DELETE FROM conversations WHERE id NOT IN (SELECT DISTINCT conversation_id FROM messages)`)
+	var convCount int64
+	if convResult != nil {
+		convCount, _ = convResult.RowsAffected()
+	}
+
+	if err := tx.Commit(); err != nil {
+		return nil, fmt.Errorf("commit purge: %w", err)
+	}
+
+	// Clean attachment files outside transaction
+	for _, hash := range hashes {
+		var count int
+		db.QueryRowContext(ctx, `SELECT COUNT(*) FROM attachments WHERE hash = ?`, hash).Scan(&count)
+		if count == 0 {
+			prefix := hash[:2]
+			filePath := filepath.Join(dataDir, "attachments", prefix, hash)
+			os.Remove(filePath)
+		}
+	}
+
+	return map[string]int64{
+		"deleted_messages":     int64(len(ids)),
+		"deleted_embeddings":   embCount,
+		"deleted_attachments":  attCount,
+		"cleaned_conversations": convCount,
+	}, nil
+}

internal/messaging/retention_test.go

@@ -0,0 +1,60 @@
+package messaging
+
+import (
+	"testing"
+	"time"
+)
+
+func TestParseRetentionPeriod(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		enabled  bool
+		duration time.Duration
+	}{
+		{"empty disables", "", false, 0},
+		{"zero disables", "0", false, 0},
+		{"12 months", "12m", true, 12 * 30 * 24 * time.Hour},
+		{"6 months", "6m", true, 6 * 30 * 24 * time.Hour},
+		{"1 month", "1m", true, 30 * 24 * time.Hour},
+		{"365 days", "365d", true, 365 * 24 * time.Hour},
+		{"30 days", "30d", true, 30 * 24 * time.Hour},
+		{"go duration hours", "8760h", true, 8760 * time.Hour},
+		{"invalid disables", "invalid", false, 0},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := ParseRetentionPeriod(tt.input)
+			if cfg.Enabled != tt.enabled {
+				t.Errorf("Enabled = %v, want %v", cfg.Enabled, tt.enabled)
+			}
+			if cfg.RetentionPeriod != tt.duration {
+				t.Errorf("RetentionPeriod = %v, want %v", cfg.RetentionPeriod, tt.duration)
+			}
+		})
+	}
+}
+
+func TestRetentionPeriodHuman(t *testing.T) {
+	tests := []struct {
+		name   string
+		config RetentionConfig
+		want   string
+	}{
+		{"disabled", RetentionConfig{Enabled: false}, "disabled"},
+		{"1 month", RetentionConfig{Enabled: true, RetentionPeriod: 30 * 24 * time.Hour}, "1 month"},
+		{"12 months", RetentionConfig{Enabled: true, RetentionPeriod: 360 * 24 * time.Hour}, "12 months"},
+		{"45 days", RetentionConfig{Enabled: true, RetentionPeriod: 45 * 24 * time.Hour}, "45 days"},
+		{"1 day", RetentionConfig{Enabled: true, RetentionPeriod: 24 * time.Hour}, "1 day"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.config.RetentionPeriodHuman()
+			if got != tt.want {
+				t.Errorf("RetentionPeriodHuman() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}

internal/messaging/service.go

@@ -355,6 +355,26 @@ func (s *MessagingService) SearchMessages(ctx context.Context, agentName, query
 	return messages, nil
 }
 
+// GetPendingDMCount returns the total count of pending DMs for an agent.
+func (s *MessagingService) GetPendingDMCount(ctx context.Context, agentName string) (int64, error) {
+	return s.store.CountPendingDMs(ctx, agentName)
+}
+
+// GetPendingDMs returns up to limit pending DMs for an agent, newest first.
+func (s *MessagingService) GetPendingDMs(ctx context.Context, agentName string, limit int) ([]*Message, error) {
+	return s.store.GetPendingDMs(ctx, agentName, limit)
+}
+
+// GetRecentMentions returns recent channel messages mentioning the agent.
+func (s *MessagingService) GetRecentMentions(ctx context.Context, agentName string, limit int) ([]*Message, error) {
+	return s.store.GetRecentMentions(ctx, agentName, limit)
+}
+
+// GetSystemNotifications returns pending messages from the "system" agent.
+func (s *MessagingService) GetSystemNotifications(ctx context.Context, agentName string, limit int) ([]*Message, error) {
+	return s.store.GetSystemNotifications(ctx, agentName, limit)
+}
+
 // GetMessageByID returns a single message by its ID.
 func (s *MessagingService) GetMessageByID(ctx context.Context, id int64) (*Message, error) {
 	msg, err := s.store.GetMessageByID(ctx, id)

internal/messaging/store.go

@@ -26,6 +26,10 @@ type MessageStore interface {
 	GetChannelMessages(ctx context.Context, channelID int64, limit int) ([]*Message, error)
 	GetDMMessages(ctx context.Context, agents []string, peerAgent string, limit int) ([]*Message, error)
 	AgentExists(ctx context.Context, agentName string) (bool, error)
+	CountPendingDMs(ctx context.Context, agentName string) (int64, error)
+	GetPendingDMs(ctx context.Context, agentName string, limit int) ([]*Message, error)
+	GetRecentMentions(ctx context.Context, agentName string, limit int) ([]*Message, error)
+	GetSystemNotifications(ctx context.Context, agentName string, limit int) ([]*Message, error)
 }
 
 // SQLiteMessageStore implements MessageStore using SQLite.
@@ -510,6 +514,84 @@ func (s *SQLiteMessageStore) GetDMMessages(ctx context.Context, agents []string,
 	return scanMessages(rows)
 }
 
+func (s *SQLiteMessageStore) CountPendingDMs(ctx context.Context, agentName string) (int64, error) {
+	var count int64
+	err := s.db.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM messages WHERE to_agent = ? AND status = 'pending' AND channel_id IS NULL`,
+		agentName,
+	).Scan(&count)
+	if err != nil {
+		return 0, fmt.Errorf("count pending DMs: %w", err)
+	}
+	return count, nil
+}
+
+func (s *SQLiteMessageStore) GetPendingDMs(ctx context.Context, agentName string, limit int) ([]*Message, error) {
+	if limit <= 0 {
+		limit = 10
+	}
+	rows, err := s.db.QueryContext(ctx,
+		`SELECT id, conversation_id, from_agent, to_agent, channel_id,
+		        body, priority, status, metadata, claimed_by, claimed_at,
+		        created_at, updated_at, reply_to
+		 FROM messages
+		 WHERE to_agent = ? AND status IN ('pending','processing') AND channel_id IS NULL AND from_agent != 'system'
+		 ORDER BY created_at DESC
+		 LIMIT ?`,
+		agentName, limit,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("get pending DMs: %w", err)
+	}
+	defer rows.Close()
+	return scanMessages(rows)
+}
+
+func (s *SQLiteMessageStore) GetRecentMentions(ctx context.Context, agentName string, limit int) ([]*Message, error) {
+	if limit <= 0 {
+		limit = 10
+	}
+	rows, err := s.db.QueryContext(ctx,
+		`SELECT m.id, m.conversation_id, m.from_agent, m.to_agent, m.channel_id,
+		        m.body, m.priority, m.status, m.metadata, m.claimed_by, m.claimed_at,
+		        m.created_at, m.updated_at, m.reply_to
+		 FROM messages m
+		 JOIN channel_members cm ON cm.channel_id = m.channel_id AND cm.agent_name = ?
+		 WHERE m.channel_id IS NOT NULL
+		   AND m.body LIKE '%@' || ? || '%'
+		   AND m.from_agent != ?
+		 ORDER BY m.created_at DESC
+		 LIMIT ?`,
+		agentName, agentName, agentName, limit,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("get recent mentions: %w", err)
+	}
+	defer rows.Close()
+	return scanMessages(rows)
+}
+
+func (s *SQLiteMessageStore) GetSystemNotifications(ctx context.Context, agentName string, limit int) ([]*Message, error) {
+	if limit <= 0 {
+		limit = 5
+	}
+	rows, err := s.db.QueryContext(ctx,
+		`SELECT id, conversation_id, from_agent, to_agent, channel_id,
+		        body, priority, status, metadata, claimed_by, claimed_at,
+		        created_at, updated_at, reply_to
+		 FROM messages
+		 WHERE to_agent = ? AND from_agent = 'system' AND status = 'pending'
+		 ORDER BY created_at DESC
+		 LIMIT ?`,
+		agentName, limit,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("get system notifications: %w", err)
+	}
+	defer rows.Close()
+	return scanMessages(rows)
+}
+
 func (s *SQLiteMessageStore) AgentExists(ctx context.Context, agentName string) (bool, error) {
 	var count int
 	err := s.db.QueryRowContext(ctx,

internal/search/store.go

@@ -242,3 +242,40 @@ func (s *EmbeddingStore) ClearQueue(ctx context.Context) error {
 	_, err := s.db.ExecContext(ctx, `DELETE FROM embedding_queue`)
 	return err
 }
+
+// FailedCount returns the number of failed items in the queue.
+func (s *EmbeddingStore) FailedCount(ctx context.Context) (int64, error) {
+	var count int64
+	err := s.db.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM embedding_queue WHERE status = 'failed'`,
+	).Scan(&count)
+	return count, err
+}
+
+// EmbeddingStats returns aggregate statistics about the embedding subsystem.
+type EmbeddingStatsResult struct {
+	Provider      string `json:"provider"`
+	TotalEmbedded int64  `json:"total_embedded"`
+	PendingCount  int64  `json:"pending_count"`
+	FailedCount   int64  `json:"failed_count"`
+	Dimensions    int    `json:"dimensions"`
+}
+
+// Stats returns aggregate embedding statistics.
+func (s *EmbeddingStore) Stats(ctx context.Context) (*EmbeddingStatsResult, error) {
+	stats := &EmbeddingStatsResult{}
+
+	// Get provider and dimensions from most recent embedding
+	err := s.db.QueryRowContext(ctx,
+		`SELECT COALESCE(provider, ''), COALESCE(dimensions, 0) FROM embeddings ORDER BY embedded_at DESC LIMIT 1`,
+	).Scan(&stats.Provider, &stats.Dimensions)
+	if err != nil && err != sql.ErrNoRows {
+		return nil, fmt.Errorf("get provider: %w", err)
+	}
+
+	stats.TotalEmbedded, _ = s.EmbeddingCount(ctx)
+	stats.PendingCount, _ = s.PendingCount(ctx)
+	stats.FailedCount, _ = s.FailedCount(ctx)
+
+	return stats, nil
+}

tests/integration/mcp_e2e_test.go

@@ -117,7 +117,7 @@ func setupEnv(t *testing.T) *testEnv {
 	con := console.NewWithWriter(io.Discard)
 
 	// Create MCP server
-	mcpSrv := mcpserver.NewMCPServer(msgService, agentService, channelService, swarmService, attService, searchService, con, nil, nil)
+	mcpSrv := mcpserver.NewMCPServer(msgService, agentService, channelService, swarmService, attService, searchService, con, nil, nil, db)
 	t.Cleanup(func() {
 		mcpSrv.Shutdown(context.Background())
 	})