Switch from Redis to a Socket.IO bridge (#6930)

Author: cte

src/core/task/Task.ts

@@ -107,6 +107,7 @@ export type TaskOptions = {
 	apiConfiguration: ProviderSettings
 	enableDiff?: boolean
 	enableCheckpoints?: boolean
+	enableTaskBridge?: boolean
 	fuzzyMatchThreshold?: number
 	consecutiveMistakeLimit?: number
 	task?: string
@@ -118,7 +119,6 @@ export type TaskOptions = {
 	parentTask?: Task
 	taskNumber?: number
 	onCreated?: (task: Task) => void
-	enableTaskBridge?: boolean
 }
 
 export class Task extends EventEmitter<TaskEvents> implements TaskLike {
@@ -239,7 +239,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 	checkpointServiceInitializing = false
 
 	// Task Bridge
-	taskBridgeService?: TaskBridgeService
+	enableTaskBridge: boolean
+	taskBridgeService: TaskBridgeService | null = null
 
 	// Streaming
 	isWaitingForFirstChunk = false
@@ -264,6 +265,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 		apiConfiguration,
 		enableDiff = false,
 		enableCheckpoints = true,
+		enableTaskBridge = false,
 		fuzzyMatchThreshold = 1.0,
 		consecutiveMistakeLimit = DEFAULT_CONSECUTIVE_MISTAKE_LIMIT,
 		task,
@@ -274,7 +276,6 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 		parentTask,
 		taskNumber = -1,
 		onCreated,
-		enableTaskBridge = false,
 	}: TaskOptions) {
 		super()
 
@@ -313,6 +314,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 		this.globalStoragePath = provider.context.globalStorageUri.fsPath
 		this.diffViewProvider = new DiffViewProvider(this.cwd, this)
 		this.enableCheckpoints = enableCheckpoints
+		this.enableTaskBridge = enableTaskBridge
 
 		this.rootTask = rootTask
 		this.parentTask = parentTask
@@ -352,11 +354,6 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 
 		this.toolRepetitionDetector = new ToolRepetitionDetector(this.consecutiveMistakeLimit)
 
-		// Initialize TaskBridgeService only if enabled
-		if (enableTaskBridge) {
-			this.taskBridgeService = TaskBridgeService.getInstance()
-		}
-
 		onCreated?.(this)
 
 		if (startTask) {
@@ -982,9 +979,20 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 	// Start / Abort / Resume
 
 	private async startTask(task?: string, images?: string[]): Promise<void> {
-		if (this.taskBridgeService) {
-			await this.taskBridgeService.initialize()
-			await this.taskBridgeService.subscribeToTask(this)
+		if (this.enableTaskBridge && CloudService.hasInstance()) {
+			if (!this.taskBridgeService) {
+				const bridgeConfig = await CloudService.instance.cloudAPI?.bridgeConfig()
+
+				if (bridgeConfig) {
+					this.taskBridgeService = await TaskBridgeService.createInstance({
+						...bridgeConfig,
+					})
+				}
+			}
+
+			if (this.taskBridgeService) {
+				await this.taskBridgeService.subscribeToTask(this)
+			}
 		}
 
 		// `conversationHistory` (for API) and `clineMessages` (for webview)
@@ -1038,9 +1046,20 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 	}
 
 	private async resumeTaskFromHistory() {
-		if (this.taskBridgeService) {
-			await this.taskBridgeService.initialize()
-			await this.taskBridgeService.subscribeToTask(this)
+		if (this.enableTaskBridge && CloudService.hasInstance()) {
+			if (!this.taskBridgeService) {
+				const bridgeConfig = await CloudService.instance.cloudAPI?.bridgeConfig()
+
+				if (bridgeConfig) {
+					this.taskBridgeService = await TaskBridgeService.createInstance({
+						...bridgeConfig,
+					})
+				}
+			}
+
+			if (this.taskBridgeService) {
+				await this.taskBridgeService.subscribeToTask(this)
+			}
 		}
 
 		const modifiedClineMessages = await this.getSavedClineMessages()
@@ -1293,6 +1312,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
 			this.taskBridgeService
 				.unsubscribeFromTask(this.taskId)
 				.catch((error) => console.error("Error unsubscribing from task bridge:", error))
+			this.taskBridgeService = null
 		}
 
 		// Release any terminals associated with this task.

src/core/webview/ClineProvider.ts

@@ -724,9 +724,6 @@ export class ClineProvider
 			throw new OrganizationAllowListViolationError(t("common:errors.violated_organization_allowlist"))
 		}
 
-		// Determine if TaskBridge should be enabled
-		const enableTaskBridge = isRemoteControlEnabled(cloudUserInfo, remoteControlEnabled)
-
 		const task = new Task({
 			provider: this,
 			apiConfiguration,
@@ -741,7 +738,7 @@ export class ClineProvider
 			parentTask,
 			taskNumber: this.clineStack.length + 1,
 			onCreated: this.taskCreationCallback,
-			enableTaskBridge,
+			enableTaskBridge: isRemoteControlEnabled(cloudUserInfo, remoteControlEnabled),
 			...options,
 		})
 
@@ -2139,26 +2136,49 @@ export class ClineProvider
 	 * Handle remote control enabled/disabled state changes
 	 * Manages ExtensionBridgeService and TaskBridgeService lifecycle
 	 */
-	public async handleRemoteControlToggle(enabled: boolean): Promise<void> {
+	public async handleRemoteControlToggle(enabled: boolean) {
 		const {
 			CloudService: CloudServiceImport,
 			ExtensionBridgeService,
 			TaskBridgeService,
 		} = await import("@roo-code/cloud")
+
 		const userInfo = CloudServiceImport.instance.getUserInfo()
 
-		// Handle ExtensionBridgeService using static method
-		await ExtensionBridgeService.handleRemoteControlState(userInfo, enabled, this, (message: string) =>
-			this.log(message),
+		const bridgeConfig = await CloudServiceImport.instance.cloudAPI?.bridgeConfig()
+
+		if (!bridgeConfig) {
+			this.log("[CloudService] Failed to get bridge config")
+			return
+		}
+
+		ExtensionBridgeService.handleRemoteControlState(
+			userInfo,
+			enabled,
+			{ ...bridgeConfig, provider: this },
+			(message: string) => this.log(message),
 		)
 
 		if (isRemoteControlEnabled(userInfo, enabled)) {
-			// Set up TaskBridgeService for the currently active task if one exists
+			// Set up TaskBridgeService for the currently active task if one exists.
 			const currentTask = this.getCurrentCline()
-			if (currentTask && !currentTask.taskBridgeService) {
+
+			if (currentTask && !currentTask.taskBridgeService && CloudService.hasInstance()) {
 				try {
-					currentTask.taskBridgeService = TaskBridgeService.getInstance()
-					await currentTask.taskBridgeService.subscribeToTask(currentTask)
+					if (!currentTask.taskBridgeService) {
+						const bridgeConfig = await CloudService.instance.cloudAPI?.bridgeConfig()
+
+						if (bridgeConfig) {
+							currentTask.taskBridgeService = await TaskBridgeService.createInstance({
+								...bridgeConfig,
+							})
+						}
+					}
+
+					if (currentTask.taskBridgeService) {
+						await currentTask.taskBridgeService.subscribeToTask(currentTask)
+					}
+
 					this.log(`[TaskBridgeService] Subscribed current task ${currentTask.taskId} to TaskBridge`)
 				} catch (error) {
 					const message = `[TaskBridgeService#subscribeToTask] ${error instanceof Error ? error.message : String(error)}`
@@ -2167,12 +2187,12 @@ export class ClineProvider
 				}
 			}
 		} else {
-			// Disconnect TaskBridgeService for all tasks in the stack
+			// Disconnect TaskBridgeService for all tasks in the stack.
 			for (const task of this.clineStack) {
 				if (task.taskBridgeService) {
 					try {
 						await task.taskBridgeService.unsubscribeFromTask(task.taskId)
-						task.taskBridgeService = undefined
+						task.taskBridgeService = null
 						this.log(`[TaskBridgeService] Unsubscribed task ${task.taskId} from TaskBridge`)
 					} catch (error) {
 						const message = `[TaskBridgeService#unsubscribeFromTask] for task ${task.taskId}: ${error instanceof Error ? error.message : String(error)}`
@@ -2181,6 +2201,8 @@ export class ClineProvider
 					}
 				}
 			}
+
+			TaskBridgeService.resetInstance()
 		}
 	}
 

src/extension.ts

@@ -131,15 +131,21 @@ export async function activate(context: vscode.ExtensionContext) {
 	cloudService.on("auth-state-changed", postStateListener)
 	cloudService.on("settings-updated", postStateListener)
 
-	cloudService.on("user-info", ({ userInfo }) => {
+	cloudService.on("user-info", async ({ userInfo }) => {
 		postStateListener()
 
-		// Check if remote control is enabled in user settings
-		const remoteControlEnabled = contextProxy.getValue("remoteControlEnabled")
+		const bridgeConfig = await cloudService.cloudAPI?.bridgeConfig()
 
-		// Handle ExtensionBridgeService state using static method
-		ExtensionBridgeService.handleRemoteControlState(userInfo, remoteControlEnabled, provider, (message: string) =>
-			outputChannel.appendLine(message),
+		if (!bridgeConfig) {
+			outputChannel.appendLine("[CloudService] Failed to get bridge config")
+			return
+		}
+
+		ExtensionBridgeService.handleRemoteControlState(
+			userInfo,
+			contextProxy.getValue("remoteControlEnabled"),
+			{ ...bridgeConfig, provider },
+			(message: string) => outputChannel.appendLine(message),
 		)
 	})