From af764a227f9c57ef129622e04262eadabff69dfa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20Fiere?= Date: Sun, 2 Aug 2026 17:04:11 +0200 Subject: [PATCH] feat(queue): merge a queued message into the message above it (#2131) --- apps/backend/internal/entityrefs/reference.go | 14 +- .../internal/entityrefs/submission_test.go | 2 +- .../orchestrator/handlers/queue_handlers.go | 62 ++- .../handlers/queue_handlers_merge_test.go | 129 +++++ .../orchestrator/messagequeue/merge.go | 97 ++++ .../orchestrator/messagequeue/repository.go | 14 + .../messagequeue/repository_memory.go | 63 +++ .../repository_postgres_merge_test.go | 104 ++++ .../messagequeue/repository_sqlite.go | 188 +++++++- .../repository_sqlite_merge_test.go | 446 ++++++++++++++++++ .../orchestrator/messagequeue/service.go | 14 + .../orchestrator/messagequeue/service_test.go | 163 +++++++ .../orchestrator/messagequeue/types.go | 16 + apps/backend/pkg/websocket/actions.go | 1 + .../task/chat/queued-ghost-list.test.tsx | 120 ++++- .../task/chat/queued-ghost-list.tsx | 71 ++- .../task/chat/queued-ghost-message.test.tsx | 174 ++++++- .../task/chat/queued-ghost-message.tsx | 203 ++++++-- apps/web/e2e/tests/chat/message-queue.spec.ts | 39 ++ .../hooks/domains/session/use-queue.test.ts | 103 ++++ apps/web/hooks/domains/session/use-queue.ts | 72 ++- apps/web/lib/api/domains/queue-api.test.ts | 65 +++ apps/web/lib/api/domains/queue-api.ts | 31 ++ apps/web/src/locales/en/chat.json | 9 + apps/web/src/locales/pseudo/chat.json | 9 + ...notifications-extend-task-pr-automation.md | 25 + docs/plans/message-queue-merge/plan.md | 244 ++++++++++ .../task-01-backend-repository.md | 58 +++ .../task-02-backend-service-handler.md | 50 ++ .../task-03-frontend-merge-ui.md | 59 +++ docs/plans/message-queue-merge/task-04-e2e.md | 41 ++ docs/public/websocket-api.md | 3 + docs/specs/INDEX.md | 1 + docs/specs/ui/message-queue-merge.md | 196 ++++++++ 34 files changed, 2806 insertions(+), 80 deletions(-) create mode 100644 apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go create mode 100644 apps/backend/internal/orchestrator/messagequeue/merge.go create mode 100644 apps/backend/internal/orchestrator/messagequeue/repository_postgres_merge_test.go create mode 100644 apps/backend/internal/orchestrator/messagequeue/repository_sqlite_merge_test.go create mode 100644 apps/web/src/locales/en/chat.json create mode 100644 apps/web/src/locales/pseudo/chat.json create mode 100644 docs/plans/message-queue-merge/plan.md create mode 100644 docs/plans/message-queue-merge/task-01-backend-repository.md create mode 100644 docs/plans/message-queue-merge/task-02-backend-service-handler.md create mode 100644 docs/plans/message-queue-merge/task-03-frontend-merge-ui.md create mode 100644 docs/plans/message-queue-merge/task-04-e2e.md create mode 100644 docs/specs/ui/message-queue-merge.md diff --git a/apps/backend/internal/entityrefs/reference.go b/apps/backend/internal/entityrefs/reference.go index beb3a60ad5..1eb2df8e5d 100644 --- a/apps/backend/internal/entityrefs/reference.go +++ b/apps/backend/internal/entityrefs/reference.go @@ -14,12 +14,18 @@ import ( apiv1 "github.com/kandev/kandev/pkg/api/v1" ) +// ErrInvalidReference is returned when an entity reference fails validation +// (malformed ref, unknown provider/kind, or over the per-message cap). var ErrInvalidReference = errors.New("invalid entity reference") var identityNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9._:-]{0,127}$`) const ( - maxReferencesPerMessage = 100 + // MaxReferencesPerMessage is the per-message entity-reference cap enforced + // by NormalizeForSubmission/NormalizePersisted. Consumers that build + // combined reference lists (e.g. message queue merges) use it to reject + // unions that would exceed the cap instead of silently dropping refs. + MaxReferencesPerMessage = 100 kandevProviderID = "kandev" kandevTaskKind = "task" ) @@ -60,11 +66,11 @@ func NormalizePersisted(raw any) []apiv1.EntityReference { } func normalize(references []apiv1.EntityReference, strict bool) ([]apiv1.EntityReference, error) { - if len(references) > maxReferencesPerMessage { + if len(references) > MaxReferencesPerMessage { if strict { - return nil, fmt.Errorf("%w: at most %d references are allowed", ErrInvalidReference, maxReferencesPerMessage) + return nil, fmt.Errorf("%w: at most %d references are allowed", ErrInvalidReference, MaxReferencesPerMessage) } - references = references[:maxReferencesPerMessage] + references = references[:MaxReferencesPerMessage] } result := make([]apiv1.EntityReference, 0, len(references)) seen := make(map[string]struct{}, len(references)) diff --git a/apps/backend/internal/entityrefs/submission_test.go b/apps/backend/internal/entityrefs/submission_test.go index d52afae5d9..1387b38e8f 100644 --- a/apps/backend/internal/entityrefs/submission_test.go +++ b/apps/backend/internal/entityrefs/submission_test.go @@ -102,7 +102,7 @@ func TestSubmissionServiceRejectsOverLimitBeforeScopeOrProviderWork(t *testing.T resolver := &fakeConversationResolver{workspaceID: "workspace-1"} authorizer := &fakeWorkspaceAuthorizer{} validator := NewSubmissionService(resolver, authorizer) - references := make([]apiv1.EntityReference, maxReferencesPerMessage+1) + references := make([]apiv1.EntityReference, MaxReferencesPerMessage+1) _, err := validator.ValidateForSubmission( context.Background(), "session-1", "task-1", references, diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers.go b/apps/backend/internal/orchestrator/handlers/queue_handlers.go index 78a198cd75..966e98fe86 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers.go +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers.go @@ -21,7 +21,11 @@ const ( queueErrorCodeEntryNotFound = "entry_not_found" queueErrorCodeSessionBusy = "session_busy" queueErrorCodeNotPromptable = "session_not_promptable" - queueInvalidReferences = "Invalid entity references" + // queueErrorCodeMergeReferenceOverflow is surfaced when a merge would push + // 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" // Payload field names — extracted to satisfy goconst (≥3 occurrences). fieldSessionID = "session_id" @@ -37,10 +41,12 @@ type QueueService interface { AppendContent(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []messagequeue.MessageAttachment) (*messagequeue.QueuedMessage, bool, error) UpdateMessageWithMetadata(ctx context.Context, sessionID, entryID, content string, attachments []messagequeue.MessageAttachment, metadataUpdates map[string]interface{}, queuedBy string) error RemoveEntry(ctx context.Context, sessionID, entryID string) error + MergeIntoAbove(ctx context.Context, sessionID, entryID, queuedBy string) (*messagequeue.QueuedMessage, error) CancelAll(ctx context.Context, sessionID string) (int, error) GetStatus(ctx context.Context, sessionID string) *messagequeue.QueueStatus } +// QueueDrainer drains a single queued entry when the session is promptable. type QueueDrainer interface { DrainQueuedMessage(ctx context.Context, sessionID string) (bool, error) } @@ -84,6 +90,7 @@ func (h *QueueHandlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionMessageQueueAppend, h.wsAppendToQueue) d.RegisterFunc(ws.ActionMessageQueueDrain, h.wsDrainQueue) d.RegisterFunc(ws.ActionMessageQueueRemove, h.wsRemoveEntry) + d.RegisterFunc(ws.ActionMessageQueueMerge, h.wsMergeIntoAbove) } type wsQueueMessageRequest struct { @@ -353,6 +360,59 @@ func (h *QueueHandlers) wsRemoveEntry(ctx context.Context, msg *ws.Message) (*ws return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{fieldEntryID: req.EntryID}) } +// wsMergeIntoAboveRequest is the payload for ActionMessageQueueMerge: the +// session whose queue is modified and the id of the entry to fold into the +// entry directly above it. user_id is forwarded for ownership checks and is +// optional (the server defaults to the reserved "user" identity). +type wsMergeIntoAboveRequest struct { + SessionID string `json:"session_id"` + EntryID string `json:"entry_id"` + UserID string `json:"user_id,omitempty"` +} + +// wsMergeIntoAbove handles ActionMessageQueueMerge, folding the referenced +// queued entry into the entry above it and broadcasting the updated queue. +func (h *QueueHandlers) wsMergeIntoAbove(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsMergeIntoAboveRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.SessionID == "" { + // Required so publishStatus can broadcast the post-merge list. + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil) + } + if req.EntryID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "entry_id is required", nil) + } + if messagequeue.IsReservedQueuedBy(req.UserID) { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, reservedIdentityError(req.UserID), nil) + } + // Default empty user_id to QueuedByUser so the merge ownership guard runs + // against a non-empty owner, mirroring wsUpdateMessage. + queuedBy := req.UserID + if queuedBy == "" { + queuedBy = messagequeue.QueuedByUser + } + + merged, err := h.queueService.MergeIntoAbove(ctx, req.SessionID, req.EntryID, queuedBy) + if err != nil { + if errors.Is(err, messagequeue.ErrEntryNotFound) { + return ws.NewError(msg.ID, msg.Action, queueErrorCodeEntryNotFound, "Queue entry was already drained or not owned by caller", nil) + } + if errors.Is(err, messagequeue.ErrNoMergeTarget) { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "No mergeable message above this entry", nil) + } + if errors.Is(err, messagequeue.ErrMergeReferenceOverflow) { + return ws.NewError(msg.ID, msg.Action, queueErrorCodeMergeReferenceOverflow, err.Error(), 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) + } + + h.publishStatus(ctx, req.SessionID) + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{fieldEntryID: merged.ID}) +} + type wsAppendToQueueRequest struct { SessionID string `json:"session_id"` TaskID string `json:"task_id"` diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go b/apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go new file mode 100644 index 0000000000..15fbacfa3d --- /dev/null +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go @@ -0,0 +1,129 @@ +package handlers + +import ( + "context" + "encoding/json" + "testing" + + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + ws "github.com/kandev/kandev/pkg/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestWsMergeIntoAbove covers the ActionMessageQueueMerge websocket path: +// happy-path merge, ownership rejection, and reserved-identity handling. +func TestWsMergeIntoAbove(t *testing.T) { + t.Run("merges a user entry into the entry above", 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) + + 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.MessageTypeResponse, response.Type) + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(response.Payload, &payload)) + assert.Equal(t, a.ID, payload["entry_id"]) + + status := svc.GetStatus(ctx, "s") + require.Equal(t, 1, status.Count) + assert.Equal(t, "first\n\nsecond", status.Entries[0].Content) + }) + + t.Run("returns validation when the target is the head", func(t *testing.T) { + handlers, svc := setupQueueHandlers(t) + ctx := context.Background() + + head, _ := svc.QueueMessage(ctx, "s", "t", "only", "", "u", false, nil) + + response, err := handlers.wsMergeIntoAbove(ctx, + createTestMessage(t, ws.ActionMessageQueueMerge, map[string]interface{}{ + "session_id": "s", + "entry_id": head.ID, + "user_id": "u", + })) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Equal(t, ws.ErrorCodeValidation, parseError(t, response).Code) + assert.Equal(t, 1, svc.GetStatus(ctx, "s").Count) + }) + + t.Run("returns validation for mismatched sender kinds", func(t *testing.T) { + handlers, svc := setupQueueHandlers(t) + ctx := context.Background() + + _, err := svc.QueueMessageWithMetadata(ctx, "s", "t", "agent prompt", "", messagequeue.QueuedByAgent, false, nil, + map[string]interface{}{messagequeue.MetadataSenderTaskID: "task-1"}) + require.NoError(t, err) + user, err := svc.QueueMessage(ctx, "s", "t", "user prompt", "", "u", false, nil) + require.NoError(t, err) + + response, err := handlers.wsMergeIntoAbove(ctx, + createTestMessage(t, ws.ActionMessageQueueMerge, map[string]interface{}{ + "session_id": "s", + "entry_id": user.ID, + "user_id": "u", + })) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Equal(t, ws.ErrorCodeValidation, parseError(t, response).Code) + assert.Equal(t, 2, svc.GetStatus(ctx, "s").Count) + }) + + t.Run("returns entry_not_found for a drained source", func(t *testing.T) { + handlers, svc := setupQueueHandlers(t) + ctx := context.Background() + + queued, _ := svc.QueueMessage(ctx, "s", "t", "x", "", "u", false, nil) + _, _ = svc.TakeQueued(ctx, "s") + + response, err := handlers.wsMergeIntoAbove(ctx, + createTestMessage(t, ws.ActionMessageQueueMerge, map[string]interface{}{ + "session_id": "s", + "entry_id": queued.ID, + "user_id": "u", + })) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Equal(t, "entry_not_found", parseError(t, response).Code) + }) + + t.Run("rejects reserved user_id", func(t *testing.T) { + handlers, _ := setupQueueHandlers(t) + response, err := handlers.wsMergeIntoAbove(context.Background(), + createTestMessage(t, ws.ActionMessageQueueMerge, map[string]interface{}{ + "session_id": "s", + "entry_id": "e", + "user_id": messagequeue.QueuedByAgent, + })) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Contains(t, parseError(t, response).Message, "may not impersonate the agent identity") + }) + + t.Run("rejects missing session_id", func(t *testing.T) { + handlers, _ := setupQueueHandlers(t) + response, err := handlers.wsMergeIntoAbove(context.Background(), + createTestMessage(t, ws.ActionMessageQueueMerge, map[string]interface{}{"entry_id": "e"})) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Contains(t, parseError(t, response).Message, "session_id is required") + }) + + t.Run("rejects missing entry_id", func(t *testing.T) { + handlers, _ := setupQueueHandlers(t) + response, err := handlers.wsMergeIntoAbove(context.Background(), + createTestMessage(t, ws.ActionMessageQueueMerge, map[string]interface{}{"session_id": "s"})) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Contains(t, parseError(t, response).Message, "entry_id is required") + }) +} diff --git a/apps/backend/internal/orchestrator/messagequeue/merge.go b/apps/backend/internal/orchestrator/messagequeue/merge.go new file mode 100644 index 0000000000..64f9e16ed0 --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/merge.go @@ -0,0 +1,97 @@ +package messagequeue + +import ( + "errors" + "fmt" + + "github.com/kandev/kandev/internal/entityrefs" + apiv1 "github.com/kandev/kandev/pkg/api/v1" +) + +// joinMergeContent concatenates the source content below the target content, +// separated by a blank line. When the target content is empty the source is +// used verbatim so there is no leading blank line. +func joinMergeContent(targetContent, sourceContent string) string { + if targetContent == "" { + return sourceContent + } + return targetContent + "\n\n" + sourceContent +} + +// ErrMergeReferenceOverflow reports a merge whose combined entity-reference +// lists would exceed entityrefs' per-message cap. The merge is rejected +// atomically — neither row is touched — rather than silently dropping +// references that were already persisted. +var ErrMergeReferenceOverflow = errors.New("merge would exceed the per-message entity reference limit") + +// mergeAllowed reports whether the source entry may be folded into the target +// entry under the caller identity queuedBy. User entries merge only into user +// entries owned by the caller; agent entries merge only into agent entries +// produced by the same sender task; workflow/server/system sources and +// reserved in-flight targets are never mergeable. +func mergeAllowed(source, target *QueuedMessage, queuedBy string) bool { + if target == nil || target.IsReservedInFlight() { + return false + } + if source.QueuedBy == QueuedByAgent { + if target.QueuedBy != QueuedByAgent { + return false + } + sourceSender := metadataString(source.Metadata, MetadataSenderTaskID) + return sourceSender != "" && sourceSender == metadataString(target.Metadata, MetadataSenderTaskID) + } + if IsReservedQueuedBy(source.QueuedBy) { + return false + } + return queuedBy != "" && !IsReservedQueuedBy(queuedBy) && + source.QueuedBy == queuedBy && target.QueuedBy == queuedBy +} + +// mergeEntryMetadata returns a copy of the target metadata with +// MetadataEntityReferences replaced by the union of both entries' references, +// normalized and deduplicated by canonical ref. The key is dropped when the +// union is empty. Returns ErrMergeReferenceOverflow when the union would +// exceed the per-message reference cap. +func mergeEntryMetadata(target, source map[string]interface{}) (map[string]interface{}, error) { + merged := make(map[string]interface{}, len(target)+1) + for key, value := range target { + merged[key] = value + } + union, err := unionEntityReferences(target, source) + if err != nil { + return nil, err + } + if len(union) == 0 { + delete(merged, MetadataEntityReferences) + return merged, nil + } + merged[MetadataEntityReferences] = union + return merged, nil +} + +// unionEntityReferences returns the deduplicated union of two entries' entity +// reference lists, normalized from either their Go-struct or JSON-roundtripped +// storage form. When the union would exceed the per-message limit entityrefs +// enforces (MaxReferencesPerMessage), it returns ErrMergeReferenceOverflow so +// the caller can reject the merge atomically instead of silently dropping +// references that were already persisted. +func unionEntityReferences(targetMetadata, sourceMetadata map[string]interface{}) ([]apiv1.EntityReference, error) { + seen := make(map[string]struct{}) + var union []apiv1.EntityReference + for _, references := range [][]apiv1.EntityReference{ + entityrefs.NormalizePersisted(targetMetadata[MetadataEntityReferences]), + entityrefs.NormalizePersisted(sourceMetadata[MetadataEntityReferences]), + } { + for _, ref := range references { + if _, duplicate := seen[ref.Ref]; duplicate { + continue + } + seen[ref.Ref] = struct{}{} + union = append(union, ref) + } + } + if len(union) > entityrefs.MaxReferencesPerMessage { + return nil, fmt.Errorf("%w: at most %d references are allowed", ErrMergeReferenceOverflow, entityrefs.MaxReferencesPerMessage) + } + return union, nil +} diff --git a/apps/backend/internal/orchestrator/messagequeue/repository.go b/apps/backend/internal/orchestrator/messagequeue/repository.go index f1874ba6ca..28b809582f 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository.go @@ -78,6 +78,20 @@ type Repository interface { // unrelated metadata remains unchanged. UpdateContentAndMetadata(ctx context.Context, sessionID, entryID, content string, attachments []MessageAttachment, metadataUpdates map[string]interface{}, queuedBy string) error + // MergeIntoAbove folds the entry identified by sourceID into the entry + // immediately above it (greatest position strictly below the source) within + // the same session. Content, attachments, and entity references are merged + // and the source row is removed, all in one transaction. + // + // The merge is allowed only between entries of the same sender kind: a + // user-owned source merges into a user-owned target owned by queuedBy; an + // agent-owned source merges into an agent-owned target whose + // sender_task_id metadata matches the source's. Anything else — head source, + // mismatched kinds, agent sender-task mismatch, caller not owning both rows, + // or a reserved in-flight lifecycle target — returns ErrNoMergeTarget. A + // missing source returns ErrEntryNotFound. + MergeIntoAbove(ctx context.Context, sessionID, sourceID, queuedBy string) (*QueuedMessage, error) + // DeleteByID removes a single entry. The session scope (`AND session_id = ?`) // is mandatory so a caller can't delete an entry by guessing its UUID across // sessions — the queue_full MCP payload deliberately discloses sibling-task diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go index 7f9f1a9373..5659758cd3 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go @@ -347,6 +347,69 @@ func (r *memoryRepository) UpdateContentAndMetadata(_ context.Context, sessionID return ErrEntryNotFound } +// MergeIntoAbove folds the source entry into the entry directly above it within +// the same session, mirroring the sqlite repository's semantics. The target is +// the entry with the greatest position strictly below the source's — the slice +// may not be position-sorted after ReplaceSession. See +// Repository.MergeIntoAbove for the merge rules and error mapping. +func (r *memoryRepository) MergeIntoAbove(_ context.Context, sessionID, sourceID, queuedBy string) (*QueuedMessage, error) { + r.mu.Lock() + defer r.mu.Unlock() + list, ok := r.entries[sessionID] + if !ok { + return nil, ErrEntryNotFound + } + sourceIndex := -1 + for i, m := range list { + if m.ID == sourceID { + sourceIndex = i + break + } + } + if sourceIndex < 0 { + return nil, ErrEntryNotFound + } + source := list[sourceIndex] + + var target *QueuedMessage + for _, m := range list { + if m.Position >= source.Position { + continue + } + if target == nil || m.Position > target.Position { + target = m + } + } + if target == nil { + return nil, ErrNoMergeTarget + } + if !mergeAllowed(source, target, queuedBy) { + return nil, ErrNoMergeTarget + } + + // Compute every merged value before mutating the target: mergeEntryMetadata + // can reject an over-cap reference union, and the target must stay untouched + // on that path so the failed merge is atomic (mirrors the sqlite + // repository's build-then-apply ordering). + content := joinMergeContent(target.Content, source.Content) + attachments := append(append([]MessageAttachment{}, target.Attachments...), source.Attachments...) + metadata, err := mergeEntryMetadata(target.Metadata, source.Metadata) + if err != nil { + return nil, err + } + target.Content = content + target.Attachments = attachments + target.Metadata = metadata + r.entries[sessionID] = append(list[:sourceIndex], list[sourceIndex+1:]...) + if len(r.entries[sessionID]) == 0 { + delete(r.entries, sessionID) + delete(r.nextPosition, sessionID) + } + + merged := *target + return &merged, nil +} + func (r *memoryRepository) DeleteByID(_ context.Context, sessionID, entryID string) error { r.mu.Lock() defer r.mu.Unlock() diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_postgres_merge_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_postgres_merge_test.go new file mode 100644 index 0000000000..22efe30fa9 --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/repository_postgres_merge_test.go @@ -0,0 +1,104 @@ +package messagequeue + +import ( + "context" + "errors" + "testing" + + _ "github.com/jackc/pgx/v5/stdlib" + + "github.com/kandev/kandev/internal/entityrefs" + "github.com/kandev/kandev/internal/testutil" +) + +// newTestPostgresRepo opens an isolated Postgres schema (skipping unless +// KANDEV_TEST_POSTGRES_DSN is set) and constructs the repository over it. The +// queue repository's SQL is driver-agnostic (Rebind'd ? placeholders, no +// SQLite-only constructs), so the same implementation exercises the merge/drain +// ordering under a real shared-database server. +func newTestPostgresRepo(t *testing.T) Repository { + t.Helper() + dsn := testutil.PostgresDSNFromEnv(t) + db := testutil.OpenIsolatedPostgres(t, dsn) + repo, err := NewSQLiteRepository(db, db) + if err != nil { + t.Fatalf("NewSQLiteRepository(postgres): %v", err) + } + return repo +} + +// TestPostgresRepository_MergeDrainOrdering_DrainWins asserts the drain-wins +// ordering of the merge/drain race: after the source drains, a merge reports +// ErrEntryNotFound and the target is untouched, so no content is lost. +func TestPostgresRepository_MergeDrainOrdering_DrainWins(t *testing.T) { + repo := newTestPostgresRepo(t) + ctx := context.Background() + + target := insertTestEntry(t, repo, "s1", "t1", "first", "user", nil, nil) + source := insertTestEntry(t, repo, "s1", "t1", "second", "user", nil, nil) + + if _, err := repo.TakeByID(ctx, "s1", source.ID); err != nil { + t.Fatalf("drain source: %v", err) + } + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user"); !errors.Is(err, ErrEntryNotFound) { + t.Fatalf("merge after drain error = %v, want ErrEntryNotFound", err) + } + kept, err := repo.ListBySession(ctx, "s1") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(kept) != 1 { + t.Fatalf("kept entries = %d, want 1", len(kept)) + } + if kept[0].ID != target.ID || kept[0].Content != "first" { + t.Errorf("kept entry = %s %q, want target %q", kept[0].ID, kept[0].Content, "first") + } +} + +// TestPostgresRepository_MergeDrainOrdering_MergeWins asserts the merge-wins +// ordering: the drain that follows picks up the merged entry with combined +// content, not the source. +func TestPostgresRepository_MergeDrainOrdering_MergeWins(t *testing.T) { + repo := newTestPostgresRepo(t) + ctx := context.Background() + + target := insertTestEntry(t, repo, "s1", "t1", "first", "user", nil, nil) + source := insertTestEntry(t, repo, "s1", "t1", "second", "user", nil, nil) + + merged, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user") + if err != nil { + t.Fatalf("merge: %v", err) + } + drained, err := repo.TakeHead(ctx, "s1") + if err != nil { + t.Fatalf("drain after merge: %v", err) + } + if drained.ID != target.ID || drained.ID != merged.ID { + t.Errorf("drained id = %s, want merged target %s", drained.ID, merged.ID) + } + if drained.Content != "first\n\nsecond" { + t.Errorf("drained content = %q, want combined content", drained.Content) + } +} + +// TestPostgresRepository_MergeIntoAbove_ReferenceOverflow asserts the overflow +// rejection is atomic under Postgres too: neither row is touched and the queue +// count stays unchanged. +func TestPostgresRepository_MergeIntoAbove_ReferenceOverflow(t *testing.T) { + repo := newTestPostgresRepo(t) + ctx := context.Background() + + target := insertTestEntry(t, repo, "s1", "t1", "first", "user", nil, + map[string]interface{}{MetadataEntityReferences: manyEntityRefsFrom(1, entityrefs.MaxReferencesPerMessage)}) + _ = target + source := insertTestEntry(t, repo, "s1", "t1", "second", "user", nil, + map[string]interface{}{MetadataEntityReferences: manyEntityRefsFrom(entityrefs.MaxReferencesPerMessage+1, entityrefs.MaxReferencesPerMessage)}) + + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user"); !errors.Is(err, ErrMergeReferenceOverflow) { + t.Fatalf("merge error = %v, want ErrMergeReferenceOverflow", err) + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 2 { + t.Errorf("count after rejected overflow merge = %d, want 2", count) + } +} diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go index cb67c60232..eeb4ff1cd7 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "sync" "time" "github.com/google/uuid" @@ -16,18 +17,39 @@ import ( type sqliteRepository struct { db *sqlx.DB // writer ro *sqlx.DB // reader + + mu sync.Mutex + sessionLocks map[string]*sync.Mutex } // NewSQLiteRepository creates a SQLite-backed Repository. The supplied writer // and reader are taken from the shared DB pool. initSchema runs idempotently. func NewSQLiteRepository(writer, reader *sqlx.DB) (Repository, error) { - r := &sqliteRepository{db: writer, ro: reader} + r := &sqliteRepository{db: writer, ro: reader, sessionLocks: make(map[string]*sync.Mutex)} if err := r.initSchema(); err != nil { return nil, fmt.Errorf("messagequeue: init schema: %w", err) } return r, nil } +// withSessionLock serializes the queue mutations of one session so a concurrent +// merge and drain cannot interleave between their reads and writes. It is +// per-session (not global), so unrelated sessions proceed in parallel. The +// affected-row checks remain the authoritative guard across processes (e.g. +// multiple backends sharing a Postgres queue); the lock makes in-process +// merge-wins/drain-wins ordering deterministic. +func (r *sqliteRepository) withSessionLock(sessionID string) func() { + r.mu.Lock() + lock := r.sessionLocks[sessionID] + if lock == nil { + lock = &sync.Mutex{} + r.sessionLocks[sessionID] = lock + } + r.mu.Unlock() + lock.Lock() + return func() { lock.Unlock() } +} + func (r *sqliteRepository) initSchema() error { _, err := r.db.Exec(` CREATE TABLE IF NOT EXISTS queued_messages ( @@ -516,6 +538,11 @@ func (r *sqliteRepository) CountBySession(ctx context.Context, sessionID string) } func (r *sqliteRepository) TakeHead(ctx context.Context, sessionID string) (*QueuedMessage, error) { + // Share the per-session lock with MergeIntoAbove so a drain and a merge on + // the same queue are serialized in-process, not just at the DB layer. + unlock := r.withSessionLock(sessionID) + defer unlock() + tx, err := r.db.BeginTxx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin take tx: %w", err) @@ -561,6 +588,9 @@ func (r *sqliteRepository) TakeHead(ctx context.Context, sessionID string) (*Que } func (r *sqliteRepository) ReserveHead(ctx context.Context, sessionID string) (*QueuedMessage, error) { + unlock := r.withSessionLock(sessionID) + defer unlock() + tx, err := r.db.BeginTxx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin reserve tx: %w", err) @@ -622,6 +652,9 @@ func (r *sqliteRepository) ReserveHead(ctx context.Context, sessionID string) (* } func (r *sqliteRepository) AcknowledgeByID(ctx context.Context, sessionID, entryID string) error { + unlock := r.withSessionLock(sessionID) + defer unlock() + res, err := r.db.ExecContext(ctx, r.db.Rebind(` DELETE FROM queued_messages WHERE id = ? AND session_id = ? `), entryID, sessionID) @@ -648,6 +681,9 @@ func (r *sqliteRepository) AcknowledgeByID(ctx context.Context, sessionID, entry // the SELECT and DELETE can't have it removed out from under the new // session — same session-scope invariant DeleteByID/UpdateContent enforce. func (r *sqliteRepository) TakeByID(ctx context.Context, sessionID, entryID string) (*QueuedMessage, error) { + unlock := r.withSessionLock(sessionID) + defer unlock() + tx, err := r.db.BeginTxx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin take tx: %w", err) @@ -736,6 +772,156 @@ func (r *sqliteRepository) UpdateContentAndMetadata(ctx context.Context, session return tx.Commit() } +// MergeIntoAbove folds the source entry into the entry directly above it within +// the same session, updating the target row and deleting the source row in one +// transaction so a concurrent drain can never observe a half-merged queue. +// The per-session lock serializes against drains (TakeHead/ReserveHead/ +// TakeByID) and the affected-row checks turn a lost write into an +// ErrEntryNotFound rollback rather than a silent content loss. See +// Repository.MergeIntoAbove for the merge rules and error mapping. +func (r *sqliteRepository) MergeIntoAbove(ctx context.Context, sessionID, sourceID, queuedBy string) (*QueuedMessage, error) { + unlock := r.withSessionLock(sessionID) + defer unlock() + + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return nil, fmt.Errorf("begin merge tx: %w", err) + } + defer func() { _ = tx.Rollback() }() + + source, err := readMergeSource(ctx, r, tx, sessionID, sourceID) + if err != nil { + return nil, err + } + + target, err := readMergeTarget(ctx, r, tx, sessionID, source.Position) + if err != nil { + return nil, err + } + + if !mergeAllowed(source, target, queuedBy) { + return nil, ErrNoMergeTarget + } + + content, attachments, metadata, err := buildMergedEntry(target, source) + if err != nil { + return nil, err + } + if err := applyMergeWrites(ctx, r, tx, target, source, content, attachments, metadata, sessionID); err != nil { + return nil, err + } + if err := tx.Commit(); err != nil { + return nil, err + } + + merged := *target + merged.Content = content + merged.Attachments = attachments + merged.Metadata = metadata + return &merged, nil +} + +// readMergeSource loads the source entry by id, mapping a missing row to +// ErrEntryNotFound so the caller can report it without wrapping. +func readMergeSource(ctx context.Context, r *sqliteRepository, tx *sqlx.Tx, sessionID, sourceID string) (*QueuedMessage, error) { + row := tx.QueryRowxContext(ctx, r.db.Rebind(` + SELECT id, session_id, task_id, position, content, model, plan_mode, + attachments_json, metadata_json, queued_at, queued_by + FROM queued_messages + WHERE id = ? AND session_id = ? + `), sourceID, sessionID) + source, err := scanQueuedRow(row) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrEntryNotFound + } + return nil, fmt.Errorf("read merge source: %w", err) + } + return source, nil +} + +// readMergeTarget loads the entry with the greatest position strictly below +// the source, mapping a missing row to ErrNoMergeTarget. +func readMergeTarget(ctx context.Context, r *sqliteRepository, tx *sqlx.Tx, sessionID string, sourcePosition int64) (*QueuedMessage, error) { + row := tx.QueryRowxContext(ctx, r.db.Rebind(` + SELECT id, session_id, task_id, position, content, model, plan_mode, + attachments_json, metadata_json, queued_at, queued_by + FROM queued_messages + WHERE session_id = ? AND position < ? + ORDER BY position DESC + LIMIT 1 + `), sessionID, sourcePosition) + target, err := scanQueuedRow(row) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrNoMergeTarget + } + return nil, fmt.Errorf("read merge target: %w", err) + } + return target, nil +} + +// buildMergedEntry computes the merged content, attachments, and metadata for +// a source folded into its target, returning the metadata marshalled to JSON +// so the caller can write both rows in the same transaction. +func buildMergedEntry(target, source *QueuedMessage) (string, []MessageAttachment, map[string]interface{}, error) { + content := joinMergeContent(target.Content, source.Content) + attachments := append(append([]MessageAttachment{}, target.Attachments...), source.Attachments...) + metadata, err := mergeEntryMetadata(target.Metadata, source.Metadata) + if err != nil { + return "", nil, nil, err + } + return content, attachments, metadata, nil +} + +// applyMergeWrites updates the target row and deletes the source row, turning +// a lost write into ErrEntryNotFound so the transaction rolls back cleanly +// instead of silently dropping a queued message. +func applyMergeWrites(ctx context.Context, r *sqliteRepository, tx *sqlx.Tx, target, source *QueuedMessage, content string, attachments []MessageAttachment, metadata map[string]interface{}, sessionID string) error { + attachmentsJSON, err := marshalAttachments(attachments) + if err != nil { + return err + } + metadataJSON, err := marshalMetadata(metadata) + if err != nil { + return err + } + res, err := tx.ExecContext(ctx, r.db.Rebind(` + UPDATE queued_messages + SET content = ?, attachments_json = ?, metadata_json = ? + WHERE id = ? AND session_id = ? + `), content, attachmentsJSON, metadataJSON, target.ID, sessionID) + if err != nil { + return fmt.Errorf("update merge target: %w", err) + } + affected, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("update merge target rows affected: %w", err) + } + if affected == 0 { + // The target was drained concurrently. Rolling back leaves the source + // untouched instead of folding into a row that no longer exists. + return ErrEntryNotFound + } + res, err = tx.ExecContext(ctx, r.db.Rebind(` + DELETE FROM queued_messages + WHERE id = ? AND session_id = ? + `), source.ID, sessionID) + if err != nil { + return fmt.Errorf("delete merge source: %w", err) + } + affected, err = res.RowsAffected() + if err != nil { + return fmt.Errorf("delete merge source rows affected: %w", err) + } + if affected == 0 { + // The source was drained concurrently; rollback restores the target + // update so the merge is all-or-nothing. + return ErrEntryNotFound + } + return nil +} + func (r *sqliteRepository) DeleteByID(ctx context.Context, sessionID, entryID string) error { res, err := r.db.ExecContext(ctx, r.db.Rebind(` DELETE FROM queued_messages diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_merge_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_merge_test.go new file mode 100644 index 0000000000..8934963008 --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_merge_test.go @@ -0,0 +1,446 @@ +package messagequeue + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/kandev/kandev/internal/entityrefs" + apiv1 "github.com/kandev/kandev/pkg/api/v1" +) + +// insertTestEntry inserts a queued message through the repository under test +// and fails the test on error, returning the inserted entry. +func insertTestEntry(t *testing.T, repo Repository, session, task, content, queuedBy string, attachments []MessageAttachment, metadata map[string]interface{}) *QueuedMessage { + t.Helper() + msg := &QueuedMessage{ + SessionID: session, + TaskID: task, + Content: content, + QueuedBy: queuedBy, + Attachments: attachments, + Metadata: metadata, + } + if err := repo.Insert(context.Background(), msg, 0); err != nil { + t.Fatalf("insert %q: %v", content, err) + } + return msg +} + +// entityRefs builds the metadata reference list for the given issue ids, +// using canonical github/acme-repo references. +func entityRefs(ids ...string) []interface{} { + out := make([]interface{}, 0, len(ids)) + for _, id := range ids { + out = append(out, map[string]interface{}{ + "version": apiv1.EntityReferenceVersion, + "ref": entityrefs.CanonicalRef("github", "issue", "acme/repo", id), + "provider": "github", + "kind": "issue", + "id": id, + "title": "Issue " + id, + "url": "https://github.com/acme/repo/issues/" + id, + "scope": "acme/repo", + }) + } + return out +} + +// manyEntityRefs returns count references starting at id 1. +func manyEntityRefs(count int) []interface{} { + return manyEntityRefsFrom(1, count) +} + +// manyEntityRefsFrom returns count references starting at the given id, +// used to build unions that straddle the per-message reference cap. +func manyEntityRefsFrom(start, count int) []interface{} { + ids := make([]string, count) + for i := range ids { + ids[i] = fmt.Sprintf("%d", start+i) + } + return entityRefs(ids...) +} + +// TestSQLiteRepository_MergeIntoAbove_ReferenceOverflow asserts a merge whose +// deduplicated reference union would exceed the per-message cap is rejected +// atomically: both rows keep their persisted references and the queue count is +// unchanged. +func TestSQLiteRepository_MergeIntoAbove_ReferenceOverflow(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + target := insertTestEntry(t, repo, "s1", "t1", "first", "user", nil, + map[string]interface{}{MetadataEntityReferences: manyEntityRefs(entityrefs.MaxReferencesPerMessage)}) + source := insertTestEntry(t, repo, "s1", "t1", "second", "user", nil, + map[string]interface{}{MetadataEntityReferences: entityRefs(fmt.Sprintf("%d", entityrefs.MaxReferencesPerMessage+1))}) + + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user"); !errors.Is(err, ErrMergeReferenceOverflow) { + t.Fatalf("merge error = %v, want ErrMergeReferenceOverflow", err) + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 2 { + t.Errorf("count after rejected overflow merge = %d, want 2", count) + } + entries, err := repo.ListBySession(ctx, "s1") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(entries) != 2 { + t.Fatalf("entries after rejected overflow merge = %d, want 2", len(entries)) + } + byID := make(map[string]*QueuedMessage, len(entries)) + for i := range entries { + byID[entries[i].ID] = &entries[i] + } + targetRow, sourceRow := byID[target.ID], byID[source.ID] + if targetRow == nil || sourceRow == nil { + t.Fatalf("missing rows after rejected overflow merge: target=%v source=%v", byID[target.ID] != nil, byID[source.ID] != nil) + } + if refs := entityrefs.NormalizePersisted(targetRow.Metadata[MetadataEntityReferences]); len(refs) != entityrefs.MaxReferencesPerMessage { + t.Errorf("target refs len = %d, want %d (untouched)", len(refs), entityrefs.MaxReferencesPerMessage) + } + if refs := entityrefs.NormalizePersisted(sourceRow.Metadata[MetadataEntityReferences]); len(refs) != 1 { + t.Errorf("source refs len = %d, want 1 (untouched)", len(refs)) + } + if sourceRow.Content != "second" { + t.Errorf("source content changed after rejected merge = %q", sourceRow.Content) + } +} + +// TestSQLiteRepository_MergeIntoAbove_ReferenceOverflow_DedupeWithinLimit +// asserts overlapping references stay within the cap after deduplication, so +// merging two cap-sized lists of identical references succeeds. +func TestSQLiteRepository_MergeIntoAbove_ReferenceOverflow_DedupeWithinLimit(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + refs := manyEntityRefs(entityrefs.MaxReferencesPerMessage) + insertTestEntry(t, repo, "s1", "t1", "first", "user", nil, + map[string]interface{}{MetadataEntityReferences: refs}) + source := insertTestEntry(t, repo, "s1", "t1", "second", "user", nil, + map[string]interface{}{MetadataEntityReferences: refs}) + + merged, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user") + if err != nil { + t.Fatalf("merge: %v", err) + } + union := entityrefs.NormalizePersisted(merged.Metadata[MetadataEntityReferences]) + if len(union) != entityrefs.MaxReferencesPerMessage { + t.Errorf("merged refs len = %d, want %d (deduped)", len(union), entityrefs.MaxReferencesPerMessage) + } +} + +// TestSQLiteRepository_MergeDrainOrdering_DrainWins drains the source before the +// merge runs, so the merge must report ErrEntryNotFound without touching the +// target or losing content — the drain-wins ordering of the merge/drain race. +func TestSQLiteRepository_MergeDrainOrdering_DrainWins(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + target := insertTestEntry(t, repo, "s1", "t1", "first", "user", nil, nil) + source := insertTestEntry(t, repo, "s1", "t1", "second", "user", nil, nil) + + if _, err := repo.TakeByID(ctx, "s1", source.ID); err != nil { + t.Fatalf("drain source: %v", err) + } + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user"); !errors.Is(err, ErrEntryNotFound) { + t.Fatalf("merge after drain error = %v, want ErrEntryNotFound", err) + } + kept, err := repo.ListBySession(ctx, "s1") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(kept) != 1 { + t.Fatalf("kept entries = %d, want 1", len(kept)) + } + if kept[0].ID != target.ID || kept[0].Content != "first" { + t.Errorf("kept entry = %s %q, want target %q", kept[0].ID, kept[0].Content, "first") + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 1 { + t.Errorf("count after drain-wins = %d, want 1", count) + } +} + +// TestSQLiteRepository_MergeDrainOrdering_MergeWins merges before the drain runs, +// so the merged entry — not the source — is what later drains, preserving the +// combined content. +func TestSQLiteRepository_MergeDrainOrdering_MergeWins(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + target := insertTestEntry(t, repo, "s1", "t1", "first", "user", nil, nil) + source := insertTestEntry(t, repo, "s1", "t1", "second", "user", nil, nil) + + merged, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user") + if err != nil { + t.Fatalf("merge: %v", err) + } + drained, err := repo.TakeHead(ctx, "s1") + if err != nil { + t.Fatalf("drain after merge: %v", err) + } + if drained.ID != target.ID || drained.ID != merged.ID { + t.Errorf("drained id = %s, want merged target %s", drained.ID, merged.ID) + } + if drained.Content != "first\n\nsecond" { + t.Errorf("drained content = %q, want combined content", drained.Content) + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 0 { + t.Errorf("count after merge-wins = %d, want 0", count) + } +} + +// TestSQLiteRepository_MergeIntoAbove_UserMergesIntoUser covers the user↔user +// happy path: combined content joined with "\n\n", attachments concatenated, +// entity references unioned and deduplicated, target identity preserved, source +// row gone, and the queue count drops by one. +func TestSQLiteRepository_MergeIntoAbove_UserMergesIntoUser(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + target := insertTestEntry(t, repo, "s1", "t1", "first", "user", + []MessageAttachment{{Type: "image", Data: "a", MimeType: "image/png"}}, + map[string]interface{}{ + MetadataEntityReferences: entityRefs("1"), + }) + source := insertTestEntry(t, repo, "s1", "t1", "second", "user", + []MessageAttachment{{Type: "file", Data: "b", MimeType: "text/plain"}}, + map[string]interface{}{ + MetadataEntityReferences: entityRefs("1", "2"), + }) + + merged, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user") + if err != nil { + t.Fatalf("merge: %v", err) + } + if merged == nil { + t.Fatal("expected merged entry, got nil") + } + if merged.ID != target.ID { + t.Errorf("merged id = %s, want target id %s", merged.ID, target.ID) + } + if merged.Content != "first\n\nsecond" { + t.Errorf("merged content = %q, want %q", merged.Content, "first\n\nsecond") + } + if len(merged.Attachments) != 2 { + t.Errorf("merged attachments len = %d, want 2", len(merged.Attachments)) + } + refs := entityrefs.NormalizePersisted(merged.Metadata[MetadataEntityReferences]) + if len(refs) != 2 { + t.Errorf("merged refs len = %d, want 2 (deduped)", len(refs)) + } + if merged.QueuedBy != "user" { + t.Errorf("merged queued_by = %q, want user", merged.QueuedBy) + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 1 { + t.Errorf("count after merge = %d, want 1", count) + } +} + +// TestSQLiteRepository_MergeIntoAbove_EmptyTargetContent ensures an empty target +// content yields the source content verbatim (no leading blank line). +func TestSQLiteRepository_MergeIntoAbove_EmptyTargetContent(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + insertTestEntry(t, repo, "s1", "t1", "", "user", nil, nil) + source := insertTestEntry(t, repo, "s1", "t1", "only source", "user", nil, nil) + + merged, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user") + if err != nil { + t.Fatalf("merge: %v", err) + } + if merged.Content != "only source" { + t.Errorf("merged content = %q, want %q", merged.Content, "only source") + } +} + +// TestSQLiteRepository_MergeIntoAbove_HeadRejected asserts the head entry has no +// target above it and the queue is untouched. +func TestSQLiteRepository_MergeIntoAbove_HeadRejected(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + head := insertTestEntry(t, repo, "s1", "t1", "head", "user", nil, nil) + if _, err := repo.MergeIntoAbove(ctx, "s1", head.ID, "user"); !errors.Is(err, ErrNoMergeTarget) { + t.Fatalf("head merge error = %v, want ErrNoMergeTarget", err) + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 1 { + t.Errorf("count after rejected head merge = %d, want 1", count) + } +} + +// TestSQLiteRepository_MergeIntoAbove_MismatchedKinds rejects merges across +// sender kinds in every direction. +func TestSQLiteRepository_MergeIntoAbove_MismatchedKinds(t *testing.T) { + cases := []struct { + name string + aboveQueuedBy string + sourceQueuedBy string + aboveMetadata map[string]interface{} + sourceMetadata map[string]interface{} + }{ + {name: "agent source above user target", aboveQueuedBy: "user", sourceQueuedBy: QueuedByAgent}, + {name: "user source above agent target", aboveQueuedBy: QueuedByAgent, sourceQueuedBy: "user"}, + {name: "workflow source", aboveQueuedBy: "user", sourceQueuedBy: QueuedByWorkflow}, + {name: "server source", aboveQueuedBy: "user", sourceQueuedBy: QueuedByServer}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + insertTestEntry(t, repo, "s1", "t1", "above", tc.aboveQueuedBy, nil, tc.aboveMetadata) + source := insertTestEntry(t, repo, "s1", "t1", "source", tc.sourceQueuedBy, nil, tc.sourceMetadata) + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user"); !errors.Is(err, ErrNoMergeTarget) { + t.Fatalf("merge error = %v, want ErrNoMergeTarget", err) + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 2 { + t.Errorf("count after rejected merge = %d, want 2", count) + } + }) + } +} + +// TestSQLiteRepository_MergeIntoAbove_AgentSameSender allows an agent entry to +// merge into the agent entry above it when both come from the same sender task. +func TestSQLiteRepository_MergeIntoAbove_AgentSameSender(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + target := insertTestEntry(t, repo, "s1", "t1", "agent first", QueuedByAgent, nil, + map[string]interface{}{MetadataSenderTaskID: "task-7"}) + source := insertTestEntry(t, repo, "s1", "t1", "agent second", QueuedByAgent, nil, + map[string]interface{}{MetadataSenderTaskID: "task-7"}) + + merged, err := repo.MergeIntoAbove(ctx, "s1", source.ID, QueuedByAgent) + if err != nil { + t.Fatalf("merge: %v", err) + } + if merged.ID != target.ID { + t.Errorf("merged id = %s, want target id %s", merged.ID, target.ID) + } + if merged.Content != "agent first\n\nagent second" { + t.Errorf("merged content = %q, want %q", merged.Content, "agent first\n\nagent second") + } + if merged.QueuedBy != QueuedByAgent { + t.Errorf("merged queued_by = %q, want agent", merged.QueuedBy) + } +} + +// TestSQLiteRepository_MergeIntoAbove_AgentDifferentSender rejects agent merges +// across different sender tasks. +func TestSQLiteRepository_MergeIntoAbove_AgentDifferentSender(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + insertTestEntry(t, repo, "s1", "t1", "agent first", QueuedByAgent, nil, + map[string]interface{}{MetadataSenderTaskID: "task-7"}) + source := insertTestEntry(t, repo, "s1", "t1", "agent second", QueuedByAgent, nil, + map[string]interface{}{MetadataSenderTaskID: "task-8"}) + + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, QueuedByAgent); !errors.Is(err, ErrNoMergeTarget) { + t.Fatalf("merge error = %v, want ErrNoMergeTarget", err) + } +} + +// TestSQLiteRepository_MergeIntoAbove_WrongCallerRejected asserts a user merge +// requires the caller identity to own both rows. +func TestSQLiteRepository_MergeIntoAbove_WrongCallerRejected(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + insertTestEntry(t, repo, "s1", "t1", "above", "alice", nil, nil) + source := insertTestEntry(t, repo, "s1", "t1", "source", "alice", nil, nil) + + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "bob"); !errors.Is(err, ErrNoMergeTarget) { + t.Fatalf("merge error = %v, want ErrNoMergeTarget", err) + } +} + +// TestSQLiteRepository_MergeIntoAbove_SourceOwnerMismatch asserts the source +// entry must be owned by the caller even when the target above is: a caller +// must not be able to fold and delete another user's queued message into one of +// their own rows. +func TestSQLiteRepository_MergeIntoAbove_SourceOwnerMismatch(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + insertTestEntry(t, repo, "s1", "t1", "above", "bob", nil, nil) + source := insertTestEntry(t, repo, "s1", "t1", "someone else's message", "alice", nil, nil) + + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "bob"); !errors.Is(err, ErrNoMergeTarget) { + t.Fatalf("merge error = %v, want ErrNoMergeTarget", err) + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 2 { + t.Errorf("count after rejected merge = %d, want 2", count) + } +} + +// TestSQLiteRepository_MergeIntoAbove_SourceMissing returns ErrEntryNotFound for +// a drained or unknown source id. +func TestSQLiteRepository_MergeIntoAbove_SourceMissing(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + if _, err := repo.MergeIntoAbove(ctx, "s1", "missing", "user"); !errors.Is(err, ErrEntryNotFound) { + t.Fatalf("missing source error = %v, want ErrEntryNotFound", err) + } +} + +// TestSQLiteRepository_MergeIntoAbove_ReservedTargetRejected refuses to merge a +// user entry into a reserved in-flight lifecycle target. +func TestSQLiteRepository_MergeIntoAbove_ReservedTargetRejected(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + insertTestEntry(t, repo, "s1", "t1", "inflight", "user", nil, map[string]interface{}{ + MetadataLifecycleDurable: true, + MetadataLifecycleReserved: true, + }) + source := insertTestEntry(t, repo, "s1", "t1", "source", "user", nil, nil) + + if _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "user"); !errors.Is(err, ErrNoMergeTarget) { + t.Fatalf("merge error = %v, want ErrNoMergeTarget", err) + } +} + +// TestSQLiteRepository_MergeIntoAbove_Chain merges C into B then B into A, +// producing a single entry with content "A\n\nB\n\nC" in order. +func TestSQLiteRepository_MergeIntoAbove_Chain(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + a := insertTestEntry(t, repo, "s1", "t1", "A", "user", nil, nil) + b := insertTestEntry(t, repo, "s1", "t1", "B", "user", nil, nil) + c := insertTestEntry(t, repo, "s1", "t1", "C", "user", nil, nil) + + merged, err := repo.MergeIntoAbove(ctx, "s1", c.ID, "user") + if err != nil { + t.Fatalf("merge C into B: %v", err) + } + if merged.ID != b.ID { + t.Errorf("after C merge id = %s, want %s", merged.ID, b.ID) + } + + merged, err = repo.MergeIntoAbove(ctx, "s1", b.ID, "user") + if err != nil { + t.Fatalf("merge B into A: %v", err) + } + if merged.ID != a.ID { + t.Errorf("after B merge id = %s, want %s", merged.ID, a.ID) + } + if merged.Content != "A\n\nB\n\nC" { + t.Errorf("chained content = %q, want %q", merged.Content, "A\n\nB\n\nC") + } + count, _ := repo.CountBySession(ctx, "s1") + if count != 1 { + t.Errorf("count after chain = %d, want 1", count) + } +} diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index 3e11c99c3f..19621c5241 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/apps/backend/internal/orchestrator/messagequeue/service.go @@ -367,6 +367,20 @@ func (s *Service) RemoveEntry(ctx context.Context, sessionID, entryID string) er return nil } +// MergeIntoAbove folds the entry identified by entryID into the entry directly +// 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) { + merged, err := s.repo.MergeIntoAbove(ctx, sessionID, entryID, queuedBy) + if err != nil { + return nil, err + } + s.logger.Info("queued entry merged into entry above", + zap.String("session_id", sessionID), + zap.String("entry_id", merged.ID)) + return merged, nil +} + // CancelAll clears every queued entry for a session. Returns the number of // rows removed. func (s *Service) CancelAll(ctx context.Context, sessionID string) (int, error) { diff --git a/apps/backend/internal/orchestrator/messagequeue/service_test.go b/apps/backend/internal/orchestrator/messagequeue/service_test.go index bec4e46cfc..185c269aac 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service_test.go +++ b/apps/backend/internal/orchestrator/messagequeue/service_test.go @@ -3,12 +3,15 @@ package messagequeue import ( "context" "errors" + "fmt" "sync" "sync/atomic" "testing" "time" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/entityrefs" + apiv1 "github.com/kandev/kandev/pkg/api/v1" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -736,3 +739,163 @@ func TestQueuedTimestamp(t *testing.T) { assert.True(t, msg.QueuedAt.After(before)) assert.True(t, msg.QueuedAt.Before(after)) } + +// TestMemoryRepository_MergeIntoAbove exercises the in-memory repository merge +// with Go-struct entity references (the form the memory repo stores without a +// JSON round-trip) alongside the shared merge rules. +func TestMemoryRepository_MergeIntoAbove(t *testing.T) { + ctx := context.Background() + repo := NewMemoryRepository() + + ref := apiv1.EntityReference{ + Version: apiv1.EntityReferenceVersion, + Ref: entityrefs.CanonicalRef("github", "issue", "acme/repo", "1"), + Provider: "github", + Kind: "issue", + ID: "1", + Title: "Issue 1", + URL: "https://github.com/acme/repo/issues/1", + Scope: "acme/repo", + } + + target := &QueuedMessage{ + SessionID: "s1", TaskID: "t1", Content: "first", QueuedBy: "alice", + Attachments: []MessageAttachment{{Type: "image", Data: "a", MimeType: "image/png"}}, + Metadata: map[string]interface{}{MetadataEntityReferences: []apiv1.EntityReference{ref}}, + } + require.NoError(t, repo.Insert(ctx, target, 0)) + source := &QueuedMessage{ + SessionID: "s1", TaskID: "t1", Content: "second", QueuedBy: "alice", + Attachments: []MessageAttachment{{Type: "file", Data: "b", MimeType: "text/plain"}}, + Metadata: map[string]interface{}{MetadataEntityReferences: []apiv1.EntityReference{ref}}, + } + require.NoError(t, repo.Insert(ctx, source, 0)) + + merged, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "alice") + require.NoError(t, err) + assert.Equal(t, target.ID, merged.ID) + assert.Equal(t, "first\n\nsecond", merged.Content) + assert.Len(t, merged.Attachments, 2) + assert.Len(t, entityrefs.NormalizePersisted(merged.Metadata[MetadataEntityReferences]), 1) + + entries, err := repo.ListBySession(ctx, "s1") + require.NoError(t, err) + assert.Len(t, entries, 1) + assert.Equal(t, "first\n\nsecond", entries[0].Content) +} + +// TestMemoryRepository_MergeIntoAbove_MixedKindsRejected covers the memory repo +// kind gate and missing-source error mapping. +func TestMemoryRepository_MergeIntoAbove_MixedKindsRejected(t *testing.T) { + ctx := context.Background() + repo := NewMemoryRepository() + + require.NoError(t, repo.Insert(ctx, &QueuedMessage{SessionID: "s1", TaskID: "t1", Content: "agent", QueuedBy: QueuedByAgent, Metadata: map[string]interface{}{MetadataSenderTaskID: "task-1"}}, 0)) + source := &QueuedMessage{SessionID: "s1", TaskID: "t1", Content: "user", QueuedBy: "alice"} + require.NoError(t, repo.Insert(ctx, source, 0)) + + _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "alice") + assert.ErrorIs(t, err, ErrNoMergeTarget) + + _, err = repo.MergeIntoAbove(ctx, "s1", "missing", "alice") + assert.ErrorIs(t, err, ErrEntryNotFound) +} + +// TestMemoryRepository_MergeIntoAbove_ReferenceOverflow asserts an over-cap +// reference union is rejected atomically in the in-memory repository: the +// target's content, attachments, and references must stay untouched so a +// failed merge does not leave a partially-folded queue behind. +func TestMemoryRepository_MergeIntoAbove_ReferenceOverflow(t *testing.T) { + ctx := context.Background() + repo := NewMemoryRepository() + + refs := make([]apiv1.EntityReference, 0, entityrefs.MaxReferencesPerMessage) + for i := 0; i < entityrefs.MaxReferencesPerMessage; i++ { + refs = append(refs, apiv1.EntityReference{ + Version: apiv1.EntityReferenceVersion, + Ref: entityrefs.CanonicalRef("github", "issue", "acme/repo", fmt.Sprintf("%d", i)), + Provider: "github", + Kind: "issue", + ID: fmt.Sprintf("%d", i), + Title: "Issue " + fmt.Sprintf("%d", i), + URL: "https://github.com/acme/repo/issues/" + fmt.Sprintf("%d", i), + Scope: "acme/repo", + }) + } + target := &QueuedMessage{ + SessionID: "s1", TaskID: "t1", Content: "first", QueuedBy: "alice", + Attachments: []MessageAttachment{{Type: "image", Data: "a", MimeType: "image/png"}}, + Metadata: map[string]interface{}{MetadataEntityReferences: refs}, + } + require.NoError(t, repo.Insert(ctx, target, 0)) + source := &QueuedMessage{ + SessionID: "s1", TaskID: "t1", Content: "second", QueuedBy: "alice", + Attachments: []MessageAttachment{{Type: "file", Data: "b", MimeType: "text/plain"}}, + Metadata: map[string]interface{}{MetadataEntityReferences: []apiv1.EntityReference{ + {Version: apiv1.EntityReferenceVersion, Ref: entityrefs.CanonicalRef("github", "issue", "acme/repo", "x"), Provider: "github", Kind: "issue", ID: "x", Title: "Issue x", URL: "https://github.com/acme/repo/issues/x", Scope: "acme/repo"}, + }}, + } + require.NoError(t, repo.Insert(ctx, source, 0)) + + _, err := repo.MergeIntoAbove(ctx, "s1", source.ID, "alice") + assert.ErrorIs(t, err, ErrMergeReferenceOverflow) + + entries, err := repo.ListBySession(ctx, "s1") + require.NoError(t, err) + require.Len(t, entries, 2) + byID := make(map[string]*QueuedMessage, len(entries)) + for i := range entries { + byID[entries[i].ID] = &entries[i] + } + assert.Equal(t, "first", byID[target.ID].Content, "target content changed after rejected overflow merge") + assert.Len(t, byID[target.ID].Attachments, 1, "target attachments changed after rejected overflow merge") + assert.Len(t, entityrefs.NormalizePersisted(byID[target.ID].Metadata[MetadataEntityReferences]), entityrefs.MaxReferencesPerMessage, "target refs changed after rejected overflow merge") + assert.Equal(t, "second", byID[source.ID].Content, "source content changed after rejected overflow merge") +} + +// TestService_MergeIntoAbove exercises the service delegation path over the +// memory repository and the mapped errors. +func TestService_MergeIntoAbove(t *testing.T) { + svc := setupService(t) + ctx := context.Background() + + above, 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) + + merged, err := svc.MergeIntoAbove(ctx, "s", source.ID, "alice") + require.NoError(t, err) + assert.Equal(t, above.ID, merged.ID) + assert.Equal(t, "above\n\nsource", merged.Content) + assert.Equal(t, 1, svc.GetStatus(ctx, "s").Count) + + _, err = svc.MergeIntoAbove(ctx, "s", "missing", "alice") + assert.ErrorIs(t, err, ErrEntryNotFound) +} + +// TestMemoryRepository_MergeIntoAbove_UnsortedSlice proves the memory repo +// selects the merge target by greatest position strictly below the source's, +// not by slice order. ReplaceSession can persist an unsorted snapshot, so a +// slice-adjacent target would be wrong: here the source (position 3) sits +// between an unrelated entry (position 1) and the true target (position 2). +func TestMemoryRepository_MergeIntoAbove_UnsortedSlice(t *testing.T) { + ctx := context.Background() + repo := NewMemoryRepository() + + err := repo.ReplaceSession(ctx, "s1", []QueuedMessage{ + {ID: "unrelated", SessionID: "s1", TaskID: "t1", Position: 1, Content: "other", QueuedBy: "alice"}, + {ID: "source", SessionID: "s1", TaskID: "t1", Position: 3, Content: "src", QueuedBy: "alice"}, + {ID: "target", SessionID: "s1", TaskID: "t1", Position: 2, Content: "tgt", QueuedBy: "alice"}, + }, nil) + require.NoError(t, err) + + merged, err := repo.MergeIntoAbove(ctx, "s1", "source", "alice") + require.NoError(t, err) + assert.Equal(t, "target", merged.ID) + assert.Equal(t, "tgt\n\nsrc", merged.Content) + + entries, err := repo.ListBySession(ctx, "s1") + require.NoError(t, err) + assert.Len(t, entries, 2) +} diff --git a/apps/backend/internal/orchestrator/messagequeue/types.go b/apps/backend/internal/orchestrator/messagequeue/types.go index 7a9fc337d9..b9542baa8e 100644 --- a/apps/backend/internal/orchestrator/messagequeue/types.go +++ b/apps/backend/internal/orchestrator/messagequeue/types.go @@ -22,6 +22,12 @@ const ( // IsReservedQueuedBy reports identities owned by backend dispatch paths. // WebSocket/MCP clients may create and mutate only user-owned entries. +// +// MergeIntoAbove is the single controlled exception: a client with session +// access may fold one agent-owned entry into the agent-owned entry above it +// when both carry the same sender_task_id. That preserves the reserved row's +// provenance (the merged entry keeps the target's identity) while consolidating +// additive prompts from one agent into a single delivery. See ADR 0051. func IsReservedQueuedBy(queuedBy string) bool { switch queuedBy { case QueuedByAgent, QueuedByWorkflow, QueuedByServer: @@ -57,6 +63,11 @@ const MetadataLifecycleGeneration = "lifecycle_queue_generation" // carries the flag. const MetadataLifecycleReserved = "lifecycle_reserved_in_flight" +// MetadataSenderTaskID identifies the task that produced an agent message. Two +// agent entries may only merge when their sender task ids match, so the merge +// never mixes prompts issued by different agents. +const MetadataSenderTaskID = "sender_task_id" + // QueueFullErrorCode is the well-known WS / MCP error code surfaced when an // insert would exceed the per-session cap. Shared between the user-side WS // handlers and the inter-task MCP handler so the wire contract stays in sync. @@ -69,6 +80,11 @@ var ( // ErrEntryNotFound is returned when an operation targets an entry that no // longer exists (e.g. it was drained between fetch and update). ErrEntryNotFound = errors.New("queue entry not found") + // ErrNoMergeTarget is returned when a merge source exists but has no valid + // entry above it: the source is the head, the sender kinds differ, agent + // 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") // 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") diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index b726448241..751a2a29ad 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -118,6 +118,7 @@ const ( ActionMessageQueueAppend = "message.queue.append" ActionMessageQueueDrain = "message.queue.drain" // Dispatch one queued entry now when the session is promptable ActionMessageQueueRemove = "message.queue.remove" // Delete a single entry by id + ActionMessageQueueMerge = "message.queue.merge" // Fold an entry into the entry above it ActionMessageQueueStatusChanged = "message.queue.status_changed" // Notification: queue status changed // Workflow template/step actions diff --git a/apps/web/components/task/chat/queued-ghost-list.test.tsx b/apps/web/components/task/chat/queued-ghost-list.test.tsx index 288f106727..f4898bb51b 100644 --- a/apps/web/components/task/chat/queued-ghost-list.test.tsx +++ b/apps/web/components/task/chat/queued-ghost-list.test.tsx @@ -1,9 +1,12 @@ import type { ReactNode } from "react"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { cleanup, fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { StateProvider } from "@/components/state-provider"; import type { QueuedMessage } from "@/lib/state/slices/session/types"; import type { EntityReference } from "@/lib/types/entity-reference"; import { entityReferenceMarkdown } from "@/lib/entity-references/message-references"; +import { QueueEntryNotFoundError } from "@/lib/api/domains/queue-api"; +import { toast } from "sonner"; const useQueueMock = vi.fn(); @@ -37,6 +40,8 @@ import { QueueAffordance } from "./queued-ghost-list"; const SESSION_ID = "sess-1"; const CHIP_ID = "queue-chip"; const PANEL_ID = "queued-ghost-list"; +const QUEUED_BY_USER = "user"; +const MERGE_BUTTON_ID = "queue-entry-merge"; function entry(overrides: Partial = {}): QueuedMessage { return { @@ -46,7 +51,7 @@ function entry(overrides: Partial = {}): QueuedMessage { content: "hello world", plan_mode: false, queued_at: "2026-05-18T00:00:00Z", - queued_by: "user-1", + queued_by: QUEUED_BY_USER, ...overrides, }; } @@ -81,12 +86,17 @@ function baseState(entries: QueuedMessage[]) { drainNext: vi.fn(async () => {}), editEntry: vi.fn(async () => {}), removeEntry: vi.fn(async () => {}), + mergeEntry: vi.fn(async () => {}), refetch: vi.fn(async () => {}), }; } const CHILD =
input
; +function renderQueue(node: ReactNode) { + return render({node}); +} + function pressQueueEscape(): ReturnType { const outerEscapeHandler = vi.fn(); document.addEventListener("keydown", outerEscapeHandler); @@ -100,6 +110,8 @@ function pressQueueEscape(): ReturnType { beforeEach(() => { useQueueMock.mockReset(); + vi.mocked(toast.error).mockClear(); + vi.mocked(toast.success).mockClear(); }); afterEach(() => { @@ -315,3 +327,109 @@ describe("QueueAffordance entity-reference edits", () => { ); }); }); + +describe("QueueAffordance merge wiring", () => { + 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 }), + entry({ id: "q-2", content: "second", queued_by: QUEUED_BY_USER }), + ]); + useQueueMock.mockReturnValue(state); + render({CHILD}); + + fireEvent.click(screen.getByTestId(CHIP_ID)); + + const mergeButtons = screen.getAllByTestId(MERGE_BUTTON_ID); + // Only the second row merges (the head row has nothing above it). + expect(mergeButtons).toHaveLength(1); + fireEvent.click(mergeButtons[0]); + expect(state.mergeEntry).toHaveBeenCalledWith("q-2"); + }); + + it("hides the merge control when the rows have mismatched sender kinds", () => { + const state = queueState([ + entry({ id: "q-1", content: "agent", queued_by: "agent" }), + entry({ id: "q-2", content: "user", queued_by: QUEUED_BY_USER }), + ]); + useQueueMock.mockReturnValue(state); + renderQueue({CHILD}); + + fireEvent.click(screen.getByTestId(CHIP_ID)); + expect(screen.queryByTestId(MERGE_BUTTON_ID)).toBeNull(); + }); + + it("hides the merge control for agent rows from different sender tasks", () => { + const state = queueState([ + entry({ + id: "q-1", + content: "agent a", + queued_by: "agent", + metadata: { sender_task_id: "task-7" }, + }), + entry({ + id: "q-2", + content: "agent b", + queued_by: "agent", + metadata: { sender_task_id: "task-8" }, + }), + ]); + useQueueMock.mockReturnValue(state); + renderQueue({CHILD}); + + fireEvent.click(screen.getByTestId(CHIP_ID)); + expect(screen.queryByTestId(MERGE_BUTTON_ID)).toBeNull(); + }); + + it("toasts an error when the merge fails", async () => { + const { toast } = await import("sonner"); + 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 }), + ]); + state.mergeEntry = vi.fn(async () => { + throw new Error("boom"); + }); + useQueueMock.mockReturnValue(state); + render({CHILD}); + + fireEvent.click(screen.getByTestId(CHIP_ID)); + fireEvent.click(screen.getAllByTestId(MERGE_BUTTON_ID)[0]); + await waitFor(() => + expect(toast.error).toHaveBeenCalledWith("Failed to merge queued messages."), + ); + }); + + it("does not toast on a drain race (QueueEntryNotFoundError)", async () => { + const { toast } = await import("sonner"); + 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 }), + ]); + state.mergeEntry = vi.fn(async () => { + throw new QueueEntryNotFoundError(); + }); + useQueueMock.mockReturnValue(state); + render({CHILD}); + + fireEvent.click(screen.getByTestId(CHIP_ID)); + fireEvent.click(screen.getAllByTestId(MERGE_BUTTON_ID)[0]); + await waitFor(() => expect(state.mergeEntry).toHaveBeenCalledTimes(1)); + expect(toast.error).not.toHaveBeenCalled(); + }); + + it("serializes rapid double-clicks on the same merge row", async () => { + 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 }), + ]); + state.mergeEntry = vi.fn(async () => {}); + useQueueMock.mockReturnValue(state); + render({CHILD}); + + fireEvent.click(screen.getByTestId(CHIP_ID)); + const button = screen.getAllByTestId(MERGE_BUTTON_ID)[0]; + fireEvent.click(button); + fireEvent.click(button); + await waitFor(() => expect(state.mergeEntry).toHaveBeenCalledTimes(1)); + }); +}); diff --git a/apps/web/components/task/chat/queued-ghost-list.tsx b/apps/web/components/task/chat/queued-ghost-list.tsx index 43de3e04dd..43d3be88bb 100644 --- a/apps/web/components/task/chat/queued-ghost-list.tsx +++ b/apps/web/components/task/chat/queued-ghost-list.tsx @@ -1,16 +1,22 @@ "use client"; -import { useCallback, useEffect, useState } from "react"; +import { useCallback, useEffect, useRef, useState } from "react"; import type { ReactNode } from "react"; import { IconLayoutList, IconPlayerPlay, IconTrash, IconX } from "@tabler/icons-react"; import { toast } from "sonner"; +import { useTranslation } from "react-i18next"; import { Button } from "@kandev/ui"; import { Collapsible, CollapsibleContent } from "@kandev/ui/collapsible"; import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip"; import { cn } from "@/lib/utils"; import { stripSystemTags } from "@/lib/utils/system-tags"; +import { MergeReferenceOverflowError, QueueEntryNotFoundError } from "@/lib/api/domains/queue-api"; import { useQueue } from "@/hooks/domains/session/use-queue"; -import { isWorkflowQueuedMessage, QueuedGhostMessage } from "./queued-ghost-message"; +import { + canMergeWithAbove, + isWorkflowQueuedMessage, + QueuedGhostMessage, +} from "./queued-ghost-message"; import type { QueuedMessage } from "@/lib/state/slices/session/types"; import type { EntityReference } from "@/lib/types/entity-reference"; @@ -80,6 +86,7 @@ type QueuePanelHandlerArgs = { entityReferences?: EntityReference[], ) => Promise; removeEntry: (entryId: string) => Promise; + mergeEntry: (entryId: string) => Promise; }; function useQueuePanelHandlers({ @@ -87,7 +94,14 @@ function useQueuePanelHandlers({ drainNext, editEntry, removeEntry, + mergeEntry, }: QueuePanelHandlerArgs) { + // Tracks merge requests still in flight so a rapid second click on the same + // row cannot fire a second request for an entry that is already gone — the + // first click's success refetch removes the row and the second would surface + // a spurious "not found" error. + const pendingMerges = useRef(new Set()); + const { t } = useTranslation(); const handleSave = useCallback( async (entryId: string, content: string, entityReferences: EntityReference[]) => { await editEntry(entryId, content, undefined, entityReferences); @@ -105,6 +119,29 @@ function useQueuePanelHandlers({ }, [removeEntry], ); + const handleMerge = useCallback( + async (entryId: string) => { + if (pendingMerges.current.has(entryId)) return; + pendingMerges.current.add(entryId); + try { + await mergeEntry(entryId); + } catch (err) { + // A drain race (QueueEntryNotFoundError) already triggered a refetch in + // mergeEntry, so the queue view is resynced — no toast needed for the + // benign, self-recovering case. + if (err instanceof QueueEntryNotFoundError) return; + if (err instanceof MergeReferenceOverflowError) { + toast.error(t("chat:mergeReferenceOverflow")); + return; + } + console.error("Failed to merge queued entry:", err); + toast.error(t("chat:failedToMergeQueuedMessages")); + } finally { + pendingMerges.current.delete(entryId); + } + }, + [mergeEntry], + ); const handleClear = useCallback(() => { clearAll().catch((err) => { console.error("Failed to clear queued messages:", err); @@ -118,7 +155,7 @@ function useQueuePanelHandlers({ }); }, [drainNext]); - return { handleSave, handleRemove, handleClear, handleDrain }; + return { handleSave, handleRemove, handleMerge, handleClear, handleDrain }; } /** @@ -135,15 +172,28 @@ export function QueueAffordance({ canDrain = false, renderStatusBar, }: QueueAffordanceProps) { - const { entries, count, max, isFull, isLoading, clearAll, drainNext, editEntry, removeEntry } = - useQueue(sessionId); - const [isOpen, setIsOpen] = useState(false); - const { handleSave, handleRemove, handleClear, handleDrain } = useQueuePanelHandlers({ + const { + entries, + count, + max, + isFull, + isLoading, clearAll, drainNext, editEntry, removeEntry, - }); + mergeEntry, + } = useQueue(sessionId); + const [isOpen, setIsOpen] = useState(false); + const { handleSave, handleRemove, handleMerge, handleClear, handleDrain } = useQueuePanelHandlers( + { + clearAll, + drainNext, + editEntry, + removeEntry, + mergeEntry, + }, + ); // Reset disclosure on session switch or full drain using render-phase state // adjustment (React docs: "Adjusting some state when a prop changes"). This @@ -206,6 +256,7 @@ export function QueueAffordance({ onDrain={handleDrain} onSave={handleSave} onRemove={handleRemove} + onMerge={handleMerge} /> @@ -280,6 +331,7 @@ type QueuePanelProps = { onDrain: () => void; onSave: (entryId: string, content: string, entityReferences: EntityReference[]) => Promise; onRemove: (entryId: string) => Promise; + onMerge: (entryId: string) => Promise; }; function QueuePanel({ @@ -294,6 +346,7 @@ function QueuePanel({ onDrain, onSave, onRemove, + onMerge, }: QueuePanelProps) { return (
onSave(entry.id, content, entityReferences)} onRemove={() => onRemove(entry.id)} + onMerge={() => onMerge(entry.id)} /> ))}
diff --git a/apps/web/components/task/chat/queued-ghost-message.test.tsx b/apps/web/components/task/chat/queued-ghost-message.test.tsx index 84dcf54b70..23cc40ff3b 100644 --- a/apps/web/components/task/chat/queued-ghost-message.test.tsx +++ b/apps/web/components/task/chat/queued-ghost-message.test.tsx @@ -1,6 +1,6 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { cleanup, fireEvent, render, screen, waitFor } from "@testing-library/react"; -import { QueuedGhostMessage } from "./queued-ghost-message"; +import { QueuedGhostMessage, canMergeEntry, canMergeWithAbove } from "./queued-ghost-message"; import { StateProvider } from "@/components/state-provider"; import { ToastProvider } from "@/components/toast-provider"; import { entityReferenceMarkdown } from "@/lib/entity-references/message-references"; @@ -25,6 +25,9 @@ const PNG_BASE64 = const ATTACHMENT_1_ALT = "Attachment 1"; const OPEN_ATTACHMENT_1_LABEL = "Open Attachment 1"; const FULL_SIZE_ATTACHMENT_1_ALT = "Full size Attachment 1"; +const MERGE_TESTID = "queue-entry-merge"; +const EDIT_TITLE = "Edit queued message"; +const MERGE_TITLE = "Merge with above"; function entry(overrides: Partial = {}): QueuedMessage { return { @@ -181,7 +184,7 @@ describe("QueuedGhostMessage attachment thumbnails", () => { onRemove={() => {}} />, ); - fireEvent.click(screen.getByTitle("Edit queued message")); + fireEvent.click(screen.getByTitle(EDIT_TITLE)); const img = screen.getByAltText(ATTACHMENT_1_ALT) as HTMLImageElement; expect(img.className).not.toContain("cursor-pointer"); expect(screen.queryByRole("button", { name: OPEN_ATTACHMENT_1_LABEL })).toBeNull(); @@ -236,7 +239,7 @@ describe("QueuedGhostMessage entity references", () => { />, ); - fireEvent.click(screen.getByTitle("Edit queued message")); + fireEvent.click(screen.getByTitle(EDIT_TITLE)); fireEvent.change(screen.getByTestId("queue-edit-textarea"), { target: { value: edited } }); fireEvent.click(screen.getByRole("button", { name: "Save" })); @@ -259,7 +262,7 @@ describe("QueuedGhostMessage entity references", () => { />, ); - fireEvent.click(screen.getByTitle("Edit queued message")); + fireEvent.click(screen.getByTitle(EDIT_TITLE)); fireEvent.change(screen.getByTestId("queue-edit-textarea"), { target: { value: "reference removed" }, }); @@ -268,3 +271,166 @@ describe("QueuedGhostMessage entity references", () => { await waitFor(() => expect(onSave).toHaveBeenCalledWith("reference removed", [])); }); }); + +describe("QueuedGhostMessage merge control", () => { + it("hides the merge button for the first entry (no entry above)", () => { + render( + {}} onRemove={() => {}} />, + ); + expect(screen.queryByTestId(MERGE_TESTID)).toBeNull(); + }); + + it("shows the merge button when canMerge is set", () => { + render( + {}} + onRemove={() => {}} + />, + ); + const button = screen.getByTestId(MERGE_TESTID); + expect(button.getAttribute("title")).toBe(MERGE_TITLE); + }); + + it("clicking the merge button invokes onMerge", () => { + const onMerge = vi.fn(); + render( + {}} + onRemove={() => {}} + />, + ); + fireEvent.click(screen.getByTestId(MERGE_TESTID)); + expect(onMerge).toHaveBeenCalledTimes(1); + }); + + it("hides the merge button for workflow entries", () => { + render( + {}} + onRemove={() => {}} + />, + ); + expect(screen.queryByTestId(MERGE_TESTID)).toBeNull(); + }); + + it("hides the merge button while editing", () => { + render( + {}} + onRemove={() => {}} + />, + ); + fireEvent.click(screen.getByTitle(EDIT_TITLE)); + expect(screen.queryByTestId(MERGE_TESTID)).toBeNull(); + }); +}); + +describe("canMergeEntry / canMergeWithAbove gating", () => { + it("treats user entries as mergeable", () => { + expect(canMergeEntry(entry({ queued_by: "user-1" }))).toBe(true); + }); + + it("treats agent entries as mergeable", () => { + expect(canMergeEntry(entry({ queued_by: "agent" }))).toBe(true); + }); + + it("treats workflow and system entries as non-mergeable", () => { + expect(canMergeEntry(entry({ queued_by: "workflow" }))).toBe(false); + expect(canMergeEntry(entry({ queued_by: "" }))).toBe(false); + }); + + it("treats server entries as non-mergeable", () => { + expect(canMergeEntry(entry({ queued_by: "server" }))).toBe(false); + }); + + it("allows user behind user owned by the caller", () => { + const above = entry({ id: "q-a", queued_by: "user-1" }); + const below = entry({ id: "q-b", queued_by: "user-1" }); + expect(canMergeWithAbove(below, above, "user-1")).toBe(true); + }); + + it("allows user rows under the default user caller identity", () => { + const above = entry({ id: "q-a", queued_by: "user" }); + const below = entry({ id: "q-b", queued_by: "user" }); + expect(canMergeWithAbove(below, above)).toBe(true); + }); + + it("rejects user behind user owned by a different owner", () => { + const above = entry({ id: "q-a", queued_by: "user-1" }); + const below = entry({ id: "q-b", queued_by: "user-2" }); + expect(canMergeWithAbove(below, above, "user-1")).toBe(false); + expect(canMergeWithAbove(below, above, "user-2")).toBe(false); + }); + + it("rejects user rows the caller does not own", () => { + const above = entry({ id: "q-a", queued_by: "user-1" }); + const below = entry({ id: "q-b", queued_by: "user-1" }); + expect(canMergeWithAbove(below, above, "user-2")).toBe(false); + }); + + it("rejects user behind a server row", () => { + const above = entry({ id: "q-a", queued_by: "server" }); + const below = entry({ id: "q-b", queued_by: "user-1" }); + expect(canMergeWithAbove(below, above, "user-1")).toBe(false); + }); + + it("allows agent behind agent from the same sender task", () => { + const above = entry({ id: "q-a", queued_by: "agent", metadata: { sender_task_id: "task-7" } }); + const below = entry({ id: "q-b", queued_by: "agent", metadata: { sender_task_id: "task-7" } }); + expect(canMergeWithAbove(below, above)).toBe(true); + }); + + it("rejects agent behind agent from different sender tasks", () => { + const above = entry({ id: "q-a", queued_by: "agent", metadata: { sender_task_id: "task-7" } }); + const below = entry({ id: "q-b", queued_by: "agent", metadata: { sender_task_id: "task-8" } }); + expect(canMergeWithAbove(below, above)).toBe(false); + }); + + it("rejects agent behind agent when sender_task_id is missing", () => { + const above = entry({ id: "q-a", queued_by: "agent" }); + const below = entry({ id: "q-b", queued_by: "agent" }); + expect(canMergeWithAbove(below, above)).toBe(false); + }); + + it("rejects user behind agent and agent behind user", () => { + const agent = entry({ id: "q-a", queued_by: "agent", metadata: { sender_task_id: "task-7" } }); + const user = entry({ id: "q-b", queued_by: "user-1" }); + expect(canMergeWithAbove(user, agent)).toBe(false); + expect(canMergeWithAbove(agent, user)).toBe(false); + }); + + it("rejects when no entry is above", () => { + expect(canMergeWithAbove(entry(), undefined)).toBe(false); + }); + + it("rejects a workflow entry behind a user entry", () => { + const above = entry({ id: "q-a", queued_by: "user-1" }); + const below = entry({ id: "q-b", queued_by: "workflow" }); + expect(canMergeWithAbove(below, above)).toBe(false); + }); + + it("rejects a server entry behind a user entry", () => { + const above = entry({ id: "q-a", queued_by: "user-1" }); + const below = entry({ id: "q-b", queued_by: "server" }); + expect(canMergeWithAbove(below, above)).toBe(false); + }); +}); diff --git a/apps/web/components/task/chat/queued-ghost-message.tsx b/apps/web/components/task/chat/queued-ghost-message.tsx index f646571139..f1648cab0c 100644 --- a/apps/web/components/task/chat/queued-ghost-message.tsx +++ b/apps/web/components/task/chat/queued-ghost-message.tsx @@ -19,9 +19,11 @@ import { IconRobot, IconUser, IconX, + IconArrowMerge, } from "@tabler/icons-react"; import ReactMarkdown from "react-markdown"; import { toast } from "sonner"; +import { useTranslation } from "react-i18next"; import { Button } from "@kandev/ui"; import { Textarea } from "@kandev/ui/textarea"; import { cn } from "@/lib/utils"; @@ -114,6 +116,39 @@ function senderKindOf(entry: QueuedMessage): SenderKind { return "user"; } +/** A message may serve as a merge source or target only when its sender kind is + * "user" or "agent". Server-owned rows (queued_by="server") are reserved + * backend rows and are never mergeable. */ +export function canMergeEntry(entry: QueuedMessage): boolean { + if (entry.queued_by === "server") return false; + const kind = senderKindOf(entry); + return kind === "user" || kind === "agent"; +} + +/** Two messages may merge when both are mergeable and share the same sender + * kind. Agent entries additionally require an identical non-empty + * sender_task_id; user entries require both rows to be owned by the caller + * identity (callerId defaults to the backend's QueuedByUser, which is what the + * SPA sends when it omits user_id). Mirrors the backend's mergeAllowed + * predicate so the button never appears for merges the server would reject. */ +export function canMergeWithAbove( + entry: QueuedMessage, + above: QueuedMessage | undefined, + callerId = "user", +): boolean { + if (!above) return false; + if (!canMergeEntry(entry) || !canMergeEntry(above)) return false; + if (senderKindOf(entry) !== senderKindOf(above)) return false; + if (senderKindOf(entry) === "agent") { + const senderTaskId = getSenderTaskInfo(entry)?.id; + return senderTaskId !== undefined && senderTaskId === getSenderTaskInfo(above)?.id; + } + if (senderKindOf(entry) === "user") { + return callerId !== "" && entry.queued_by === callerId && above.queued_by === callerId; + } + return false; +} + function senderLabel(entry: QueuedMessage): string { const kind = senderKindOf(entry); if (kind === "agent") { @@ -256,8 +291,10 @@ type DisplayViewProps = { entityReferences: readonly EntityReference[]; positionLabel: string; canEdit: boolean; + canMerge: boolean; onStartEdit: () => void; onRemove: () => void; + onMerge?: () => void | Promise; }; /** Rough threshold above which we offer a per-row expand toggle. Two lines of @@ -270,13 +307,102 @@ function shouldOfferExpand(text: string): boolean { return text.length > EXPAND_THRESHOLD || text.includes("\n"); } +type RowActionsProps = { + canExpand: boolean; + expanded: boolean; + canMerge: boolean; + canEdit: boolean; + onToggleExpand: () => void; + onMerge?: () => void | Promise; + onStartEdit: () => void; + onRemove: () => void; +}; + +function RowActions({ + canExpand, + expanded, + canMerge, + canEdit, + onToggleExpand, + onMerge, + onStartEdit, + onRemove, +}: RowActionsProps) { + const { t } = useTranslation(); + return ( +
+ {canExpand && ( + + )} + {canMerge && onMerge && ( + + )} + {canEdit && ( + <> + + + + )} +
+ ); +} + function DisplayView({ entry, entityReferences, positionLabel, canEdit, + canMerge, onStartEdit, onRemove, + onMerge, }: DisplayViewProps) { const visible = stripSystemTags(entry.content); const attachments = (entry.attachments ?? []) as QueuedAttachment[]; @@ -320,55 +446,16 @@ function DisplayView({ )} -
- {canExpand && ( - - )} - {canEdit && ( - <> - - - - )} -
+ setExpanded((v) => !v)} + onMerge={onMerge} + onStartEdit={onStartEdit} + onRemove={onRemove} + /> ); } @@ -382,14 +469,24 @@ type QueuedGhostMessageProps = { * entry's queued_by. Inter-task entries are visible but read-only. */ canEdit: boolean; + /** + * Merge-with-above control; only shown when this row and the one above it + * share a sender kind. Optional so standalone renders (tests) can omit it. + */ + canMerge?: boolean; onSave: (content: string, entityReferences: EntityReference[]) => Promise; onRemove: () => void | Promise; + /** Fold this entry into the one above it. */ + onMerge?: () => void | Promise; /** Called after edit save/cancel so the parent can refocus the chat input. */ onEditComplete?: () => void; }; export const QueuedGhostMessage = forwardRef( - function QueuedGhostMessage({ entry, index, canEdit, onSave, onRemove, onEditComplete }, ref) { + function QueuedGhostMessage( + { entry, index, canEdit, canMerge = false, onSave, onRemove, onMerge, onEditComplete }, + ref, + ) { const [editing, setEditing] = useState(false); const [value, setValue] = useState(entry.content); const [saving, setSaving] = useState(false); @@ -398,6 +495,10 @@ export const QueuedGhostMessage = forwardRef entityReferencesFromMetadata(entry.metadata), [entry.metadata], ); + // A row is mergeable only when both the caller gate and the entry itself + // pass: workflow/server/system rows are reserved and never mergeable, even + // if a parent computed canMerge for a different row. + const effectiveCanMerge = canMerge && canMergeEntry(entry); useEffect(() => { if (!editing) setValue(entry.content); @@ -480,8 +581,10 @@ export const QueuedGhostMessage = forwardRef )} diff --git a/apps/web/e2e/tests/chat/message-queue.spec.ts b/apps/web/e2e/tests/chat/message-queue.spec.ts index e494e0f547..94ed4761c8 100644 --- a/apps/web/e2e/tests/chat/message-queue.spec.ts +++ b/apps/web/e2e/tests/chat/message-queue.spec.ts @@ -325,6 +325,45 @@ test.describe("Task session queue", () => { expect(metrics.overflowY).toBe("auto"); }); + test("merges a queued message into the message above it", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(90_000); + + const session = await seedTaskAndWaitForIdle(testPage, apiClient, seedData, "Queue merge test"); + + // Send a slow command to keep the agent busy while we queue two messages. + await session.sendMessage("/slow 10s"); + await expect(session.agentStatus()).toBeVisible({ timeout: 15_000 }); + await testPage.waitForTimeout(500); + + const editor = testPage.locator(".tiptap.ProseMirror").first(); + await typeWhileBusy(testPage, editor, "first message"); + const submitBtn = testPage.getByTestId("submit-message-button"); + await expect(submitBtn).toBeVisible({ timeout: 5_000 }); + await submitBtn.click(); + + await typeWhileBusy(testPage, editor, "second message"); + await expect(submitBtn).toBeVisible({ timeout: 5_000 }); + await submitBtn.click(); + + // Expand the queue panel; the second row offers the merge control. + await openQueuePanel(testPage); + const mergeButtons = testPage.getByTestId("queue-entry-merge"); + await expect(mergeButtons).toHaveCount(1, { timeout: 5_000 }); + await mergeButtons.click(); + + // After the merge the queue holds a single entry whose text contains both + // messages, and the panel header reflects one queued message. + const entries = testPage.getByTestId("queue-entry-text"); + await expect(entries).toHaveCount(1, { timeout: 5_000 }); + await expect(entries.first()).toContainText("first message"); + await expect(entries.first()).toContainText("second message"); + await expect(testPage.getByTestId("queued-ghost-list")).toContainText(/1 of 10/); + }); + test("queue message with plan mode enabled via submit button", async ({ testPage, apiClient, diff --git a/apps/web/hooks/domains/session/use-queue.test.ts b/apps/web/hooks/domains/session/use-queue.test.ts index 31f1b1aaf7..1260e486d8 100644 --- a/apps/web/hooks/domains/session/use-queue.test.ts +++ b/apps/web/hooks/domains/session/use-queue.test.ts @@ -13,6 +13,7 @@ const queueApiMock = vi.hoisted(() => { getQueueStatus: vi.fn(), updateQueuedMessage: vi.fn(), removeQueuedEntry: vi.fn(), + mergeQueuedEntry: vi.fn(), }; }); @@ -203,3 +204,105 @@ describe("useQueue", () => { }); }); }); + +describe("useQueue mergeEntry", () => { + beforeEach(() => { + resetMockState(); + setDocumentVisibility("visible"); + queueApiMock.getQueueStatus.mockResolvedValue({ entries: [], count: 0, max: 10 }); + }); + + it("merges an entry and refetches the queue", async () => { + queueApiMock.mergeQueuedEntry.mockResolvedValue({ entry_id: "q-1" }); + const { result } = renderHook(() => useQueue(SESSION_ID)); + await waitFor(() => expect(queueApiMock.getQueueStatus).toHaveBeenCalled()); + queueApiMock.getQueueStatus.mockClear(); + + await act(async () => { + await result.current.mergeEntry("q-2"); + }); + + expect(queueApiMock.mergeQueuedEntry).toHaveBeenCalledWith({ + session_id: SESSION_ID, + entry_id: "q-2", + }); + expect(queueApiMock.getQueueStatus).toHaveBeenCalledWith(SESSION_ID); + }); + + it("forwards an explicit user_id on merge", async () => { + queueApiMock.mergeQueuedEntry.mockResolvedValue({ entry_id: "q-1" }); + const { result } = renderHook(() => useQueue(SESSION_ID)); + await waitFor(() => expect(queueApiMock.getQueueStatus).toHaveBeenCalled()); + + await act(async () => { + await result.current.mergeEntry("q-2", "alice"); + }); + + expect(queueApiMock.mergeQueuedEntry).toHaveBeenCalledWith({ + session_id: SESSION_ID, + entry_id: "q-2", + user_id: "alice", + }); + }); + + it("refetches the queue when the merge target was already drained", async () => { + queueApiMock.mergeQueuedEntry.mockRejectedValue(new queueApiMock.QueueEntryNotFoundError()); + const { result } = renderHook(() => useQueue(SESSION_ID)); + await waitFor(() => expect(queueApiMock.getQueueStatus).toHaveBeenCalled()); + queueApiMock.getQueueStatus.mockClear(); + + await act(async () => { + await expect(result.current.mergeEntry("q-2")).rejects.toThrow( + queueApiMock.QueueEntryNotFoundError, + ); + }); + + expect(queueApiMock.getQueueStatus).toHaveBeenCalledWith(SESSION_ID); + }); +}); + +describe("useQueue clearAll", () => { + beforeEach(() => { + resetMockState(); + setDocumentVisibility("visible"); + queueApiMock.getQueueStatus.mockResolvedValue({ entries: [], count: 0, max: 10 }); + queueApiMock.clearQueue.mockResolvedValue(undefined); + }); + + it("discards an in-flight refetch that resolves after the clear", async () => { + const { result } = renderHook(() => useQueue(SESSION_ID)); + await waitFor(() => expect(queueApiMock.getQueueStatus).toHaveBeenCalled()); + + // A refetch starts before the clear and resolves afterwards with the + // pre-clear entries. + let resolveStale: (status: { entries: QueuedMessage[]; count: number; max: number }) => void; + queueApiMock.getQueueStatus.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveStale = resolve; + }), + ); + let staleRefetch: Promise; + act(() => { + staleRefetch = result.current.refetch(); + }); + expect(mockState.setQueueEntries).toHaveBeenCalledTimes(1); + + await act(async () => { + await result.current.clearAll(); + }); + + await act(async () => { + resolveStale!({ entries: [entry({ id: "pre-clear" })], count: 1, max: 10 }); + await staleRefetch!; + }); + + // The stale pre-clear snapshot must never be applied; the empty snapshot + // from clearAll stays the last one written. + expect(mockState.setQueueEntries).not.toHaveBeenCalledWith( + SESSION_ID, + [entry({ id: "pre-clear" })], + { count: 1, max: 10 }, + ); + }); +}); diff --git a/apps/web/hooks/domains/session/use-queue.ts b/apps/web/hooks/domains/session/use-queue.ts index f114ce71da..8598c48f84 100644 --- a/apps/web/hooks/domains/session/use-queue.ts +++ b/apps/web/hooks/domains/session/use-queue.ts @@ -1,4 +1,4 @@ -import { useEffect, useCallback } from "react"; +import { useEffect, useCallback, useRef } from "react"; import { useAppStore } from "@/components/state-provider"; import { useForegroundRefresh } from "@/hooks/use-foreground-refresh"; import { @@ -8,6 +8,7 @@ import { getQueueStatus, updateQueuedMessage, removeQueuedEntry, + mergeQueuedEntry, QueueEntryNotFoundError, } from "@/lib/api/domains/queue-api"; import type { QueuedMessage } from "@/lib/state/slices/session/types"; @@ -82,14 +83,25 @@ function useQueueActions({ setQueueLoading, metaMax, }: QueueActionsArgs) { + // Monotonic per-session refetch version. Concurrent mutations (merge, edit, + // remove, drain) all refetch; without a guard an older in-flight + // getQueueStatus response can land after a newer one and restore a stale + // snapshot (e.g. the pre-merge row). Discard any response that is no longer + // the latest issued for its session. + const refetchVersion = useRef>({}); const refetch = useCallback( async (sid: string) => { + const version = (refetchVersion.current[sid] ?? 0) + 1; + refetchVersion.current[sid] = version; try { setQueueLoading(sid, true); const status = await getQueueStatus(sid); + if (refetchVersion.current[sid] !== version) return; setQueueEntries(sid, status.entries ?? [], { count: status.count, max: status.max }); } finally { - setQueueLoading(sid, false); + if (refetchVersion.current[sid] === version) { + setQueueLoading(sid, false); + } } }, [setQueueEntries, setQueueLoading], @@ -129,6 +141,10 @@ function useQueueActions({ setQueueLoading(sessionId, true); try { await clearQueue(sessionId); + // Invalidate any in-flight refetch so a pre-clear getQueueStatus + // response cannot land after the empty snapshot and restore the + // entries that were just drained. + refetchVersion.current[sessionId] = (refetchVersion.current[sessionId] ?? 0) + 1; // Reset to a neutral capacity snapshot; the next status_changed event // will replace it with the authoritative server value. Using the // pre-clear entry count as a fallback for `max` was wrong (it would @@ -141,6 +157,24 @@ function useQueueActions({ const drainNext = useDrainNextAction(sessionId, setQueueLoading, refetch); + const { editEntry, removeEntry, mergeEntry } = useEntryMutations({ + sessionId, + removeQueueEntry, + refetch, + }); + + return { refetch, queue, clearAll, drainNext, editEntry, removeEntry, mergeEntry }; +} + +type EntryMutationsArgs = { + sessionId: string | null; + removeQueueEntry: ReturnType["removeQueueEntry"]; + refetch: (sid: string) => Promise; +}; + +/** Entry-level mutations (edit / remove / merge) that refetch on success and + * resync on a drain race (QueueEntryNotFoundError). */ +function useEntryMutations({ sessionId, removeQueueEntry, refetch }: EntryMutationsArgs) { const editEntry = useCallback( async ( entryId: string, @@ -183,7 +217,23 @@ function useQueueActions({ [sessionId, refetch, removeQueueEntry], ); - return { refetch, queue, clearAll, drainNext, editEntry, removeEntry }; + const mergeEntry = useCallback( + async (entryId: string, userId?: string) => { + if (!sessionId) return; + try { + await mergeQueuedEntry({ session_id: sessionId, entry_id: entryId, user_id: userId }); + await refetch(sessionId); + } catch (err) { + if (err instanceof QueueEntryNotFoundError) { + await refetch(sessionId); + } + throw err; + } + }, + [sessionId, refetch], + ); + + return { editEntry, removeEntry, mergeEntry }; } /** @@ -198,13 +248,14 @@ export function useQueue(sessionId: string | null) { const state = useQueueState(sessionId); const { entries, meta, isLoading } = state; const connectionStatus = useAppStore((appState) => appState.connection.status); - const { refetch, queue, clearAll, drainNext, editEntry, removeEntry } = useQueueActions({ - sessionId, - setQueueEntries: state.setQueueEntries, - removeQueueEntry: state.removeQueueEntry, - setQueueLoading: state.setQueueLoading, - metaMax: meta?.max, - }); + const { refetch, queue, clearAll, drainNext, editEntry, removeEntry, mergeEntry } = + useQueueActions({ + sessionId, + setQueueEntries: state.setQueueEntries, + removeQueueEntry: state.removeQueueEntry, + setQueueLoading: state.setQueueLoading, + metaMax: meta?.max, + }); useEffect(() => { if (!sessionId) return; @@ -242,6 +293,7 @@ export function useQueue(sessionId: string | null) { drainNext, editEntry, removeEntry, + mergeEntry, refetch: refetchBound, }; } diff --git a/apps/web/lib/api/domains/queue-api.test.ts b/apps/web/lib/api/domains/queue-api.test.ts index f502b36d7c..1173ca816a 100644 --- a/apps/web/lib/api/domains/queue-api.test.ts +++ b/apps/web/lib/api/domains/queue-api.test.ts @@ -8,8 +8,10 @@ vi.mock("@/lib/ws/connection", () => ({ })); import { + MergeReferenceOverflowError, QueueFullError, QueueEntryNotFoundError, + mergeQueuedEntry, queueMessage, rethrowQueueError, updateQueuedMessage, @@ -77,6 +79,15 @@ describe("rethrowQueueError", () => { ).toThrow(QueueEntryNotFoundError); }); + it("maps merge_reference_overflow errors to MergeReferenceOverflowError", () => { + expect(() => + rethrowQueueError({ + code: "merge_reference_overflow", + message: "merge would exceed the per-message entity reference limit", + }), + ).toThrow(MergeReferenceOverflowError); + }); + it("rethrows non-queue WS errors as plain Error instances", () => { let caught: unknown; try { @@ -168,3 +179,57 @@ describe("queue reference payloads", () => { }); }); }); + +describe("mergeQueuedEntry", () => { + it("forwards session, entry, and caller identity through message.queue.merge", async () => { + const request = vi.fn().mockResolvedValue({ entry_id: "q-a" }); + getWebSocketClientMock.mockReturnValue({ request }); + + await mergeQueuedEntry({ + session_id: "session-1", + entry_id: "q-b", + user_id: "user-1", + }); + + expect(request).toHaveBeenCalledWith("message.queue.merge", { + session_id: "session-1", + entry_id: "q-b", + user_id: "user-1", + }); + }); + + it("omits user_id when the caller identity is unknown", async () => { + const request = vi.fn().mockResolvedValue({ entry_id: "q-a" }); + getWebSocketClientMock.mockReturnValue({ request }); + + await mergeQueuedEntry({ session_id: "session-1", entry_id: "q-b" }); + + expect(request).toHaveBeenCalledWith("message.queue.merge", { + session_id: "session-1", + entry_id: "q-b", + }); + }); + + it("maps a drained-source rejection to QueueEntryNotFoundError", async () => { + const request = vi + .fn() + .mockRejectedValue({ code: "entry_not_found", message: "Already drained" }); + getWebSocketClientMock.mockReturnValue({ request }); + + await expect( + mergeQueuedEntry({ session_id: "session-1", entry_id: "q-b" }), + ).rejects.toBeInstanceOf(QueueEntryNotFoundError); + }); + + it("maps a reference-overflow rejection to MergeReferenceOverflowError", async () => { + const request = vi.fn().mockRejectedValue({ + code: "merge_reference_overflow", + message: "merge would exceed the per-message entity reference limit", + }); + getWebSocketClientMock.mockReturnValue({ request }); + + await expect( + mergeQueuedEntry({ session_id: "session-1", entry_id: "q-b" }), + ).rejects.toBeInstanceOf(MergeReferenceOverflowError); + }); +}); diff --git a/apps/web/lib/api/domains/queue-api.ts b/apps/web/lib/api/domains/queue-api.ts index f2878cb122..2541d7b1e2 100644 --- a/apps/web/lib/api/domains/queue-api.ts +++ b/apps/web/lib/api/domains/queue-api.ts @@ -27,6 +27,17 @@ export class QueueEntryNotFoundError extends Error { } } +/** Error thrown when a merge would push the combined entity references past + * the per-message cap; the server rejects the merge atomically instead of + * dropping references that were already persisted. */ +export class MergeReferenceOverflowError extends Error { + readonly code = "merge_reference_overflow"; + constructor() { + super("Merging would exceed the per-message entity reference limit."); + this.name = "MergeReferenceOverflowError"; + } +} + type WSError = { code?: string; message?: string; @@ -57,6 +68,9 @@ export function rethrowQueueError(err: unknown): never { if (wsErr?.code === "entry_not_found") { throw new QueueEntryNotFoundError(); } + if (wsErr?.code === "merge_reference_overflow") { + throw new MergeReferenceOverflowError(); + } if (wsErr?.message) { throw new Error(wsErr.message); } @@ -187,3 +201,20 @@ export async function removeQueuedEntry(params: { rethrowQueueError(err); } } + +/** Fold a queued entry into the entry directly above it. Throws QueueEntryNotFoundError if drained. */ +export async function mergeQueuedEntry(params: { + session_id: string; + entry_id: string; + user_id?: string; +}): Promise<{ entry_id: string }> { + const client = getWebSocketClient(); + if (!client) { + throw new Error(WS_CLIENT_UNAVAILABLE); + } + try { + return await client.request<{ entry_id: string }>("message.queue.merge", params); + } catch (err) { + rethrowQueueError(err); + } +} diff --git a/apps/web/src/locales/en/chat.json b/apps/web/src/locales/en/chat.json new file mode 100644 index 0000000000..c2f6ef707b --- /dev/null +++ b/apps/web/src/locales/en/chat.json @@ -0,0 +1,9 @@ +{ + "collapseMessage": "Collapse message", + "editQueuedMessage": "Edit queued message", + "expandMessage": "Expand message", + "failedToMergeQueuedMessages": "Failed to merge queued messages.", + "mergeReferenceOverflow": "Too many entity references to merge these queued messages.", + "mergeWithAbove": "Merge with above", + "removeQueuedMessage": "Remove queued message" +} diff --git a/apps/web/src/locales/pseudo/chat.json b/apps/web/src/locales/pseudo/chat.json new file mode 100644 index 0000000000..a01902db89 --- /dev/null +++ b/apps/web/src/locales/pseudo/chat.json @@ -0,0 +1,9 @@ +{ + "collapseMessage": "Ćōĺĺàƥśē ḿēśśàĝē", + "editQueuedMessage": "Ēďĩţ qũēũēď ḿēśśàĝē", + "expandMessage": "Ēxƥàńď ḿēśśàĝē", + "failedToMergeQueuedMessages": "Ƒàĩĺēď ţō ḿēŕĝē qũēũēď ḿēśśàĝēś.", + "mergeReferenceOverflow": "Ţōō ḿàńŷ ēńţĩţŷ ŕēƒēŕēńćēś ţō ḿēŕĝē ţĥēśē qũēũēď ḿēśśàĝēś.", + "mergeWithAbove": "Ḿēŕĝē ŵĩţĥ àƀōvē", + "removeQueuedMessage": "Ŕēḿōvē qũēũēď ḿēśśàĝē" +} diff --git a/docs/decisions/0051-pr-agent-notifications-extend-task-pr-automation.md b/docs/decisions/0051-pr-agent-notifications-extend-task-pr-automation.md index 8f8fd760ed..d69d5a8c90 100644 --- a/docs/decisions/0051-pr-agent-notifications-extend-task-pr-automation.md +++ b/docs/decisions/0051-pr-agent-notifications-extend-task-pr-automation.md @@ -184,3 +184,28 @@ error. 3. **Event-specific workflow-step transitions in the same change.** Deferred because today's task PR automation contract is boolean task-level options. Step selection needs its own workflow interaction and UI design. + +## Amendment: message-queue-merge agent carve-out + +**Date:** 2026-08-01 + +The `message.queue.merge` action (fold one queued message into the message above +it) is the single controlled exception to this ADR's rule that agent-, +workflow-, and server-owned entries are reserved for backend dispatch paths and +clients may only create, edit, and remove user-owned entries. + +MergeIntoAbove permits folding an agent-owned source into the agent-owned target +above it only when both rows carry the same non-empty `sender_task_id` metadata, +and the caller must have session access. This preserves the reserved row's +provenance — the merged entry keeps the target's identity, sender kind, and +queued-by of `agent` — so the operation consolidates additive one-way prompts +from a single inter-task agent without letting clients move, delete, or reorder +agent-owned rows outside that exact fold. The caller identity is not the gate +for agent merges (matching how the entry's ownership belongs to the sending +task, not the UI user); the identical-sender-task gate is. All other merge paths +remain strictly user-owned: a user-kind source merges only into a user-kind +target whose `queued_by` equals the caller's identity. + +Workflow- and server-owned rows remain entirely non-mergeable, and the reference- +overflow guard (a merge is rejected atomically when the combined entity +references exceed the per-message cap) applies to every sender kind. diff --git a/docs/plans/message-queue-merge/plan.md b/docs/plans/message-queue-merge/plan.md new file mode 100644 index 0000000000..c2e72f77dd --- /dev/null +++ b/docs/plans/message-queue-merge/plan.md @@ -0,0 +1,244 @@ +--- +spec: docs/specs/ui/message-queue-merge.md +created: 2026-07-31 +status: completed +--- + +# Implementation Plan: Merge Enqueued Messages Individually + +## Overview + +Add a per-entry **Merge with above** control to the queued-message panel. The +control folds a queued message into the message directly above it (same sender +kind only: user↔user, agent↔agent from the same sender task), concatenating +content, attachments, and entity references, then removing the folded entry. +The merge is implemented as a new atomic WebSocket action `message.queue.merge` +so a concurrent drain can never observe a half-merged queue. + +Order: repository contract and transaction first, then the service + WS handler +that expose it, then the frontend API/hook/components, then E2E. + +--- + +## Backend + +### Contract (`apps/backend/pkg/websocket/actions.go`) + +- Add `ActionMessageQueueMerge = "message.queue.merge"` beside the other + `ActionMessageQueue*` constants (line ~120). + +### Message-queue package (`apps/backend/internal/orchestrator/messagequeue/`) + +- `types.go` + - Add `MetadataSenderTaskID = "sender_task_id"` constant next to the other + `Metadata*` constants (reused by the merge sender-task equality check). + - Add sentinel error `ErrNoMergeTarget = errors.New("no mergeable message above")` + next to `ErrEntryNotFound` — signals the source exists but has no valid + target above it (head, different sender kind, agent sender-task mismatch, + or a reserved/in-flight target). +- `repository.go` + - Add to the `Repository` interface: + `MergeIntoAbove(ctx context.Context, sessionID, sourceID string, queuedBy string) (*QueuedMessage, error)`. +- `repository_sqlite.go` + - Implement `MergeIntoAbove` inside one transaction: + 1. SELECT the source row by `(id, session_id)` → `ErrEntryNotFound` when + missing. + 2. SELECT the target = the entry in the session with the greatest + `position` strictly below the source's `position` → `ErrNoMergeTarget` + when none (source is head). + 3. Kind/ownership gate (reject with `ErrNoMergeTarget` unless all hold): + - source `queued_by == QueuedByUser`: target `queued_by == queuedBy` + (caller, non-empty, non-reserved) — user merges into user. + - source `queued_by == QueuedByAgent`: target `queued_by == QueuedByAgent` + and both `metadata[sender_task_id]` values are equal — agent merges + into agent from the same sender task. + - otherwise (workflow/server/system, mismatched kinds, agent sender + mismatch) → `ErrNoMergeTarget`. + - a target that `IsReservedInFlight()` is not a valid target → + `ErrNoMergeTarget`. + 4. Merged fields: + - content = `joinMergeContent(target.Content, source.Content)` — join + with `"\n\n"`, but use `source.Content` verbatim when the target + content is empty (no leading blank line). + - attachments = `append(target.Attachments, source.Attachments...)`. + - metadata = copy of target metadata with `MetadataEntityReferences` + replaced by the union of both entries' reference lists, deduplicated + by canonical `ref` (`normalizeEntityReferences`-style dedupe — reuse + the existing `MetadataEntityReferences` key; see `types.go` and the + memory repo's `applyMetadataUpdates`). The deduplicated union must + not exceed the shared per-message cap + (`entityrefs.MaxReferencesPerMessage`): an over-cap union fails with + `ErrMergeReferenceOverflow` before either row is written, leaving + both entries unchanged (atomic rejection, sqlite and memory alike). + 5. `UPDATE` the target row (content/attachments/metadata, scoped by + `id` + `session_id`), then `DELETE` the source row (scoped the same), + then commit. Return the merged target entry. + - Keep the function ≤80 lines/≤50 statements; extract the kind gate and the + reference-union helper if needed. +- `repository_memory.go` + - Mirror the same logic under the repository mutex, returning the same errors + and merged semantics (used by service/handler tests and the memory repo + tests). +- `service.go` + - Add `MergeIntoAbove(ctx, sessionID, entryID, queuedBy string) (*QueuedMessage, error)` + delegating to `s.repo.MergeIntoAbove` with the same `s.logger.Info` log + pattern as `RemoveEntry`. + +### WS handler (`apps/backend/internal/orchestrator/handlers/queue_handlers.go`) + +- Extend the `QueueService` interface with `MergeIntoAbove`. +- Register `ws.ActionMessageQueueMerge` → `wsMergeIntoAbove` in + `RegisterHandlers`. +- New request type `wsMergeIntoAboveRequest{SessionID, EntryID, UserID}`. +- `wsMergeIntoAbove`: + - validate `session_id` / `entry_id` required (validation error), reject + reserved `user_id` via `IsReservedQueuedBy` + `reservedIdentityError` + (mirrors `wsUpdateMessage`), default empty `user_id` to `QueuedByUser`. + - call `queueService.MergeIntoAbove`; map `ErrEntryNotFound` → + `entry_not_found`, `ErrNoMergeTarget` → `ws.ErrorCodeValidation`, else + internal error. + - `publishStatus(ctx, sessionID)` on success; respond `{entry_id}`. +- Add a `queueErrorCodeEntryNotFound`-style mapping reuse — the existing + constants at the top of the file already cover both codes. + +## Frontend + +### API client (`apps/web/lib/api/domains/queue-api.ts`) + +- Add `mergeQueuedEntry(params: { session_id: string; entry_id: string }): Promise<{ entry_id: string }>` + sending `message.queue.merge` through `client.request`, wrapped in + `rethrowQueueError` like `removeQueuedEntry`. + +### Hook (`apps/web/hooks/domains/session/use-queue.ts`) + +- Add `mergeEntry(entryId: string)` action in `useQueueActions`: call + `mergeQueuedEntry({ session_id, entry_id })`, then `refetch(sessionId)`; on + `QueueEntryNotFoundError` refetch (drain race) and rethrow so the caller can + toast, matching the `editEntry` pattern. No optimistic store mutation (the + merged entry's content isn't computed client-side). Return it from `useQueue`. + +### Components (`apps/web/components/task/chat/`) + +- `queued-ghost-message.tsx` + - Export a mergeability helper: + `canMergeEntry(entry)` → sender kind is `"user"` or `"agent"`, plus — for + agent entries — a non-empty `metadata[sender_task_id]`; `canMergeWithAbove(entry, + above)` → both mergeable and same sender kind **and** equal non-empty + `sender_task_id` values for agent entries (mirrors the backend + `mergeAllowed` rule so the UI never offers a merge the backend will + reject). Reuse the existing `senderKindOf`. + - Add optional `canMerge: boolean` + `onMerge: () => void | Promise` + props to `QueuedGhostMessage`; pass through to `DisplayView`. + - Render a **Merge with above** ghost button (title `t("chat:mergeWithAbove")`, + `data-testid="queue-entry-merge"`, `IconArrowMerge` from + `@tabler/icons-react`, same styling as Edit/Remove) before the Edit + button, only when `canMerge` and the row is not in edit mode. +- `queued-ghost-list.tsx` + - In `QueuePanel`, compute each entry's `canMerge` as + `canMergeWithAbove(entry, entries[index - 1])` and pass + `onMerge={() => onMerge(entry.id)}`. + - Thread `onMerge` through `QueuePanelProps` and `useQueuePanelHandlers` (new + `handleMerge` → `mergeEntry`; on failure, `toast.error(t("chat:failedToMergeQueuedMessages"))` + or, for an over-cap reference union, `toast.error(t("chat:mergeReferenceOverflow"))`, + matching `handleRemove`). + - Pull `mergeEntry` from `useQueue(sessionId)` in `QueueAffordance`. + +## Tests + +- **Repository — user↔user happy path** (`repository_sqlite_merge_test.go`, + `service_test.go`): merge a user entry into the user entry above it → + combined content joined with `\n\n`, attachments concatenated, entity + references unioned/deduped, target identity (id/position/queued_by/queued_at) + preserved, source row gone, queue count drops by one. +- **Repository — empty target content** (`repository_sqlite_merge_test.go`): target + with empty content → merged content is exactly the source content (no leading + blank line). +- **Repository — head rejected** (`repository_sqlite_merge_test.go`): merging the + head entry → `ErrNoMergeTarget`, queue unchanged. +- **Repository — mismatched kinds rejected** (`repository_sqlite_merge_test.go`): + user source above agent target → `ErrNoMergeTarget`; agent source above user + target → `ErrNoMergeTarget`; workflow/system source → `ErrNoMergeTarget`. +- **Repository — agent↔agent** (`repository_sqlite_merge_test.go`): same + `sender_task_id` → allowed, merged entry keeps target's agent identity; + different `sender_task_id` → `ErrNoMergeTarget`. +- **Repository — ownership guard** (`repository_sqlite_merge_test.go`): user merge + with caller `queuedBy` not equal to both rows → `ErrNoMergeTarget`. +- **Repository — reference overflow rejected atomically** + (`repository_sqlite_merge_test.go`, `repository_postgres_merge_test.go`, + `service_test.go`): a deduplicated reference union over + `entityrefs.MaxReferencesPerMessage` → `ErrMergeReferenceOverflow`, both + rows' content/attachments/references unchanged, queue count unchanged + (sqlite, postgres, and in-memory repositories all assert the target is + untouched on the rejected path). +- **Repository — missing source** (`repository_sqlite_merge_test.go`): drained source + id → `ErrEntryNotFound`. +- **Repository — reserved in-flight target** (`repository_sqlite_merge_test.go`): a + durable lifecycle target with the reservation marker → `ErrNoMergeTarget`. +- **Repository — three-entry chain** (`repository_sqlite_merge_test.go`): merge `C` + into `B`, then `B` into `A` → one entry with `A+B+C` in order. +- **Service** (`service_test.go`): `MergeIntoAbove` delegates and returns the + merged entry (or the mapped error) through the memory repo. +- **WS handler** (`queue_handlers_merge_test.go`): happy path returns `{entry_id}` and + publishes status; missing `session_id` / `entry_id` → validation; head merge → + validation; reserved `user_id` rejected; drained source → `entry_not_found`. +- **Frontend hook** (`use-queue` via `queued-ghost-list.test.tsx` or a hook + test): `mergeEntry` calls the API and refetches; refetch on + `QueueEntryNotFoundError`. +- **Frontend gating** (`queued-ghost-message.test.tsx`): merge button hidden for + the first entry, hidden for mismatched sender kinds, hidden for + workflow/system entries, shown for a user entry behind a user entry and for + an agent entry behind an agent entry, and clicking it calls `onMerge`. +- **Frontend list wiring** (`queued-ghost-list.test.tsx`): `onMerge` passes the + clicked entry id to `mergeEntry`. + +## E2E Tests + +- **Scenario** (spec): two queued messages `A` and `B` with `B` behind `A`; user + clicks **Merge with above** on `B` → queue contains one message `A + B`, count + drops by one. + - **File:** `apps/web/e2e/tests/chat/message-queue.spec.ts` + - **What to verify:** queue two messages while the agent is busy, open the + panel, click `queue-entry-merge` on the second row, assert the panel now + shows a single entry whose text contains both messages and the chip/label + reflects one queued message. Agent-kind merging is covered by backend + unit + tests (inter-task dispatch is not practical in this E2E harness). + +## Implementation Waves And Parallel Candidates + +```text +Wave 1: +- [x] [task-01-backend-repository](task-01-backend-repository.md) + +Wave 2: +- [x] [task-02-backend-service-handler](task-02-backend-service-handler.md) + +Wave 3: +- [x] [task-03-frontend-merge-ui](task-03-frontend-merge-ui.md) + +Wave 4: +- [x] [task-04-e2e](task-04-e2e.md) +``` + +All four tasks execute sequentially in the primary conversation. The default is +sequential; `task-01` and `task-03` touch disjoint files and are parallel-safe +candidates **only with a frozen backend contract** — task-03 consumes the +`message.queue.merge` WS action and `mergeEntry` hook from task-02, so it must +not merge before task-02 lands. No subagents run unless the user explicitly +asks. + +## Verification commands + +All commands pass on the final head (`2b657e61d`, 2026-08-02); CI run +[#30742525141](https://github.com/kdlbs/kandev/actions/runs/30742525141) +(Frontend: lint, i18n checks + ratchet, tests, build — ✓) and +[#30742525164](https://github.com/kdlbs/kandev/actions/runs/30742525164) +(Backend: gofmt, golangci-lint, `go test -race` incl. Postgres — ✓), E2E +[#30742525129](https://github.com/kdlbs/kandev/actions/runs/30742525129) (all +14 shards + containers + desktop smoke — ✓). + +- `make fmt` — ✓ 2026-08-02 (gofmt clean on changed Go files) +- `make typecheck test lint` (from repo root per AGENTS.md) — ✓ via CI runs + above +- Targeted Go: `cd apps/backend && go test -race ./internal/orchestrator/messagequeue/... ./internal/orchestrator/handlers/...` — ✓ 2026-08-02 +- Targeted frontend: `cd apps && pnpm --filter @kandev/web test -- --run components/task/chat/queued-ghost-message.test.tsx components/task/chat/queued-ghost-list.test.tsx` — ✓ 2026-08-02 +- E2E: `cd apps/web && pnpm e2e:run --host tests/chat/message-queue.spec.ts --project=chromium` — ✓ 2026-08-02 (in E2E CI shards above) diff --git a/docs/plans/message-queue-merge/task-01-backend-repository.md b/docs/plans/message-queue-merge/task-01-backend-repository.md new file mode 100644 index 0000000000..c683aab11d --- /dev/null +++ b/docs/plans/message-queue-merge/task-01-backend-repository.md @@ -0,0 +1,58 @@ +--- +id: message-queue-merge-01 +title: Backend repository merge +status: completed +wave: 1 +depends_on: None +plan: plan.md +spec: ../../specs/ui/message-queue-merge.md +--- + +# Task 01: Backend repository merge + +## Acceptance + +1. `Repository.MergeIntoAbove(ctx, sessionID, sourceID, queuedBy)` is defined + in `internal/orchestrator/messagequeue/repository.go` and implemented in + both `repository_sqlite.go` (single transaction) and `repository_memory.go`, + with `ErrNoMergeTarget` and `MetadataSenderTaskID` added to `types.go`. +2. The same-sender-kind rules hold: user merges only into a user entry owned by + the caller; agent merges only into an agent entry with the identical + `sender_task_id`; everything else (head, workflow/server/system, mismatched + kind, in-flight reserved target) returns `ErrNoMergeTarget`; a drained + source returns `ErrEntryNotFound`. +3. Merged entry keeps the target's identity (id/position/queued_by/queued_at); + content joins with `\n\n` (no leading blank line when the target content is + empty), attachments concatenate, `entity_references` metadata is the + ref-deduplicated union; the source row is deleted; ordering of remaining + entries is preserved. + +## Verification + +- `cd apps/backend && go test -race ./internal/orchestrator/messagequeue/...` + +## Files + +- `apps/backend/internal/orchestrator/messagequeue/types.go` +- `apps/backend/internal/orchestrator/messagequeue/repository.go` +- `apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go` +- `apps/backend/internal/orchestrator/messagequeue/repository_memory.go` +- `apps/backend/internal/orchestrator/messagequeue/repository_sqlite_merge_test.go` +- `apps/backend/internal/orchestrator/messagequeue/service_test.go` + +## Inputs + +- Spec sections: `API surface`, `Data model`, `Permissions`, `Failure modes`, + `Scenarios` (all merge scenarios). +- Plan sections: `Backend → Message-queue package`, `Tests` (repository rows). +- Pattern to mirror: `UpdateContentAndMetadata` guard + tx in + `repository_sqlite.go:691`; `append`/`applyMetadataUpdates` helpers; the + memory repo's locking style. + +## Risks + +- Function size/complexity limits (≤80 lines, ≤50 statements, complexity ≤15) + — extract the kind gate and the reference-union into helpers. +- goconst: reuse a new `MetadataSenderTaskID` constant for `"sender_task_id"`. +- Reserved lifecycle rows: a reserved-in-flight target must be rejected, never + merged into. diff --git a/docs/plans/message-queue-merge/task-02-backend-service-handler.md b/docs/plans/message-queue-merge/task-02-backend-service-handler.md new file mode 100644 index 0000000000..f86b1f1af9 --- /dev/null +++ b/docs/plans/message-queue-merge/task-02-backend-service-handler.md @@ -0,0 +1,50 @@ +--- +id: message-queue-merge-02 +title: Backend service and WS handler +status: completed +wave: 2 +depends_on: ["message-queue-merge-01"] +plan: plan.md +spec: ../../specs/ui/message-queue-merge.md +--- + +# Task 02: Backend service and WS handler + +## Acceptance + +1. `ActionMessageQueueMerge = "message.queue.merge"` exists in + `pkg/websocket/actions.go` and `Service.MergeIntoAbove` delegates to the + repository (with the same logger pattern as `RemoveEntry`). +2. `QueueHandlers` registers `message.queue.merge`, validates `session_id` / + `entry_id`, rejects reserved `user_id`, defaults `user_id` to `QueuedByUser`, + and maps `ErrEntryNotFound` → `entry_not_found` / `ErrNoMergeTarget` → + validation, publishing `message.queue.status_changed` on success. +3. Handler tests cover the happy path, missing fields, head/mismatch → validation, + reserved `user_id` rejection, and drained source → `entry_not_found`. + +## Verification + +- `cd apps/backend && go test -race ./internal/orchestrator/handlers/... ./internal/orchestrator/messagequeue/...` + +## Files + +- `apps/backend/pkg/websocket/actions.go` +- `apps/backend/internal/orchestrator/messagequeue/service.go` +- `apps/backend/internal/orchestrator/handlers/queue_handlers.go` +- `apps/backend/internal/orchestrator/handlers/queue_handlers_merge_test.go` +- `apps/backend/internal/orchestrator/messagequeue/service_test.go` + +## Inputs + +- Spec sections: `API surface`, `Failure modes`, `Permissions`. +- Plan sections: `Backend → Contract`, `Backend → WS handler`, `Tests` (service + + handler rows). +- Patterns to mirror: `wsUpdateMessage` validation/error mapping + (`queue_handlers.go:243`), `wsRemoveEntry` shape (`queue_handlers.go:332`), + `publishStatus`. + +## Risks + +- Any other `QueueService` implementor in the codebase must gain the new + `MergeIntoAbove` method — grep for `QueueService` implementations (e.g. mocks + in tests) and update them. diff --git a/docs/plans/message-queue-merge/task-03-frontend-merge-ui.md b/docs/plans/message-queue-merge/task-03-frontend-merge-ui.md new file mode 100644 index 0000000000..4fdd9b6dc3 --- /dev/null +++ b/docs/plans/message-queue-merge/task-03-frontend-merge-ui.md @@ -0,0 +1,59 @@ +--- +id: message-queue-merge-03 +title: Frontend merge control +status: completed +wave: 3 +depends_on: ["message-queue-merge-02"] +plan: plan.md +spec: ../../specs/ui/message-queue-merge.md +--- + +# Task 03: Frontend merge control + +## Acceptance + +1. `mergeQueuedEntry` in `lib/api/domains/queue-api.ts` sends + `message.queue.merge` and rethrows queue errors; `useQueue` exposes a + `mergeEntry(entryId)` action that refetches after success and refetches on + `QueueEntryNotFoundError`. +2. The queue panel renders a **Merge with above** ghost button + (`data-testid="queue-entry-merge"`, title `Merge with above`, + `IconArrowMerge`) on an entry only when the entry and the one above it are + both mergeable and share the same sender kind (`user`↔`user`, `agent`↔ + `agent`); agent rows additionally require an identical non-empty + `sender_task_id` in metadata (mirroring the backend's `mergeAllowed`), and + the control is hidden for the first entry, mismatched kinds, and + workflow/system entries. +3. Unit tests cover button gating and the click → `mergeEntry` wiring. + +## Verification + +- `cd apps && pnpm --filter @kandev/web test -- --run components/task/chat/queued-ghost-message.test.tsx components/task/chat/queued-ghost-list.test.tsx` + +## Files + +- `apps/web/lib/api/domains/queue-api.ts` +- `apps/web/hooks/domains/session/use-queue.ts` +- `apps/web/components/task/chat/queued-ghost-message.tsx` +- `apps/web/components/task/chat/queued-ghost-list.tsx` +- `apps/web/components/task/chat/queued-ghost-message.test.tsx` +- `apps/web/components/task/chat/queued-ghost-list.test.tsx` + +## Inputs + +- Spec sections: `What`, `Permissions`, `Scenarios`. +- Plan sections: `Frontend`. +- Patterns to mirror: `removeEntry` in `use-queue.ts`, the Edit/Remove ghost + buttons in `queued-ghost-message.tsx`, `handleRemove`/`handleSave` in + `queued-ghost-list.tsx`, and the existing `queued-ghost-list.test.tsx` / + `queued-ghost-message.test.tsx` harnesses. +- Verify `IconArrowMerge` exists in the installed `@tabler/icons-react` + (^3.36.1) before importing; otherwise use `IconArrowsJoin` or the closest + merge glyph. + +## Risks + +- Frontend code-quality limits (≤100-line functions, complexity) — keep the + mergeability helpers tiny and pure. +- `senderKindOf` is currently private in `queued-ghost-message.tsx`; export the + merge helpers from the same module. diff --git a/docs/plans/message-queue-merge/task-04-e2e.md b/docs/plans/message-queue-merge/task-04-e2e.md new file mode 100644 index 0000000000..f493fb783e --- /dev/null +++ b/docs/plans/message-queue-merge/task-04-e2e.md @@ -0,0 +1,41 @@ +--- +id: message-queue-merge-04 +title: E2E merge flow +status: completed +wave: 4 +depends_on: ["message-queue-merge-03"] +plan: plan.md +spec: ../../specs/ui/message-queue-merge.md +--- + +# Task 04: E2E merge flow + +## Acceptance + +1. A Playwright test in the chat message-queue spec queues two messages while + the agent is busy, opens the queue panel, clicks **Merge with above** on the + second entry, and asserts the panel now shows a single entry whose text + contains both messages and the queue indicator reflects one queued message. + +## Verification + +- `cd apps/web && pnpm e2e:run --host tests/chat/message-queue.spec.ts` + +## Files + +- `apps/web/e2e/tests/chat/message-queue.spec.ts` + +## Inputs + +- Spec scenario: "two queued messages `A` and `B` with `B` behind `A` … one + message whose content is `A` + blank line + `B`". +- Plan section: `E2E Tests`. +- Pattern to mirror: the existing `seedTaskAndWaitForIdle` + `openQueuePanel` + helpers in `message-queue.spec.ts` and `typeWhileBusy`. + +## Risks + +- Agent-kind merging is not exercised in E2E (inter-task dispatch is not + practical in this harness) — backend + unit tests cover it. +- Timing: the first message must be slow enough (`/slow 10s`) that the second + stays queued; reuse the existing retry + timeout conventions of the suite. diff --git a/docs/public/websocket-api.md b/docs/public/websocket-api.md index c18f159949..c763c8076d 100644 --- a/docs/public/websocket-api.md +++ b/docs/public/websocket-api.md @@ -239,6 +239,8 @@ A successful response returns the normalized query and an ordered `groups` array `message.queue.add` accepts the same optional array. `message.queue.update` also accepts it and treats the array as a replacement: send `[]` after removing every generated reference link so stale metadata is cleared. Queue status and message responses return validated entries under `metadata.entity_references`. +`message.queue.merge` folds a queued entry into the entry directly above it. The payload requires `session_id` and `entry_id` (the source entry being merged away); `user_id` is optional and defaults to `user`. On success the response carries the surviving merged entry's `entry_id` (the target's id) and the server broadcasts an updated `message.queue.status_changed`. The request is rejected with `entry_not_found` when the entry was already drained or is not owned by the caller, with a validation error when no mergeable entry exists above it, and with `merge_reference_overflow` when the combined entity-reference lists would exceed the per-message cap (the merge is rejected atomically — neither row changes). + If the agent is busy, use the `message.queue.*` operations rather than retrying `message.add`. Permission prompts are represented in persisted/session message data; answer one with `permission.respond`. Its payload requires `session_id` and `pending_id`, plus `option_id` unless `cancelled:true`; optional `rejected:true` distinguishes an explicit denial from dismissing the prompt. For a raw diagnostic session, install a WebSocket client such as `websocat`, connect, then paste one request object per line: @@ -339,6 +341,7 @@ message.queue.append message.queue.cancel message.queue.drain message.queue.get +message.queue.merge message.queue.remove message.queue.update message.search diff --git a/docs/specs/INDEX.md b/docs/specs/INDEX.md index 8d4e8223bc..027f03b30e 100644 --- a/docs/specs/INDEX.md +++ b/docs/specs/INDEX.md @@ -138,6 +138,7 @@ Subscription quota tracking and per-agent cheap-model profile routing. | [slash-command-composer](ui/slash-command-composer.md) | shipped | | [entity-reference-composer](ui/entity-reference-composer.md) | draft | | [mermaid-rendering](ui/mermaid-rendering.md) | shipped | +| [message-queue-merge](ui/message-queue-merge.md) | shipped | | [settings-manual-save](ui/settings-manual-save.md) | shipped | | [transcript-navigation-settings](ui/transcript-navigation-settings.md) | shipped | | [app-status-bar](ui/app-status-bar.md) | shipped | diff --git a/docs/specs/ui/message-queue-merge.md b/docs/specs/ui/message-queue-merge.md new file mode 100644 index 0000000000..2a406cfd06 --- /dev/null +++ b/docs/specs/ui/message-queue-merge.md @@ -0,0 +1,196 @@ +--- +status: shipped +created: 2026-07-31 +owner: kandev +--- + +# Merge Enqueued Messages Individually + +## Why + +When a session is busy, users queue follow-up prompts that drain one-per-turn. A +burst of small, related prompts (a correction, a follow-up question, extra +context) is dispatched as several separate turns, which wastes agent turns and +fragments the conversation. Users want to collapse queued messages they no +longer want delivered separately — folding one queued message into the one +above it — before the agent picks them up. + +## What + +- A queued message that has another queued message above it (i.e. it is not the + head of the queue) offers a **Merge with above** control in the queue panel, + next to the existing Edit and Remove controls. +- Clicking **Merge with above** folds the message into the message directly + above it: the above message keeps its identity (id, position, sender, + queued-at time) and its content is concatenated with the folded message's + content, separated by a blank line. +- The folded message is removed from the queue; the queue count decreases by + one and remaining entries keep their FIFO order. +- Attachments from both messages are carried over to the merged message. +- Entity references from both messages are carried over to the merged message, + deduplicated by canonical reference. +- Merging is **rejected atomically** when the deduplicated reference union + would exceed the per-message reference cap (100, mirroring + `entityrefs.maxReferencesPerMessage`). Nothing is truncated: both rows keep + the references they already persisted, and the merge reports + `merge_reference_overflow` instead of silently dropping source references at + dispatch time. +- The merge is atomic and race-safe: a concurrent queue drain can either win + entirely (both rows untouched) or lose entirely (both rows updated); it can + never interleave between updating the above message and removing the folded + one. +- A message may only be merged into a message of the **same sender kind** — + mismatching kinds never merge. A user-typed message merges only into another + user-typed message; an agent (inter-task) message merges only into another + agent message. Workflow and system messages are never a merge source or + target. +- Two agent messages merge only when they come from the **same sender task** + (identical `sender_task_id` in metadata), so the merged message never loses + or conflates message provenance. The merged message keeps the target's + sender identity. +- The head message (first in the queue) has no merge control: there is no + message above it. +- The merge button is hidden entirely when only one message is queued, or when + the message above it has a different sender kind (or is not mergeable at + all). + +## API surface + +New WebSocket action `message.queue.merge` (registered alongside the existing +`message.queue.update` / `message.queue.remove` actions in +`internal/orchestrator/handlers/queue_handlers.go`). + +Request payload: + +``` +session_id string required +entry_id string required — the message to fold into the one above it +user_id string optional — defaults to "user"; reserved identities rejected +``` + +Response payload: + +``` +entry_id string the id of the surviving (above) message +``` + +Errors: + +- `entry_not_found` — the folded message was already drained or does not + exist, matching the `message.queue.update` / `message.queue.remove` + semantics. +- `validation` — the folded message is the head of the queue (nothing above + it), the message above it is not a valid merge target (different sender kind, + an agent message from a different sender task), the folded message is a + non-mergeable kind (workflow/system), or `session_id` / `entry_id` is + missing. +- `merge_reference_overflow` — the combined entity references exceed the + per-message cap; both rows are left untouched. + +After a successful merge the backend publishes the standard +`message.queue.status_changed` event so all connected clients refresh their +queue view, exactly like the other queue mutations. + +## Data model + +No schema change. The merge reuses the existing `queued_messages` table: + +- The above entry's `content`, `attachments_json`, and `metadata_json` are + rewritten in place (identity, `position`, `queued_by`, `queued_at`, `model`, + `plan_mode`, `task_id` unchanged). +- The folded entry's row is deleted (same rule as `DeleteByID`: rows with + `queued_by` in the reserved set are never deleted by this operation). +- Positions are not renumbered; gaps are allowed and ordering stays correct + because the queue orders by `(session_id, position)`. + +## Permissions + +- The folded (source) entry and the above (target) entry must have the **same + sender kind** — user merges into user, agent merges into agent. +- For a user-kind merge, both entries must have `queued_by` equal to the + caller's identity and must not be in the reserved set (`agent`, `workflow`, + `server`), mirroring the existing `UpdateContentAndMetadata` guard. +- For an agent-kind merge, both entries must have `queued_by` equal to `agent` + and the source's `sender_task_id` metadata must match the target's; the merge + is otherwise rejected because the merged entry keeps the target's sender + identity. This is a deliberate carve-out of the ADR 0051 boundary that + reserves agent-owned rows for backend mutation: consolidating additive + one-way prompts from a single inter-task agent preserves the reserved row's + provenance (the merged entry keeps the target's identity and the source's + `sender_task_id`-matched lineage) while reducing deliveries. The gate is + strict — identical non-empty `sender_task_id`, never caller-tunable — so a + client still cannot move, delete, or reorder agent-owned rows outside that + exact fold. +- Workflow and system messages are never a merge source or target. +- The WS handler rejects client-supplied reserved `user_id` values, exactly like + the update handler. + +## Failure modes + +- **Drain race:** if the agent drains the folded message (or the above message) + between the UI rendering and the merge request, the backend returns + `entry_not_found` and the frontend refetches the queue to resync, matching the + existing edit/remove resync path. +- **Drain interleave:** the update-above and delete-source writes run inside one + transaction (serialized per session by the repository's per-session lock and + guarded by affected-row counts), so a concurrent drain can never observe a + half-merged queue. +- **Reference overflow:** a merge whose deduplicated reference union exceeds the + per-message cap is rejected atomically (`merge_reference_overflow`); the + frontend surfaces a focused toast and the queue is unchanged. +- **Merge failure:** the frontend shows the existing toast error pattern and + refetches; no partial state is left behind. + +## Scenarios + +- **GIVEN** two queued messages `A` and `B` with `B` behind `A`, **WHEN** the + user clicks **Merge with above** on `B`, **THEN** the queue contains one + message whose content is `A` + blank line + `B`, the queue count drops by + one, and both messages' attachments and entity references are present. +- **GIVEN** a single queued message, **WHEN** the user opens the queue panel, + **THEN** no **Merge with above** control is shown on it. +- **GIVEN** a queued message at the head of the queue, **WHEN** the merge + action targets it, **THEN** the backend rejects it with a validation error + and the queue is unchanged. +- **GIVEN** two consecutive user messages `A` and `B` with `B` behind `A`, + **WHEN** the user clicks **Merge with above** on `B`, **THEN** the queue + contains one user message whose content is `A` + blank line + `B`, and the + merged entry keeps `A`'s identity. +- **GIVEN** two consecutive agent messages from the same sender task (same + `sender_task_id`), **WHEN** the user clicks **Merge with above** on the later + one, **THEN** the queue contains one agent message with the combined content + that keeps the earlier entry's sender identity. +- **GIVEN** two consecutive agent messages from different sender tasks, + **WHEN** the merge action targets the later one, **THEN** the backend rejects + it and the queue is unchanged. +- **GIVEN** a user message whose above message is agent-owned, **WHEN** the + user opens the queue panel, **THEN** no **Merge with above** control is shown + on that message, and a direct merge request is rejected with a validation + error. +- **GIVEN** a message queued by an agent (inter-task) whose above message is a + user message, **WHEN** the user opens the queue panel, **THEN** no **Merge + with above** control is shown on that message. +- **GIVEN** a merge request for a message that was just drained, **WHEN** the + request arrives, **THEN** the backend returns `entry_not_found` and the + frontend refetches the queue. +- **GIVEN** three queued messages `A`, `B`, `C`, **WHEN** the user merges `C` + into `B` and then merges the resulting message into `A`, **THEN** the queue + contains a single message with all three contents concatenated in order. +- **GIVEN** an above message with empty content (attachments only), **WHEN** + the user merges a message into it, **THEN** the merged content is exactly the + folded message's content (no leading blank line) and attachments combine. +- **GIVEN** two messages whose deduplicated entity-reference union exceeds 100, + **WHEN** the user clicks **Merge with above** on the later one, **THEN** the + backend returns `merge_reference_overflow`, both rows keep their references, + and the frontend shows a focused error toast. + +## Out of scope + +- Merging delivered (already sent) messages — this operates only on the pending + queue. +- Reordering queued messages (e.g. drag-to-move); only merge-into-above is + offered. +- Merging messages of mismatching sender kinds (e.g. a user message into an + agent message), or agent messages from different sender tasks. +- Merging workflow or system (server-owned) messages. +- A preview of the merged result before applying.