From 3b899cecd21ef17505e9f57f4f1950c20f68f8ce Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 19 Aug 2026 22:43:06 +0800 Subject: [PATCH 01/27] feat(api): abort signal support for openai-native provider (completePrompt + createMessage) - completePrompt now uses a request-local signal merged from options.abortSignal and options.timeoutMs via mergeAbortSignalAndTimeout, no longer clobbering the streaming this.abortController; AbortError is rethrown as-is so callers can identify cancellations - createMessage paths (executeRequest and makeResponsesApiRequest fallback) bridge metadata.abortSignal into the internal controller using the Bedrock pattern (pre-aborted guard + { once: true } listener) Tests: abort signal passthrough, timeout abort, streaming-controller isolation, merged-signal abort, pre-aborted AbortError, fallback fetch pre-aborted/mid-request abort, non-Error rethrow, gpt-5.1 request-body coverage, response id/encrypted content accessors --- .../providers/__tests__/openai-native.spec.ts | 253 +++++++++++++++++- src/api/providers/openai-native.ts | 46 +++- 2 files changed, 293 insertions(+), 6 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 8c3398d443..15f1472b11 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -18,7 +18,11 @@ import { ApiProviderError, OpenAiServiceTier, SERVICE_TIER_KEY, serviceTiers } f import { OpenAiNativeHandler } from "../openai-native" import { ApiHandlerOptions } from "../../../shared/api" import { Package } from "../../../shared/package" -import { expectRequestObjectContaining, makeApiHandlerOptions } from "../../../test-utils/api" +import { + expectRequestObjectContaining, + makeApiHandlerOptions, + makeCreateMessageMetadata, +} from "../../../test-utils/api" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { deleteGlobalFetch } from "../../../test-utils/reset" @@ -375,6 +379,62 @@ describe("OpenAiNativeHandler", () => { } }).rejects.toThrow("OpenAI service error") }) + + it("should reject with AbortError when the external abortSignal is already aborted (fallback path)", async () => { + const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { + if (options?.signal?.aborted) { + const error = new Error("This operation was aborted") + error.name = "AbortError" + return Promise.reject(error) + } + return new Promise(() => {}) + }) + global.fetch = mockFetch as typeof fetch + + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + + const controller = new AbortController() + controller.abort() + + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(collectStream(stream)).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("should abort the fallback fetch when the external abortSignal is aborted mid-request", async () => { + const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { + return new Promise((_resolve, reject) => { + options?.signal?.addEventListener( + "abort", + () => { + const error = new Error("This operation was aborted") + error.name = "AbortError" + reject(error) + }, + { once: true }, + ) + }) + }) + global.fetch = mockFetch as typeof fetch + + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + + const controller = new AbortController() + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const collected = collectStream(stream) + setTimeout(() => controller.abort(), 10) + + await expect(collected).rejects.toMatchObject({ name: "AbortError" }) + }) }) describe("completePrompt", () => { @@ -483,6 +543,197 @@ describe("OpenAiNativeHandler", () => { expect(result).toBe("") }) + it("should pass the external abort signal through to the SDK request", async () => { + mockResponsesCreate.mockResolvedValue({ + output: [ + { + type: "message", + content: [{ type: "output_text", text: "response" }], + }, + ], + }) + + const controller = new AbortController() + await handler.completePrompt("Test prompt", { abortSignal: controller.signal }) + + // Without a timeout the merged signal is the external signal itself + expect(mockResponsesCreate.mock.calls[0][1].signal).toBe(controller.signal) + }) + + it("should work without options (backward compatible)", async () => { + mockResponsesCreate.mockResolvedValue({ + output: [ + { + type: "message", + content: [{ type: "output_text", text: "response" }], + }, + ], + }) + + const result = await handler.completePrompt("Test prompt") + + expect(result).toBe("response") + expect(mockResponsesCreate.mock.calls[0][1].signal).toBeInstanceOf(AbortSignal) + }) + + it("completePrompt should abort its request signal when timeoutMs is reached", async () => { + // Node's AbortSignal.timeout() uses internal timers that vi.useFakeTimers() does not + // intercept, so this relies on a short real timeout instead of fake timers. + let requestSignal: AbortSignal | undefined + mockResponsesCreate.mockImplementationOnce(async (_body: unknown, options: { signal?: AbortSignal }) => { + requestSignal = options.signal + // Stay pending until the merged timeout signal aborts the request + await new Promise((resolve) => { + options.signal?.addEventListener("abort", () => resolve(), { once: true }) + }) + return { + output: [ + { + type: "message", + content: [{ type: "output_text", text: "response" }], + }, + ], + } + }) + + const result = await handler.completePrompt("Test prompt", { timeoutMs: 50 }) + + expect(result).toBe("response") + expect(requestSignal).toBeInstanceOf(AbortSignal) + expect(requestSignal?.aborted).toBe(true) + }) + + it("completePrompt should not replace an active streaming abort controller", async () => { + const activeStreamingController = new AbortController() + handler["abortController"] = activeStreamingController + mockResponsesCreate.mockResolvedValue({ + output: [ + { + type: "message", + content: [{ type: "output_text", text: "response" }], + }, + ], + }) + + await handler.completePrompt("Test prompt") + + expect(handler["abortController"]).toBe(activeStreamingController) + }) + + it("completePrompt should merge the external signal and timeoutMs together", async () => { + const controller = new AbortController() + mockResponsesCreate.mockResolvedValue({ + output: [ + { + type: "message", + content: [{ type: "output_text", text: "response" }], + }, + ], + }) + + await handler.completePrompt("Test prompt", { abortSignal: controller.signal, timeoutMs: 10000 }) + + const mergedSignal = mockResponsesCreate.mock.calls[0][1].signal as AbortSignal + expect(mergedSignal).toBeInstanceOf(AbortSignal) + + // Aborting the external signal must abort the merged signal synchronously + controller.abort() + expect(mergedSignal.aborted).toBe(true) + }) + + it("completePrompt should reject with AbortError when the abortSignal is already aborted", async () => { + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + const error = new Error("This operation was aborted") + error.name = "AbortError" + return Promise.reject(error) + } + return Promise.resolve({ + output: [ + { + type: "message", + content: [{ type: "output_text", text: "response" }], + }, + ], + }) + }) + + const controller = new AbortController() + controller.abort() + + await expect( + handler.completePrompt("Test prompt", { abortSignal: controller.signal }), + ).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("completePrompt should rethrow non-Error failures after telemetry", async () => { + mockResponsesCreate.mockRejectedValue("string failure") + + await expect(handler.completePrompt("Test prompt")).rejects.toBe("string failure") + expect(mockCaptureException).toHaveBeenCalledWith( + expect.objectContaining({ + message: "string failure", + provider: "OpenAI Native", + modelId: "gpt-4.1", + operation: "completePrompt", + }), + ) + }) + + it("completePrompt should return direct response text fallback", async () => { + mockResponsesCreate.mockResolvedValue({ text: "fallback response" }) + + const result = await handler.completePrompt("Test prompt") + + expect(result).toBe("fallback response") + }) + + it("completePrompt should include supported service tier, reasoning, verbosity, and prompt cache retention", async () => { + const configuredHandler = new OpenAiNativeHandler({ + ...mockOptions, + apiModelId: "gpt-5.1", + openAiNativeServiceTier: "flex", + enableResponsesReasoningSummary: true, + }) + mockResponsesCreate.mockResolvedValue({ + output: [ + { + type: "message", + content: [{ type: "output_text", text: "response" }], + }, + ], + }) + + await configuredHandler.completePrompt("Test prompt") + + const requestBody = mockResponsesCreate.mock.calls[0][0] + expect(requestBody.service_tier).toBe("flex") + expect(requestBody.include).toEqual(["reasoning.encrypted_content"]) + expect(requestBody.reasoning).toEqual({ effort: "medium", summary: "auto" }) + expect(requestBody.text).toEqual({ verbosity: "medium" }) + expect(requestBody.prompt_cache_retention).toBe("24h") + }) + + it("should expose response id and encrypted reasoning content", () => { + handler["lastResponseId"] = "resp_123" + handler["lastResponseOutput"] = [ + { type: "message" }, + { type: "reasoning", encrypted_content: "encrypted", id: "reasoning_1" }, + ] + + expect(handler.getResponseId()).toBe("resp_123") + expect(handler.getEncryptedContent()).toEqual({ encrypted_content: "encrypted", id: "reasoning_1" }) + }) + + it("should return undefined when encrypted reasoning content is absent", () => { + expect(handler.getEncryptedContent()).toBeUndefined() + + handler["lastResponseOutput"] = [{ type: "reasoning" }] + + expect(handler.getEncryptedContent()).toBeUndefined() + }) }) describe("getModel", () => { diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index a919e2c82b..9cd74fcc05 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -32,6 +32,7 @@ import { NOT_PROVIDED } from "./constants" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { isMcpTool } from "../../utils/mcp-name" import { sanitizeOpenAiCallId } from "../../utils/tool-id" +import { mergeAbortSignalAndTimeout } from "./utils/abort-signal" export type OpenAiNativeModel = ReturnType @@ -413,6 +414,18 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Create AbortController for cancellation this.abortController = new AbortController() + // Bridge external abort signal to our internal controller using the Bedrock pattern: + // - pre-aborted guard: check if already aborted before adding listener + // - { once: true }: remove listener after first abort to avoid leaks + const externalAbortSignal = metadata?.abortSignal + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + this.abortController.abort() + } else { + externalAbortSignal.addEventListener("abort", () => this.abortController?.abort(), { once: true }) + } + } + // Build per-request headers using taskId when available, falling back to sessionId const taskId = metadata?.taskId const userAgent = `zoo-code/${Package.version} (${os.platform()} ${os.release()}; ${os.arch()}) node/${process.version.slice(1)}` @@ -563,6 +576,18 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Create AbortController for cancellation this.abortController = new AbortController() + // Bridge external abort signal to our internal controller using the Bedrock pattern: + // - pre-aborted guard: check if already aborted before adding listener + // - { once: true }: remove listener after first abort to avoid leaks + const externalAbortSignal = metadata?.abortSignal + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + this.abortController.abort() + } else { + externalAbortSignal.addEventListener("abort", () => this.abortController?.abort(), { once: true }) + } + } + // Build per-request headers using taskId when available, falling back to sessionId const taskId = metadata?.taskId const userAgent = `zoo-code/${Package.version} (${os.platform()} ${os.release()}; ${os.arch()}) node/${process.version.slice(1)}` @@ -644,6 +669,11 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Handle streaming response yield* this.handleStreamResponse(response.body, model) } catch (error) { + // Re-throw abort errors as-is so callers can identify cancellations + if (error instanceof Error && error.name === "AbortError") { + throw error + } + const model = this.getModel() const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "createMessage") @@ -1489,9 +1519,12 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio return this.lastResponseId } async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { - try { - this.abortController = new AbortController() + // Request-local abort signal: merges the external abort signal with an optional + // timeout without touching this.abortController (owned by streaming requests). + const requestSignal = + mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) ?? new AbortController().signal + try { const model = this.getModel() const { verbosity } = model @@ -1550,7 +1583,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Make the non-streaming request const response = await (this.client as any).responses.create(requestBody, { - signal: this.abortController.signal, + signal: requestSignal, }) // Extract text from the response @@ -1573,6 +1606,11 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio return "" } catch (error) { + // Re-throw abort errors as-is so callers can identify cancellations + if (error instanceof Error && error.name === "AbortError") { + throw error + } + const errorModel = this.getModel() const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, errorModel.id, "completePrompt") @@ -1582,8 +1620,6 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio throw new Error(`OpenAI Native completion error: ${error.message}`) } throw error - } finally { - this.abortController = undefined } } } From d53b955698600b23c68b3e064645a75b3ced59ac Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 19 Aug 2026 22:48:06 +0800 Subject: [PATCH 02/27] feat(api): abort signal support for openai-compatible provider (completePrompt + createMessage) - completePrompt merges options.abortSignal and options.timeoutMs via mergeAbortSignalAndTimeout and forwards the merged signal to the AI SDK generateText abortSignal option - createMessage forwards metadata.abortSignal to streamText so in-flight streams are aborted on task cancellation Tests: new openai-compatible.spec.ts covering completePrompt signal/timeout passthrough, timeoutMs <= 0 disabled, pre-aborted AbortError, error propagation, and createMessage abortSignal bridging (pass-through, absent metadata, pre-aborted, mid-request abort) --- .../__tests__/openai-compatible.spec.ts | 228 ++++++++++++++++++ src/api/providers/openai-compatible.ts | 18 +- 2 files changed, 244 insertions(+), 2 deletions(-) create mode 100644 src/api/providers/__tests__/openai-compatible.spec.ts diff --git a/src/api/providers/__tests__/openai-compatible.spec.ts b/src/api/providers/__tests__/openai-compatible.spec.ts new file mode 100644 index 0000000000..38f13935f4 --- /dev/null +++ b/src/api/providers/__tests__/openai-compatible.spec.ts @@ -0,0 +1,228 @@ +// npx vitest run api/providers/__tests__/openai-compatible.spec.ts + +import { OpenAICompatibleHandler } from "../openai-compatible" +import { makeApiHandlerOptions } from "../../../test-utils/api" +import { collectStream } from "../../../test-utils/stream" + +const mockGenerateText = vitest.fn() +const mockStreamText = vitest.fn() + +// The factory must not touch the mock bindings at factory-execution time (vi.mock is +// hoisted above the consts), so forward lazily through wrapper functions. +vitest.mock("ai", () => ({ + generateText: (...args: unknown[]) => mockGenerateText(...(args as [])), + streamText: (...args: unknown[]) => mockStreamText(...(args as [])), +})) + +// Concrete test implementation of the abstract OpenAI-compatible base class +class TestOpenAICompatibleHandler extends OpenAICompatibleHandler { + constructor(apiKey: string) { + super(makeApiHandlerOptions({ apiModelId: "test-model" }), { + providerName: "TestProvider", + baseURL: "https://test.example.com/v1", + apiKey, + modelId: "test-model", + modelInfo: { + maxTokens: 4096, + contextWindow: 128000, + supportsImages: false, + supportsPromptCache: false, + inputPrice: 0.5, + outputPrice: 1.5, + }, + }) + } + + override getModel() { + return { id: "test-model", info: this.config.modelInfo } + } +} + +function makeEmptyStreamResult() { + return { + fullStream: { + [Symbol.asyncIterator]: async function* () { + // Emit no parts + yield* [] + }, + }, + usage: Promise.resolve(undefined), + } +} + +describe("OpenAICompatibleHandler", () => { + let handler: TestOpenAICompatibleHandler + + beforeEach(() => { + vi.clearAllMocks() + handler = new TestOpenAICompatibleHandler("test-api-key") + }) + + describe("completePrompt", () => { + it("should return message content from successful response", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + const result = await handler.completePrompt("test prompt") + + expect(result).toBe("response") + expect(mockGenerateText).toHaveBeenCalledTimes(1) + expect(mockGenerateText.mock.calls[0][0].prompt).toBe("test prompt") + }) + + it("should pass abortSignal through to generateText", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + const controller = new AbortController() + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + + expect(mockGenerateText.mock.calls[0][0].abortSignal).toBe(controller.signal) + }) + + it("should pass timeoutMs through to generateText as a timeout abort signal", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + await handler.completePrompt("test prompt", { timeoutMs: 5000 }) + + const { abortSignal } = mockGenerateText.mock.calls[0][0] + expect(abortSignal).toBeInstanceOf(AbortSignal) + expect(abortSignal.aborted).toBe(false) + }) + + it("should merge signal and timeout when both are provided", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + const controller = new AbortController() + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 10000 }) + + const mergedSignal = mockGenerateText.mock.calls[0][0].abortSignal as AbortSignal + expect(mergedSignal).toBeInstanceOf(AbortSignal) + + // Aborting the external signal must abort the merged signal synchronously + controller.abort() + expect(mergedSignal.aborted).toBe(true) + }) + + it("should work without options (backward compatible)", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + const result = await handler.completePrompt("test prompt") + + expect(result).toBe("response") + expect(mockGenerateText.mock.calls[0][0].abortSignal).toBeUndefined() + }) + + it("should treat timeoutMs <= 0 as disabled", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + expect(mockGenerateText.mock.calls[0][0].abortSignal).toBeUndefined() + + await handler.completePrompt("test prompt", { timeoutMs: -1 }) + expect(mockGenerateText.mock.calls[1][0].abortSignal).toBeUndefined() + }) + + it("should reject with AbortError when abortSignal is already aborted before request", async () => { + mockGenerateText.mockImplementation((options: { abortSignal?: AbortSignal }) => { + if (options.abortSignal?.aborted) { + const error = new Error("This operation was aborted") + error.name = "AbortError" + return Promise.reject(error) + } + return Promise.resolve({ text: "response" }) + }) + + const controller = new AbortController() + controller.abort() + + await expect( + handler.completePrompt("test prompt", { abortSignal: controller.signal }), + ).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("should throw handled error when API call fails", async () => { + mockGenerateText.mockRejectedValue(new Error("Network error")) + + await expect(handler.completePrompt("test prompt")).rejects.toThrow("Network error") + }) + }) + + describe("createMessage", () => { + it("should pass the external abortSignal to streamText", async () => { + mockStreamText.mockReturnValue(makeEmptyStreamResult()) + + const controller = new AbortController() + const stream = handler.createMessage("You are helpful.", [], { + taskId: "test", + abortSignal: controller.signal, + }) + await collectStream(stream) + + expect(mockStreamText).toHaveBeenCalledTimes(1) + expect(mockStreamText.mock.calls[0][0].abortSignal).toBe(controller.signal) + }) + + it("should not set an abortSignal when metadata has none", async () => { + mockStreamText.mockReturnValue(makeEmptyStreamResult()) + + const stream = handler.createMessage("You are helpful.", [], { taskId: "test" }) + await collectStream(stream) + + expect(mockStreamText.mock.calls[0][0].abortSignal).toBeUndefined() + }) + + it("should reject with AbortError when the external abortSignal is pre-aborted", async () => { + mockStreamText.mockImplementation((options: { abortSignal?: AbortSignal }) => ({ + fullStream: { + [Symbol.asyncIterator]: async function* () { + if (options.abortSignal?.aborted) { + const error = new Error("This operation was aborted") + error.name = "AbortError" + throw error + } + yield* [] + }, + }, + usage: Promise.resolve(undefined), + })) + + const controller = new AbortController() + controller.abort() + + const stream = handler.createMessage("You are helpful.", [], { + taskId: "test", + abortSignal: controller.signal, + }) + await expect(collectStream(stream)).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("should abort the stream when the external abortSignal is aborted mid-request", async () => { + mockStreamText.mockImplementation((options: { abortSignal?: AbortSignal }) => ({ + fullStream: { + [Symbol.asyncIterator]: async function* () { + // Emulate a slow model response that ends when the request is aborted + await new Promise((resolve) => { + options.abortSignal?.addEventListener("abort", () => resolve(), { once: true }) + }) + yield* [] + const error = new Error("This operation was aborted") + error.name = "AbortError" + throw error + }, + }, + usage: Promise.resolve(undefined), + })) + + const controller = new AbortController() + const stream = handler.createMessage("You are helpful.", [], { + taskId: "test", + abortSignal: controller.signal, + }) + const collected = collectStream(stream) + setTimeout(() => controller.abort(), 10) + + await expect(collected).rejects.toMatchObject({ name: "AbortError" }) + }) + }) +}) diff --git a/src/api/providers/openai-compatible.ts b/src/api/providers/openai-compatible.ts index 308652804a..21776ddd0c 100644 --- a/src/api/providers/openai-compatible.ts +++ b/src/api/providers/openai-compatible.ts @@ -18,6 +18,7 @@ import { ApiStream, ApiStreamUsageChunk } from "../transform/stream" import { DEFAULT_HEADERS } from "./constants" import { BaseProvider } from "./base-provider" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" +import { mergeAbortSignalAndTimeout } from "./utils/abort-signal" /** * Configuration options for creating an OpenAI-compatible provider. @@ -176,6 +177,12 @@ export abstract class OpenAICompatibleHandler extends BaseProvider implements Si toolChoice: this.mapToolChoice(metadata?.tool_choice), } + // Forward the external abort signal (e.g. task cancellation) so the in-flight + // stream is aborted when the request is cancelled + if (metadata?.abortSignal) { + requestOptions.abortSignal = metadata.abortSignal + } + // Use streamText for streaming responses const result = streamText(requestOptions) @@ -200,12 +207,19 @@ export abstract class OpenAICompatibleHandler extends BaseProvider implements Si async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { const languageModel = this.getLanguageModel() - const { text } = await generateText({ + const generateOptions: Parameters[0] & { abortSignal?: AbortSignal } = { model: languageModel, prompt, maxOutputTokens: this.getMaxOutputTokens(), temperature: this.config.temperature ?? 0, - }) + } + + const mergedAbortSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) + if (mergedAbortSignal) { + generateOptions.abortSignal = mergedAbortSignal + } + + const { text } = await generateText(generateOptions) return text } From 2296b74a74ecc35cdd10993d7ba9cfe12de2accc Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 02:43:01 +0800 Subject: [PATCH 03/27] fix(api): address CodeRabbit review on openai-native and openai-compatible abort handling - openai-native.ts: bridge external abort signal to a request-local controller in executeRequest and makeResponsesApiRequest; detach the { once: true } listener in finally so a late abort from an earlier request cannot cancel a later request's controller (listener closures no longer read the mutable this.abortController field) - openai-native.spec.ts: regression test - first stream completes normally, second stream runs with a different external signal, aborting the FIRST signal must not cancel the second stream - openai-compatible.spec.ts: timeout tests now assert the generated signal actually fires on its own ~50ms timeout (a never-expiring signal can no longer pass), the merged-signal timeout component fires independently of the caller signal, and caller signals pass through by identity when timeoutMs is 0 or negative --- .../__tests__/openai-compatible.spec.ts | 61 +++++++++++++- .../providers/__tests__/openai-native.spec.ts | 82 +++++++++++++++++++ src/api/providers/openai-native.ts | 66 +++++++++++---- 3 files changed, 188 insertions(+), 21 deletions(-) diff --git a/src/api/providers/__tests__/openai-compatible.spec.ts b/src/api/providers/__tests__/openai-compatible.spec.ts index 38f13935f4..2dc46a750a 100644 --- a/src/api/providers/__tests__/openai-compatible.spec.ts +++ b/src/api/providers/__tests__/openai-compatible.spec.ts @@ -78,30 +78,66 @@ describe("OpenAICompatibleHandler", () => { expect(mockGenerateText.mock.calls[0][0].abortSignal).toBe(controller.signal) }) - it("should pass timeoutMs through to generateText as a timeout abort signal", async () => { + it("should pass timeoutMs through to generateText as a working timeout abort signal", async () => { mockGenerateText.mockResolvedValue({ text: "response" }) - await handler.completePrompt("test prompt", { timeoutMs: 5000 }) + await handler.completePrompt("test prompt", { timeoutMs: 50 }) const { abortSignal } = mockGenerateText.mock.calls[0][0] expect(abortSignal).toBeInstanceOf(AbortSignal) expect(abortSignal.aborted).toBe(false) + + // A never-expiring signal (or a pre-aborted one) would fail this check: + // the signal must fire on its own ~50ms timeout without any external abort. + const fired = await Promise.race([ + new Promise((resolve) => { + abortSignal.addEventListener("abort", () => resolve(true), { once: true }) + }), + new Promise((resolve) => { + setTimeout(() => resolve(false), 1000) + }), + ]) + expect(fired).toBe(true) + expect(abortSignal.aborted).toBe(true) }) it("should merge signal and timeout when both are provided", async () => { mockGenerateText.mockResolvedValue({ text: "response" }) const controller = new AbortController() - await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 10000 }) + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 50 }) const mergedSignal = mockGenerateText.mock.calls[0][0].abortSignal as AbortSignal expect(mergedSignal).toBeInstanceOf(AbortSignal) + expect(mergedSignal.aborted).toBe(false) // Aborting the external signal must abort the merged signal synchronously controller.abort() expect(mergedSignal.aborted).toBe(true) }) + it("should let the timeout component of a merged signal fire without the caller signal", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + await handler.completePrompt("test prompt", { abortSignal: new AbortController().signal, timeoutMs: 50 }) + + const mergedSignal = mockGenerateText.mock.calls[0][0].abortSignal as AbortSignal + expect(mergedSignal).toBeInstanceOf(AbortSignal) + expect(mergedSignal.aborted).toBe(false) + + // With the caller signal left untouched, only the timeout component can fire + const fired = await Promise.race([ + new Promise((resolve) => { + mergedSignal.addEventListener("abort", () => resolve(true), { once: true }) + }), + new Promise((resolve) => { + setTimeout(() => resolve(false), 1000) + }), + ]) + expect(fired).toBe(true) + expect(mergedSignal.aborted).toBe(true) + }) + it("should work without options (backward compatible)", async () => { mockGenerateText.mockResolvedValue({ text: "response" }) @@ -121,6 +157,25 @@ describe("OpenAICompatibleHandler", () => { expect(mockGenerateText.mock.calls[1][0].abortSignal).toBeUndefined() }) + it("should pass the caller signal unchanged when timeoutMs is 0", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + const controller = new AbortController() + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 0 }) + + // A disabled timeout must not drop the caller's cancellation signal + expect(mockGenerateText.mock.calls[0][0].abortSignal).toBe(controller.signal) + }) + + it("should pass the caller signal unchanged when timeoutMs is negative", async () => { + mockGenerateText.mockResolvedValue({ text: "response" }) + + const controller = new AbortController() + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: -1 }) + + expect(mockGenerateText.mock.calls[0][0].abortSignal).toBe(controller.signal) + }) + it("should reject with AbortError when abortSignal is already aborted before request", async () => { mockGenerateText.mockImplementation((options: { abortSignal?: AbortSignal }) => { if (options.abortSignal?.aborted) { diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 15f1472b11..a03604304b 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -435,6 +435,88 @@ describe("OpenAiNativeHandler", () => { await expect(collected).rejects.toMatchObject({ name: "AbortError" }) }) + + it("should not let a late abort from an earlier request cancel a later request", async () => { + // Regression: the external-signal bridge must detach on request completion. + // With a lingering listener (or one reading the mutable this.abortController + // field), aborting the FIRST request's signal after completion would cancel + // the SECOND request's controller. + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + + type OpenStream = { + controller?: ReadableStreamDefaultController + fetchSignal: AbortSignal + } + const openStreams: OpenStream[] = [] + const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { + const entry: OpenStream = { fetchSignal: options?.signal as AbortSignal } + const body = new ReadableStream({ + start: (controller) => { + entry.controller = controller + }, + }) + openStreams.push(entry) + return Promise.resolve({ + ok: true, + body, + }) + }) + global.fetch = mockFetch as typeof fetch + + const requireController = (index: number): ReadableStreamDefaultController => { + const entry = openStreams[index] + if (!entry?.controller) { + throw new Error("expected fallback fetch to have started") + } + return entry.controller + } + + const firstController = new AbortController() + const secondController = new AbortController() + + // First request: completes normally. + const firstStream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: firstController.signal }), + ) + const firstCollected = collectStream(firstStream) + await new Promise((resolve) => setTimeout(resolve, 10)) + requireController(0).enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"one"}\n\n'), + ) + requireController(0).enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + requireController(0).close() + + const firstChunks = await firstCollected + expect(firstChunks.some((chunk) => chunk.type === "text" && chunk.text === "one")).toBe(true) + + // Second request with a different external signal, left in-flight. + const secondStream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: secondController.signal }), + ) + const secondCollected = collectStream(secondStream) + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(openStreams).toHaveLength(2) + + // Aborting the FIRST request's signal must not leak into the second request. + firstController.abort() + + // The second request's internal fetch signal must remain active... + expect(openStreams[1].fetchSignal.aborted).toBe(false) + + // ...and the second stream must still complete normally. + requireController(1).enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"two"}\n\n'), + ) + requireController(1).enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + requireController(1).close() + + const secondChunks = await secondCollected + expect(secondChunks.some((chunk) => chunk.type === "text" && chunk.text === "two")).toBe(true) + }) }) describe("completePrompt", () => { diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index 9cd74fcc05..7f0f409161 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -411,18 +411,25 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio systemPrompt?: string, messages?: Anthropic.Messages.MessageParam[], ): ApiStream { - // Create AbortController for cancellation - this.abortController = new AbortController() - - // Bridge external abort signal to our internal controller using the Bedrock pattern: + // Create a request-local AbortController for cancellation. It is exposed via + // this.abortController so the stop-button path can observe it, but all bridging + // below captures the local reference so a late abort from an earlier request can + // never reach a later request's controller. + const requestController = new AbortController() + this.abortController = requestController + + // Bridge external abort signal to the request controller using the Bedrock pattern: // - pre-aborted guard: check if already aborted before adding listener - // - { once: true }: remove listener after first abort to avoid leaks + // - { once: true }: removes the listener after the first abort to avoid leaks + // The listener is removed in the finally block when the request completes. + let abortListener: (() => void) | undefined const externalAbortSignal = metadata?.abortSignal if (externalAbortSignal) { if (externalAbortSignal.aborted) { - this.abortController.abort() + requestController.abort() } else { - externalAbortSignal.addEventListener("abort", () => this.abortController?.abort(), { once: true }) + abortListener = () => requestController.abort() + externalAbortSignal.addEventListener("abort", abortListener, { once: true }) } } @@ -438,7 +445,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio try { // Use the official SDK with per-request headers const stream = (await (this.client as any).responses.create(requestBody, { - signal: this.abortController.signal, + signal: requestController.signal, headers: requestHeaders, })) as AsyncIterable @@ -450,7 +457,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio for await (const event of stream) { // Check if request was aborted - if (this.abortController.signal.aborted) { + if (requestController.signal.aborted) { break } @@ -462,7 +469,16 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // For errors, fallback to manual SSE via fetch yield* this.makeResponsesApiRequest(requestBody, model, metadata, systemPrompt, messages) } finally { - this.abortController = undefined + // Detach the bridging listener so a late abort from this request cannot + // cancel a later request's controller. + if (abortListener) { + externalAbortSignal?.removeEventListener("abort", abortListener) + } + // Only clear the field if this request still owns it (the fallback path may + // have installed its own controller, which it clears itself). + if (this.abortController === requestController) { + this.abortController = undefined + } } } @@ -573,18 +589,25 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio const baseUrl = this.options.openAiNativeBaseUrl || "https://api.openai.com" const url = `${baseUrl}/v1/responses` - // Create AbortController for cancellation - this.abortController = new AbortController() + // Create a request-local AbortController for cancellation. It is exposed via + // this.abortController so the stop-button path can observe it, but the bridging + // listener captures the local reference so a late abort from an earlier request + // can never reach a later request's controller. + const requestController = new AbortController() + this.abortController = requestController - // Bridge external abort signal to our internal controller using the Bedrock pattern: + // Bridge external abort signal to the request controller using the Bedrock pattern: // - pre-aborted guard: check if already aborted before adding listener - // - { once: true }: remove listener after first abort to avoid leaks + // - { once: true }: removes the listener after the first abort to avoid leaks + // The listener is removed in the finally block when the request completes. + let abortListener: (() => void) | undefined const externalAbortSignal = metadata?.abortSignal if (externalAbortSignal) { if (externalAbortSignal.aborted) { - this.abortController.abort() + requestController.abort() } else { - externalAbortSignal.addEventListener("abort", () => this.abortController?.abort(), { once: true }) + abortListener = () => requestController.abort() + externalAbortSignal.addEventListener("abort", abortListener, { once: true }) } } @@ -603,7 +626,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio "User-Agent": userAgent, }, body: JSON.stringify(requestBody), - signal: this.abortController.signal, + signal: requestController.signal, }) if (!response.ok) { @@ -690,7 +713,14 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Handle non-Error objects throw new Error(`Unexpected error connecting to Responses API`) } finally { - this.abortController = undefined + // Detach the bridging listener so a late abort from this request cannot + // cancel a later request's controller. + if (abortListener) { + externalAbortSignal?.removeEventListener("abort", abortListener) + } + if (this.abortController === requestController) { + this.abortController = undefined + } } } From 176a0fa05422b6c7e39e86c6fd7dc5fe9862d8b3 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 04:28:42 +0800 Subject: [PATCH 04/27] test(api): clear leftover guard timers in openai-compatible abort tests --- .../__tests__/openai-compatible.spec.ts | 26 ++++++++++++++----- 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/src/api/providers/__tests__/openai-compatible.spec.ts b/src/api/providers/__tests__/openai-compatible.spec.ts index 2dc46a750a..aa3289a1a8 100644 --- a/src/api/providers/__tests__/openai-compatible.spec.ts +++ b/src/api/providers/__tests__/openai-compatible.spec.ts @@ -89,16 +89,23 @@ describe("OpenAICompatibleHandler", () => { // A never-expiring signal (or a pre-aborted one) would fail this check: // the signal must fire on its own ~50ms timeout without any external abort. + let guardTimer: ReturnType | undefined const fired = await Promise.race([ new Promise((resolve) => { abortSignal.addEventListener("abort", () => resolve(true), { once: true }) }), new Promise((resolve) => { - setTimeout(() => resolve(false), 1000) + guardTimer = setTimeout(() => resolve(false), 1000) }), ]) - expect(fired).toBe(true) - expect(abortSignal.aborted).toBe(true) + try { + expect(fired).toBe(true) + expect(abortSignal.aborted).toBe(true) + } finally { + // Clear the guard timer so a winning abort event does not leave an + // active timer behind that delays worker teardown. + clearTimeout(guardTimer) + } }) it("should merge signal and timeout when both are provided", async () => { @@ -126,16 +133,23 @@ describe("OpenAICompatibleHandler", () => { expect(mergedSignal.aborted).toBe(false) // With the caller signal left untouched, only the timeout component can fire + let guardTimer: ReturnType | undefined const fired = await Promise.race([ new Promise((resolve) => { mergedSignal.addEventListener("abort", () => resolve(true), { once: true }) }), new Promise((resolve) => { - setTimeout(() => resolve(false), 1000) + guardTimer = setTimeout(() => resolve(false), 1000) }), ]) - expect(fired).toBe(true) - expect(mergedSignal.aborted).toBe(true) + try { + expect(fired).toBe(true) + expect(mergedSignal.aborted).toBe(true) + } finally { + // Clear the guard timer so a winning abort event does not leave an + // active timer behind that delays worker teardown. + clearTimeout(guardTimer) + } }) it("should work without options (backward compatible)", async () => { From cb153ec293ac10d8ccbc29262e9828dbbf1f3151 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 17:55:15 +0800 Subject: [PATCH 05/27] refactor(api): adopt RequestConfigBuilder in feat/abort-r1-openai-native-compat abort wiring --- .../__tests__/request-config-builder.spec.ts | 31 +++++++++++++++++++ .../config-builder/request-config-builder.ts | 15 +++++++++ src/api/providers/openai-compatible.ts | 7 +++-- src/api/providers/openai-native.ts | 5 +-- 4 files changed, 54 insertions(+), 4 deletions(-) diff --git a/src/api/providers/__tests__/request-config-builder.spec.ts b/src/api/providers/__tests__/request-config-builder.spec.ts index 977b09df6b..80203890b2 100644 --- a/src/api/providers/__tests__/request-config-builder.spec.ts +++ b/src/api/providers/__tests__/request-config-builder.spec.ts @@ -505,4 +505,35 @@ describe("RequestConfigBuilder", () => { expect(config.maxTokens).toBe(2000) }) }) + + describe("static merge helpers (canonical abort-signal entry points)", () => { + it("returns undefined from mergeAbortSignalAndTimeout when no external signal and no valid timeout", () => { + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(undefined, undefined)).toBeUndefined() + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(undefined, 0)).toBeUndefined() + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(undefined, -5)).toBeUndefined() + }) + + it("returns the external signal directly when no timeout is merged", () => { + const controller = new AbortController() + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(controller.signal, undefined)).toBe( + controller.signal, + ) + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(controller.signal, 0)).toBe(controller.signal) + }) + + it("returns the primary signal directly from mergeAbortSignals when there is no secondary", () => { + const controller = new AbortController() + expect(RequestConfigBuilder.mergeAbortSignals(controller.signal)).toBe(controller.signal) + expect(RequestConfigBuilder.mergeAbortSignals(controller.signal, undefined)).toBe(controller.signal) + }) + + it("delegates to AbortSignal.any when two distinct signals are merged", () => { + const a = new AbortController() + const b = new AbortController() + const merged = RequestConfigBuilder.mergeAbortSignals(a.signal, b.signal) + expect(merged.aborted).toBe(false) + b.abort() + expect(merged.aborted).toBe(true) + }) + }) }) diff --git a/src/api/providers/config-builder/request-config-builder.ts b/src/api/providers/config-builder/request-config-builder.ts index 2201d735bc..a3c1ba7ebb 100644 --- a/src/api/providers/config-builder/request-config-builder.ts +++ b/src/api/providers/config-builder/request-config-builder.ts @@ -163,4 +163,19 @@ export class RequestConfigBuilder @@ -1552,7 +1552,8 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Request-local abort signal: merges the external abort signal with an optional // timeout without touching this.abortController (owned by streaming requests). const requestSignal = - mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) ?? new AbortController().signal + RequestConfigBuilder.mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) ?? + new AbortController().signal try { const model = this.getModel() From 470273876fa457ffb629404c11cda121f1ac0e00 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 18:27:48 +0800 Subject: [PATCH 06/27] test(api): cover primary-signal abort in builder merge regression tests --- .../providers/__tests__/request-config-builder.spec.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/api/providers/__tests__/request-config-builder.spec.ts b/src/api/providers/__tests__/request-config-builder.spec.ts index 80203890b2..f9557a123f 100644 --- a/src/api/providers/__tests__/request-config-builder.spec.ts +++ b/src/api/providers/__tests__/request-config-builder.spec.ts @@ -535,5 +535,14 @@ describe("RequestConfigBuilder", () => { b.abort() expect(merged.aborted).toBe(true) }) + + it("aborts the merged signal when the primary signal aborts", () => { + const a = new AbortController() + const b = new AbortController() + const merged = RequestConfigBuilder.mergeAbortSignals(a.signal, b.signal) + expect(merged.aborted).toBe(false) + a.abort() + expect(merged.aborted).toBe(true) + }) }) }) From b16d5ad0db2c86132312f4cb406c5c669ad83b7f Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 12:24:58 +0800 Subject: [PATCH 07/27] test(api): kill abort-signal mutation survivors in openai-native/openai-compatible Add focused abort-signal bridging tests for the bedrock-pattern bridge in OpenAiNativeHandler (executeRequest + makeResponsesApiRequest): once-only listener wiring and detachment, pre-aborted guard, mid-flight abort propagation, mid-stream break, late-abort listener detachment, and request-local controller ownership across concurrent SDK/fallback requests. Also assert the abortSignal property is left absent (not just undefined) in OpenAICompatibleHandler when no signal is supplied. Kills 26 of the 28 mutation-diff survivors on the PR diff; the two remaining OptionalChaining mutants (openai-native.ts L475/L719, externalAbortSignal?.removeEventListener) are unreachable variants: abortListener is only assigned when the signal is present, so optional chaining is behaviorally identical on every reachable path. --- .../__tests__/openai-compatible.spec.ts | 18 + .../providers/__tests__/openai-native.spec.ts | 517 ++++++++++++++++++ 2 files changed, 535 insertions(+) diff --git a/src/api/providers/__tests__/openai-compatible.spec.ts b/src/api/providers/__tests__/openai-compatible.spec.ts index aa3289a1a8..ca0e2c7e56 100644 --- a/src/api/providers/__tests__/openai-compatible.spec.ts +++ b/src/api/providers/__tests__/openai-compatible.spec.ts @@ -159,6 +159,9 @@ describe("OpenAICompatibleHandler", () => { expect(result).toBe("response") expect(mockGenerateText.mock.calls[0][0].abortSignal).toBeUndefined() + // The property must be absent entirely: an unconditional assignment would + // leave it present with an undefined value, which `toBeUndefined` cannot see. + expect("abortSignal" in mockGenerateText.mock.calls[0][0]).toBe(false) }) it("should treat timeoutMs <= 0 as disabled", async () => { @@ -239,6 +242,21 @@ describe("OpenAICompatibleHandler", () => { await collectStream(stream) expect(mockStreamText.mock.calls[0][0].abortSignal).toBeUndefined() + // The property must be absent entirely (an unconditional assignment would + // leave it present with an undefined value). + expect("abortSignal" in mockStreamText.mock.calls[0][0]).toBe(false) + }) + + it("should complete without metadata and leave the abortSignal property unset", async () => { + mockStreamText.mockReturnValue(makeEmptyStreamResult()) + + const stream = handler.createMessage("You are helpful.", []) + await collectStream(stream) + + expect(mockStreamText).toHaveBeenCalledTimes(1) + // createMessage without metadata must not throw and must leave the + // property absent: metadata?.abortSignal is undefined when metadata is absent. + expect("abortSignal" in mockStreamText.mock.calls[0][0]).toBe(false) }) it("should reject with AbortError when the external abortSignal is pre-aborted", async () => { diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index a03604304b..b891881e93 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -16,6 +16,7 @@ import OpenAI from "openai" import { ApiProviderError, OpenAiServiceTier, SERVICE_TIER_KEY, serviceTiers } from "@roo-code/types" import { OpenAiNativeHandler } from "../openai-native" +import type { ApiStreamChunk, ApiStreamTextChunk } from "../../../api/transform/stream" import { ApiHandlerOptions } from "../../../shared/api" import { Package } from "../../../shared/package" import { @@ -517,6 +518,522 @@ describe("OpenAiNativeHandler", () => { const secondChunks = await secondCollected expect(secondChunks.some((chunk) => chunk.type === "text" && chunk.text === "two")).toBe(true) }) + + describe("abort-signal bridging", () => { + // The bedrock-pattern bridge in executeRequest and makeResponsesApiRequest forwards + // metadata?.abortSignal onto a request-local AbortController. These tests make every + // branch observable: a resolving SDK mock exercises the executeRequest bridge + // directly, a rejecting one exercises the fetch fallback bridge, and the + // request-local signal handed to the SDK/fetch is captured for assertions. + + function makeAbortError(): Error { + const error = new Error("This operation was aborted") + error.name = "AbortError" + return error + } + + const tick = (): Promise => new Promise((resolve) => setTimeout(resolve, 0)) + + function untilSignalAborted(signal: AbortSignal, timeoutMs = 200): Promise { + return new Promise((resolve) => { + if (signal.aborted) { + resolve() + return + } + const timer = setTimeout(() => resolve(), timeoutMs) + signal.addEventListener( + "abort", + () => { + clearTimeout(timer) + resolve() + }, + { once: true }, + ) + }) + } + + function textChunks(chunks: ApiStreamChunk[]): ApiStreamTextChunk[] { + return chunks.filter((chunk): chunk is ApiStreamTextChunk => chunk.type === "text") + } + + function makeOpenStreamFetchMock() { + type OpenStream = { + controller?: ReadableStreamDefaultController + fetchSignal: AbortSignal + } + const openStreams: OpenStream[] = [] + const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { + const entry: OpenStream = { fetchSignal: options?.signal as AbortSignal } + const body = new ReadableStream({ + start: (controller) => { + entry.controller = controller + }, + }) + openStreams.push(entry) + return Promise.resolve({ + ok: true, + body, + }) + }) + const requireController = (index: number): ReadableStreamDefaultController => { + const entry = openStreams[index] + if (!entry?.controller) { + throw new Error("expected fallback fetch to have started") + } + return entry.controller + } + return { openStreams, mockFetch, requireController } + } + + it("should register a once-only abort listener on the external signal and detach it when the SDK request completes", async () => { + const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, "addEventListener") + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + let sdkSignal: AbortSignal | undefined + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + sdkSignal = options?.signal + return Promise.resolve( + asyncStreamFrom([ + { type: "response.output_text.delta", delta: "one" }, + { type: "response.output_text.delta", delta: " two" }, + ]), + ) + }) + + try { + const chunks = await collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + + expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["one", " two"]) + expect(sdkSignal?.aborted).toBe(false) + // The bridge must listen for the "abort" event with { once: true } ... + expect(addSpy).toHaveBeenCalledTimes(1) + expect(addSpy).toHaveBeenCalledWith("abort", expect.any(Function), { once: true }) + // ... and detach that exact listener when the request completes. + const registeredListener = addSpy.mock.calls.find(([type]) => type === "abort")?.[1] + expect(registeredListener).toBeDefined() + expect(removeSpy).toHaveBeenCalledTimes(1) + expect(removeSpy).toHaveBeenCalledWith("abort", registeredListener) + // The request-local controller is cleared once the request is done. + expect(handler["abortController"]).toBeUndefined() + } finally { + addSpy.mockRestore() + removeSpy.mockRestore() + } + }) + + it("should abort the SDK request immediately when the external signal is already aborted", async () => { + const controller = new AbortController() + controller.abort() + const addSpy = vi.spyOn(controller.signal, "addEventListener") + let sdkSignal: AbortSignal | undefined + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + sdkSignal = options?.signal + // A real SDK rejects immediately when its request signal is pre-aborted. + if (options?.signal?.aborted) { + return Promise.reject(makeAbortError()) + } + return Promise.resolve(asyncStreamFrom([{ type: "response.output_text.delta", delta: "one" }])) + }) + const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { + if (options?.signal?.aborted) { + return Promise.reject(makeAbortError()) + } + return new Promise(() => {}) + }) + global.fetch = mockFetch as typeof fetch + + try { + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + await expect(collectStream(stream)).rejects.toMatchObject({ name: "AbortError" }) + + // The bridge must have pre-aborted the request-local controller ... + expect(sdkSignal?.aborted).toBe(true) + // ... instead of registering a listener on the already-aborted signal. + expect(addSpy).not.toHaveBeenCalled() + } finally { + addSpy.mockRestore() + } + }) + + it("should not pre-abort the SDK request for a pending external signal and should abort it mid-flight", async () => { + const controller = new AbortController() + let sdkSignal: AbortSignal | undefined + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + sdkSignal = options?.signal + const signal = options?.signal + return Promise.resolve( + (async function* () { + yield { type: "response.output_text.delta", delta: "one" } + if (signal) { + await untilSignalAborted(signal, 200) + } + })(), + ) + }) + + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + const collected = collectStream(stream) + await tick() + + expect(sdkSignal).toBeDefined() + // A pending external signal must not abort the request up front. + expect(sdkSignal?.aborted).toBe(false) + + controller.abort() + await tick() + // ... but it must abort the request as soon as it fires. + expect(sdkSignal?.aborted).toBe(true) + + const chunks = await collected + expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["one"]) + }) + + it("should stop consuming the SDK stream once the external signal aborts the request", async () => { + const controller = new AbortController() + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + const signal = options?.signal + return Promise.resolve( + (async function* () { + yield { type: "response.output_text.delta", delta: "first" } + if (signal) { + await untilSignalAborted(signal, 200) + } + yield { type: "response.output_text.delta", delta: "second" } + })(), + ) + }) + + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + const collected = collectStream(stream) + await tick() + + controller.abort() + + const chunks = await collected + expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["first"]) + }) + + it("should detach the external abort listener on completion so a late abort cannot abort the request signal", async () => { + const controller = new AbortController() + let openGate: (() => void) | undefined + const gate = new Promise((resolve) => { + openGate = resolve + }) + let sdkSignal: AbortSignal | undefined + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + sdkSignal = options?.signal + return Promise.resolve( + (async function* () { + yield { type: "response.output_text.delta", delta: "one" } + await gate + })(), + ) + }) + + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + const collected = collectStream(stream) + await tick() + + // Let the request complete normally, then abort the external signal late. + if (!openGate) { + throw new Error("expected the stream gate to be ready") + } + openGate() + const chunks = await collected + expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["one"]) + + controller.abort() + await tick() + + // The bridging listener must have been detached: the late abort must + // not reach the already-completed request's controller. + expect(sdkSignal?.aborted).toBe(false) + expect(handler["abortController"]).toBeUndefined() + }) + + it("should not call removeEventListener on the external signal when no listener was registered", async () => { + const controller = new AbortController() + controller.abort() + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + return Promise.reject(makeAbortError()) + } + return Promise.resolve(asyncStreamFrom([{ type: "response.output_text.delta", delta: "one" }])) + }) + const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { + if (options?.signal?.aborted) { + return Promise.reject(makeAbortError()) + } + return new Promise(() => {}) + }) + global.fetch = mockFetch as typeof fetch + + try { + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + await expect(collectStream(stream)).rejects.toMatchObject({ name: "AbortError" }) + + // A pre-aborted signal registers no listener, so nothing may be removed. + expect(removeSpy).not.toHaveBeenCalled() + } finally { + removeSpy.mockRestore() + } + }) + + it("should preserve a later fallback request's controller when an earlier SDK request completes", async () => { + // Request A: SDK path, in flight. Request B: SDK fails, so its fallback + // fetch installs the handler's controller. When A completes, its finally + // must not clear the controller owned by B's fallback. + let aGateOpen: (() => void) | undefined + const aGate = new Promise((resolve) => { + aGateOpen = resolve + }) + let aSdkSignal: AbortSignal | undefined + let sdkCalls = 0 + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + sdkCalls += 1 + if (sdkCalls === 1) { + aSdkSignal = options?.signal + return Promise.resolve( + (async function* () { + yield { type: "response.output_text.delta", delta: "a-one" } + await aGate + })(), + ) + } + return Promise.reject(new Error("SDK not available")) + }) + const { openStreams, mockFetch, requireController } = makeOpenStreamFetchMock() + global.fetch = mockFetch as typeof fetch + + const streamA = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: new AbortController().signal }), + ) + const collectedA = collectStream(streamA) + await tick() + expect(aSdkSignal?.aborted).toBe(false) + + const streamB = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: new AbortController().signal }), + ) + const collectedB = collectStream(streamB) + await tick() + expect(openStreams).toHaveLength(1) + + // Complete A while B's fallback owns the handler's controller. + if (!aGateOpen) { + throw new Error("expected the stream gate to be ready") + } + aGateOpen() + const chunksA = await collectedA + expect(textChunks(chunksA).map((chunk) => chunk.text)).toEqual(["a-one"]) + expect(handler["abortController"]?.signal).toBe(openStreams[0].fetchSignal) + + // Let B finish; its finally chain clears the controller. + requireController(0).enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"b-one"}\n\n'), + ) + requireController(0).enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + requireController(0).close() + const chunksB = await collectedB + expect(textChunks(chunksB).map((chunk) => chunk.text)).toEqual(["b-one"]) + expect(handler["abortController"]).toBeUndefined() + }) + + it("should not clear a later SDK request's controller when a fallback request completes", async () => { + // Mirror of the previous test: request B (fallback) starts first and + // request A (SDK) takes over the handler's controller. When B's fallback + // completes, its finally must not clear A's controller. + let aGateOpen: (() => void) | undefined + const aGate = new Promise((resolve) => { + aGateOpen = resolve + }) + let aSdkSignal: AbortSignal | undefined + let sdkCalls = 0 + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + sdkCalls += 1 + if (sdkCalls === 1) { + return Promise.reject(new Error("SDK not available")) + } + aSdkSignal = options?.signal + return Promise.resolve( + (async function* () { + yield { type: "response.output_text.delta", delta: "a-one" } + await aGate + })(), + ) + }) + const { openStreams, mockFetch, requireController } = makeOpenStreamFetchMock() + global.fetch = mockFetch as typeof fetch + + const streamB = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: new AbortController().signal }), + ) + const collectedB = collectStream(streamB) + await tick() + expect(openStreams).toHaveLength(1) + + const streamA = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: new AbortController().signal }), + ) + const collectedA = collectStream(streamA) + await tick() + expect(aSdkSignal?.aborted).toBe(false) + + // Let B's fallback complete while A owns the handler's controller. + requireController(0).enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"b-one"}\n\n'), + ) + requireController(0).enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + requireController(0).close() + const chunksB = await collectedB + expect(textChunks(chunksB).map((chunk) => chunk.text)).toEqual(["b-one"]) + expect(handler["abortController"]?.signal).toBe(aSdkSignal) + + // Let A finish; its finally clears the controller. + if (!aGateOpen) { + throw new Error("expected the stream gate to be ready") + } + aGateOpen() + const chunksA = await collectedA + expect(textChunks(chunksA).map((chunk) => chunk.text)).toEqual(["a-one"]) + expect(handler["abortController"]).toBeUndefined() + }) + + it("should clear the handler's abortController after a fallback request completes", async () => { + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + body: new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"one"}\n\n'), + ) + controller.enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + controller.close() + }, + }), + }) + global.fetch = mockFetch as typeof fetch + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + + const chunks = await collectStream(handler.createMessage(systemPrompt, messages)) + + expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["one"]) + // The fallback installs its own controller and must clear it when done. + expect(handler["abortController"]).toBeUndefined() + }) + + it("should register a once-only abort listener in the fallback path and detach it on completion", async () => { + const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, "addEventListener") + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + body: new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"one"}\n\n'), + ) + controller.enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + controller.close() + }, + }), + }) + global.fetch = mockFetch as typeof fetch + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + + try { + const chunks = await collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + + expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["one"]) + // Both bridges (SDK path and fallback) listen for "abort" with + // { once: true }, and both detach their own listener on completion. + expect(addSpy).toHaveBeenCalledTimes(2) + expect(removeSpy).toHaveBeenCalledTimes(2) + for (const call of addSpy.mock.calls) { + expect(call[0]).toBe("abort") + expect(call[2]).toEqual({ once: true }) + } + for (const call of removeSpy.mock.calls) { + expect(call[0]).toBe("abort") + } + expect(handler["abortController"]).toBeUndefined() + } finally { + addSpy.mockRestore() + removeSpy.mockRestore() + } + }) + + it("should detach the fallback's external abort listener on completion so a late abort cannot abort the fetch signal", async () => { + const { openStreams, mockFetch, requireController } = makeOpenStreamFetchMock() + global.fetch = mockFetch as typeof fetch + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + + const controller = new AbortController() + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + const collected = collectStream(stream) + await tick() + expect(openStreams).toHaveLength(1) + + requireController(0).enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"one"}\n\n'), + ) + requireController(0).enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + requireController(0).close() + + const chunks = await collected + expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["one"]) + + // A late abort must not reach this request's own fetch signal. + controller.abort() + await tick() + expect(openStreams[0].fetchSignal.aborted).toBe(false) + }) + }) }) describe("completePrompt", () => { From 118b921c7ad4819a2c262b8b4761c95f256e97ed Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 12:42:05 +0800 Subject: [PATCH 08/27] test(api): exclude proven-equivalent openai-native OptionalChaining mutants from the gate --- src/api/providers/openai-native.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index 62e148b351..8ef16858b8 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -472,6 +472,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Detach the bridging listener so a late abort from this request cannot // cancel a later request's controller. if (abortListener) { + // Stryker disable next-line OptionalChaining: abortListener is only assigned inside the if (externalAbortSignal) block, so whenever the finally guard is true the signal is non-nullish; dropping the optional chain is behaviorally identical on every reachable path (covered by the abort-signal bridging specs) externalAbortSignal?.removeEventListener("abort", abortListener) } // Only clear the field if this request still owns it (the fallback path may @@ -716,6 +717,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Detach the bridging listener so a late abort from this request cannot // cancel a later request's controller. if (abortListener) { + // Stryker disable next-line OptionalChaining: abortListener is only assigned inside the if (externalAbortSignal) block, so whenever the finally guard is true the signal is non-nullish; dropping the optional chain is behaviorally identical on every reachable path (covered by the abort-signal bridging specs) externalAbortSignal?.removeEventListener("abort", abortListener) } if (this.abortController === requestController) { From ba72b77c10d855a9f96dad59aacc7463299bc3d4 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 13:16:23 +0800 Subject: [PATCH 09/27] fix(api): address reviewer abort-signal findings in openai-native and openai-compatible - openai-native: handleStreamResponse now checks the request-local controller before wrapping stream errors, so a user stop surfaces the contract AbortError ("The OpenAI Native request was aborted") exactly once instead of a wrapped error plus a double captureException. Regression tests: "should surface the contract AbortError once when the fallback stream read rejects on external abort" and "should not convert a non-abort stream error into an AbortError". - openai-native: extract the duplicated external-abort bridge from executeRequest and makeResponsesApiRequest into attachExternalAbort; the finally blocks call the returned cleanup (detach?.()), which removes the two Stryker OptionalChaining directives added in 118b921c7. All 11 abort-signal bridging specs pass unchanged. - request-config-builder: delete the unused mergeAbortSignals static and its three spec cases (zero production callers); mergeAbortSignalAndTimeout stays, with production callers in openai-native.ts and #1290's openai-codex.ts. - openai-native spec: replace the trivial "completePrompt should not replace an active streaming abort controller" assertion with the observable "should not let an earlier request's external abort affect a later request on the same handler" test. - openai-compatible spec: the pre-aborted completePrompt test now rejects with a real DOMException AbortError and asserts the exact message, since openai-compatible.ts passes SDK errors through without normalization. --- .../__tests__/openai-compatible.spec.ts | 12 +- .../providers/__tests__/openai-native.spec.ts | 159 ++++++++++++++++-- .../__tests__/request-config-builder.spec.ts | 24 --- .../config-builder/request-config-builder.ts | 14 +- src/api/providers/openai-native.ts | 94 ++++++----- 5 files changed, 211 insertions(+), 92 deletions(-) diff --git a/src/api/providers/__tests__/openai-compatible.spec.ts b/src/api/providers/__tests__/openai-compatible.spec.ts index ca0e2c7e56..0ecf28758f 100644 --- a/src/api/providers/__tests__/openai-compatible.spec.ts +++ b/src/api/providers/__tests__/openai-compatible.spec.ts @@ -193,12 +193,15 @@ describe("OpenAICompatibleHandler", () => { expect(mockGenerateText.mock.calls[0][0].abortSignal).toBe(controller.signal) }) - it("should reject with AbortError when abortSignal is already aborted before request", async () => { + it("should reject with the real DOMException AbortError when the signal is pre-aborted", async () => { + // Emulate the real AI SDK: a pre-aborted signal makes the request reject + // with the fetch stack's DOMException abort error rather than a fabricated + // Error, so this exercises the provider's pass-through of a real SDK + // abort. openai-compatible.ts has no normalization layer, so the + // DOMException must surface unchanged (name and message). mockGenerateText.mockImplementation((options: { abortSignal?: AbortSignal }) => { if (options.abortSignal?.aborted) { - const error = new Error("This operation was aborted") - error.name = "AbortError" - return Promise.reject(error) + return Promise.reject(new DOMException("The operation was aborted.", "AbortError")) } return Promise.resolve({ text: "response" }) }) @@ -210,6 +213,7 @@ describe("OpenAICompatibleHandler", () => { handler.completePrompt("test prompt", { abortSignal: controller.signal }), ).rejects.toMatchObject({ name: "AbortError", + message: "The operation was aborted.", }) }) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index b891881e93..dfe9b8668b 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -1033,6 +1033,148 @@ describe("OpenAiNativeHandler", () => { await tick() expect(openStreams[0].fetchSignal.aborted).toBe(false) }) + + it("should surface the contract AbortError once when the fallback stream read rejects on external abort", async () => { + // Force the fallback path and emulate undici: the body's reader.read() + // stays pending and rejects with a DOMException AbortError when the + // request signal aborts. Before the fix, handleStreamResponse wrapped + // that error in a plain Error (defeating the caller's AbortError guard) + // and the request was captured as an exception twice. + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + let fetchSignal: AbortSignal | undefined + const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { + const signal = options?.signal + if (!signal) { + return Promise.reject(new Error("expected the fallback fetch to carry a request signal")) + } + fetchSignal = signal + const body = new ReadableStream({ + pull: () => { + return new Promise((_resolve, reject) => { + if (signal.aborted) { + reject(new DOMException("This operation was aborted", "AbortError")) + return + } + signal.addEventListener( + "abort", + () => reject(new DOMException("This operation was aborted", "AbortError")), + { once: true }, + ) + }) + }, + }) + return Promise.resolve({ ok: true, body }) + }) + global.fetch = mockFetch as typeof fetch + + const controller = new AbortController() + const collected = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + await tick() + expect(fetchSignal).toBeDefined() + + // A user stop mid-stream must surface exactly one error, contract-named. + controller.abort() + await expect(collected).rejects.toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) + // The provider must not report a user-triggered stop as an exception. + expect(mockCaptureException).not.toHaveBeenCalled() + }) + + it("should not convert a non-abort stream error into an AbortError", async () => { + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + const mockFetch = vitest.fn().mockImplementation(() => { + const body = new ReadableStream({ + pull: () => Promise.reject(new Error("socket hang up")), + }) + return Promise.resolve({ ok: true, body }) + }) + global.fetch = mockFetch as typeof fetch + + await expect(collectStream(handler.createMessage(systemPrompt, messages))).rejects.toThrow( + "Error processing response stream: socket hang up", + ) + }) + + it("should not let an earlier request's external abort affect a later request on the same handler", async () => { + // Request 1 streams under external signal A while request 2 starts under + // external signal B. Aborting A while both are in flight must abort only + // request 1's request-local controller; request 2 completes normally. + const firstExternal = new AbortController() + const secondExternal = new AbortController() + let firstGateOpen: (() => void) | undefined + const firstGate = new Promise((resolve) => { + firstGateOpen = resolve + }) + let firstSdkSignal: AbortSignal | undefined + let secondSdkSignal: AbortSignal | undefined + let sdkCalls = 0 + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + sdkCalls += 1 + if (sdkCalls === 1) { + firstSdkSignal = options?.signal + return Promise.resolve( + (async function* () { + yield { type: "response.output_text.delta", delta: "one" } + await firstGate + })(), + ) + } + secondSdkSignal = options?.signal + return Promise.resolve( + asyncStreamFrom([ + { type: "response.output_text.delta", delta: "two-a" }, + { type: "response.output_text.delta", delta: " two-b" }, + ]), + ) + }) + + const collected1 = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: firstExternal.signal }), + ), + ) + await tick() + expect(firstSdkSignal).toBeDefined() + + const collected2 = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: secondExternal.signal }), + ), + ) + await tick() + expect(secondSdkSignal).toBeDefined() + + // Abort the first request's external signal while the second is in flight. + firstExternal.abort() + await tick() + expect(firstSdkSignal?.aborted).toBe(true) + expect(secondSdkSignal?.aborted).toBe(false) + + // The second request completes normally with its own content. + const chunks2 = await collected2 + expect(textChunks(chunks2).map((chunk) => chunk.text)).toEqual(["two-a", " two-b"]) + expect(secondExternal.signal.aborted).toBe(false) + + // Let the first request wind down; its stream simply ends. + if (!firstGateOpen) { + throw new Error("expected the first stream gate to be ready") + } + firstGateOpen() + const chunks1 = await collected1 + expect(textChunks(chunks1).map((chunk) => chunk.text)).toEqual(["one"]) + }) }) }) @@ -1202,23 +1344,6 @@ describe("OpenAiNativeHandler", () => { expect(requestSignal?.aborted).toBe(true) }) - it("completePrompt should not replace an active streaming abort controller", async () => { - const activeStreamingController = new AbortController() - handler["abortController"] = activeStreamingController - mockResponsesCreate.mockResolvedValue({ - output: [ - { - type: "message", - content: [{ type: "output_text", text: "response" }], - }, - ], - }) - - await handler.completePrompt("Test prompt") - - expect(handler["abortController"]).toBe(activeStreamingController) - }) - it("completePrompt should merge the external signal and timeoutMs together", async () => { const controller = new AbortController() mockResponsesCreate.mockResolvedValue({ diff --git a/src/api/providers/__tests__/request-config-builder.spec.ts b/src/api/providers/__tests__/request-config-builder.spec.ts index f9557a123f..4ec2d3abc4 100644 --- a/src/api/providers/__tests__/request-config-builder.spec.ts +++ b/src/api/providers/__tests__/request-config-builder.spec.ts @@ -520,29 +520,5 @@ describe("RequestConfigBuilder", () => { ) expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(controller.signal, 0)).toBe(controller.signal) }) - - it("returns the primary signal directly from mergeAbortSignals when there is no secondary", () => { - const controller = new AbortController() - expect(RequestConfigBuilder.mergeAbortSignals(controller.signal)).toBe(controller.signal) - expect(RequestConfigBuilder.mergeAbortSignals(controller.signal, undefined)).toBe(controller.signal) - }) - - it("delegates to AbortSignal.any when two distinct signals are merged", () => { - const a = new AbortController() - const b = new AbortController() - const merged = RequestConfigBuilder.mergeAbortSignals(a.signal, b.signal) - expect(merged.aborted).toBe(false) - b.abort() - expect(merged.aborted).toBe(true) - }) - - it("aborts the merged signal when the primary signal aborts", () => { - const a = new AbortController() - const b = new AbortController() - const merged = RequestConfigBuilder.mergeAbortSignals(a.signal, b.signal) - expect(merged.aborted).toBe(false) - a.abort() - expect(merged.aborted).toBe(true) - }) }) }) diff --git a/src/api/providers/config-builder/request-config-builder.ts b/src/api/providers/config-builder/request-config-builder.ts index a3c1ba7ebb..38a6c36b08 100644 --- a/src/api/providers/config-builder/request-config-builder.ts +++ b/src/api/providers/config-builder/request-config-builder.ts @@ -165,17 +165,13 @@ export class RequestConfigBuilder void) | undefined { + if (!externalAbortSignal) { + return undefined + } + if (externalAbortSignal.aborted) { + requestController.abort() + return undefined + } + const abortListener = () => requestController.abort() + externalAbortSignal.addEventListener("abort", abortListener, { once: true }) + return () => { + externalAbortSignal.removeEventListener("abort", abortListener) + } + } + private async *executeRequest( requestBody: any, model: OpenAiNativeModel, @@ -418,20 +444,10 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio const requestController = new AbortController() this.abortController = requestController - // Bridge external abort signal to the request controller using the Bedrock pattern: - // - pre-aborted guard: check if already aborted before adding listener - // - { once: true }: removes the listener after the first abort to avoid leaks - // The listener is removed in the finally block when the request completes. - let abortListener: (() => void) | undefined - const externalAbortSignal = metadata?.abortSignal - if (externalAbortSignal) { - if (externalAbortSignal.aborted) { - requestController.abort() - } else { - abortListener = () => requestController.abort() - externalAbortSignal.addEventListener("abort", abortListener, { once: true }) - } - } + // Bridge the external abort signal onto the request controller; the returned + // cleanup detaches the listener in the finally block so a late abort from this + // request cannot cancel a later request's controller. + const detach = this.attachExternalAbort(metadata?.abortSignal, requestController) // Build per-request headers using taskId when available, falling back to sessionId const taskId = metadata?.taskId @@ -471,10 +487,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } finally { // Detach the bridging listener so a late abort from this request cannot // cancel a later request's controller. - if (abortListener) { - // Stryker disable next-line OptionalChaining: abortListener is only assigned inside the if (externalAbortSignal) block, so whenever the finally guard is true the signal is non-nullish; dropping the optional chain is behaviorally identical on every reachable path (covered by the abort-signal bridging specs) - externalAbortSignal?.removeEventListener("abort", abortListener) - } + detach?.() // Only clear the field if this request still owns it (the fallback path may // have installed its own controller, which it clears itself). if (this.abortController === requestController) { @@ -597,20 +610,10 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio const requestController = new AbortController() this.abortController = requestController - // Bridge external abort signal to the request controller using the Bedrock pattern: - // - pre-aborted guard: check if already aborted before adding listener - // - { once: true }: removes the listener after the first abort to avoid leaks - // The listener is removed in the finally block when the request completes. - let abortListener: (() => void) | undefined - const externalAbortSignal = metadata?.abortSignal - if (externalAbortSignal) { - if (externalAbortSignal.aborted) { - requestController.abort() - } else { - abortListener = () => requestController.abort() - externalAbortSignal.addEventListener("abort", abortListener, { once: true }) - } - } + // Bridge the external abort signal onto the request controller; the returned + // cleanup detaches the listener in the finally block so a late abort from this + // request cannot cancel a later request's controller. + const detach = this.attachExternalAbort(metadata?.abortSignal, requestController) // Build per-request headers using taskId when available, falling back to sessionId const taskId = metadata?.taskId @@ -691,7 +694,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } // Handle streaming response - yield* this.handleStreamResponse(response.body, model) + yield* this.handleStreamResponse(response.body, model, requestController) } catch (error) { // Re-throw abort errors as-is so callers can identify cancellations if (error instanceof Error && error.name === "AbortError") { @@ -716,10 +719,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } finally { // Detach the bridging listener so a late abort from this request cannot // cancel a later request's controller. - if (abortListener) { - // Stryker disable next-line OptionalChaining: abortListener is only assigned inside the if (externalAbortSignal) block, so whenever the finally guard is true the signal is non-nullish; dropping the optional chain is behaviorally identical on every reachable path (covered by the abort-signal bridging specs) - externalAbortSignal?.removeEventListener("abort", abortListener) - } + detach?.() if (this.abortController === requestController) { this.abortController = undefined } @@ -732,8 +732,17 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio * This function iterates through the Server-Sent Events (SSE) stream, parses each event, * and yields structured data chunks (`ApiStream`). It handles a wide variety of event types, * including text deltas, reasoning, usage data, and various status/tool events. + * + * @param requestController - The request-local controller. When it aborts (external + * abort or request timeout), a stream error is surfaced as the contract AbortError + * instead of being wrapped, so a cancellation is never misreported as a provider + * error and is not captured twice. */ - private async *handleStreamResponse(body: ReadableStream, model: OpenAiNativeModel): ApiStream { + private async *handleStreamResponse( + body: ReadableStream, + model: OpenAiNativeModel, + requestController: AbortController, + ): ApiStream { const reader = body.getReader() const decoder = new TextDecoder() let buffer = "" @@ -1185,6 +1194,15 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // If we didn't get any content, don't throw - the API might have returned an empty response // This can happen in certain edge cases and shouldn't break the flow } catch (error) { + // The request-local controller is only aborted on external abort or request + // timeout, both legitimate cancellations: surface the contract abort error + // before any wrapping so the caller's AbortError guard sees one correctly + // named error instead of a wrapped DOMException, and the stop is not + // captured as an exception here or again by the caller. + if (requestController.signal.aborted) { + throw new DOMException(`The ${this.providerName} request was aborted`, "AbortError") + } + const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "createMessage") TelemetryService.instance.captureException(apiError) From 0488a39b70063908171440dcc4dd254394888a46 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 14:23:25 +0800 Subject: [PATCH 10/27] test(api): reuse the open-stream fetch fixture and assert listener identity in openai-native specs --- .../providers/__tests__/openai-native.spec.ts | 96 ++++++++----------- 1 file changed, 38 insertions(+), 58 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 1d2edbdabf..dd5d044070 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -137,6 +137,35 @@ describe("OpenAiNativeHandler", () => { }) describe("createMessage", () => { + function makeOpenStreamFetchMock() { + type OpenStream = { + controller?: ReadableStreamDefaultController + fetchSignal: AbortSignal + } + const openStreams: OpenStream[] = [] + const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { + const entry: OpenStream = { fetchSignal: options?.signal as AbortSignal } + const body = new ReadableStream({ + start: (controller) => { + entry.controller = controller + }, + }) + openStreams.push(entry) + return Promise.resolve({ + ok: true, + body, + }) + }) + const requireController = (index: number): ReadableStreamDefaultController => { + const entry = openStreams[index] + if (!entry?.controller) { + throw new Error("expected fallback fetch to have started") + } + return entry.controller + } + return { openStreams, mockFetch, requireController } + } + it("shapes GPT-6 Astra requests for Responses tool calling", () => { const astraHandler = new OpenAiNativeHandler({ ...mockOptions, @@ -495,34 +524,9 @@ describe("OpenAiNativeHandler", () => { // the SECOND request's controller. mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) - type OpenStream = { - controller?: ReadableStreamDefaultController - fetchSignal: AbortSignal - } - const openStreams: OpenStream[] = [] - const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { - const entry: OpenStream = { fetchSignal: options?.signal as AbortSignal } - const body = new ReadableStream({ - start: (controller) => { - entry.controller = controller - }, - }) - openStreams.push(entry) - return Promise.resolve({ - ok: true, - body, - }) - }) + const { openStreams, mockFetch, requireController } = makeOpenStreamFetchMock() global.fetch = mockFetch as typeof fetch - const requireController = (index: number): ReadableStreamDefaultController => { - const entry = openStreams[index] - if (!entry?.controller) { - throw new Error("expected fallback fetch to have started") - } - return entry.controller - } - const firstController = new AbortController() const secondController = new AbortController() @@ -607,35 +611,6 @@ describe("OpenAiNativeHandler", () => { return chunks.filter((chunk): chunk is ApiStreamTextChunk => chunk.type === "text") } - function makeOpenStreamFetchMock() { - type OpenStream = { - controller?: ReadableStreamDefaultController - fetchSignal: AbortSignal - } - const openStreams: OpenStream[] = [] - const mockFetch = vitest.fn().mockImplementation((_url: string, options?: RequestInit) => { - const entry: OpenStream = { fetchSignal: options?.signal as AbortSignal } - const body = new ReadableStream({ - start: (controller) => { - entry.controller = controller - }, - }) - openStreams.push(entry) - return Promise.resolve({ - ok: true, - body, - }) - }) - const requireController = (index: number): ReadableStreamDefaultController => { - const entry = openStreams[index] - if (!entry?.controller) { - throw new Error("expected fallback fetch to have started") - } - return entry.controller - } - return { openStreams, mockFetch, requireController } - } - it("should register a once-only abort listener on the external signal and detach it when the SDK request completes", async () => { const controller = new AbortController() const addSpy = vi.spyOn(controller.signal, "addEventListener") @@ -1045,9 +1020,14 @@ describe("OpenAiNativeHandler", () => { expect(call[0]).toBe("abort") expect(call[2]).toEqual({ once: true }) } - for (const call of removeSpy.mock.calls) { - expect(call[0]).toBe("abort") - } + const registered = addSpy.mock.calls.map(([, listener]) => listener) + const removed = removeSpy.mock.calls.map(([type, listener]) => { + expect(type).toBe("abort") + return listener + }) + // Each bridge must detach its own listener exactly once. + expect(new Set(removed).size).toBe(2) + expect(new Set(removed)).toEqual(new Set(registered)) expect(handler["abortController"]).toBeUndefined() } finally { addSpy.mockRestore() From e94868065e323c3be7f47f3b6a7d8912382c289b Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Mon, 28 Sep 2026 17:31:08 +0800 Subject: [PATCH 11/27] fix(api): per-request abort check in fallback SSE loop and release body on early exit --- .../providers/__tests__/openai-native.spec.ts | 145 ++++++++++++++++++ src/api/providers/openai-native.ts | 91 ++++++++++- 2 files changed, 229 insertions(+), 7 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index dd5d044070..8747dd3461 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -1206,6 +1206,151 @@ describe("OpenAiNativeHandler", () => { const chunks1 = await collected1 expect(textChunks(chunks1).map((chunk) => chunk.text)).toEqual(["one"]) }) + + it("should not issue a fallback POST when the SDK request was aborted by this request's own signal", async () => { + // Regression: the SDK catch path used to fall through to the manual SSE + // fallback for every error, including the AbortError raised when this + // request's own controller aborted — issuing a second POST for a request + // the caller already cancelled. + const mockFetch = vitest.fn() + global.fetch = mockFetch as typeof fetch + const controller = new AbortController() + let sdkSignal: AbortSignal | undefined + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + sdkSignal = options?.signal + // Emulate the SDK: an aborted in-flight request rejects with AbortError. + return new Promise((_resolve, reject) => { + if (!sdkSignal) { + reject(new Error("expected the SDK request to carry a request signal")) + return + } + if (sdkSignal.aborted) { + reject(makeAbortError()) + return + } + sdkSignal.addEventListener("abort", () => reject(makeAbortError()), { once: true }) + }) + }) + + const collected = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + await tick() + expect(sdkSignal).toBeDefined() + expect(sdkSignal?.aborted).toBe(false) + + controller.abort() + await expect(collected).rejects.toMatchObject({ name: "AbortError" }) + + // The fallback must not run: a cancelled request must not trigger a second POST. + expect(mockFetch).not.toHaveBeenCalled() + }) + + it("should cancel only the aborted fallback body when overlapping requests share a handler", async () => { + // Regression: the SSE loop gated on the shared this.abortController + // field, so with two overlapping fallback reads an abort of one + // request's signal could terminate (or be ignored by) the other + // request's stream. Each request must react only to its own + // request-local controller, and the aborted request's body must be + // cancelled before the reader lock is released so the socket is + // torn down. + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + + const firstExternal = new AbortController() + const secondExternal = new AbortController() + let firstReadSettled = false + const cancelCalls = [false, false] + let fetchCall = 0 + const mockFetch = vitest.fn().mockImplementation(() => { + fetchCall += 1 + if (fetchCall === 1) { + // First request's body: one chunk now, then a pull that stays + // pending until this request's external signal aborts, where it + // enqueues an empty [DONE] chunk so the loop's next per-request + // abort check can break it. + const body = new ReadableStream({ + start: (startController) => { + startController.enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"one"}\n\n'), + ) + }, + pull: (pullController) => + new Promise((resolve) => { + const onAbort = () => { + if (!firstReadSettled) { + firstReadSettled = true + pullController.enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + } + resolve() + } + if (firstExternal.signal.aborted) { + onAbort() + return + } + firstExternal.signal.addEventListener("abort", onAbort, { once: true }) + }), + cancel: () => { + cancelCalls[0] = true + }, + }) + return Promise.resolve({ ok: true, body }) + } + // Second request's body: completes normally. + const body = new ReadableStream({ + start: (startController) => { + startController.enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"two"}\n\n'), + ) + startController.enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + startController.close() + }, + cancel: () => { + cancelCalls[1] = true + }, + }) + return Promise.resolve({ ok: true, body }) + }) + global.fetch = mockFetch as typeof fetch + + const collected1 = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: firstExternal.signal }), + ), + ) + await tick() + const collected2 = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: secondExternal.signal }), + ), + ) + await tick() + + // Abort only the first request's signal while both fallback reads are open. + firstExternal.abort() + + // The second request must complete normally with its own content and + // must not have its body cancelled. + const chunks2 = await collected2 + expect(textChunks(chunks2).map((chunk) => chunk.text)).toEqual(["two"]) + expect(secondExternal.signal.aborted).toBe(false) + expect(cancelCalls[1]).toBe(false) + + // The first request reacts to its own abort: the loop breaks once the + // pending read settles, and its body is cancelled before the lock is + // released. No second POST is issued for the aborted request. + const chunks1 = await collected1 + expect(textChunks(chunks1).map((chunk) => chunk.text)).toEqual(["one"]) + expect(cancelCalls[0]).toBe(true) + expect(fetchCall).toBe(2) + }) }) }) diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index 0ab0c644c7..7e3b5a3b66 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -436,6 +436,23 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } } + /** + * Streams a Responses API request through the official OpenAI SDK. + * + * The task's external abort signal is bridged onto a request-local + * controller; the shared this.abortController field only exposes that + * controller so the stop-button path can observe it. If the SDK path + * fails, the request is retried once through the manual SSE fallback + * (makeResponsesApiRequest) — except when the SDK error is the cancellation + * itself (an AbortError from this request's aborted controller), which is + * rethrown as-is so a cancelled request never triggers a second POST. + * + * @param requestBody - The serialized Responses API request body + * @param model - The resolved model configuration + * @param metadata - Per-request metadata (external abortSignal, taskId, ...) + * @param systemPrompt - Optional system prompt forwarded to the fallback path + * @param messages - Optional message history forwarded to the fallback path + */ private async *executeRequest( requestBody: any, model: OpenAiNativeModel, @@ -488,6 +505,15 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } } } catch (sdkErr: any) { + // If this request's own controller aborted and the SDK error is the + // cancellation itself (an AbortError from the aborted request signal), + // rethrow it unchanged instead of issuing a second POST through the + // fallback path for a request the caller already cancelled. Other SDK + // failures (including a pre-aborted signal hitting a broken SDK) still + // take the fallback, whose own aborted fetch surfaces the AbortError. + if (requestController.signal.aborted && sdkErr instanceof Error && sdkErr.name === "AbortError") { + throw sdkErr + } // For errors, fallback to manual SSE via fetch yield* this.makeResponsesApiRequest(requestBody, model, metadata, systemPrompt, messages) } finally { @@ -598,6 +624,23 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio return formattedInput } + /** + * Fallback streaming path: POSTs the Responses API request with fetch and + * streams the raw SSE body. + * + * Used when the SDK path is unavailable or fails. Like executeRequest, the + * external abort signal is bridged onto a request-local controller so an + * overlapping request can never cancel this one; the bridging listener is + * detached in the finally block and abort errors are rethrown as-is so + * callers can identify cancellations. systemPrompt and messages are + * retained for signature compatibility with the SDK path. + * + * @param requestBody - The serialized Responses API request body + * @param model - The resolved model configuration + * @param metadata - Per-request metadata (external abortSignal, taskId, ...) + * @param systemPrompt - Unused; retained for signature compatibility + * @param messages - Unused; retained for signature compatibility + */ private async *makeResponsesApiRequest( requestBody: any, model: OpenAiNativeModel, @@ -739,10 +782,15 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio * and yields structured data chunks (`ApiStream`). It handles a wide variety of event types, * including text deltas, reasoning, usage data, and various status/tool events. * - * @param requestController - The request-local controller. When it aborts (external - * abort or request timeout), a stream error is surfaced as the contract AbortError - * instead of being wrapped, so a cancellation is never misreported as a provider - * error and is not captured twice. + * The loop only reacts to the request-local requestController (never the shared + * this.abortController field), so overlapping requests can only cancel + * themselves. When it aborts (external abort or request timeout), a stream + * error is surfaced as the contract AbortError instead of being wrapped, so a + * cancellation is never misreported as a provider error and is not captured + * twice. On any early exit the body reader is cancelled before the lock is + * released so the underlying socket is torn down. + * + * @param requestController - The request-local controller for this stream */ private async *handleStreamResponse( body: ReadableStream, @@ -753,18 +801,25 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio const decoder = new TextDecoder() let buffer = "" let hasContent = false + let streamCompleted = false const totalInputTokens = 0 const totalOutputTokens = 0 try { while (true) { - // Check if request was aborted - if (this.abortController?.signal.aborted) { + // Check if THIS request's own controller was aborted. The shared + // this.abortController field must not gate the loop: with overlapping + // requests it reflects the most recently created request, so reading it + // here would let one request's cancellation terminate another's stream. + if (requestController.signal.aborted) { break } const { done, value } = await reader.read() - if (done) break + if (done) { + streamCompleted = true + break + } buffer += decoder.decode(value, { stream: true }) const lines = buffer.split("\n") @@ -1218,6 +1273,15 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } throw new Error("Unexpected error processing response stream") } finally { + // An early exit (abort, error, or the consumer abandoning the generator) + // leaves the body still open: cancel it before releasing the lock so the + // socket is torn down and a dangling stream cannot be terminated by — or + // terminate — an overlapping request's stream. Cancel is a no-op once the + // stream has already completed or failed; the guarded catch keeps this + // best-effort cleanup from masking the original error. + if (!streamCompleted) { + await reader.cancel().catch(() => undefined) + } reader.releaseLock() } } @@ -1585,6 +1649,19 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio getResponseId(): string | undefined { return this.lastResponseId } + /** + * Completes a prompt using the Responses API (non-streaming). + * + * The request signal is built request-locally from the caller's abort + * signal and optional timeout (RequestConfigBuilder.mergeAbortSignalAndTimeout), + * so cancellations never touch this.abortController, which streaming + * requests own. Abort errors are rethrown as-is so callers can identify + * cancellations. + * + * @param prompt - The user prompt to complete + * @param options - Optional abortSignal/timeoutMs controlling the request + * @returns The completed text content + */ async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { // Request-local abort signal: merges the external abort signal with an optional // timeout without touching this.abortController (owned by streaming requests). From df16983aa554d12f3534de4b2b3e8ccde1479534 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Mon, 28 Sep 2026 18:10:58 +0800 Subject: [PATCH 12/27] fix(api): normalize completePrompt SDK aborts to the contract AbortError --- .../providers/__tests__/openai-native.spec.ts | 28 +++++++++++++++++++ src/api/providers/openai-native.ts | 18 ++++++++---- 2 files changed, 41 insertions(+), 5 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 8747dd3461..6bb6acd807 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -1568,6 +1568,34 @@ describe("OpenAiNativeHandler", () => { }) }) + it("completePrompt should normalize SDK APIUserAbortError to the contract AbortError without telemetry", async () => { + // The locked OpenAI SDK raises APIUserAbortError (message + // "Request was aborted.") instead of a native AbortError when the + // request signal aborts. Cancellations must surface as the contract + // AbortError and must never be captured as provider telemetry. + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + const error = new Error("Request was aborted.") + error.name = "APIUserAbortError" + if (options?.signal?.aborted) { + reject(error) + return + } + options?.signal?.addEventListener("abort", () => reject(error), { once: true }) + }) + }) + + const controller = new AbortController() + const pending = handler.completePrompt("Test prompt", { abortSignal: controller.signal }) + await new Promise((resolve) => setTimeout(resolve, 0)) + controller.abort() + + await expect(pending).rejects.toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) + expect(mockCaptureException).not.toHaveBeenCalled() + }) it("completePrompt should rethrow non-Error failures after telemetry", async () => { mockResponsesCreate.mockRejectedValue("string failure") diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index 7e3b5a3b66..3f8c4cb5c3 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -33,6 +33,7 @@ import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, Complete import { isMcpTool } from "../../utils/mcp-name" import { sanitizeOpenAiCallId } from "../../utils/tool-id" import { RequestConfigBuilder } from "./config-builder/request-config-builder" +import { createAbortError, isRequestAborted } from "./utils/abort-signal" export type OpenAiNativeModel = ReturnType @@ -1655,8 +1656,10 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio * The request signal is built request-locally from the caller's abort * signal and optional timeout (RequestConfigBuilder.mergeAbortSignalAndTimeout), * so cancellations never touch this.abortController, which streaming - * requests own. Abort errors are rethrown as-is so callers can identify - * cancellations. + * requests own. Cancellations are normalized to the contract AbortError via + * isRequestAborted/createAbortError (the SDK's own APIUserAbortError does + * not satisfy the contract) so callers can identify them and they are never + * captured as provider errors. * * @param prompt - The user prompt to complete * @param options - Optional abortSignal/timeoutMs controlling the request @@ -1751,9 +1754,14 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio return "" } catch (error) { - // Re-throw abort errors as-is so callers can identify cancellations - if (error instanceof Error && error.name === "AbortError") { - throw error + // Normalize cancellations to the contract AbortError before anything + // else: the locked OpenAI SDK raises APIUserAbortError (or its + // "Request was aborted." message) instead of a native AbortError when + // requestSignal aborts, and the raw SDK error does not satisfy the + // Task's abort contract — rethrowing it would surface a cancellation + // as a wrapped provider error and capture it as telemetry. + if (isRequestAborted(error, requestSignal)) { + throw createAbortError(this.providerName) } const errorModel = this.getModel() From bfc880c991a6307dcfd7288b4c6313783d4f4d85 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Mon, 28 Sep 2026 19:23:23 +0800 Subject: [PATCH 13/27] fix(api): classify SDK cancellations from the error alone, never from the signal --- .../providers/__tests__/openai-native.spec.ts | 58 +++++++++++++++++++ src/api/providers/openai-native.ts | 50 ++++++++++------ 2 files changed, 89 insertions(+), 19 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 6bb6acd807..01eada86a5 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -1250,6 +1250,50 @@ describe("OpenAiNativeHandler", () => { expect(mockFetch).not.toHaveBeenCalled() }) + it("should normalize SDK APIUserAbortError without entering the fallback when the request's own signal aborts", async () => { + // Regression: the SDK catch path only recognized native AbortErrors, so the + // OpenAI SDK's own APIUserAbortError fell through into the manual SSE + // fallback — an unnecessary fallback path for a request the caller already + // cancelled. + const mockFetch = vitest.fn().mockImplementation((_url: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + const error = new Error("This operation was aborted") + error.name = "AbortError" + return Promise.reject(error) + } + return new Promise(() => undefined) + }) + global.fetch = mockFetch as typeof fetch + const controller = new AbortController() + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + const error = new Error("Request was aborted.") + error.name = "APIUserAbortError" + if (options?.signal?.aborted) { + reject(error) + return + } + options?.signal?.addEventListener("abort", () => reject(error), { once: true }) + }) + }) + + const collected = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + await tick() + controller.abort() + + await expect(collected).rejects.toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) + expect(mockFetch).not.toHaveBeenCalled() + }) + it("should cancel only the aborted fallback body when overlapping requests share a handler", async () => { // Regression: the SSE loop gated on the shared this.abortController // field, so with two overlapping fallback reads an abort of one @@ -1596,6 +1640,20 @@ describe("OpenAiNativeHandler", () => { }) expect(mockCaptureException).not.toHaveBeenCalled() }) + it("completePrompt should not reclassify a terminal SDK error as a cancellation when the signal aborts", async () => { + // Regression: classifying cancellation from requestSignal instead of the + // error let a signal that aborts after a terminal SDK error convert that + // error into an AbortError and skip telemetry. + mockResponsesCreate.mockRejectedValue(new Error("API Error")) + const controller = new AbortController() + controller.abort() + + await expect(handler.completePrompt("Test prompt", { abortSignal: controller.signal })).rejects.toThrow( + "OpenAI Native completion error: API Error", + ) + expect(mockCaptureException).toHaveBeenCalledTimes(1) + }) + it("completePrompt should rethrow non-Error failures after telemetry", async () => { mockResponsesCreate.mockRejectedValue("string failure") diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index 3f8c4cb5c3..ca2891c70b 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -445,8 +445,10 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio * controller so the stop-button path can observe it. If the SDK path * fails, the request is retried once through the manual SSE fallback * (makeResponsesApiRequest) — except when the SDK error is the cancellation - * itself (an AbortError from this request's aborted controller), which is - * rethrown as-is so a cancelled request never triggers a second POST. + * itself (recognized from the error alone via isRequestAborted while this + * request's controller is aborted): native AbortErrors are rethrown as-is + * and other recognized SDK abort errors are normalized to the contract + * AbortError, so a cancelled request never triggers a second POST. * * @param requestBody - The serialized Responses API request body * @param model - The resolved model configuration @@ -507,13 +509,20 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } } catch (sdkErr: any) { // If this request's own controller aborted and the SDK error is the - // cancellation itself (an AbortError from the aborted request signal), - // rethrow it unchanged instead of issuing a second POST through the - // fallback path for a request the caller already cancelled. Other SDK - // failures (including a pre-aborted signal hitting a broken SDK) still - // take the fallback, whose own aborted fetch surfaces the AbortError. - if (requestController.signal.aborted && sdkErr instanceof Error && sdkErr.name === "AbortError") { - throw sdkErr + // cancellation itself — classified from the error alone via + // isRequestAborted, never from the signal, so a terminal SDK error + // that merely races the abort is not reclassified — surface it as + // the contract AbortError instead of entering the fallback path for + // a request the caller already cancelled: native AbortErrors are + // rethrown as-is, other recognized SDK abort errors are normalized. + // Other SDK failures (including a pre-aborted signal hitting a broken + // SDK) still take the fallback, whose own aborted fetch surfaces the + // AbortError. + if (requestController.signal.aborted && isRequestAborted(sdkErr)) { + if (sdkErr instanceof Error && sdkErr.name === "AbortError") { + throw sdkErr + } + throw createAbortError(this.providerName) } // For errors, fallback to manual SSE via fetch yield* this.makeResponsesApiRequest(requestBody, model, metadata, systemPrompt, messages) @@ -1656,10 +1665,10 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio * The request signal is built request-locally from the caller's abort * signal and optional timeout (RequestConfigBuilder.mergeAbortSignalAndTimeout), * so cancellations never touch this.abortController, which streaming - * requests own. Cancellations are normalized to the contract AbortError via - * isRequestAborted/createAbortError (the SDK's own APIUserAbortError does - * not satisfy the contract) so callers can identify them and they are never - * captured as provider errors. + * requests own. Cancellations are classified from the error alone via + * isRequestAborted (the SDK's own APIUserAbortError never satisfies the + * contract) and normalized with createAbortError, so callers can identify + * them and terminal SDK errors are never misreported as cancellations. * * @param prompt - The user prompt to complete * @param options - Optional abortSignal/timeoutMs controlling the request @@ -1755,12 +1764,15 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio return "" } catch (error) { // Normalize cancellations to the contract AbortError before anything - // else: the locked OpenAI SDK raises APIUserAbortError (or its - // "Request was aborted." message) instead of a native AbortError when - // requestSignal aborts, and the raw SDK error does not satisfy the - // Task's abort contract — rethrowing it would surface a cancellation - // as a wrapped provider error and capture it as telemetry. - if (isRequestAborted(error, requestSignal)) { + // Normalize cancellations to the contract AbortError before anything + // else. Cancellation is classified from the error alone — never from + // requestSignal — because the locked OpenAI SDK raises APIUserAbortError + // (or its "Request was aborted." message) when requestSignal aborts, + // and a signal that aborts after a terminal SDK error must not turn + // that error into a cancellation. The raw SDK abort error does not + // satisfy the Task's abort contract, so it is normalized instead of + // rethrown; terminal errors keep the telemetry + wrapping below. + if (isRequestAborted(error)) { throw createAbortError(this.providerName) } From 3717280c00b886d35245e42081286902ffdb6834 Mon Sep 17 00:00:00 2001 From: easonLiangWorldedtech Date: Sat, 3 Oct 2026 01:40:52 +0800 Subject: [PATCH 14/27] preflight synthetic head From f9d5d00baddb836d124148e800ad08170a4b9769 Mon Sep 17 00:00:00 2001 From: easonLiangWorldedtech Date: Sat, 3 Oct 2026 01:41:34 +0800 Subject: [PATCH 15/27] test: kill catch-path mutants + document equivalent streamCompleted cleanup --- .../providers/__tests__/openai-native.spec.ts | 77 +++++++++++++++++++ src/api/providers/openai-native.ts | 2 + 2 files changed, 79 insertions(+) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 01eada86a5..1f46517180 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -1294,6 +1294,83 @@ describe("OpenAiNativeHandler", () => { expect(mockFetch).not.toHaveBeenCalled() }) + it("should take the fallback when the request's own signal aborted but the SDK error is not a cancellation", async () => { + // Regression guard for the catch-path condition: a terminal SDK error that + // merely races the abort must not be reclassified as a cancellation, so the + // fallback still runs instead of throwing an AbortError. + const mockFetch = vitest.fn().mockImplementation((_url: unknown, options: { signal?: AbortSignal }) => { + const body = new ReadableStream({ + start: (startController) => { + startController.enqueue( + new TextEncoder().encode('data: {"type":"response.text.delta","delta":"fallback"}\n\n'), + ); + startController.enqueue(new TextEncoder().encode("data: [DONE]\n\n")); + startController.close(); + }, + }); + return Promise.resolve({ ok: true, body }); + }); + global.fetch = mockFetch as typeof fetch; + const controller = new AbortController(); + const terminalError = new Error("SDK connection reset"); + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + if (options?.signal?.aborted) { + reject(terminalError); + return; + } + options?.signal?.addEventListener("abort", () => reject(terminalError), { once: true }); + }); + }); + + const collected = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ); + await tick(); + controller.abort(); + + // The fallback is entered (the fetch is issued) rather than an AbortError + // being thrown; the loop's own aborted check then breaks before reading. + await collected; + expect(mockFetch).toHaveBeenCalled(); + }) + + it("should rethrow a native AbortError from the SDK as-is instead of normalizing it", async () => { + // Regression guard for the name check: a native AbortError from the SDK must + // be rethrown as-is, not replaced by the normalized contract error. + const mockFetch = vitest.fn(); + global.fetch = mockFetch as typeof fetch; + const controller = new AbortController(); + const nativeError = new Error("socket closed by abort"); + nativeError.name = "AbortError"; + mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + if (options?.signal?.aborted) { + reject(nativeError); + return; + } + options?.signal?.addEventListener("abort", () => reject(nativeError), { once: true }); + }); + }); + + const collected = collectStream( + handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ); + await tick(); + controller.abort(); + + await expect(collected).rejects.toBe(nativeError); + expect(mockFetch).not.toHaveBeenCalled(); + }) + it("should cancel only the aborted fallback body when overlapping requests share a handler", async () => { // Regression: the SSE loop gated on the shared this.abortController // field, so with two overlapping fallback reads an abort of one diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index ca2891c70b..d98e6a8086 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -827,6 +827,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio const { done, value } = await reader.read() if (done) { + // Stryker disable next-line BooleanLiteral: done from reader.read() implies the stream is closed, and cancel() on a closed stream never invokes the underlying cancel algorithm, so this flag cannot change observable behavior. streamCompleted = true break } @@ -1289,6 +1290,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // terminate — an overlapping request's stream. Cancel is a no-op once the // stream has already completed or failed; the guarded catch keeps this // best-effort cleanup from masking the original error. + // Stryker disable next-line ConditionalExpression: when streamCompleted is true the stream is already closed, so reader.cancel() is a no-op; forcing this branch true cannot change observable behavior. if (!streamCompleted) { await reader.cancel().catch(() => undefined) } From fff974d9f0c68b7c7145ab3b64b72f7561939387 Mon Sep 17 00:00:00 2001 From: easonLiangWorldedtech Date: Sat, 3 Oct 2026 01:41:35 +0800 Subject: [PATCH 16/27] preflight synthetic head From 5e983df64dd2c092c617b6689a341a3779103a8c Mon Sep 17 00:00:00 2001 From: easonLiangWorldedtech Date: Sat, 3 Oct 2026 02:37:49 +0800 Subject: [PATCH 17/27] chore: re-trigger CI (flaky e2e-mock restart-persistence timeout) From 8389c31ab58bd44c6c86266801c4e3e0de72b0ca Mon Sep 17 00:00:00 2001 From: easonLiangWorldedtech Date: Sat, 3 Oct 2026 03:02:35 +0800 Subject: [PATCH 18/27] test(abort-r1): assert the fallback request signal and the empty stream after cancellation --- .../providers/__tests__/openai-native.spec.ts | 89 ++++++++++--------- 1 file changed, 48 insertions(+), 41 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 1f46517180..aacea681d7 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -1303,74 +1303,81 @@ describe("OpenAiNativeHandler", () => { start: (startController) => { startController.enqueue( new TextEncoder().encode('data: {"type":"response.text.delta","delta":"fallback"}\n\n'), - ); - startController.enqueue(new TextEncoder().encode("data: [DONE]\n\n")); - startController.close(); + ) + startController.enqueue(new TextEncoder().encode("data: [DONE]\n\n")) + startController.close() }, - }); - return Promise.resolve({ ok: true, body }); - }); - global.fetch = mockFetch as typeof fetch; - const controller = new AbortController(); - const terminalError = new Error("SDK connection reset"); + }) + return Promise.resolve({ ok: true, body }) + }) + global.fetch = mockFetch as typeof fetch + const controller = new AbortController() + const terminalError = new Error("SDK connection reset") mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { return new Promise((_resolve, reject) => { if (options?.signal?.aborted) { - reject(terminalError); - return; + reject(terminalError) + return } - options?.signal?.addEventListener("abort", () => reject(terminalError), { once: true }); - }); - }); - + options?.signal?.addEventListener("abort", () => reject(terminalError), { once: true }) + }) + }) + const collected = collectStream( handler.createMessage( systemPrompt, messages, makeCreateMessageMetadata({ abortSignal: controller.signal }), ), - ); - await tick(); - controller.abort(); - + ) + await tick() + controller.abort() + // The fallback is entered (the fetch is issued) rather than an AbortError // being thrown; the loop's own aborted check then breaks before reading. - await collected; - expect(mockFetch).toHaveBeenCalled(); + // Both halves of the contract must hold: the fallback request carries an + // already-aborted signal (so a real fetch would reject instead of issuing a + // second POST) and the stream yields nothing after cancellation. + const chunks = await collected + expect(mockFetch).toHaveBeenCalled() + const fetchOptions = mockFetch.mock.calls[0]?.[1] as { signal?: AbortSignal } | undefined + expect(fetchOptions?.signal).toBeInstanceOf(AbortSignal) + expect(fetchOptions?.signal?.aborted).toBe(true) + expect(chunks).toHaveLength(0) }) - + it("should rethrow a native AbortError from the SDK as-is instead of normalizing it", async () => { // Regression guard for the name check: a native AbortError from the SDK must // be rethrown as-is, not replaced by the normalized contract error. - const mockFetch = vitest.fn(); - global.fetch = mockFetch as typeof fetch; - const controller = new AbortController(); - const nativeError = new Error("socket closed by abort"); - nativeError.name = "AbortError"; + const mockFetch = vitest.fn() + global.fetch = mockFetch as typeof fetch + const controller = new AbortController() + const nativeError = new Error("socket closed by abort") + nativeError.name = "AbortError" mockResponsesCreate.mockImplementation((_body: unknown, options: { signal?: AbortSignal }) => { return new Promise((_resolve, reject) => { if (options?.signal?.aborted) { - reject(nativeError); - return; + reject(nativeError) + return } - options?.signal?.addEventListener("abort", () => reject(nativeError), { once: true }); - }); - }); - + options?.signal?.addEventListener("abort", () => reject(nativeError), { once: true }) + }) + }) + const collected = collectStream( handler.createMessage( systemPrompt, messages, makeCreateMessageMetadata({ abortSignal: controller.signal }), ), - ); - await tick(); - controller.abort(); - - await expect(collected).rejects.toBe(nativeError); - expect(mockFetch).not.toHaveBeenCalled(); - }) - + ) + await tick() + controller.abort() + + await expect(collected).rejects.toBe(nativeError) + expect(mockFetch).not.toHaveBeenCalled() + }) + it("should cancel only the aborted fallback body when overlapping requests share a handler", async () => { // Regression: the SSE loop gated on the shared this.abortController // field, so with two overlapping fallback reads an abort of one From 32bbf6b1dc623b1815ec1d375a4c8b9be22df9d8 Mon Sep 17 00:00:00 2001 From: easonLiangWorldedtech Date: Sat, 3 Oct 2026 03:53:06 +0800 Subject: [PATCH 19/27] test(abort-r1): pin the fallback fetch to exactly one call --- src/api/providers/__tests__/openai-native.spec.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index aacea681d7..3ef8d2b7ba 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -1339,7 +1339,7 @@ describe("OpenAiNativeHandler", () => { // already-aborted signal (so a real fetch would reject instead of issuing a // second POST) and the stream yields nothing after cancellation. const chunks = await collected - expect(mockFetch).toHaveBeenCalled() + expect(mockFetch).toHaveBeenCalledTimes(1) const fetchOptions = mockFetch.mock.calls[0]?.[1] as { signal?: AbortSignal } | undefined expect(fetchOptions?.signal).toBeInstanceOf(AbortSignal) expect(fetchOptions?.signal?.aborted).toBe(true) From c942a4574be8c3b775d6a2f50994e5c6a7878b75 Mon Sep 17 00:00:00 2001 From: easonLiangWorldedtech Date: Sat, 3 Oct 2026 04:23:38 +0800 Subject: [PATCH 20/27] fix(abort-r1): hand back the contract AbortError on a mid-stream abort --- src/api/providers/__tests__/openai-native.spec.ts | 9 +++++++-- src/api/providers/openai-native.ts | 6 ++++-- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 3ef8d2b7ba..0ea0057f78 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -753,8 +753,13 @@ describe("OpenAiNativeHandler", () => { controller.abort() - const chunks = await collected - expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["first"]) + // The abort lands while the generator is parked, so the next iteration hands back + // the contract AbortError instead of the buffered "second" chunk: the stream stops + // consuming and reports the cancellation as an AbortError. + await expect(collected).rejects.toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) }) it("should detach the external abort listener on completion so a late abort cannot abort the request signal", async () => { diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index d98e6a8086..169accf012 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -498,9 +498,11 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } for await (const event of stream) { - // Check if request was aborted + // An abort observed mid-stream must hand back the contract AbortError instead of + // resolving with a partial stream: cancellation is identified through an error named + // "AbortError", the same way the inner loop and the SSE fallback path in this file do. if (requestController.signal.aborted) { - break + throw createAbortError(this.providerName) } for await (const outChunk of this.processEvent(event, model)) { From 25417473a38b0022b04e899328b66d13fd5405d6 Mon Sep 17 00:00:00 2001 From: easonLiangWorldedtech Date: Sat, 3 Oct 2026 04:56:57 +0800 Subject: [PATCH 21/27] fix(abort-r1): report the contract AbortError when the SDK iterator ends after an abort --- .../providers/__tests__/openai-native.spec.ts | 44 ++++++++++++++----- src/api/providers/openai-native.ts | 8 ++++ 2 files changed, 42 insertions(+), 10 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 0ea0057f78..403e6a069f 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -713,6 +713,10 @@ describe("OpenAiNativeHandler", () => { makeCreateMessageMetadata({ abortSignal: controller.signal }), ) const collected = collectStream(stream) + // Attach the handler before the abort fires: the post-loop check rejects the stream + // as soon as the generator ends, so a late handler would surface as an unhandled + // rejection. + const settled = collected.catch((error) => error) await tick() expect(sdkSignal).toBeDefined() @@ -724,8 +728,14 @@ describe("OpenAiNativeHandler", () => { // ... but it must abort the request as soon as it fires. expect(sdkSignal?.aborted).toBe(true) - const chunks = await collected - expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["one"]) + // The SDK iterator returns normally when the abort lands while it awaits the + // next event, so the post-loop check must report the contract AbortError instead + // of resolving with the partial stream. + const error = await settled + expect(error).toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) }) it("should stop consuming the SDK stream once the external signal aborts the request", async () => { @@ -748,18 +758,29 @@ describe("OpenAiNativeHandler", () => { messages, makeCreateMessageMetadata({ abortSignal: controller.signal }), ) - const collected = collectStream(stream) + const chunks: ApiStreamChunk[] = [] + // Consume from before the abort so both halves of the contract are visible: the stream + // stops consuming before the post-abort event is processed, and it reports the + // cancellation as the contract AbortError. + const consumed = (async () => { + try { + for await (const chunk of stream) { + chunks.push(chunk) + } + } catch (error) { + return error + } + return undefined + })() await tick() controller.abort() - // The abort lands while the generator is parked, so the next iteration hands back - // the contract AbortError instead of the buffered "second" chunk: the stream stops - // consuming and reports the cancellation as an AbortError. - await expect(collected).rejects.toMatchObject({ + expect(await consumed).toMatchObject({ name: "AbortError", message: "The OpenAI Native request was aborted", }) + expect(textChunks(chunks).map((chunk) => chunk.text)).toEqual(["first"]) }) it("should detach the external abort listener on completion so a late abort cannot abort the request signal", async () => { @@ -1203,13 +1224,16 @@ describe("OpenAiNativeHandler", () => { expect(textChunks(chunks2).map((chunk) => chunk.text)).toEqual(["two-a", " two-b"]) expect(secondExternal.signal.aborted).toBe(false) - // Let the first request wind down; its stream simply ends. + // Let the first request wind down: it is the aborted one, so it hands back + // the contract AbortError rather than ending as a partial stream. if (!firstGateOpen) { throw new Error("expected the first stream gate to be ready") } firstGateOpen() - const chunks1 = await collected1 - expect(textChunks(chunks1).map((chunk) => chunk.text)).toEqual(["one"]) + await expect(collected1).rejects.toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) }) it("should not issue a fallback POST when the SDK request was aborted by this request's own signal", async () => { diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index 169accf012..6c63f32ee3 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -509,6 +509,14 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio yield outChunk } } + + // The resolved openai SDK swallows the abort inside its iterator and returns + // normally, so a cancellation that lands while it awaits the next event never runs + // the in-loop check. Report the contract AbortError for that path too, otherwise + // createMessage resolves with a partial stream. + if (requestController.signal.aborted) { + throw createAbortError(this.providerName) + } } catch (sdkErr: any) { // If this request's own controller aborted and the SDK error is the // cancellation itself — classified from the error alone via From 863c5e7aa13e223935864a14764e39bb3980b59b Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Mon, 12 Oct 2026 23:06:51 +0800 Subject: [PATCH 22/27] test(providers): pin in-flight abort rejection for openai-compatible completePrompt CodeRabbit flagged that the completePrompt abort tests resolve the mock immediately, so the in-flight cancellation path was never observed. The new test keeps generateText pending, rejects with the real DOMException when the signal aborts, and asserts the provider's rejection after a mid-request abort - pinning the merged catch's abort guard (negative control: removing the guard turns this test red, the wrap would rename the error). 32/32 spec passes, eslint 0, prettier clean. --- .../__tests__/openai-compatible.spec.ts | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/src/api/providers/__tests__/openai-compatible.spec.ts b/src/api/providers/__tests__/openai-compatible.spec.ts index fbc3073298..e78d1cdc08 100644 --- a/src/api/providers/__tests__/openai-compatible.spec.ts +++ b/src/api/providers/__tests__/openai-compatible.spec.ts @@ -599,6 +599,33 @@ describe("OpenAICompatibleHandler abort wiring", () => { expect(mergedSignal.aborted).toBe(true) }) + it("should reject with AbortError when the caller aborts while the request is pending", async () => { + // Keep generateText pending and reject only when the signal aborts, so the + // assertion covers the provider's in-flight cancellation path: a mock that + // resolves immediately would settle completePrompt before any cancellation + // is observable. This also pins the merged catch's abort guard - the raw + // DOMException must survive without the provider error-wrap. + mockGenerateText.mockImplementation((options: { abortSignal?: AbortSignal }) => { + return new Promise<{ text: string }>((resolve, reject) => { + const onAbort = () => reject(new DOMException("The operation was aborted.", "AbortError")) + if (options.abortSignal?.aborted) { + onAbort() + return + } + options.abortSignal?.addEventListener("abort", onAbort, { once: true }) + }) + }) + + const controller = new AbortController() + const pending = handler.completePrompt("test prompt", { abortSignal: controller.signal }) + controller.abort() + + await expect(pending).rejects.toMatchObject({ + name: "AbortError", + message: "The operation was aborted.", + }) + }) + it("should let the timeout component of a merged signal fire without the caller signal", async () => { mockGenerateText.mockResolvedValue({ text: "response" }) From d282fa3594029fef38ea1246346ef686e6633936 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Tue, 13 Oct 2026 11:27:04 +0800 Subject: [PATCH 23/27] Address review: throw the contract AbortError at the SSE loop boundary The between-pulls abort check in handleStreamResponse broke out of the loop, so a cancellation landing after a chunk was handed out and before the next pull resolved createMessage with partial output instead of surfacing the cancellation. The check now throws the contract AbortError, mirroring the SDK path's own between-pulls checks. Two existing tests encoded the break behavior (empty/partial resolution after cancellation) and are updated to expect the contract AbortError; a new test pins cancellation between successive pulls with a body that hangs after one chunk (negative control verified: reverting the throw to break fails the new test). Verification at this head: 114/114 green in openai-native.spec.ts; tsc --noEmit 0 errors; eslint --max-warnings=0 clean on touched files; prettier clean. --- .../providers/__tests__/openai-native.spec.ts | 82 ++++++++++++++++--- src/api/providers/openai-native.ts | 6 +- 2 files changed, 74 insertions(+), 14 deletions(-) diff --git a/src/api/providers/__tests__/openai-native.spec.ts b/src/api/providers/__tests__/openai-native.spec.ts index 4b84354a9d..b75f75cfe4 100644 --- a/src/api/providers/__tests__/openai-native.spec.ts +++ b/src/api/providers/__tests__/openai-native.spec.ts @@ -1145,6 +1145,55 @@ describe("OpenAiNativeHandler", () => { expect(mockCaptureException).not.toHaveBeenCalled() }) + it("should reject with the contract AbortError when the signal aborts between successive SSE pulls", async () => { + // Force the fallback path with a body that hands out exactly one SSE + // chunk and then hangs on the next pull, so only the loop-boundary + // abort check can settle the stream. Before the fix that check broke + // out of the loop and createMessage resolved with the partial output + // instead of surfacing the cancellation. + mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) + const mockFetch = vitest.fn().mockImplementation(() => { + const encoder = new TextEncoder() + let pulls = 0 + const body = new ReadableStream({ + pull: (streamController) => { + pulls++ + if (pulls === 1) { + streamController.enqueue( + encoder.encode('data: {"type":"response.output_text.delta","delta":"partial"}\n\n'), + ) + return + } + // Never settles: the between-pulls abort check is the only + // exit for a cancelled consumer. + return new Promise(() => {}) + }, + }) + return Promise.resolve({ ok: true, body }) + }) + global.fetch = mockFetch as typeof fetch + + const signalController = new AbortController() + const stream = handler.createMessage( + systemPrompt, + messages, + makeCreateMessageMetadata({ abortSignal: signalController.signal }), + ) + const iterator = stream[Symbol.asyncIterator]() + const first = await iterator.next() + expect(first.done).toBe(false) + expect(first.value).toMatchObject({ type: "text", text: "partial" }) + + // The cancellation lands after a chunk was handed out and before the + // next pull: the loop-boundary check must throw, not break. + signalController.abort() + await expect(iterator.next()).rejects.toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) + expect(mockCaptureException).not.toHaveBeenCalled() + }) + it("should not convert a non-abort stream error into an AbortError", async () => { mockResponsesCreate.mockRejectedValue(new Error("SDK not available")) const mockFetch = vitest.fn().mockImplementation(() => { @@ -1362,17 +1411,20 @@ describe("OpenAiNativeHandler", () => { await tick() controller.abort() - // The fallback is entered (the fetch is issued) rather than an AbortError - // being thrown; the loop's own aborted check then breaks before reading. - // Both halves of the contract must hold: the fallback request carries an - // already-aborted signal (so a real fetch would reject instead of issuing a - // second POST) and the stream yields nothing after cancellation. - const chunks = await collected + // The fallback is entered (the fetch is issued) rather than the SDK error + // being reclassified as a cancellation. Both halves of the contract must + // hold: the fallback request carries an already-aborted signal (so a real + // fetch would reject instead of issuing a second POST), and the loop's + // aborted check surfaces the cancellation as the contract AbortError + // instead of resolving with an empty stream. + await expect(collected).rejects.toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) expect(mockFetch).toHaveBeenCalledTimes(1) const fetchOptions = mockFetch.mock.calls[0]?.[1] as { signal?: AbortSignal } | undefined expect(fetchOptions?.signal).toBeInstanceOf(AbortSignal) expect(fetchOptions?.signal?.aborted).toBe(true) - expect(chunks).toHaveLength(0) }) it("should rethrow a native AbortError from the SDK as-is instead of normalizing it", async () => { @@ -1428,7 +1480,7 @@ describe("OpenAiNativeHandler", () => { // First request's body: one chunk now, then a pull that stays // pending until this request's external signal aborts, where it // enqueues an empty [DONE] chunk so the loop's next per-request - // abort check can break it. + // abort check is the only exit. const body = new ReadableStream({ start: (startController) => { startController.enqueue( @@ -1500,11 +1552,15 @@ describe("OpenAiNativeHandler", () => { expect(secondExternal.signal.aborted).toBe(false) expect(cancelCalls[1]).toBe(false) - // The first request reacts to its own abort: the loop breaks once the - // pending read settles, and its body is cancelled before the lock is - // released. No second POST is issued for the aborted request. - const chunks1 = await collected1 - expect(textChunks(chunks1).map((chunk) => chunk.text)).toEqual(["one"]) + // The first request reacts to its own abort: once the pending read + // settles, the loop-boundary check throws the contract AbortError (a + // cancelled request must not resolve with its partial output), and its + // body is cancelled before the lock is released. No second POST is + // issued for the aborted request. + await expect(collected1).rejects.toMatchObject({ + name: "AbortError", + message: "The OpenAI Native request was aborted", + }) expect(cancelCalls[0]).toBe(true) expect(fetchCall).toBe(2) }) diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index fc22ed97cf..d515b58f1a 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -831,8 +831,12 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // this.abortController field must not gate the loop: with overlapping // requests it reflects the most recently created request, so reading it // here would let one request's cancellation terminate another's stream. + // Throw rather than break: a cancellation that lands after a chunk was + // handed out and before the next pull must surface as the contract + // AbortError, not resolve with a partial stream (same rule the SDK + // path enforces with its own between-pulls checks). if (requestController.signal.aborted) { - break + throw createAbortError(this.providerName) } const { done, value } = await reader.read() From 45254961206fdb7e0ecbbe866f7b9bad1114d91b Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 18:22:33 +0800 Subject: [PATCH 24/27] fix(api): restore arrow glyphs in compat spec and type the native catch as unknown CodeRabbit follow-up on the main refresh: - openai-compatible.spec.ts: three comment arrows were mangled to '??' by an encoding roundtrip during the merge; restore '->'-style arrow glyph (the original '->' arrow character). - openai-native.ts: catch (sdkErr: unknown) instead of any; the body already handles unknown (isRequestAborted takes unknown and instanceof Error narrows before .name). The no-explicit-any suppression for the file returns to upstream's count of 27. --- src/api/providers/__tests__/openai-compatible.spec.ts | 6 +++--- src/api/providers/openai-native.ts | 2 +- src/eslint-suppressions.json | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/api/providers/__tests__/openai-compatible.spec.ts b/src/api/providers/__tests__/openai-compatible.spec.ts index 9d806c55ab..d065efead2 100644 --- a/src/api/providers/__tests__/openai-compatible.spec.ts +++ b/src/api/providers/__tests__/openai-compatible.spec.ts @@ -257,7 +257,7 @@ describe("OpenAICompatibleHandler", () => { }) // Test 1: createMessage() with a real AI SDK APICallError 429 (the AI SDK - // exposes the HTTP status as statusCode, not status) ??verify the wrapped error + // exposes the HTTP status as statusCode, not status) →verify the wrapped error // surfaces .status === 429 and the provider name in its message. it("should throw error with .status 429 when API returns 429", async () => { const rateLimitError = new APICallError({ @@ -292,7 +292,7 @@ describe("OpenAICompatibleHandler", () => { expect(thrownError.message).toContain("TestProvider") }) - // Test 2: createMessage() with mock 500 response ??verify error is properly tagged + // Test 2: createMessage() with mock 500 response →verify error is properly tagged it("should throw error with .status 500 and provider name when API returns 500", async () => { const serverError = Object.assign(new Error("Internal Server Error"), { status: 500 }) @@ -445,7 +445,7 @@ describe("OpenAICompatibleHandler", () => { ) }) - // Test 3: completePrompt() with mock 4xx/5xx ??verify error carries .status and provider name + // Test 3: completePrompt() with mock 4xx/5xx →verify error carries .status and provider name it("should throw error with .status and provider name when generateText throws 400", async () => { const badRequestError = Object.assign(new Error("Bad Request"), { status: 400 }) diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index 50c3eb2698..2bdc9fbd6d 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -516,7 +516,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio if (requestController.signal.aborted) { throw createAbortError(this.providerName) } - } catch (sdkErr: any) { + } catch (sdkErr: unknown) { // If this request's own controller aborted and the SDK error is the // cancellation itself — classified from the error alone via // isRequestAborted, never from the signal, so a terminal SDK error diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index b361584426..f1b2590f7f 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -396,7 +396,7 @@ }, "api/providers/openai-native.ts": { "@typescript-eslint/no-explicit-any": { - "count": 28 + "count": 27 } }, "api/providers/openai.ts": { From 6df4ff65fea1c30bdbbbd4594750a15c90e7f23c Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 20:26:29 +0800 Subject: [PATCH 25/27] fix(api): reject non-finite and cap overflowing mergeAbortSignalAndTimeout delays Port of the #1300 timeout guard finding (CodeRabbit): Node's AbortSignal.timeout throws ERR_OUT_OF_RANGE for non-integer delays (Infinity) and clamps delays above 2^31-1 ms to 1 ms, so a non-finite timeoutMs threw a RangeError instead of running the request, and an overflowing one aborted almost immediately. The helper now treats non-finite values as "no timeout" and caps finite delays at 2^31-1 ms. Regression tests added; negative controls (reverting the guard) fail exactly the two new tests. --- .../utils/__tests__/abort-signal.spec.ts | 23 +++++++++++++++++++ src/api/providers/utils/abort-signal.ts | 11 ++++++--- 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index f779d48e1f..65e511ed87 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -275,6 +275,29 @@ describe("abort-signal utilities", () => { await vi.waitFor(() => expect(result?.aborted).toBe(true)) }) + + it("treats a non-finite timeoutMs as no timeout", () => { + const controller = new AbortController() + + // Infinity must not reach AbortSignal.timeout(), which throws + // ERR_OUT_OF_RANGE for non-integer delays. + expect(() => mergeAbortSignalAndTimeout(undefined, Infinity)).not.toThrow() + expect(mergeAbortSignalAndTimeout(undefined, Infinity)).toBeUndefined() + expect(mergeAbortSignalAndTimeout(controller.signal, Infinity)).toBe(controller.signal) + }) + + it("caps an overflowing timeoutMs instead of aborting immediately", async () => { + // Node clamps AbortSignal.timeout delays above 2^31-1 ms to 1 ms; + // the helper caps the delay so an overflowing timeout does not abort + // the request within the first millisecond. + const result = mergeAbortSignalAndTimeout(undefined, 2_147_483_648) + + expect(result).toBeInstanceOf(AbortSignal) + expect(result?.aborted).toBe(false) + + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(result?.aborted).toBe(false) + }) }) describe("mergeAbortSignals", () => { diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index cbb7a13aaf..435af3b29e 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -1,18 +1,23 @@ /** * Merge an optional external abort signal with an optional timeout. * - * Timeout values <= 0 are treated as disabled. The timeout is created via the + * Timeout values <= 0 and non-finite values are treated as disabled; finite + * delays above 2^31-1 ms are capped (Node clamps them to 1 ms otherwise). The timeout is created via the * native AbortSignal.timeout() API, which self-manages its timer — callers do * not need to (and cannot) clear it manually. */ export function mergeAbortSignalAndTimeout(externalSignal?: AbortSignal, timeoutMs?: number): AbortSignal | undefined { - const hasTimeout = typeof timeoutMs === "number" && timeoutMs > 0 + // Node's AbortSignal.timeout() throws ERR_OUT_OF_RANGE for non-integer + // delays (Infinity) and clamps delays above 2^31-1 ms to 1 ms, which would + // abort almost immediately; treat non-finite values as "no timeout" and + // cap the rest. + const hasTimeout = typeof timeoutMs === "number" && Number.isFinite(timeoutMs) && timeoutMs > 0 if (!hasTimeout) { return externalSignal } - const timeoutSignal = AbortSignal.timeout(timeoutMs) + const timeoutSignal = AbortSignal.timeout(Math.min(timeoutMs, 2_147_483_647)) if (!externalSignal) { return timeoutSignal From 609667533bb739b9e073b895f6e4d3c2f8aa136a Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 21:43:38 +0800 Subject: [PATCH 26/27] fix(api): round fractional timeouts before AbortSignal.timeout CodeRabbit follow-up on the timeout guard: Node validates AbortSignal.timeout delays as uint32, so finite fractions (1500.5, 0.5) threw ERR_OUT_OF_RANGE just like Infinity did. The helper now rounds delays up (a positive fraction never becomes 0 ms) before capping at 2^31-1 ms. Regression test mergeAbortSignalAndTimeout (undefined, 1.5); negative control (reverting the rounding) fails exactly that test. --- .../providers/utils/__tests__/abort-signal.spec.ts | 14 ++++++++++++++ src/api/providers/utils/abort-signal.ts | 7 ++++--- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index 65e511ed87..c9fc051a9b 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -286,6 +286,20 @@ describe("abort-signal utilities", () => { expect(mergeAbortSignalAndTimeout(controller.signal, Infinity)).toBe(controller.signal) }) + it("rounds a fractional timeoutMs up before scheduling", async () => { + // Node validates AbortSignal.timeout delays as uint32: 1.5 or 0.5 + // would throw ERR_OUT_OF_RANGE. The helper rounds up, so a positive + // fraction never becomes 0 ms either. + let result: AbortSignal | undefined + expect(() => { + result = mergeAbortSignalAndTimeout(undefined, 1.5) + }).not.toThrow() + expect(result).toBeInstanceOf(AbortSignal) + expect(result?.aborted).toBe(false) + await vi.waitFor(() => expect(result?.aborted).toBe(true)) + expect(() => mergeAbortSignalAndTimeout(undefined, 0.5)).not.toThrow() + }) + it("caps an overflowing timeoutMs instead of aborting immediately", async () => { // Node clamps AbortSignal.timeout delays above 2^31-1 ms to 1 ms; // the helper caps the delay so an overflowing timeout does not abort diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index 435af3b29e..763a97c4c0 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -9,15 +9,16 @@ export function mergeAbortSignalAndTimeout(externalSignal?: AbortSignal, timeoutMs?: number): AbortSignal | undefined { // Node's AbortSignal.timeout() throws ERR_OUT_OF_RANGE for non-integer // delays (Infinity) and clamps delays above 2^31-1 ms to 1 ms, which would - // abort almost immediately; treat non-finite values as "no timeout" and - // cap the rest. + // abort almost immediately; treat non-finite values as "no timeout", + // round fractional delays up to whole milliseconds (uint32 validation + // throws for 1500.5), and cap the rest. const hasTimeout = typeof timeoutMs === "number" && Number.isFinite(timeoutMs) && timeoutMs > 0 if (!hasTimeout) { return externalSignal } - const timeoutSignal = AbortSignal.timeout(Math.min(timeoutMs, 2_147_483_647)) + const timeoutSignal = AbortSignal.timeout(Math.min(Math.max(1, Math.ceil(timeoutMs)), 2_147_483_647)) if (!externalSignal) { return timeoutSignal From 441ccf8509ae33f8a5a5ba26a8a42825e4d82284 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 23:21:51 +0800 Subject: [PATCH 27/27] test(api): assert the rounded delay reaching AbortSignal.timeout CodeRabbit: the fractional-timeout test only proved the call did not throw and eventually aborted - a round-down mutation (1.5 -> 1) would still pass. The test now spies AbortSignal.timeout and pins the received delay: 2 for 1.5 and 1 for 0.5 (never 0). Negative control: mutating Math.ceil to Math.floor fails exactly this test. --- .../utils/__tests__/abort-signal.spec.ts | 25 ++++++++++++++----- 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index c9fc051a9b..3cd64e8fd1 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -288,16 +288,29 @@ describe("abort-signal utilities", () => { it("rounds a fractional timeoutMs up before scheduling", async () => { // Node validates AbortSignal.timeout delays as uint32: 1.5 or 0.5 - // would throw ERR_OUT_OF_RANGE. The helper rounds up, so a positive - // fraction never becomes 0 ms either. + // would throw ERR_OUT_OF_RANGE. Pin the rounded delay that actually + // reaches AbortSignal.timeout (1.5 -> 2, 0.5 -> 1, never 0) so a + // round-down mutation reddens this test instead of still aborting + // eventually. + const timeoutSpy = vi.spyOn(AbortSignal, "timeout") let result: AbortSignal | undefined - expect(() => { - result = mergeAbortSignalAndTimeout(undefined, 1.5) - }).not.toThrow() + let tiny: AbortSignal | undefined + try { + expect(() => { + result = mergeAbortSignalAndTimeout(undefined, 1.5) + }).not.toThrow() + expect(timeoutSpy).toHaveBeenCalledWith(2) + expect(() => { + tiny = mergeAbortSignalAndTimeout(undefined, 0.5) + }).not.toThrow() + expect(timeoutSpy).toHaveBeenCalledWith(1) + } finally { + timeoutSpy.mockRestore() + } expect(result).toBeInstanceOf(AbortSignal) expect(result?.aborted).toBe(false) + expect(tiny).toBeInstanceOf(AbortSignal) await vi.waitFor(() => expect(result?.aborted).toBe(true)) - expect(() => mergeAbortSignalAndTimeout(undefined, 0.5)).not.toThrow() }) it("caps an overflowing timeoutMs instead of aborting immediately", async () => {