-
Notifications
You must be signed in to change notification settings - Fork 191
chore: enforce no-floating-promises in core/task/ #253
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 3 commits
1691955
c5d8d4a
e57c772
8fa009c
6686bbf
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -356,6 +356,29 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| */ | ||
| assistantMessageSavedToHistory = false | ||
|
|
||
| /** | ||
| * Fire-and-forget wrapper around `presentAssistantMessage` that swallows the | ||
| * expected cancellation rejection (the presenter throws when `this.abort` is set) | ||
| * and logs any other failure. Keeping it non-blocking preserves the streaming | ||
| * presenter's self-locking semantics while preventing unhandled promise rejections | ||
| * from crashing the extension host. | ||
| */ | ||
| private presentAssistantMessageSafe(): void { | ||
| void presentAssistantMessage(this).catch((error) => { | ||
| // Discriminate on the error message rather than `this.abort` state, | ||
| // which can flip between the throw and the catch microtask running: | ||
| // a real failure followed by an abort flip would otherwise be | ||
| // silently swallowed, and a stale abort error logged as a failure. | ||
| // The abort throw site in presentAssistantMessage emits a message | ||
| // ending in "aborted" (matching the other abort-throw contracts in | ||
| // this file), so we suppress exactly that. | ||
| if (error instanceof Error && error.message.endsWith("aborted")) { | ||
| return | ||
| } | ||
| console.error(`[Task#presentAssistantMessage] task ${this.taskId}.${this.instanceId} failed:`, error) | ||
| }) | ||
| } | ||
|
|
||
| /** | ||
| * Push a tool_result block to userMessageContent, preventing duplicates. | ||
| * Duplicate tool_use_ids cause API errors. | ||
|
|
@@ -522,7 +545,15 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| this.messageQueueStateChangedHandler = () => { | ||
| this.emit(RooCodeEventName.TaskUserMessage, this.taskId) | ||
| this.emit(RooCodeEventName.QueuedMessagesUpdated, this.taskId, this.messageQueueService.messages) | ||
| this.providerRef.deref()?.postStateToWebviewWithoutTaskHistory() | ||
| void this.providerRef | ||
| .deref() | ||
| ?.postStateToWebviewWithoutTaskHistory() | ||
| .catch((error) => { | ||
| console.error( | ||
| "[Task#messageQueueStateChangedHandler] postStateToWebviewWithoutTaskHistory failed:", | ||
| error, | ||
| ) | ||
| }) | ||
| } | ||
|
|
||
| this.messageQueueService.on("stateChanged", this.messageQueueStateChangedHandler) | ||
|
|
@@ -567,9 +598,13 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| if (startTask) { | ||
| this._started = true | ||
| if (task || images) { | ||
| this.startTask(task, images) | ||
| void this.startTask(task, images).catch((error) => { | ||
| console.error("[Task#constructor] startTask failed:", error) | ||
| }) | ||
| } else if (historyItem) { | ||
| this.resumeTaskFromHistory() | ||
| void this.resumeTaskFromHistory().catch((error) => { | ||
| console.error("[Task#constructor] resumeTaskFromHistory failed:", error) | ||
| }) | ||
| } else { | ||
| throw new Error("Either historyItem or task/images must be provided") | ||
| } | ||
|
|
@@ -1162,7 +1197,13 @@ export class Task extends EventEmitter<TaskEvents> 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. | ||
| this.updateClineMessage(lastMessage) | ||
| // 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) | ||
| }) | ||
| // console.log("Task#ask: current ask promise was ignored (#1)") | ||
| throw new AskIgnoredError("updating existing partial") | ||
| } else { | ||
|
|
@@ -1203,7 +1244,11 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| lastMessage.isAnswered = true | ||
| } | ||
| await this.saveClineMessages() | ||
| this.updateClineMessage(lastMessage) | ||
| // Fire-and-forget: see updateClineMessage call above for the | ||
| // rationale on the .catch arm. | ||
| this.updateClineMessage(lastMessage).catch((error) => { | ||
| console.error("[Task#ask] updateClineMessage failed:", error) | ||
| }) | ||
| } else { | ||
| // This is a new and complete message, so add it like normal. | ||
| this.askResponse = undefined | ||
|
|
@@ -1274,7 +1319,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| if (message) { | ||
| this.interactiveAsk = message | ||
| this.emit(RooCodeEventName.TaskInteractive, this.taskId) | ||
| provider?.postMessageToWebview({ type: "interactionRequired" }) | ||
| /* v8 ignore next -- fire-and-forget webview notification inside setTimeout; rejection is benign */ | ||
| void provider?.postMessageToWebview({ type: "interactionRequired" }) | ||
| } | ||
| }, statusMutationTimeout), | ||
| ) | ||
|
|
@@ -1662,7 +1708,13 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| lastMessage.images = images | ||
| lastMessage.partial = partial | ||
| lastMessage.progressStatus = progressStatus | ||
| this.updateClineMessage(lastMessage) | ||
| // 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) | ||
| }) | ||
| } else { | ||
| // This is a new partial message, so add it with partial state. | ||
| const sayTs = Date.now() | ||
|
|
@@ -1701,7 +1753,11 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| await this.saveClineMessages() | ||
|
|
||
| // More performant than an entire `postStateToWebview`. | ||
| this.updateClineMessage(lastMessage) | ||
| // Fire-and-forget: see updateClineMessage call above for the | ||
| // rationale on the .catch arm. | ||
| this.updateClineMessage(lastMessage).catch((error) => { | ||
| console.error("[Task#say] updateClineMessage failed:", error) | ||
| }) | ||
| } else { | ||
| // This is a new and complete message, so add it like normal. | ||
| const sayTs = Date.now() | ||
|
|
@@ -1810,7 +1866,9 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| const { task, images } = this.metadata | ||
|
|
||
| if (task || images) { | ||
| this.startTask(task ?? undefined, images ?? undefined) | ||
| void this.startTask(task ?? undefined, images ?? undefined).catch((error) => { | ||
| console.error("[Task#start] startTask failed:", error) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -2351,7 +2409,11 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
|
|
||
| private async initiateTaskLoop(userContent: Anthropic.Messages.ContentBlockParam[]): Promise<void> { | ||
| // Kicks off the checkpoints initialization process in the background. | ||
| getCheckpointService(this) | ||
| // `getCheckpointService` wraps its full body in a try/catch and returns | ||
| // `undefined` on failure (see src/core/checkpoints/index.ts), so the | ||
| // returned promise cannot reject. `void` is sufficient — no `.catch` | ||
| // arm needed. | ||
| void getCheckpointService(this) | ||
|
|
||
| let nextUserContent = userContent | ||
| let includeFileDetails = true | ||
|
|
@@ -2688,9 +2750,13 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| if (signal.aborted) { | ||
| reject(new Error("Request cancelled by user")) | ||
| } else { | ||
| signal.addEventListener("abort", () => { | ||
| reject(new Error("Request cancelled by user")) | ||
| }, { once: true }) | ||
| signal.addEventListener( | ||
| "abort", | ||
| () => { | ||
| reject(new Error("Request cancelled by user")) | ||
| }, | ||
| { once: true }, | ||
| ) | ||
|
Comment on lines
+2757
to
+2763
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win Remove abort listeners when the raced chunk wins.
Suggested cleanup pattern- signal.addEventListener(
- "abort",
- () => {
- reject(new Error("Request cancelled by user"))
- },
- { once: true },
- )
+ const onAbort = () => reject(new Error("Request cancelled by user"))
+ signal.addEventListener("abort", onAbort, { once: true })Then wrap the Also applies to: 4288-4294 🤖 Prompt for AI Agents |
||
| } | ||
| }) | ||
| return await Promise.race([nextPromise, abortPromise]) | ||
|
|
@@ -2794,7 +2860,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| // Add to content and present | ||
| this.assistantMessageContent.push(partialToolUse) | ||
| this.userMessageContentReady = false | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| } else if (event.type === "tool_call_delta") { | ||
| // Process chunk using streaming JSON parser | ||
| const partialToolUse = NativeToolCallParser.processStreamingChunk( | ||
|
|
@@ -2813,7 +2880,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| this.assistantMessageContent[toolUseIndex] = partialToolUse | ||
|
|
||
| // Present updated tool use | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| } | ||
| } | ||
| } else if (event.type === "tool_call_end") { | ||
|
|
@@ -2839,7 +2907,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| this.userMessageContentReady = false | ||
|
|
||
| // Present the finalized tool call | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| } else if (toolUseIndex !== undefined) { | ||
| // finalizeStreamingToolCall returned null (malformed JSON or missing args) | ||
| // Mark the tool as non-partial so it's presented as complete, but execution | ||
|
|
@@ -2858,7 +2927,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| this.userMessageContentReady = false | ||
|
|
||
| // Present the tool call - validation will handle missing params | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -2891,7 +2961,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
|
|
||
| // Present the tool call to user - presentAssistantMessage will execute | ||
| // tools sequentially and accumulate all results in userMessageContent | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| break | ||
| } | ||
| case "text": { | ||
|
|
@@ -2910,7 +2981,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| }) | ||
| this.userMessageContentReady = false | ||
| } | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| break | ||
| } | ||
| } | ||
|
|
@@ -3237,7 +3309,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| this.userMessageContentReady = false | ||
|
|
||
| // Present the finalized tool call | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| } else if (toolUseIndex !== undefined) { | ||
| // finalizeStreamingToolCall returned null (malformed JSON or missing args) | ||
| // We still need to mark the tool as non-partial so it gets executed | ||
|
|
@@ -3256,7 +3329,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| this.userMessageContentReady = false | ||
|
|
||
| // Present the tool call - validation will handle missing params | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -3455,7 +3529,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| // If there is content to update then it will complete and | ||
| // update `this.userMessageContentReady` to true, which we | ||
| // `pWaitFor` before making the next request. | ||
| presentAssistantMessage(this) | ||
| /* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */ | ||
| this.presentAssistantMessageSafe() | ||
| } | ||
|
|
||
| if (hasTextContent || hasToolUses) { | ||
|
|
@@ -4191,10 +4266,14 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| const iterator = stream[Symbol.asyncIterator]() | ||
|
|
||
| // Set up abort handling - when the signal is aborted, clean up the controller reference | ||
| abortSignal.addEventListener("abort", () => { | ||
| console.log(`[Task#${this.taskId}.${this.instanceId}] AbortSignal triggered for current request`) | ||
| this.currentRequestAbortController = undefined | ||
| }, { once: true }) | ||
| abortSignal.addEventListener( | ||
| "abort", | ||
| () => { | ||
| console.log(`[Task#${this.taskId}.${this.instanceId}] AbortSignal triggered for current request`) | ||
| this.currentRequestAbortController = undefined | ||
| }, | ||
| { once: true }, | ||
| ) | ||
|
|
||
| try { | ||
| // Awaiting first chunk to see if it will throw an error. | ||
|
|
@@ -4206,9 +4285,13 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike { | |
| if (abortSignal.aborted) { | ||
| reject(new Error("Request cancelled by user")) | ||
| } else { | ||
| abortSignal.addEventListener("abort", () => { | ||
| reject(new Error("Request cancelled by user")) | ||
| }, { once: true }) | ||
| abortSignal.addEventListener( | ||
| "abort", | ||
| () => { | ||
| reject(new Error("Request cancelled by user")) | ||
| }, | ||
| { once: true }, | ||
| ) | ||
| } | ||
| }) | ||
|
|
||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.