Move message queue to the extension host

Author: cte

packages/types/src/message.ts

@@ -248,14 +248,9 @@ export type TokenUsage = z.infer<typeof tokenUsageSchema>
  * QueuedMessage
  */
 
-/**
- * Represents a message that is queued to be sent when sending is enabled
- */
 export interface QueuedMessage {
-	/** Unique identifier for the queued message */
+	timestamp: number
 	id: string
-	/** The text content of the message */
 	text: string
-	/** Array of image data URLs attached to the message */
-	images: string[]
+	images?: string[]
 }

src/core/message-queue/MessageQueueService.ts

@@ -0,0 +1,100 @@
+import { EventEmitter } from "events"
+
+import { v4 as uuidv4 } from "uuid"
+
+import { QueuedMessage } from "@roo-code/types"
+
+export interface MessageQueueState {
+	messages: QueuedMessage[]
+	isProcessing: boolean
+	isPaused: boolean
+}
+
+export interface QueueEvents {
+	stateChanged: [messages: QueuedMessage[]]
+}
+
+export class MessageQueueService extends EventEmitter<QueueEvents> {
+	private _messages: QueuedMessage[]
+
+	constructor() {
+		super()
+
+		this._messages = []
+	}
+
+	private findMessage(id: string) {
+		const index = this._messages.findIndex((msg) => msg.id === id)
+
+		if (index === -1) {
+			return { index, message: undefined }
+		}
+
+		return { index, message: this._messages[index] }
+	}
+
+	public addMessage(text: string, images?: string[]): QueuedMessage | undefined {
+		if (!text && !images?.length) {
+			return undefined
+		}
+
+		const message: QueuedMessage = {
+			timestamp: Date.now(),
+			id: uuidv4(),
+			text,
+			images,
+		}
+
+		this._messages.push(message)
+		this.emit("stateChanged", this._messages)
+
+		return message
+	}
+
+	public removeMessage(id: string): boolean {
+		const { index, message } = this.findMessage(id)
+
+		if (!message) {
+			return false
+		}
+
+		console.log("removeMessage", message)
+		this._messages.splice(index, 1)
+		this.emit("stateChanged", this._messages)
+		return true
+	}
+
+	public updateMessage(id: string, text: string, images?: string[]): boolean {
+		const { message } = this.findMessage(id)
+
+		if (!message) {
+			return false
+		}
+
+		console.log("updateMessage", message)
+		message.timestamp = Date.now()
+		message.text = text
+		message.images = images
+		this.emit("stateChanged", this._messages)
+		return true
+	}
+
+	public dequeueMessage(): QueuedMessage | undefined {
+		const message = this._messages.shift()
+		this.emit("stateChanged", this._messages)
+		return message
+	}
+
+	public get messages(): QueuedMessage[] {
+		return this._messages
+	}
+
+	public isEmpty(): boolean {
+		return this._messages.length === 0
+	}
+
+	public dispose(): void {
+		this._messages = []
+		this.removeAllListeners()
+	}
+}

src/core/task-persistence/index.ts

@@ -1,3 +1,3 @@
-export { readApiMessages, saveApiMessages } from "./apiMessages"
+export { type ApiMessage, readApiMessages, saveApiMessages } from "./apiMessages"
 export { readTaskMessages, saveTaskMessages } from "./taskMessages"
 export { taskMetadata } from "./taskMetadata"

src/core/task/Task.ts

@@ -10,7 +10,6 @@ import pWaitFor from "p-wait-for"
 import { serializeError } from "serialize-error"
 
 import {
-	type RooCodeSettings,
 	type TaskLike,
 	type TaskMetadata,
 	type TaskEvents,
@@ -42,6 +41,7 @@ import { CloudService, BridgeOrchestrator } from "@roo-code/cloud"
 // api
 import { ApiHandler, ApiHandlerCreateMessageMetadata, buildApiHandler } from "../../api"
 import { ApiStream } from "../../api/transform/stream"
+import { maybeRemoveImageBlocks } from "../../api/transform/image-cleaning"
 
 // shared
 import { findLastIndex } from "../../shared/array"
@@ -79,6 +79,7 @@ import { SYSTEM_PROMPT } from "../prompts/system"
 
 // core modules
 import { ToolRepetitionDetector } from "../tools/ToolRepetitionDetector"
+import { restoreTodoListForTask } from "../tools/updateTodoListTool"
 import { FileContextTracker } from "../context-tracking/FileContextTracker"
 import { RooIgnoreController } from "../ignore/RooIgnoreController"
 import { RooProtectedController } from "../protect/RooProtectedController"
@@ -88,7 +89,14 @@ import { truncateConversationIfNeeded } from "../sliding-window"
 import { ClineProvider } from "../webview/ClineProvider"
 import { MultiSearchReplaceDiffStrategy } from "../diff/strategies/multi-search-replace"
 import { MultiFileSearchReplaceDiffStrategy } from "../diff/strategies/multi-file-search-replace"
-import { readApiMessages, saveApiMessages, readTaskMessages, saveTaskMessages, taskMetadata } from "../task-persistence"
+import {
+	type ApiMessage,
+	readApiMessages,
+	saveApiMessages,
+	readTaskMessages,
+	saveTaskMessages,
+	taskMetadata,
+} from "../task-persistence"
 import { getEnvironmentDetails } from "../environment/getEnvironmentDetails"
 import { checkContextWindowExceededError } from "../context/context-management/context-error-handling"
 import {
@@ -100,12 +108,11 @@ import {
 	checkpointDiff,
 } from "../checkpoints"
 import { processUserContentMentions } from "../mentions/processUserContentMentions"
-import { ApiMessage } from "../task-persistence/apiMessages"
 import { getMessagesSinceLastSummary, summarizeConversation } from "../condense"
-import { maybeRemoveImageBlocks } from "../../api/transform/image-cleaning"
-import { restoreTodoListForTask } from "../tools/updateTodoListTool"
-import { AutoApprovalHandler } from "./AutoApprovalHandler"
 import { Gpt5Metadata, ClineMessageWithMetadata } from "./types"
+import { MessageQueueService } from "../message-queue/MessageQueueService"
+
+import { AutoApprovalHandler } from "./AutoApprovalHandler"
 
 const MAX_EXPONENTIAL_BACKOFF_SECONDS = 600 // 10 minutes
 const DEFAULT_USAGE_COLLECTION_TIMEOUT_MS = 5000 // 5 seconds
@@ -259,6 +266,9 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 	// Task Bridge
 	enableBridge: boolean
 
+	// Message Queue Service
+	public readonly messageQueueService: MessageQueueService
+
 	// Streaming
 	isWaitingForFirstChunk = false
 	isStreaming = false
@@ -356,9 +366,12 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 			TelemetryService.instance.captureTaskCreated(this.taskId)
 		}
 
-		// Initialize the assistant message parser
+		// Initialize the assistant message parser.
 		this.assistantMessageParser = new AssistantMessageParser()
 
+		this.messageQueueService = new MessageQueueService()
+		this.messageQueueService.on("stateChanged", () => this.providerRef.deref()?.postStateToWebview())
+
 		// Only set up diff strategy if diff is enabled.
 		if (this.diffEnabled) {
 			// Default to old strategy, will be updated if experiment is enabled.
@@ -759,10 +772,14 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 		// The state is mutable if the message is complete and the task will
 		// block (via the `pWaitFor`).
 		const isBlocking = !(this.askResponse !== undefined || this.lastMessageTs !== askTs)
-		const isStatusMutable = !partial && isBlocking
+		const isMessageQueued = !this.messageQueueService.isEmpty()
+		const isStatusMutable = !partial && isBlocking && !isMessageQueued
 		let statusMutationTimeouts: NodeJS.Timeout[] = []
+		let messageQueueTimeout: NodeJS.Timeout | undefined
 
 		if (isStatusMutable) {
+			console.log(`Task#ask will block -> type: ${type}`)
+
 			if (isInteractiveAsk(type)) {
 				statusMutationTimeouts.push(
 					setTimeout(() => {
@@ -797,9 +814,19 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 					}, 1_000),
 				)
 			}
+		} else if (isMessageQueued) {
+			console.log("Task#ask will process message queue after a timeout")
+
+			messageQueueTimeout = setTimeout(() => {
+				const message = this.messageQueueService.dequeueMessage()
+
+				if (message) {
+					this.submitUserMessage(message.text, message.images)
+				}
+			}, 1_000)
 		}
 
-		// Wait for askResponse to be set
+		// Wait for askResponse to be set.
 		await pWaitFor(() => this.askResponse !== undefined || this.lastMessageTs !== askTs, { interval: 100 })
 
 		if (this.lastMessageTs !== askTs) {
@@ -817,6 +844,10 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 		// Cancel the timeouts if they are still running.
 		statusMutationTimeouts.forEach((timeout) => clearTimeout(timeout))
 
+		if (messageQueueTimeout) {
+			clearTimeout(messageQueueTimeout)
+		}
+
 		// Switch back to an active state.
 		if (this.idleAsk || this.resumableAsk || this.interactiveAsk) {
 			this.idleAsk = undefined
@@ -1406,8 +1437,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 			newUserContent.push(...formatResponse.imageBlocks(responseImages))
 		}
 
-		// Ensure we have at least some content to send to the API
-		// If newUserContent is empty, add a minimal resumption message
+		// Ensure we have at least some content to send to the API.
+		// If newUserContent is empty, add a minimal resumption message.
 		if (newUserContent.length === 0) {
 			newUserContent.push({
 				type: "text",
@@ -1417,14 +1448,20 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 
 		await this.overwriteApiConversationHistory(modifiedApiConversationHistory)
 
-		// Task resuming from history item
-
+		// Task resuming from history item.
 		await this.initiateTaskLoop(newUserContent)
 	}
 
 	public dispose(): void {
 		console.log(`[Task#dispose] disposing task ${this.taskId}.${this.instanceId}`)
 
+		// Dispose message queue.
+		try {
+			this.messageQueueService.dispose()
+		} catch (error) {
+			console.error("Error disposing message queue:", error)
+		}
+
 		// Remove all event listeners to prevent memory leaks.
 		try {
 			this.removeAllListeners()