[Perf] Throttle partial message updates while streaming

src/core/task/Task.ts

@@ -154,6 +154,9 @@
 
 const MAX_EXPONENTIAL_BACKOFF_SECONDS = 600 // 10 minutes
 const DEFAULT_USAGE_COLLECTION_TIMEOUT_MS = 5000 // 5 seconds
+// Streamed deltas arrive far faster than the webview can usefully re-render a growing message
+// (tables and code fences reflow on nearly every update), so partial updates post at most this often.
+const PARTIAL_MESSAGE_UPDATE_INTERVAL_MS = 120
 // Upper bound on awaiting lazily loaded model metadata. Some model-catalog
 // fetchers (e.g. OpenRouter's bare axios GET) have no request timeout, so a
 // hung endpoint must not stall streaming, condense, or context-window
@@ -1734,6 +1737,65 @@
 	 * Updates a Cline message in the webview and emits an event.
 	 * Non-partial messages are synced to cloud telemetry if not already synced.
 	 */
+	private lastPartialMessagePostTs = 0
+	private pendingPartialMessageTimer: ReturnType<typeof setTimeout> | undefined
+	private pendingPartialMessage: ClineMessage | undefined
+
+	/**
+	 * Posts a streaming (partial) message update, at most once per
+	 * `PARTIAL_MESSAGE_UPDATE_INTERVAL_MS`. The message object is always kept current by the
+	 * caller; a trailing post makes sure the latest text is delivered once the burst ends.
+	 */
+	private postPartialMessageUpdate(message: ClineMessage, source: "ask" | "say"): void {
+		const post = (target: ClineMessage) => {
+			this.lastPartialMessagePostTs = Date.now()
+			// Fire-and-forget: the `RooCodeEventName.Message` emit can synchronously throw via a
+			// consumer-attached listener. Surface that as a log, not an unhandled rejection.
+			this.updateClineMessage(target).catch((error) => {
+				console.error(`[Task#${source}] updateClineMessage failed:`, error)
+			})
+		}
+
+		// A queued update for a different message (e.g. reasoning followed by text) must not be
+		// overwritten by this one, so deliver it now.
+		const queued = this.pendingPartialMessage
+		if (queued && queued.ts !== message.ts) {
+			this.cancelPendingPartialMessageUpdate()
+			post(queued)
+		}
+
+		const elapsed = Date.now() - this.lastPartialMessagePostTs
+		if (elapsed >= PARTIAL_MESSAGE_UPDATE_INTERVAL_MS) {
+			this.cancelPendingPartialMessageUpdate()
+			post(message)
+			return
+		}
+
+		this.pendingPartialMessage = message
+		if (this.pendingPartialMessageTimer) {
+			return
+		}
+
+		this.pendingPartialMessageTimer = setTimeout(() => {
+			const pending = this.pendingPartialMessage
+			this.pendingPartialMessageTimer = undefined
+			this.pendingPartialMessage = undefined
+
+			if (pending?.partial && !this.abort && !this.abandoned) {
+				post(pending)
+			}
+		}, PARTIAL_MESSAGE_UPDATE_INTERVAL_MS - elapsed)
+	}
+
+	/** Drops a queued trailing partial post, e.g. because the message was finalized or the task is ending. */
+	private cancelPendingPartialMessageUpdate(): void {
+		if (this.pendingPartialMessageTimer) {
+			clearTimeout(this.pendingPartialMessageTimer)
+		}
+		this.pendingPartialMessageTimer = undefined
+		this.pendingPartialMessage = undefined
+	}
+
 	private async updateClineMessage(message: ClineMessage) {
 		const provider = this.providerRef.deref()
 		await provider?.postMessageToWebview({ type: "messageUpdated", clineMessage: message })
@@ -1941,13 +2003,7 @@
 					// data or one whole message at a time so ignore partial for
 					// saves, and only post parts of partial message instead of
 					// whole array in new listener.
-					// Fire-and-forget: the webview post is internally guarded, but
-					// the `RooCodeEventName.Message` emit can synchronously throw
-					// if any consumer-attached listener does, which would surface
-					// here as an unhandled rejection. Log it instead.
-					this.updateClineMessage(lastMessage).catch((error) => {
-						console.error("[Task#ask] updateClineMessage failed:", error)
-					})
+					this.postPartialMessageUpdate(lastMessage, "ask")
 					// console.log("Task#ask: current ask promise was ignored (#1)")
 					throw new AskIgnoredError("updating existing partial")
 				} else {
@@ -1988,9 +2044,9 @@
 						lastMessage.isAnswered = true
 						lastMessage.autoApprovalDecision = autoApprovalDecision
 					}
+					this.cancelPendingPartialMessageUpdate()
 					await this.saveClineMessages()
-					// Fire-and-forget: see updateClineMessage call above for the
-					// rationale on the .catch arm.
+					// Fire-and-forget: see postPartialMessageUpdate for the rationale on the .catch arm.
 					this.updateClineMessage(lastMessage).catch((error) => {
 						console.error("[Task#ask] updateClineMessage failed:", error)
 					})
@@ -2779,13 +2835,7 @@
 					lastMessage.images = images
 					lastMessage.partial = partial
 					lastMessage.progressStatus = progressStatus
-					// Fire-and-forget: webview post is internally guarded, but the
-					// `RooCodeEventName.Message` emit can synchronously throw via a
-					// consumer-attached listener. Surface that as a log, not an
-					// unhandled rejection.
-					this.updateClineMessage(lastMessage).catch((error) => {
-						console.error("[Task#say] updateClineMessage failed:", error)
-					})
+					this.postPartialMessageUpdate(lastMessage, "say")
 				} else {
 					// This is a new partial message, so add it with partial state.
 					const sayTs = Date.now()
@@ -2821,6 +2871,7 @@
 
 					// Instead of streaming partialMessage events, we do a save
 					// and post like normal to persist to disk.
+					this.cancelPendingPartialMessageUpdate()
 					await this.saveClineMessages()
 
 					// More performant than an entire `postStateToWebview`.
@@ -3623,6 +3674,7 @@
 	private async disposeOnce(): Promise<void> {
 		console.log(`[Task#dispose] disposing task ${this.taskId}.${this.instanceId}`)
 		this.cancelAssistantMessagePersistence()
+		this.cancelPendingPartialMessageUpdate()
 
 		// Release handler-owned host resources (e.g. VS Code configuration listeners).
 		// This is the fallback for leases that never settle; later lease releases

src/core/task/__tests__/Task.spec.ts

@@ -4366,6 +4366,154 @@ describe("Cline", () => {
 		})
 	})
 
+	describe("partial message update throttling", () => {
+		function createStreamingTask() {
+			// Task mutates one ClineMessage across deltas and finalization, so snapshot the
+			// fields of each messageUpdated post when it happens instead of keeping the reference.
+			const messageUpdates: Array<Pick<ClineMessage, "ts" | "text" | "partial">> = []
+			vi.mocked(mockProvider.postMessageToWebview).mockImplementation(async (message) => {
+				if (message.type === "messageUpdated" && message.clineMessage) {
+					const { ts, text, partial } = message.clineMessage
+					messageUpdates.push({ ts, text, partial })
+				}
+			})
+
+			const task = new Task({
+				provider: mockProvider,
+				apiConfiguration: mockApiConfig,
+				task: "test task",
+				startTask: false,
+			})
+			vi.spyOn(getTaskTestAccess(task), "saveClineMessages").mockResolvedValue(true)
+			return { task, messageUpdates }
+		}
+
+		beforeEach(() => {
+			vi.useFakeTimers()
+		})
+
+		afterEach(() => {
+			vi.useRealTimers()
+		})
+
+		it("coalesces a burst of partial updates and delivers the latest text last", async () => {
+			const { task, messageUpdates } = createStreamingTask()
+
+			await task.say("text", "chunk 0", undefined, true)
+			for (let i = 1; i <= 50; i++) {
+				await task.say("text", `chunk ${i}`, undefined, true)
+				vi.advanceTimersByTime(5)
+			}
+			await vi.advanceTimersByTimeAsync(500)
+
+			// 50 deltas over ~250ms: far fewer than one post per delta.
+			expect(messageUpdates.length).toBeGreaterThan(0)
+			expect(messageUpdates.length).toBeLessThan(10)
+			expect(messageUpdates.at(-1)).toMatchObject({ text: "chunk 50", partial: true })
+		})
+
+		it("lets the finalized message through and drops the queued partial post", async () => {
+			const { task, messageUpdates } = createStreamingTask()
+
+			await task.say("text", "first", undefined, true)
+			await task.say("text", "second", undefined, true)
+			await task.say("text", "third", undefined, true)
+			// "third" is queued behind the throttle; finalizing must cancel that queued post.
+			expect(vi.getTimerCount()).toBe(1)
+			await task.say("text", "final text", undefined, false)
+			expect(vi.getTimerCount()).toBe(0)
+			await vi.advanceTimersByTimeAsync(500)
+
+			expect(messageUpdates.map(({ text, partial }) => ({ text, partial }))).toEqual([
+				{ text: "second", partial: true },
+				{ text: "final text", partial: false },
+			])
+		})
+
+		it("drops a queued partial ask update when the ask is finalized", async () => {
+			const { task, messageUpdates } = createStreamingTask()
+
+			// Partial asks always reject with AskIgnoredError.
+			await task.ask("tool", "first", true).catch(() => {})
+			await task.ask("tool", "second", true).catch(() => {})
+			await task.ask("tool", "third", true).catch(() => {})
+			expect(task["pendingPartialMessageTimer"]).toBeDefined()
+
+			// A complete ask resolves only after a response, so leave it pending. Step 1ms per check
+			// so the final post lands well before the queued timer would fire on its own.
+			void task.ask("tool", "final", false).catch(() => {})
+			await vi.waitFor(() => expect(messageUpdates.at(-1)?.partial).toBe(false), { interval: 1 })
+			expect(task["pendingPartialMessageTimer"]).toBeUndefined()
+			await vi.advanceTimersByTimeAsync(500)
+
+			expect(messageUpdates.map(({ text, partial }) => ({ text, partial }))).toEqual([
+				{ text: "second", partial: true },
+				{ text: "final", partial: false },
+			])
+		})
+
+		it("drops a queued partial update when the task is disposed", async () => {
+			const { task, messageUpdates } = createStreamingTask()
+
+			await task.say("text", "first", undefined, true)
+			await task.say("text", "second", undefined, true)
+			await task.say("text", "third", undefined, true)
+			expect(task["pendingPartialMessageTimer"]).toBeDefined()
+
+			await task.dispose()
+			expect(task["pendingPartialMessageTimer"]).toBeUndefined()
+			await vi.advanceTimersByTimeAsync(500)
+
+			expect(messageUpdates.map(({ text }) => text)).toEqual(["second"])
+		})
+
+		it("holds a queued update until the 120ms interval elapses", async () => {
+			const { task, messageUpdates } = createStreamingTask()
+
+			await task.say("text", "first", undefined, true)
+			await task.say("text", "second", undefined, true)
+			await task.say("text", "third", undefined, true)
+			expect(messageUpdates.map(({ text }) => text)).toEqual(["second"])
+
+			await vi.advanceTimersByTimeAsync(119)
+			expect(messageUpdates.map(({ text }) => text)).toEqual(["second"])
+
+			await vi.advanceTimersByTimeAsync(1)
+			expect(messageUpdates.map(({ text }) => text)).toEqual(["second", "third"])
+		})
+
+		it("delivers a queued update for one message before updating a different message", async () => {
+			const { task, messageUpdates } = createStreamingTask()
+
+			await task.say("reasoning", "r1", undefined, true)
+			await task.say("reasoning", "r2", undefined, true)
+			await task.say("reasoning", "r3", undefined, true)
+			expect(messageUpdates.map(({ text }) => text)).toEqual(["r2"])
+
+			// A new message starts, then streams: its update must not replace the queued "r3".
+			// Message timestamps are Date.now(), which fake timers freeze, so step the clock.
+			await vi.advanceTimersByTimeAsync(1)
+			await task.say("text", "t1", undefined, true)
+			await task.say("text", "t2", undefined, true)
+			expect(messageUpdates.map(({ text }) => text)).toEqual(["r2", "r3"])
+
+			await vi.advanceTimersByTimeAsync(500)
+			expect(messageUpdates.map(({ text }) => text)).toEqual(["r2", "r3", "t2"])
+		})
+
+		it("posts updates that are far enough apart immediately", async () => {
+			const { task, messageUpdates } = createStreamingTask()
+
+			await task.say("text", "one", undefined, true)
+			await task.say("text", "two", undefined, true)
+			await vi.advanceTimersByTimeAsync(1000)
+			await task.say("text", "three", undefined, true)
+
+			// "one" creates the message; "two" and "three" are each the first update in their window.
+			expect(messageUpdates.map(({ text }) => text)).toEqual(["two", "three"])
+		})
+	})
+
 	describe("abortTask", () => {
 		it("should set abort flag and emit TaskAborted event", async () => {
 			const task = new Task({