Move message queue to the extension host (RooCodeInc#7604)

Co-authored-by: roomote[bot] <219738659+roomote[bot]@users.noreply.github.com>

Author: 2 people

packages/cloud/src/bridge/ExtensionChannel.ts

@@ -175,6 +175,7 @@ export class ExtensionChannel extends BaseChannel<
 			{ from: RooCodeEventName.TaskInteractive, to: ExtensionBridgeEventName.TaskInteractive },
 			{ from: RooCodeEventName.TaskResumable, to: ExtensionBridgeEventName.TaskResumable },
 			{ from: RooCodeEventName.TaskIdle, to: ExtensionBridgeEventName.TaskIdle },
+			{ from: RooCodeEventName.TaskUserMessage, to: ExtensionBridgeEventName.TaskUserMessage },
 		] as const
 
 		eventMapping.forEach(({ from, to }) => {
@@ -220,6 +221,8 @@ export class ExtensionChannel extends BaseChannel<
 				? {
 						taskId: task.taskId,
 						taskStatus: task.taskStatus,
+						taskAsk: task?.taskAsk,
+						queuedMessages: task.queuedMessages,
 						...task.metadata,
 					}
 				: { taskId: "", taskStatus: TaskStatus.None },

packages/cloud/src/bridge/__tests__/ExtensionChannel.test.ts

@@ -101,6 +101,7 @@ describe("ExtensionChannel", () => {
 				RooCodeEventName.TaskInteractive,
 				RooCodeEventName.TaskResumable,
 				RooCodeEventName.TaskIdle,
+				RooCodeEventName.TaskUserMessage,
 			]
 
 			// Check that on() was called for each event
@@ -230,7 +231,7 @@ describe("ExtensionChannel", () => {
 			}
 
 			// Listeners should still be the same count (not accumulated)
-			const expectedEventCount = 10 // Number of events we listen to
+			const expectedEventCount = 11 // Number of events we listen to (including TaskUserMessage)
 			expect(eventListeners.size).toBe(expectedEventCount)
 
 			// Each event should have exactly 1 listener

packages/types/npm/package.metadata.json

@@ -1,6 +1,6 @@
 {
 	"name": "@roo-code/types",
-	"version": "1.66.0",
+	"version": "1.67.0",
 	"description": "TypeScript type definitions for Roo Code.",
 	"publishConfig": {
 		"access": "public",

packages/types/src/cloud.ts

@@ -7,7 +7,7 @@ import { TaskStatus, taskMetadataSchema } from "./task.js"
 import { globalSettingsSchema } from "./global-settings.js"
 import { providerSettingsWithIdSchema } from "./provider-settings.js"
 import { mcpMarketplaceItemSchema } from "./marketplace.js"
-import { clineMessageSchema } from "./message.js"
+import { clineMessageSchema, queuedMessageSchema } from "./message.js"
 import { staticAppPropertiesSchema, gitPropertiesSchema } from "./telemetry.js"
 
 /**
@@ -359,6 +359,8 @@ export const INSTANCE_TTL_SECONDS = 60
 const extensionTaskSchema = z.object({
 	taskId: z.string(),
 	taskStatus: z.nativeEnum(TaskStatus),
+	taskAsk: clineMessageSchema.optional(),
+	queuedMessages: z.array(queuedMessageSchema).optional(),
 	...taskMetadataSchema.shape,
 })
 
@@ -402,6 +404,8 @@ export enum ExtensionBridgeEventName {
 	TaskResumable = RooCodeEventName.TaskResumable,
 	TaskIdle = RooCodeEventName.TaskIdle,
 
+	TaskUserMessage = RooCodeEventName.TaskUserMessage,
+
 	ModeChanged = RooCodeEventName.ModeChanged,
 	ProviderProfileChanged = RooCodeEventName.ProviderProfileChanged,
 
@@ -461,31 +465,39 @@ export const extensionBridgeEventSchema = z.discriminatedUnion("type", [
 		instance: extensionInstanceSchema,
 		timestamp: z.number(),
 	}),
+
 	z.object({
-		type: z.literal(ExtensionBridgeEventName.InstanceRegistered),
+		type: z.literal(ExtensionBridgeEventName.TaskUserMessage),
 		instance: extensionInstanceSchema,
 		timestamp: z.number(),
 	}),
+
 	z.object({
-		type: z.literal(ExtensionBridgeEventName.InstanceUnregistered),
+		type: z.literal(ExtensionBridgeEventName.ModeChanged),
 		instance: extensionInstanceSchema,
+		mode: z.string(),
 		timestamp: z.number(),
 	}),
 	z.object({
-		type: z.literal(ExtensionBridgeEventName.HeartbeatUpdated),
+		type: z.literal(ExtensionBridgeEventName.ProviderProfileChanged),
 		instance: extensionInstanceSchema,
+		providerProfile: z.object({ name: z.string(), provider: z.string().optional() }),
 		timestamp: z.number(),
 	}),
+
 	z.object({
-		type: z.literal(ExtensionBridgeEventName.ModeChanged),
+		type: z.literal(ExtensionBridgeEventName.InstanceRegistered),
 		instance: extensionInstanceSchema,
-		mode: z.string(),
 		timestamp: z.number(),
 	}),
 	z.object({
-		type: z.literal(ExtensionBridgeEventName.ProviderProfileChanged),
+		type: z.literal(ExtensionBridgeEventName.InstanceUnregistered),
+		instance: extensionInstanceSchema,
+		timestamp: z.number(),
+	}),
+	z.object({
+		type: z.literal(ExtensionBridgeEventName.HeartbeatUpdated),
 		instance: extensionInstanceSchema,
-		providerProfile: z.object({ name: z.string(), provider: z.string().optional() }),
 		timestamp: z.number(),
 	}),
 ])

packages/types/src/events.ts

@@ -31,6 +31,7 @@ export enum RooCodeEventName {
 	Message = "message",
 	TaskModeSwitched = "taskModeSwitched",
 	TaskAskResponded = "taskAskResponded",
+	TaskUserMessage = "taskUserMessage",
 
 	// Task Analytics
 	TaskTokenUsageUpdated = "taskTokenUsageUpdated",
@@ -82,6 +83,7 @@ export const rooCodeEventsSchema = z.object({
 	]),
 	[RooCodeEventName.TaskModeSwitched]: z.tuple([z.string(), z.string()]),
 	[RooCodeEventName.TaskAskResponded]: z.tuple([z.string()]),
+	[RooCodeEventName.TaskUserMessage]: z.tuple([z.string()]),
 
 	[RooCodeEventName.TaskToolFailed]: z.tuple([z.string(), toolNamesSchema, z.string()]),
 	[RooCodeEventName.TaskTokenUsageUpdated]: z.tuple([z.string(), tokenUsageSchema]),

packages/types/src/message.ts

@@ -248,14 +248,11 @@ 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 */
-	id: string
-	/** The text content of the message */
-	text: string
-	/** Array of image data URLs attached to the message */
-	images: string[]
-}
+export const queuedMessageSchema = z.object({
+	timestamp: z.number(),
+	id: z.string(),
+	text: z.string(),
+	images: z.array(z.string()).optional(),
+})
+
+export type QueuedMessage = z.infer<typeof queuedMessageSchema>

packages/types/src/task.ts

@@ -2,7 +2,7 @@ import { z } from "zod"
 
 import { RooCodeEventName } from "./events.js"
 import type { RooCodeSettings } from "./global-settings.js"
-import type { ClineMessage, TokenUsage } from "./message.js"
+import type { ClineMessage, QueuedMessage, TokenUsage } from "./message.js"
 import type { ToolUsage, ToolName } from "./tool.js"
 import type { StaticAppProperties, GitProperties, TelemetryProperties } from "./telemetry.js"
 import type { TodoItem } from "./todo.js"
@@ -59,8 +59,6 @@ export interface TaskProviderLike {
 
 export type TaskProviderEvents = {
 	[RooCodeEventName.TaskCreated]: [task: TaskLike]
-
-	// Proxied from the Task EventEmitter.
 	[RooCodeEventName.TaskStarted]: [taskId: string]
 	[RooCodeEventName.TaskCompleted]: [taskId: string, tokenUsage: TokenUsage, toolUsage: ToolUsage]
 	[RooCodeEventName.TaskAborted]: [taskId: string]
@@ -71,6 +69,9 @@ export type TaskProviderEvents = {
 	[RooCodeEventName.TaskResumable]: [taskId: string]
 	[RooCodeEventName.TaskIdle]: [taskId: string]
 	[RooCodeEventName.TaskSpawned]: [taskId: string]
+
+	[RooCodeEventName.TaskUserMessage]: [taskId: string]
+
 	[RooCodeEventName.ModeChanged]: [mode: string]
 	[RooCodeEventName.ProviderProfileChanged]: [config: { name: string; provider?: string }]
 }
@@ -105,11 +106,12 @@ export type TaskMetadata = z.infer<typeof taskMetadataSchema>
 
 export interface TaskLike {
 	readonly taskId: string
-	readonly taskStatus: TaskStatus
-	readonly taskAsk: ClineMessage | undefined
+	readonly rootTask?: TaskLike
 	readonly metadata: TaskMetadata
 
-	readonly rootTask?: TaskLike
+	readonly taskStatus: TaskStatus
+	readonly taskAsk: ClineMessage | undefined
+	readonly queuedMessages: QueuedMessage[]
 
 	on<K extends keyof TaskEvents>(event: K, listener: (...args: TaskEvents[K]) => void | Promise<void>): this
 	off<K extends keyof TaskEvents>(event: K, listener: (...args: TaskEvents[K]) => void | Promise<void>): this
@@ -141,6 +143,7 @@ export type TaskEvents = {
 	[RooCodeEventName.Message]: [{ action: "created" | "updated"; message: ClineMessage }]
 	[RooCodeEventName.TaskModeSwitched]: [taskId: string, mode: string]
 	[RooCodeEventName.TaskAskResponded]: []
+	[RooCodeEventName.TaskUserMessage]: [taskId: string]
 
 	// Task Analytics
 	[RooCodeEventName.TaskToolFailed]: [taskId: string, tool: ToolName, error: string]

src/core/message-queue/MessageQueueService.ts

@@ -0,0 +1,98 @@
+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
+		}
+
+		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
+		}
+
+		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"