Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 23 additions & 6 deletions packages/types/src/vscode-extension-host.ts
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,12 @@ export interface ExtensionMessage {
| "theme"
| "workspaceUpdated"
| "invoke"
| "messageUpdated"
| "clineMessageAppended"
| "clineMessageUpdated"
| "clineMessagesSnapshotStart"
| "clineMessagesSnapshotChunk"
| "clineMessagesSnapshotEnd"
| "messageUpdated" // Legacy: a patched webview requests a full resync instead of applying this.
| "mcpServers"
| "enhancedPrompt"
| "commitSearchResults"
Expand Down Expand Up @@ -138,7 +143,13 @@ export interface ExtensionMessage {
isActive: boolean
path?: string
}>
taskId?: string
clineMessage?: ClineMessage
clineMessages?: ClineMessage[]
clineMessagesSeq?: number
snapshotId?: string
snapshotStartIndex?: number
snapshotTotal?: number
routerModels?: RouterModels
openAiModels?: string[]
ollamaModels?: ModelRecord
Expand Down Expand Up @@ -334,7 +345,11 @@ export type ExtensionState = Pick<
lockApiConfigAcrossModes?: boolean
version: string
clineMessages: ClineMessage[]
currentTaskId?: string
/**
* Focused task identity. Omitted means this partial state update does not
* change task focus; null authoritatively means no task is focused.
*/
currentTaskId?: string | null
currentTaskItem?: HistoryItem
currentTaskTodos?: TodoItem[] // Initial todos for the current task
apiConfiguration: ProviderSettings
Expand Down Expand Up @@ -426,10 +441,9 @@ export type ExtensionState = Pick<
arch?: string

/**
* Monotonically increasing sequence number for clineMessages state pushes.
* When present, the frontend should only apply clineMessages from a state push
* if its seq is greater than the last applied seq. This prevents stale state
* (captured during async getStateToPostToWebview) from overwriting newer messages.
* Last sequence applied by the dedicated task-scoped transcript transport.
* Generic `state` messages intentionally omit this field and `clineMessages`;
* snapshots and append/update messages carry both transcript data and sequence.
*/
clineMessagesSeq?: number
}
Expand Down Expand Up @@ -646,8 +660,11 @@ export interface WebviewMessage {
| "openRuleFile"
| "openRulesDirectory"
| "themeFixtureProbeResponse"
| "requestClineMessagesResync"
text?: string
taskId?: string
expectedSeq?: number
receivedSeq?: number
editedMessageContent?: string
tab?: "settings" | "history" | "mcp" | "modes" | "chat" | "marketplace" | "cloud"
disabled?: boolean
Expand Down
4 changes: 4 additions & 0 deletions src/__tests__/helpers/provider-stub.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,9 @@ import { type Task } from "../../core/task/Task"
type ProviderStubFields = {
delegationTransitionLocks?: Map<string, Promise<void>>
cancelledDelegationChildIds?: Set<string>
clineMessagesSeqByTaskId?: Map<string, number>
log?: ReturnType<typeof vi.fn>
syncFocusedTaskToWebview?: ReturnType<typeof vi.fn>
taskHistoryStore?: { get: (id: string) => unknown }
taskRegistry?: TaskRegistry
clineStack?: Task[]
Expand Down Expand Up @@ -36,7 +38,9 @@ export function makeProviderStub<T extends object>(stub: T): ClineProvider {
const proto = ClineProvider.prototype as unknown as PrivateProviderMethods
s.delegationTransitionLocks ??= new Map()
s.cancelledDelegationChildIds ??= new Set()
s.clineMessagesSeqByTaskId ??= new Map()
s.log ??= vi.fn()
s.syncFocusedTaskToWebview ??= vi.fn().mockResolvedValue(undefined)
s.taskHistoryStore ??= { get: () => undefined }

// Convert legacy clineStack array into a TaskRegistry
Expand Down
2 changes: 2 additions & 0 deletions src/__tests__/single-open-invariant.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -269,6 +269,7 @@ describe("Single-open-task invariant", () => {
taskScheduler: { schedule: schedulespy },
taskEventListeners: new WeakMap(),
performPreparationTasks: vi.fn().mockResolvedValue(undefined),
syncFocusedTaskToWebview: vi.fn().mockResolvedValue(undefined),
context: { extension: { packageJSON: {} }, globalStorageUri: { fsPath: "/tmp" } },
contextProxy: {
extensionUri: {},
Expand Down Expand Up @@ -341,6 +342,7 @@ describe("Single-open-task invariant", () => {
taskScheduler: { schedule: schedulespy },
taskEventListeners: new WeakMap(),
performPreparationTasks: vi.fn().mockResolvedValue(undefined),
syncFocusedTaskToWebview: vi.fn().mockResolvedValue(undefined),
context: { extension: { packageJSON: {} }, globalStorageUri: { fsPath: "/tmp" } },
contextProxy: {
extensionUri: {},
Expand Down
64 changes: 40 additions & 24 deletions src/core/task/Task.ts
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,7 @@ function queuedResponseForAsk(type: ClineAsk, text?: string): QueuedAskResolutio

const FORCED_CONTEXT_REDUCTION_PERCENT = 75 // Keep 75% of context (remove 25%) on context window errors
const MAX_CONTEXT_WINDOW_RETRIES = 3 // Maximum retries for context window errors
const PARTIAL_MESSAGE_UPDATE_DEBOUNCE_MS = 500

export interface TaskOptions extends CreateTaskOptions {
provider: ClineProvider
Expand Down Expand Up @@ -477,6 +478,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
// Token Usage Throttling - Debounced emit function
private readonly TOKEN_USAGE_EMIT_INTERVAL_MS = 2000 // 2 seconds
private debouncedEmitTokenUsage: ReturnType<typeof debounce>
private debouncedPostPartialMessageUpdate: ReturnType<typeof debounce>

// Historical cloud sync tracking retained only to avoid task resume churn.
private cloudSyncedMessageTimestamps: Set<number> = new Set()
Expand Down Expand Up @@ -640,6 +642,20 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
this.TOKEN_USAGE_EMIT_INTERVAL_MS,
{ leading: true, trailing: true, maxWait: this.TOKEN_USAGE_EMIT_INTERVAL_MS },
)
this.debouncedPostPartialMessageUpdate = debounce(
(message: ClineMessage) => {
const provider = this.providerRef.deref()
if (!provider) {
return
}

void provider.postClineMessageUpdated(this.taskId, message).catch((error) => {
console.error("[Task#updateClineMessage] incremental post failed:", error)
})
},
PARTIAL_MESSAGE_UPDATE_DEBOUNCE_MS,
{ leading: false, trailing: true },
)

onCreated?.(this)

Expand Down Expand Up @@ -1148,20 +1164,10 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
private async addToClineMessages(message: ClineMessage) {
this.clineMessages.push(message)
const provider = this.providerRef.deref()
// Unanswered asks must reach the webview before Message listeners can respond against its state.
const requiresImmediateState =
message.partial === true || (message.type === "ask" && message.isAnswered !== true)
try {
await provider?.postStateToWebviewThrottled()
await provider?.postClineMessageAppended(this.taskId, message)
} catch (error) {
console.error("[Task#addToClineMessages] postStateToWebviewThrottled failed:", error)
}
if (requiresImmediateState) {
try {
await provider?.flushPostStateToWebviewThrottled()
} catch (error) {
console.error("[Task#addToClineMessages] flushPostStateToWebviewThrottled failed:", error)
}
console.error("[Task#addToClineMessages] incremental post failed:", error)
}
this.emit(RooCodeEventName.Message, { action: "created", message })
await this.saveClineMessages()
Expand Down Expand Up @@ -1191,11 +1197,16 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
this.cloudSyncedMessageTimestamps.add(msg.ts)
}
}
await this.providerRef.deref()?.postClineMessagesSnapshot(this.taskId, { bumpSeq: true })
}

private async updateClineMessage(message: ClineMessage) {
const provider = this.providerRef.deref()
await provider?.postMessageToWebview({ type: "messageUpdated", clineMessage: message })
if (message.partial === true) {
this.debouncedPostPartialMessageUpdate(message)
} else {
this.debouncedPostPartialMessageUpdate.cancel()
await this.providerRef.deref()?.postClineMessageUpdated(this.taskId, message)
}
this.emit(RooCodeEventName.Message, { action: "updated", message })
Comment thread
Gh0st352 marked this conversation as resolved.

// Check if we should sync to cloud and haven't already synced this message
Expand Down Expand Up @@ -1288,7 +1299,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {

let askTs: number

// Resolve auto-approval before adding the message so the state snapshot
// Resolve auto-approval before adding the message so the incremental append
// sent to the webview already carries isAnswered:true when the ask will
// be immediately resolved. This eliminates the race between the state
// update (which shows approval buttons) and the former separate
Expand Down Expand Up @@ -1322,10 +1333,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
lastMessage.partial = partial
lastMessage.progressStatus = progressStatus
lastMessage.isProtected = isProtected
// TODO: Be more efficient about saving and posting only new
// 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.
// Persist partial messages only when they become complete; the
// dedicated transport can still update one in-memory message at a time.
// 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
Expand Down Expand Up @@ -1578,6 +1587,9 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
if (lastFollowUpIndex !== -1) {
// Mark this follow-up as answered
this.clineMessages[lastFollowUpIndex].isAnswered = true
void this.updateClineMessage(this.clineMessages[lastFollowUpIndex]).catch((error) => {
console.error("[Task#handleWebviewAskResponse] follow-up delta failed:", error)
})
// Save the updated messages
this.saveClineMessages().catch((error) => {
console.error("Failed to save answered follow-up state:", error)
Expand Down Expand Up @@ -2053,7 +2065,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
// The todo list is already set in the constructor if initialTodos were provided
// No need to add any messages - the todoList property is already set

await this.providerRef.deref()?.postStateToWebviewWithoutTaskHistory()
await this.providerRef.deref()?.postClineMessagesSnapshot(this.taskId, { bumpSeq: true })

await this.say("text", task, images)

Expand Down Expand Up @@ -2204,7 +2216,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {

this.isInitialized = true

const { response, text, images } = await this.ask(askType) // Calls `postStateToWebview`.
const { response, text, images } = await this.ask(askType)

let responseText: string | undefined
let responseImages: string[] | undefined
Expand Down Expand Up @@ -2516,6 +2528,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {

public dispose(): void {
console.log(`[Task#dispose] disposing task ${this.taskId}.${this.instanceId}`)
this.debouncedPostPartialMessageUpdate.cancel()

// Stop the idle telemetry check and report any unflushed activity as a
// shutdown installment, so a task torn down mid-work (panel closed, task
Expand Down Expand Up @@ -2897,7 +2910,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
} satisfies ClineApiReqInfo)

await this.saveClineMessages()
await this.providerRef.deref()?.postStateToWebviewWithoutTaskHistory()
await this.updateClineMessage(this.clineMessages[lastApiReqIndex])

try {
let cacheWriteTokens = 0
Expand Down Expand Up @@ -2968,12 +2981,16 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
if (lastMessage && lastMessage.partial) {
// lastMessage.ts = Date.now() DO NOT update ts since it is used as a key for virtuoso list
lastMessage.partial = false
// instead of streaming partialMessage events, we do a save and post like normal to persist to disk
await this.updateClineMessage(lastMessage)
}

// Update `api_req_started` to have cancelled and cost, so that
// we can display the cost of the partial stream and the cancellation reason
updateApiReqMsg(cancelReason, streamingFailedMessage)
const apiRequestMessage = this.clineMessages[lastApiReqIndex]
if (apiRequestMessage) {
await this.updateClineMessage(apiRequestMessage)
}
await this.saveClineMessages()

// Signals to provider that it can retrieve the saved messages
Expand Down Expand Up @@ -3651,7 +3668,6 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
}

await this.saveClineMessages()
await this.providerRef.deref()?.postStateToWebviewWithoutTaskHistory()

// No legacy text-stream tool parser state to reset.

Expand Down
3 changes: 3 additions & 0 deletions src/core/task/__tests__/Task.persistence.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,9 @@ describe("Task persistence", () => {
mockProvider.postMessageToWebview = vi.fn().mockResolvedValue(undefined)
mockProvider.postStateToWebview = vi.fn().mockResolvedValue(undefined)
mockProvider.postStateToWebviewWithoutTaskHistory = vi.fn().mockResolvedValue(undefined)
mockProvider.postClineMessageAppended = vi.fn().mockResolvedValue(undefined)
mockProvider.postClineMessageUpdated = vi.fn().mockResolvedValue(undefined)
mockProvider.postClineMessagesSnapshot = vi.fn().mockResolvedValue(undefined)
mockProvider.updateTaskHistory = vi.fn().mockResolvedValue(undefined)
mockProvider.log = vi.fn()
})
Expand Down
Loading
Loading