feat(queue): add admin toggle for queued message merging (#2342)

apps/backend/internal/backendapp/message_queue_settings_test.go

@@ -44,6 +44,48 @@ func TestResolveQueueMaxPerSessionPrecedence(t *testing.T) {
 	}
 }
 
+func TestResolveQueueMergeEnabledPrecedence(t *testing.T) {
+	tests := []struct {
+		name      string
+		persisted *bool
+		want      bool
+	}{
+		{name: "no persisted setting defaults to enabled", want: true},
+		{name: "persisted disabled", persisted: new(false), want: false},
+		{name: "persisted enabled", persisted: new(true), want: true},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			pool := newMessageQueueSettingsTestPool(t)
+			if tc.persisted != nil {
+				raw, err := systemsettings.NewStore(pool)
+				if err != nil {
+					t.Fatalf("new system settings store: %v", err)
+				}
+				store := queuesettings.NewStore(raw)
+				if err := store.Save(context.Background(), queuesettings.Settings{
+					MaxPerSession: queuesettings.DefaultMaxPerSession, MergeEnabled: *tc.persisted,
+				}); err != nil {
+					t.Fatalf("save setting: %v", err)
+				}
+			}
+
+			if got := resolveQueueMergeEnabled(pool, testLogger(t)); got != tc.want {
+				t.Fatalf("merge enabled = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestResolveQueueMergeEnabledWithoutPool asserts the nil-pool fallback path
+// (mirrors resolveQueueMaxPerSession's own nil-pool behavior) still defaults
+// merging to enabled.
+func TestResolveQueueMergeEnabledWithoutPool(t *testing.T) {
+	if got := resolveQueueMergeEnabled(nil, testLogger(t)); !got {
+		t.Fatalf("merge enabled = %v, want true", got)
+	}
+}
+
 func newMessageQueueSettingsTestPool(t *testing.T) *db.Pool {
 	t.Helper()
 	connection, err := sqlx.Open("sqlite3", ":memory:")

apps/backend/internal/backendapp/orchestrator.go

@@ -94,9 +94,12 @@ func provideOrchestrator(
 		return nil, nil, fmt.Errorf("init message queue repo: %w", err)
 	}
 	maxPerSession := resolveQueueMaxPerSession(pool, log)
+	mergeEnabled := resolveQueueMergeEnabled(pool, log)
 	msgQueue := messagequeue.NewService(queueRepo, maxPerSession, log)
+	msgQueue.SetMergeEnabled(mergeEnabled)
 	log.Info("Message queue initialized",
-		zap.Int("max_per_session", maxPerSession))
+		zap.Int("max_per_session", maxPerSession),
+		zap.Bool("merge_enabled", mergeEnabled))
 	if taskSvc.AttachmentService() == nil && taskSvc.AttachmentRepository() != nil {
 		attachmentSvc, attachmentErr := taskservice.NewAttachmentService(
 			taskSvc.AttachmentRepository(), cfg.ResolvedHomeDir(), taskSvc.AuthorizeWorkspaceAccess, log,
@@ -267,6 +270,21 @@ func githubCredentialBrokerEndpoint(cfg *config.Config) string {
 // Values <= 0 disable the cap entirely (callers can still flood queues — only
 // useful in tests / specialized deployments).
 func resolveQueueMaxPerSession(pool *db.Pool, log *logger.Logger) int {
+	return resolveQueueSettings(pool, log).Effective.MaxPerSession
+}
+
+// resolveQueueMergeEnabled honors the persisted message queue setting,
+// falling back to enabled (the shipped default) when unset or invalid.
+// Unlike max_per_session it has no environment override.
+func resolveQueueMergeEnabled(pool *db.Pool, log *logger.Logger) bool {
+	return resolveQueueSettings(pool, log).Effective.MergeEnabled
+}
+
+// resolveQueueSettings loads the persisted message queue settings — falling
+// back to defaults when unset, invalid, or the store is unavailable — and
+// resolves them against the KANDEV_QUEUE_MAX_PER_SESSION environment
+// override.
+func resolveQueueSettings(pool *db.Pool, log *logger.Logger) queuesettings.Resolution {
 	var configured *queuesettings.Settings
 	if pool != nil {
 		rawStore, err := systemsettings.NewStore(pool)
@@ -282,14 +300,20 @@ func resolveQueueMaxPerSession(pool *db.Pool, log *logger.Logger) int {
 	}
 	resolution, err := queuesettings.Resolve(configured, queuesettings.ReadEnvironment())
 	if err != nil {
-		log.Warn("Failed to resolve message queue capacity, using default", zap.Error(err))
-		return messagequeue.DefaultMaxPerSession
+		log.Warn("Failed to resolve message queue settings, using defaults", zap.Error(err))
+		return queuesettings.Resolution{Response: queuesettings.Response{
+			Settings: queuesettings.DefaultSettings(),
+			Effective: queuesettings.Effective{
+				MaxPerSession: messagequeue.DefaultMaxPerSession,
+				MergeEnabled:  true,
+			},
+		}}
 	}
 	if resolution.InvalidEnvironment {
 		log.Warn("Ignoring invalid message queue capacity environment value",
 			zap.String("environment_variable", queuesettings.EnvironmentVariable))
 	}
-	return resolution.Effective.MaxPerSession
+	return resolution
 }
 
 func resolveEventNamespace(cfg *config.Config) string {

apps/backend/internal/orchestrator/handlers/queue_handlers.go

@@ -33,8 +33,11 @@ const (
 	// the combined entity references past the per-message cap; the merge is
 	// rejected atomically instead of dropping persisted references.
 	queueErrorCodeMergeReferenceOverflow = "merge_reference_overflow"
-	queueInvalidReferences               = "Invalid entity references"
-	queueAccessDenied                    = "Session not found"
+	// queueErrorCodeMergeDisabled is surfaced when queued-message merging is
+	// disabled via the message queue system setting.
+	queueErrorCodeMergeDisabled = "merge_disabled"
+	queueInvalidReferences      = "Invalid entity references"
+	queueAccessDenied           = "Session not found"
 
 	// Payload field names — extracted to satisfy goconst (≥3 occurrences).
 	fieldSessionID = "session_id"
@@ -726,6 +729,9 @@ func (h *QueueHandlers) wsMergeIntoAbove(ctx context.Context, msg *ws.Message) (
 		if errors.Is(err, messagequeue.ErrMergeReferenceOverflow) {
 			return ws.NewError(msg.ID, msg.Action, queueErrorCodeMergeReferenceOverflow, err.Error(), nil)
 		}
+		if errors.Is(err, messagequeue.ErrMergeDisabled) {
+			return ws.NewError(msg.ID, msg.Action, queueErrorCodeMergeDisabled, "Message merging is disabled", nil)
+		}
 		h.logger.Error("failed to merge queued message", zap.Error(err))
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to merge queued message", nil)
 	}
@@ -834,10 +840,11 @@ func (h *QueueHandlers) publishStatus(ctx context.Context, sessionID string) {
 	}
 	status := h.queueService.GetStatus(ctx, sessionID)
 	eventData := map[string]interface{}{
-		fieldSessionID: sessionID,
-		"entries":      status.Entries,
-		"count":        status.Count,
-		fieldMax:       status.Max,
+		fieldSessionID:  sessionID,
+		"entries":       status.Entries,
+		"count":         status.Count,
+		fieldMax:        status.Max,
+		"merge_enabled": status.MergeEnabled,
 	}
 	if h.sessionTaskResolver != nil {
 		if taskID, err := h.sessionTaskResolver(ctx, sessionID); err != nil {

apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go

@@ -96,6 +96,29 @@ func TestWsMergeIntoAbove(t *testing.T) {
 		assert.Equal(t, "entry_not_found", parseError(t, response).Code)
 	})
 
+	t.Run("returns merge_disabled when merging is turned off", func(t *testing.T) {
+		handlers, svc := setupQueueHandlers(t)
+		ctx := context.Background()
+
+		a, _ := svc.QueueMessage(ctx, "s", "t", "first", "", "user-1", false, nil)
+		b, _ := svc.QueueMessage(ctx, "s", "t", "second", "", "user-1", false, nil)
+		svc.SetMergeEnabled(false)
+
+		response, err := handlers.wsMergeIntoAbove(ctx,
+			createTestMessage(t, ws.ActionMessageQueueMerge, map[string]interface{}{
+				"session_id": "s",
+				"entry_id":   b.ID,
+				"user_id":    "user-1",
+			}))
+		require.NoError(t, err)
+		assert.Equal(t, ws.MessageTypeError, response.Type)
+		assert.Equal(t, "merge_disabled", parseError(t, response).Code)
+
+		status := svc.GetStatus(ctx, "s")
+		require.Equal(t, 2, status.Count)
+		assert.Equal(t, a.ID, status.Entries[0].ID)
+	})
+
 	t.Run("rejects reserved user_id", func(t *testing.T) {
 		handlers, _ := setupQueueHandlers(t)
 		response, err := handlers.wsMergeIntoAbove(context.Background(),

apps/backend/internal/orchestrator/messagequeue/service.go

@@ -24,6 +24,7 @@ import (
 type Service struct {
 	repo          Repository
 	maxPerSession atomic.Int64
+	mergeEnabled  atomic.Bool
 	logger        *logger.Logger
 	admissionMu   sync.Mutex
 	admissions    map[string]*sessionAdmission
@@ -51,6 +52,7 @@ func NewService(repo Repository, maxPerSession int, log *logger.Logger) *Service
 		admissions: make(map[string]*sessionAdmission),
 	}
 	service.SetMaxPerSession(maxPerSession)
+	service.mergeEnabled.Store(true)
 	return service
 }
 
@@ -78,6 +80,17 @@ func (s *Service) SetMaxPerSession(maxPerSession int) {
 	s.maxPerSession.Store(int64(maxPerSession))
 }
 
+// MergeEnabled reports whether MergeIntoAbove is currently allowed. Enabled
+// by default; see SetMergeEnabled.
+func (s *Service) MergeEnabled() bool { return s.mergeEnabled.Load() }
+
+// SetMergeEnabled toggles whether queued messages may be folded into the
+// entry above them via MergeIntoAbove. Disabling it does not affect merges
+// already applied.
+func (s *Service) SetMergeEnabled(enabled bool) {
+	s.mergeEnabled.Store(enabled)
+}
+
 // WithSessionAdmission runs fn under the per-session queue admission lock.
 // All queue insertion paths use the same lock. The callback must complete
 // synchronously; queue methods called with its context reuse the held lock.
@@ -600,6 +613,9 @@ func (s *Service) RemoveEntry(ctx context.Context, sessionID, entryID string) er
 // above it within the same session. See Repository.MergeIntoAbove for the merge
 // rules and error mapping (ErrEntryNotFound / ErrNoMergeTarget).
 func (s *Service) MergeIntoAbove(ctx context.Context, sessionID, entryID, queuedBy string) (*QueuedMessage, error) {
+	if !s.MergeEnabled() {
+		return nil, ErrMergeDisabled
+	}
 	merged, err := s.repo.MergeIntoAbove(ctx, sessionID, entryID, queuedBy)
 	if err != nil {
 		return nil, err
@@ -626,12 +642,13 @@ func (s *Service) CancelAll(ctx context.Context, sessionID string) (int, error)
 // GetStatus returns the full pending list and capacity info for a session.
 func (s *Service) GetStatus(ctx context.Context, sessionID string) *QueueStatus {
 	maxPerSession := s.MaxPerSession()
+	mergeEnabled := s.MergeEnabled()
 	entries, err := s.repo.ListBySession(ctx, sessionID)
 	if err != nil {
 		s.logger.Error("list queued failed",
 			zap.String("session_id", sessionID),
 			zap.Error(err))
-		return &QueueStatus{Entries: []QueuedMessage{}, Count: 0, Max: maxPerSession}
+		return &QueueStatus{Entries: []QueuedMessage{}, Count: 0, Max: maxPerSession, MergeEnabled: mergeEnabled}
 	}
 	pending := make([]QueuedMessage, 0, len(entries))
 	for _, entry := range entries {
@@ -643,9 +660,10 @@ func (s *Service) GetStatus(ctx context.Context, sessionID string) *QueueStatus
 		pending = append(pending, entry)
 	}
 	return &QueueStatus{
-		Entries: pending,
-		Count:   len(pending),
-		Max:     maxPerSession,
+		Entries:      pending,
+		Count:        len(pending),
+		Max:          maxPerSession,
+		MergeEnabled: mergeEnabled,
 	}
 }
 

apps/backend/internal/orchestrator/messagequeue/service_merge_enabled_test.go

@@ -0,0 +1,54 @@
+package messagequeue
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestService_MergeEnabledDefaultsToTrue asserts a freshly constructed
+// Service allows merges out of the box, matching the "enabled by default"
+// product requirement for queued-message merging.
+func TestService_MergeEnabledDefaultsToTrue(t *testing.T) {
+	svc := setupService(t)
+	assert.True(t, svc.MergeEnabled())
+}
+
+// TestService_MergeIntoAbove_DisabledRejectsMerge asserts SetMergeEnabled(false)
+// makes MergeIntoAbove fail with ErrMergeDisabled and leaves the queue
+// untouched, without needing per-call plumbing through every merge call site.
+func TestService_MergeIntoAbove_DisabledRejectsMerge(t *testing.T) {
+	svc := setupService(t)
+	ctx := context.Background()
+
+	_, err := svc.QueueMessage(ctx, "s", "t", "above", "", "alice", false, nil)
+	require.NoError(t, err)
+	source, err := svc.QueueMessage(ctx, "s", "t", "source", "", "alice", false, nil)
+	require.NoError(t, err)
+
+	svc.SetMergeEnabled(false)
+
+	_, err = svc.MergeIntoAbove(ctx, "s", source.ID, "alice")
+	assert.ErrorIs(t, err, ErrMergeDisabled)
+	assert.Equal(t, 2, svc.GetStatus(ctx, "s").Count)
+
+	svc.SetMergeEnabled(true)
+	merged, err := svc.MergeIntoAbove(ctx, "s", source.ID, "alice")
+	require.NoError(t, err)
+	assert.Equal(t, "above\n\nsource", merged.Content)
+}
+
+// TestGetStatus_ReflectsMergeEnabled asserts QueueStatus.MergeEnabled mirrors
+// the live flag, so clients can hide the merge affordance without a separate
+// settings round trip.
+func TestGetStatus_ReflectsMergeEnabled(t *testing.T) {
+	svc := setupService(t)
+	ctx := context.Background()
+
+	assert.True(t, svc.GetStatus(ctx, "s").MergeEnabled)
+
+	svc.SetMergeEnabled(false)
+	assert.False(t, svc.GetStatus(ctx, "s").MergeEnabled)
+}

apps/backend/internal/orchestrator/messagequeue/types.go

@@ -89,6 +89,10 @@ var (
 	// sender tasks differ, the caller does not own the rows, or the target is a
 	// reserved in-flight lifecycle entry.
 	ErrNoMergeTarget = errors.New("no mergeable message above")
+	// ErrMergeDisabled is returned when a merge is attempted while queued
+	// message merging is disabled (see Service.SetMergeEnabled). The setting
+	// is admin-controlled and enabled by default.
+	ErrMergeDisabled = errors.New("queued message merging is disabled")
 	// ErrTaskInactive means a lifecycle prompt could not be accepted because
 	// its task was deleted or archived before the queue transaction claimed it.
 	ErrTaskInactive = errors.New("queue task is inactive")
@@ -187,6 +191,9 @@ type QueueStatus struct {
 	Entries []QueuedMessage `json:"entries"`
 	Count   int             `json:"count"`
 	Max     int             `json:"max"`
+	// MergeEnabled mirrors Service.MergeEnabled so clients can hide the
+	// "Merge with above" affordance without a separate settings fetch.
+	MergeEnabled bool `json:"merge_enabled"`
 }
 
 // PendingMove represents a workflow step move requested by an agent (via

apps/backend/internal/system/queuesettings/handler.go

@@ -25,14 +25,17 @@ func handleGet(service *Service) gin.HandlerFunc {
 	}
 }
 
+// handleUpdate applies a partial SettingsPatch: the request body may name
+// only the fields it wants to change, and any field it omits keeps its
+// current persisted value (see Service.Update).
 func handleUpdate(service *Service) gin.HandlerFunc {
 	return func(ctx *gin.Context) {
-		var settings Settings
-		if err := ctx.ShouldBindJSON(&settings); err != nil {
+		var patch SettingsPatch
+		if err := ctx.ShouldBindJSON(&patch); err != nil {
 			ctx.JSON(http.StatusBadRequest, gin.H{responseErrorKey: "invalid request"})
 			return
 		}
-		response, err := service.Update(ctx.Request.Context(), settings)
+		response, err := service.Update(ctx.Request.Context(), patch)
 		if err != nil {
 			switch {
 			case errors.Is(err, ErrValidation):

apps/backend/internal/system/queuesettings/queuesettings_test.go

@@ -24,12 +24,19 @@ func TestResolvePrecedenceAndNormalization(t *testing.T) {
 		want        Response
 		invalidEnv  bool
 	}{
-		{name: "default", want: responseFor(10, 10, SourceDefault, false)},
-		{name: "setting", configured: &Settings{MaxPerSession: 6}, want: responseFor(6, 6, SourceSetting, false)},
-		{name: "environment", configured: &Settings{MaxPerSession: 6}, environment: Environment{Value: "20", Present: true}, want: responseFor(6, 20, SourceEnvironment, true)},
-		{name: "zero environment is unlimited", configured: &Settings{MaxPerSession: 6}, environment: Environment{Value: "0", Present: true}, want: responseFor(6, 0, SourceEnvironment, true)},
-		{name: "negative environment is unlimited", configured: &Settings{MaxPerSession: 6}, environment: Environment{Value: "-3", Present: true}, want: responseFor(6, 0, SourceEnvironment, true)},
-		{name: "invalid environment is ignored", configured: &Settings{MaxPerSession: 6}, environment: Environment{Value: "many", Present: true}, want: responseFor(6, 6, SourceSetting, false), invalidEnv: true},
+		{name: "default", want: responseFor(10, 10, SourceDefault, false, true)},
+		{name: "setting", configured: &Settings{MaxPerSession: 6, MergeEnabled: true}, want: responseFor(6, 6, SourceSetting, false, true)},
+		{name: "environment", configured: &Settings{MaxPerSession: 6, MergeEnabled: true}, environment: Environment{Value: "20", Present: true}, want: responseFor(6, 20, SourceEnvironment, true, true)},
+		{name: "zero environment is unlimited", configured: &Settings{MaxPerSession: 6, MergeEnabled: true}, environment: Environment{Value: "0", Present: true}, want: responseFor(6, 0, SourceEnvironment, true, true)},
+		{name: "negative environment is unlimited", configured: &Settings{MaxPerSession: 6, MergeEnabled: true}, environment: Environment{Value: "-3", Present: true}, want: responseFor(6, 0, SourceEnvironment, true, true)},
+		{name: "invalid environment is ignored", configured: &Settings{MaxPerSession: 6, MergeEnabled: true}, environment: Environment{Value: "many", Present: true}, want: responseFor(6, 6, SourceSetting, false, true), invalidEnv: true},
+		{name: "merge disabled setting", configured: &Settings{MaxPerSession: 6, MergeEnabled: false}, want: responseFor(6, 6, SourceSetting, false, false)},
+		{
+			name:        "merge disabled setting survives an environment override of max_per_session",
+			configured:  &Settings{MaxPerSession: 6, MergeEnabled: false},
+			environment: Environment{Value: "20", Present: true},
+			want:        responseFor(6, 20, SourceEnvironment, true, false),
+		},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
@@ -58,11 +65,11 @@ func TestStoreRoundTripAndValidation(t *testing.T) {
 	if err != nil || loaded != nil {
 		t.Fatalf("missing load = %+v, %v", loaded, err)
 	}
-	if err := store.Save(context.Background(), Settings{MaxPerSession: 8}); err != nil {
+	if err := store.Save(context.Background(), Settings{MaxPerSession: 8, MergeEnabled: true}); err != nil {
 		t.Fatalf("save: %v", err)
 	}
 	loaded, err = store.Load(context.Background())
-	if err != nil || loaded == nil || loaded.MaxPerSession != 8 {
+	if err != nil || loaded == nil || loaded.MaxPerSession != 8 || !loaded.MergeEnabled {
 		t.Fatalf("round trip = %+v, %v", loaded, err)
 	}
 	if err := store.Save(context.Background(), Settings{MaxPerSession: -1}); !errors.Is(err, ErrValidation) {
@@ -70,12 +77,49 @@ func TestStoreRoundTripAndValidation(t *testing.T) {
 	}
 }
 
+// TestStoreLoadDefaultsMergeEnabledForPreExistingRecords guards a real
+// upgrade hazard: an installation that persisted max_per_session before
+// merge_enabled existed has a stored JSON object with no "merge_enabled"
+// key. json.Unmarshal into a plain bool decodes that as false, which would
+// silently disable merging on upgrade instead of leaving it enabled by
+// default. Store.Load must treat the missing key as "unset" (-> true), not
+// "explicitly false".
+func TestStoreLoadDefaultsMergeEnabledForPreExistingRecords(t *testing.T) {
+	raw := &fakeRawStore{raw: []byte(`{"max_per_session":6}`), found: true}
+	store := NewStore(raw)
+
+	loaded, err := store.Load(context.Background())
+	if err != nil {
+		t.Fatalf("load pre-existing record: %v", err)
+	}
+	if loaded == nil || loaded.MaxPerSession != 6 || !loaded.MergeEnabled {
+		t.Fatalf("loaded = %+v, want {MaxPerSession:6 MergeEnabled:true}", loaded)
+	}
+}
+
+// TestStoreLoadPreservesExplicitMergeDisabled is the companion case: once a
+// record has been re-saved through this code path with merge_enabled
+// explicitly false, Load must not treat that as "unset" and re-default it to
+// true.
+func TestStoreLoadPreservesExplicitMergeDisabled(t *testing.T) {
+	raw := &fakeRawStore{raw: []byte(`{"max_per_session":6,"merge_enabled":false}`), found: true}
+	store := NewStore(raw)
+
+	loaded, err := store.Load(context.Background())
+	if err != nil {
+		t.Fatalf("load explicit-false record: %v", err)
+	}
+	if loaded == nil || loaded.MergeEnabled {
+		t.Fatalf("loaded = %+v, want MergeEnabled:false preserved", loaded)
+	}
+}
+
 func TestServiceUpdatePersistsBeforeLiveApply(t *testing.T) {
 	raw := &fakeRawStore{}
 	target := &fakeTarget{max: 10}
 	service := NewService(NewStore(raw), target, func() Environment { return Environment{} }, testLogger(t))
 
-	response, err := service.Update(context.Background(), Settings{MaxPerSession: 4})
+	response, err := service.Update(context.Background(), SettingsPatch{MaxPerSession: new(4)})
 	if err != nil {
 		t.Fatalf("update: %v", err)
 	}
@@ -84,7 +128,7 @@ func TestServiceUpdatePersistsBeforeLiveApply(t *testing.T) {
 	}
 
 	raw.saveErr = errors.New("disk full")
-	_, err = service.Update(context.Background(), Settings{MaxPerSession: 2})
+	_, err = service.Update(context.Background(), SettingsPatch{MaxPerSession: new(2)})
 	if err == nil {
 		t.Fatal("expected persistence error")
 	}
@@ -93,6 +137,39 @@ func TestServiceUpdatePersistsBeforeLiveApply(t *testing.T) {
 	}
 }
 
+// TestServiceUpdatePatchPreservesUnspecifiedFields guards against a patch
+// that touches only one field silently resetting the other to its zero
+// value: a max_per_session-only patch must not disable merging, and a
+// merge_enabled-only patch must not reset the capacity.
+func TestServiceUpdatePatchPreservesUnspecifiedFields(t *testing.T) {
+	raw := &fakeRawStore{}
+	target := &fakeTarget{max: 10}
+	service := NewService(NewStore(raw), target, func() Environment { return Environment{} }, testLogger(t))
+
+	if _, err := service.Update(context.Background(), SettingsPatch{MergeEnabled: new(true)}); err != nil {
+		t.Fatalf("seed merge_enabled: %v", err)
+	}
+
+	response, err := service.Update(context.Background(), SettingsPatch{MaxPerSession: new(4)})
+	if err != nil {
+		t.Fatalf("max-only update: %v", err)
+	}
+	if !response.Settings.MergeEnabled || !target.MergeEnabled() {
+		t.Fatalf("max-only patch reset merge_enabled: response=%+v target=%v", response, target.MergeEnabled())
+	}
+
+	response, err = service.Update(context.Background(), SettingsPatch{MergeEnabled: new(false)})
+	if err != nil {
+		t.Fatalf("merge-only update: %v", err)
+	}
+	if response.Settings.MaxPerSession != 4 || target.MaxPerSession() != 4 {
+		t.Fatalf("merge-only patch reset max_per_session: response=%+v target=%d", response, target.MaxPerSession())
+	}
+	if response.Settings.MergeEnabled || target.MergeEnabled() {
+		t.Fatalf("merge-only patch did not apply: response=%+v target=%v", response, target.MergeEnabled())
+	}
+}
+
 func TestServiceRecoversFromInvalidPersistedSettingAndAllowsReplacement(t *testing.T) {
 	tests := []struct {
 		name string
@@ -113,15 +190,15 @@ func TestServiceRecoversFromInvalidPersistedSettingAndAllowsReplacement(t *testi
 			if err != nil {
 				t.Fatalf("get with invalid persisted setting: %v", err)
 			}
-			if response != responseFor(10, 10, SourceDefault, false) {
+			if response != responseFor(10, 10, SourceDefault, false, true) {
 				t.Fatalf("fallback response = %+v", response)
 			}
 
-			response, err = service.Update(context.Background(), Settings{MaxPerSession: 7})
+			response, err = service.Update(context.Background(), SettingsPatch{MaxPerSession: new(7)})
 			if err != nil {
 				t.Fatalf("replace invalid persisted setting: %v", err)
 			}
-			if response != responseFor(7, 7, SourceSetting, false) || target.max != 7 {
+			if response != responseFor(7, 7, SourceSetting, false, true) || target.max != 7 {
 				t.Fatalf("replacement response=%+v target=%d", response, target.max)
 			}
 		})
@@ -138,14 +215,14 @@ func TestServiceSerializesPersistenceAndLiveApply(t *testing.T) {
 
 	firstDone := make(chan error, 1)
 	go func() {
-		_, err := service.Update(context.Background(), Settings{MaxPerSession: 7})
+		_, err := service.Update(context.Background(), SettingsPatch{MaxPerSession: new(7)})
 		firstDone <- err
 	}()
 	<-raw.firstSaveWritten
 
 	secondDone := make(chan error, 1)
 	go func() {
-		_, err := service.Update(context.Background(), Settings{MaxPerSession: 9})
+		_, err := service.Update(context.Background(), SettingsPatch{MaxPerSession: new(9)})
 		secondDone <- err
 	}()
 
@@ -181,7 +258,7 @@ func TestServiceEnvironmentLockRejectsUpdate(t *testing.T) {
 		return Environment{Value: "20", Present: true}
 	}, testLogger(t))
 
-	_, err := service.Update(context.Background(), Settings{MaxPerSession: 4})
+	_, err := service.Update(context.Background(), SettingsPatch{MaxPerSession: new(4)})
 	if !errors.Is(err, ErrEnvironmentLocked) {
 		t.Fatalf("update error = %v, want environment lock", err)
 	}
@@ -190,6 +267,25 @@ func TestServiceEnvironmentLockRejectsUpdate(t *testing.T) {
 	}
 }
 
+// TestServiceEnvironmentLockAllowsMergeOnlyUpdate asserts the environment
+// lock only blocks a patch that touches max_per_session — merge_enabled has
+// no environment override and must stay editable.
+func TestServiceEnvironmentLockAllowsMergeOnlyUpdate(t *testing.T) {
+	raw := &fakeRawStore{}
+	target := &fakeTarget{max: 20}
+	service := NewService(NewStore(raw), target, func() Environment {
+		return Environment{Value: "20", Present: true}
+	}, testLogger(t))
+
+	response, err := service.Update(context.Background(), SettingsPatch{MergeEnabled: new(false)})
+	if err != nil {
+		t.Fatalf("merge-only update under max_per_session lock: %v", err)
+	}
+	if response.Settings.MergeEnabled || target.MergeEnabled() {
+		t.Fatalf("merge-only update did not apply: response=%+v target=%v", response, target.MergeEnabled())
+	}
+}
+
 func TestHandlerReturnsConflictForEnvironmentLock(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 	raw := &fakeRawStore{}
@@ -200,7 +296,7 @@ func TestHandlerReturnsConflictForEnvironmentLock(t *testing.T) {
 	group := router.Group("/api/v1/system")
 	RegisterRoutes(group, group, service)
 
-	body, _ := json.Marshal(Settings{MaxPerSession: 4})
+	body, _ := json.Marshal(SettingsPatch{MaxPerSession: new(4)})
 	request := httptest.NewRequest(http.MethodPatch, "/api/v1/system/message-queue/settings", bytes.NewReader(body))
 	request.Header.Set("Content-Type", "application/json")
 	response := httptest.NewRecorder()
@@ -214,7 +310,7 @@ func TestHandlerGetReturnsConfiguredAndEffectiveValues(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 	raw := &fakeRawStore{}
 	store := NewStore(raw)
-	if err := store.Save(context.Background(), Settings{MaxPerSession: 6}); err != nil {
+	if err := store.Save(context.Background(), Settings{MaxPerSession: 6, MergeEnabled: true}); err != nil {
 		t.Fatalf("save baseline: %v", err)
 	}
 	service := NewService(store, &fakeTarget{max: 20}, func() Environment {
@@ -235,7 +331,7 @@ func TestHandlerGetReturnsConfiguredAndEffectiveValues(t *testing.T) {
 	if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil {
 		t.Fatalf("decode response: %v", err)
 	}
-	want := responseFor(6, 20, SourceEnvironment, true)
+	want := responseFor(6, 20, SourceEnvironment, true, true)
 	if payload != want {
 		t.Fatalf("response = %+v, want %+v", payload, want)
 	}
@@ -262,10 +358,45 @@ func TestHandlerRejectsNegativeCapacity(t *testing.T) {
 	}
 }
 
-func responseFor(configured, effective int, source Source, locked bool) Response {
+// TestHandlerUpdateOmittingMergeEnabledPreservesIt is the HTTP-level
+// regression test for the patch-not-replace contract: a PATCH body that
+// only names max_per_session must never flip a previously-saved
+// merge_enabled=true back to false.
+func TestHandlerUpdateOmittingMergeEnabledPreservesIt(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	raw := &fakeRawStore{}
+	target := &fakeTarget{max: 10, mergeEnabled: true}
+	service := NewService(NewStore(raw), target, func() Environment { return Environment{} }, testLogger(t))
+	router := gin.New()
+	group := router.Group("/api/v1/system")
+	RegisterRoutes(group, group, service)
+
+	request := httptest.NewRequest(
+		http.MethodPatch, "/api/v1/system/message-queue/settings",
+		bytes.NewBufferString(`{"max_per_session":5}`),
+	)
+	request.Header.Set("Content-Type", "application/json")
+	response := httptest.NewRecorder()
+	router.ServeHTTP(response, request)
+	if response.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200; body=%s", response.Code, response.Body.String())
+	}
+	var payload Response
+	if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if !payload.Settings.MergeEnabled || !target.MergeEnabled() {
+		t.Fatalf("max_per_session-only PATCH disabled merging: payload=%+v target=%v", payload, target.MergeEnabled())
+	}
+	if payload.Settings.MaxPerSession != 5 || target.MaxPerSession() != 5 {
+		t.Fatalf("max_per_session was not applied: payload=%+v target=%d", payload, target.MaxPerSession())
+	}
+}
+
+func responseFor(configured, effective int, source Source, locked bool, mergeEnabled bool) Response {
 	return Response{
-		Settings:  Settings{MaxPerSession: configured},
-		Effective: Effective{MaxPerSession: effective, Source: source, Locked: locked},
+		Settings:  Settings{MaxPerSession: configured, MergeEnabled: mergeEnabled},
+		Effective: Effective{MaxPerSession: effective, Source: source, Locked: locked, MergeEnabled: mergeEnabled},
 	}
 }
 
@@ -285,23 +416,26 @@ func (f *fakeRawStore) Save(_ context.Context, _ string, value []byte) error {
 	if f.saveErr != nil {
 		return f.saveErr
 	}
-	f.raw = append([]byte(nil), value...)
+	f.raw = value
 	f.found = true
 	return nil
 }
 
 type fakeTarget struct {
-	max int
+	max          int
+	mergeEnabled bool
 }
 
 func (f *fakeTarget) MaxPerSession() int     { return f.max }
 func (f *fakeTarget) SetMaxPerSession(n int) { f.max = n }
+func (f *fakeTarget) MergeEnabled() bool     { return f.mergeEnabled }
+func (f *fakeTarget) SetMergeEnabled(v bool) { f.mergeEnabled = v }
 
 type blockingRawStore struct {
 	mu                sync.Mutex
 	raw               []byte
 	found             bool
-	saveCalls         int
+	saveCount         int
 	firstSaveWritten  chan struct{}
 	secondSaveEntered chan struct{}
 	releaseFirst      chan struct{}
@@ -318,40 +452,48 @@ func newBlockingRawStore() *blockingRawStore {
 func (s *blockingRawStore) Get(context.Context, string) ([]byte, bool, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	return append([]byte(nil), s.raw...), s.found, nil
+	return s.raw, s.found, nil
 }
 
 func (s *blockingRawStore) Save(_ context.Context, _ string, value []byte) error {
 	s.mu.Lock()
-	s.saveCalls++
-	call := s.saveCalls
-	s.raw = append([]byte(nil), value...)
-	s.found = true
-	switch call {
-	case 1:
-		close(s.firstSaveWritten)
-	case 2:
-		close(s.secondSaveEntered)
-	}
+	s.saveCount++
+	count := s.saveCount
 	s.mu.Unlock()
-	if call == 1 {
+
+	if count == 1 {
+		s.mu.Lock()
+		s.raw = value
+		s.found = true
+		s.mu.Unlock()
+		close(s.firstSaveWritten)
 		<-s.releaseFirst
+		return nil
 	}
+
+	close(s.secondSaveEntered)
+	s.mu.Lock()
+	s.raw = value
+	s.found = true
+	s.mu.Unlock()
 	return nil
 }
 
 type atomicTarget struct {
-	max atomic.Int64
+	max          atomic.Int64
+	mergeEnabled atomic.Bool
 }
 
 func (t *atomicTarget) MaxPerSession() int     { return int(t.max.Load()) }
 func (t *atomicTarget) SetMaxPerSession(n int) { t.max.Store(int64(n)) }
+func (t *atomicTarget) MergeEnabled() bool     { return t.mergeEnabled.Load() }
+func (t *atomicTarget) SetMergeEnabled(v bool) { t.mergeEnabled.Store(v) }
 
 func testLogger(t *testing.T) *logger.Logger {
 	t.Helper()
 	log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console", OutputPath: "stderr"})
 	if err != nil {
-		t.Fatalf("logger: %v", err)
+		t.Fatalf("new logger: %v", err)
 	}
 	return log
 }

apps/backend/internal/system/queuesettings/resolver.go

@@ -23,6 +23,7 @@ func Resolve(configured *Settings, environment Environment) (Resolution, error)
 		Effective: Effective{
 			MaxPerSession: settings.MaxPerSession,
 			Source:        source,
+			MergeEnabled:  settings.MergeEnabled,
 		},
 	}}
 	raw := strings.TrimSpace(environment.Value)
@@ -41,6 +42,7 @@ func Resolve(configured *Settings, environment Environment) (Resolution, error)
 		MaxPerSession: value,
 		Source:        SourceEnvironment,
 		Locked:        true,
+		MergeEnabled:  settings.MergeEnabled,
 	}
 	return resolution, nil
 }

apps/backend/internal/system/queuesettings/service.go

@@ -13,6 +13,8 @@ import (
 type Target interface {
 	MaxPerSession() int
 	SetMaxPerSession(int)
+	MergeEnabled() bool
+	SetMergeEnabled(bool)
 }
 
 type EnvironmentReader func() Environment
@@ -53,13 +55,15 @@ func (s *Service) Get(ctx context.Context) (Response, error) {
 	return resolution.Response, nil
 }
 
-func (s *Service) Update(ctx context.Context, settings Settings) (Response, error) {
+// Update applies a partial patch to the persisted settings: fields the caller
+// omits keep their current effective value rather than resetting to zero, so
+// a client that edits only one field can never silently clobber another (see
+// SettingsPatch). The environment lock only blocks a patch that actually
+// attempts to change max_per_session — it does not control merge_enabled.
+func (s *Service) Update(ctx context.Context, patch SettingsPatch) (Response, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	if err := Validate(settings); err != nil {
-		return Response{}, err
-	}
 	environment := s.readEnvironment()
 	current, err := s.loadConfigured(ctx)
 	if err != nil {
@@ -70,9 +74,13 @@ func (s *Service) Update(ctx context.Context, settings Settings) (Response, erro
 		return Response{}, err
 	}
 	s.warnInvalidEnvironment(resolution)
-	if resolution.Effective.Locked {
+	if patch.MaxPerSession != nil && resolution.Effective.Locked {
 		return Response{}, ErrEnvironmentLocked
 	}
+	settings := patch.Apply(resolution.Settings)
+	if err := Validate(settings); err != nil {
+		return Response{}, err
+	}
 	if s.target == nil {
 		return Response{}, ErrTargetUnavailable
 	}
@@ -80,6 +88,7 @@ func (s *Service) Update(ctx context.Context, settings Settings) (Response, erro
 		return Response{}, err
 	}
 	s.target.SetMaxPerSession(settings.MaxPerSession)
+	s.target.SetMergeEnabled(settings.MergeEnabled)
 	updated, err := Resolve(&settings, environment)
 	if err != nil {
 		return Response{}, err

apps/backend/internal/system/queuesettings/store.go

@@ -19,6 +19,27 @@ func NewStore(raw RawStore) *Store {
 	return &Store{raw: raw}
 }
 
+// storedSettings mirrors Settings for JSON decoding but keeps MergeEnabled as
+// a pointer so a missing key ("unset" — a record saved before this field
+// existed) can be told apart from an explicit `false`. Without this, an
+// installation that persisted max_per_session before merge_enabled shipped
+// would decode its stale record as merge_enabled=false and silently disable
+// merging on upgrade instead of leaving it enabled by default.
+type storedSettings struct {
+	MaxPerSession int   `json:"max_per_session"`
+	MergeEnabled  *bool `json:"merge_enabled"`
+}
+
+// toSettings normalizes a decoded record into Settings, defaulting
+// MergeEnabled to true when the persisted JSON has no merge_enabled key.
+func (s storedSettings) toSettings() Settings {
+	mergeEnabled := true
+	if s.MergeEnabled != nil {
+		mergeEnabled = *s.MergeEnabled
+	}
+	return Settings{MaxPerSession: s.MaxPerSession, MergeEnabled: mergeEnabled}
+}
+
 func (s *Store) Load(ctx context.Context) (*Settings, error) {
 	raw, found, err := s.raw.Get(ctx, SettingsKey)
 	if err != nil {
@@ -27,10 +48,11 @@ func (s *Store) Load(ctx context.Context) (*Settings, error) {
 	if !found {
 		return nil, nil
 	}
-	var settings Settings
-	if err := json.Unmarshal(raw, &settings); err != nil {
+	var stored storedSettings
+	if err := json.Unmarshal(raw, &stored); err != nil {
 		return nil, fmt.Errorf("%w: decode JSON: %v", ErrInvalidPersisted, err)
 	}
+	settings := stored.toSettings()
 	if err := Validate(settings); err != nil {
 		return nil, fmt.Errorf("%w: %v", ErrInvalidPersisted, err)
 	}

apps/backend/internal/system/queuesettings/types.go

@@ -29,18 +29,43 @@ var (
 )
 
 type Settings struct {
-	MaxPerSession int `json:"max_per_session"`
+	MaxPerSession int  `json:"max_per_session"`
+	MergeEnabled  bool `json:"merge_enabled"`
+}
+
+// SettingsPatch is a partial update to Settings: a nil field means "leave
+// unchanged" rather than "reset to zero value". Without this distinction a
+// client that PATCHes only max_per_session would silently reset
+// merge_enabled to false, since Settings has no way to represent omission.
+type SettingsPatch struct {
+	MaxPerSession *int  `json:"max_per_session"`
+	MergeEnabled  *bool `json:"merge_enabled"`
+}
+
+// Apply returns base with every non-nil patch field overlaid.
+func (p SettingsPatch) Apply(base Settings) Settings {
+	if p.MaxPerSession != nil {
+		base.MaxPerSession = *p.MaxPerSession
+	}
+	if p.MergeEnabled != nil {
+		base.MergeEnabled = *p.MergeEnabled
+	}
+	return base
+}
+
+type Response struct {
+	Settings  Settings  `json:"settings"`
+	Effective Effective `json:"effective"`
 }
 
 type Effective struct {
 	MaxPerSession int    `json:"max_per_session"`
 	Source        Source `json:"source"`
 	Locked        bool   `json:"locked"`
-}
-
-type Response struct {
-	Settings  Settings  `json:"settings"`
-	Effective Effective `json:"effective"`
+	// MergeEnabled always mirrors Settings.MergeEnabled: unlike MaxPerSession
+	// it has no environment override, so there is no separate source/lock to
+	// track.
+	MergeEnabled bool `json:"merge_enabled"`
 }
 
 type Environment struct {
@@ -53,8 +78,10 @@ type Resolution struct {
 	InvalidEnvironment bool
 }
 
+// DefaultSettings returns the shipped defaults: no session-count cap and
+// merging enabled.
 func DefaultSettings() Settings {
-	return Settings{MaxPerSession: DefaultMaxPerSession}
+	return Settings{MaxPerSession: DefaultMaxPerSession, MergeEnabled: true}
 }
 
 func Validate(settings Settings) error {

apps/backend/internal/system/system_routes_test.go

@@ -158,11 +158,14 @@ func (s *queueSettingsRawStore) Save(_ context.Context, _ string, value []byte)
 }
 
 type queueSettingsTarget struct {
-	max int
+	max          int
+	mergeEnabled bool
 }
 
 func (t *queueSettingsTarget) MaxPerSession() int     { return t.max }
 func (t *queueSettingsTarget) SetMaxPerSession(n int) { t.max = n }
+func (t *queueSettingsTarget) MergeEnabled() bool     { return t.mergeEnabled }
+func (t *queueSettingsTarget) SetMergeEnabled(v bool) { t.mergeEnabled = v }
 
 type systemSleepRawStore struct {
 	raw   []byte

apps/web/components/settings/system/message-queue-settings.test.tsx

@@ -8,6 +8,7 @@ const updateSettingsMock = vi.fn();
 let saveContributor: SettingsSaveContributor | null = null;
 let currentRole: "admin" | "member" | undefined;
 const MAXIMUM_LABEL = "Maximum messages per session";
+const MERGE_TOGGLE_LABEL = "Enable queued message merging";
 const EFFECTIVE_VALUE_ID = "message-queue-effective-value";
 
 vi.mock("@/lib/api/domains/settings-api", () => ({
@@ -26,6 +27,28 @@ vi.mock("../settings-save-provider", () => ({
   },
 }));
 
+vi.mock("@kandev/ui/switch", () => ({
+  Switch: ({
+    checked,
+    disabled,
+    "aria-label": ariaLabel,
+    onCheckedChange,
+  }: {
+    checked: boolean;
+    disabled: boolean;
+    "aria-label": string;
+    onCheckedChange: (checked: boolean) => void;
+  }) => (
+    <button
+      aria-label={ariaLabel}
+      aria-pressed={checked}
+      disabled={disabled}
+      type="button"
+      onClick={() => onCheckedChange(!checked)}
+    />
+  ),
+}));
+
 import { MessageQueueSettings } from "./message-queue-settings";
 
 function response(
@@ -33,13 +56,26 @@ function response(
   effective = configured,
   source: MessageQueueSettingsResponse["effective"]["source"] = "default",
   locked = false,
+  mergeEnabled = true,
 ): MessageQueueSettingsResponse {
   return {
-    settings: { max_per_session: configured },
-    effective: { max_per_session: effective, source, locked },
+    settings: { max_per_session: configured, merge_enabled: mergeEnabled },
+    effective: { max_per_session: effective, source, locked, merge_enabled: mergeEnabled },
   };
 }
 
+/** Every save/discard test needs the contributor the component just
+ * registered; centralizing the "not registered yet" guard keeps that
+ * assertion's message out of `sonarjs/no-duplicate-string` territory. */
+function requireContributor(): SettingsSaveContributor {
+  if (!saveContributor) throw new Error("save contributor was not registered");
+  return saveContributor;
+}
+
+function mergeToggle(): HTMLElement {
+  return screen.getByRole("button", { name: MERGE_TOGGLE_LABEL });
+}
+
 beforeEach(() => {
   fetchSettingsMock.mockReset();
   updateSettingsMock.mockReset();
@@ -53,7 +89,7 @@ afterEach(() => {
   vi.clearAllMocks();
 });
 
-describe("MessageQueueSettings", () => {
+describe("MessageQueueSettings — per-session limit", () => {
   it("shows loading, configured, effective, and default-source states", async () => {
     let resolveLoad: (value: MessageQueueSettingsResponse) => void = () => {};
     fetchSettingsMock.mockReturnValueOnce(
@@ -97,9 +133,9 @@ describe("MessageQueueSettings", () => {
     expect(saveContributor?.isDirty).toBe(true);
     expect(screen.queryByRole("button", { name: "Save" })).toBeNull();
     expect(screen.queryByRole("button", { name: "Cancel" })).toBeNull();
-    if (!saveContributor) throw new Error("save contributor was not registered");
+    const contributor = requireContributor();
 
-    await act(async () => saveContributor?.save(saveContributor.revision));
+    await act(async () => contributor.save(contributor.revision));
 
     expect(updateSettingsMock).toHaveBeenCalledWith({ max_per_session: 25 });
     expect(screen.getByTestId(EFFECTIVE_VALUE_ID).textContent).toBe("25");
@@ -154,10 +190,10 @@ describe("MessageQueueSettings", () => {
     render(<MessageQueueSettings />);
     const input = await screen.findByLabelText(MAXIMUM_LABEL);
     fireEvent.change(input, { target: { value: "20" } });
-    if (!saveContributor) throw new Error("save contributor was not registered");
+    const contributor = requireContributor();
 
     await act(async () => {
-      await expect(saveContributor?.save(saveContributor.revision)).rejects.toThrow("save failed");
+      await expect(contributor.save(contributor.revision)).rejects.toThrow("save failed");
     });
 
     expect(screen.getByText("Failed to save message queue settings.")).toBeTruthy();
@@ -169,11 +205,93 @@ describe("MessageQueueSettings", () => {
     render(<MessageQueueSettings />);
     const input = await screen.findByLabelText(MAXIMUM_LABEL);
     fireEvent.change(input, { target: { value: "30" } });
-    if (!saveContributor) throw new Error("save contributor was not registered");
+    const contributor = requireContributor();
 
-    act(() => saveContributor?.discard());
+    act(() => contributor.discard());
 
     expect(input.getAttribute("value")).toBe("12");
     expect(saveContributor?.isDirty).toBe(false);
   });
 });
+
+describe("MessageQueueSettings — merge toggle", () => {
+  it("is enabled by default and shows its limitations notice", async () => {
+    render(<MessageQueueSettings />);
+    await screen.findByLabelText(MAXIMUM_LABEL);
+
+    expect(mergeToggle().getAttribute("aria-pressed")).toBe("true");
+    expect(
+      screen.getByText(/Lets you fold a queued message into the message directly above it/),
+    ).toBeTruthy();
+    expect(
+      screen.getByText(/Only adjacent messages from the same sender can be merged/),
+    ).toBeTruthy();
+    expect(
+      screen.getByText(/combined, deduplicated entity references would exceed 100/),
+    ).toBeTruthy();
+  });
+
+  it("stages a toggle change and PATCHes merge_enabled without touching max_per_session", async () => {
+    updateSettingsMock.mockResolvedValueOnce(response(10, 10, "default", false, false));
+    render(<MessageQueueSettings />);
+    await screen.findByLabelText(MAXIMUM_LABEL);
+
+    fireEvent.click(mergeToggle());
+
+    expect(updateSettingsMock).not.toHaveBeenCalled();
+    expect(saveContributor?.isDirty).toBe(true);
+    const contributor = requireContributor();
+
+    await act(async () => contributor.save(contributor.revision));
+
+    expect(updateSettingsMock).toHaveBeenCalledWith({ merge_enabled: false });
+    await waitFor(() => expect(saveContributor?.isDirty).toBe(false));
+    expect(mergeToggle().getAttribute("aria-pressed")).toBe("false");
+  });
+
+  it("sends both fields in one PATCH when max_per_session and merge_enabled are both dirty", async () => {
+    updateSettingsMock.mockResolvedValueOnce(response(25, 25, "setting", false, false));
+    render(<MessageQueueSettings />);
+    const input = await screen.findByLabelText(MAXIMUM_LABEL);
+
+    fireEvent.change(input, { target: { value: "25" } });
+    fireEvent.click(mergeToggle());
+    const contributor = requireContributor();
+
+    await act(async () => contributor.save(contributor.revision));
+
+    expect(updateSettingsMock).toHaveBeenCalledWith({
+      max_per_session: 25,
+      merge_enabled: false,
+    });
+  });
+
+  it("stays editable for admins even when max_per_session is environment-locked", async () => {
+    fetchSettingsMock.mockResolvedValueOnce(response(25, 50, "environment", true, true));
+    render(<MessageQueueSettings />);
+    await screen.findByLabelText(MAXIMUM_LABEL);
+
+    expect(mergeToggle()).toHaveProperty("disabled", false);
+  });
+
+  it("is disabled for members", async () => {
+    currentRole = "member";
+    render(<MessageQueueSettings />);
+    await screen.findByLabelText(MAXIMUM_LABEL);
+
+    expect(mergeToggle()).toHaveProperty("disabled", true);
+  });
+
+  it("discards a toggle change back to the authoritative value", async () => {
+    fetchSettingsMock.mockResolvedValueOnce(response(10, 10, "default", false, true));
+    render(<MessageQueueSettings />);
+    await screen.findByLabelText(MAXIMUM_LABEL);
+    fireEvent.click(mergeToggle());
+    const contributor = requireContributor();
+
+    act(() => contributor.discard());
+
+    expect(mergeToggle().getAttribute("aria-pressed")).toBe("true");
+    expect(saveContributor?.isDirty).toBe(false);
+  });
+});

apps/web/components/settings/system/message-queue-settings.tsx

@@ -8,7 +8,8 @@ import { CardContent, CardHeader, CardTitle } from "@kandev/ui/card";
 import { Input } from "@kandev/ui/input";
 import { Label } from "@kandev/ui/label";
 import { Spinner } from "@kandev/ui/spinner";
-import { IconAlertCircle, IconLock } from "@tabler/icons-react";
+import { Switch } from "@kandev/ui/switch";
+import { IconAlertCircle, IconInfoCircle, IconLock } from "@tabler/icons-react";
 import { useTranslation } from "react-i18next";
 import { useAppStore } from "@/components/state-provider";
 import { SettingsCard } from "@/components/settings/settings-card";
@@ -21,6 +22,8 @@ import type { MessageQueueSettingsResponse, MessageQueueSettingsSource } from "@
 
 const ENVIRONMENT_VARIABLE = "KANDEV_QUEUE_MAX_PER_SESSION";
 
+/** Parses the max-per-session draft text into a non-negative safe integer,
+ * or `null` when the text isn't a valid whole number. */
 function parseMaximum(value: string): number | null {
   const trimmed = value.trim();
   if (!/^\d+$/.test(trimmed)) return null;
@@ -28,6 +31,7 @@ function parseMaximum(value: string): number | null {
   return Number.isSafeInteger(parsed) ? parsed : null;
 }
 
+/** Maps an effective-settings source enum to its i18n label key. */
 function sourceLabelKey(source: MessageQueueSettingsSource): string {
   switch (source) {
     case "setting":
@@ -39,14 +43,16 @@ function sourceLabelKey(source: MessageQueueSettingsSource): string {
   }
 }
 
-function useMessageQueueSettingsDraft() {
-  const { t } = useTranslation();
-  const role = useAppStore((state) => state.auth.user?.role);
+/** Loads the persisted snapshot and owns the two editable drafts (the
+ * max-per-session text field and the merge-enabled toggle). Split out of
+ * `useMessageQueueSettingsDraft` so neither function needs to grow past the
+ * project's per-function line limit as the page gains fields. */
+function useMessageQueueSettingsLoad() {
   const [snapshot, setSnapshot] = useState<MessageQueueSettingsResponse | null>(null);
   const [draft, setDraft] = useState("");
+  const [mergeDraft, setMergeDraft] = useState(true);
   const [loading, setLoading] = useState(true);
   const [loadFailed, setLoadFailed] = useState(false);
-  const [saveFailed, setSaveFailed] = useState(false);
   const loadVersion = useRef(0);
 
   const reload = useCallback(async () => {
@@ -58,6 +64,7 @@ function useMessageQueueSettingsDraft() {
       if (version !== loadVersion.current) return;
       setSnapshot(response);
       setDraft(String(response.settings.max_per_session));
+      setMergeDraft(response.settings.merge_enabled);
     } catch {
       if (version === loadVersion.current) setLoadFailed(true);
     } finally {
@@ -72,37 +79,73 @@ function useMessageQueueSettingsDraft() {
     };
   }, [reload]);
 
+  return {
+    snapshot,
+    setSnapshot,
+    draft,
+    setDraft,
+    mergeDraft,
+    setMergeDraft,
+    loading,
+    loadFailed,
+    reload,
+  };
+}
+
+/** Derives dirty/validity/permission state from the loaded snapshot and the
+ * two drafts, and registers the combined save/discard contributor that the
+ * page's floating save bar drives. */
+function useMessageQueueSettingsDraft() {
+  const { t } = useTranslation();
+  const role = useAppStore((state) => state.auth.user?.role);
+  const [saveFailed, setSaveFailed] = useState(false);
+  const load = useMessageQueueSettingsLoad();
+  const { snapshot, setSnapshot, draft, setDraft, mergeDraft, setMergeDraft } = load;
+
   const parsed = parseMaximum(draft);
   const baseline = snapshot?.settings.max_per_session;
-  const isDirty = baseline !== undefined && draft !== String(baseline);
+  const mergeBaseline = snapshot?.settings.merge_enabled;
+  const isMaxDirty = baseline !== undefined && draft !== String(baseline);
+  const isMergeDirty = mergeBaseline !== undefined && mergeDraft !== mergeBaseline;
+  const isDirty = isMaxDirty || isMergeDirty;
   const isAdmin = role === undefined || role === "admin";
   const isLocked = snapshot?.effective.locked === true;
+  // The environment lock only governs max_per_session; merging has no
+  // environment override, so an admin may always toggle it.
   const canEdit = isAdmin && !isLocked;
+  const canEditMerge = isAdmin;
   let invalidReason: string | undefined;
   if (parsed === null) invalidReason = t("system:messageQueueValidation");
   else if (!isAdmin) invalidReason = t("system:messageQueueAdminOnly");
-  else if (isLocked) {
-    invalidReason = t("system:messageQueueEnvironmentLocked", {
-      variable: ENVIRONMENT_VARIABLE,
-    });
+  else if (isLocked && isMaxDirty) {
+    invalidReason = t("system:messageQueueEnvironmentLocked", { variable: ENVIRONMENT_VARIABLE });
   }
 
   useSettingsSaveContributor({
     id: "system-message-queue",
-    revision: draft,
+    revision: `${draft}:${mergeDraft}`,
     isDirty,
-    canSave: parsed !== null && canEdit,
+    canSave: parsed !== null && (isMaxDirty ? canEdit : canEditMerge),
     invalidReason,
     save: async () => {
-      if (parsed === null || !canEdit) throw new Error(invalidReason);
+      if (parsed === null || (isMaxDirty && !canEdit) || (isMergeDirty && !canEditMerge)) {
+        throw new Error(invalidReason);
+      }
       const submitted = draft;
+      const submittedMerge = mergeDraft;
       setSaveFailed(false);
       try {
-        const response = await updateMessageQueueSettings({ max_per_session: parsed });
+        const response = await updateMessageQueueSettings({
+          ...(isMaxDirty ? { max_per_session: parsed } : {}),
+          ...(isMergeDirty ? { merge_enabled: mergeDraft } : {}),
+        });
         setSnapshot(response);
         setDraft((current) =>
           current === submitted ? String(response.settings.max_per_session) : current,
         );
+        setMergeDraft((current) =>
+          current === submittedMerge ? response.settings.merge_enabled : current,
+        );
       } catch (error) {
         setSaveFailed(true);
         throw error;
@@ -110,6 +153,7 @@ function useMessageQueueSettingsDraft() {
     },
     discard: () => {
       if (baseline !== undefined) setDraft(String(baseline));
+      if (mergeBaseline !== undefined) setMergeDraft(mergeBaseline);
       setSaveFailed(false);
     },
   });
@@ -118,16 +162,118 @@ function useMessageQueueSettingsDraft() {
     snapshot,
     draft,
     setDraft,
-    loading,
-    loadFailed,
+    mergeDraft,
+    setMergeDraft,
+    loading: load.loading,
+    loadFailed: load.loadFailed,
     saveFailed,
     isDirty,
     isAdmin,
     isLocked,
-    reload,
+    reload: load.reload,
   };
 }
 
+type QueueLimitFieldsProps = {
+  draft: string;
+  onDraftChange: (value: string) => void;
+  disabled: boolean;
+  configured: number;
+  effectiveValue: string;
+  source: MessageQueueSettingsSource;
+};
+
+/** Renders the per-session limit input plus the configured/effective summary
+ * badge; a pure presentational block extracted so `MessageQueueSettings`
+ * stays under the per-function line limit. */
+function QueueLimitFields({
+  draft,
+  onDraftChange,
+  disabled,
+  configured,
+  effectiveValue,
+  source,
+}: QueueLimitFieldsProps) {
+  const { t } = useTranslation();
+  return (
+    <>
+      <div className="min-w-0 space-y-2">
+        <Label htmlFor="message-queue-max-per-session">
+          {t("system:messageQueueMaximumLabel")}
+        </Label>
+        <Input
+          id="message-queue-max-per-session"
+          data-testid="message-queue-max-per-session"
+          type="number"
+          inputMode="numeric"
+          min={0}
+          step={1}
+          value={draft}
+          disabled={disabled}
+          onChange={(event) => onDraftChange(event.target.value)}
+          className="h-11 w-full max-w-xs"
+        />
+        <p className="text-xs text-muted-foreground">{t("system:messageQueueUnlimitedHelp")}</p>
+      </div>
+      <div className="rounded-md border border-border/70 bg-muted/20 p-3 text-sm">
+        <div className="flex flex-wrap items-center gap-2">
+          <span className="text-muted-foreground">{t("system:messageQueueConfigured")}</span>
+          <span>{configured}</span>
+          <span className="text-muted-foreground">{t("system:messageQueueEffective")}</span>
+          <strong data-testid="message-queue-effective-value">{effectiveValue}</strong>
+          <Badge variant="secondary" data-testid="message-queue-source">
+            {t(sourceLabelKey(source))}
+          </Badge>
+        </div>
+        <p className="mt-2 text-xs text-muted-foreground">
+          {t("system:messageQueueEffectiveHelp")}
+        </p>
+      </div>
+    </>
+  );
+}
+
+type MergeToggleFieldsProps = {
+  enabled: boolean;
+  onChange: (next: boolean) => void;
+  disabled: boolean;
+};
+
+/** Renders the merge-enabled switch plus its limitations notice; extracted
+ * for the same per-function line-limit reason as `QueueLimitFields`. */
+function MergeToggleFields({ enabled, onChange, disabled }: MergeToggleFieldsProps) {
+  const { t } = useTranslation();
+  return (
+    <div className="min-w-0 space-y-3 border-t border-border/70 pt-5">
+      <div className="flex flex-wrap items-start justify-between gap-3">
+        <div className="min-w-0 space-y-1">
+          <Label htmlFor="message-queue-merge-enabled">
+            {t("system:messageQueueMergeToggleLabel")}
+          </Label>
+          <p className="text-sm text-muted-foreground">
+            {t("system:messageQueueMergeDescription")}
+          </p>
+        </div>
+        <Switch
+          id="message-queue-merge-enabled"
+          data-testid="message-queue-merge-enabled"
+          checked={enabled}
+          disabled={disabled}
+          onCheckedChange={onChange}
+          aria-label={t("system:messageQueueMergeToggleLabel")}
+          className="cursor-pointer disabled:cursor-not-allowed"
+        />
+      </div>
+      <div className="flex gap-2 rounded-md border border-border/70 bg-muted/20 p-3 text-sm text-muted-foreground">
+        <IconInfoCircle className="size-4 shrink-0" />
+        <p>{t("system:messageQueueMergeNotice")}</p>
+      </div>
+    </div>
+  );
+}
+
+/** Settings → General → Message Queue page: the per-session queue limit and
+ * the queued-message merge toggle, sharing one save contributor. */
 export function MessageQueueSettings() {
   const { t } = useTranslation();
   const state = useMessageQueueSettingsDraft();
@@ -165,39 +311,14 @@ export function MessageQueueSettings() {
       </CardHeader>
       <CardContent className="min-w-0 space-y-5">
         <p className="text-sm text-muted-foreground">{t("system:messageQueueLimitDescription")}</p>
-        <div className="min-w-0 space-y-2">
-          <Label htmlFor="message-queue-max-per-session">
-            {t("system:messageQueueMaximumLabel")}
-          </Label>
-          <Input
-            id="message-queue-max-per-session"
-            data-testid="message-queue-max-per-session"
-            type="number"
-            inputMode="numeric"
-            min={0}
-            step={1}
-            value={state.draft}
-            disabled={!state.isAdmin || state.isLocked}
-            onChange={(event) => state.setDraft(event.target.value)}
-            className="h-11 w-full max-w-xs"
-          />
-          <p className="text-xs text-muted-foreground">{t("system:messageQueueUnlimitedHelp")}</p>
-        </div>
-
-        <div className="rounded-md border border-border/70 bg-muted/20 p-3 text-sm">
-          <div className="flex flex-wrap items-center gap-2">
-            <span className="text-muted-foreground">{t("system:messageQueueConfigured")}</span>
-            <span>{settings.max_per_session}</span>
-            <span className="text-muted-foreground">{t("system:messageQueueEffective")}</span>
-            <strong data-testid="message-queue-effective-value">{effectiveValue}</strong>
-            <Badge variant="secondary" data-testid="message-queue-source">
-              {t(sourceLabelKey(effective.source))}
-            </Badge>
-          </div>
-          <p className="mt-2 text-xs text-muted-foreground">
-            {t("system:messageQueueEffectiveHelp")}
-          </p>
-        </div>
+        <QueueLimitFields
+          draft={state.draft}
+          onDraftChange={state.setDraft}
+          disabled={!state.isAdmin || state.isLocked}
+          configured={settings.max_per_session}
+          effectiveValue={effectiveValue}
+          source={effective.source}
+        />
 
         {state.isLocked && (
           <Alert>
@@ -217,6 +338,12 @@ export function MessageQueueSettings() {
             <AlertDescription>{t("system:messageQueueSaveFailed")}</AlertDescription>
           </Alert>
         )}
+
+        <MergeToggleFields
+          enabled={state.mergeDraft}
+          onChange={state.setMergeDraft}
+          disabled={!state.isAdmin}
+        />
       </CardContent>
     </SettingsCard>
   );

apps/web/components/task/chat/queued-ghost-list.test.tsx

@@ -83,6 +83,7 @@ function baseState(entries: QueuedMessage[]) {
     count: entries.length,
     max: 10,
     isFull: false,
+    mergeEnabled: true,
     isLoading: false,
     queue: vi.fn(async () => {}),
     clearAll: vi.fn(async () => {}),
@@ -403,7 +404,7 @@ describe("QueueAffordance entity-reference edits", () => {
   });
 });
 
-describe("QueueAffordance merge wiring", () => {
+describe("QueueAffordance merge wiring — eligibility", () => {
   it("shows a merge control on the second row and calls mergeEntry with its id", () => {
     const state = queueState([
       entry({ id: "q-1", content: "first", queued_by: QUEUED_BY_USER }),
@@ -454,7 +455,9 @@ describe("QueueAffordance merge wiring", () => {
     fireEvent.click(screen.getByTestId(CHIP_ID));
     expect(screen.queryByTestId(MERGE_BUTTON_ID)).toBeNull();
   });
+});
 
+describe("QueueAffordance merge wiring — dispatch", () => {
   it("toasts an error when the merge fails", async () => {
     const { toast } = await import("sonner");
     const state = queueState([
@@ -507,4 +510,19 @@ describe("QueueAffordance merge wiring", () => {
     fireEvent.click(button);
     await waitFor(() => expect(state.mergeEntry).toHaveBeenCalledTimes(1));
   });
+
+  it("hides the merge control entirely when merging is disabled", () => {
+    const state = queueState(
+      [
+        entry({ id: "q-1", content: "first", queued_by: QUEUED_BY_USER }),
+        entry({ id: "q-2", content: "second", queued_by: QUEUED_BY_USER }),
+      ],
+      { mergeEnabled: false },
+    );
+    useQueueMock.mockReturnValue(state);
+    render(<QueueAffordance sessionId={SESSION_ID}>{CHILD}</QueueAffordance>);
+
+    fireEvent.click(screen.getByTestId(CHIP_ID));
+    expect(screen.queryAllByTestId(MERGE_BUTTON_ID)).toHaveLength(0);
+  });
 });

apps/web/components/task/chat/queued-ghost-list.tsx

@@ -226,6 +226,7 @@ type QueuePanelDisclosureProps = {
   canDrain: boolean;
   isLoading: boolean;
   cancellationPending: boolean;
+  mergeEnabled: boolean;
   onClose: () => void;
   onClear: () => void;
   onDrain: () => void;
@@ -236,6 +237,7 @@ type QueuePanelDisclosureProps = {
   onSendEntryNow: (entryId: string) => void;
 };
 
+/** Wraps QueuePanel in the collapsible open/close animation shell. */
 function QueuePanelDisclosure({
   isOpen,
   onOpenChange,
@@ -246,6 +248,7 @@ function QueuePanelDisclosure({
   canDrain,
   isLoading,
   cancellationPending,
+  mergeEnabled,
   onClose,
   onClear,
   onDrain,
@@ -271,6 +274,7 @@ function QueuePanelDisclosure({
           canDrain={canDrain}
           isLoading={isLoading}
           cancellationPending={cancellationPending}
+          mergeEnabled={mergeEnabled}
           onClose={onClose}
           onClear={onClear}
           onDrain={onDrain}
@@ -319,6 +323,7 @@ export function QueueAffordance({
     count,
     max,
     isFull,
+    mergeEnabled,
     isLoading,
     clearAll,
     drainNext,
@@ -389,6 +394,7 @@ export function QueueAffordance({
         canDrain={canDrain}
         isLoading={isLoading}
         cancellationPending={cancellationPending}
+        mergeEnabled={mergeEnabled}
         onClose={close}
         onClear={handleClear}
         onDrain={handleDrain}
@@ -467,6 +473,7 @@ type QueuePanelProps = {
   canDrain: boolean;
   isLoading: boolean;
   cancellationPending: boolean;
+  mergeEnabled: boolean;
   onClose: () => void;
   onClear: () => void;
   onDrain: () => void;
@@ -477,6 +484,8 @@ type QueuePanelProps = {
   onSendEntryNow: (entryId: string) => void;
 };
 
+/** Renders the expanded queue list: header controls plus one QueuedGhostMessage
+ * row per pending entry, gating each row's merge control on `mergeEnabled`. */
 function QueuePanel({
   entries,
   count,
@@ -485,6 +494,7 @@ function QueuePanel({
   canDrain,
   isLoading,
   cancellationPending,
+  mergeEnabled,
   onClose,
   onClear,
   onDrain,
@@ -530,7 +540,7 @@ function QueuePanel({
             index={index}
             canEdit={canUserEditEntry(entry)}
             canRemove
-            canMerge={canMergeWithAbove(entry, entries[index - 1])}
+            canMerge={mergeEnabled && canMergeWithAbove(entry, entries[index - 1])}
             onSave={(content, entityReferences) => onSave(entry.id, content, entityReferences)}
             onRemove={() => onRemove(entry.id)}
             onMerge={() => onMerge(entry.id)}

apps/web/e2e/tests/system/message-queue-settings.spec.ts

@@ -1,19 +1,18 @@
 import { test, expect } from "../../fixtures/test-base";
 import type { ApiClient } from "../../helpers/api-client";
-import type { MessageQueueSettingsResponse } from "../../../lib/types/system";
+import type {
+  MessageQueueSettingsPatch,
+  MessageQueueSettingsResponse,
+} from "../../../lib/types/system";
 
 const SETTINGS_PATH = "/api/v1/system/message-queue/settings";
 
 async function requestSettings(
   apiClient: ApiClient,
   method: "GET" | "PATCH",
-  maxPerSession?: number,
+  patch?: MessageQueueSettingsPatch,
 ): Promise<MessageQueueSettingsResponse> {
-  const response = await apiClient.rawRequest(
-    method,
-    SETTINGS_PATH,
-    maxPerSession === undefined ? undefined : { max_per_session: maxPerSession },
-  );
+  const response = await apiClient.rawRequest(method, SETTINGS_PATH, patch);
   if (!response.ok) {
     throw new Error(
       `${method} ${SETTINGS_PATH} failed (${response.status}): ${await response.text()}`,
@@ -24,15 +23,22 @@ async function requestSettings(
 
 test.describe.serial("Message Queue general settings", () => {
   let baseline: number | undefined;
+  let mergeBaseline: boolean | undefined;
 
   test.beforeEach(async ({ apiClient }) => {
-    baseline = (await requestSettings(apiClient, "GET")).settings.max_per_session;
+    const current = (await requestSettings(apiClient, "GET")).settings;
+    baseline = current.max_per_session;
+    mergeBaseline = current.merge_enabled;
   });
 
   test.afterEach(async ({ apiClient }) => {
-    if (baseline === undefined) return;
-    await requestSettings(apiClient, "PATCH", baseline);
+    if (baseline === undefined || mergeBaseline === undefined) return;
+    await requestSettings(apiClient, "PATCH", {
+      max_per_session: baseline,
+      merge_enabled: mergeBaseline,
+    });
     baseline = undefined;
+    mergeBaseline = undefined;
   });
 
   test("admin navigates to Message Queue, saves, and reloads the live setting", async ({
@@ -68,14 +74,54 @@ test.describe.serial("Message Queue general settings", () => {
     await expect(testPage.getByTestId("message-queue-effective-value")).toHaveText(String(updated));
   });
 
+  test("admin toggles queued message merging without touching the session limit", async ({
+    testPage,
+  }) => {
+    await testPage.goto("/settings/general/message-queue");
+    const toggle = testPage.getByTestId("message-queue-merge-enabled");
+    await expect(toggle).toBeVisible();
+    const wasEnabled = mergeBaseline === true;
+    await expect(toggle).toHaveAttribute("aria-checked", String(wasEnabled));
+    await expect(
+      testPage.getByText(/Only adjacent messages from the same sender can be merged/),
+    ).toBeVisible();
+    await expect(
+      testPage.getByText(/combined, deduplicated entity references would exceed 100/),
+    ).toBeVisible();
+
+    const saveResponse = testPage.waitForResponse(
+      (response) =>
+        response.request().method() === "PATCH" &&
+        new URL(response.url()).pathname === SETTINGS_PATH,
+    );
+    await toggle.click();
+    await expect(testPage.getByTestId("settings-floating-save")).toBeVisible();
+    await testPage.getByRole("button", { name: "Save changes" }).click();
+    const response = await saveResponse;
+    expect(response.status()).toBe(200);
+    expect(response.request().postDataJSON()).toEqual({ merge_enabled: !wasEnabled });
+    await expect(toggle).toHaveAttribute("aria-checked", String(!wasEnabled));
+
+    await testPage.reload();
+    await expect(testPage.getByTestId("message-queue-merge-enabled")).toHaveAttribute(
+      "aria-checked",
+      String(!wasEnabled),
+    );
+  });
+
   test("environment override is visible and read-only", async ({ testPage }) => {
     await testPage.route(`**${SETTINGS_PATH}`, async (route) => {
       await route.fulfill({
         status: 200,
         contentType: "application/json",
         body: JSON.stringify({
-          settings: { max_per_session: baseline },
-          effective: { max_per_session: 41, source: "environment", locked: true },
+          settings: { max_per_session: baseline, merge_enabled: true },
+          effective: {
+            max_per_session: 41,
+            source: "environment",
+            locked: true,
+            merge_enabled: true,
+          },
         } satisfies MessageQueueSettingsResponse),
       });
     });
@@ -86,6 +132,8 @@ test.describe.serial("Message Queue general settings", () => {
     await expect(testPage.getByTestId("message-queue-source")).toHaveText("Environment");
     await expect(testPage.getByText(/KANDEV_QUEUE_MAX_PER_SESSION/).first()).toBeVisible();
     await expect(testPage.getByTestId("settings-floating-save")).toHaveCount(0);
+    // The environment lock only governs max_per_session; merging stays editable.
+    await expect(testPage.getByTestId("message-queue-merge-enabled")).toBeEnabled();
   });
 
   test("member can navigate to the setting but cannot edit it", async ({ testPage }) => {
@@ -111,6 +159,7 @@ test.describe.serial("Message Queue general settings", () => {
     await settingsNav.getByRole("link", { name: "Message Queue" }).click();
 
     await expect(testPage.getByTestId("message-queue-max-per-session")).toBeDisabled();
+    await expect(testPage.getByTestId("message-queue-merge-enabled")).toBeDisabled();
     await expect(testPage.getByText("Only administrators can change this setting.")).toBeVisible();
     await expect(testPage.getByTestId("settings-floating-save")).toHaveCount(0);
   });

apps/web/hooks/domains/session/use-queue.ts

@@ -73,6 +73,7 @@ type QueueActionsArgs = {
   removeQueueEntry: ReturnType<typeof useQueueState>["removeQueueEntry"];
   setQueueLoading: ReturnType<typeof useQueueState>["setQueueLoading"];
   metaMax: number | undefined;
+  metaMergeEnabled: boolean | undefined;
 };
 
 function useDrainNextAction(
@@ -130,6 +131,9 @@ function useSendNowAction(
   );
 }
 
+/** Fetches the authoritative queue snapshot and writes it into the slice,
+ * discarding a response that arrives after a newer request for the same
+ * session was already issued (out-of-order network resolution). */
 function useQueueRefetch(
   setQueueEntries: ReturnType<typeof useQueueState>["setQueueEntries"],
   setQueueLoading: ReturnType<typeof useQueueState>["setQueueLoading"],
@@ -146,7 +150,11 @@ function useQueueRefetch(
         setQueueLoading(sid, true);
         const status = await getQueueStatus(sid);
         if (refetchVersion.current[sid] !== version) return;
-        setQueueEntries(sid, status.entries ?? [], { count: status.count, max: status.max });
+        setQueueEntries(sid, status.entries ?? [], {
+          count: status.count,
+          max: status.max,
+          mergeEnabled: status.merge_enabled,
+        });
       } finally {
         if (refetchVersion.current[sid] === version) setQueueLoading(sid, false);
       }
@@ -163,6 +171,7 @@ function useQueueActions({
   removeQueueEntry,
   setQueueLoading,
   metaMax,
+  metaMergeEnabled,
 }: QueueActionsArgs) {
   const { refetch, invalidate: invalidateRefetch } = useQueueRefetch(
     setQueueEntries,
@@ -212,7 +221,11 @@ function useQueueActions({
         // optimistic empty state. The explicit refetch below then owns the
         // newest version and replaces it with the backend's exact result.
         invalidateRefetch(sessionId);
-        setQueueEntries(sessionId, [], { count: 0, max: metaMax ?? 0 });
+        setQueueEntries(sessionId, [], {
+          count: 0,
+          max: metaMax ?? 0,
+          mergeEnabled: metaMergeEnabled ?? true,
+        });
       } catch (err) {
         mutationFailed = true;
         mutationError = err;
@@ -230,7 +243,15 @@ function useQueueActions({
     } finally {
       setQueueLoading(sessionId, false);
     }
-  }, [sessionId, setQueueEntries, setQueueLoading, metaMax, refetch, invalidateRefetch]);
+  }, [
+    sessionId,
+    setQueueEntries,
+    setQueueLoading,
+    metaMax,
+    metaMergeEnabled,
+    refetch,
+    invalidateRefetch,
+  ]);
 
   const drainNext = useDrainNextAction(sessionId, setQueueLoading, refetch);
 
@@ -373,6 +394,7 @@ export function useQueue(sessionId: string | null) {
     removeQueueEntry: state.removeQueueEntry,
     setQueueLoading: state.setQueueLoading,
     metaMax: meta?.max,
+    metaMergeEnabled: meta?.mergeEnabled,
   });
 
   useEffect(() => {
@@ -405,6 +427,7 @@ export function useQueue(sessionId: string | null) {
     count: meta?.count ?? entries.length,
     max: meta?.max ?? 0,
     isFull: meta ? meta.count >= meta.max && meta.max > 0 : false,
+    mergeEnabled: meta?.mergeEnabled ?? true,
     isLoading,
     cancellationPending,
     queue,

apps/web/lib/api/domains/settings-api.ts

@@ -21,8 +21,8 @@ import type {
   DynamicModelsResponse,
 } from "@/lib/types/http";
 import type {
+  MessageQueueSettingsPatch,
   MessageQueueSettingsResponse,
-  MessageQueueSettingsValue,
   SystemMetricsGlobalSettings,
   SystemMetricsSettingsResponse,
 } from "@/lib/types/system";
@@ -75,6 +75,7 @@ export async function updateSystemMetricsSettings(
 
 const MESSAGE_QUEUE_SETTINGS_PATH = "/api/v1/system/message-queue/settings";
 
+/** Fetches the current message queue settings (configured + effective). */
 export async function fetchMessageQueueSettings(
   options?: ApiRequestOptions,
 ): Promise<MessageQueueSettingsResponse> {
@@ -84,8 +85,10 @@ export async function fetchMessageQueueSettings(
   });
 }
 
+/** Applies a partial update to message queue settings: fields omitted from
+ * `payload` keep their current persisted value on the backend. */
 export async function updateMessageQueueSettings(
-  payload: MessageQueueSettingsValue,
+  payload: MessageQueueSettingsPatch,
   options?: ApiRequestOptions,
 ): Promise<MessageQueueSettingsResponse> {
   return fetchJson<MessageQueueSettingsResponse>(MESSAGE_QUEUE_SETTINGS_PATH, {

apps/web/lib/state/slices/session/session-slice.upsert.test.ts

@@ -366,16 +366,34 @@ describe("queue actions", () => {
       makeEntry({ id: "e2", content: "second" }),
     ];
 
-    store.getState().setQueueEntries(SESSION_ID, entries, { count: 2, max: 10 });
+    store
+      .getState()
+      .setQueueEntries(SESSION_ID, entries, { count: 2, max: 10, mergeEnabled: true });
 
     expect(store.getState().queue.bySessionId[SESSION_ID]).toEqual(entries);
-    expect(store.getState().queue.metaBySessionId[SESSION_ID]).toEqual({ count: 2, max: 10 });
+    expect(store.getState().queue.metaBySessionId[SESSION_ID]).toEqual({
+      count: 2,
+      max: 10,
+      mergeEnabled: true,
+    });
+  });
+
+  it("retains a disabled merge state", () => {
+    const store = makeStore();
+    store.getState().setQueueEntries(SESSION_ID, [makeEntry()], {
+      count: 1,
+      max: 10,
+      mergeEnabled: false,
+    });
+    expect(store.getState().queue.metaBySessionId[SESSION_ID]?.mergeEnabled).toBe(false);
   });
 
   it("removeQueueEntry drops a single entry by id and refreshes meta.count", () => {
     const store = makeStore();
     const entries = [makeEntry({ id: "e1" }), makeEntry({ id: "e2" }), makeEntry({ id: "e3" })];
-    store.getState().setQueueEntries(SESSION_ID, entries, { count: 3, max: 10 });
+    store
+      .getState()
+      .setQueueEntries(SESSION_ID, entries, { count: 3, max: 10, mergeEnabled: true });
 
     store.getState().removeQueueEntry(SESSION_ID, "e2");
 
@@ -392,7 +410,9 @@ describe("queue actions", () => {
 
   it("clearQueueStatus removes both entries and meta", () => {
     const store = makeStore();
-    store.getState().setQueueEntries(SESSION_ID, [makeEntry()], { count: 1, max: 10 });
+    store
+      .getState()
+      .setQueueEntries(SESSION_ID, [makeEntry()], { count: 1, max: 10, mergeEnabled: true });
 
     store.getState().clearQueueStatus(SESSION_ID);
 

apps/web/lib/state/slices/session/types.ts

@@ -137,12 +137,16 @@ export type QueuedMessage = {
 export type QueueMeta = {
   count: number;
   max: number;
+  /** Mirrors the server's message queue merge_enabled setting; hides the
+   * "Merge with above" affordance without a separate settings fetch. */
+  mergeEnabled: boolean;
 };
 
 export type QueueStatus = {
   entries: QueuedMessage[];
   count: number;
   max: number;
+  merge_enabled: boolean;
 };
 
 export type QueueState = {

apps/web/lib/types/backend.ts

@@ -1,3 +1,4 @@
+import type { QueuedMessage } from "@/lib/state/slices/session/types";
 import type { TaskPlanEventPayload, TaskPlanRevisionEventPayload } from "./task-plan-events";
 import type { CaptureRequest } from "@/lib/logger/capture";
 
@@ -482,20 +483,12 @@ export {
 
 export type { TaskPlanEventPayload, TaskPlanRevisionEventPayload } from "./task-plan-events";
 
-export type QueuedMessagePayload = {
-  content: string;
-  model?: string;
-  plan_mode?: boolean;
-  task_id: string;
-  user_id?: string;
-  queued_at: string;
-};
-
 export type QueueStatusChangedPayload = {
   session_id: string;
-  entries?: QueuedMessagePayload[] | null;
+  entries?: QueuedMessage[] | null;
   count?: number;
   max?: number;
+  merge_enabled?: boolean;
 };
 
 export type TaskStatusSummaryUpdatedPayload = {

apps/web/lib/types/system.ts

@@ -188,8 +188,12 @@ export interface SystemMetricsSettingsResponse {
 
 export interface MessageQueueSettingsValue {
   max_per_session: number;
+  merge_enabled: boolean;
 }
 
+/** Partial PATCH payload: omitted fields are left unchanged server-side. */
+export type MessageQueueSettingsPatch = Partial<MessageQueueSettingsValue>;
+
 export type MessageQueueSettingsSource = "default" | "setting" | "environment";
 
 export interface MessageQueueEffectiveSettings extends MessageQueueSettingsValue {

apps/web/lib/ws/handlers/agent-session.ts

@@ -10,7 +10,7 @@ import {
   type TaskSession,
   type TaskSessionState,
 } from "@/lib/types/http";
-import type { QueuedMessage } from "@/lib/state/slices/session/types";
+import type { QueueStatusChangedPayload } from "@/lib/types/backend";
 import { syncKanbanPrimarySessionState } from "@/lib/ws/handlers/agent-session-kanban-sync";
 import { parseContextWindowEntry } from "@/lib/state/slices/session-runtime/context-window";
 
@@ -668,20 +668,27 @@ function handleCancellationPendingMessage(
   applyCancellationPending(store, payload);
 }
 
+/** Writes a message.queue.status_changed broadcast into the queue slice,
+ * defaulting count/max/mergeEnabled when the backend omits them. */
+function handleQueueStatusChangedMessage(
+  store: StoreApi<AppState>,
+  payload: QueueStatusChangedPayload,
+): void {
+  if (!payload.session_id) {
+    console.warn("[Queue] Missing session_id in queue status change event");
+    return;
+  }
+  const entries = payload.entries ?? [];
+  const count = typeof payload.count === "number" ? payload.count : entries.length;
+  const max = typeof payload.max === "number" ? payload.max : 0;
+  const mergeEnabled = typeof payload.merge_enabled === "boolean" ? payload.merge_enabled : true;
+  store.getState().setQueueEntries(payload.session_id, entries, { count, max, mergeEnabled });
+}
+
 export function registerTaskSessionHandlers(store: StoreApi<AppState>): WsHandlers {
   return {
-    "message.queue.status_changed": (message) => {
-      const payload = message.payload;
-      if (!payload?.session_id) {
-        console.warn("[Queue] Missing session_id in queue status change event");
-        return;
-      }
-      const sessionId = payload.session_id;
-      const entries = (payload.entries as QueuedMessage[] | null | undefined) ?? [];
-      const count = typeof payload.count === "number" ? payload.count : entries.length;
-      const max = typeof payload.max === "number" ? payload.max : 0;
-      store.getState().setQueueEntries(sessionId, entries, { count, max });
-    },
+    "message.queue.status_changed": (message) =>
+      handleQueueStatusChangedMessage(store, message.payload),
     "session.state_changed": (message) => {
       const payload = message.payload;
       if (!payload?.task_id) return;

apps/web/src/locales/en/system.json

@@ -174,6 +174,9 @@
   "messageQueueLoadFailed": "Message queue settings could not be loaded.",
   "messageQueueLoading": "Loading message queue settings…",
   "messageQueueMaximumLabel": "Maximum messages per session",
+  "messageQueueMergeDescription": "Lets you fold a queued message into the message directly above it before the agent picks them up, combining bursts of related follow-ups into one turn.",
+  "messageQueueMergeNotice": "Only adjacent messages from the same sender can be merged: your own queued messages merge into each other, and messages queued by another task's agent merge only when they share the same originating task. Workflow, system, and in-flight messages can never be merged. A merge is also rejected if the two messages' combined, deduplicated entity references would exceed 100, the shared per-message limit. Turning this off hides the merge control immediately and does not affect messages already merged.",
+  "messageQueueMergeToggleLabel": "Enable queued message merging",
   "messageQueueRetry": "Retry",
   "messageQueueSaveFailed": "Failed to save message queue settings.",
   "messageQueueSourceDefault": "Default",

apps/web/src/locales/pseudo/system.json

@@ -174,6 +174,9 @@
   "messageQueueLoadFailed": "Ḿēśśàĝē qũēũē śēţţĩńĝś ćōũĺď ńōţ ƀē ĺōàďēď.",
   "messageQueueLoading": "Ĺōàďĩńĝ ḿēśśàĝē qũēũē śēţţĩńĝś…",
   "messageQueueMaximumLabel": "Ḿàxĩḿũḿ ḿēśśàĝēś ƥēŕ śēśśĩōń",
+  "messageQueueMergeDescription": "Ĺēţś ŷōũ ƒōĺď à qũēũēď ḿēśśàĝē ĩńţō ţĥē ḿēśśàĝē ďĩŕēćţĺŷ àƀōvē ĩţ ƀēƒōŕē ţĥē àĝēńţ ƥĩćķś ţĥēḿ ũƥ, ćōḿƀĩńĩńĝ ƀũŕśţś ōƒ ŕēĺàţēď ƒōĺĺōŵ-ũƥś ĩńţō ōńē ţũŕń.",
+  "messageQueueMergeNotice": "Ōńĺŷ àďĵàćēńţ ḿēśśàĝēś ƒŕōḿ ţĥē śàḿē śēńďēŕ ćàń ƀē ḿēŕĝēď: ŷōũŕ ōŵń qũēũēď ḿēśśàĝēś ḿēŕĝē ĩńţō ēàćĥ ōţĥēŕ, àńď ḿēśśàĝēś qũēũēď ƀŷ àńōţĥēŕ ţàśķ'ś àĝēńţ ḿēŕĝē ōńĺŷ ŵĥēń ţĥēŷ śĥàŕē ţĥē śàḿē ōŕĩĝĩńàţĩńĝ ţàśķ. Ŵōŕķƒĺōŵ, śŷśţēḿ, àńď ĩń-ƒĺĩĝĥţ ḿēśśàĝēś ćàń ńēvēŕ ƀē ḿēŕĝēď. À ḿēŕĝē ĩś àĺśō ŕēĵēćţēď ĩƒ ţĥē ţŵō ḿēśśàĝēś' ćōḿƀĩńēď, ďēďũƥĺĩćàţēď ēńţĩţŷ ŕēƒēŕēńćēś ŵōũĺď ēxćēēď 100, ţĥē śĥàŕēď ƥēŕ-ḿēśśàĝē ĺĩḿĩţ. Ţũŕńĩńĝ ţĥĩś ōƒƒ ĥĩďēś ţĥē ḿēŕĝē ćōńţŕōĺ ĩḿḿēďĩàţēĺŷ àńď ďōēś ńōţ àƒƒēćţ ḿēśśàĝēś àĺŕēàďŷ ḿēŕĝēď.",
+  "messageQueueMergeToggleLabel": "Ēńàƀĺē qũēũēď ḿēśśàĝē ḿēŕĝĩńĝ",
   "messageQueueRetry": "Ŕēţŕŷ",
   "messageQueueSaveFailed": "Ƒàĩĺēď ţō śàvē ḿēśśàĝē qũēũē śēţţĩńĝś.",
   "messageQueueSourceDefault": "Ďēƒàũĺţ",