diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index e573a7f107..a5995f576a 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -154,6 +154,9 @@ import { 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 @@ export class Task extends EventEmitter implements TaskLike { * 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 | 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 @@ export class Task extends EventEmitter implements TaskLike { // 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 @@ export class Task extends EventEmitter implements TaskLike { 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 @@ export class Task extends EventEmitter implements TaskLike { 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 @@ export class Task extends EventEmitter implements TaskLike { // 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 @@ export class Task extends EventEmitter implements TaskLike { private async disposeOnce(): Promise { 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 diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index ad07230204..695248c755 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/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> = [] + 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({