From 5994929110b9026685633d91153de320853d2050 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 08:58:27 +0800 Subject: [PATCH 1/3] feat(api): abort signal support for openrouter, requesty, poe (completePrompt + createMessage) --- .../providers/__tests__/openrouter.spec.ts | 155 ++++- src/api/providers/__tests__/poe.spec.ts | 193 +++++ src/api/providers/__tests__/requesty.spec.ts | 189 ++++- src/api/providers/openrouter.ts | 657 ++++++++++-------- src/api/providers/poe.ts | 220 ++++-- src/api/providers/requesty.ts | 199 ++++-- 6 files changed, 1156 insertions(+), 457 deletions(-) diff --git a/src/api/providers/__tests__/openrouter.spec.ts b/src/api/providers/__tests__/openrouter.spec.ts index 1e422e4ba8..a2130be039 100644 --- a/src/api/providers/__tests__/openrouter.spec.ts +++ b/src/api/providers/__tests__/openrouter.spec.ts @@ -21,7 +21,7 @@ import { providerIdentifiers } from "@roo-code/types" import { OpenRouterHandler } from "../openrouter" import { Package } from "../../../shared/package" -import { makeApiHandlerOptions } from "../../../test-utils/api" +import { makeApiHandlerOptions, makeCreateMessageMetadata } from "../../../test-utils/api" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { clearAllMocks } from "../../../test-utils/reset" @@ -289,7 +289,10 @@ describe("OpenRouterHandler", () => { temperature: 0, top_p: undefined, }), - { headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" } }, + { + headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" }, + signal: expect.any(AbortSignal), + }, ) }) @@ -332,7 +335,10 @@ describe("OpenRouterHandler", () => { }), ]), }), - { headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" } }, + { + headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" }, + signal: expect.any(AbortSignal), + }, ) }) @@ -539,6 +545,69 @@ describe("OpenRouterHandler", () => { expect(endChunks).toHaveLength(1) expect(endChunks[0].id).toBe("call_openrouter_test") }) + it("rejects with AbortError when the external signal is pre-aborted", async () => { + const handler = new OpenRouterHandler(mockOptions) + const mockCreate = vitest.fn().mockResolvedValue({ choices: [{ message: { content: "response" } }] }) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const controller = new AbortController() + controller.abort() + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + + await expect( + handler.createMessage("test", [{ role: "user" as const, content: "hi" }], metadata).next(), + ).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("aborts the in-flight stream and rejects with AbortError when the external signal aborts", async () => { + const handler = new OpenRouterHandler(mockOptions) + const controller = new AbortController() + + let requestSignal: AbortSignal | undefined + const mockCreate = vitest + .fn() + .mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + requestSignal = options?.signal + // Emulate the OpenAI SDK: the first chunk arrives, then the in-flight + // response body rejects once the request signal aborts. + return (async function* () { + yield { id: "1", choices: [{ delta: { content: "first" } }] } + await new Promise((resolve) => { + if (requestSignal?.aborted) { + resolve() + } else { + requestSignal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The user aborted a request") + abortError.name = "AbortError" + throw abortError + })() + }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("test", [{ role: "user" as const, content: "hi" }], metadata) + + const chunks: unknown[] = [] + const iteration = (async () => { + for await (const chunk of generator) { + chunks.push(chunk) + if (chunk.type === "text") { + // Abort while the stream is still in flight. + controller.abort() + } + } + })() + + await expect(iteration).rejects.toMatchObject({ name: "AbortError" }) + expect(chunks).toContainEqual({ type: "text", text: "first" }) + }) }) describe("completePrompt", () => { @@ -711,5 +780,85 @@ describe("OpenRouterHandler", () => { }), ) }) + it("should pass abort signal through to client", async () => { + const handler = new OpenRouterHandler(mockOptions) + const controller = new AbortController() + const mockCreate = vitest.fn().mockResolvedValue({ choices: [{ message: { content: "response" } }] }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ signal: controller.signal }), + ) + }) + + it("should pass timeout through to client", async () => { + const handler = new OpenRouterHandler(mockOptions) + const mockCreate = vitest.fn().mockResolvedValue({ choices: [{ message: { content: "response" } }] }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + await handler.completePrompt("test prompt", { timeoutMs: 5000 }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 5000 }), + ) + }) + + it("should work without options (backward compatible)", async () => { + const handler = new OpenRouterHandler(mockOptions) + const mockCreate = vitest.fn().mockResolvedValue({ choices: [{ message: { content: "response" } }] }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + }) + + it("rejects with AbortError when the signal is pre-aborted", async () => { + const handler = new OpenRouterHandler(mockOptions) + const mockCreate = vitest.fn().mockResolvedValue({ choices: [{ message: { content: "response" } }] }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const controller = new AbortController() + controller.abort() + + await expect( + handler.completePrompt("test prompt", { abortSignal: controller.signal }), + ).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("rejects with AbortError when aborted mid-flight", async () => { + const handler = new OpenRouterHandler(mockOptions) + const controller = new AbortController() + + const mockCreate = vitest + .fn() + .mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + // Emulate the OpenAI SDK: the in-flight request rejects when the signal aborts. + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The user aborted a request") + abortError.name = "AbortError" + throw abortError + }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal }) + controller.abort() + + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + }) }) }) diff --git a/src/api/providers/__tests__/poe.spec.ts b/src/api/providers/__tests__/poe.spec.ts index 00712924f5..f636a13bef 100644 --- a/src/api/providers/__tests__/poe.spec.ts +++ b/src/api/providers/__tests__/poe.spec.ts @@ -3,6 +3,7 @@ import { poeDefaultModelId, providerIdentifiers } from "@roo-code/types" import { PoeHandler } from "../poe" import { getModelsFromCache } from "../fetchers/modelCache" +import { makeCreateMessageMetadata } from "../../../test-utils/api" import { clearAllMocks } from "../../../test-utils/reset" const { mockStreamText, mockGenerateText, mockCreatePoe, mockGetModelsFromCache, mockCaptureException } = @@ -237,6 +238,70 @@ describe("PoeHandler", () => { }), ) }) + + it("rejects with AbortError when the external signal is pre-aborted", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + mockStreamText.mockReturnValue({ + fullStream: (async function* () {})(), + usage: Promise.resolve(undefined), + }) + + const controller = new AbortController() + controller.abort() + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + + await expect( + handler.createMessage("system", [{ role: "user" as const, content: "hi" }], metadata).next(), + ).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("aborts the in-flight stream and rejects with AbortError when the external signal aborts", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + + let requestSignal: AbortSignal | undefined + mockStreamText.mockImplementationOnce((args: { abortSignal?: AbortSignal }) => { + requestSignal = args.abortSignal + // Emulate the AI SDK: the first chunk arrives, then the stream errors once + // the abort signal fires. + const fullStream = (async function* () { + yield { type: "text-delta", text: "Hello " } + await new Promise((resolve) => { + if (requestSignal?.aborted) { + resolve() + } else { + requestSignal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The operation was aborted") + abortError.name = "AbortError" + throw abortError + })() + return { fullStream, usage: Promise.resolve(undefined) } + }) + + const controller = new AbortController() + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + + const chunks: unknown[] = [] + const iteration = (async () => { + for await (const chunk of handler.createMessage( + "system", + [{ role: "user" as const, content: "hi" }], + metadata, + )) { + chunks.push(chunk) + if (chunk.type === "text") { + // Abort while the stream is still in flight. + controller.abort() + } + } + })() + + await expect(iteration).rejects.toMatchObject({ name: "AbortError" }) + expect(chunks).toContainEqual({ type: "text", text: "Hello " }) + }) }) describe("reasoning", () => { @@ -398,5 +463,133 @@ describe("PoeHandler", () => { }), ) }) + + it("completePrompt should pass abort signal through to generateText", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + const controller = new AbortController() + mockGenerateText.mockResolvedValueOnce({ text: "response" }) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockGenerateText).toHaveBeenCalledWith( + expect.objectContaining({ + model: mockLanguageModel, + prompt: "test prompt", + abortSignal: controller.signal, + }), + ) + }) + + it("completePrompt should work without options (backward compatible)", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + mockGenerateText.mockResolvedValueOnce({ text: "response" }) + + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + expect(mockGenerateText).toHaveBeenCalledWith( + expect.objectContaining({ + model: mockLanguageModel, + prompt: "test prompt", + }), + ) + }) + + it("completePrompt should merge signal and timeoutMs into combined abortSignal", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + const controller = new AbortController() + mockGenerateText.mockResolvedValueOnce({ text: "response" }) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) + expect(mockGenerateText).toHaveBeenCalledWith( + expect.objectContaining({ + model: mockLanguageModel, + prompt: "test prompt", + abortSignal: expect.any(AbortSignal), + }), + ) + // The abortSignal should be a merged signal (not the original controller.signal) + const callArgs = mockGenerateText.mock.calls[0][0] + expect(callArgs.abortSignal).toBeDefined() + expect(callArgs.abortSignal).toBeInstanceOf(AbortSignal) + }) + + it("completePrompt should use AbortSignal.timeout when only timeoutMs is provided", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + mockGenerateText.mockResolvedValueOnce({ text: "response" }) + + await handler.completePrompt("test prompt", { timeoutMs: 3000 }) + expect(mockGenerateText).toHaveBeenCalledWith( + expect.objectContaining({ + model: mockLanguageModel, + prompt: "test prompt", + abortSignal: expect.any(AbortSignal), + }), + ) + const callArgs = mockGenerateText.mock.calls[0][0] + expect(callArgs.abortSignal).toBeDefined() + expect(callArgs.abortSignal).toBeInstanceOf(AbortSignal) + }) + + it("completePrompt should prefer signal over timeoutMs when both are provided", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + const controller = new AbortController() + mockGenerateText.mockResolvedValueOnce({ text: "response" }) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) + const callArgs = mockGenerateText.mock.calls[0][0] + // Should have a merged abortSignal (not the original controller.signal) + expect(callArgs.abortSignal).toBeInstanceOf(AbortSignal) + expect(callArgs.abortSignal).not.toBe(controller.signal) + }) + + it("completePrompt rejects with AbortError when the external signal aborts mid-flight", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + const controller = new AbortController() + // Emulate the AI SDK: the in-flight generation rejects when the abort signal fires. + mockGenerateText.mockImplementationOnce(async (args: { abortSignal?: AbortSignal }) => { + await new Promise((resolve) => { + if (args.abortSignal?.aborted) { + resolve() + } else { + args.abortSignal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The operation was aborted") + abortError.name = "AbortError" + throw abortError + }) + + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) + const callArgs = mockGenerateText.mock.calls[0][0] + expect(callArgs.abortSignal).toBeDefined() + + // Abort the external signal before the generation settles. + controller.abort() + + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + // The merged signal should be aborted once the user signal aborts. + expect(callArgs.abortSignal.aborted).toBe(true) + }) + + it("completePrompt should handle timeoutMs=0 as no timeout", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + mockGenerateText.mockResolvedValueOnce({ text: "response" }) + + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + expect(mockGenerateText).toHaveBeenCalledWith( + expect.objectContaining({ + model: mockLanguageModel, + prompt: "test prompt", + }), + ) + const callArgs = mockGenerateText.mock.calls[0][0] + expect(callArgs.abortSignal).toBeUndefined() + }) + + it("completePrompt should handle non-Error values in catch block", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + mockGenerateText.mockRejectedValueOnce("not an error") + + await expect(handler.completePrompt("test prompt")).rejects.toThrow() + }) }) }) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index c685da0ed2..325cfaf6cd 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -12,7 +12,7 @@ import OpenAI from "openai" import { RequestyHandler } from "../requesty" import { Package } from "../../../shared/package" import { ApiHandlerCreateMessageMetadata } from "../../index" -import { makeApiHandlerOptions } from "../../../test-utils/api" +import { makeApiHandlerOptions, makeCreateMessageMetadata } from "../../../test-utils/api" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { clearAllMocks } from "../../../test-utils/reset" @@ -241,6 +241,7 @@ describe("RequestyHandler", () => { stream_options: { include_usage: true }, temperature: 0, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -274,6 +275,7 @@ describe("RequestyHandler", () => { thinking: { type: "adaptive" }, temperature: undefined, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -307,6 +309,7 @@ describe("RequestyHandler", () => { thinking: { type: "adaptive" }, temperature: undefined, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -340,6 +343,7 @@ describe("RequestyHandler", () => { thinking: { type: "adaptive" }, temperature: undefined, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -478,6 +482,7 @@ describe("RequestyHandler", () => { ]), tool_choice: "auto", }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -559,6 +564,62 @@ describe("RequestyHandler", () => { }) }) }) + it("rejects with AbortError when the external signal is pre-aborted", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockResolvedValue(asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "response" } }] }])) + + const controller = new AbortController() + controller.abort() + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + + await expect( + handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata).next(), + ).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("aborts the in-flight stream and rejects with AbortError when the external signal aborts", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + let requestSignal: AbortSignal | undefined + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + requestSignal = options?.signal + // Emulate the OpenAI SDK: the first chunk arrives, then the in-flight + // response body rejects once the request signal aborts. + return (async function* () { + yield { id: "1", choices: [{ delta: { content: "first" } }] } + await new Promise((resolve) => { + if (requestSignal?.aborted) { + resolve() + } else { + requestSignal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The user aborted a request") + abortError.name = "AbortError" + throw abortError + })() + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + const chunks: unknown[] = [] + const iteration = (async () => { + for await (const chunk of generator) { + chunks.push(chunk) + if (chunk.type === "text") { + // Abort while the stream is still in flight. + controller.abort() + } + } + })() + + await expect(iteration).rejects.toMatchObject({ name: "AbortError" }) + expect(chunks).toContainEqual({ type: "text", text: "first" }) + }) }) describe("completePrompt", () => { @@ -572,12 +633,15 @@ describe("RequestyHandler", () => { expect(result).toBe("test completion") - expect(mockCreate).toHaveBeenCalledWith({ - model: mockOptions.requestyModelId, - max_tokens: 8192, - messages: [{ role: "system", content: "test prompt" }], - temperature: 0, - }) + expect(mockCreate).toHaveBeenCalledWith( + { + model: mockOptions.requestyModelId, + max_tokens: 8192, + messages: [{ role: "system", content: "test prompt" }], + temperature: 0, + }, + {}, + ) }) it("omits temperature for Claude Fable 5 in completePrompt", async () => { @@ -591,12 +655,15 @@ describe("RequestyHandler", () => { await handler.completePrompt("test prompt") - expect(mockCreate).toHaveBeenCalledWith({ - model: "anthropic/claude-fable-5", - max_tokens: 8192, - messages: [{ role: "system", content: "test prompt" }], - temperature: undefined, - }) + expect(mockCreate).toHaveBeenCalledWith( + { + model: "anthropic/claude-fable-5", + max_tokens: 8192, + messages: [{ role: "system", content: "test prompt" }], + temperature: undefined, + }, + {}, + ) }) it("omits temperature for Claude Sonnet 5 in completePrompt", async () => { @@ -610,12 +677,15 @@ describe("RequestyHandler", () => { await handler.completePrompt("test prompt") - expect(mockCreate).toHaveBeenCalledWith({ - model: "anthropic/claude-sonnet-5", - max_tokens: 8192, - messages: [{ role: "system", content: "test prompt" }], - temperature: undefined, - }) + expect(mockCreate).toHaveBeenCalledWith( + { + model: "anthropic/claude-sonnet-5", + max_tokens: 8192, + messages: [{ role: "system", content: "test prompt" }], + temperature: undefined, + }, + {}, + ) }) it("omits temperature for Claude Opus 5 in completePrompt", async () => { @@ -629,12 +699,15 @@ describe("RequestyHandler", () => { await handler.completePrompt("test prompt") - expect(mockCreate).toHaveBeenCalledWith({ - model: "anthropic/claude-opus-5", - max_tokens: 8192, - messages: [{ role: "system", content: "test prompt" }], - temperature: undefined, - }) + expect(mockCreate).toHaveBeenCalledWith( + { + model: "anthropic/claude-opus-5", + max_tokens: 8192, + messages: [{ role: "system", content: "test prompt" }], + temperature: undefined, + }, + {}, + ) }) it("handles API errors", async () => { @@ -651,5 +724,71 @@ describe("RequestyHandler", () => { await expect(handler.completePrompt("test prompt")).rejects.toThrow("Unexpected error") }) + it("should pass abort signal through to client", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + signal: controller.signal, + }) + }) + + it("should pass timeout through to client", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) + + await handler.completePrompt("test prompt", { timeoutMs: 5000 }) + expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + timeout: 5000, + }) + }) + + it("should work without options (backward compatible)", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) + + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + }) + + it("rejects with AbortError when the signal is pre-aborted", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) + + const controller = new AbortController() + controller.abort() + + await expect( + handler.completePrompt("test prompt", { abortSignal: controller.signal }), + ).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("rejects with AbortError when aborted mid-flight", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + // Emulate the OpenAI SDK: the in-flight request rejects when the signal aborts. + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The user aborted a request") + abortError.name = "AbortError" + throw abortError + }) + + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal }) + controller.abort() + + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + }) }) }) diff --git a/src/api/providers/openrouter.ts b/src/api/providers/openrouter.ts index f61e007214..2ed7094761 100644 --- a/src/api/providers/openrouter.ts +++ b/src/api/providers/openrouter.ts @@ -139,6 +139,16 @@ interface CompletionUsage { } } +/** + * Create a DOM-standard AbortError so callers can detect aborted requests + * (matches the error name produced by native abort-based APIs). + */ +function createAbortError(message: string): Error { + const error = new Error(message) + error.name = "AbortError" + return error +} + export class OpenRouterHandler extends BaseProvider implements SingleCompletionHandler { protected options: ApiHandlerOptions private client: OpenAI @@ -212,330 +222,381 @@ export class OpenRouterHandler extends BaseProvider implements SingleCompletionH messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): AsyncGenerator { - const model = await this.fetchModel() - - let { id: modelId, maxTokens, temperature, topP, reasoning } = model - - // Reset reasoning_details accumulator for this request - this.currentReasoningDetails = [] - - // OpenRouter sends reasoning tokens by default for Gemini 2.5 Pro models - // even if you don't request them. This is not the default for - // other providers (including Gemini), so we need to explicitly disable - // them unless the user has explicitly configured reasoning. - // Note: Gemini 3 models use reasoning_details format with thought signatures, - // but we handle this via skip_thought_signature_validator injection below. - if ( - (modelId === "google/gemini-2.5-pro-preview" || modelId === "google/gemini-2.5-pro") && - typeof reasoning === "undefined" - ) { - reasoning = { exclude: true } + // Per-request AbortController: external aborts cancel the in-flight request + // without replacing the client-level timeout, which remains the default safety net. + const controller = new AbortController() + + // Bridge the external abort signal into the per-request controller: + // - pre-aborted guard: abort immediately when the signal is already aborted + // - { once: true }: the listener removes itself after the first abort + // - explicit removal in finally: the listener must not outlive a request that + // completes (or fails) without being aborted + const externalAbortSignal = metadata?.abortSignal + let removeExternalAbortListener: (() => void) | undefined + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + controller.abort() + } else { + const onExternalAbort = () => controller.abort() + externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) + removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) + } } - // Convert Anthropic messages to OpenAI format. - // Pass normalization function for Mistral compatibility (requires 9-char alphanumeric IDs) - const isMistral = modelId.toLowerCase().includes("mistral") - let openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ - { role: "system", content: systemPrompt }, - ...convertToOpenAiMessages( - messages, - isMistral ? { normalizeToolCallId: normalizeMistralToolCallId } : undefined, - ), - ] - - // DeepSeek highly recommends using user instead of system role. - if (modelId.startsWith("deepseek/deepseek-r1") || modelId === "perplexity/sonar-reasoning") { - openAiMessages = convertToR1Format([{ role: "user", content: systemPrompt }, ...messages]) - } + try { + // The request was already aborted before we started: fail fast without calling the API. + if (controller.signal.aborted) { + throw createAbortError("OpenRouter request aborted") + } - // Process reasoning_details when switching models to Gemini. - const isGemini = modelId.startsWith("google/gemini") - - // For Gemini models with native protocol: - // 1. Sanitize messages to handle thought signature validation issues. - // This must happen BEFORE fake encrypted block injection to avoid injecting for - // tool calls that will be dropped due to missing/mismatched reasoning_details. - // 2. Inject fake reasoning.encrypted block for tool calls without existing encrypted reasoning. - // This is required when switching from other models to Gemini to satisfy API validation. - // Per OpenRouter documentation (conversation with Toven, Nov 2025): - // - Create ONE reasoning_details entry per assistant message with tool calls - // - Set `id` to the FIRST tool call's ID from the tool_calls array - // - Set `data` to "skip_thought_signature_validator" to bypass signature validation - // - Set `index` to 0 - // See: https://github.com/cline/cline/issues/8214 - if (isGemini) { - // Step 1: Sanitize messages - filter out tool calls with missing/mismatched reasoning_details - openAiMessages = sanitizeGeminiMessages(openAiMessages, modelId) - - // Step 2: Inject fake reasoning.encrypted block for tool calls that survived sanitization - openAiMessages = openAiMessages.map((msg) => { - if (msg.role === "assistant") { - const toolCalls = (msg as any).tool_calls as any[] | undefined - const existingDetails = (msg as any).reasoning_details as any[] | undefined - - // Only inject if there are tool calls and no existing encrypted reasoning - if (toolCalls && toolCalls.length > 0) { - const hasEncrypted = existingDetails?.some((d) => d.type === "reasoning.encrypted") ?? false - - if (!hasEncrypted) { - // Create ONE fake encrypted block with the FIRST tool call's ID - // This is the documented format from OpenRouter for skipping thought signature validation - const fakeEncrypted = { - type: "reasoning.encrypted", - data: "skip_thought_signature_validator", - id: toolCalls[0].id, - format: "google-gemini-v1", - index: 0, - } + const model = await this.fetchModel() + + let { id: modelId, maxTokens, temperature, topP, reasoning } = model + + // Reset reasoning_details accumulator for this request + this.currentReasoningDetails = [] + + // OpenRouter sends reasoning tokens by default for Gemini 2.5 Pro models + // even if you don't request them. This is not the default for + // other providers (including Gemini), so we need to explicitly disable + // them unless the user has explicitly configured reasoning. + // Note: Gemini 3 models use reasoning_details format with thought signatures, + // but we handle this via skip_thought_signature_validator injection below. + if ( + (modelId === "google/gemini-2.5-pro-preview" || modelId === "google/gemini-2.5-pro") && + typeof reasoning === "undefined" + ) { + reasoning = { exclude: true } + } - return { - ...msg, - reasoning_details: [...(existingDetails ?? []), fakeEncrypted], + // Convert Anthropic messages to OpenAI format. + // Pass normalization function for Mistral compatibility (requires 9-char alphanumeric IDs) + const isMistral = modelId.toLowerCase().includes("mistral") + let openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ + { role: "system", content: systemPrompt }, + ...convertToOpenAiMessages( + messages, + isMistral ? { normalizeToolCallId: normalizeMistralToolCallId } : undefined, + ), + ] + + // DeepSeek highly recommends using user instead of system role. + if (modelId.startsWith("deepseek/deepseek-r1") || modelId === "perplexity/sonar-reasoning") { + openAiMessages = convertToR1Format([{ role: "user", content: systemPrompt }, ...messages]) + } + + // Process reasoning_details when switching models to Gemini. + const isGemini = modelId.startsWith("google/gemini") + + // For Gemini models with native protocol: + // 1. Sanitize messages to handle thought signature validation issues. + // This must happen BEFORE fake encrypted block injection to avoid injecting for + // tool calls that will be dropped due to missing/mismatched reasoning_details. + // 2. Inject fake reasoning.encrypted block for tool calls without existing encrypted reasoning. + // This is required when switching from other models to Gemini to satisfy API validation. + // Per OpenRouter documentation (conversation with Toven, Nov 2025): + // - Create ONE reasoning_details entry per assistant message with tool calls + // - Set `id` to the FIRST tool call's ID from the tool_calls array + // - Set `data` to "skip_thought_signature_validator" to bypass signature validation + // - Set `index` to 0 + // See: https://github.com/cline/cline/issues/8214 + if (isGemini) { + // Step 1: Sanitize messages - filter out tool calls with missing/mismatched reasoning_details + openAiMessages = sanitizeGeminiMessages(openAiMessages, modelId) + + // Step 2: Inject fake reasoning.encrypted block for tool calls that survived sanitization + openAiMessages = openAiMessages.map((msg) => { + if (msg.role === "assistant") { + const toolCalls = (msg as any).tool_calls as any[] | undefined + const existingDetails = (msg as any).reasoning_details as any[] | undefined + + // Only inject if there are tool calls and no existing encrypted reasoning + if (toolCalls && toolCalls.length > 0) { + const hasEncrypted = existingDetails?.some((d) => d.type === "reasoning.encrypted") ?? false + + if (!hasEncrypted) { + // Create ONE fake encrypted block with the FIRST tool call's ID + // This is the documented format from OpenRouter for skipping thought signature validation + const fakeEncrypted = { + type: "reasoning.encrypted", + data: "skip_thought_signature_validator", + id: toolCalls[0].id, + format: "google-gemini-v1", + index: 0, + } + + return { + ...msg, + reasoning_details: [...(existingDetails ?? []), fakeEncrypted], + } } } } - } - return msg - }) - } - - // https://openrouter.ai/docs/features/prompt-caching - // TODO: Add a `promptCacheStratey` field to `ModelInfo`. - if (OPEN_ROUTER_PROMPT_CACHING_MODELS.has(modelId)) { - if (modelId.startsWith("google")) { - addGeminiCacheBreakpoints(systemPrompt, openAiMessages) - } else { - addAnthropicCacheBreakpoints(systemPrompt, openAiMessages) + return msg + }) } - } - - // https://openrouter.ai/docs/transforms - const completionParams: OpenRouterChatCompletionParams = { - model: modelId, - ...(maxTokens && maxTokens > 0 && { max_tokens: maxTokens }), - temperature, - top_p: topP, - messages: openAiMessages, - stream: true, - stream_options: { include_usage: true }, - // Only include provider if openRouterSpecificProvider is not "[default]". - ...(this.options.openRouterSpecificProvider && - this.options.openRouterSpecificProvider !== OPENROUTER_DEFAULT_PROVIDER_NAME && { - provider: { - order: [this.options.openRouterSpecificProvider], - only: [this.options.openRouterSpecificProvider], - allow_fallbacks: false, - }, - }), - ...(reasoning && { reasoning }), - tools: this.convertToolsForOpenAI(metadata?.tools), - tool_choice: metadata?.tool_choice, - } - // Add Anthropic beta header for fine-grained tool streaming when using Anthropic models - const requestOptions = modelId.startsWith("anthropic/") - ? { headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" } } - : undefined + // https://openrouter.ai/docs/features/prompt-caching + // TODO: Add a `promptCacheStratey` field to `ModelInfo`. + if (OPEN_ROUTER_PROMPT_CACHING_MODELS.has(modelId)) { + if (modelId.startsWith("google")) { + addGeminiCacheBreakpoints(systemPrompt, openAiMessages) + } else { + addAnthropicCacheBreakpoints(systemPrompt, openAiMessages) + } + } - let stream - try { - stream = await this.client.chat.completions.create(completionParams, requestOptions) - } catch (error) { - // Try to parse as OpenRouter error structure using Zod - const parseResult = OpenRouterErrorResponseSchema.safeParse(error) + // https://openrouter.ai/docs/transforms + const completionParams: OpenRouterChatCompletionParams = { + model: modelId, + ...(maxTokens && maxTokens > 0 && { max_tokens: maxTokens }), + temperature, + top_p: topP, + messages: openAiMessages, + stream: true, + stream_options: { include_usage: true }, + // Only include provider if openRouterSpecificProvider is not "[default]". + ...(this.options.openRouterSpecificProvider && + this.options.openRouterSpecificProvider !== OPENROUTER_DEFAULT_PROVIDER_NAME && { + provider: { + order: [this.options.openRouterSpecificProvider], + only: [this.options.openRouterSpecificProvider], + allow_fallbacks: false, + }, + }), + ...(reasoning && { reasoning }), + tools: this.convertToolsForOpenAI(metadata?.tools), + tool_choice: metadata?.tool_choice, + } - if (parseResult.success && parseResult.data.error) { - const openRouterError = parseResult.data - const rawString = openRouterError.error?.metadata?.raw - const parsedError = extractErrorFromMetadataRaw(rawString) - const rawErrorMessage = parsedError || openRouterError.error?.message || "Unknown error" + // Add Anthropic beta header for fine-grained tool streaming when using Anthropic models + // and pass the per-request signal so external aborts cancel the in-flight stream. + const requestOptions: OpenAI.RequestOptions = { + ...(modelId.startsWith("anthropic/") + ? { headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" } } + : undefined), + signal: controller.signal, + } - const apiError = Object.assign( - new ApiProviderError( - rawErrorMessage, + let stream + try { + stream = await this.client.chat.completions.create(completionParams, requestOptions) + } catch (error) { + // Aborted requests are user-initiated: surface them as AbortError instead of + // a completion error (and keep them out of exception telemetry). + if (controller.signal.aborted) { + throw createAbortError("OpenRouter request aborted") + } + // Try to parse as OpenRouter error structure using Zod + const parseResult = OpenRouterErrorResponseSchema.safeParse(error) + + if (parseResult.success && parseResult.data.error) { + const openRouterError = parseResult.data + const rawString = openRouterError.error?.metadata?.raw + const parsedError = extractErrorFromMetadataRaw(rawString) + const rawErrorMessage = parsedError || openRouterError.error?.message || "Unknown error" + + const apiError = Object.assign( + new ApiProviderError( + rawErrorMessage, + providerIdentifiers.openrouter, + modelId, + "createMessage", + openRouterError.error?.code, + ), + { + status: openRouterError.error?.code, + error: openRouterError.error, + }, + ) + + TelemetryService.instance.captureException(apiError) + throw handleOpenAIError(error, this.providerName) + } else { + // Fallback for non-OpenRouter errors + const errorMessage = error instanceof Error ? error.message : String(error) + const apiError = new ApiProviderError( + errorMessage, providerIdentifiers.openrouter, modelId, "createMessage", - openRouterError.error?.code, - ), - { - status: openRouterError.error?.code, - error: openRouterError.error, - }, - ) - - TelemetryService.instance.captureException(apiError) - throw handleOpenAIError(error, this.providerName) - } else { - // Fallback for non-OpenRouter errors - const errorMessage = error instanceof Error ? error.message : String(error) - const apiError = new ApiProviderError( - errorMessage, - providerIdentifiers.openrouter, - modelId, - "createMessage", - ) - TelemetryService.instance.captureException(apiError) - throw handleOpenAIError(error, this.providerName) - } - } - - let lastUsage: CompletionUsage | undefined = undefined - // Accumulator for reasoning_details FROM the API. - // We preserve the original shape of reasoning_details to prevent malformed responses. - const reasoningDetailsAccumulator = new Map< - string, - { - type: string - text?: string - summary?: string - data?: string - id?: string | null - format?: string - signature?: string - index: number - } - >() - - // Track whether we've yielded displayable text from reasoning_details. - // When reasoning_details has displayable content (reasoning.text or reasoning.summary), - // we skip yielding the top-level reasoning field to avoid duplicate display. - let hasYieldedReasoningFromDetails = false - - for await (const chunk of stream) { - // OpenRouter returns an error object instead of the OpenAI SDK throwing an error. - if ("error" in chunk) { - this.handleStreamingError(chunk.error as OpenRouterError, modelId, "createMessage") + ) + TelemetryService.instance.captureException(apiError) + throw handleOpenAIError(error, this.providerName) + } } - const delta = chunk.choices[0]?.delta - const finishReason = chunk.choices[0]?.finish_reason - - if (delta) { - // Handle reasoning_details array format (used by Gemini 3, Claude, OpenAI o-series, etc.) - // See: https://openrouter.ai/docs/use-cases/reasoning-tokens#preserving-reasoning-blocks - // Priority: Check for reasoning_details first, as it's the newer format - const deltaWithReasoning = delta as typeof delta & { - reasoning_details?: Array<{ - type: string - text?: string - summary?: string - data?: string - id?: string | null - format?: string - signature?: string - index?: number - }> + let lastUsage: CompletionUsage | undefined = undefined + // Accumulator for reasoning_details FROM the API. + // We preserve the original shape of reasoning_details to prevent malformed responses. + const reasoningDetailsAccumulator = new Map< + string, + { + type: string + text?: string + summary?: string + data?: string + id?: string | null + format?: string + signature?: string + index: number } + >() + + // Track whether we've yielded displayable text from reasoning_details. + // When reasoning_details has displayable content (reasoning.text or reasoning.summary), + // we skip yielding the top-level reasoning field to avoid duplicate display. + let hasYieldedReasoningFromDetails = false + + try { + for await (const chunk of stream) { + // OpenRouter returns an error object instead of the OpenAI SDK throwing an error. + if ("error" in chunk) { + this.handleStreamingError(chunk.error as OpenRouterError, modelId, "createMessage") + } - if (deltaWithReasoning.reasoning_details && Array.isArray(deltaWithReasoning.reasoning_details)) { - for (const detail of deltaWithReasoning.reasoning_details) { - const index = detail.index ?? 0 - const key = `${detail.type}-${index}` - const existing = reasoningDetailsAccumulator.get(key) + const delta = chunk.choices[0]?.delta + const finishReason = chunk.choices[0]?.finish_reason + + if (delta) { + // Handle reasoning_details array format (used by Gemini 3, Claude, OpenAI o-series, etc.) + // See: https://openrouter.ai/docs/use-cases/reasoning-tokens#preserving-reasoning-blocks + // Priority: Check for reasoning_details first, as it's the newer format + const deltaWithReasoning = delta as typeof delta & { + reasoning_details?: Array<{ + type: string + text?: string + summary?: string + data?: string + id?: string | null + format?: string + signature?: string + index?: number + }> + } - if (existing) { - // Accumulate text/summary/data for existing reasoning detail - if (detail.text !== undefined) { - existing.text = (existing.text || "") + detail.text + if ( + deltaWithReasoning.reasoning_details && + Array.isArray(deltaWithReasoning.reasoning_details) + ) { + for (const detail of deltaWithReasoning.reasoning_details) { + const index = detail.index ?? 0 + const key = `${detail.type}-${index}` + const existing = reasoningDetailsAccumulator.get(key) + + if (existing) { + // Accumulate text/summary/data for existing reasoning detail + if (detail.text !== undefined) { + existing.text = (existing.text || "") + detail.text + } + if (detail.summary !== undefined) { + existing.summary = (existing.summary || "") + detail.summary + } + if (detail.data !== undefined) { + existing.data = (existing.data || "") + detail.data + } + // Update other fields if provided + if (detail.id !== undefined) existing.id = detail.id + if (detail.format !== undefined) existing.format = detail.format + if (detail.signature !== undefined) existing.signature = detail.signature + } else { + // Start new reasoning detail accumulation + reasoningDetailsAccumulator.set(key, { + type: detail.type, + text: detail.text, + summary: detail.summary, + data: detail.data, + id: detail.id, + format: detail.format, + signature: detail.signature, + index, + }) + } + + // Yield text for display (still fragmented for live streaming) + // Only reasoning.text and reasoning.summary have displayable content + // reasoning.encrypted is intentionally skipped as it contains redacted content + let reasoningText: string | undefined + if (detail.type === "reasoning.text" && typeof detail.text === "string") { + reasoningText = detail.text + } else if (detail.type === "reasoning.summary" && typeof detail.summary === "string") { + reasoningText = detail.summary + } + + if (reasoningText) { + hasYieldedReasoningFromDetails = true + yield { type: "reasoning", text: reasoningText } + } } - if (detail.summary !== undefined) { - existing.summary = (existing.summary || "") + detail.summary - } - if (detail.data !== undefined) { - existing.data = (existing.data || "") + detail.data + } + + // Handle top-level reasoning field for UI display. + // Skip if we've already yielded from reasoning_details to avoid duplicate display. + if ("reasoning" in delta && delta.reasoning && typeof delta.reasoning === "string") { + if (!hasYieldedReasoningFromDetails) { + yield { type: "reasoning", text: delta.reasoning } } - // Update other fields if provided - if (detail.id !== undefined) existing.id = detail.id - if (detail.format !== undefined) existing.format = detail.format - if (detail.signature !== undefined) existing.signature = detail.signature - } else { - // Start new reasoning detail accumulation - reasoningDetailsAccumulator.set(key, { - type: detail.type, - text: detail.text, - summary: detail.summary, - data: detail.data, - id: detail.id, - format: detail.format, - signature: detail.signature, - index, - }) } - // Yield text for display (still fragmented for live streaming) - // Only reasoning.text and reasoning.summary have displayable content - // reasoning.encrypted is intentionally skipped as it contains redacted content - let reasoningText: string | undefined - if (detail.type === "reasoning.text" && typeof detail.text === "string") { - reasoningText = detail.text - } else if (detail.type === "reasoning.summary" && typeof detail.summary === "string") { - reasoningText = detail.summary + // Emit raw tool call chunks - NativeToolCallParser handles state management + if ("tool_calls" in delta && Array.isArray(delta.tool_calls)) { + for (const toolCall of delta.tool_calls) { + yield { + type: "tool_call_partial", + index: toolCall.index, + id: toolCall.id, + name: toolCall.function?.name, + arguments: toolCall.function?.arguments, + } + } } - if (reasoningText) { - hasYieldedReasoningFromDetails = true - yield { type: "reasoning", text: reasoningText } + if (delta.content) { + yield { type: "text", text: delta.content } } } - } - // Handle top-level reasoning field for UI display. - // Skip if we've already yielded from reasoning_details to avoid duplicate display. - if ("reasoning" in delta && delta.reasoning && typeof delta.reasoning === "string") { - if (!hasYieldedReasoningFromDetails) { - yield { type: "reasoning", text: delta.reasoning } + // Process finish_reason to emit tool_call_end events + // This ensures tool calls are finalized even if the stream doesn't properly close + if (finishReason) { + const endEvents = NativeToolCallParser.processFinishReason(finishReason) + for (const event of endEvents) { + yield event + } } - } - // Emit raw tool call chunks - NativeToolCallParser handles state management - if ("tool_calls" in delta && Array.isArray(delta.tool_calls)) { - for (const toolCall of delta.tool_calls) { - yield { - type: "tool_call_partial", - index: toolCall.index, - id: toolCall.id, - name: toolCall.function?.name, - arguments: toolCall.function?.arguments, - } + if (chunk.usage) { + lastUsage = chunk.usage } } - if (delta.content) { - yield { type: "text", text: delta.content } + // After streaming completes, consolidate and store reasoning_details from the API. + // This filters out corrupted encrypted blocks (missing `data`) and consolidates by index. + if (reasoningDetailsAccumulator.size > 0) { + const rawDetails = Array.from(reasoningDetailsAccumulator.values()) + this.currentReasoningDetails = consolidateReasoningDetails(rawDetails) } - } - // Process finish_reason to emit tool_call_end events - // This ensures tool calls are finalized even if the stream doesn't properly close - if (finishReason) { - const endEvents = NativeToolCallParser.processFinishReason(finishReason) - for (const event of endEvents) { - yield event + if (lastUsage) { + yield { + type: "usage", + inputTokens: lastUsage.prompt_tokens || 0, + outputTokens: lastUsage.completion_tokens || 0, + cacheReadTokens: lastUsage.prompt_tokens_details?.cached_tokens, + reasoningTokens: lastUsage.completion_tokens_details?.reasoning_tokens, + totalCost: (lastUsage.cost_details?.upstream_inference_cost || 0) + (lastUsage.cost || 0), + } } + } catch (error) { + // Normalize abort-driven stream failures (SDK abort or timeout errors) to a + // DOM-standard AbortError so callers can detect the aborted request. + if (controller.signal.aborted) { + throw createAbortError("OpenRouter request aborted") + } + throw error } - - if (chunk.usage) { - lastUsage = chunk.usage - } - } - - // After streaming completes, consolidate and store reasoning_details from the API. - // This filters out corrupted encrypted blocks (missing `data`) and consolidates by index. - if (reasoningDetailsAccumulator.size > 0) { - const rawDetails = Array.from(reasoningDetailsAccumulator.values()) - this.currentReasoningDetails = consolidateReasoningDetails(rawDetails) - } - - if (lastUsage) { - yield { - type: "usage", - inputTokens: lastUsage.prompt_tokens || 0, - outputTokens: lastUsage.completion_tokens || 0, - cacheReadTokens: lastUsage.prompt_tokens_details?.cached_tokens, - reasoningTokens: lastUsage.completion_tokens_details?.reasoning_tokens, - totalCost: (lastUsage.cost_details?.upstream_inference_cost || 0) + (lastUsage.cost || 0), - } + } finally { + removeExternalAbortListener?.() } } @@ -602,15 +663,28 @@ export class OpenRouterHandler extends BaseProvider implements SingleCompletionH } // Add Anthropic beta header for fine-grained tool streaming when using Anthropic models - const requestOptions = modelId.startsWith("anthropic/") - ? { headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" } } - : undefined + // and forward the caller's abort signal / per-request timeout to the SDK. The client-level + // timeout remains as the default safety net; timeoutMs <= 0 disables the per-request timeout. + const requestOptions: OpenAI.RequestOptions = { + ...(modelId.startsWith("anthropic/") + ? { headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" } } + : undefined), + ...(options?.abortSignal && { signal: options.abortSignal }), + ...(typeof options?.timeoutMs === "number" && options.timeoutMs > 0 && { timeout: options.timeoutMs }), + } + + const requestAbortSignal = options?.abortSignal let response try { response = await this.client.chat.completions.create(completionParams, requestOptions) } catch (error) { + // Aborted requests are user-initiated: surface them as AbortError (this also covers + // timeouts, which abort the same signal) instead of a completion error. + if (requestAbortSignal?.aborted) { + throw createAbortError("OpenRouter completion aborted") + } // Try to parse as OpenRouter error structure using Zod const parseResult = OpenRouterErrorResponseSchema.safeParse(error) @@ -650,6 +724,11 @@ export class OpenRouterHandler extends BaseProvider implements SingleCompletionH } } + if (requestAbortSignal?.aborted) { + // The response resolved after the request was aborted: do not return the late result. + throw createAbortError("OpenRouter completion aborted") + } + if ("error" in response) { this.handleStreamingError(response.error as OpenRouterError, modelId, "completePrompt") } diff --git a/src/api/providers/poe.ts b/src/api/providers/poe.ts index fb3255c572..8c15d5f4a2 100644 --- a/src/api/providers/poe.ts +++ b/src/api/providers/poe.ts @@ -22,9 +22,20 @@ import { BaseProvider } from "./base-provider" import { NOT_PROVIDED } from "./constants" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { getModelsFromCache } from "./fetchers/modelCache" +import { mergeAbortSignalAndTimeout } from "./utils/abort-signal" const DEFAULT_THINKING_BUDGET = 8192 +/** + * Create a DOM-standard AbortError so callers can detect aborted requests + * (matches the error name produced by native abort-based APIs). + */ +function createAbortError(message: string): Error { + const error = new Error(message) + error.name = "AbortError" + return error +} + export class PoeHandler extends BaseProvider implements SingleCompletionHandler { protected options: ApiHandlerOptions private poe: PoeProvider @@ -54,105 +65,160 @@ export class PoeHandler extends BaseProvider implements SingleCompletionHandler messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { - const { id, info } = this.getModel() - const languageModel = this.poe(id) - - const aiSdkMessages = convertToAiSdkMessages(messages) - const openAiTools = this.convertToolsForOpenAI(metadata?.tools) - const aiSdkTools = convertToolsForAiSdk(openAiTools) as ToolSet | undefined - - const useBudget = shouldUseReasoningBudget({ model: info, settings: this.options }) - const useEffort = !useBudget && shouldUseReasoningEffort({ model: info, settings: this.options }) - - // Only pass temperature when the user explicitly configured it. - let temperature: number | undefined = this.options.modelTemperature ?? undefined - let maxOutputTokens: number | undefined - const providerOptions: NonNullable[0]["providerOptions"]> & { - poe?: PoeScopedProviderOptions - } = {} - - if (useBudget) { - const requestedBudget = this.options.modelMaxThinkingTokens ?? DEFAULT_THINKING_BUDGET - // maxOutputTokens is the text-only budget; reasoningBudgetTokens is - // separate, so total output = maxOutputTokens + reasoningBudgetTokens. - maxOutputTokens = this.options.modelMaxTokens ?? Math.max(0, (info.maxTokens ?? 0) - requestedBudget) - providerOptions.poe = { - reasoningBudgetTokens: requestedBudget, - } - temperature = 1.0 - } else if (useEffort) { - let effort = (this.options.reasoningEffort ?? info.reasoningEffort ?? "medium") as ReasoningEffortExtended - // Validate that the effort level is actually supported by the current model - const supportedEfforts = info.supportsReasoningEffort - if (Array.isArray(supportedEfforts) && !supportedEfforts.includes(effort as any)) { - effort = (info.reasoningEffort as ReasoningEffortExtended) ?? "medium" - } - providerOptions.poe = { - reasoningEffort: effort, - reasoningSummary: "auto", - } - if (this.options.modelMaxTokens) { - maxOutputTokens = this.options.modelMaxTokens + // Per-request AbortController: external aborts cancel the in-flight AI SDK request + // (the AI SDK aborts the underlying fetch when its abortSignal fires). + const controller = new AbortController() + + // Bridge the external abort signal into the per-request controller: + // - pre-aborted guard: abort immediately when the signal is already aborted + // - { once: true }: the listener removes itself after the first abort + // - explicit removal in finally: the listener must not outlive a request that + // completes (or fails) without being aborted + const externalAbortSignal = metadata?.abortSignal + let removeExternalAbortListener: (() => void) | undefined + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + controller.abort() + } else { + const onExternalAbort = () => controller.abort() + externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) + removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) } } - let result try { - result = streamText({ - model: languageModel, - system: systemPrompt, - messages: aiSdkMessages, - temperature, - maxOutputTokens, - tools: aiSdkTools, - toolChoice: mapToolChoice(metadata?.tool_choice as any), - ...(Object.keys(providerOptions).length > 0 && { providerOptions }), - }) - } catch (error) { - const errorMessage = error instanceof Error ? error.message : String(error) - TelemetryService.instance.captureException( - new ApiProviderError(errorMessage, providerIdentifiers.poe, id, "createMessage"), - ) - throw new Error(`Poe completion error: ${errorMessage}`) - } + // The request was already aborted before we started: fail fast without calling the API. + if (controller.signal.aborted) { + throw createAbortError("Poe request aborted") + } - try { - for await (const part of result.fullStream) { - for (const chunk of processAiSdkStreamPart(part)) { - yield chunk + const { id, info } = this.getModel() + const languageModel = this.poe(id) + const aiSdkMessages = convertToAiSdkMessages(messages) + const openAiTools = this.convertToolsForOpenAI(metadata?.tools) + const aiSdkTools = convertToolsForAiSdk(openAiTools) as ToolSet | undefined + + const useBudget = shouldUseReasoningBudget({ model: info, settings: this.options }) + const useEffort = !useBudget && shouldUseReasoningEffort({ model: info, settings: this.options }) + + // Only pass temperature when the user explicitly configured it. + let temperature: number | undefined = this.options.modelTemperature ?? undefined + let maxOutputTokens: number | undefined + const providerOptions: NonNullable[0]["providerOptions"]> & { + poe?: PoeScopedProviderOptions + } = {} + + if (useBudget) { + const requestedBudget = this.options.modelMaxThinkingTokens ?? DEFAULT_THINKING_BUDGET + // maxOutputTokens is the text-only budget; reasoningBudgetTokens is + // separate, so total output = maxOutputTokens + reasoningBudgetTokens. + maxOutputTokens = this.options.modelMaxTokens ?? Math.max(0, (info.maxTokens ?? 0) - requestedBudget) + providerOptions.poe = { + reasoningBudgetTokens: requestedBudget, + } + temperature = 1.0 + } else if (useEffort) { + let effort = (this.options.reasoningEffort ?? + info.reasoningEffort ?? + "medium") as ReasoningEffortExtended + // Validate that the effort level is actually supported by the current model + const supportedEfforts = info.supportsReasoningEffort + if (Array.isArray(supportedEfforts) && !supportedEfforts.includes(effort as any)) { + effort = (info.reasoningEffort as ReasoningEffortExtended) ?? "medium" + } + providerOptions.poe = { + reasoningEffort: effort, + reasoningSummary: "auto", + } + if (this.options.modelMaxTokens) { + maxOutputTokens = this.options.modelMaxTokens } } - const usage = await result.usage - if (usage) { - const metrics = extractUsageMetrics(usage as any) - yield { - type: "usage" as const, - inputTokens: metrics.inputTokens, - outputTokens: metrics.outputTokens, - cacheReadTokens: metrics.cacheReadTokens, - cacheWriteTokens: metrics.cacheWriteTokens, - reasoningTokens: metrics.reasoningTokens, + let result + try { + result = streamText({ + model: languageModel, + system: systemPrompt, + messages: aiSdkMessages, + temperature, + maxOutputTokens, + tools: aiSdkTools, + toolChoice: mapToolChoice(metadata?.tool_choice as any), + ...(Object.keys(providerOptions).length > 0 && { providerOptions }), + abortSignal: controller.signal, + }) + } catch (error) { + // Aborted requests are user-initiated: surface them as AbortError instead of + // a completion error. + if (controller.signal.aborted) { + throw createAbortError("Poe request aborted") } + const errorMessage = error instanceof Error ? error.message : String(error) + TelemetryService.instance.captureException( + new ApiProviderError(errorMessage, providerIdentifiers.poe, id, "createMessage"), + ) + throw new Error(`Poe completion error: ${errorMessage}`) } - } catch (error) { - const errorMessage = error instanceof Error ? error.message : String(error) - TelemetryService.instance.captureException( - new ApiProviderError(errorMessage, providerIdentifiers.poe, id, "createMessage"), - ) - throw new Error(`Poe streaming error: ${errorMessage}`) + + try { + for await (const part of result.fullStream) { + for (const chunk of processAiSdkStreamPart(part)) { + yield chunk + } + } + + const usage = await result.usage + if (usage) { + const metrics = extractUsageMetrics(usage as any) + yield { + type: "usage" as const, + inputTokens: metrics.inputTokens, + outputTokens: metrics.outputTokens, + cacheReadTokens: metrics.cacheReadTokens, + cacheWriteTokens: metrics.cacheWriteTokens, + reasoningTokens: metrics.reasoningTokens, + } + } + } catch (error) { + // Aborted requests are user-initiated: surface them as AbortError instead of + // a completion error. + if (controller.signal.aborted) { + throw createAbortError("Poe request aborted") + } + const errorMessage = error instanceof Error ? error.message : String(error) + TelemetryService.instance.captureException( + new ApiProviderError(errorMessage, providerIdentifiers.poe, id, "createMessage"), + ) + throw new Error(`Poe streaming error: ${errorMessage}`) + } + } finally { + removeExternalAbortListener?.() } } async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { const { id } = this.getModel() + // Merge the caller's abort signal with the per-request timeout (timeoutMs <= 0 disables it). + const mergedAbortSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) try { const { text } = await generateText({ model: this.poe(id), prompt, + ...(mergedAbortSignal && { abortSignal: mergedAbortSignal }), }) + + if (mergedAbortSignal?.aborted) { + // The response resolved after the request was aborted: do not return the late result. + throw createAbortError("Poe completion aborted") + } return text } catch (error) { + // Aborted requests are user-initiated: surface them as AbortError (this also covers + // timeouts, which abort the same signal) instead of a completion error. + if (mergedAbortSignal?.aborted) { + throw createAbortError("Poe completion aborted") + } const errorMessage = error instanceof Error ? error.message : String(error) TelemetryService.instance.captureException( new ApiProviderError(errorMessage, providerIdentifiers.poe, id, "completePrompt"), diff --git a/src/api/providers/requesty.ts b/src/api/providers/requesty.ts index 1ba0771ce2..3b1301b955 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -56,6 +56,16 @@ type RequestyChatCompletionParams = OpenAI.Chat.ChatCompletionCreateParams & { thinking?: AnthropicProviderReasoningParams } +/** + * Create a DOM-standard AbortError so callers can detect aborted requests + * (matches the error name produced by native abort-based APIs). + */ +function createAbortError(message: string): Error { + const error = new Error(message) + error.name = "AbortError" + return error +} + export class RequestyHandler extends BaseProvider implements SingleCompletionHandler { protected options: ApiHandlerOptions protected models: ModelRecord = {} @@ -133,80 +143,124 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { - const { - id: model, - info, - maxTokens: max_tokens, - temperature, - reasoningEffort: reasoning_effort, - reasoning: thinking, - } = await this.fetchModel() - - const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ - { role: "system", content: systemPrompt }, - ...convertToOpenAiMessages(messages), - ] - - // Map extended efforts to OpenAI Chat Completions-accepted values (omit unsupported) - const allowedEffort = (["low", "medium", "high"] as const).includes(reasoning_effort as any) - ? (reasoning_effort as OpenAI.Chat.Completions.ChatCompletionCreateParamsStreaming["reasoning_effort"]) - : undefined - - const completionParams: RequestyChatCompletionParamsStreaming = { - messages: openAiMessages, - model, - max_tokens, - temperature, - ...(allowedEffort && { reasoning_effort: allowedEffort }), - ...(thinking && { thinking }), - stream: true, - stream_options: { include_usage: true }, - requesty: { trace_id: metadata?.taskId, extra: { mode: metadata?.mode } }, - tools: this.convertToolsForOpenAI(metadata?.tools), - tool_choice: metadata?.tool_choice, + // Per-request AbortController: external aborts cancel the in-flight request + // without replacing the client-level timeout, which remains the default safety net. + const controller = new AbortController() + + // Bridge the external abort signal into the per-request controller: + // - pre-aborted guard: abort immediately when the signal is already aborted + // - { once: true }: the listener removes itself after the first abort + // - explicit removal in finally: the listener must not outlive a request that + // completes (or fails) without being aborted + const externalAbortSignal = metadata?.abortSignal + let removeExternalAbortListener: (() => void) | undefined + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + controller.abort() + } else { + const onExternalAbort = () => controller.abort() + externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) + removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) + } } - let stream try { - // With streaming params type, SDK returns an async iterable stream - stream = await this.client.chat.completions.create(completionParams) - } catch (error) { - throw handleOpenAIError(error, this.providerName) - } - let lastUsage: any = undefined - - for await (const chunk of stream) { - const delta = chunk.choices[0]?.delta + // The request was already aborted before we started: fail fast without calling the API. + if (controller.signal.aborted) { + throw createAbortError("Requesty request aborted") + } - if (delta?.content) { - yield { type: "text", text: delta.content } + const { + id: model, + info, + maxTokens: max_tokens, + temperature, + reasoningEffort: reasoning_effort, + reasoning: thinking, + } = await this.fetchModel() + + const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ + { role: "system", content: systemPrompt }, + ...convertToOpenAiMessages(messages), + ] + + // Map extended efforts to OpenAI Chat Completions-accepted values (omit unsupported) + const allowedEffort = (["low", "medium", "high"] as const).includes(reasoning_effort as any) + ? (reasoning_effort as OpenAI.Chat.Completions.ChatCompletionCreateParamsStreaming["reasoning_effort"]) + : undefined + + const completionParams: RequestyChatCompletionParamsStreaming = { + messages: openAiMessages, + model, + max_tokens, + temperature, + ...(allowedEffort && { reasoning_effort: allowedEffort }), + ...(thinking && { thinking }), + stream: true, + stream_options: { include_usage: true }, + requesty: { trace_id: metadata?.taskId, extra: { mode: metadata?.mode } }, + tools: this.convertToolsForOpenAI(metadata?.tools), + tool_choice: metadata?.tool_choice, } - const reasoningText = extractReasoningFromDelta(delta) - if (reasoningText) { - yield { type: "reasoning", text: reasoningText } + let stream + try { + // With streaming params type, SDK returns an async iterable stream + stream = await this.client.chat.completions.create(completionParams, { signal: controller.signal }) + } catch (error) { + // Aborted requests are user-initiated: surface them as AbortError instead of + // a completion error. + if (controller.signal.aborted) { + throw createAbortError("Requesty request aborted") + } + throw handleOpenAIError(error, this.providerName) } + try { + let lastUsage: any = undefined + + for await (const chunk of stream) { + const delta = chunk.choices[0]?.delta + + if (delta?.content) { + yield { type: "text", text: delta.content } + } - // Handle native tool calls - if (delta && "tool_calls" in delta && Array.isArray(delta.tool_calls)) { - for (const toolCall of delta.tool_calls) { - yield { - type: "tool_call_partial", - index: toolCall.index, - id: toolCall.id, - name: toolCall.function?.name, - arguments: toolCall.function?.arguments, + const reasoningText = extractReasoningFromDelta(delta) + if (reasoningText) { + yield { type: "reasoning", text: reasoningText } + } + + // Handle native tool calls + if (delta && "tool_calls" in delta && Array.isArray(delta.tool_calls)) { + for (const toolCall of delta.tool_calls) { + yield { + type: "tool_call_partial", + index: toolCall.index, + id: toolCall.id, + name: toolCall.function?.name, + arguments: toolCall.function?.arguments, + } + } + } + + if (chunk.usage) { + lastUsage = chunk.usage } } - } - if (chunk.usage) { - lastUsage = chunk.usage + if (lastUsage) { + yield this.processUsageMetrics(lastUsage, info) + } + } catch (error) { + // Normalize abort-driven stream failures (SDK abort or timeout errors) to a + // DOM-standard AbortError so callers can detect the aborted request. + if (controller.signal.aborted) { + throw createAbortError("Requesty request aborted") + } + throw error } - } - - if (lastUsage) { - yield this.processUsageMetrics(lastUsage, info) + } finally { + removeExternalAbortListener?.() } } @@ -222,12 +276,31 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan temperature: temperature, } + const requestAbortSignal = options?.abortSignal + + // Forward the caller's abort signal / per-request timeout to the SDK. The client-level + // timeout remains as the default safety net; timeoutMs <= 0 disables the per-request timeout. + const createOptions: OpenAI.RequestOptions = { + ...(requestAbortSignal && { signal: requestAbortSignal }), + ...(typeof options?.timeoutMs === "number" && options.timeoutMs > 0 && { timeout: options.timeoutMs }), + } + let response: OpenAI.Chat.ChatCompletion try { - response = await this.client.chat.completions.create(completionParams) + response = await this.client.chat.completions.create(completionParams, createOptions) } catch (error) { + // Aborted requests are user-initiated: surface them as AbortError (this also covers + // timeouts, which abort the same signal) instead of a completion error. + if (requestAbortSignal?.aborted) { + throw createAbortError("Requesty completion aborted") + } throw handleOpenAIError(error, this.providerName) } + + if (requestAbortSignal?.aborted) { + // The response resolved after the request was aborted: do not return the late result. + throw createAbortError("Requesty completion aborted") + } return response.choices[0]?.message.content || "" } } From 4856f5e61cd1c1170f3c2dede07092b81de581f9 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 15:01:44 +0800 Subject: [PATCH 2/3] fix(api): normalize openrouter/requesty timeouts to AbortError + cover new abort paths --- .../providers/__tests__/openrouter.spec.ts | 287 ++++++++++++++++++ src/api/providers/__tests__/poe.spec.ts | 90 ++++++ src/api/providers/__tests__/requesty.spec.ts | 108 ++++++- src/api/providers/openrouter.ts | 13 +- src/api/providers/requesty.ts | 8 +- 5 files changed, 495 insertions(+), 11 deletions(-) diff --git a/src/api/providers/__tests__/openrouter.spec.ts b/src/api/providers/__tests__/openrouter.spec.ts index a2130be039..5716930e1f 100644 --- a/src/api/providers/__tests__/openrouter.spec.ts +++ b/src/api/providers/__tests__/openrouter.spec.ts @@ -608,6 +608,224 @@ describe("OpenRouterHandler", () => { await expect(iteration).rejects.toMatchObject({ name: "AbortError" }) expect(chunks).toContainEqual({ type: "text", text: "first" }) }) + it("excludes reasoning for Gemini 2.5 Pro models by default", async () => { + const handler = new OpenRouterHandler( + makeApiHandlerOptions({ + ...mockOptions, + openRouterModelId: "google/gemini-2.5-pro-preview", + }), + ) + const mockCreate = vitest + .fn() + .mockResolvedValue(asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "ok" } }] }])) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const stream = handler.createMessage("system", [{ role: "user" as const, content: "hi" }]) + await collectStream(stream) + + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ reasoning: { exclude: true } }), + expect.any(Object), + ) + }) + + it("uses user role for the system prompt with DeepSeek R1 models", async () => { + const handler = new OpenRouterHandler( + makeApiHandlerOptions({ + ...mockOptions, + openRouterModelId: "deepseek/deepseek-r1", + }), + ) + const mockCreate = vitest + .fn() + .mockResolvedValue(asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "ok" } }] }])) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const stream = handler.createMessage("system prompt", [{ role: "user" as const, content: "hi" }]) + await collectStream(stream) + + const params = mockCreate.mock.calls[0][0] as { messages: { role: string; content: unknown }[] } + expect(params.messages[0].role).toBe("user") + expect(params.messages.map((m) => m.role)).not.toContain("system") + }) + + it("injects a fake encrypted reasoning block for Gemini tool calls without encrypted reasoning", async () => { + const handler = new OpenRouterHandler( + makeApiHandlerOptions({ + ...mockOptions, + openRouterModelId: "google/gemini-2.5-flash", + }), + ) + const mockCreate = vitest + .fn() + .mockResolvedValue(asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "ok" } }] }])) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + // reasoning_details is an OpenRouter extension field round-tripped on assistant + // messages; the Anthropic SDK types do not include it, hence the structural cast. + const assistantMessage = { + role: "assistant" as const, + content: [{ type: "tool_use" as const, id: "toolu_01", name: "get_weather", input: { city: "SF" } }], + reasoning_details: [{ type: "reasoning.text", id: "toolu_01", text: "thinking", index: 0 }], + } + const stream = handler.createMessage("system", [ + assistantMessage as unknown as Anthropic.Messages.MessageParam, + ]) + await collectStream(stream) + + const params = mockCreate.mock.calls[0][0] as { + messages: { + role: string + tool_calls?: { id: string }[] + reasoning_details?: { type: string; id: string; data: string }[] + }[] + } + const assistant = params.messages.find((m) => m.role === "assistant") + expect(assistant?.tool_calls).toHaveLength(1) + const encrypted = assistant?.reasoning_details?.find((d) => d.type === "reasoning.encrypted") + expect(encrypted).toMatchObject({ + id: "toolu_01", + data: "skip_thought_signature_validator", + }) + }) + + it("accumulates and yields reasoning_details from streamed chunks", async () => { + const handler = new OpenRouterHandler(mockOptions) + const mockCreate = vitest.fn().mockResolvedValue( + asyncStreamFrom([ + { id: "1", choices: [{ delta: { reasoning: "top-level thinking" } }] }, + { + id: "2", + choices: [ + { delta: { reasoning_details: [{ type: "reasoning.text", index: 0, text: "thinking " }] } }, + ], + }, + { + id: "3", + choices: [ + { + delta: { + reasoning_details: [ + { + type: "reasoning.text", + index: 0, + text: "more", + id: "r1", + format: "google-gemini-v1", + signature: "sig", + }, + ], + }, + }, + ], + }, + { + id: "4", + choices: [ + { delta: { reasoning_details: [{ type: "reasoning.summary", index: 1, summary: "sum" }] } }, + ], + }, + { id: "5", choices: [{ delta: { content: "hello" } }] }, + { + id: "6", + choices: [ + { + delta: { + reasoning_details: [{ type: "reasoning.summary", index: 1, summary: " more" }], + }, + }, + ], + }, + { + id: "7", + choices: [ + { delta: { reasoning_details: [{ type: "reasoning.encrypted", index: 2, data: "enc-" }] } }, + ], + }, + { + id: "8", + choices: [ + { + delta: { + reasoning_details: [{ type: "reasoning.encrypted", index: 2, data: "rypted" }], + }, + }, + ], + }, + ]), + ) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const chunks = await collectStream( + handler.createMessage("system", [{ role: "user" as const, content: "hi" }]), + ) + + expect(chunks).toContainEqual({ type: "reasoning", text: "top-level thinking" }) + expect(chunks).toContainEqual({ type: "reasoning", text: "thinking " }) + expect(chunks).toContainEqual({ type: "reasoning", text: "sum" }) + expect(chunks).toContainEqual({ type: "reasoning", text: " more" }) + expect(chunks).toContainEqual({ type: "text", text: "hello" }) + + const details = handler.getReasoningDetails() + expect(details).toHaveLength(3) + expect(details?.find((d) => d.type === "reasoning.summary")?.summary).toBe("sum more") + expect(details?.find((d) => d.type === "reasoning.encrypted")?.data).toBe("enc-rypted") + }) + + it("rejects with AbortError when the external signal aborts during request creation", async () => { + const handler = new OpenRouterHandler(mockOptions) + const controller = new AbortController() + + const mockCreate = vitest + .fn() + .mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + // Emulate the OpenAI SDK: the pending request rejects when the signal aborts. + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The user aborted a request") + abortError.name = "AbortError" + throw abortError + }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("system", [{ role: "user" as const, content: "hi" }], metadata) + + const nextPromise = generator.next() + // Let the generator reach the pending create() call, then abort. + await new Promise((resolve) => setTimeout(resolve, 10)) + controller.abort() + + await expect(nextPromise).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("reports OpenRouter structured errors in createMessage with telemetry", async () => { + const handler = new OpenRouterHandler(mockOptions) + const mockCreate = vitest.fn().mockRejectedValueOnce({ + error: { + message: "Model not found", + code: 404, + metadata: { raw: '{"message":"upstream: model not found"}' }, + }, + }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const generator = handler.createMessage("system", [{ role: "user" as const, content: "hi" }]) + + await expect(generator.next()).rejects.toThrow(/completion error/) + expect(mockCaptureException).toHaveBeenCalledTimes(1) + }) }) describe("completePrompt", () => { @@ -860,5 +1078,74 @@ describe("OpenRouterHandler", () => { await expect(promise).rejects.toMatchObject({ name: "AbortError" }) }) + it("rejects with AbortError when only a timeout is provided and it elapses", async () => { + // Non-Anthropic model: also exercises the no-beta-header branch of requestOptions. + const handler = new OpenRouterHandler( + makeApiHandlerOptions({ + ...mockOptions, + openRouterModelId: "openai/gpt-4o", + }), + ) + const mockCreate = vitest + .fn() + .mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + // Emulate the OpenAI SDK: the in-flight request rejects when the signal times out. + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const timeoutError = new Error("TimeoutError: Request timed out.") + timeoutError.name = "TimeoutError" + throw timeoutError + }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + await expect(handler.completePrompt("test prompt", { timeoutMs: 50 })).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("rejects with AbortError when both an abort signal and a timeout are provided", async () => { + const handler = new OpenRouterHandler(mockOptions) + const controller = new AbortController() + + let requestSignal: AbortSignal | undefined + const mockCreate = vitest + .fn() + .mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + requestSignal = options?.signal + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The user aborted a request") + abortError.name = "AbortError" + throw abortError + }) + const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } + client.chat = { completions: { create: mockCreate } } + + const promise = handler.completePrompt("test prompt", { + abortSignal: controller.signal, + timeoutMs: 100_000, + }) + controller.abort() + + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + // The SDK received a merged signal (not the caller's signal) plus the timeout. + expect(requestSignal).toBeDefined() + expect(requestSignal).not.toBe(controller.signal) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 100_000 }), + ) + }) }) }) diff --git a/src/api/providers/__tests__/poe.spec.ts b/src/api/providers/__tests__/poe.spec.ts index f636a13bef..7a08bc08e1 100644 --- a/src/api/providers/__tests__/poe.spec.ts +++ b/src/api/providers/__tests__/poe.spec.ts @@ -5,6 +5,7 @@ import { getModelsFromCache } from "../fetchers/modelCache" import { makeCreateMessageMetadata } from "../../../test-utils/api" import { clearAllMocks } from "../../../test-utils/reset" +import { collectStream } from "../../../test-utils/stream" const { mockStreamText, mockGenerateText, mockCreatePoe, mockGetModelsFromCache, mockCaptureException } = vitest.hoisted(() => ({ @@ -302,6 +303,52 @@ describe("PoeHandler", () => { await expect(iteration).rejects.toMatchObject({ name: "AbortError" }) expect(chunks).toContainEqual({ type: "text", text: "Hello " }) }) + it("rejects with AbortError when the external signal aborts during request creation", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + const controller = new AbortController() + + mockStreamText.mockImplementationOnce(() => { + // Emulate the AI SDK failing synchronously: abort the external signal first so + // the catch normalizes the failure to a DOM-standard AbortError. + controller.abort() + const abortError = new Error("The operation was aborted") + abortError.name = "AbortError" + throw abortError + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const nextPromise = handler + .createMessage("system", [{ role: "user" as const, content: "hi" }], metadata) + .next() + + await expect(nextPromise).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("rejects with a completion error when request creation fails without abort", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + mockStreamText.mockImplementationOnce(() => { + throw new Error("boom") + }) + + await expect( + handler.createMessage("system", [{ role: "user" as const, content: "hi" }]).next(), + ).rejects.toThrow("Poe completion error: boom") + }) + + it("rejects with a streaming error when the stream fails without abort", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + mockStreamText.mockReturnValueOnce({ + fullStream: (async function* () { + yield { type: "text-delta", text: "Hello " } + throw new Error("stream broke") + })(), + usage: Promise.resolve(undefined), + }) + + await expect( + collectStream(handler.createMessage("system", [{ role: "user" as const, content: "hi" }])), + ).rejects.toThrow("Poe streaming error: stream broke") + }) }) describe("reasoning", () => { @@ -591,5 +638,48 @@ describe("PoeHandler", () => { await expect(handler.completePrompt("test prompt")).rejects.toThrow() }) + it("completePrompt rejects with AbortError when the response resolves after abort", async () => { + const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/gpt-4o" }) + const controller = new AbortController() + mockGenerateText.mockImplementationOnce(async (args: { abortSignal?: AbortSignal }) => { + // The generation only settles once the abort signal has fired (late result). + await new Promise((resolve) => { + if (args.abortSignal?.aborted) { + resolve() + } else { + args.abortSignal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + return { text: "late result" } + }) + + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal }) + controller.abort() + + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("completePrompt should pass reasoning effort for effort-capable models", async () => { + const handler = new PoeHandler({ + poeApiKey: "key", + apiModelId: "openai/o3", + enableReasoningEffort: true, + reasoningEffort: "low", + modelMaxTokens: 8192, + }) + mockStreamText.mockReturnValueOnce({ + fullStream: (async function* () {})(), + usage: Promise.resolve(undefined), + }) + + await handler.createMessage("system", [{ role: "user" as const, content: "hi" }]).next() + + expect(mockStreamText).toHaveBeenCalledWith( + expect.objectContaining({ + maxOutputTokens: 8192, + providerOptions: { poe: { reasoningEffort: "low", reasoningSummary: "auto" } }, + }), + ) + }) }) }) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index 325cfaf6cd..565793511f 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -620,6 +620,48 @@ describe("RequestyHandler", () => { await expect(iteration).rejects.toMatchObject({ name: "AbortError" }) expect(chunks).toContainEqual({ type: "text", text: "first" }) }) + it("rejects with AbortError when the external signal aborts during request creation", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + // Emulate the OpenAI SDK: the pending request rejects when the signal aborts. + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The user aborted a request") + abortError.name = "AbortError" + throw abortError + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + const nextPromise = generator.next() + // Let the generator reach the pending create() call, then abort. + await new Promise((resolve) => setTimeout(resolve, 10)) + controller.abort() + + await expect(nextPromise).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("rethrows non-abort stream errors from createMessage", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockImplementationOnce(async () => { + return (async function* () { + yield { id: "1", choices: [{ delta: { content: "first" } }] } + throw new Error("stream broke") + })() + }) + + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }]) + + await expect(collectStream(generator)).rejects.toThrow("stream broke") + }) }) describe("completePrompt", () => { @@ -740,9 +782,12 @@ describe("RequestyHandler", () => { mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) await handler.completePrompt("test prompt", { timeoutMs: 5000 }) - expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { - timeout: 5000, - }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ + timeout: 5000, + }), + ) }) it("should work without options (backward compatible)", async () => { @@ -790,5 +835,62 @@ describe("RequestyHandler", () => { await expect(promise).rejects.toMatchObject({ name: "AbortError" }) }) + it("rejects with AbortError when only a timeout is provided and it elapses", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + // Emulate the OpenAI SDK: the in-flight request rejects when the signal times out. + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const timeoutError = new Error("TimeoutError: Request timed out.") + timeoutError.name = "TimeoutError" + throw timeoutError + }) + + await expect(handler.completePrompt("test prompt", { timeoutMs: 50 })).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("rejects with AbortError when both an abort signal and a timeout are provided", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + let requestSignal: AbortSignal | undefined + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + requestSignal = options?.signal + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + const abortError = new Error("The user aborted a request") + abortError.name = "AbortError" + throw abortError + }) + + const promise = handler.completePrompt("test prompt", { + abortSignal: controller.signal, + timeoutMs: 100_000, + }) + controller.abort() + + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + // The SDK received a merged signal (not the caller's signal) plus the timeout. + expect(requestSignal).toBeDefined() + expect(requestSignal).not.toBe(controller.signal) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ + timeout: 100_000, + }), + ) + }) }) }) diff --git a/src/api/providers/openrouter.ts b/src/api/providers/openrouter.ts index 2ed7094761..1c6a0de401 100644 --- a/src/api/providers/openrouter.ts +++ b/src/api/providers/openrouter.ts @@ -38,6 +38,7 @@ import { DEFAULT_HEADERS, NOT_PROVIDED } from "./constants" import { BaseProvider } from "./base-provider" import type { ApiHandlerCreateMessageMetadata, CompletePromptOptions, SingleCompletionHandler } from "../index" import { handleOpenAIError } from "./utils/error-handler" +import { mergeAbortSignalAndTimeout } from "./utils/abort-signal" import { generateImageWithProvider, ImageGenerationResult } from "./utils/image-generation" import { applyRouterToolPreferences } from "./utils/router-tool-preferences" @@ -663,18 +664,20 @@ export class OpenRouterHandler extends BaseProvider implements SingleCompletionH } // Add Anthropic beta header for fine-grained tool streaming when using Anthropic models - // and forward the caller's abort signal / per-request timeout to the SDK. The client-level - // timeout remains as the default safety net; timeoutMs <= 0 disables the per-request timeout. + // and forward the caller's abort signal / per-request timeout to the SDK. The merged signal + // aborts when either the caller's signal or the timeout fires, so timeouts are normalized to + // AbortError in the catch below. The client-level timeout remains the default safety net; + // timeoutMs <= 0 disables the per-request timeout, and 0 is never passed to the SDK. + const requestAbortSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) + const requestOptions: OpenAI.RequestOptions = { ...(modelId.startsWith("anthropic/") ? { headers: { "x-anthropic-beta": "fine-grained-tool-streaming-2025-05-14" } } : undefined), - ...(options?.abortSignal && { signal: options.abortSignal }), + ...(requestAbortSignal && { signal: requestAbortSignal }), ...(typeof options?.timeoutMs === "number" && options.timeoutMs > 0 && { timeout: options.timeoutMs }), } - const requestAbortSignal = options?.abortSignal - let response try { diff --git a/src/api/providers/requesty.ts b/src/api/providers/requesty.ts index 3b1301b955..3f3fa4f2c8 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -23,6 +23,7 @@ import { BaseProvider } from "./base-provider" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { toRequestyServiceUrl } from "../../shared/utils/requesty" import { handleOpenAIError } from "./utils/error-handler" +import { mergeAbortSignalAndTimeout } from "./utils/abort-signal" import { applyRouterToolPreferences } from "./utils/router-tool-preferences" import { extractReasoningFromDelta } from "./utils/extract-reasoning" @@ -276,10 +277,11 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan temperature: temperature, } - const requestAbortSignal = options?.abortSignal + // Merge the caller's abort signal with the per-request timeout (timeoutMs <= 0 disables it) + // so both abort and timeout reject with a DOM-standard AbortError in the catch below. The + // client-level timeout remains the default safety net; 0 is never passed to the SDK timeout. + const requestAbortSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) - // Forward the caller's abort signal / per-request timeout to the SDK. The client-level - // timeout remains as the default safety net; timeoutMs <= 0 disables the per-request timeout. const createOptions: OpenAI.RequestOptions = { ...(requestAbortSignal && { signal: requestAbortSignal }), ...(typeof options?.timeoutMs === "number" && options.timeoutMs > 0 && { timeout: options.timeoutMs }), From 078715141d823569739193fecadb5a432e858046 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 16:12:47 +0800 Subject: [PATCH 3/3] test(api): align gateway-a test names and deterministic abort synchronization --- src/api/providers/__tests__/openrouter.spec.ts | 8 ++++++++ src/api/providers/__tests__/poe.spec.ts | 2 +- src/api/providers/__tests__/requesty.spec.ts | 10 ++++++++-- 3 files changed, 17 insertions(+), 3 deletions(-) diff --git a/src/api/providers/__tests__/openrouter.spec.ts b/src/api/providers/__tests__/openrouter.spec.ts index 5716930e1f..00aea2aedd 100644 --- a/src/api/providers/__tests__/openrouter.spec.ts +++ b/src/api/providers/__tests__/openrouter.spec.ts @@ -618,6 +618,7 @@ describe("OpenRouterHandler", () => { const mockCreate = vitest .fn() .mockResolvedValue(asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "ok" } }] }])) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } client.chat = { completions: { create: mockCreate } } @@ -640,6 +641,7 @@ describe("OpenRouterHandler", () => { const mockCreate = vitest .fn() .mockResolvedValue(asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "ok" } }] }])) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } client.chat = { completions: { create: mockCreate } } @@ -661,6 +663,7 @@ describe("OpenRouterHandler", () => { const mockCreate = vitest .fn() .mockResolvedValue(asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "ok" } }] }])) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } client.chat = { completions: { create: mockCreate } } @@ -757,6 +760,7 @@ describe("OpenRouterHandler", () => { }, ]), ) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } client.chat = { completions: { create: mockCreate } } @@ -795,6 +799,7 @@ describe("OpenRouterHandler", () => { abortError.name = "AbortError" throw abortError }) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } client.chat = { completions: { create: mockCreate } } @@ -818,6 +823,7 @@ describe("OpenRouterHandler", () => { metadata: { raw: '{"message":"upstream: model not found"}' }, }, }) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } client.chat = { completions: { create: mockCreate } } @@ -1101,6 +1107,7 @@ describe("OpenRouterHandler", () => { timeoutError.name = "TimeoutError" throw timeoutError }) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } client.chat = { completions: { create: mockCreate } } @@ -1129,6 +1136,7 @@ describe("OpenRouterHandler", () => { abortError.name = "AbortError" throw abortError }) + // The auto-mocked OpenAI client is injected via a structural type to avoid `any` casts. const client = handler["client"] as unknown as { chat: { completions: { create: typeof mockCreate } } } client.chat = { completions: { create: mockCreate } } diff --git a/src/api/providers/__tests__/poe.spec.ts b/src/api/providers/__tests__/poe.spec.ts index 7a08bc08e1..d87b3f8ec0 100644 --- a/src/api/providers/__tests__/poe.spec.ts +++ b/src/api/providers/__tests__/poe.spec.ts @@ -659,7 +659,7 @@ describe("PoeHandler", () => { await expect(promise).rejects.toMatchObject({ name: "AbortError" }) }) - it("completePrompt should pass reasoning effort for effort-capable models", async () => { + it("passes reasoning effort to streamText via createMessage", async () => { const handler = new PoeHandler({ poeApiKey: "key", apiModelId: "openai/o3", diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index 565793511f..ca0d4d309d 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -624,7 +624,14 @@ describe("RequestyHandler", () => { const handler = new RequestyHandler(mockOptions) const controller = new AbortController() + // Synchronize on request startup (instead of a fixed sleep) so the abort + // deterministically lands while the request is in flight. + let notifyCreateStarted!: () => void + const createStarted = new Promise((resolve) => { + notifyCreateStarted = resolve + }) mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + notifyCreateStarted() // Emulate the OpenAI SDK: the pending request rejects when the signal aborts. await new Promise((resolve) => { if (options?.signal?.aborted) { @@ -642,8 +649,7 @@ describe("RequestyHandler", () => { const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) const nextPromise = generator.next() - // Let the generator reach the pending create() call, then abort. - await new Promise((resolve) => setTimeout(resolve, 10)) + await createStarted controller.abort() await expect(nextPromise).rejects.toMatchObject({ name: "AbortError" })