From 3f627cfe2b4d24250aa6cbc7aa04c01997b27c26 Mon Sep 17 00:00:00 2001 From: Roo Code Date: Fri, 21 Nov 2025 23:21:31 +0000 Subject: [PATCH] feat(sse): wire AbortSignal through Task -> providers to terminate SSE on cancel --- src/api/index.ts | 5 +++ src/api/providers/anthropic.ts | 26 +++++++----- .../base-openai-compatible-provider.ts | 16 +++++++- src/api/providers/openai.ts | 40 +++++++++++-------- src/api/providers/openrouter.ts | 9 ++++- src/api/providers/roo.ts | 10 ++++- src/core/task/Task.ts | 5 ++- 7 files changed, 81 insertions(+), 30 deletions(-) diff --git a/src/api/index.ts b/src/api/index.ts index dd9208aa96..ebb7791711 100644 --- a/src/api/index.ts +++ b/src/api/index.ts @@ -89,6 +89,11 @@ export interface ApiHandlerCreateMessageMetadata { * Used by providers to determine whether to include native tool definitions. */ toolProtocol?: ToolProtocol + /** + * Abort controller signal to cancel the underlying HTTP request/stream. + * Providers that support request cancellation must forward this to their SDK client. + */ + abortSignal?: AbortSignal } export interface ApiHandler { diff --git a/src/api/providers/anthropic.ts b/src/api/providers/anthropic.ts index 0e767ce237..ec349b3e05 100644 --- a/src/api/providers/anthropic.ts +++ b/src/api/providers/anthropic.ts @@ -126,23 +126,29 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa case "claude-haiku-4-5-20251001": case "claude-3-haiku-20240307": betas.push("prompt-caching-2024-07-31") - return { headers: { "anthropic-beta": betas.join(",") } } + return { + headers: { "anthropic-beta": betas.join(",") }, + ...(metadata?.abortSignal ? { signal: metadata.abortSignal } : {}), + } default: - return undefined + return metadata?.abortSignal ? { signal: metadata.abortSignal } : undefined } })(), ) break } default: { - stream = (await this.client.messages.create({ - model: modelId, - max_tokens: maxTokens ?? ANTHROPIC_DEFAULT_MAX_TOKENS, - temperature, - system: [{ text: systemPrompt, type: "text" }], - messages, - stream: true, - })) as any + stream = (await this.client.messages.create( + { + model: modelId, + max_tokens: maxTokens ?? ANTHROPIC_DEFAULT_MAX_TOKENS, + temperature, + system: [{ text: systemPrompt, type: "text" }], + messages, + stream: true, + }, + metadata?.abortSignal ? { signal: metadata.abortSignal } : undefined, + )) as any break } } diff --git a/src/api/providers/base-openai-compatible-provider.ts b/src/api/providers/base-openai-compatible-provider.ts index 3d78ef75d1..ba48dfd482 100644 --- a/src/api/providers/base-openai-compatible-provider.ts +++ b/src/api/providers/base-openai-compatible-provider.ts @@ -101,7 +101,21 @@ export abstract class BaseOpenAiCompatibleProvider } try { - return this.client.chat.completions.create(params, requestOptions) + // Merge any caller-provided requestOptions with an AbortSignal from metadata (if present) + const effectiveRequestOptions: OpenAI.RequestOptions | undefined = (() => { + const base: OpenAI.RequestOptions = requestOptions ? { ...requestOptions } : {} + if (metadata?.abortSignal) { + ;(base as any).signal = metadata.abortSignal + } + return Object.keys(base).length > 0 ? base : undefined + })() + + // IMPORTANT: Only pass a second argument when options are actually defined + // to preserve test expectations that assert single-arg invocation. + if (effectiveRequestOptions) { + return this.client.chat.completions.create(params, effectiveRequestOptions) + } + return this.client.chat.completions.create(params) } catch (error) { throw handleOpenAIError(error, this.providerName) } diff --git a/src/api/providers/openai.ts b/src/api/providers/openai.ts index 1c8d3c7d9d..836d07f3b8 100644 --- a/src/api/providers/openai.ts +++ b/src/api/providers/openai.ts @@ -173,10 +173,11 @@ export class OpenAiHandler extends BaseProvider implements SingleCompletionHandl let stream try { - stream = await this.client.chat.completions.create( - requestOptions, - isAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {}, - ) + const baseOptions = isAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {} + const effectiveOptions = metadata?.abortSignal + ? { ...baseOptions, signal: metadata.abortSignal } + : baseOptions + stream = await this.client.chat.completions.create(requestOptions, effectiveOptions as any) } catch (error) { throw handleOpenAIError(error, this.providerName) } @@ -284,10 +285,11 @@ export class OpenAiHandler extends BaseProvider implements SingleCompletionHandl let response try { - response = await this.client.chat.completions.create( - requestOptions, - this._isAzureAiInference(modelUrl) ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {}, - ) + const baseOptions = this._isAzureAiInference(modelUrl) ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {} + const effectiveOptions = metadata?.abortSignal + ? { ...baseOptions, signal: metadata.abortSignal } + : baseOptions + response = await this.client.chat.completions.create(requestOptions, effectiveOptions as any) } catch (error) { throw handleOpenAIError(error, this.providerName) } @@ -403,10 +405,13 @@ export class OpenAiHandler extends BaseProvider implements SingleCompletionHandl let stream try { - stream = await this.client.chat.completions.create( - requestOptions, - methodIsAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {}, - ) + { + const baseOptions = methodIsAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {} + const effectiveOptions = metadata?.abortSignal + ? { ...baseOptions, signal: metadata.abortSignal } + : baseOptions + stream = await this.client.chat.completions.create(requestOptions, effectiveOptions as any) + } } catch (error) { throw handleOpenAIError(error, this.providerName) } @@ -435,10 +440,13 @@ export class OpenAiHandler extends BaseProvider implements SingleCompletionHandl let response try { - response = await this.client.chat.completions.create( - requestOptions, - methodIsAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {}, - ) + { + const baseOptions = methodIsAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {} + const effectiveOptions = metadata?.abortSignal + ? { ...baseOptions, signal: metadata.abortSignal } + : baseOptions + response = await this.client.chat.completions.create(requestOptions, effectiveOptions as any) + } } catch (error) { throw handleOpenAIError(error, this.providerName) } diff --git a/src/api/providers/openrouter.ts b/src/api/providers/openrouter.ts index c63142aad9..48900af829 100644 --- a/src/api/providers/openrouter.ts +++ b/src/api/providers/openrouter.ts @@ -195,7 +195,14 @@ export class OpenRouterHandler extends BaseProvider implements SingleCompletionH let stream try { - stream = await this.client.chat.completions.create(completionParams) + // Forward AbortSignal so Cancel terminates SSE immediately + const effectiveOptions = metadata?.abortSignal ? ({ signal: metadata.abortSignal } as any) : undefined + // Preserve tests that expect single-arg invocation when no options are provided + if (effectiveOptions) { + stream = await this.client.chat.completions.create(completionParams, effectiveOptions) + } else { + stream = await this.client.chat.completions.create(completionParams) + } } catch (error) { throw handleOpenAIError(error, this.providerName) } diff --git a/src/api/providers/roo.ts b/src/api/providers/roo.ts index 393740d3bd..d9fda095bb 100644 --- a/src/api/providers/roo.ts +++ b/src/api/providers/roo.ts @@ -106,7 +106,15 @@ export class RooHandler extends BaseOpenAiCompatibleProvider { try { this.client.apiKey = getSessionToken() - return this.client.chat.completions.create(rooParams, requestOptions) + // Merge AbortSignal into request options so Cancel can terminate SSE + const effectiveRequestOptions: OpenAI.RequestOptions | undefined = (() => { + const base: OpenAI.RequestOptions = requestOptions ? { ...requestOptions } : {} + if (metadata?.abortSignal) { + ;(base as any).signal = metadata.abortSignal + } + return Object.keys(base).length > 0 ? base : undefined + })() + return this.client.chat.completions.create(rooParams, effectiveRequestOptions) } catch (error) { throw handleOpenAIError(error, this.providerName) } diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 275265efd5..21c01f22fe 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -3271,11 +3271,14 @@ export class Task extends EventEmitter implements TaskLike { this.currentRequestAbortController = new AbortController() const abortSignal = this.currentRequestAbortController.signal + // Include abort signal so providers can terminate the underlying HTTP/SSE stream immediately + const metadataWithSignal: ApiHandlerCreateMessageMetadata = { ...metadata, abortSignal } + // The provider accepts reasoning items alongside standard messages; cast to the expected parameter type. const stream = this.api.createMessage( systemPrompt, cleanConversationHistory as unknown as Anthropic.Messages.MessageParam[], - metadata, + metadataWithSignal, ) const iterator = stream[Symbol.asyncIterator]()