feat(queue): merge a queued message into the message above it (#2131)

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