diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 2103dacb27..ac944e1373 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -131,6 +131,10 @@ export class Task extends EventEmitter implements TaskLike { readonly taskNumber: number readonly workspacePath: string + // Message queue for handling user messages + private messageQueue: Array<{ text: string; images?: string[] }> = [] + private isProcessingMessage: boolean = false + /** * The mode associated with this task. Persisted across sessions * to maintain user context when reopening tasks from history. @@ -742,11 +746,48 @@ export class Task extends EventEmitter implements TaskLike { } handleWebviewAskResponse(askResponse: ClineAskResponse, text?: string, images?: string[]) { + // If we're currently processing a message and this is a new user message, + // queue it instead of processing immediately + if (this.isProcessingMessage && askResponse === "messageResponse" && (text || images?.length)) { + this.messageQueue.push({ text: text || "", images }) + this.providerRef + .deref() + ?.log(`[Task#handleWebviewAskResponse] Message queued. Queue size: ${this.messageQueue.length}`) + + // Notify the user that their message has been queued + this.providerRef.deref()?.postMessageToWebview({ + type: "messageQueued", + queueSize: this.messageQueue.length, + }) + return + } + this.askResponse = askResponse this.askResponseText = text this.askResponseImages = images } + // Process the next message in the queue + private async processNextQueuedMessage() { + if (this.messageQueue.length === 0 || this.isProcessingMessage) { + return + } + + const nextMessage = this.messageQueue.shift() + if (!nextMessage) { + return + } + + this.providerRef + .deref() + ?.log( + `[Task#processNextQueuedMessage] Processing queued message. Remaining in queue: ${this.messageQueue.length}`, + ) + + // Submit the queued message as a new user message + this.submitUserMessage(nextMessage.text, nextMessage.images) + } + public submitUserMessage(text: string, images?: string[]): void { try { const trimmed = (text ?? "").trim() @@ -1409,30 +1450,37 @@ export class Task extends EventEmitter implements TaskLike { let includeFileDetails = true this.emit(RooCodeEventName.TaskStarted) + this.isProcessingMessage = true - while (!this.abort) { - const didEndLoop = await this.recursivelyMakeClineRequests(nextUserContent, includeFileDetails) - includeFileDetails = false // We only need file details the first time. + try { + while (!this.abort) { + const didEndLoop = await this.recursivelyMakeClineRequests(nextUserContent, includeFileDetails) + includeFileDetails = false // We only need file details the first time. - // The way this agentic loop works is that cline will be given a - // task that he then calls tools to complete. Unless there's an - // attempt_completion call, we keep responding back to him with his - // tool's responses until he either attempt_completion or does not - // use anymore tools. If he does not use anymore tools, we ask him - // to consider if he's completed the task and then call - // attempt_completion, otherwise proceed with completing the task. - // There is a MAX_REQUESTS_PER_TASK limit to prevent infinite - // requests, but Cline is prompted to finish the task as efficiently - // as he can. + // The way this agentic loop works is that cline will be given a + // task that he then calls tools to complete. Unless there's an + // attempt_completion call, we keep responding back to him with his + // tool's responses until he either attempt_completion or does not + // use anymore tools. If he does not use anymore tools, we ask him + // to consider if he's completed the task and then call + // attempt_completion, otherwise proceed with completing the task. + // There is a MAX_REQUESTS_PER_TASK limit to prevent infinite + // requests, but Cline is prompted to finish the task as efficiently + // as he can. - if (didEndLoop) { - // For now a task never 'completes'. This will only happen if - // the user hits max requests and denies resetting the count. - break - } else { - nextUserContent = [{ type: "text", text: formatResponse.noToolsUsed() }] - this.consecutiveMistakeCount++ + if (didEndLoop) { + // For now a task never 'completes'. This will only happen if + // the user hits max requests and denies resetting the count. + break + } else { + nextUserContent = [{ type: "text", text: formatResponse.noToolsUsed() }] + this.consecutiveMistakeCount++ + } } + } finally { + this.isProcessingMessage = false + // Process any queued messages after the current task completes + await this.processNextQueuedMessage() } } @@ -1444,6 +1492,9 @@ export class Task extends EventEmitter implements TaskLike { throw new Error(`[RooCode#recursivelyMakeRooRequests] task ${this.taskId}.${this.instanceId} aborted`) } + // Mark that we're processing a message + this.isProcessingMessage = true + if (this.consecutiveMistakeLimit > 0 && this.consecutiveMistakeCount >= this.consecutiveMistakeLimit) { const { response, text, images } = await this.ask( "mistake_limit_reached", @@ -1889,8 +1940,19 @@ export class Task extends EventEmitter implements TaskLike { }) } + // Mark processing complete and check for queued messages + this.isProcessingMessage = false + + // Process next queued message if any + if (this.messageQueue.length > 0) { + await this.processNextQueuedMessage() + } + return didEndLoop // Will always be false for now. } catch (error) { + // Mark processing complete even on error + this.isProcessingMessage = false + // This should never happen since the only thing that can throw an // error is the attemptApiRequest, which is wrapped in a try catch // that sends an ask where if noButtonClicked, will clear current diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index 39df433814..ea825d6d46 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -1613,5 +1613,249 @@ describe("Cline", () => { consoleErrorSpy.mockRestore() }) }) + + describe("Message Queueing", () => { + let task: Task + let mockProvider: any + + beforeEach(() => { + mockProvider = { + context: { + globalStorageUri: { fsPath: "/test/storage" }, + }, + getState: vi.fn().mockResolvedValue({}), + postMessageToWebview: vi.fn().mockResolvedValue(undefined), + postStateToWebview: vi.fn().mockResolvedValue(undefined), + updateTaskHistory: vi.fn().mockResolvedValue(undefined), + log: vi.fn(), // Add log method to mock provider + } + + task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "test task", + startTask: false, + }) + }) + + it("should queue messages when already processing", async () => { + // Access private properties via any cast for testing + const taskAny = task as any + + // Set processing state + taskAny.isProcessingMessage = true + + // Spy on postMessageToWebview + const postMessageSpy = vi.spyOn(mockProvider, "postMessageToWebview") + + // Try to handle a new message while processing + await task.handleWebviewAskResponse("messageResponse", "new message while processing") + + // Verify message was queued + expect(taskAny.messageQueue).toHaveLength(1) + expect(taskAny.messageQueue[0]).toEqual({ + text: "new message while processing", + images: undefined, + }) + + // Verify queue notification was sent + expect(postMessageSpy).toHaveBeenCalledWith({ + type: "messageQueued", + queueSize: 1, + }) + }) + + it("should process messages immediately when not processing", async () => { + // Access private properties via any cast + const taskAny = task as any + + // Directly test the queueing logic + taskAny.isProcessingMessage = false + + // Call handleWebviewAskResponse + await task.handleWebviewAskResponse("messageResponse", "test message") + + // Since we're not processing, it should start processing immediately + // The message should not be in the queue + expect(taskAny.messageQueue).toHaveLength(0) + }) + + it("should process queued messages in FIFO order", async () => { + // Access private properties via any cast + const taskAny = task as any + + // Set processing state + taskAny.isProcessingMessage = true + + // Queue multiple messages + await task.handleWebviewAskResponse("messageResponse", "first message") + await task.handleWebviewAskResponse("messageResponse", "second message") + await task.handleWebviewAskResponse("messageResponse", "third message") + + // Verify all messages were queued in order + expect(taskAny.messageQueue).toHaveLength(3) + expect(taskAny.messageQueue[0].text).toBe("first message") + expect(taskAny.messageQueue[1].text).toBe("second message") + expect(taskAny.messageQueue[2].text).toBe("third message") + + // Reset processing state and remove first message to simulate processing + taskAny.isProcessingMessage = false + const firstMessage = taskAny.messageQueue.shift() + + // Verify FIFO order is maintained + expect(firstMessage.text).toBe("first message") + expect(taskAny.messageQueue).toHaveLength(2) + expect(taskAny.messageQueue[0].text).toBe("second message") + }) + + it("should handle empty queue gracefully", async () => { + // Access private properties via any cast + const taskAny = task as any + + // Ensure queue is empty + taskAny.messageQueue = [] + + // Mock initiateTaskLoop + const initiateTaskLoopSpy = vi.spyOn(taskAny, "initiateTaskLoop").mockResolvedValue(undefined) + + // Try to process next queued message + await taskAny.processNextQueuedMessage() + + // Verify no processing occurred + expect(initiateTaskLoopSpy).not.toHaveBeenCalled() + expect(taskAny.isProcessingMessage).toBe(false) + }) + + it("should set processing state correctly during message handling", async () => { + // Access private properties via any cast + const taskAny = task as any + + // Initially not processing + expect(taskAny.isProcessingMessage).toBe(false) + + // Mock recursivelyMakeClineRequests to track processing + let processingDuringRequest = false + vi.spyOn(taskAny, "recursivelyMakeClineRequests").mockImplementation(async () => { + processingDuringRequest = taskAny.isProcessingMessage + return true + }) + + // Mock other required methods + vi.spyOn(task, "say").mockResolvedValue(undefined) + vi.spyOn(taskAny, "addToApiConversationHistory").mockReturnValue(undefined) + vi.spyOn(taskAny, "saveApiConversationHistory").mockResolvedValue(undefined) + + // Start processing a message + await taskAny.initiateTaskLoop([{ type: "text", text: "test message" }]) + + // Verify processing state was set during request + expect(processingDuringRequest).toBe(true) + + // Verify processing state is reset after completion + expect(taskAny.isProcessingMessage).toBe(false) + }) + + it("should send queue size updates when queueing multiple messages", async () => { + // Access private properties via any cast + const taskAny = task as any + + // Set processing state + taskAny.isProcessingMessage = true + + // Spy on postMessageToWebview + const postMessageSpy = vi.spyOn(mockProvider, "postMessageToWebview") + + // Queue multiple messages + await task.handleWebviewAskResponse("messageResponse", "message 1") + await task.handleWebviewAskResponse("messageResponse", "message 2") + await task.handleWebviewAskResponse("messageResponse", "message 3") + + // Verify queue notifications were sent with correct sizes + expect(postMessageSpy).toHaveBeenNthCalledWith(1, { + type: "messageQueued", + queueSize: 1, + }) + expect(postMessageSpy).toHaveBeenNthCalledWith(2, { + type: "messageQueued", + queueSize: 2, + }) + expect(postMessageSpy).toHaveBeenNthCalledWith(3, { + type: "messageQueued", + queueSize: 3, + }) + }) + + it("should handle rapid message submissions correctly", async () => { + // Access private properties via any cast + const taskAny = task as any + + // Set processing state + taskAny.isProcessingMessage = true + + // Simulate rapid message submissions + const promises = [] + for (let i = 1; i <= 10; i++) { + promises.push(task.handleWebviewAskResponse("messageResponse", `rapid message ${i}`)) + } + + // Wait for all to complete + await Promise.all(promises) + + // Verify all messages were queued + expect(taskAny.messageQueue).toHaveLength(10) + for (let i = 0; i < 10; i++) { + expect(taskAny.messageQueue[i].text).toBe(`rapid message ${i + 1}`) + } + }) + + it("should reset processing state when recursivelyMakeClineRequests completes", async () => { + // Access private properties via any cast + const taskAny = task as any + + // Mock the recursive method to track state changes + vi.spyOn(taskAny, "recursivelyMakeClineRequests").mockResolvedValue(true) + + // Mock other required methods + vi.spyOn(task, "say").mockResolvedValue(undefined) + vi.spyOn(taskAny, "addToApiConversationHistory").mockReturnValue(undefined) + vi.spyOn(taskAny, "saveApiConversationHistory").mockResolvedValue(undefined) + + // Queue a message for after processing + taskAny.messageQueue = [{ text: "queued message" }] + + // Mock processNextQueuedMessage + const processNextSpy = vi.spyOn(taskAny, "processNextQueuedMessage").mockResolvedValue(undefined) + + // Start processing + await taskAny.initiateTaskLoop([{ type: "text", text: "initial message" }]) + + // Verify processing state was reset + expect(taskAny.isProcessingMessage).toBe(false) + + // Verify next queued message was triggered + expect(processNextSpy).toHaveBeenCalled() + }) + + it("should handle undefined provider reference when sending queue notifications", async () => { + // Access private properties via any cast + const taskAny = task as any + + // Set processing state + taskAny.isProcessingMessage = true + + // Simulate weakref returning undefined + Object.defineProperty(task, "providerRef", { + value: { deref: () => undefined }, + writable: false, + configurable: true, + }) + + // Try to queue a message - this should not throw + await task.handleWebviewAskResponse("messageResponse", "message without provider") + + // Verify message was still queued + expect(taskAny.messageQueue).toHaveLength(1) + }) + }) }) }) diff --git a/src/shared/ExtensionMessage.ts b/src/shared/ExtensionMessage.ts index ebdc137432..ec453790d2 100644 --- a/src/shared/ExtensionMessage.ts +++ b/src/shared/ExtensionMessage.ts @@ -66,6 +66,7 @@ export interface ExtensionMessage { | "workspaceUpdated" | "invoke" | "messageUpdated" + | "messageQueued" | "mcpServers" | "enhancedPrompt" | "commitSearchResults" @@ -193,6 +194,7 @@ export interface ExtensionMessage { messageTs?: number context?: string commands?: Command[] + queueSize?: number } export type ExtensionState = Pick<