diff --git a/apps/backend/internal/backendapp/message_queue_settings_test.go b/apps/backend/internal/backendapp/message_queue_settings_test.go index 068fe27d05..433a3a85fd 100644 --- a/apps/backend/internal/backendapp/message_queue_settings_test.go +++ b/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:") diff --git a/apps/backend/internal/backendapp/orchestrator.go b/apps/backend/internal/backendapp/orchestrator.go index 106d2e5dd7..862de17064 100644 --- a/apps/backend/internal/backendapp/orchestrator.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers.go b/apps/backend/internal/orchestrator/handlers/queue_handlers.go index f3bd15a17c..0a762ad4da 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go b/apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go index 15fbacfa3d..a828390364 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go +++ b/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(), diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index 239937968f..81edcc0b35 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/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, } } diff --git a/apps/backend/internal/orchestrator/messagequeue/service_merge_enabled_test.go b/apps/backend/internal/orchestrator/messagequeue/service_merge_enabled_test.go new file mode 100644 index 0000000000..78dab90bda --- /dev/null +++ b/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) +} diff --git a/apps/backend/internal/orchestrator/messagequeue/types.go b/apps/backend/internal/orchestrator/messagequeue/types.go index 111d38fd35..904d8a2bc4 100644 --- a/apps/backend/internal/orchestrator/messagequeue/types.go +++ b/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 diff --git a/apps/backend/internal/system/queuesettings/handler.go b/apps/backend/internal/system/queuesettings/handler.go index eb36d7018d..a076c36239 100644 --- a/apps/backend/internal/system/queuesettings/handler.go +++ b/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): diff --git a/apps/backend/internal/system/queuesettings/queuesettings_test.go b/apps/backend/internal/system/queuesettings/queuesettings_test.go index b4d7f92bb2..fe8fd5b1ce 100644 --- a/apps/backend/internal/system/queuesettings/queuesettings_test.go +++ b/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 } diff --git a/apps/backend/internal/system/queuesettings/resolver.go b/apps/backend/internal/system/queuesettings/resolver.go index 90a49fe8d0..2fc3a47864 100644 --- a/apps/backend/internal/system/queuesettings/resolver.go +++ b/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 } diff --git a/apps/backend/internal/system/queuesettings/service.go b/apps/backend/internal/system/queuesettings/service.go index 13bc9831d5..81fa456046 100644 --- a/apps/backend/internal/system/queuesettings/service.go +++ b/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 diff --git a/apps/backend/internal/system/queuesettings/store.go b/apps/backend/internal/system/queuesettings/store.go index 85f765de06..943d24aed1 100644 --- a/apps/backend/internal/system/queuesettings/store.go +++ b/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) } diff --git a/apps/backend/internal/system/queuesettings/types.go b/apps/backend/internal/system/queuesettings/types.go index 69a1c63541..b31d97ee7d 100644 --- a/apps/backend/internal/system/queuesettings/types.go +++ b/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 { diff --git a/apps/backend/internal/system/system_routes_test.go b/apps/backend/internal/system/system_routes_test.go index 1934025ebb..2653e2b00e 100644 --- a/apps/backend/internal/system/system_routes_test.go +++ b/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 diff --git a/apps/web/components/settings/system/message-queue-settings.test.tsx b/apps/web/components/settings/system/message-queue-settings.test.tsx index b33c57f1a1..000aa2569d 100644 --- a/apps/web/components/settings/system/message-queue-settings.test.tsx +++ b/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; + }) => ( +