From bd9505499792b1279e6e38acd04827b9b3357fcd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20Fiere?= Date: Sun, 9 Aug 2026 13:00:21 +0200 Subject: [PATCH] feat: reorder queued messages by drag and drop (#2420) Co-authored-by: Carlos Florencio --- .../orchestrator/handlers/queue_handlers.go | 95 ++++- .../handlers/queue_handlers_reorder_test.go | 131 +++++++ .../handlers/queue_handlers_test.go | 3 + .../orchestrator/messagequeue/reorder.go | 64 ++++ .../orchestrator/messagequeue/repository.go | 12 + .../messagequeue/repository_memory.go | 75 ++++ .../messagequeue/repository_reorder_test.go | 283 +++++++++++++++ .../messagequeue/repository_sqlite.go | 142 ++++++++ .../orchestrator/messagequeue/service.go | 27 ++ .../orchestrator/messagequeue/types.go | 5 + apps/backend/pkg/websocket/actions.go | 1 + .../task/chat/queued-ghost-list.test.tsx | 157 ++++++++ .../task/chat/queued-ghost-list.tsx | 144 +++++++- .../task/chat/queued-ghost-message.test.tsx | 96 +++++ .../task/chat/queued-ghost-message.tsx | 149 +++++++- .../tests/chat/message-queue-reorder.spec.ts | 192 ++++++++++ .../chat/mobile-message-queue-reorder.spec.ts | 166 +++++++++ .../hooks/domains/session/use-queue.test.ts | 94 +++++ apps/web/hooks/domains/session/use-queue.ts | 164 +++++++-- apps/web/lib/api/domains/queue-api.test.ts | 40 +++ apps/web/lib/api/domains/queue-api.ts | 39 ++ apps/web/src/locales/en/chat.json | 3 + apps/web/src/locales/pseudo/chat.json | 3 + docs/plans/message-queue-reorder/plan.md | 339 ++++++++++++++++++ .../task-01-backend-repository-reorder.md | 56 +++ ...task-02-backend-service-handler-reorder.md | 57 +++ .../task-03-frontend-api-hook-reorder.md | 62 ++++ .../task-04-frontend-dnd-ui.md | 72 ++++ .../task-05-e2e-reorder.md | 72 ++++ .../task-06-docs-out-of-scope-updates.md | 63 ++++ docs/specs/INDEX.md | 7 +- docs/specs/ui/message-queue-management.md | 3 +- docs/specs/ui/message-queue-reorder.md | 210 +++++++++++ docs/specs/ui/message-queue-send-now.md | 4 +- 34 files changed, 2972 insertions(+), 58 deletions(-) create mode 100644 apps/backend/internal/orchestrator/handlers/queue_handlers_reorder_test.go create mode 100644 apps/backend/internal/orchestrator/messagequeue/reorder.go create mode 100644 apps/backend/internal/orchestrator/messagequeue/repository_reorder_test.go create mode 100644 apps/web/e2e/tests/chat/message-queue-reorder.spec.ts create mode 100644 apps/web/e2e/tests/chat/mobile-message-queue-reorder.spec.ts create mode 100644 docs/plans/message-queue-reorder/plan.md create mode 100644 docs/plans/message-queue-reorder/task-01-backend-repository-reorder.md create mode 100644 docs/plans/message-queue-reorder/task-02-backend-service-handler-reorder.md create mode 100644 docs/plans/message-queue-reorder/task-03-frontend-api-hook-reorder.md create mode 100644 docs/plans/message-queue-reorder/task-04-frontend-dnd-ui.md create mode 100644 docs/plans/message-queue-reorder/task-05-e2e-reorder.md create mode 100644 docs/plans/message-queue-reorder/task-06-docs-out-of-scope-updates.md create mode 100644 docs/specs/ui/message-queue-reorder.md diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers.go b/apps/backend/internal/orchestrator/handlers/queue_handlers.go index 0a762ad4da..88a9307b54 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers.go +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers.go @@ -20,11 +20,15 @@ import ( const ( // queueErrorCodeEntryNotFound is surfaced when an edit/remove targets an entry // that has already been drained (atomic-take won the race). - queueErrorCodeEntryNotFound = "entry_not_found" - queueErrorCodeSessionBusy = "session_busy" - queueErrorCodeNotPromptable = "session_not_promptable" - queueErrorCodeSendNowQueueEmpty = "queue_empty" - queueErrorCodeSendNowQueueChanged = "queue_changed" + queueErrorCodeEntryNotFound = "entry_not_found" + queueErrorCodeSessionBusy = "session_busy" + queueErrorCodeNotPromptable = "session_not_promptable" + queueErrorCodeSendNowQueueEmpty = "queue_empty" + queueErrorCodeSendNowQueueChanged = "queue_changed" + // queueErrorCodeQueueChanged is the shared "your snapshot is stale" signal + // for reorder drift; the wire value matches the send-now code so clients + // reconcile with one handler. + queueErrorCodeQueueChanged = "queue_changed" queueErrorCodeSendNowConflict = "send_now_conflict" queueErrorCodeSendNowTurnChanged = "turn_changed" queueErrorCodeSendNowAttachmentOverflow = "send_now_attachment_overflow" @@ -55,6 +59,7 @@ type QueueService interface { 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) + ReorderEntries(ctx context.Context, sessionID string, orderedIDs []string) error CancelAll(ctx context.Context, sessionID string) (int, error) GetStatus(ctx context.Context, sessionID string) *messagequeue.QueueStatus } @@ -158,6 +163,7 @@ func (h *QueueHandlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionMessageQueueSendNow, h.wsSendNow) d.RegisterFunc(ws.ActionMessageQueueRemove, h.wsRemoveEntry) d.RegisterFunc(ws.ActionMessageQueueMerge, h.wsMergeIntoAbove) + d.RegisterFunc(ws.ActionMessageQueueReorder, h.wsReorder) } type wsQueueMessageRequest struct { @@ -172,6 +178,7 @@ type wsQueueMessageRequest struct { UserID string `json:"user_id,omitempty"` } +// wsQueueMessage handles ActionMessageQueueAdd, appending a new entry to the session queue. func (h *QueueHandlers) wsQueueMessage(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsQueueMessageRequest if err := msg.ParsePayload(&req); err != nil { @@ -248,6 +255,7 @@ type wsCancelAllRequest struct { SessionID string `json:"session_id"` } +// wsCancelAll handles ActionMessageQueueCancel, clearing every pending entry for a session. func (h *QueueHandlers) wsCancelAll(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsCancelAllRequest if err := msg.ParsePayload(&req); err != nil { @@ -277,6 +285,7 @@ type wsDrainQueueRequest struct { SessionID string `json:"session_id"` } +// wsDrainQueue handles ActionMessageQueueDrain, dispatching one queued entry when the session is promptable. func (h *QueueHandlers) wsDrainQueue(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsDrainQueueRequest if err := msg.ParsePayload(&req); err != nil { @@ -318,6 +327,7 @@ type wsSendNowRequest struct { EntryID string `json:"entry_id,omitempty"` } +// wsSendNow handles ActionMessageQueueSendNow, interrupting the active turn with an exact queue selection. func (h *QueueHandlers) wsSendNow(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsSendNowRequest if err := msg.ParsePayload(&req); err != nil { @@ -349,6 +359,7 @@ func (h *QueueHandlers) wsSendNow(ctx context.Context, msg *ws.Message) (*ws.Mes }) } +// validateSendNowRequest validates the send-now scope and entry id combination. func validateSendNowRequest(req wsSendNowRequest) string { switch { case req.Scope != orchestrator.QueueSendNowScopeEntry && req.Scope != orchestrator.QueueSendNowScopeAll: @@ -362,6 +373,7 @@ func validateSendNowRequest(req wsSendNowRequest) string { } } +// sendNowErrorResponse maps send-now failures to their stable websocket error codes. func (h *QueueHandlers) sendNowErrorResponse(msg *ws.Message, sessionID string, err error) (*ws.Message, error) { switch { case errors.Is(err, orchestrator.ErrSendNowEntryNotFound): @@ -390,6 +402,7 @@ type wsGetQueueStatusRequest struct { SessionID string `json:"session_id"` } +// wsGetQueueStatus handles ActionMessageQueueGet, returning the pending list and capacity. func (h *QueueHandlers) wsGetQueueStatus(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsGetQueueStatusRequest if err := msg.ParsePayload(&req); err != nil { @@ -416,6 +429,7 @@ type wsUpdateMessageRequest struct { UserID string `json:"user_id,omitempty"` } +// wsUpdateMessage handles ActionMessageQueueUpdate, replacing a queued entry's content. func (h *QueueHandlers) wsUpdateMessage(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsUpdateMessageRequest if err := msg.ParsePayload(&req); err != nil { @@ -513,6 +527,7 @@ func (h *QueueHandlers) wsUpdateMessage(ctx context.Context, msg *ws.Message) (* return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{fieldEntryID: req.EntryID}) } +// newlyAddedQueueAttachments returns attachment descriptors that were not present before. func newlyAddedQueueAttachments(previous, replacement []messagequeue.MessageAttachment) []messagequeue.MessageAttachment { retained := make(map[string]struct{}, len(previous)) for _, attachment := range previous { @@ -532,6 +547,7 @@ func newlyAddedQueueAttachments(previous, replacement []messagequeue.MessageAtta return newlyAdded } +// supersededQueueAttachments returns attachment descriptors dropped by the replacement. func supersededQueueAttachments(previous, replacement []messagequeue.MessageAttachment) []messagequeue.MessageAttachment { retained := make(map[string]struct{}, len(replacement)) for _, attachment := range replacement { @@ -551,6 +567,7 @@ func supersededQueueAttachments(previous, replacement []messagequeue.MessageAtta return superseded } +// validateSubmittedReferences runs the entity-reference submission validator when configured. func (h *QueueHandlers) validateSubmittedReferences( ctx context.Context, sessionID, taskID string, @@ -565,6 +582,7 @@ func (h *QueueHandlers) validateSubmittedReferences( return h.referenceValidator.ValidateForSubmission(ctx, sessionID, taskID, references) } +// rollbackQueuedAttachmentClaim releases attachments claimed for a queue entry that failed to persist. func (h *QueueHandlers) rollbackQueuedAttachmentClaim(ctx context.Context, sessionID, entryID string) error { if err := h.queueService.RemoveEntry(ctx, sessionID, entryID); err == nil { return nil @@ -580,6 +598,7 @@ func (h *QueueHandlers) rollbackQueuedAttachmentClaim(ctx context.Context, sessi return err } +// firstInvalidDeliveryMode returns the index of the first attachment with an unknown delivery mode. func firstInvalidDeliveryMode(attachments []messagequeue.MessageAttachment) int { for i, att := range attachments { if att.DeliveryMode != "" && att.DeliveryMode != "prompt" && att.DeliveryMode != "path" { @@ -589,6 +608,7 @@ func firstInvalidDeliveryMode(attachments []messagequeue.MessageAttachment) int return -1 } +// firstInvalidAttachment returns the index of the first structurally invalid attachment. func firstInvalidAttachment(attachments []messagequeue.MessageAttachment) int { if len(attachments) > models.MaxMessageAttachmentCount { return models.MaxMessageAttachmentCount @@ -610,6 +630,7 @@ func firstInvalidAttachment(attachments []messagequeue.MessageAttachment) int { return -1 } +// attachmentPayloadBytes returns the decoded payload size and whether the base64 is valid. func attachmentPayloadBytes(attachment messagequeue.MessageAttachment) (int64, bool) { if attachment.AttachmentID != "" { if attachment.Data != "" || attachment.Name == "" || attachment.MimeType == "" { @@ -630,6 +651,7 @@ func attachmentPayloadBytes(attachment messagequeue.MessageAttachment) (int64, b return int64(len(decoded)), true } +// queueAttachmentsToV1 converts queue attachments to the API v1 representation. func queueAttachmentsToV1(attachments []messagequeue.MessageAttachment) []v1.MessageAttachment { if len(attachments) == 0 { return nil @@ -654,6 +676,7 @@ type wsRemoveEntryRequest struct { EntryID string `json:"entry_id"` } +// wsRemoveEntry handles ActionMessageQueueRemove, deleting a single queued entry. func (h *QueueHandlers) wsRemoveEntry(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsRemoveEntryRequest if err := msg.ParsePayload(&req); err != nil { @@ -740,6 +763,51 @@ func (h *QueueHandlers) wsMergeIntoAbove(ctx context.Context, msg *ws.Message) ( return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{fieldEntryID: merged.ID}) } +// wsReorderRequest is the payload for ActionMessageQueueReorder: the session +// whose queue is modified and the complete ordered list of visible pending +// entry ids the caller wants as the new FIFO order. +type wsReorderRequest struct { + SessionID string `json:"session_id"` + OrderedIDs []string `json:"ordered_ids"` +} + +// wsReorder handles ActionMessageQueueReorder, rewriting the session's visible +// pending order to match ordered_ids and broadcasting the updated queue. Any +// drift from the persisted visible set (a drain/remove/merge raced the drag) +// is rejected atomically with queue_changed so the client refetches. +func (h *QueueHandlers) wsReorder(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsReorderRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.SessionID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil) + } + if denied := h.authorizeSession(ctx, msg, req.SessionID); denied != nil { + return denied, nil + } + if len(req.OrderedIDs) == 0 { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "ordered_ids is required", nil) + } + if hasDuplicateIDs(req.OrderedIDs) { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "ordered_ids must not contain duplicates", nil) + } + + if err := h.queueService.ReorderEntries(ctx, req.SessionID, req.OrderedIDs); err != nil { + if errors.Is(err, messagequeue.ErrQueueChanged) { + return ws.NewError(msg.ID, msg.Action, queueErrorCodeQueueChanged, "Queue changed before the reorder could be applied", nil) + } + h.logger.Error("failed to reorder queued messages", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to reorder queued messages", nil) + } + + h.publishStatus(ctx, req.SessionID) + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + fieldSessionID: req.SessionID, + "reordered": len(req.OrderedIDs), + }) +} + type wsAppendToQueueRequest struct { SessionID string `json:"session_id"` TaskID string `json:"task_id"` @@ -749,6 +817,7 @@ type wsAppendToQueueRequest struct { UserID string `json:"user_id,omitempty"` } +// wsAppendToQueue handles ActionMessageQueueAppend, appending or inserting a user message. func (h *QueueHandlers) wsAppendToQueue(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsAppendToQueueRequest if err := msg.ParsePayload(&req); err != nil { @@ -796,6 +865,19 @@ func (h *QueueHandlers) wsAppendToQueue(ctx context.Context, msg *ws.Message) (* }) } +// hasDuplicateIDs reports whether ids contains any id more than once. +func hasDuplicateIDs(ids []string) bool { + seen := make(map[string]struct{}, len(ids)) + for _, id := range ids { + if _, duplicate := seen[id]; duplicate { + return true + } + seen[id] = struct{}{} + } + return false +} + +// reservedIdentityError builds the validation message for reserved caller identities. func reservedIdentityError(queuedBy string) string { if queuedBy == messagequeue.QueuedByAgent { return "user_id may not impersonate the agent identity" @@ -803,6 +885,7 @@ func reservedIdentityError(queuedBy string) string { return "user_id may not impersonate a reserved identity" } +// authorizeSession denies the request when the caller cannot access the session. func (h *QueueHandlers) authorizeSession(ctx context.Context, msg *ws.Message, sessionID string) *ws.Message { if h.accessAuthorizer == nil { return queueAccessDeniedResponse(msg) @@ -813,6 +896,7 @@ func (h *QueueHandlers) authorizeSession(ctx context.Context, msg *ws.Message, s return nil } +// authorizeTaskSession denies the request when the caller cannot access the task/session pair. func (h *QueueHandlers) authorizeTaskSession( ctx context.Context, msg *ws.Message, @@ -827,6 +911,7 @@ func (h *QueueHandlers) authorizeTaskSession( return nil } +// queueAccessDeniedResponse builds the non-enumerating session-not-found error response. func queueAccessDeniedResponse(msg *ws.Message) *ws.Message { response, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, queueAccessDenied, nil) return response diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers_reorder_test.go b/apps/backend/internal/orchestrator/handlers/queue_handlers_reorder_test.go new file mode 100644 index 0000000000..e5f6a898fd --- /dev/null +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers_reorder_test.go @@ -0,0 +1,131 @@ +package handlers + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + ws "github.com/kandev/kandev/pkg/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestWsReorder covers the ActionMessageQueueReorder websocket path: happy +// path (publish + reversed order), validation, access denial, and drift +// mapping to queue_changed. +func TestWsReorder(t *testing.T) { + t.Run("reorders the visible queue and publishes status", func(t *testing.T) { + handlers, svc := setupQueueHandlers(t) + ctx := context.Background() + first, err := svc.QueueMessage(ctx, "s", "t", "first", "", messagequeue.QueuedByUser, false, nil) + require.NoError(t, err) + second, err := svc.QueueMessage(ctx, "s", "t", "second", "", messagequeue.QueuedByUser, false, nil) + require.NoError(t, err) + + events := &mockEventBus{} + handlers.eventBus = events + + response, err := handlers.wsReorder(ctx, createTestMessage(t, ws.ActionMessageQueueReorder, map[string]interface{}{ + "session_id": "s", + "ordered_ids": []string{second.ID, first.ID}, + })) + 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, "s", payload["session_id"]) + assert.Equal(t, float64(2), payload["reordered"]) + + status := svc.GetStatus(ctx, "s") + require.Len(t, status.Entries, 2) + assert.Equal(t, second.ID, status.Entries[0].ID) + assert.Equal(t, first.ID, status.Entries[1].ID) + assert.Equal(t, 1, events.published) + }) + + t.Run("rejects empty and duplicate ordered_ids", func(t *testing.T) { + handlers, svc := setupQueueHandlers(t) + ctx := context.Background() + first, err := svc.QueueMessage(ctx, "s", "t", "first", "", messagequeue.QueuedByUser, false, nil) + require.NoError(t, err) + + events := &mockEventBus{} + handlers.eventBus = events + + empty := createTestMessage(t, ws.ActionMessageQueueReorder, map[string]interface{}{ + "session_id": "s", + "ordered_ids": []string{}, + }) + response, err := handlers.wsReorder(ctx, empty) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Equal(t, ws.ErrorCodeValidation, parseError(t, response).Code) + assert.Contains(t, parseError(t, response).Message, "ordered_ids is required") + + duplicate := createTestMessage(t, ws.ActionMessageQueueReorder, map[string]interface{}{ + "session_id": "s", + "ordered_ids": []string{first.ID, first.ID}, + }) + response, err = handlers.wsReorder(ctx, duplicate) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Equal(t, ws.ErrorCodeValidation, parseError(t, response).Code) + assert.Contains(t, parseError(t, response).Message, "duplicates") + + assert.Equal(t, 0, events.published) + assert.Equal(t, 1, svc.GetStatus(ctx, "s").Count) + }) + + t.Run("rejects drift with queue_changed and no mutation", func(t *testing.T) { + handlers, svc := setupQueueHandlers(t) + ctx := context.Background() + first, err := svc.QueueMessage(ctx, "s", "t", "first", "", messagequeue.QueuedByUser, false, nil) + require.NoError(t, err) + second, err := svc.QueueMessage(ctx, "s", "t", "second", "", messagequeue.QueuedByUser, false, nil) + require.NoError(t, err) + // A drain wins between the client's snapshot and the reorder. + _, ok := svc.TakeQueued(ctx, "s") + require.True(t, ok) + + events := &mockEventBus{} + handlers.eventBus = events + + response, err := handlers.wsReorder(ctx, createTestMessage(t, ws.ActionMessageQueueReorder, map[string]interface{}{ + "session_id": "s", + "ordered_ids": []string{first.ID, second.ID}, + })) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Equal(t, queueErrorCodeQueueChanged, parseError(t, response).Code) + assert.Equal(t, 0, events.published) + status := svc.GetStatus(ctx, "s") + require.Len(t, status.Entries, 1) + assert.Equal(t, second.ID, status.Entries[0].ID) + }) + + t.Run("denies when the session is not authorized", func(t *testing.T) { + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console", OutputPath: "stderr"}) + require.NoError(t, err) + svc := messagequeue.NewServiceMemory(log) + first, err := svc.QueueMessage(context.Background(), "s", "t", "first", "", messagequeue.QueuedByUser, false, nil) + require.NoError(t, err) + access := &fakeQueueAccessAuthorizer{sessionErr: errors.New("secret denial")} + events := &mockEventBus{} + handlers := NewQueueHandlers(svc, events, log, nil, access, nil) + + response, err := handlers.wsReorder(context.Background(), createTestMessage(t, ws.ActionMessageQueueReorder, map[string]interface{}{ + "session_id": "s", + "ordered_ids": []string{first.ID}, + })) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + assert.Equal(t, ws.ErrorCodeNotFound, parseError(t, response).Code) + assert.NotContains(t, string(response.Payload), "secret denial") + assert.Equal(t, 1, access.sessionCalls) + assert.Equal(t, 0, events.published) + }) +} diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go b/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go index a1fe1de187..115e0f7e78 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go @@ -190,6 +190,9 @@ func TestQueueHandlersDenyUnauthorizedSessionActions(t *testing.T) { {name: "merge", action: ws.ActionMessageQueueMerge, call: (*QueueHandlers).wsMergeIntoAbove, body: func(id string) map[string]interface{} { return map[string]interface{}{"session_id": "s", "entry_id": id} }}, + {name: "reorder", action: ws.ActionMessageQueueReorder, call: (*QueueHandlers).wsReorder, body: func(id string) map[string]interface{} { + return map[string]interface{}{"session_id": "s", "ordered_ids": []string{id, "q-other"}} + }}, } for _, tc := range tests { diff --git a/apps/backend/internal/orchestrator/messagequeue/reorder.go b/apps/backend/internal/orchestrator/messagequeue/reorder.go new file mode 100644 index 0000000000..8533035231 --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/reorder.go @@ -0,0 +1,64 @@ +package messagequeue + +// validateReorderInput rejects empty or duplicate reorder submissions before +// they reach the repository. Duplicate ids are a client bug but map to the +// same ErrQueueChanged drift signal so direct callers reconcile from the +// authoritative queue (the WS handler reports duplicates as a validation +// error before this check runs). +func validateReorderInput(orderedIDs []string) error { + if len(orderedIDs) == 0 { + return ErrQueueChanged + } + seen := make(map[string]struct{}, len(orderedIDs)) + for _, id := range orderedIDs { + if _, duplicate := seen[id]; duplicate { + return ErrQueueChanged + } + seen[id] = struct{}{} + } + return nil +} + +// splitVisibleAndReserved partitions a position-ordered row list into visible +// pending entries and reserved in-flight lifecycle rows. Reordering validates +// against the visible set exactly as GetStatus presents it; reserved rows are +// hidden from clients and never reordered. +func splitVisibleAndReserved(stored []*QueuedMessage) (visible, reserved []*QueuedMessage) { + visible = make([]*QueuedMessage, 0, len(stored)) + for _, msg := range stored { + if msg.IsReservedInFlight() { + reserved = append(reserved, msg) + } else { + visible = append(visible, msg) + } + } + return visible, reserved +} + +// validateReorderSet checks that orderedIDs is an exact permutation of the +// visible pending entry ids and returns the entries in the submitted order. +// Any drift — a length mismatch, a missing or unknown id, or a duplicate — +// returns ErrQueueChanged so callers reject the reorder atomically. +func validateReorderSet(visible []*QueuedMessage, orderedIDs []string) ([]*QueuedMessage, error) { + if len(orderedIDs) != len(visible) { + return nil, ErrQueueChanged + } + byID := make(map[string]*QueuedMessage, len(visible)) + for _, msg := range visible { + byID[msg.ID] = msg + } + ordered := make([]*QueuedMessage, 0, len(visible)) + seen := make(map[string]struct{}, len(orderedIDs)) + for _, id := range orderedIDs { + if _, duplicate := seen[id]; duplicate { + return nil, ErrQueueChanged + } + seen[id] = struct{}{} + msg, ok := byID[id] + if !ok { + return nil, ErrQueueChanged + } + ordered = append(ordered, msg) + } + return ordered, nil +} diff --git a/apps/backend/internal/orchestrator/messagequeue/repository.go b/apps/backend/internal/orchestrator/messagequeue/repository.go index 057419420f..de618a8e3d 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository.go @@ -116,6 +116,17 @@ type Repository interface { // missing source returns ErrEntryNotFound. MergeIntoAbove(ctx context.Context, sessionID, sourceID, queuedBy string) (*QueuedMessage, error) + // ReorderEntries atomically rewrites the FIFO positions of the session's + // pending entries to match orderedIDs. The submitted list must be an exact + // permutation of the currently visible pending (not reserved-in-flight) + // entry ids ordered as the caller wants them; any drift — missing, extra, + // or duplicate ids, or an id belonging to a reserved in-flight row — + // returns ErrQueueChanged and leaves the queue untouched. Reserved + // in-flight rows keep their place in the sequence; visible rows are + // interleaved in the submitted order and all positions are compacted to + // 1..N in one transaction. + ReorderEntries(ctx context.Context, sessionID string, orderedIDs []string) 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 @@ -149,6 +160,7 @@ type Repository interface { TakePendingMove(ctx context.Context, sessionID string) (*PendingMove, error) } +// applyMetadataUpdates merges metadata key updates into current; a nil value removes the key. func applyMetadataUpdates(current, updates map[string]interface{}) map[string]interface{} { merged := make(map[string]interface{}, len(current)+len(updates)) for key, value := range current { diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go index 3ed72bf405..e0efc92afa 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go @@ -30,12 +30,14 @@ func NewMemoryRepository() Repository { } } +// LifecycleGeneration returns the current archive/delete generation for a task. func (r *memoryRepository) LifecycleGeneration(_ context.Context, taskID string) (int64, error) { r.mu.Lock() defer r.mu.Unlock() return r.generation[taskID], nil } +// PurgeTask removes all task rows and advances its generation. func (r *memoryRepository) PurgeTask(_ context.Context, taskID string) (int, error) { r.mu.Lock() defer r.mu.Unlock() @@ -85,12 +87,14 @@ func (r *memoryRepository) CountPendingByTaskIDs(_ context.Context, taskIDs []st return counts, nil } +// Insert appends a new entry at the tail of the session's FIFO queue. func (r *memoryRepository) Insert(_ context.Context, msg *QueuedMessage, maxPerSession int) error { r.mu.Lock() defer r.mu.Unlock() return r.insertLocked(msg, maxPerSession) } +// Restore reinserts a previously dequeued entry at its original FIFO position. func (r *memoryRepository) Restore(_ context.Context, msg *QueuedMessage, maxPerSession int) error { r.mu.Lock() defer r.mu.Unlock() @@ -169,6 +173,7 @@ func (r *memoryRepository) AppendOrInsertTail(_ context.Context, sessionID, task return msg, false, nil } +// InsertOrReplaceByCoalesceKey replaces an entry with the same session/queued_by/coalesce key, or inserts when allowInsert is set. func (r *memoryRepository) InsertOrReplaceByCoalesceKey(_ context.Context, msg *QueuedMessage, coalesceKey string, maxPerSession int, allowInsert bool) (*QueuedMessage, bool, error) { r.mu.Lock() defer r.mu.Unlock() @@ -237,6 +242,7 @@ func (r *memoryRepository) InsertOrReplaceLifecycleByCoalesceKey(ctx context.Con return msg, false, nil } +// ListBySession returns all entries for a session ordered by position ascending. func (r *memoryRepository) ListBySession(_ context.Context, sessionID string) ([]QueuedMessage, error) { r.mu.Lock() defer r.mu.Unlock() @@ -249,12 +255,14 @@ func (r *memoryRepository) ListBySession(_ context.Context, sessionID string) ([ return out, nil } +// CountBySession returns the number of entries for a session. func (r *memoryRepository) CountBySession(_ context.Context, sessionID string) (int, error) { r.mu.Lock() defer r.mu.Unlock() return len(r.entries[sessionID]), nil } +// TakeHead atomically returns and deletes the lowest-position entry for the session. func (r *memoryRepository) TakeHead(_ context.Context, sessionID string) (*QueuedMessage, error) { r.mu.Lock() defer r.mu.Unlock() @@ -272,6 +280,7 @@ func (r *memoryRepository) TakeHead(_ context.Context, sessionID string) (*Queue return &out, nil } +// ReserveHead returns the lowest-position entry, deleting ordinary rows and reserving durable lifecycle rows. func (r *memoryRepository) ReserveHead(_ context.Context, sessionID string) (*QueuedMessage, error) { r.mu.Lock() defer r.mu.Unlock() @@ -298,6 +307,7 @@ func (r *memoryRepository) ReserveHead(_ context.Context, sessionID string) (*Qu return &out, nil } +// AcknowledgeByID removes a reserved durable entry after executor acceptance. func (r *memoryRepository) AcknowledgeByID(_ context.Context, sessionID, entryID string) error { r.mu.Lock() defer r.mu.Unlock() @@ -341,6 +351,7 @@ func (r *memoryRepository) TakeByID(_ context.Context, sessionID, entryID string return nil, nil } +// ClaimSendNow atomically claims the exact ordered source snapshot for a send-now dispatch. func (r *memoryRepository) ClaimSendNow(_ context.Context, sessionID string, expected []QueuedMessage) (*SendNowClaim, error) { r.mu.Lock() defer r.mu.Unlock() @@ -403,6 +414,7 @@ func (r *memoryRepository) ClaimSendNow(_ context.Context, sessionID string, exp return &SendNowClaim{Sources: sources, Dispatch: *envelope, SourceGenerations: generations}, nil } +// RestoreSendNowClaim puts every claimed source back at its original position. func (r *memoryRepository) RestoreSendNowClaim(_ context.Context, claim *SendNowClaim) error { if claim == nil || len(claim.Sources) == 0 { return ErrSendNowEmpty @@ -442,6 +454,7 @@ func (r *memoryRepository) RestoreSendNowClaim(_ context.Context, claim *SendNow return nil } +// validateMemorySendNowRestore verifies the stored entries still match the claim before restoring. func validateMemorySendNowRestore( claim *SendNowClaim, sessionID string, @@ -469,6 +482,7 @@ func validateMemorySendNowRestore( return nil } +// AcknowledgeSendNowClaim removes every durable source after the replacement prompt is accepted. func (r *memoryRepository) AcknowledgeSendNowClaim(_ context.Context, claim *SendNowClaim) error { if claim == nil || len(claim.Sources) == 0 { return ErrSendNowEmpty @@ -510,6 +524,7 @@ func (r *memoryRepository) AcknowledgeSendNowClaim(_ context.Context, claim *Sen return nil } +// cloneSendNowSources deep-copies the claimed sources so later mutation cannot alter the claim. func cloneSendNowSources(entries []*QueuedMessage) []QueuedMessage { sources := make([]QueuedMessage, 0, len(entries)) for _, entry := range entries { @@ -523,6 +538,7 @@ func cloneSendNowSources(entries []*QueuedMessage) []QueuedMessage { return sources } +// cloneQueuedMessage deep-copies a queued message including metadata and attachments. func cloneQueuedMessage(entry *QueuedMessage) *QueuedMessage { if entry == nil { return nil @@ -533,6 +549,7 @@ func cloneQueuedMessage(entry *QueuedMessage) *QueuedMessage { return &clone } +// sameQueuedMessageContent reports whether two entries carry identical content, attachments, and metadata. func sameQueuedMessageContent(left, right *QueuedMessage) bool { if left == nil || right == nil { return false @@ -546,10 +563,12 @@ func sameQueuedMessageContent(left, right *QueuedMessage) bool { return reflect.DeepEqual(leftCopy, rightCopy) } +// UpdateContent replaces the content of an entry owned by queuedBy. func (r *memoryRepository) UpdateContent(ctx context.Context, sessionID, entryID, content string, attachments []MessageAttachment, queuedBy string) error { return r.UpdateContentAndMetadata(ctx, sessionID, entryID, content, attachments, nil, queuedBy) } +// UpdateContentAndMetadata replaces content and applies metadata updates to an entry owned by queuedBy. func (r *memoryRepository) UpdateContentAndMetadata(_ context.Context, sessionID, entryID, content string, attachments []MessageAttachment, metadataUpdates map[string]interface{}, queuedBy string) error { r.mu.Lock() defer r.mu.Unlock() @@ -641,6 +660,56 @@ func (r *memoryRepository) MergeIntoAbove(_ context.Context, sessionID, sourceID return &merged, nil } +// ReorderEntries rewrites the session's visible pending order to match +// orderedIDs, mirroring the sqlite repository's semantics: reserved in-flight +// rows keep their place in the sequence, visible rows appear in the submitted +// order, and positions are compacted to 1..N. The stored slice is reassigned +// to the new sequence so TakeHead/ReserveHead (which consume list[0]) drain in +// the reordered order, not the pre-reorder slice order. Any drift returns +// ErrQueueChanged and leaves every entry untouched. +func (r *memoryRepository) ReorderEntries(_ context.Context, sessionID string, orderedIDs []string) error { + r.mu.Lock() + defer r.mu.Unlock() + stored := r.entries[sessionID] + + // The stored slice is not guaranteed position-sorted after ReplaceSession; + // sort a copy so reserved rows interleave at their persisted places. + sorted := make([]*QueuedMessage, len(stored)) + copy(sorted, stored) + sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].Position < sorted[j].Position }) + + visible, _ := splitVisibleAndReserved(sorted) + ordered, err := validateReorderSet(visible, orderedIDs) + if err != nil { + return err + } + + sequence := make([]*QueuedMessage, 0, len(sorted)) + visibleCursor := 0 + for _, msg := range sorted { + if msg.IsReservedInFlight() { + sequence = append(sequence, msg) + } else { + sequence = append(sequence, ordered[visibleCursor]) + visibleCursor++ + } + } + for i, msg := range sequence { + msg.Position = int64(i + 1) + } + if len(sequence) > 0 { + // The entries slice is consumed by index (TakeHead/ReserveHead take + // list[0]); it must be stored in the new position order or drains + // would still return the pre-reorder head. + r.entries[sessionID] = sequence + if n := len(sequence); int64(n) > r.nextPosition[sessionID] { + r.nextPosition[sessionID] = int64(n) + } + } + return nil +} + +// DeleteByID removes a single pending entry scoped to its session. func (r *memoryRepository) DeleteByID(_ context.Context, sessionID, entryID string) error { r.mu.Lock() defer r.mu.Unlock() @@ -665,6 +734,7 @@ func (r *memoryRepository) DeleteByID(_ context.Context, sessionID, entryID stri return ErrEntryNotFound } +// DeleteAllBySession removes every pending entry for a session, keeping reserved in-flight rows. func (r *memoryRepository) DeleteAllBySession(_ context.Context, sessionID string) (int, error) { r.mu.Lock() defer r.mu.Unlock() @@ -687,6 +757,7 @@ func (r *memoryRepository) DeleteAllBySession(_ context.Context, sessionID strin return removed, nil } +// TransferSession moves all entries (and any pending move) from one session to another. func (r *memoryRepository) TransferSession(_ context.Context, oldSessionID, newSessionID string) error { r.mu.Lock() defer r.mu.Unlock() @@ -725,6 +796,7 @@ func (r *memoryRepository) TransferSession(_ context.Context, oldSessionID, newS return nil } +// ReplaceSession replaces a session's queue with the supplied snapshot. func (r *memoryRepository) ReplaceSession(_ context.Context, sessionID string, entries []QueuedMessage, pendingMove *PendingMove) error { r.mu.Lock() defer r.mu.Unlock() @@ -754,6 +826,7 @@ func (r *memoryRepository) ReplaceSession(_ context.Context, sessionID string, e return nil } +// SetPendingMove upserts the deferred workflow move for a session. func (r *memoryRepository) SetPendingMove(_ context.Context, sessionID string, move *PendingMove) error { r.mu.Lock() defer r.mu.Unlock() @@ -765,6 +838,7 @@ func (r *memoryRepository) SetPendingMove(_ context.Context, sessionID string, m return nil } +// GetPendingMove returns the deferred workflow move for a session, or nil when absent. func (r *memoryRepository) GetPendingMove(_ context.Context, sessionID string) (*PendingMove, error) { r.mu.Lock() defer r.mu.Unlock() @@ -776,6 +850,7 @@ func (r *memoryRepository) GetPendingMove(_ context.Context, sessionID string) ( return &clone, nil } +// TakePendingMove returns and removes the deferred workflow move for a session. func (r *memoryRepository) TakePendingMove(_ context.Context, sessionID string) (*PendingMove, error) { r.mu.Lock() defer r.mu.Unlock() diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_reorder_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_reorder_test.go new file mode 100644 index 0000000000..fc60773097 --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/repository_reorder_test.go @@ -0,0 +1,283 @@ +package messagequeue + +import ( + "context" + "errors" + "reflect" + "sync" + "testing" + + "github.com/kandev/kandev/internal/testutil" +) + +// reorderRepos returns the repository constructors exercised by the reorder tests. +func reorderRepos(t *testing.T) []struct { + name string + new func(*testing.T) Repository +} { + t.Helper() + return []struct { + name string + new func(*testing.T) Repository + }{ + {name: "memory", new: func(*testing.T) Repository { return NewMemoryRepository() }}, + {name: "sqlite", new: newTestSQLiteRepo}, + {name: "postgres", new: newTestPostgresRepo}, + } +} + +// assertOrderedIDs fails the test unless entries match want in exact order. +func assertOrderedIDs(t *testing.T, entries []QueuedMessage, want ...string) { + t.Helper() + if len(entries) != len(want) { + t.Fatalf("entry count = %d, want %d (entries %#v)", len(entries), len(want), entries) + } + for i, id := range want { + if entries[i].ID != id { + t.Fatalf("order[%d] = %q, want %q (full order %v)", i, entries[i].ID, id, entryIDs(entries)) + } + } +} + +// entryIDs extracts the ordered id list from entries. +func entryIDs(entries []QueuedMessage) []string { + ids := make([]string, len(entries)) + for i, e := range entries { + ids[i] = e.ID + } + return ids +} + +// snapshotEntries lists the session queue for assertions. +func snapshotEntries(t *testing.T, repo Repository, sessionID string) []QueuedMessage { + t.Helper() + entries, err := repo.ListBySession(context.Background(), sessionID) + if err != nil { + t.Fatalf("list %q: %v", sessionID, err) + } + return entries +} + +func TestReorderEntriesRewritesVisibleOrder(t *testing.T) { + for _, tt := range reorderRepos(t) { + t.Run(tt.name, func(t *testing.T) { + repo := tt.new(t) + ctx := context.Background() + first := insertTestEntry(t, repo, "session-1", "task-1", "first", QueuedByUser, nil, nil) + second := insertTestEntry(t, repo, "session-1", "task-1", "second", QueuedByUser, nil, nil) + third := insertTestEntry(t, repo, "session-1", "task-1", "third", QueuedByUser, nil, nil) + + if err := repo.ReorderEntries(ctx, "session-1", []string{third.ID, first.ID, second.ID}); err != nil { + t.Fatalf("reorder: %v", err) + } + entries := snapshotEntries(t, repo, "session-1") + assertOrderedIDs(t, entries, third.ID, first.ID, second.ID) + for i, e := range entries { + if e.Position != int64(i+1) { + t.Fatalf("position[%d] = %d, want %d", i, e.Position, i+1) + } + if e.Content == "" || e.QueuedBy != QueuedByUser || e.SessionID != "session-1" { + t.Fatalf("reorder lost entry data: %#v", e) + } + } + }) + } +} + +func TestReorderEntriesNoOpKeepsPositions(t *testing.T) { + for _, tt := range reorderRepos(t) { + t.Run(tt.name, func(t *testing.T) { + repo := tt.new(t) + ctx := context.Background() + first := insertTestEntry(t, repo, "session-1", "task-1", "first", QueuedByUser, nil, nil) + second := insertTestEntry(t, repo, "session-1", "task-1", "second", QueuedByUser, nil, nil) + before := snapshotEntries(t, repo, "session-1") + + if err := repo.ReorderEntries(ctx, "session-1", []string{first.ID, second.ID}); err != nil { + t.Fatalf("no-op reorder: %v", err) + } + after := snapshotEntries(t, repo, "session-1") + if !reflect.DeepEqual(before, after) { + t.Fatalf("no-op reorder changed the queue:\nbefore %#v\nafter %#v", before, after) + } + }) + } +} + +func TestReorderEntriesRejectsDriftAtomically(t *testing.T) { + tests := []struct { + name string + ids func(first, second *QueuedMessage) []string + }{ + {name: "missing id", ids: func(first, _ *QueuedMessage) []string { return []string{first.ID} }}, + {name: "unknown id", ids: func(first, second *QueuedMessage) []string { return []string{first.ID, second.ID, "ghost"} }}, + {name: "duplicate id", ids: func(first, second *QueuedMessage) []string { return []string{first.ID, first.ID} }}, + {name: "empty list", ids: func(_, _ *QueuedMessage) []string { return nil }}, + } + for _, tt := range reorderRepos(t) { + t.Run(tt.name, func(t *testing.T) { + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + repo := tt.new(t) + ctx := context.Background() + first := insertTestEntry(t, repo, "session-1", "task-1", "first", QueuedByUser, nil, nil) + second := insertTestEntry(t, repo, "session-1", "task-1", "second", QueuedByUser, nil, nil) + before := snapshotEntries(t, repo, "session-1") + + err := repo.ReorderEntries(ctx, "session-1", tc.ids(first, second)) + if !errors.Is(err, ErrQueueChanged) { + t.Fatalf("reorder error = %v, want ErrQueueChanged", err) + } + after := snapshotEntries(t, repo, "session-1") + if !reflect.DeepEqual(before, after) { + t.Fatalf("failed reorder mutated the queue:\nbefore %#v\nafter %#v", before, after) + } + }) + } + }) + } +} + +func TestReorderEntriesKeepsReservedRowInPlace(t *testing.T) { + for _, tt := range reorderRepos(t) { + t.Run(tt.name, func(t *testing.T) { + repo := tt.new(t) + ctx := context.Background() + // A durable lifecycle row already reserved in flight is hidden from + // clients and must keep its place in the sequence. + reserved := insertTestEntry(t, repo, "session-1", "task-1", "durable", QueuedByWorkflow, nil, + map[string]interface{}{MetadataLifecycleDurable: true, MetadataLifecycleReserved: true}) + first := insertTestEntry(t, repo, "session-1", "task-1", "first", QueuedByUser, nil, nil) + second := insertTestEntry(t, repo, "session-1", "task-1", "second", QueuedByUser, nil, nil) + + if err := repo.ReorderEntries(ctx, "session-1", []string{second.ID, first.ID}); err != nil { + t.Fatalf("reorder visible rows: %v", err) + } + entries := snapshotEntries(t, repo, "session-1") + assertOrderedIDs(t, entries, reserved.ID, second.ID, first.ID) + if entries[0].Position != 1 || entries[1].Position != 2 || entries[2].Position != 3 { + t.Fatalf("positions = %d/%d/%d, want 1/2/3", entries[0].Position, entries[1].Position, entries[2].Position) + } + if !entries[0].IsReservedInFlight() { + t.Fatalf("reserved row lost its in-flight marker: %#v", entries[0]) + } + + // Submitting a reserved id is drift: it is not part of the visible set. + if err := repo.ReorderEntries(ctx, "session-1", []string{reserved.ID, first.ID, second.ID}); !errors.Is(err, ErrQueueChanged) { + t.Fatalf("reorder with reserved id error = %v, want ErrQueueChanged", err) + } + after := snapshotEntries(t, repo, "session-1") + if !reflect.DeepEqual(entries, after) { + t.Fatalf("failed reorder mutated the queue:\nbefore %#v\nafter %#v", entries, after) + } + }) + } +} + +// TestReorderEntriesDrainOrderAfterReorder verifies that drains consume the +// queue in the NEW order: TakeHead/ReserveHead take the stored slice head, so +// the memory repository must keep its slice in position order after a reorder +// (ListBySession sorts and would mask a stale slice order). +func TestReorderEntriesDrainOrderAfterReorder(t *testing.T) { + for _, tt := range reorderRepos(t) { + t.Run(tt.name, func(t *testing.T) { + repo := tt.new(t) + ctx := context.Background() + first := insertTestEntry(t, repo, "session-1", "task-1", "first", QueuedByUser, nil, nil) + second := insertTestEntry(t, repo, "session-1", "task-1", "second", QueuedByUser, nil, nil) + + if err := repo.ReorderEntries(ctx, "session-1", []string{second.ID, first.ID}); err != nil { + t.Fatalf("reorder: %v", err) + } + head, err := repo.TakeHead(ctx, "session-1") + if err != nil { + t.Fatalf("take head: %v", err) + } + if head == nil || head.ID != second.ID { + t.Fatalf("after reorder TakeHead returned %v, want %q", head, second.ID) + } + next, err := repo.TakeHead(ctx, "session-1") + if err != nil { + t.Fatalf("take next head: %v", err) + } + if next == nil || next.ID != first.ID { + t.Fatalf("after reorder second TakeHead returned %v, want %q", next, first.ID) + } + }) + } +} + +// TestReorderEntriesCrossInstanceStaleReorderRejected runs two repository +// instances over one shared database (Postgres; skipped unless +// KANDEV_TEST_POSTGRES_DSN is set) racing conflicting reorders. The +// per-session lock is per-instance, so MVCC can hand both transactions the +// pre-reorder snapshot; the position compare-and-swap must reject the stale +// writer with ErrQueueChanged instead of silently clobbering the newer order. +func TestReorderEntriesCrossInstanceStaleReorderRejected(t *testing.T) { + dsn := testutil.PostgresDSNFromEnv(t) + db := testutil.OpenIsolatedPostgres(t, dsn) + repo1, err := NewSQLiteRepository(db, db) + if err != nil { + t.Fatalf("NewSQLiteRepository(1): %v", err) + } + repo2, err := NewSQLiteRepository(db, db) + if err != nil { + t.Fatalf("NewSQLiteRepository(2): %v", err) + } + + ctx := context.Background() + first := insertTestEntry(t, repo1, "session-1", "task-1", "first", QueuedByUser, nil, nil) + second := insertTestEntry(t, repo1, "session-1", "task-1", "second", QueuedByUser, nil, nil) + + var wg sync.WaitGroup + results := make([]error, 2) + wg.Add(2) + go func() { + defer wg.Done() + results[0] = repo1.ReorderEntries(ctx, "session-1", []string{second.ID, first.ID}) + }() + go func() { + defer wg.Done() + results[1] = repo2.ReorderEntries(ctx, "session-1", []string{first.ID, second.ID}) + }() + wg.Wait() + + for i, reorderErr := range results { + if reorderErr != nil && !errors.Is(reorderErr, ErrQueueChanged) { + t.Fatalf("instance %d reorder error = %v, want nil or ErrQueueChanged", i, reorderErr) + } + } + entries, err := repo1.ListBySession(ctx, "session-1") + if err != nil { + t.Fatalf("list: %v", err) + } + // A lost update would leave a third, pre-reorder order; the final order + // must be exactly one of the two submitted orders. + order := entryIDs(entries) + if !reflect.DeepEqual(order, []string{first.ID, second.ID}) && + !reflect.DeepEqual(order, []string{second.ID, first.ID}) { + t.Fatalf("final order %v is neither submitted order", order) + } +} + +func TestReorderEntriesDrainRaceRejected(t *testing.T) { + for _, tt := range reorderRepos(t) { + t.Run(tt.name, func(t *testing.T) { + repo := tt.new(t) + ctx := context.Background() + first := insertTestEntry(t, repo, "session-1", "task-1", "first", QueuedByUser, nil, nil) + second := insertTestEntry(t, repo, "session-1", "task-1", "second", QueuedByUser, nil, nil) + + // A drain wins between the client's snapshot and the reorder: the + // submitted set no longer matches the visible set. + if _, err := repo.TakeHead(ctx, "session-1"); err != nil { + t.Fatalf("take head: %v", err) + } + if err := repo.ReorderEntries(ctx, "session-1", []string{first.ID, second.ID}); !errors.Is(err, ErrQueueChanged) { + t.Fatalf("reorder after drain error = %v, want ErrQueueChanged", err) + } + entries := snapshotEntries(t, repo, "session-1") + assertOrderedIDs(t, entries, second.ID) + }) + } +} diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go index 04bc86bc9c..d9c8e52f66 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go @@ -50,6 +50,7 @@ func (r *sqliteRepository) withSessionLock(sessionID string) func() { return func() { lock.Unlock() } } +// initSchema creates the queue tables and indexes idempotently. func (r *sqliteRepository) initSchema() error { _, err := r.db.Exec(` CREATE TABLE IF NOT EXISTS queued_messages ( @@ -85,6 +86,7 @@ func (r *sqliteRepository) initSchema() error { return err } +// Insert appends a new entry at the tail of the session's FIFO queue. func (r *sqliteRepository) Insert(ctx context.Context, msg *QueuedMessage, maxPerSession int) error { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -136,6 +138,7 @@ func (r *sqliteRepository) Insert(ctx context.Context, msg *QueuedMessage, maxPe return tx.Commit() } +// Restore reinserts a previously dequeued entry at its original FIFO position. func (r *sqliteRepository) Restore(ctx context.Context, msg *QueuedMessage, maxPerSession int) error { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -178,6 +181,7 @@ func (r *sqliteRepository) Restore(ctx context.Context, msg *QueuedMessage, maxP return tx.Commit() } +// AppendOrInsertTail concatenates onto the tail entry when its owner matches, otherwise inserts a new entry. func (r *sqliteRepository) AppendOrInsertTail(ctx context.Context, sessionID, taskID, content, model, queuedBy string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, maxPerSession int) (*QueuedMessage, bool, error) { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -253,6 +257,7 @@ func (r *sqliteRepository) AppendOrInsertTail(ctx context.Context, sessionID, ta return msg, false, nil } +// InsertOrReplaceByCoalesceKey replaces an entry with the same session/queued_by/coalesce key, or inserts when allowInsert is set. func (r *sqliteRepository) InsertOrReplaceByCoalesceKey(ctx context.Context, msg *QueuedMessage, coalesceKey string, maxPerSession int, allowInsert bool) (*QueuedMessage, bool, error) { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -350,6 +355,7 @@ func (r *sqliteRepository) InsertOrReplaceLifecycleByCoalesceKey(ctx context.Con return msg, false, nil } +// LifecycleGeneration returns the current archive/delete generation for a task. func (r *sqliteRepository) LifecycleGeneration(ctx context.Context, taskID string) (int64, error) { var generation int64 err := r.ro.GetContext(ctx, &generation, r.ro.Rebind(` @@ -364,6 +370,7 @@ func (r *sqliteRepository) LifecycleGeneration(ctx context.Context, taskID strin return generation, nil } +// PurgeTask removes all task rows and advances its generation. func (r *sqliteRepository) PurgeTask(ctx context.Context, taskID string) (int, error) { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -380,6 +387,7 @@ func (r *sqliteRepository) PurgeTask(ctx context.Context, taskID string) (int, e return removed, nil } +// lifecycleGenerationInTx reads the lifecycle generation inside an existing transaction. func lifecycleGenerationInTx(ctx context.Context, tx *sqlx.Tx, db *sqlx.DB, taskID string) (int64, error) { var generation int64 err := tx.GetContext(ctx, &generation, db.Rebind(` @@ -415,6 +423,7 @@ func PurgeTaskInTransaction(ctx context.Context, tx *sqlx.Tx, db *sqlx.DB, taskI return int(removed), nil } +// replaceCoalesced overwrites the existing coalesced row with msg inside the transaction. func (r *sqliteRepository) replaceCoalesced(ctx context.Context, tx *sqlx.Tx, existing, msg *QueuedMessage) (*QueuedMessage, error) { if msg.QueuedAt.IsZero() { msg.QueuedAt = time.Now().UTC() @@ -457,6 +466,7 @@ func (r *sqliteRepository) replaceCoalesced(ctx context.Context, tx *sqlx.Tx, ex return existing, nil } +// insertCoalesced inserts msg as a new tail entry inside the transaction, honoring the capacity cap. func (r *sqliteRepository) insertCoalesced(ctx context.Context, tx *sqlx.Tx, msg *QueuedMessage, maxPerSession int) error { if err := r.ensureQueueCapacity(ctx, tx, msg.SessionID, maxPerSession); err != nil { return err @@ -493,6 +503,7 @@ func (r *sqliteRepository) insertCoalesced(ctx context.Context, tx *sqlx.Tx, msg return nil } +// ensureQueueCapacity rejects the insert when the session already holds maxPerSession entries. func (r *sqliteRepository) ensureQueueCapacity(ctx context.Context, tx *sqlx.Tx, sessionID string, maxPerSession int) error { if maxPerSession <= 0 { return nil @@ -507,6 +518,7 @@ func (r *sqliteRepository) ensureQueueCapacity(ctx context.Context, tx *sqlx.Tx, return nil } +// ListBySession returns all entries for a session ordered by position ascending. func (r *sqliteRepository) ListBySession(ctx context.Context, sessionID string) ([]QueuedMessage, error) { rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(` SELECT id, session_id, task_id, position, content, model, plan_mode, @@ -531,6 +543,7 @@ func (r *sqliteRepository) ListBySession(ctx context.Context, sessionID string) return out, rows.Err() } +// CountBySession returns the number of entries for a session. func (r *sqliteRepository) CountBySession(ctx context.Context, sessionID string) (int, error) { var n int err := r.ro.GetContext(ctx, &n, r.ro.Rebind(`SELECT COUNT(*) FROM queued_messages WHERE session_id = ?`), sessionID) @@ -577,6 +590,7 @@ func (r *sqliteRepository) CountPendingByTaskIDs(ctx context.Context, taskIDs [] return counts, nil } +// TakeHead atomically returns and deletes the lowest-position entry for the session. 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. @@ -627,6 +641,7 @@ func (r *sqliteRepository) TakeHead(ctx context.Context, sessionID string) (*Que return msg, nil } +// ReserveHead returns the lowest-position entry, deleting ordinary rows and reserving durable lifecycle rows. func (r *sqliteRepository) ReserveHead(ctx context.Context, sessionID string) (*QueuedMessage, error) { unlock := r.withSessionLock(sessionID) defer unlock() @@ -700,6 +715,7 @@ func (r *sqliteRepository) ReserveHead(ctx context.Context, sessionID string) (* return msg, nil } +// AcknowledgeByID removes a reserved durable entry after executor acceptance. func (r *sqliteRepository) AcknowledgeByID(ctx context.Context, sessionID, entryID string) error { unlock := r.withSessionLock(sessionID) defer unlock() @@ -769,6 +785,7 @@ func (r *sqliteRepository) TakeByID(ctx context.Context, sessionID, entryID stri return msg, nil } +// ClaimSendNow atomically claims the exact ordered source snapshot for a send-now dispatch. func (r *sqliteRepository) ClaimSendNow(ctx context.Context, sessionID string, expected []QueuedMessage) (*SendNowClaim, error) { if len(expected) == 0 { return nil, ErrSendNowEmpty @@ -820,6 +837,7 @@ func (r *sqliteRepository) ClaimSendNow(ctx context.Context, sessionID string, e return &SendNowClaim{Sources: sources, Dispatch: *envelope, SourceGenerations: generations}, nil } +// RestoreSendNowClaim puts every claimed source back at its original position. func (r *sqliteRepository) RestoreSendNowClaim(ctx context.Context, claim *SendNowClaim) error { tx, sessionID, unlock, err := r.beginSendNowClaimTx(ctx, claim, "restore") if err != nil { @@ -857,6 +875,7 @@ func (r *sqliteRepository) RestoreSendNowClaim(ctx context.Context, claim *SendN return tx.Commit() } +// AcknowledgeSendNowClaim removes every durable source after the replacement prompt is accepted. func (r *sqliteRepository) AcknowledgeSendNowClaim(ctx context.Context, claim *SendNowClaim) error { tx, sessionID, unlock, err := r.beginSendNowClaimTx(ctx, claim, "acknowledge") if err != nil { @@ -879,6 +898,7 @@ func (r *sqliteRepository) AcknowledgeSendNowClaim(ctx context.Context, claim *S return tx.Commit() } +// beginSendNowClaimTx starts the claim transaction and takes the per-session lock. func (r *sqliteRepository) beginSendNowClaimTx( ctx context.Context, claim *SendNowClaim, @@ -903,11 +923,13 @@ type storedQueueEntry struct { attachmentsJSON string } +// listStoredSessionEntries reads a session's entries by id inside a transaction. func (r *sqliteRepository) listStoredSessionEntries(ctx context.Context, tx *sqlx.Tx, sessionID string) (map[string]storedQueueEntry, error) { _, entries, err := r.listOrderedStoredSessionEntries(ctx, tx, sessionID) return entries, err } +// listOrderedStoredSessionEntries reads a session's entries ordered by position, plus an id index. func (r *sqliteRepository) listOrderedStoredSessionEntries(ctx context.Context, tx *sqlx.Tx, sessionID string) ([]storedQueueEntry, map[string]storedQueueEntry, error) { rows, err := tx.QueryxContext(ctx, r.db.Rebind(` SELECT id, session_id, task_id, position, content, model, plan_mode, @@ -941,6 +963,7 @@ func (r *sqliteRepository) listOrderedStoredSessionEntries(ctx context.Context, return ordered, entries, nil } +// selectSQLiteSendNowSources picks the requested sources in FIFO order and validates them against the expected snapshot. func selectSQLiteSendNowSources( ordered []storedQueueEntry, byID map[string]storedQueueEntry, @@ -973,6 +996,7 @@ func selectSQLiteSendNowSources( return cloneSendNowSources(selected), nil } +// applySQLiteSendNowClaim removes or reserves every claimed source inside the transaction. func (r *sqliteRepository) applySQLiteSendNowClaim( ctx context.Context, tx *sqlx.Tx, @@ -995,6 +1019,7 @@ func (r *sqliteRepository) applySQLiteSendNowClaim( return nil } +// reserveSQLiteSendNowSource marks a durable lifecycle source as in flight. func (r *sqliteRepository) reserveSQLiteSendNowSource( ctx context.Context, tx *sqlx.Tx, @@ -1023,6 +1048,7 @@ func (r *sqliteRepository) reserveSQLiteSendNowSource( return nil } +// removeSQLiteSendNowSource deletes an ordinary source, failing when its stored content changed. func (r *sqliteRepository) removeSQLiteSendNowSource( ctx context.Context, tx *sqlx.Tx, @@ -1047,6 +1073,7 @@ func (r *sqliteRepository) removeSQLiteSendNowSource( return nil } +// validateSQLiteSendNowRestore verifies stored entries still match the claim before restoring. func validateSQLiteSendNowRestore( claim *SendNowClaim, sessionID string, @@ -1074,6 +1101,7 @@ func validateSQLiteSendNowRestore( return nil } +// restoreSQLiteSendNowSource reinserts or unmarks a claimed source at its original position. func (r *sqliteRepository) restoreSQLiteSendNowSource( ctx context.Context, tx *sqlx.Tx, @@ -1109,6 +1137,7 @@ func (r *sqliteRepository) restoreSQLiteSendNowSource( return nil } +// insertSQLiteSendNowSource reinserts a restored source row. func (r *sqliteRepository) insertSQLiteSendNowSource(ctx context.Context, tx *sqlx.Tx, source QueuedMessage) error { attachmentsJSON, err := marshalAttachments(source.Attachments) if err != nil { @@ -1129,6 +1158,7 @@ func (r *sqliteRepository) insertSQLiteSendNowSource(ctx context.Context, tx *sq return nil } +// validateSQLiteSendNowAcknowledge verifies every durable source is still reserved before acknowledgement. func validateSQLiteSendNowAcknowledge(claim *SendNowClaim, sessionID string, stored map[string]storedQueueEntry) error { for _, source := range claim.Sources { if source.SessionID != sessionID { @@ -1144,6 +1174,7 @@ func validateSQLiteSendNowAcknowledge(claim *SendNowClaim, sessionID string, sto return nil } +// acknowledgeSQLiteSendNowSource deletes a reserved durable source row. func (r *sqliteRepository) acknowledgeSQLiteSendNowSource( ctx context.Context, tx *sqlx.Tx, @@ -1174,10 +1205,12 @@ func (r *sqliteRepository) acknowledgeSQLiteSendNowSource( return nil } +// UpdateContent replaces the content of an entry owned by queuedBy. func (r *sqliteRepository) UpdateContent(ctx context.Context, sessionID, entryID, content string, attachments []MessageAttachment, queuedBy string) error { return r.UpdateContentAndMetadata(ctx, sessionID, entryID, content, attachments, nil, queuedBy) } +// UpdateContentAndMetadata replaces content and applies metadata updates to an entry owned by queuedBy. func (r *sqliteRepository) UpdateContentAndMetadata(ctx context.Context, sessionID, entryID, content string, attachments []MessageAttachment, metadataUpdates map[string]interface{}, queuedBy string) error { if queuedBy == "" || IsReservedQueuedBy(queuedBy) { return ErrEntryNotFound @@ -1275,6 +1308,100 @@ func (r *sqliteRepository) MergeIntoAbove(ctx context.Context, sessionID, source return &merged, nil } +// ReorderEntries atomically rewrites the FIFO positions of the session's +// visible pending entries to match orderedIDs. Reserved in-flight lifecycle +// rows keep their place in the sequence; visible rows are interleaved in the +// submitted order and all positions are compacted to 1..N in one transaction. +// Any drift — a missing, extra, or duplicate id, or an id belonging to a +// reserved in-flight row — returns ErrQueueChanged and leaves the queue +// untouched. The per-session lock plus the in-transaction read make the +// validation and the position rewrite one atomic snapshot in-process; across +// backend instances sharing the same database (Postgres), each UPDATE carries +// the row's read-time position as a compare-and-swap precondition, so a stale +// reorder rolls back with ErrQueueChanged instead of clobbering a newer order. +// A row that vanished mid-transaction (a drain racing the lock) fails the same +// precondition and rolls back. +func (r *sqliteRepository) ReorderEntries(ctx context.Context, sessionID string, orderedIDs []string) error { + unlock := r.withSessionLock(sessionID) + defer unlock() + + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return fmt.Errorf("begin reorder tx: %w", err) + } + defer func() { _ = tx.Rollback() }() + + rows, err := tx.QueryxContext(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 = ? + ORDER BY position ASC + `), sessionID) + if err != nil { + return fmt.Errorf("read reorder rows: %w", err) + } + stored := make([]*QueuedMessage, 0, 4) + for rows.Next() { + msg, scanErr := scanQueuedRow(rows) + if scanErr != nil { + _ = rows.Close() + return fmt.Errorf("scan reorder row: %w", scanErr) + } + stored = append(stored, msg) + } + if err := rows.Close(); err != nil { + return fmt.Errorf("close reorder rows: %w", err) + } + + visible, _ := splitVisibleAndReserved(stored) + ordered, err := validateReorderSet(visible, orderedIDs) + if err != nil { + return err + } + + // Interleave reserved rows at their current places; visible rows emit in + // the submitted order. Positions are compacted to 1..N. + sequence := make([]*QueuedMessage, 0, len(stored)) + visibleCursor := 0 + for _, msg := range stored { + if msg.IsReservedInFlight() { + sequence = append(sequence, msg) + } else { + sequence = append(sequence, ordered[visibleCursor]) + visibleCursor++ + } + } + + for i, msg := range sequence { + res, err := tx.ExecContext(ctx, r.db.Rebind(` + UPDATE queued_messages + SET position = ? + WHERE id = ? AND session_id = ? AND position = ? + `), int64(i+1), msg.ID, sessionID, msg.Position) + if err != nil { + return fmt.Errorf("reorder position %d: %w", i+1, err) + } + affected, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("reorder position %d rows affected: %w", i+1, err) + } + if affected == 0 { + // The row vanished or its position drifted since the in-transaction + // read — a concurrent drain, or another backend instance committed a + // reorder for this session in between. The per-session lock only + // serializes in-process; the position precondition is the + // cross-instance compare-and-swap, so a stale reorder rolls back + // with ErrQueueChanged instead of clobbering the newer order. + return ErrQueueChanged + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit reorder: %w", err) + } + return 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) { @@ -1376,6 +1503,7 @@ func applyMergeWrites(ctx context.Context, r *sqliteRepository, tx *sqlx.Tx, tar return nil } +// DeleteByID removes a single pending entry scoped to its session. func (r *sqliteRepository) DeleteByID(ctx context.Context, sessionID, entryID string) error { unlock := r.withSessionLock(sessionID) defer unlock() @@ -1421,6 +1549,7 @@ func (r *sqliteRepository) DeleteByID(ctx context.Context, sessionID, entryID st return tx.Commit() } +// DeleteAllBySession removes every pending entry for a session, keeping reserved in-flight rows. func (r *sqliteRepository) DeleteAllBySession(ctx context.Context, sessionID string) (int, error) { unlock := r.withSessionLock(sessionID) defer unlock() @@ -1461,6 +1590,7 @@ type cancellationCandidate struct { metadataJSON string } +// cancellationCandidates lists pending entries eligible for cancellation, excluding reserved in-flight rows. func cancellationCandidates( ctx context.Context, tx *sqlx.Tx, @@ -1494,6 +1624,7 @@ func cancellationCandidates( return candidates, nil } +// isReservedMetadataJSON reports whether serialized metadata marks a reserved in-flight row. func isReservedMetadataJSON(metadataJSON string) (bool, error) { metadata := make(map[string]interface{}) if metadataJSON != "" && metadataJSON != "{}" { @@ -1504,6 +1635,7 @@ func isReservedMetadataJSON(metadataJSON string) (bool, error) { return (&QueuedMessage{Metadata: metadata}).IsReservedInFlight(), nil } +// TransferSession moves all entries (and any pending move) from one session to another. func (r *sqliteRepository) TransferSession(ctx context.Context, oldSessionID, newSessionID string) error { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -1538,6 +1670,7 @@ func (r *sqliteRepository) TransferSession(ctx context.Context, oldSessionID, ne return tx.Commit() } +// ReplaceSession replaces a session's queue with the supplied snapshot. func (r *sqliteRepository) ReplaceSession(ctx context.Context, sessionID string, entries []QueuedMessage, pendingMove *PendingMove) error { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -1594,6 +1727,7 @@ func (r *sqliteRepository) ReplaceSession(ctx context.Context, sessionID string, return tx.Commit() } +// SetPendingMove upserts the deferred workflow move for a session. func (r *sqliteRepository) SetPendingMove(ctx context.Context, sessionID string, move *PendingMove) error { if move.QueuedAt.IsZero() { move.QueuedAt = time.Now().UTC() @@ -1616,6 +1750,7 @@ func (r *sqliteRepository) SetPendingMove(ctx context.Context, sessionID string, return nil } +// GetPendingMove returns the deferred workflow move for a session, or nil when absent. func (r *sqliteRepository) GetPendingMove(ctx context.Context, sessionID string) (*PendingMove, error) { var ( taskID, workflowID, workflowStepID string @@ -1640,6 +1775,7 @@ func (r *sqliteRepository) GetPendingMove(ctx context.Context, sessionID string) }, nil } +// TakePendingMove returns and removes the deferred workflow move for a session. func (r *sqliteRepository) TakePendingMove(ctx context.Context, sessionID string) (*PendingMove, error) { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -1697,6 +1833,7 @@ func (r *sqliteRepository) scanTail(ctx context.Context, tx *sqlx.Tx, sessionID return msg, nil } +// findCoalesced locates the entry matching the session, owner, and coalesce key inside a transaction. func (r *sqliteRepository) findCoalesced(ctx context.Context, tx *sqlx.Tx, sessionID, queuedBy, coalesceKey string) (*QueuedMessage, error) { rows, err := tx.QueryxContext(ctx, r.db.Rebind(` SELECT id, session_id, task_id, position, content, model, plan_mode, @@ -1730,6 +1867,7 @@ func scanQueuedRow(scanner interface{ Scan(dest ...any) error }) (*QueuedMessage return msg, err } +// scanQueuedRowWithMetadataJSON scans a queue row plus its raw metadata JSON. func scanQueuedRowWithMetadataJSON( scanner interface{ Scan(dest ...any) error }, ) (*QueuedMessage, string, error) { @@ -1758,6 +1896,7 @@ func scanQueuedRowWithMetadataJSON( return &msg, metaJSON, nil } +// marshalAttachments serializes attachments for storage. func marshalAttachments(att []MessageAttachment) (string, error) { if len(att) == 0 { return "[]", nil @@ -1769,6 +1908,7 @@ func marshalAttachments(att []MessageAttachment) (string, error) { return string(b), nil } +// marshalMetadata serializes metadata for storage. func marshalMetadata(meta map[string]interface{}) (string, error) { if len(meta) == 0 { return "{}", nil @@ -1780,6 +1920,7 @@ func marshalMetadata(meta map[string]interface{}) (string, error) { return string(b), nil } +// boolToInt converts a boolean to its integer storage form. func boolToInt(b bool) int { if b { return 1 @@ -1787,6 +1928,7 @@ func boolToInt(b bool) int { return 0 } +// metadataString reads a string metadata key, returning empty when absent. func metadataString(meta map[string]interface{}, key string) string { if meta == nil { return "" diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index 81edcc0b35..b4f7533d86 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/apps/backend/internal/orchestrator/messagequeue/service.go @@ -146,6 +146,7 @@ func (s *Service) QueueMessageWithMetadata(ctx context.Context, sessionID, taskI ) } +// queueMessageWithMetadata is the admission-checked core of QueueMessageWithMetadata. func (s *Service) queueMessageWithMetadata(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, maxPerSession int) (*QueuedMessage, error) { var queued *QueuedMessage err := s.WithSessionAdmission(ctx, sessionID, func(admittedCtx context.Context) error { @@ -158,6 +159,7 @@ func (s *Service) queueMessageWithMetadata(ctx context.Context, sessionID, taskI return queued, err } +// insertQueueMessageWithMetadata inserts a message with metadata under the per-session admission lock. func (s *Service) insertQueueMessageWithMetadata(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, maxPerSession int) (*QueuedMessage, error) { metadataCopy := copyMessageMetadata(metadata, 0) msg := &QueuedMessage{ @@ -203,6 +205,7 @@ func (s *Service) RestoreMessage(ctx context.Context, msg *QueuedMessage) (*Queu return restored, err } +// restoreMessage is the admission-checked core of RestoreMessage. func (s *Service) restoreMessage(ctx context.Context, msg *QueuedMessage) (*QueuedMessage, error) { restored := *msg restored.Metadata = copyMessageMetadata(msg.Metadata, 0) @@ -228,6 +231,7 @@ func (s *Service) QueueMessageWithCoalesceKey(ctx context.Context, sessionID, ta ) } +// queueMessageWithCoalesceKey is the admission-checked core of QueueMessageWithCoalesceKey. func (s *Service) queueMessageWithCoalesceKey(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, coalesceKey string, allowInsert bool, maxPerSession int) (*QueuedMessage, bool, error) { var queued *QueuedMessage var replaced bool @@ -242,6 +246,7 @@ func (s *Service) queueMessageWithCoalesceKey(ctx context.Context, sessionID, ta return queued, replaced, err } +// insertQueueMessageWithCoalesceKey inserts or coalesce-replaces an entry under the admission lock. func (s *Service) insertQueueMessageWithCoalesceKey(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, coalesceKey string, allowInsert bool, maxPerSession int) (*QueuedMessage, bool, error) { metadataCopy := copyMessageMetadata(metadata, 1) metadataCopy[MetadataCoalesceKey] = coalesceKey @@ -308,6 +313,7 @@ func (s *Service) RequeueLifecycleMessageWithCoalesceKey(ctx context.Context, se return s.queueLifecycleMessageWithCoalesceKey(ctx, sessionID, taskID, content, model, userID, planMode, attachments, metadata, coalesceKey, allowInsert, true) } +// queueLifecycleMessageWithCoalesceKey is the lifecycle-guarded core of the lifecycle queue methods. func (s *Service) queueLifecycleMessageWithCoalesceKey(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, coalesceKey string, allowInsert, isRetry bool) (*QueuedMessage, bool, bool, error) { var queued *QueuedMessage var replaced, accepted bool @@ -322,6 +328,7 @@ func (s *Service) queueLifecycleMessageWithCoalesceKey(ctx context.Context, sess return queued, replaced, accepted, err } +// insertLifecycleMessageWithCoalesceKey inserts or replaces a lifecycle entry under the admission lock. func (s *Service) insertLifecycleMessageWithCoalesceKey(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, coalesceKey string, allowInsert, isRetry bool) (*QueuedMessage, bool, bool, error) { metadataCopy := clearReservedMetadata(metadata) generation, err := s.repo.LifecycleGeneration(ctx, taskID) @@ -364,6 +371,7 @@ func (s *Service) PurgeTask(ctx context.Context, taskID string) (int, error) { return s.repo.PurgeTask(ctx, taskID) } +// lifecycleGenerationFromMetadata reads the lifecycle generation captured on an entry. func lifecycleGenerationFromMetadata(metadata map[string]interface{}) (int64, bool) { if metadata == nil { return 0, false @@ -431,6 +439,7 @@ func (s *Service) IsCurrentLifecycleReservation(ctx context.Context, msg *Queued return false } +// copyMessageMetadata clones metadata, optionally growing the map capacity. func copyMessageMetadata(metadata map[string]interface{}, extraCapacity int) map[string]interface{} { if len(metadata) == 0 && extraCapacity == 0 { return nil @@ -626,6 +635,24 @@ func (s *Service) MergeIntoAbove(ctx context.Context, sessionID, entryID, queued return merged, nil } +// ReorderEntries rewrites the pending FIFO order of a session's queue to +// match orderedIDs. Returns ErrQueueChanged when the submitted order is empty, +// contains duplicates, or does not match the current visible pending set (a +// drain/remove/merge raced the request); callers refetch the authoritative +// queue. +func (s *Service) ReorderEntries(ctx context.Context, sessionID string, orderedIDs []string) error { + if err := validateReorderInput(orderedIDs); err != nil { + return err + } + if err := s.repo.ReorderEntries(ctx, sessionID, orderedIDs); err != nil { + return err + } + s.logger.Info("queued entries reordered", + zap.String("session_id", sessionID), + zap.Int("entry_count", len(orderedIDs))) + return 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/types.go b/apps/backend/internal/orchestrator/messagequeue/types.go index 904d8a2bc4..70e53b00e2 100644 --- a/apps/backend/internal/orchestrator/messagequeue/types.go +++ b/apps/backend/internal/orchestrator/messagequeue/types.go @@ -93,6 +93,11 @@ var ( // message merging is disabled (see Service.SetMergeEnabled). The setting // is admin-controlled and enabled by default. ErrMergeDisabled = errors.New("queued message merging is disabled") + // ErrQueueChanged is returned when a reorder's submitted id set does not + // match the session's current visible pending entries — an entry was + // drained, removed, merged, or newly queued since the client's snapshot. + // The reorder is rejected atomically with no partial position rewrite. + ErrQueueChanged = errors.New("queue changed during reorder") // 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 a737240917..b4aeb639c0 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -120,6 +120,7 @@ const ( ActionMessageQueueSendNow = "message.queue.send_now" // Interrupt and replace the active turn with an exact queue selection ActionMessageQueueRemove = "message.queue.remove" // Delete a single entry by id ActionMessageQueueMerge = "message.queue.merge" // Fold an entry into the entry above it + ActionMessageQueueReorder = "message.queue.reorder" // Rewrite the visible pending order for a session 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 f7e92005a1..55a56fa0e1 100644 --- a/apps/web/components/task/chat/queued-ghost-list.test.tsx +++ b/apps/web/components/task/chat/queued-ghost-list.test.tsx @@ -91,6 +91,7 @@ function baseState(entries: QueuedMessage[]) { editEntry: vi.fn(async () => {}), removeEntry: vi.fn(async () => {}), mergeEntry: vi.fn(async () => {}), + reorderEntries: vi.fn(async () => {}), sendEntryNow: vi.fn(async () => {}), sendAllNow: vi.fn(async () => {}), cancellationPending: false, @@ -123,6 +124,7 @@ beforeEach(() => { afterEach(() => { cleanup(); + vi.restoreAllMocks(); }); describe("QueueAffordance", () => { @@ -526,3 +528,158 @@ describe("QueueAffordance merge wiring — dispatch", () => { expect(screen.queryAllByTestId(MERGE_BUTTON_ID)).toHaveLength(0); }); }); + +describe("QueueAffordance reorder", () => { + const GRAB_HANDLE_ID = "queue-grab-handle"; + + it("renders a localized grab handle on every row when multiple entries are queued", () => { + const state = queueState([ + entry({ id: "q-1", content: "first" }), + entry({ id: "q-2", content: "second" }), + ]); + useQueueMock.mockReturnValue(state); + renderQueue({CHILD}); + fireEvent.click(screen.getByTestId(CHIP_ID)); + + const handles = screen.getAllByTestId(GRAB_HANDLE_ID); + expect(handles).toHaveLength(2); + for (const handle of handles) { + expect(handle.getAttribute("aria-label")).toBe("Reorder queued message"); + expect(handle.getAttribute("aria-roledescription")).toBe("sortable"); + expect((handle as HTMLButtonElement).disabled).toBe(false); + } + const rowShell = screen.getAllByTestId("queue-entry")[0].parentElement!; + expect(rowShell.className).toContain("pl-5"); + expect(rowShell.className).toContain("[@media(pointer:coarse)]:pl-11"); + }); + + it("does not render a handle when only one entry is queued", () => { + useQueueMock.mockReturnValue(queueState([entry()])); + renderQueue({CHILD}); + fireEvent.click(screen.getByTestId(CHIP_ID)); + + expect(screen.queryByTestId(GRAB_HANDLE_ID)).toBeNull(); + }); + + it("keeps compact row padding when no drag handle is shown", () => { + useQueueMock.mockReturnValue(queueState([entry()])); + renderQueue({CHILD}); + fireEvent.click(screen.getByTestId(CHIP_ID)); + + const rowShell = screen.getAllByTestId("queue-entry")[0].parentElement!; + expect(rowShell.className).toContain("pl-2"); + expect(rowShell.className).not.toContain("pl-5"); + expect(rowShell.className).not.toContain("[@media(pointer:coarse)]:pl-11"); + }); + + it("disables every handle while a queue mutation or cancellation is pending", () => { + const state = queueState([entry({ id: "q-1" }), entry({ id: "q-2", content: "second" })], { + isLoading: true, + cancellationPending: true, + }); + useQueueMock.mockReturnValue(state); + renderQueue({CHILD}); + fireEvent.click(screen.getByTestId(CHIP_ID)); + + const handles = screen.getAllByTestId(GRAB_HANDLE_ID) as HTMLButtonElement[]; + expect(handles).toHaveLength(2); + for (const handle of handles) expect(handle.disabled).toBe(true); + }); + + it("removes the handle while the row is being edited", () => { + const state = queueState([ + entry({ id: "q-1", queued_by: QUEUED_BY_USER }), + entry({ id: "q-2", content: "second" }), + ]); + useQueueMock.mockReturnValue(state); + renderQueue({CHILD}); + fireEvent.click(screen.getByTestId(CHIP_ID)); + + fireEvent.click(screen.getAllByTestId(EDIT_BUTTON_ID)[0]); + expect(screen.getByTestId("queue-edit-textarea")).toBeTruthy(); + expect(screen.getAllByTestId(GRAB_HANDLE_ID)).toHaveLength(1); + }); + + it("drags the last row onto the first and calls reorderEntries with the new order", async () => { + const state = queueState([ + entry({ id: "q-1", content: "first" }), + entry({ id: "q-2", content: "second" }), + entry({ id: "q-3", content: "third" }), + ]); + state.reorderEntries = vi.fn(async () => {}); + useQueueMock.mockReturnValue(state); + renderQueue({CHILD}); + fireEvent.click(screen.getByTestId(CHIP_ID)); + + const handles = screen.getAllByTestId(GRAB_HANDLE_ID); + simulateReorderDrag(handles[2], handles.length, 10); + + await waitFor(() => expect(state.reorderEntries).toHaveBeenCalledWith(["q-3", "q-1", "q-2"])); + }); +}); + +/** + * Drives a dnd-kit PointerSensor drag in happy-dom: mocks per-row droppable + * rects for collision detection, patches isPrimary (happy-dom omits the + * primary-pointer computation), and dispatches down / move / move / up. + * The first move only activates the drag (distance constraint); the second + * dispatches DragMove so collision detection resolves the target. + */ +function simulateReorderDrag(handle: HTMLElement, rowCount: number, targetClientY: number) { + const original = HTMLElement.prototype.getBoundingClientRect; + vi.spyOn(HTMLElement.prototype, "getBoundingClientRect").mockImplementation(function ( + this: HTMLElement, + ) { + const inner = this.querySelector?.('[data-testid="queue-entry"]') as HTMLElement | null; + const rows = screen.queryAllByTestId("queue-entry"); + const index = inner ? rows.indexOf(inner) : -1; + if (index >= 0) { + const top = index * 40; + return { + x: 0, + y: top, + top, + bottom: top + 36, + left: 0, + right: 400, + width: 400, + height: 36, + } as DOMRect; + } + return original.call(this); + }); + + // isPrimary is an own property created by the PointerEvent constructor + // (prototype patching is shadowed), so a capture-phase listener rewrites it + // before React's root handler runs. + const patchPrimary = (e: Event) => { + Object.defineProperty(e, "isPrimary", { configurable: true, value: true }); + }; + document.addEventListener("pointerdown", patchPrimary, true); + fireEvent.pointerDown(handle, { + pointerId: 0, + pointerType: "mouse", + button: 0, + clientX: 10, + clientY: (rowCount - 1) * 40 + 10, + }); + document.removeEventListener("pointerdown", patchPrimary, true); + fireEvent.pointerMove(document.body, { + pointerId: 0, + pointerType: "mouse", + clientX: 10, + clientY: targetClientY, + }); + fireEvent.pointerMove(document.body, { + pointerId: 0, + pointerType: "mouse", + clientX: 10, + clientY: targetClientY + 2, + }); + fireEvent.pointerUp(document.body, { + pointerId: 0, + pointerType: "mouse", + clientX: 10, + clientY: targetClientY + 2, + }); +} diff --git a/apps/web/components/task/chat/queued-ghost-list.tsx b/apps/web/components/task/chat/queued-ghost-list.tsx index 91fae873f2..5e976b6b27 100644 --- a/apps/web/components/task/chat/queued-ghost-list.tsx +++ b/apps/web/components/task/chat/queued-ghost-list.tsx @@ -1,8 +1,24 @@ "use client"; -import { useCallback, useEffect, useRef, useState } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import type { ReactNode } from "react"; import { IconLayoutList } from "@tabler/icons-react"; +import { + DndContext, + KeyboardSensor, + PointerSensor, + closestCenter, + useSensor, + useSensors, + type DragEndEvent, + type DragStartEvent, +} from "@dnd-kit/core"; +import { + SortableContext, + arrayMove, + sortableKeyboardCoordinates, + verticalListSortingStrategy, +} from "@dnd-kit/sortable"; import { toast } from "@/lib/toast/sonner"; import { useTranslation } from "react-i18next"; import { Collapsible, CollapsibleContent } from "@kandev/ui/collapsible"; @@ -12,6 +28,7 @@ import { stripSystemTags } from "@/lib/utils/system-tags"; import { MergeReferenceOverflowError, QueueEntryNotFoundError, + QueueReorderError, QueueSendNowError, } from "@/lib/api/domains/queue-api"; import { useQueue } from "@/hooks/domains/session/use-queue"; @@ -87,6 +104,7 @@ type QueuePanelHandlerArgs = { ) => Promise; removeEntry: (entryId: string) => Promise; mergeEntry: (entryId: string) => Promise; + reorderEntries: (orderedIds: string[]) => Promise; sendEntryNow: (entryId: string) => Promise; sendAllNow: () => Promise; }; @@ -139,6 +157,7 @@ function useQueuePanelHandlers({ editEntry, removeEntry, mergeEntry, + reorderEntries, sendEntryNow, sendAllNow, }: QueuePanelHandlerArgs) { @@ -204,6 +223,20 @@ function useQueuePanelHandlers({ toast.error(t("chat:failedToRunQueuedMessage")); }); }, [drainNext, t]); + const handleReorder = useCallback( + async (orderedIds: string[]) => { + try { + await reorderEntries(orderedIds); + } catch (err) { + // A queue_changed race already refetched and reconciled to the + // authoritative order, so it is a successful UI outcome — no toast. + if (err instanceof QueueReorderError) return; + console.error("Failed to reorder queued entries:", err); + toast.error(t("chat:failedToReorderQueuedMessages")); + } + }, + [reorderEntries, t], + ); return { handleSave, @@ -211,6 +244,7 @@ function useQueuePanelHandlers({ handleMerge, handleClear, handleDrain, + handleReorder, handleSendEntryNow, handleSendAllNow, }; @@ -234,6 +268,7 @@ type QueuePanelDisclosureProps = { onSave: (entryId: string, content: string, refs: EntityReference[]) => Promise; onRemove: (entryId: string) => Promise; onMerge: (entryId: string) => Promise; + onReorder: (orderedIds: string[]) => void; onSendEntryNow: (entryId: string) => void; }; @@ -256,6 +291,7 @@ function QueuePanelDisclosure({ onSave, onRemove, onMerge, + onReorder, onSendEntryNow, }: QueuePanelDisclosureProps) { return ( @@ -282,6 +318,7 @@ function QueuePanelDisclosure({ onSave={onSave} onRemove={onRemove} onMerge={onMerge} + onReorder={onReorder} onSendEntryNow={onSendEntryNow} /> @@ -330,6 +367,7 @@ export function QueueAffordance({ editEntry, removeEntry, mergeEntry, + reorderEntries, sendEntryNow, sendAllNow, cancellationPending, @@ -342,6 +380,7 @@ export function QueueAffordance({ handleMerge, handleClear, handleDrain, + handleReorder, handleSendEntryNow, handleSendAllNow, } = useQueuePanelHandlers({ @@ -350,6 +389,7 @@ export function QueueAffordance({ editEntry, removeEntry, mergeEntry, + reorderEntries, sendEntryNow, sendAllNow, }); @@ -402,6 +442,7 @@ export function QueueAffordance({ onSave={handleSave} onRemove={handleRemove} onMerge={handleMerge} + onReorder={handleReorder} onSendEntryNow={handleSendEntryNow} /> {!renderStatusBar && chipNode && ( @@ -481,11 +522,58 @@ type QueuePanelProps = { onSave: (entryId: string, content: string, entityReferences: EntityReference[]) => Promise; onRemove: (entryId: string) => Promise; onMerge: (entryId: string) => Promise; + onReorder: (orderedIds: string[]) => void; onSendEntryNow: (entryId: string) => void; }; /** Renders the expanded queue list: header controls plus one QueuedGhostMessage * row per pending entry, gating each row's merge control on `mergeEnabled`. */ +type QueueReorderArgs = { + entries: QueuedMessage[]; + canReorder: boolean; + onReorder: (orderedIds: string[]) => void; +}; + +/** Sortable-list state for the queue panel: sensors, the id list, and the + * drag lifecycle handlers that translate a drop into the new ordered ids. */ +function useQueueReorder({ entries, canReorder, onReorder }: QueueReorderArgs) { + const [activeId, setActiveId] = useState(null); + const sensors = useSensors( + useSensor(PointerSensor, { activationConstraint: { distance: 4 } }), + useSensor(KeyboardSensor, { coordinateGetter: sortableKeyboardCoordinates }), + ); + const ids = useMemo(() => entries.map((entry) => entry.id), [entries]); + + const handleDragStart = useCallback((event: DragStartEvent) => { + setActiveId(String(event.active.id)); + }, []); + + const handleDragEnd = useCallback( + (event: DragEndEvent) => { + setActiveId(null); + const { active, over } = event; + if (!over || active.id === over.id) return; + const oldIndex = entries.findIndex((entry) => entry.id === active.id); + const newIndex = entries.findIndex((entry) => entry.id === over.id); + if (oldIndex === -1 || newIndex === -1) return; + onReorder(arrayMove(ids, oldIndex, newIndex)); + }, + [entries, ids, onReorder], + ); + + const handleDragCancel = useCallback(() => setActiveId(null), []); + + return { + ids, + sensors, + activeId, + canReorder, + handleDragStart, + handleDragEnd, + handleDragCancel, + }; +} + function QueuePanel({ entries, count, @@ -502,9 +590,20 @@ function QueuePanel({ onSave, onRemove, onMerge, + onReorder, onSendEntryNow, }: QueuePanelProps) { const { t } = useTranslation(); + // Reordering is disabled while a queue mutation or backend cancellation is + // in flight, matching the Send Now gate; the server stays authoritative, so + // the optimistic order is reconciled by the refetch after the drop. + const { ids, sensors, activeId, canReorder, handleDragStart, handleDragEnd, handleDragCancel } = + useQueueReorder({ + entries, + canReorder: !isLoading && !cancellationPending, + onReorder, + }); + return (
- {entries.map((entry, index) => ( - onSave(entry.id, content, entityReferences)} - onRemove={() => onRemove(entry.id)} - onMerge={() => onMerge(entry.id)} - onSendNow={() => onSendEntryNow(entry.id)} - sendNowDisabled={isLoading || cancellationPending} - /> - ))} + + + {entries.map((entry, index) => ( + 1} + isDragging={activeId === entry.id} + onSave={(content, entityReferences) => onSave(entry.id, content, entityReferences)} + onRemove={() => onRemove(entry.id)} + onMerge={() => onMerge(entry.id)} + onSendNow={() => onSendEntryNow(entry.id)} + sendNowDisabled={isLoading || cancellationPending} + /> + ))} + +
); 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 fa99e3d7ae..6b96e02076 100644 --- a/apps/web/components/task/chat/queued-ghost-message.test.tsx +++ b/apps/web/components/task/chat/queued-ghost-message.test.tsx @@ -81,6 +81,102 @@ function renderWithProviders(node: React.ReactNode) { ); } +describe("QueuedGhostMessage reorder handle", () => { + const HANDLE_TESTID = "queue-grab-handle"; + + function renderRow( + overrides: Partial = {}, + props: Partial> = {}, + ) { + return render( + , + ); + } + + it("renders a dotted grab handle with localized labels", () => { + renderRow(); + const handle = screen.getByTestId(HANDLE_TESTID); + expect(handle.getAttribute("aria-label")).toBe("Reorder queued message"); + expect(handle.getAttribute("aria-roledescription")).toBe("sortable"); + // 2x3 dot grid reads as a dotted grip. + expect(handle.querySelectorAll('[aria-hidden="true"] span').length).toBe(6); + }); + + it("shows the handle without hovering on a fine pointer", () => { + renderRow(); + const handle = screen.getByTestId(HANDLE_TESTID); + // A mounted handle is an always-visible drag affordance on every pointer. + expect(handle.className).toContain("opacity-100"); + expect(handle.className).not.toContain("opacity-0"); + expect(handle.className).not.toContain("group-hover:opacity-100"); + }); + + it("keeps the fine-pointer grip small and unobtrusive", () => { + renderRow(); + const handle = screen.getByTestId(HANDLE_TESTID); + expect(handle.className).toContain("left-1"); + expect(handle.className).toContain("h-3 w-3"); + expect(handle.className).toContain("border-0"); + expect(handle.className).toContain("bg-transparent"); + expect(handle.className).toContain("shadow-none"); + const grip = handle.querySelector('[aria-hidden="true"]')!; + expect(grip.className).toContain("gap-0.5"); + const dot = handle.querySelector('[aria-hidden="true"] span')!; + expect(dot.className).toContain("h-px w-px"); + }); + + it("attaches the coarse-pointer handle to the box edge in the background", () => { + renderRow(); + const handle = screen.getByTestId(HANDLE_TESTID); + // Coarse pointers: flush at the left edge, behind the content (z-0, no + // chip surface) so the dots read as part of the box edge. + expect(handle.className).toContain("[@media(pointer:coarse)]:left-0"); + expect(handle.className).toContain("[@media(pointer:coarse)]:z-0"); + expect(handle.className).toContain("[@media(pointer:coarse)]:bg-transparent"); + expect(handle.className).toContain("[@media(pointer:coarse)]:shadow-none"); + expect(handle.className).toContain("[@media(pointer:coarse)]:h-11"); + expect(handle.className).toContain("[@media(pointer:coarse)]:w-11"); + }); + + it("keeps the position label painted above the handle and pointer-transparent", () => { + renderRow(); + // The label/icon cluster sits above the background handle (relative) and + // passes touches through to it (pointer-events-none). + const label = screen.getByLabelText("Position #1"); + const cluster = label.parentElement!; + expect(cluster.className).toContain("relative"); + expect(cluster.className).toContain("pointer-events-none"); + }); + + it("centers the position, sender icon, and message content", () => { + renderRow(); + const row = screen.getByTestId("queue-entry"); + expect(row.className).toContain("items-center"); + expect(row.className).not.toContain("items-start"); + const label = screen.getByLabelText("Position #1"); + expect(label.parentElement!.className).not.toContain("mt-0.5"); + }); + + it("disables the handle while reordering is unavailable", () => { + renderRow({}, { canDrag: false }); + const handle = screen.getByTestId(HANDLE_TESTID) as HTMLButtonElement; + expect(handle.disabled).toBe(true); + expect(handle.className).toContain("cursor-not-allowed"); + }); + + it("hides the handle while the row is being edited", () => { + renderRow({ queued_by: "user-1" }, { canEdit: true }); + fireEvent.click(screen.getByTestId(EDIT_TESTID)); + expect(screen.queryByTestId(HANDLE_TESTID)).toBeNull(); + }); +}); + describe("QueuedGhostMessage workflow badge", () => { it("renders workflow metadata as a workflow step badge", () => { render( diff --git a/apps/web/components/task/chat/queued-ghost-message.tsx b/apps/web/components/task/chat/queued-ghost-message.tsx index 5d57ebf650..37e6861672 100644 --- a/apps/web/components/task/chat/queued-ghost-message.tsx +++ b/apps/web/components/task/chat/queued-ghost-message.tsx @@ -9,8 +9,12 @@ import { useRef, useState, } from "react"; +import type { ReactNode } from "react"; import { IconCheck, IconFile, IconInfoCircle, IconRobot, IconUser } from "@tabler/icons-react"; import ReactMarkdown from "react-markdown"; +import { useSortable } from "@dnd-kit/sortable"; +import { CSS } from "@dnd-kit/utilities"; +import type { DraggableAttributes, DraggableSyntheticListeners } from "@dnd-kit/core"; import { toast } from "@/lib/toast/sonner"; import { useTranslation } from "react-i18next"; import { Button } from "@kandev/ui"; @@ -335,8 +339,11 @@ function DisplayView({ [entityReferences], ); return ( -
- +
+ {/* Position label + sender icon are non-interactive; relative keeps them + * painted above the coarse-pointer touch target, and pointer-events-none + * lets touches reach the handle behind them. */} + void; +}; + +/** + * Dotted drag handle that floats over the row's left edge. It is absolutely + * positioned so nothing in the row shifts to make room. It is always visible + * when the queue has multiple entries, attached flush to the box's left edge + * and painted behind the row content on coarse pointers (no chip surface — + * the dots read as part of the box edge). + * Dragging starts only from this handle. + */ +function QueueGrabHandle({ + canDrag, + attributes, + listeners, + setActivatorNodeRef, +}: QueueGrabHandleProps) { + const { t } = useTranslation(); + const dragProps = canDrag ? { ...attributes, ...listeners } : {}; + return ( + + ); +} + +type SortableRowShellProps = { + id: string; + /** useSortable is disabled while the row is edited or reordering is off. */ + disabled: boolean; + canDrag: boolean; + /** The grab handle is hidden while the row is being edited. */ + showHandle: boolean; + isDragging: boolean; + children: ReactNode; +}; + +/** Sortable row wrapper: owns the dnd-kit node ref, transform, and the dotted + * grab handle overlay, keeping QueuedGhostMessage focused on content. */ +function SortableRowShell({ + id, + disabled, + canDrag, + showHandle, + isDragging, + children, +}: SortableRowShellProps) { + const { attributes, listeners, setActivatorNodeRef, setNodeRef, transform, transition } = + useSortable({ + id, + disabled, + }); + return ( +
+ {showHandle && ( + + )} + {children} +
+ ); +} + type QueuedGhostMessageProps = { entry: QueuedMessage; /** Zero-based render position; combined with entry.position to label the row. */ @@ -399,6 +517,17 @@ type QueuedGhostMessageProps = { * share a sender kind. Optional so standalone renders (tests) can omit it. */ canMerge?: boolean; + /** + * Reordering is enabled unless a queue mutation or backend cancellation is + * in flight; the row's own edit view also suppresses the handle. The queue + * panel hides it when there is only one entry. Optional so standalone renders + * default to showing the handle. + */ + canDrag?: boolean; + /** Whether this row belongs to a queue with enough entries to reorder. */ + showDragHandle?: boolean; + /** Set while this row is the active drag target, for dimming/z-index. */ + isDragging?: boolean; onSave: (content: string, entityReferences: EntityReference[]) => Promise; onRemove: () => void | Promise; /** Fold this entry into the one above it. */ @@ -479,6 +608,9 @@ export const QueuedGhostMessage = forwardRef {editing ? ( )} -
+ ); }, ); diff --git a/apps/web/e2e/tests/chat/message-queue-reorder.spec.ts b/apps/web/e2e/tests/chat/message-queue-reorder.spec.ts new file mode 100644 index 0000000000..3f4832fde0 --- /dev/null +++ b/apps/web/e2e/tests/chat/message-queue-reorder.spec.ts @@ -0,0 +1,192 @@ +import { expect, type Locator, type Page } from "@playwright/test"; +import { test } from "../../fixtures/test-base"; +import type { SeedData } from "../../fixtures/test-base"; +import type { ApiClient } from "../../helpers/api-client"; +import { typeWhileBusy } from "../../helpers/type-while-busy"; +import { SessionPage } from "../../pages/session-page"; + +/** Create a task whose agent finishes its intro turn and becomes idle. */ +async function seedIdleTask( + testPage: Page, + apiClient: ApiClient, + seedData: SeedData, + title: string, +): Promise { + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + title, + seedData.agentProfileId, + { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await session.waitForChatIdle({ timeout: 60_000 }); + return session; +} + +/** Keep the agent busy with a slow command, then queue one message per entry. */ +async function queueMessagesWhileBusy( + page: Page, + editor: Locator, + messages: string[], +): Promise { + const submit = page.getByTestId("submit-message-button"); + for (const message of messages) { + await typeWhileBusy(page, editor, message); + await expect(submit).toBeVisible({ timeout: 5_000 }); + await submit.click(); + } +} + +async function openQueuePanel(session: SessionPage): Promise { + const chip = session.activeChat().getByTestId("queue-chip"); + await expect(chip).toBeVisible({ timeout: 10_000 }); + await chip.click(); + const panel = session.activeChat().getByTestId("queued-ghost-list"); + await expect(panel).toBeVisible({ timeout: 5_000 }); + return panel; +} + +/** Grab handle of the row whose visible text contains `text`. */ +function rowHandle(panel: Locator, text: string): Locator { + const row = panel.getByTestId("queue-entry").filter({ hasText: text }); + return row.locator("xpath=..").getByTestId("queue-grab-handle"); +} + +/** + * Assert that `contents` occupy exactly the consecutive positions starting at + * the head after removing the `/slow` keep-busy command, and that the `/slow` + * row — when present — sits at the head. This pins the global order instead of + * normalizing it away: a foreign row interleaved between the expected contents + * or a `/slow` row anywhere but the head fails. + */ +async function expectRelativeOrder(panel: Locator, ...contents: string[]): Promise { + await expect + .poll( + async () => { + const texts = (await panel.getByTestId("queue-entry-text").allTextContents()).map((t) => + t.replace(/\s+/g, " ").trim(), + ); + const slowIndex = texts.findIndex((t) => t.startsWith("/slow")); + const withoutSlow = texts.filter((_, index) => index !== slowIndex); + return { + positions: contents.map((content) => withoutSlow.findIndex((t) => t.includes(content))), + slowHead: slowIndex === -1 || slowIndex === 0, + }; + }, + { timeout: 10_000 }, + ) + .toEqual({ positions: contents.map((_, index) => index), slowHead: true }); +} + +test.describe("Queue reorder", () => { + test.describe.configure({ retries: 1 }); + + test("drag-and-drop reorders queued messages and persists the order", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(120_000); + + const session = await seedIdleTask(testPage, apiClient, seedData, "Queue reorder drag"); + await session.sendMessage("/slow 30s"); + await expect(session.agentStatus()).toBeVisible({ timeout: 15_000 }); + await testPage.waitForTimeout(500); + + const editor = session.activeChat().locator(".tiptap.ProseMirror:visible").first(); + await queueMessagesWhileBusy(testPage, editor, [ + "reorder first", + "reorder second", + "reorder third", + ]); + + const panel = await openQueuePanel(session); + await expectRelativeOrder(panel, "reorder first", "reorder second", "reorder third"); + + // Multiple queued rows expose their dotted grab handles immediately on a + // fine pointer, so the drag affordance is discoverable without hover. + const firstHandle = rowHandle(panel, "reorder first"); + await expect(firstHandle).toHaveCSS("opacity", "1", { timeout: 5_000 }); + const firstContent = panel.getByTestId("queue-entry").filter({ hasText: "reorder first" }); + const [handleBox, contentBox] = await Promise.all([ + firstHandle.boundingBox(), + firstContent.boundingBox(), + ]); + expect(handleBox).not.toBeNull(); + expect(contentBox).not.toBeNull(); + expect(handleBox!.width).toBeLessThanOrEqual(12); + expect(contentBox!.x).toBeGreaterThanOrEqual(handleBox!.x + handleBox!.width + 4); + + // Drag the third row's handle onto the first row. Playwright's dragTo + // fires a single mousemove, which only activates dnd-kit's PointerSensor + // (distance constraint) without dispatching DragMove — the drop would + // resolve no target. Stepped moves keep the sensor's move events flowing. + const sourceBox = (await rowHandle(panel, "reorder third").boundingBox())!; + const targetBox = (await firstHandle.boundingBox())!; + await testPage.mouse.move( + sourceBox.x + sourceBox.width / 2, + sourceBox.y + sourceBox.height / 2, + ); + await testPage.mouse.down(); + await testPage.mouse.move( + targetBox.x + targetBox.width / 2, + targetBox.y + targetBox.height / 2, + { steps: 12 }, + ); + await testPage.mouse.up(); + await expectRelativeOrder(panel, "reorder third", "reorder first", "reorder second"); + + // The order is persisted server-side and survives a reload. + await testPage.reload(); + await session.waitForLoad(); + const panelAfter = await openQueuePanel(session); + await expectRelativeOrder(panelAfter, "reorder third", "reorder first", "reorder second"); + }); + + test("keyboard reordering moves a queued message one position", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(120_000); + + const session = await seedIdleTask(testPage, apiClient, seedData, "Queue reorder keyboard"); + await session.sendMessage("/slow 30s"); + await expect(session.agentStatus()).toBeVisible({ timeout: 15_000 }); + await testPage.waitForTimeout(500); + + const editor = session.activeChat().locator(".tiptap.ProseMirror:visible").first(); + await queueMessagesWhileBusy(testPage, editor, [ + "reorder first", + "reorder second", + "reorder third", + ]); + + const panel = await openQueuePanel(session); + await expectRelativeOrder(panel, "reorder first", "reorder second"); + + // dnd-kit's keyboard sensor: Space starts the drag, arrows move, Space + // drops. Wait for the drag to actually start (the row dims) before moving. + const handle = rowHandle(panel, "reorder first"); + const row = handle.locator("xpath=.."); + await handle.focus(); + await testPage.keyboard.press("Space"); + await expect(row).toHaveClass(/opacity-40/, { timeout: 5_000 }); + // Droppable measuring happens asynchronously after the drag starts; an + // arrow press before it completes computes no target. Settle briefly. + await testPage.waitForTimeout(200); + await testPage.keyboard.press("ArrowDown"); + await testPage.waitForTimeout(100); + await testPage.keyboard.press("Space"); + await expect(row).not.toHaveClass(/opacity-40/, { timeout: 5_000 }); + + await expectRelativeOrder(panel, "reorder second", "reorder first"); + }); +}); diff --git a/apps/web/e2e/tests/chat/mobile-message-queue-reorder.spec.ts b/apps/web/e2e/tests/chat/mobile-message-queue-reorder.spec.ts new file mode 100644 index 0000000000..8ae65f9633 --- /dev/null +++ b/apps/web/e2e/tests/chat/mobile-message-queue-reorder.spec.ts @@ -0,0 +1,166 @@ +import { expect, type Locator, type Page } from "@playwright/test"; +import { test } from "../../fixtures/test-base"; +import type { SeedData } from "../../fixtures/test-base"; +import type { ApiClient } from "../../helpers/api-client"; +import { typeWhileBusy } from "../../helpers/type-while-busy"; +import { SessionPage } from "../../pages/session-page"; + +async function expectTouchTarget(locator: Locator): Promise { + await expect(locator).toBeVisible(); + const box = await locator.boundingBox(); + expect(box).not.toBeNull(); + // Chromium can report a 44px CSS target a fraction below 44 after layout. + expect(box!.width).toBeGreaterThanOrEqual(43.5); + expect(box!.height).toBeGreaterThanOrEqual(43.5); +} + +async function seedBusyQueueTask( + testPage: Page, + apiClient: ApiClient, + seedData: SeedData, +): Promise { + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Mobile queue reorder", + seedData.agentProfileId, + { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await session.waitForChatIdle({ timeout: 60_000 }); + await session.sendMessageViaButton("/slow 30s"); + await session.agentStatus().waitFor({ state: "visible", timeout: 15_000 }); + await testPage.waitForTimeout(500); + return session; +} + +/** Grab handle of the row whose visible text contains `text`. */ +function rowHandle(panel: Locator, text: string): Locator { + const row = panel.getByTestId("queue-entry").filter({ hasText: text }); + return row.locator("xpath=..").getByTestId("queue-grab-handle"); +} + +/** + * Assert that `contents` occupy exactly the consecutive positions starting at + * the head after removing the `/slow` keep-busy command, and that the `/slow` + * row — when present — sits at the head. This pins the global order instead of + * normalizing it away: a foreign row interleaved between the expected contents + * or a `/slow` row anywhere but the head fails. + */ +async function expectRelativeOrder(panel: Locator, ...contents: string[]): Promise { + await expect + .poll( + async () => { + const texts = (await panel.getByTestId("queue-entry-text").allTextContents()).map((t) => + t.replace(/\s+/g, " ").trim(), + ); + const slowIndex = texts.findIndex((t) => t.startsWith("/slow")); + const withoutSlow = texts.filter((_, index) => index !== slowIndex); + return { + positions: contents.map((content) => withoutSlow.findIndex((t) => t.includes(content))), + slowHead: slowIndex === -1 || slowIndex === 0, + }; + }, + { timeout: 10_000 }, + ) + .toEqual({ positions: contents.map((_, index) => index), slowHead: true }); +} + +/** + * Touch-pointer drag for dnd-kit's PointerSensor: pointerdown on the handle, + * two pointermoves (the first only activates the drag, the second dispatches + * DragMove so collision detection resolves), then pointerup. isPrimary is set + * explicitly because Playwright's dispatchEvent constructs untrusted events. + */ +async function touchDragTo( + page: Page, + source: Locator, + targetBox: { x: number; y: number }, +): Promise { + const sourceBox = (await source.boundingBox())!; + const start = { x: sourceBox.x + sourceBox.width / 2, y: sourceBox.y + sourceBox.height / 2 }; + const end = { x: targetBox.x, y: targetBox.y }; + const pointerId = 5; + const down = { + pointerId, + pointerType: "touch", + isPrimary: true, + button: 0, + clientX: start.x, + clientY: start.y, + }; + await source.dispatchEvent("pointerdown", down); + // Each event is a separate task; the gaps let React commit the drag start + // and dnd-kit finish droppable measuring before the move dispatches + // DragMove — otherwise the drop resolves no target. + await page.waitForTimeout(100); + await page.dispatchEvent("body", "pointermove", { ...down, clientX: end.x, clientY: end.y }); + await page.waitForTimeout(100); + await page.dispatchEvent("body", "pointermove", { + ...down, + clientX: end.x, + clientY: end.y + 2, + }); + await page.waitForTimeout(100); + await page.dispatchEvent("body", "pointerup", { + ...down, + clientX: end.x, + clientY: end.y + 2, + }); +} + +test.describe.configure({ retries: 1 }); + +test("mobile touch drag reorders queued messages with an always-visible handle", async ({ + testPage, + apiClient, + seedData, +}) => { + test.setTimeout(120_000); + + const session = await seedBusyQueueTask(testPage, apiClient, seedData); + const chat = session.activeChat(); + const editor = chat.locator(".tiptap.ProseMirror:visible").first(); + const submit = testPage.getByTestId("submit-message-button"); + for (const message of ["reorder first", "reorder second", "reorder third"]) { + await typeWhileBusy(testPage, editor, message); + await expect(submit).toBeVisible({ timeout: 5_000 }); + await submit.tap(); + } + + const chip = chat.getByTestId("queue-chip"); + await expect(chip).toBeVisible({ timeout: 10_000 }); + await chip.tap(); + const panel = chat.getByTestId("queued-ghost-list"); + await expect(panel).toBeVisible({ timeout: 5_000 }); + await expectRelativeOrder(panel, "reorder first", "reorder second", "reorder third"); + + // No hover on touch: the handle is always visible and touch-sized. + const firstHandle = rowHandle(panel, "reorder first"); + const lastHandle = rowHandle(panel, "reorder third"); + await expect(firstHandle).toHaveCSS("opacity", "1"); + await expectTouchTarget(lastHandle); + + const firstBox = (await firstHandle.boundingBox())!; + await touchDragTo(testPage, lastHandle, { + x: firstBox.x + firstBox.width / 2, + y: firstBox.y + firstBox.height / 2, + }); + + await expectRelativeOrder(panel, "reorder third", "reorder first", "reorder second"); + + // The reorder is persisted server-side and visible after a reload. + await testPage.reload(); + await session.waitForLoad(); + await expect(chat.getByTestId("queue-chip")).toBeVisible({ timeout: 10_000 }); + await chat.getByTestId("queue-chip").tap(); + const panelAfter = chat.getByTestId("queued-ghost-list"); + await expect(panelAfter).toBeVisible({ timeout: 5_000 }); + await expectRelativeOrder(panelAfter, "reorder third", "reorder first", "reorder second"); +}); diff --git a/apps/web/hooks/domains/session/use-queue.test.ts b/apps/web/hooks/domains/session/use-queue.test.ts index 241f378cfd..ac0b061147 100644 --- a/apps/web/hooks/domains/session/use-queue.test.ts +++ b/apps/web/hooks/domains/session/use-queue.test.ts @@ -6,9 +6,11 @@ import type { EntityReference } from "@/lib/types/entity-reference"; const queueApiMock = vi.hoisted(() => { class QueueEntryNotFoundError extends Error {} class QueueSendNowError extends Error {} + class QueueReorderError extends Error {} return { QueueEntryNotFoundError, QueueSendNowError, + QueueReorderError, queueMessage: vi.fn(), clearQueue: vi.fn(), drainQueuedMessage: vi.fn(), @@ -16,6 +18,7 @@ const queueApiMock = vi.hoisted(() => { updateQueuedMessage: vi.fn(), removeQueuedEntry: vi.fn(), mergeQueuedEntry: vi.fn(), + reorderQueuedEntries: vi.fn(), sendQueuedNow: vi.fn(), }; }); @@ -497,3 +500,94 @@ describe("useQueue removeEntry", () => { }); }); }); + +describe("useQueue reorderEntries", () => { + beforeEach(() => { + resetMockState(); + setDocumentVisibility("visible"); + queueApiMock.reorderQueuedEntries.mockReset(); + queueApiMock.getQueueStatus.mockResolvedValue({ entries: [], count: 0, max: 10 }); + }); + + it("optimistically reorders the store and refetches the authoritative queue", async () => { + const first = entry({ id: "q-1", content: "first" }); + const second = entry({ id: "q-2", content: "second" }); + // The test harness's setQueueEntries is a no-op spy, so seed the store + // directly: the hook's optimistic reorder reads the current entries. + mockState.queue.bySessionId[SESSION_ID] = [first, second]; + mockState.queue.metaBySessionId[SESSION_ID] = { count: 2, max: 10 }; + queueApiMock.getQueueStatus.mockResolvedValue({ + entries: [first, second], + count: 2, + max: 10, + }); + queueApiMock.reorderQueuedEntries.mockResolvedValue({ + session_id: SESSION_ID, + reordered: 2, + }); + + const { result } = renderHook(() => useQueue(SESSION_ID)); + await waitFor(() => expect(queueApiMock.getQueueStatus).toHaveBeenCalled()); + queueApiMock.getQueueStatus.mockClear(); + + await act(async () => { + await result.current.reorderEntries([second.id, first.id]); + }); + + expect(queueApiMock.reorderQueuedEntries).toHaveBeenCalledWith({ + session_id: SESSION_ID, + ordered_ids: [second.id, first.id], + }); + expect(mockState.setQueueEntries).toHaveBeenCalledWith(SESSION_ID, [second, first], { + count: 2, + max: 10, + mergeEnabled: true, + }); + expect(queueApiMock.getQueueStatus).toHaveBeenCalledWith(SESSION_ID); + }); + + it("refetches and rethrows when the reorder fails", async () => { + const first = entry({ id: "q-1", content: "first" }); + const second = entry({ id: "q-2", content: "second" }); + queueApiMock.getQueueStatus.mockResolvedValue({ + entries: [first, second], + count: 2, + max: 10, + }); + queueApiMock.reorderQueuedEntries.mockRejectedValueOnce(new Error("reorder failed")); + const { result } = renderHook(() => useQueue(SESSION_ID)); + await waitFor(() => expect(queueApiMock.getQueueStatus).toHaveBeenCalled()); + queueApiMock.getQueueStatus.mockClear(); + queueApiMock.getQueueStatus.mockResolvedValueOnce({ + entries: [first, second], + count: 2, + max: 10, + }); + + await act(async () => { + await expect(result.current.reorderEntries([second.id, first.id])).rejects.toThrow( + "reorder failed", + ); + }); + + expect(queueApiMock.getQueueStatus).toHaveBeenCalledWith(SESSION_ID); + }); + + it("refetches and rethrows a QueueReorderError so the panel can swallow it", async () => { + const first = entry({ id: "q-1", content: "first" }); + queueApiMock.getQueueStatus.mockResolvedValue({ entries: [first], count: 1, max: 10 }); + queueApiMock.reorderQueuedEntries.mockRejectedValueOnce(new queueApiMock.QueueReorderError()); + + const { result } = renderHook(() => useQueue(SESSION_ID)); + await waitFor(() => expect(queueApiMock.getQueueStatus).toHaveBeenCalled()); + queueApiMock.getQueueStatus.mockClear(); + + await act(async () => { + await expect(result.current.reorderEntries([first.id])).rejects.toThrow( + queueApiMock.QueueReorderError, + ); + }); + + expect(queueApiMock.getQueueStatus).toHaveBeenCalledWith(SESSION_ID); + }); +}); diff --git a/apps/web/hooks/domains/session/use-queue.ts b/apps/web/hooks/domains/session/use-queue.ts index ea345146b6..5ee2e39990 100644 --- a/apps/web/hooks/domains/session/use-queue.ts +++ b/apps/web/hooks/domains/session/use-queue.ts @@ -9,6 +9,7 @@ import { updateQueuedMessage, removeQueuedEntry, mergeQueuedEntry, + reorderQueuedEntries, QueueEntryNotFoundError, sendQueuedNow, } from "@/lib/api/domains/queue-api"; @@ -69,6 +70,7 @@ function useQueueState(sessionId: string | null) { type QueueActionsArgs = { sessionId: string | null; + entries: QueuedMessage[]; setQueueEntries: ReturnType["setQueueEntries"]; removeQueueEntry: ReturnType["removeQueueEntry"]; setQueueLoading: ReturnType["setQueueLoading"]; @@ -167,6 +169,7 @@ function useQueueRefetch( /** Build an action set bound to the supplied session + slice setters. */ function useQueueActions({ sessionId, + entries, setQueueEntries, removeQueueEntry, setQueueLoading, @@ -209,7 +212,85 @@ function useQueueActions({ [sessionId, refetch, setQueueLoading], ); - const clearAll = useCallback(async () => { + const clearAll = useClearAllAction({ + sessionId, + setQueueEntries, + setQueueLoading, + metaMax, + metaMergeEnabled, + refetch, + invalidateRefetch, + }); + + const drainNext = useDrainNextAction(sessionId, setQueueLoading, refetch); + + const sendNow = useSendNowAction(sessionId, setQueueLoading, refetch); + + const sendEntryNow = useCallback((entryId: string) => sendNow("entry", entryId), [sendNow]); + const sendAllNow = useCallback(() => sendNow("all"), [sendNow]); + + const { editEntry, removeEntry, mergeEntry } = useEntryMutations({ + sessionId, + removeQueueEntry, + refetch, + }); + + const reorderEntries = useReorderEntriesAction({ + sessionId, + entries, + setQueueEntries, + setQueueLoading, + metaMax, + metaMergeEnabled, + refetch, + }); + + return { + refetch, + queue, + clearAll, + drainNext, + sendEntryNow, + sendAllNow, + editEntry, + removeEntry, + mergeEntry, + reorderEntries, + }; +} + +type ReorderEntriesArgs = { + sessionId: string | null; + entries: QueuedMessage[]; + setQueueEntries: ReturnType["setQueueEntries"]; + setQueueLoading: ReturnType["setQueueLoading"]; + metaMax: number | undefined; + metaMergeEnabled: boolean | undefined; + refetch: (sid: string) => Promise; +}; + +type ClearAllArgs = { + sessionId: string | null; + setQueueEntries: ReturnType["setQueueEntries"]; + setQueueLoading: ReturnType["setQueueLoading"]; + metaMax: number | undefined; + metaMergeEnabled: boolean | undefined; + refetch: (sid: string) => Promise; + invalidateRefetch: (sid: string) => void; +}; + +/** Clears every pending entry: optimistic empty state, then the backend's + * authoritative result. Reconcile errors preserve the mutation error. */ +function useClearAllAction({ + sessionId, + setQueueEntries, + setQueueLoading, + metaMax, + metaMergeEnabled, + refetch, + invalidateRefetch, +}: ClearAllArgs) { + return useCallback(async () => { if (!sessionId) return; setQueueLoading(sessionId, true); let mutationFailed = false; @@ -252,31 +333,63 @@ function useQueueActions({ refetch, invalidateRefetch, ]); +} - const drainNext = useDrainNextAction(sessionId, setQueueLoading, refetch); +/** Rewrites the visible pending order, optimistically at first, then + * reconciles to the backend's authoritative order. A queue_changed drift + * (drain/remove/merge raced the drag) is refetched and rethrown as + * QueueReorderError so the panel can swallow it without a toast. */ +function useReorderEntriesAction({ + sessionId, + entries, + setQueueEntries, + setQueueLoading, + metaMax, + metaMergeEnabled, + refetch, +}: ReorderEntriesArgs) { + return useCallback( + async (orderedIds: string[]) => { + if (!sessionId) return; + // Optimistic: reflect the drop immediately; the refetch below reconciles + // to the backend's authoritative order and positions. + const byId = new Map(entries.map((entry) => [entry.id, entry])); + const reordered = orderedIds + .map((id) => byId.get(id)) + .filter((entry): entry is QueuedMessage => entry !== undefined); + if (reordered.length === entries.length) { + setQueueEntries(sessionId, reordered, { + count: reordered.length, + max: metaMax ?? 0, + mergeEnabled: metaMergeEnabled ?? true, + }); + } + setQueueLoading(sessionId, true); + let mutationFailed = false; + let mutationError: unknown; + try { + try { + await reorderQueuedEntries({ session_id: sessionId, ordered_ids: orderedIds }); + } catch (err) { + mutationFailed = true; + mutationError = err; + } - const sendNow = useSendNowAction(sessionId, setQueueLoading, refetch); - - const sendEntryNow = useCallback((entryId: string) => sendNow("entry", entryId), [sendNow]); - const sendAllNow = useCallback(() => sendNow("all"), [sendNow]); - - const { editEntry, removeEntry, mergeEntry } = useEntryMutations({ - sessionId, - removeQueueEntry, - refetch, - }); - - return { - refetch, - queue, - clearAll, - drainNext, - sendEntryNow, - sendAllNow, - editEntry, - removeEntry, - mergeEntry, - }; + try { + await refetch(sessionId); + } catch (reconcileError) { + // Preserve the mutation error when both operations fail; it best + // explains why the requested order did not land. + if (mutationFailed) throw mutationError; + throw reconcileError; + } + if (mutationFailed) throw mutationError; + } finally { + setQueueLoading(sessionId, false); + } + }, + [sessionId, entries, setQueueEntries, setQueueLoading, metaMax, refetch], + ); } type EntryMutationsArgs = { @@ -388,8 +501,10 @@ export function useQueue(sessionId: string | null) { editEntry, removeEntry, mergeEntry, + reorderEntries, } = useQueueActions({ sessionId, + entries, setQueueEntries: state.setQueueEntries, removeQueueEntry: state.removeQueueEntry, setQueueLoading: state.setQueueLoading, @@ -438,6 +553,7 @@ export function useQueue(sessionId: string | null) { editEntry, removeEntry, mergeEntry, + reorderEntries, 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 2a24a453e2..94a14496d2 100644 --- a/apps/web/lib/api/domains/queue-api.test.ts +++ b/apps/web/lib/api/domains/queue-api.test.ts @@ -12,8 +12,10 @@ import { QueueSendNowError, QueueFullError, QueueEntryNotFoundError, + QueueReorderError, mergeQueuedEntry, queueMessage, + reorderQueuedEntries, rethrowQueueError, sendQueuedNow, updateQueuedMessage, @@ -308,3 +310,41 @@ describe("sendQueuedNow", () => { ); }); }); + +describe("reorderQueuedEntries", () => { + it("sends the full ordered id list through message.queue.reorder", async () => { + const request = vi.fn().mockResolvedValue({ session_id: "session-1", reordered: 3 }); + getWebSocketClientMock.mockReturnValue({ request }); + + await reorderQueuedEntries({ + session_id: "session-1", + ordered_ids: ["q-c", "q-a", "q-b"], + }); + + expect(request).toHaveBeenCalledWith("message.queue.reorder", { + session_id: "session-1", + ordered_ids: ["q-c", "q-a", "q-b"], + }); + }); + + it("maps a queue_changed rejection to QueueReorderError", async () => { + const request = vi.fn().mockRejectedValue({ + code: "queue_changed", + message: "Queue changed before the reorder could be applied", + }); + getWebSocketClientMock.mockReturnValue({ request }); + + await expect( + reorderQueuedEntries({ session_id: "session-1", ordered_ids: ["q-a"] }), + ).rejects.toBeInstanceOf(QueueReorderError); + }); + + it("falls through to the shared error mapping for other failures", async () => { + const request = vi.fn().mockRejectedValue({ code: "entry_not_found", message: "Gone" }); + getWebSocketClientMock.mockReturnValue({ request }); + + await expect( + reorderQueuedEntries({ session_id: "session-1", ordered_ids: ["q-a"] }), + ).rejects.toBeInstanceOf(QueueEntryNotFoundError); + }); +}); diff --git a/apps/web/lib/api/domains/queue-api.ts b/apps/web/lib/api/domains/queue-api.ts index 06c2036a80..bb3c8654a7 100644 --- a/apps/web/lib/api/domains/queue-api.ts +++ b/apps/web/lib/api/domains/queue-api.ts @@ -66,6 +66,19 @@ export class QueueSendNowError extends Error { } } +/** Error thrown when a reorder's submitted id set no longer matches the + * visible pending queue (an entry was drained, removed, merged, or newly + * queued since the client's snapshot). The reorder was rejected atomically; + * callers refetch the authoritative queue. */ +export class QueueReorderError extends Error { + readonly code = "queue_changed" as const; + + constructor(message?: string) { + super(message ?? "The queue changed before the reorder could be applied."); + this.name = "QueueReorderError"; + } +} + type WSError = { code?: string; message?: string; @@ -286,3 +299,29 @@ export async function mergeQueuedEntry(params: { rethrowQueueError(err); } } + +/** Rewrite the visible pending order of a session's queue. Throws + * QueueReorderError when the queue changed since the client's snapshot; the + * reorder was rejected atomically and the caller should refetch. */ +export async function reorderQueuedEntries(params: { + session_id: string; + ordered_ids: string[]; +}): Promise<{ session_id: string; reordered: number }> { + const client = getWebSocketClient(); + if (!client) { + throw new Error(WS_CLIENT_UNAVAILABLE); + } + try { + return await client.request<{ session_id: string; reordered: number }>( + "message.queue.reorder", + params, + ); + } catch (err) { + // The shared queue_changed mapping targets Send Now; reorder drift needs + // its own error class so callers reconcile silently instead of toasting. + if (asWSError(err)?.code === "queue_changed") { + throw new QueueReorderError(); + } + rethrowQueueError(err); + } +} diff --git a/apps/web/src/locales/en/chat.json b/apps/web/src/locales/en/chat.json index a939a946bf..201e9eb95b 100644 --- a/apps/web/src/locales/en/chat.json +++ b/apps/web/src/locales/en/chat.json @@ -43,6 +43,7 @@ "failedToDeleteQuickChat": "Failed to delete quick chat", "failedToMergeQueuedMessages": "Failed to merge queued messages.", "failedToRemoveQueuedMessage": "Failed to remove queued message.", + "failedToReorderQueuedMessages": "Failed to reorder queued messages.", "failedToRunQueuedMessage": "Failed to run queued message.", "failedToStartQuickChat": "Failed to start quick chat", "fileTreeActions": "File actions", @@ -96,6 +97,7 @@ "quickChatIntroSecondary": "Quick chats stay outside your task board.", "removeQueuedMessage": "Remove queued message", "renameChat": "Rename chat", + "reorderQueuedMessage": "Reorder queued message", "renameSavedOnThisDeviceOnly": "Rename saved on this device only", "renameSyncFailedDescription": "We could not sync the new name to your other devices.", "repositories": "Repositories", @@ -123,6 +125,7 @@ "sendNowTurnChanged": "The active turn changed before Send Now could start.", "sendNowAttachmentOverflow": "Combined attachments exceed the message limits.", "sendNowReferenceOverflow": "Combined entity references exceed the message limit.", + "sortable": "sortable", "technicalDetails": "Technical details", "theAssociatedWorktree": "The associated worktree", "theTaskAndItsData": "The task and its data", diff --git a/apps/web/src/locales/pseudo/chat.json b/apps/web/src/locales/pseudo/chat.json index 527e57310c..9079cbfec9 100644 --- a/apps/web/src/locales/pseudo/chat.json +++ b/apps/web/src/locales/pseudo/chat.json @@ -43,6 +43,7 @@ "failedToDeleteQuickChat": "Ƒàĩĺēď ţō ďēĺēţē qũĩćķ ćĥàţ", "failedToMergeQueuedMessages": "Ƒàĩĺēď ţō ḿēŕĝē qũēũēď ḿēśśàĝēś.", "failedToRemoveQueuedMessage": "Ƒàĩĺēď ţō ŕēḿōvē qũēũēď ḿēśśàĝē.", + "failedToReorderQueuedMessages": "Ƒàĩĺēď ţō ŕēōŕďēŕ qũēũēď ḿēśśàĝēś.", "failedToRunQueuedMessage": "Ƒàĩĺēď ţō ŕũń qũēũēď ḿēśśàĝē.", "failedToStartQuickChat": "Ƒàĩĺēď ţō śţàŕţ qũĩćķ ćĥàţ", "fileTreeActions": "Ƒĩĺē àćţĩōńś", @@ -83,6 +84,7 @@ "quickChatIntroSecondary": "Qũĩćķ ćĥàţś śţàŷ ōũţśĩďē ŷōũŕ ţàśķ ƀōàŕď.", "removeQueuedMessage": "Ŕēḿōvē qũēũēď ḿēśśàĝē", "renameChat": "Ŕēńàḿē ćĥàţ", + "reorderQueuedMessage": "Ŕēōŕďēŕ qũēũēď ḿēśśàĝē", "renameSavedOnThisDeviceOnly": "Ŕēńàḿē śàvēď ōń ţĥĩś ďēvĩćē ōńĺŷ", "renameSyncFailedDescription": "Ŵē ćōũĺď ńōţ śŷńć ţĥē ńēŵ ńàḿē ţō ŷōũŕ ōţĥēŕ ďēvĩćēś.", "repositories": "Ŕēƥōśĩţōŕĩēś", @@ -110,6 +112,7 @@ "sendNowTurnChanged": "Ţĥē àćţĩvē ţũŕń ćĥàńĝēď ƀēƒōŕē Śēńď Ńōŵ ćōũĺď śţàŕţ.", "sendNowAttachmentOverflow": "Ćōḿƀĩńēď àţţàćĥḿēńţś ēxćēēď ţĥē ḿēśśàĝē ĺĩḿĩţś.", "sendNowReferenceOverflow": "Ćōḿƀĩńēď ēńţĩţŷ ŕēƒēŕēńćēś ēxćēēď ţĥē ḿēśśàĝē ĺĩḿĩţ.", + "sortable": "śōŕţàƀĺē", "technicalDetails": "Ţēćĥńĩćàĺ ďēţàĩĺś", "theAssociatedWorktree": "Ţĥē àśśōćĩàţēď ŵōŕķţŕēē", "theTaskAndItsData": "Ţĥē ţàśķ àńď ĩţś ďàţà", diff --git a/docs/plans/message-queue-reorder/plan.md b/docs/plans/message-queue-reorder/plan.md new file mode 100644 index 0000000000..13901c46f4 --- /dev/null +++ b/docs/plans/message-queue-reorder/plan.md @@ -0,0 +1,339 @@ +--- +spec: docs/specs/ui/message-queue-reorder.md +created: 2026-08-07 +status: draft +--- + +# Implementation Plan: Reorder Queued Messages + +## Overview + +Add a durable, server-persisted reorder operation for the per-session message +queue. The backend gains `message.queue.reorder` (full-order submission, +atomically validated against the visible pending set, positions rewritten in +one transaction). The frontend wires a new `reorderEntries` hook action +(optimistic reorder + reconcile) and converts the queue panel rows to +`@dnd-kit/sortable` with a dotted grab handle that floats over the row's left +edge on hover (always visible on touch). E2E covers desktop pointer and +mobile touch drags; two shipped specs' out-of-scope lines are updated. + +Order: repository → service/handler → frontend API/hook → frontend UI → +E2E → docs. + +--- + +## Backend + +### Repository contract — `apps/backend/internal/orchestrator/messagequeue/repository.go` + +Add to the `Repository` interface: + +```go +// ReorderEntries atomically rewrites the FIFO positions of the session's +// pending entries to match orderedIDs. The submitted list must be an exact +// permutation of the currently visible pending (not reserved-in-flight) +// entry ids ordered as the caller wants them; any drift — missing, extra, +// or duplicate ids, or an id belonging to a reserved in-flight row — +// returns ErrQueueChanged and leaves the queue untouched. Reserved +// in-flight rows keep their place in the sequence; visible rows are +// interleaved in the submitted order and all positions are compacted to +// 1..N in one transaction. +ReorderEntries(ctx context.Context, sessionID string, orderedIDs []string) error +``` + +Add `ErrQueueChanged = errors.New("queue changed during reorder")` beside the +other package errors in `types.go`. + +### SQLite — `apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go` + +`ReorderEntries`: take `r.withSessionLock(sessionID)`, begin a tx (pattern: +`MergeIntoAbove`, `applyMergeWrites`). Inside the tx: + +1. Read all rows ordered by position (`SELECT id, ... ORDER BY position ASC`). +2. Split into `visible` (not `IsReservedInFlight()`) and `reserved`. +3. Validate `orderedIDs` is a permutation of the visible ids (same set, no + duplicates) → else `ErrQueueChanged`. +4. Build the merged sequence: walk the original order; reserved rows emit as + themselves, visible rows emit from `orderedIDs` in order. +5. Rewrite every row's position to `1..N` (`UPDATE queued_messages SET + position = ? WHERE id = ? AND session_id = ?`; check `RowsAffected` per + row — a vanished row during the tx means a drain raced the lock and + `ErrQueueChanged` rolls back). +6. Commit. Defer `tx.Rollback()` as in the merge path. + +Do not reuse `ListBySession` inside the tx (it reads from the read-only +handle); select within the tx so validation and writes see one snapshot under +the session lock. + +### Memory — `apps/backend/internal/orchestrator/messagequeue/repository_memory.go` + +Mirror under `r.mu.Lock()`: same visible/reserved split and permutation +validation against the stored slice (slice is not guaranteed position-sorted +after `ReplaceSession` — sort a copy by position first), then rewrite each +entry's `Position` to `1..N` in the merged sequence and bump +`r.nextPosition[sessionID]` to at least `N`. + +### Tests — `apps/backend/internal/orchestrator/messagequeue/repository_reorder_test.go` + +Table-driven over both repos (pattern: `repository_send_now_test.go` builds a +table of `new func(*testing.T) Repository`). Cases: + +- reorder reverses a 3-entry queue; positions become 1..3, content/metadata/ + queued_by untouched (compare full entries). +- no-op order (same as current) succeeds and leaves positions unchanged. +- drift: missing id, extra id, duplicate id, empty list → `ErrQueueChanged`, + queue unchanged (assert full snapshot identical). +- a reserved in-flight lifecycle row (insert via metadata + `lifecycle_reserved_in_flight: true`) is not in the visible set: submitting + only visible ids succeeds, the reserved row keeps its position, visible + rows interleave around it; submitting the reserved id → `ErrQueueChanged`. +- concurrent drain during reorder: take head after building the submitted + set, then reorder → `ErrQueueChanged` (validates the atomic read). + +--- + +### Service — `apps/backend/internal/orchestrator/messagequeue/service.go` + +```go +// ReorderEntries rewrites the pending FIFO order of a session's queue to +// match orderedIDs. Returns ErrQueueChanged when the submitted order does +// not match the current visible pending set (a drain/remove/merge raced +// the request); callers refetch the authoritative queue. +func (s *Service) ReorderEntries(ctx context.Context, sessionID string, orderedIDs []string) error +``` + +Validation before the repo call: non-empty, no duplicate ids → `ErrQueueChanged` +or a dedicated validation error (handler maps both to `queue_changed`). +Delegate to `s.repo.ReorderEntries`; wrap only unexpected errors. + +### WebSocket action — `apps/backend/pkg/websocket/actions.go` + +Add next to the queue actions: + +```go +ActionMessageQueueReorder = "message.queue.reorder" // Rewrite the visible pending order for a session +``` + +### Handler — `apps/backend/internal/orchestrator/handlers/queue_handlers.go` + +- Add `ReorderEntries(ctx context.Context, sessionID string, orderedIDs []string) error` + to the `QueueService` interface (real impl is `messagequeue.Service` via + `orchestratorSvc.GetMessageQueue()` — `apps/backend/internal/backendapp/gateway.go`). +- Register `d.RegisterFunc(ws.ActionMessageQueueReorder, h.wsReorder)`. +- `wsReorder`: parse `{session_id, ordered_ids}`; validation errors for + missing session, empty/duplicate `ordered_ids`; `authorizeSession`; call + `ReorderEntries`; map `ErrQueueChanged` (and the service-level validation + error) to error code `queue_changed`; other errors → `500`-style WS error. + On success respond `{session_id, reordered: N}` and call `publishStatus`. +- Add `queueErrorCodeQueueChanged = "queue_changed"` constant (leave the + existing send-now constant untouched; wire value is identical). + +### Tests — handler + service + +- `apps/backend/internal/orchestrator/handlers/queue_handlers_reorder_test.go`: + happy path (mock `QueueService` + `mockEventBus` asserts `publishStatus` + ran), validation errors (missing session, empty ids, duplicate ids), + access denied, `ErrQueueChanged` → `queue_changed` error code, and a real + `messagequeue.NewServiceMemory` round-trip (queue 2 entries via + `QueueMessageWithMetadata`, reorder reversed, assert response + status). + Add the action to the table in `queue_handlers_test.go` (mirror the + `merge` row). +- Service-level validation covered in the repository tests plus a small + `service_test.go` case for duplicate/empty input. + +--- + +## Frontend + +### API client — `apps/web/lib/api/domains/queue-api.ts` + +```ts +export class QueueReorderError extends Error { + readonly code = "queue_changed" as const; + constructor(message?: string) { ... this.name = "QueueReorderError"; } +} + +export async function reorderQueuedEntries(params: { + session_id: string; + ordered_ids: string[]; +}): Promise<{ session_id: string; reordered: number }> +``` + +`reorderQueuedEntries` sends `{ action: "message.queue.reorder", payload }` +via `getWebSocketClient()`. Because `rethrowQueueError` currently maps +`queue_changed` to `QueueSendNowError`, check `asWSError(err)?.code === +"queue_changed"` in `reorderQueuedEntries` and throw `QueueReorderError` +before falling through to `rethrowQueueError`. + +### Hook — `apps/web/hooks/domains/session/use-queue.ts` + +Add `reorderEntries(orderedIds: string[])`: + +1. Optimistic: reorder the current `entries` by id using `setQueueEntries` + (keep the same `meta`). +2. `await reorderQueuedEntries({ session_id, ordered_ids })`; on + `QueueReorderError` refetch and swallow (self-recovering race, mirrors + `mergeEntry`'s `QueueEntryNotFoundError` handling); on any other error + refetch and rethrow. +3. On success refetch to reconcile (authoritative order + positions). + +Expose `reorderEntries` from `useQueue` and from +`useQueuePanelHandlers`/`QueueAffordance` as `handleReorder(orderedIds)` that +toasts `chat:failedToReorderQueuedMessages` for non-race errors. + +### State slice + +No changes: `setQueueEntries` and `metaBySessionId` already exist in +`apps/web/lib/state/slices/session/types.ts` / the session store. + +### i18n — `apps/web/src/locales/en/chat.json` (+ regenerate pseudo) + +- `"reorderQueuedMessage": "Reorder queued message"` — handle aria-label +- `"sortable": "sortable"` — handle `aria-roledescription` +- `"failedToReorderQueuedMessages": "Failed to reorder queued messages."` — + error toast + +Run `pnpm i18n:pseudo` to regenerate `src/locales/pseudo/chat.json`, then +`pnpm i18n:check` and `pnpm i18n:ratchet`. No new copy may be a hardcoded +literal in JSX. + +--- + +### Queue panel — `apps/web/components/task/chat/queued-ghost-list.tsx` + +In `QueuePanel`, wrap the `queue-scroll-region` list with dnd-kit (pattern: +`apps/web/components/kanban/swimlane-container.tsx`): + +```tsx + setActiveId(String(active.id))} + onDragEnd={handleDragEnd} + onDragCancel={() => setActiveId(null)} +> + + {entries.map((entry, index) => ( + + ))} + + +``` + +- `sensors = useSensors(useSensor(PointerSensor, { activationConstraint: { distance: 4 } }), useSensor(KeyboardSensor, { coordinateGetter: sortableKeyboardCoordinates }))`. +- `handleDragEnd`: no-op when `over` is null or ids match; otherwise + `arrayMove(ids, oldIndex, newIndex)` → `onReorder(newIds)`. +- `onReorder` = `handleReorder` from `QueueAffordance` (threaded through + `QueuePanelDisclosure` + `QueuePanel` props, like `onMerge`). + +### Row — `apps/web/components/task/chat/queued-ghost-message.tsx` + +- `useSortable({ id: entry.id, disabled: editing || !canDrag })` on the row + wrapper; apply `setNodeRef`, `style={{ transform: CSS.Transform.toString(transform), transition }}`, + add `group relative` + `z-10 opacity-40` while `isDragging`. +- Add `canDrag` + `isDragging` props; `canDrag` defaults `true` for + standalone renders/tests. +- New `QueueGrabHandle` component rendered when `!editing`: an absolutely + positioned `