From 3b899cecd21ef17505e9f57f4f1950c20f68f8ce Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 19 Aug 2026 22:43:06 +0800 Subject: [PATCH 01/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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()