diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers.go b/apps/backend/internal/orchestrator/handlers/queue_handlers.go index a59661c856..32a0eea08f 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers.go +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers.go @@ -17,6 +17,7 @@ type QueueService interface { CancelQueued(ctx context.Context, sessionID string) (*messagequeue.QueuedMessage, error) GetStatus(ctx context.Context, sessionID string) *messagequeue.QueueStatus UpdateMessage(ctx context.Context, sessionID, content string) error + AppendContent(ctx context.Context, sessionID, content string) error } // QueueHandlers handles message queue operations @@ -41,6 +42,7 @@ func (h *QueueHandlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionMessageQueueCancel, h.wsCancelQueue) d.RegisterFunc(ws.ActionMessageQueueGet, h.wsGetQueueStatus) d.RegisterFunc(ws.ActionMessageQueueUpdate, h.wsUpdateMessage) + d.RegisterFunc(ws.ActionMessageQueueAppend, h.wsAppendToQueue) } // WebSocket Handlers @@ -192,3 +194,58 @@ func (h *QueueHandlers) wsUpdateMessage(ctx context.Context, msg *ws.Message) (* return ws.NewResponse(msg.ID, msg.Action, status) } + +type wsAppendToQueueRequest struct { + SessionID string `json:"session_id"` + TaskID string `json:"task_id"` + Content string `json:"content"` + Model string `json:"model,omitempty"` + PlanMode bool `json:"plan_mode,omitempty"` + UserID string `json:"user_id,omitempty"` +} + +func (h *QueueHandlers) wsAppendToQueue(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsAppendToQueueRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + + if req.SessionID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + if req.Content == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "content is required", nil) + } + + // Try to append to existing queued message; if none exists, create a new one + if err := h.queueService.AppendContent(ctx, req.SessionID, req.Content); err != nil { + _, queueErr := h.queueService.QueueMessage( + ctx, req.SessionID, req.TaskID, req.Content, req.Model, req.UserID, req.PlanMode, nil, + ) + if queueErr != nil { + h.logger.Error("failed to queue message on append fallback", zap.Error(queueErr)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to queue message", nil) + } + } + + status := h.queueService.GetStatus(ctx, req.SessionID) + + // Publish queue status changed event + if h.eventBus != nil { + eventData := map[string]interface{}{ + "session_id": req.SessionID, + "is_queued": status.IsQueued, + "message": status.Message, + } + _ = h.eventBus.Publish(ctx, events.MessageQueueStatusChanged, bus.NewEvent( + events.MessageQueueStatusChanged, + "queue-handlers", + eventData, + )) + } + + return ws.NewResponse(msg.ID, msg.Action, status) +} diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go b/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go index 648df69d54..5a862c218a 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go @@ -21,6 +21,7 @@ type mockQueueService struct { cancelQueuedFunc func(ctx context.Context, sessionID string) (*messagequeue.QueuedMessage, error) getStatusFunc func(ctx context.Context, sessionID string) *messagequeue.QueueStatus updateMessageFunc func(ctx context.Context, sessionID, content string) error + appendContentFunc func(ctx context.Context, sessionID, content string) error } func (m *mockQueueService) QueueMessage(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []messagequeue.MessageAttachment) (*messagequeue.QueuedMessage, error) { @@ -51,6 +52,13 @@ func (m *mockQueueService) UpdateMessage(ctx context.Context, sessionID, content return nil } +func (m *mockQueueService) AppendContent(ctx context.Context, sessionID, content string) error { + if m.appendContentFunc != nil { + return m.appendContentFunc(ctx, sessionID, content) + } + return errors.New("no queued message") +} + // mockEventBus is a simple mock event bus type mockEventBus struct { publishFunc func(ctx context.Context, subject string, event *bus.Event) error @@ -456,6 +464,120 @@ func TestWsUpdateMessage(t *testing.T) { }) } +func TestWsAppendToQueue(t *testing.T) { + t.Run("appends to existing queued message", func(t *testing.T) { + handlers, mockQueue, mockBus := setupQueueHandlers(t) + ctx := context.Background() + + mockQueue.appendContentFunc = func(ctx context.Context, sessionID, content string) error { + assert.Equal(t, "session-1", sessionID) + assert.Equal(t, "new comment", content) + return nil + } + + mockQueue.getStatusFunc = func(ctx context.Context, sessionID string) *messagequeue.QueueStatus { + return &messagequeue.QueueStatus{ + IsQueued: true, + Message: &messagequeue.QueuedMessage{ + ID: "queue-1", + Content: "original\n\n---\n\nnew comment", + }, + } + } + + mockBus.publishFunc = func(ctx context.Context, subject string, event *bus.Event) error { + return nil + } + + msg := createTestMessage(t, ws.ActionMessageQueueAppend, map[string]interface{}{ + "session_id": "session-1", + "task_id": "task-1", + "content": "new comment", + }) + + response, err := handlers.wsAppendToQueue(ctx, msg) + + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeResponse, response.Type) + }) + + t.Run("creates new queued message when none exists", func(t *testing.T) { + handlers, mockQueue, mockBus := setupQueueHandlers(t) + ctx := context.Background() + + // AppendContent returns error (no existing message) + mockQueue.appendContentFunc = func(ctx context.Context, sessionID, content string) error { + return errors.New("no queued message") + } + + // Falls back to QueueMessage + mockQueue.queueMessageFunc = func(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []messagequeue.MessageAttachment) (*messagequeue.QueuedMessage, error) { + assert.Equal(t, "session-1", sessionID) + assert.Equal(t, "task-1", taskID) + assert.Equal(t, "new comment", content) + return &messagequeue.QueuedMessage{ID: "queue-1", Content: content}, nil + } + + mockQueue.getStatusFunc = func(ctx context.Context, sessionID string) *messagequeue.QueueStatus { + return &messagequeue.QueueStatus{ + IsQueued: true, + Message: &messagequeue.QueuedMessage{ID: "queue-1", Content: "new comment"}, + } + } + + mockBus.publishFunc = func(ctx context.Context, subject string, event *bus.Event) error { + return nil + } + + msg := createTestMessage(t, ws.ActionMessageQueueAppend, map[string]interface{}{ + "session_id": "session-1", + "task_id": "task-1", + "content": "new comment", + }) + + response, err := handlers.wsAppendToQueue(ctx, msg) + + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeResponse, response.Type) + }) + + t.Run("returns error when content is empty", func(t *testing.T) { + handlers, _, _ := setupQueueHandlers(t) + ctx := context.Background() + + msg := createTestMessage(t, ws.ActionMessageQueueAppend, map[string]interface{}{ + "session_id": "session-1", + "task_id": "task-1", + "content": "", + }) + + response, err := handlers.wsAppendToQueue(ctx, msg) + + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + + var errorPayload ws.ErrorPayload + err = json.Unmarshal(response.Payload, &errorPayload) + require.NoError(t, err) + assert.Contains(t, errorPayload.Message, "content is required") + }) + + t.Run("returns error when session_id missing", func(t *testing.T) { + handlers, _, _ := setupQueueHandlers(t) + ctx := context.Background() + + msg := createTestMessage(t, ws.ActionMessageQueueAppend, map[string]interface{}{ + "task_id": "task-1", + "content": "test", + }) + + response, err := handlers.wsAppendToQueue(ctx, msg) + + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeError, response.Type) + }) +} + func TestRegisterHandlers(t *testing.T) { t.Run("registers all queue handlers", func(t *testing.T) { handlers, _, _ := setupQueueHandlers(t) diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index 5cb70cb1d5..7c41e9f0c9 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/apps/backend/internal/orchestrator/messagequeue/service.go @@ -102,6 +102,26 @@ func (s *Service) GetStatus(ctx context.Context, sessionID string) *QueueStatus } } +// AppendContent appends content to an existing queued message. +// Returns an error if no message is queued (caller should use QueueMessage instead). +func (s *Service) AppendContent(ctx context.Context, sessionID, content string) error { + s.mu.Lock() + defer s.mu.Unlock() + + msg, exists := s.queued[sessionID] + if !exists { + return fmt.Errorf("no queued message for session %s", sessionID) + } + + msg.Content = msg.Content + "\n\n---\n\n" + content + s.logger.Info("content appended to queued message", + zap.String("session_id", sessionID), + zap.Int("appended_length", len(content)), + zap.Int("total_length", len(msg.Content))) + + return nil +} + // UpdateMessage updates an existing queued message (for arrow up editing) func (s *Service) UpdateMessage(ctx context.Context, sessionID, content string) error { s.mu.Lock() diff --git a/apps/backend/internal/orchestrator/messagequeue/service_test.go b/apps/backend/internal/orchestrator/messagequeue/service_test.go index 177f09a7bd..e72131304f 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service_test.go +++ b/apps/backend/internal/orchestrator/messagequeue/service_test.go @@ -192,6 +192,60 @@ func TestGetStatus(t *testing.T) { }) } +func TestAppendContent(t *testing.T) { + t.Run("appends content to existing queued message", func(t *testing.T) { + svc := setupService(t) + ctx := context.Background() + + _, err := svc.QueueMessage(ctx, "session-1", "task-1", "original content", "model-1", "user-1", false, nil) + require.NoError(t, err) + + err = svc.AppendContent(ctx, "session-1", "appended content") + require.NoError(t, err) + + status := svc.GetStatus(ctx, "session-1") + assert.True(t, status.IsQueued) + assert.Equal(t, "original content\n\n---\n\nappended content", status.Message.Content) + }) + + t.Run("appends multiple times", func(t *testing.T) { + svc := setupService(t) + ctx := context.Background() + + _, err := svc.QueueMessage(ctx, "session-1", "task-1", "first", "model-1", "user-1", false, nil) + require.NoError(t, err) + + require.NoError(t, svc.AppendContent(ctx, "session-1", "second")) + require.NoError(t, svc.AppendContent(ctx, "session-1", "third")) + + status := svc.GetStatus(ctx, "session-1") + assert.Equal(t, "first\n\n---\n\nsecond\n\n---\n\nthird", status.Message.Content) + }) + + t.Run("returns error when no message queued", func(t *testing.T) { + svc := setupService(t) + ctx := context.Background() + + err := svc.AppendContent(ctx, "session-1", "content") + assert.Error(t, err) + assert.Contains(t, err.Error(), "no queued message") + }) + + t.Run("preserves message ID after append", func(t *testing.T) { + svc := setupService(t) + ctx := context.Background() + + msg, err := svc.QueueMessage(ctx, "session-1", "task-1", "original", "model-1", "user-1", false, nil) + require.NoError(t, err) + originalID := msg.ID + + require.NoError(t, svc.AppendContent(ctx, "session-1", "extra")) + + status := svc.GetStatus(ctx, "session-1") + assert.Equal(t, originalID, status.Message.ID) + }) +} + func TestUpdateMessage(t *testing.T) { t.Run("updates existing queued message content", func(t *testing.T) { svc := setupService(t) diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 15d0d01c6c..333a177379 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -111,6 +111,7 @@ const ( ActionMessageQueueCancel = "message.queue.cancel" ActionMessageQueueGet = "message.queue.get" ActionMessageQueueUpdate = "message.queue.update" + ActionMessageQueueAppend = "message.queue.append" ActionMessageQueueStatusChanged = "message.queue.status_changed" // Notification: queue status changed // Workflow template/step actions diff --git a/apps/web/components/diff/comment-display.tsx b/apps/web/components/diff/comment-display.tsx index a4a15b142c..05dd33dcc7 100644 --- a/apps/web/components/diff/comment-display.tsx +++ b/apps/web/components/diff/comment-display.tsx @@ -1,7 +1,7 @@ "use client"; import { Button } from "@kandev/ui/button"; -import { IconTrash, IconEdit, IconMessage } from "@tabler/icons-react"; +import { IconTrash, IconEdit, IconMessage, IconPlayerPlay } from "@tabler/icons-react"; import type { DiffComment } from "@/lib/diff/types"; import { formatLineRange } from "@/lib/diff"; @@ -12,16 +12,85 @@ interface CommentDisplayProps { onDelete?: () => void; /** Callback to edit the comment */ onEdit?: () => void; + /** Callback to run/send the comment to the agent immediately */ + onRun?: () => void; /** Whether to show the code content */ showCode?: boolean; /** Compact mode for inline display */ compact?: boolean; } +type ActionButtonsProps = { + onRun?: () => void; + onEdit?: () => void; + onDelete?: () => void; + size: "compact" | "normal"; + stopPropagation?: boolean; +}; + +function CommentActionButtons({ + onRun, + onEdit, + onDelete, + size, + stopPropagation, +}: ActionButtonsProps) { + const btnClass = size === "compact" ? "h-4 w-4 p-0" : "h-5 w-5 p-0"; + const wrap = (fn?: () => void) => + stopPropagation + ? (e: React.MouseEvent) => { + e.stopPropagation(); + fn?.(); + } + : fn; + + return ( + <> + {onRun && ( + + )} + {onEdit && ( + + )} + {onDelete && ( + + )} + + ); +} + export function CommentDisplay({ comment, onDelete, onEdit, + onRun, showCode = false, compact = false, }: CommentDisplayProps) { @@ -40,32 +109,13 @@ export function CommentDisplay({ {comment.text}
- {onEdit && ( - - )} - {onDelete && ( - - )} +
); @@ -73,7 +123,6 @@ export function CommentDisplay({ return (
- {/* Header */}
@@ -83,37 +132,14 @@ export function CommentDisplay({
- {onEdit && ( - - )} - {onDelete && ( - - )} +
- - {/* Code preview */} {showCode && comment.codeContent && (
           {comment.codeContent}
         
)} - - {/* Comment text */}

{comment.text}

); diff --git a/apps/web/components/diff/comment-form.tsx b/apps/web/components/diff/comment-form.tsx index 8bd00a79e1..e00e213f36 100644 --- a/apps/web/components/diff/comment-form.tsx +++ b/apps/web/components/diff/comment-form.tsx @@ -3,25 +3,94 @@ import { useState, useRef, useEffect } from "react"; import { Button } from "@kandev/ui/button"; import { Textarea } from "@kandev/ui/textarea"; -import { IconSend, IconX } from "@tabler/icons-react"; +import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from "@kandev/ui/tooltip"; +import { IconPlayerPlay, IconSend, IconX } from "@tabler/icons-react"; interface CommentFormProps { - /** Initial content for editing */ initialContent?: string; - /** Callback when comment is submitted */ onSubmit: (content: string) => void; - /** Callback when form is cancelled */ onCancel: () => void; - /** Whether the form is in editing mode */ + onSubmitAndRun?: (content: string) => void; isEditing?: boolean; - /** Auto focus the textarea */ autoFocus?: boolean; } +function ActionButtons({ + disabled, + modKey, + isEditing, + showRunButton, + onCancel, + onSubmit, + onSubmitAndRun, +}: { + disabled: boolean; + modKey: string; + isEditing: boolean; + showRunButton: boolean; + onCancel: () => void; + onSubmit: () => void; + onSubmitAndRun?: () => void; +}) { + return ( + +
+ +
+ + + + + +

Save comment for review ({modKey}+Enter)

+
+
+ {showRunButton && onSubmitAndRun && ( + + + + + +

Save and send to agent ({modKey}+Shift+Enter)

+
+
+ )} +
+
+
+ ); +} + export function CommentForm({ initialContent = "", onSubmit, onCancel, + onSubmitAndRun, isEditing = false, autoFocus = true, }: CommentFormProps) { @@ -29,9 +98,7 @@ export function CommentForm({ const textareaRef = useRef(null); useEffect(() => { - if (autoFocus && textareaRef.current) { - textareaRef.current.focus(); - } + if (autoFocus && textareaRef.current) textareaRef.current.focus(); }, [autoFocus]); const handleSubmit = () => { @@ -42,16 +109,31 @@ export function CommentForm({ } }; + const handleSubmitAndRun = () => { + const trimmed = content.trim(); + if (trimmed && onSubmitAndRun) { + onSubmitAndRun(trimmed); + setContent(""); + } + }; + + const isMac = typeof navigator !== "undefined" && navigator.platform?.includes("Mac"); + const modKey = isMac ? "⌘" : "Ctrl"; + const handleKeyDown = (e: React.KeyboardEvent) => { if (e.key === "Enter" && (e.metaKey || e.ctrlKey)) { e.preventDefault(); - handleSubmit(); + if (e.shiftKey && onSubmitAndRun) handleSubmitAndRun(); + else handleSubmit(); } else if (e.key === "Escape") { e.preventDefault(); onCancel(); } }; + const disabled = !content.trim(); + const showRunButton = !!onSubmitAndRun && !isEditing; + return (