From d298d4a6f6b835eac62471c010ddfa98bbd1c881 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 22:44:24 +0800 Subject: [PATCH 01/10] feat(api): abort signal support for requesty (completePrompt + shared helpers) --- src/api/providers/__tests__/requesty.spec.ts | 351 ++++++++++++++++-- src/api/providers/requesty.ts | 48 ++- .../utils/__tests__/abort-signal.spec.ts | 96 +++++ src/api/providers/utils/abort-signal.ts | 32 ++ 4 files changed, 501 insertions(+), 26 deletions(-) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index f256e5b1c1..bbad8dbd94 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -9,6 +9,8 @@ const MOCK_TIMEOUT_MS = 300_000 import { Anthropic } from "@anthropic-ai/sdk" import OpenAI from "openai" +import type { ModelRecord } from "@roo-code/types" + import { RequestyHandler } from "../requesty" import { Package } from "../../../shared/package" import { ApiHandlerCreateMessageMetadata } from "../../index" @@ -16,6 +18,33 @@ import { makeApiHandlerOptions } from "../../../test-utils/api" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { clearAllMocks } from "../../../test-utils/reset" +/** + * Stryker guard: fails fast if `promise` does not settle within `ms`. + * + * Stryker's per-mutant cutoff (timeoutMS 5s x timeoutFactor 1.5 ~= 7.5s) is shorter + * than vitest's testTimeout (20s). A mutant that removes a settle call (or an abort + * listener) leaves an awaited promise pending forever; without this guard the test + * would outlive the cutoff and the mutant would be reported as Timeout (inconclusive). + * Settling the guard at 500ms turns those mutants into fast failures (KILLED). + */ +function withSettleGuard(promise: Promise, ms = 500): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new Error(`settle guard timed out after ${ms}ms`)) + }, ms) + void promise.then( + (value) => { + clearTimeout(timer) + resolve(value) + }, + (error) => { + clearTimeout(timer) + reject(error) + }, + ) + }) +} + const mockCreate = vitest.fn() vitest.mock("openai", () => { @@ -613,6 +642,13 @@ describe("RequestyHandler", () => { }) describe("completePrompt", () => { + // The createMessage tests leave behind a persistent stream mock plus queued + // one-shot implementations; reset so each completePrompt test starts from a clean + // mock (its own mockSetup below is authoritative). + beforeEach(() => { + mockCreate.mockReset() + }) + it("returns correct response", async () => { const handler = new RequestyHandler(mockOptions) const mockResponse = { choices: [{ message: { content: "test completion" } }] } @@ -623,12 +659,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 () => { @@ -642,12 +681,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 () => { @@ -661,12 +703,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 () => { @@ -680,12 +725,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 () => { @@ -702,5 +750,260 @@ 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) }), + expect.objectContaining({ + 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", + message: "This operation was aborted", + }) + }) + + it("rejects with AbortError when the signal aborts during model lookup", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + // Model discovery is deferred and never settles: with rejectOnAbort racing the + // lookup, the abort must end the request before the lookup resolves. + let notifyLookupStarted!: () => void + const lookupStarted = new Promise((resolve) => { + notifyLookupStarted = resolve + }) + const deferredModelLookup = new Promise(() => {}) + const { getModels } = await import("../fetchers/modelCache") + vitest.mocked(getModels).mockImplementationOnce(() => { + notifyLookupStarted() + return deferredModelLookup + }) + + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal }) + // Defensive: the fast-fail path may reject before the barrier below settles, + // which would otherwise surface as an unhandled rejection. + void promise.catch(() => {}) + await withSettleGuard(lookupStarted) + controller.abort() + + await expect(withSettleGuard(promise)).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + expect(mockCreate).not.toHaveBeenCalled() + }) + + it("rethrows non-abort model lookup failures from completePrompt", async () => { + const handler = new RequestyHandler(mockOptions) + const { getModels } = await import("../fetchers/modelCache") + const lookupError = new Error("lookup failed") + vitest.mocked(getModels).mockImplementationOnce(() => { + return Promise.reject(lookupError) + }) + + await expect(handler.completePrompt("test prompt")).rejects.toThrow("lookup failed") + }) + + it("normalizes raw AbortError lookup failures to the provider AbortError", async () => { + const handler = new RequestyHandler(mockOptions) + const { getModels } = await import("../fetchers/modelCache") + const rawAbort = new Error("The user aborted a request") + rawAbort.name = "AbortError" + vitest.mocked(getModels).mockImplementationOnce(() => { + return Promise.reject(rawAbort) + }) + + await expect(handler.completePrompt("test prompt")).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + }) + + it("rejects with AbortError when aborted mid-flight", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + // Deterministic synchronization (mirrors the Requesty test): the mock notifies the + // test when the request actually starts, so the abort lands mid-flight (after model + // lookup) instead of winning the race at model discovery on a slow runner. + let notifyCreateStarted!: () => void + const createStarted = new Promise((resolve) => { + notifyCreateStarted = resolve + }) + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + notifyCreateStarted() + // 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 }) + // Defensive: the fast-fail path may reject before the barrier below settles, + // which would otherwise surface as an unhandled rejection. + void promise.catch(() => {}) + // Abort only once create() has actually started (after model lookup). + await withSettleGuard(createStarted) + controller.abort() + + await expect(withSettleGuard(promise)).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + }) + 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("does not return a late result when the response resolves after abort", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + // Emulate the OpenAI SDK: the pending request resolves once the signal aborts, + // i.e. after the caller has already cancelled. + let notifyCreateStarted!: () => void + const createStarted = new Promise((resolve) => { + notifyCreateStarted = resolve + }) + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + notifyCreateStarted() + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve() + } else { + options?.signal?.addEventListener("abort", () => resolve(), { once: true }) + } + }) + return { choices: [{ message: { content: "late" } }] } + }) + + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal }) + // Defensive: the fast-fail path may reject before the barrier below settles, + // which would otherwise surface as an unhandled rejection. + void promise.catch(() => {}) + // Abort while the request is in flight; the resolved response is late and must + // be discarded instead of returned. + await withSettleGuard(createStarted) + controller.abort() + + await expect(withSettleGuard(promise)).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + }) + + it("does not forward a non-positive timeout to the client", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) + + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), {}) + }) + + 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 notifyCreateStarted!: () => void + const createStarted = new Promise((resolve) => { + notifyCreateStarted = resolve + }) + let requestSignal: AbortSignal | undefined + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + notifyCreateStarted() + 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, + }) + // Defensive: the fast-fail path may reject before the barrier below settles, + // which would otherwise surface as an unhandled rejection. + void promise.catch(() => {}) + // Abort only once create() has actually started (after model lookup). + await withSettleGuard(createStarted) + controller.abort() + + await expect(withSettleGuard(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/requesty.ts b/src/api/providers/requesty.ts index 2c1092d303..f77beba9cd 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -23,6 +23,13 @@ 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 { + createAbortError, + isRequestAborted, + mergeAbortSignalAndTimeout, + rejectOnAbort, + throwIfAborted, +} from "./utils/abort-signal" import { applyRouterToolPreferences } from "./utils/router-tool-preferences" import { extractReasoningFromDelta } from "./utils/extract-reasoning" @@ -211,7 +218,25 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan } async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { - const { id: model, maxTokens: max_tokens, temperature } = await this.fetchModel() + // Establish the cancellation scope before model lookup: a pre-aborted call, or + // one aborted while model metadata is loading, must reject promptly instead of + // waiting for the lookup to settle. The configured timeoutMs covers the lookup + // as well. + const requestAbortSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) + throwIfAborted(requestAbortSignal) + + let modelData: Awaited> + try { + modelData = requestAbortSignal + ? await rejectOnAbort(this.fetchModel(), requestAbortSignal, this.providerName) + : await this.fetchModel() + } catch (error) { + if (isRequestAborted(error, requestAbortSignal)) { + throw createAbortError(this.providerName) + } + throw error + } + const { id: model, maxTokens: max_tokens, temperature } = modelData const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [{ role: "system", content: prompt }] @@ -222,12 +247,31 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan temperature: temperature, } + // The merged abort signal (established before model lookup, above) is forwarded to the + // SDK 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 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") + } 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") + } return response.choices[0]?.message.content || "" } } diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index 1692f71e63..c95b800d8c 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -3,9 +3,105 @@ import { isRequestAborted, mergeAbortSignalAndTimeout, mergeAbortSignals, + rejectOnAbort, throwIfAborted, } from "../abort-signal" +/** + * Stryker guard: fails fast if `promise` does not settle within `ms`. + * + * Stryker's per-mutant cutoff (timeoutMS 5s x timeoutFactor 1.5 ~= 7.5s) is shorter + * than vitest's testTimeout (20s). A mutant that removes a settle call (or an abort + * listener) leaves an awaited promise pending forever; without this guard the test + * would outlive the cutoff and the mutant would be reported as Timeout (inconclusive). + * Settling the guard at 500ms turns those mutants into fast failures (KILLED). + */ +function withSettleGuard(promise: Promise, ms = 500): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new Error(`settle guard timed out after ${ms}ms`)) + }, ms) + void promise.then( + (value) => { + clearTimeout(timer) + resolve(value) + }, + (error) => { + clearTimeout(timer) + reject(error) + }, + ) + }) +} + +describe("rejectOnAbort", () => { + it("resolves with the pending value when it settles before the signal aborts", async () => { + const controller = new AbortController() + + await expect( + withSettleGuard(rejectOnAbort(Promise.resolve("done"), controller.signal, "TestProvider")), + ).resolves.toBe("done") + expect(controller.signal.aborted).toBe(false) + }) + + it("rejects with the provider abort error when the signal aborts first", async () => { + const controller = new AbortController() + // Never settles: the race must end purely via the abort. + const pending = new Promise(() => {}) + const race = rejectOnAbort(pending, controller.signal, "TestProvider") + controller.abort() + + await expect(withSettleGuard(race)).rejects.toMatchObject({ + name: "AbortError", + message: "The TestProvider request was aborted", + }) + }) + + it("rejects immediately when the signal is already aborted", async () => { + const controller = new AbortController() + controller.abort() + const pending = new Promise(() => {}) + + await expect(withSettleGuard(rejectOnAbort(pending, controller.signal, "TestProvider"))).rejects.toMatchObject({ + name: "AbortError", + }) + }) + + it("propagates the pending rejection when the signal stays active", async () => { + const controller = new AbortController() + const boom = new Error("lookup failed") + + await expect( + withSettleGuard(rejectOnAbort(Promise.reject(boom), controller.signal, "TestProvider")), + ).rejects.toBe(boom) + }) + + it("detaches the abort listener once the pending settles", async () => { + const controller = new AbortController() + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + + await expect( + withSettleGuard(rejectOnAbort(Promise.resolve("done"), controller.signal, "TestProvider")), + ).resolves.toBe("done") + + expect(removeSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + removeSpy.mockRestore() + }) + + it("detaches the abort listener when the pending rejects", async () => { + const controller = new AbortController() + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + const lookupError = new Error("lookup failed") + + await expect( + withSettleGuard(rejectOnAbort(Promise.reject(lookupError), controller.signal, "TestProvider")), + ).rejects.toBe(lookupError) + + expect(removeSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + removeSpy.mockRestore() + }) +}) + describe("abort-signal utilities", () => { describe("mergeAbortSignalAndTimeout", () => { it("returns undefined when no signal or positive timeout is provided", () => { diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index 26f57c3e9a..bd7d579b00 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -93,3 +93,35 @@ export function createAbortError(providerName: string): Error { abortError.name = "AbortError" return abortError } + +/** + * Await `pending` but reject with the provider's abort error when `signal` + * aborts first. For async phases that have no native signal support (model + * discovery) yet must still settle promptly on cancellation. The underlying + * promise keeps running (its settlement is ignored) — cancellation is + * cooperative at this boundary. + * + * The abort listener is detached once `pending` settles (success or + * failure), so repeated calls on one signal do not accumulate listeners. + */ +export function rejectOnAbort(pending: Promise, signal: AbortSignal, providerName: string): Promise { + if (signal.aborted) { + return Promise.reject(createAbortError(providerName)) + } + + return new Promise((resolve, reject) => { + const onAbort = () => reject(createAbortError(providerName)) + // Stryker disable next-line ObjectLiteral,BooleanLiteral: a signal fires its abort event exactly once and the settle handler removes this listener, so the once flag is unobservable + signal.addEventListener("abort", onAbort, { once: true }) + void pending.then( + (value) => { + signal.removeEventListener("abort", onAbort) + resolve(value) + }, + (error) => { + signal.removeEventListener("abort", onAbort) + reject(error) + }, + ) + }) +} From ebd5404df6c5b415e2507211c74d1aa810a4513e Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sun, 6 Sep 2026 01:04:58 +0800 Subject: [PATCH 02/10] test(api): strengthen requesty abort assertions and share settle guard helper --- src/api/providers/__tests__/requesty.spec.ts | 61 ++++++++----------- src/api/providers/requesty.ts | 4 +- .../utils/__tests__/abort-signal.spec.ts | 51 +++++++--------- src/test-utils/settle-guard.ts | 26 ++++++++ 4 files changed, 77 insertions(+), 65 deletions(-) create mode 100644 src/test-utils/settle-guard.ts diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index bbad8dbd94..4a3faeb34a 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -17,33 +17,7 @@ import { ApiHandlerCreateMessageMetadata } from "../../index" import { makeApiHandlerOptions } from "../../../test-utils/api" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { clearAllMocks } from "../../../test-utils/reset" - -/** - * Stryker guard: fails fast if `promise` does not settle within `ms`. - * - * Stryker's per-mutant cutoff (timeoutMS 5s x timeoutFactor 1.5 ~= 7.5s) is shorter - * than vitest's testTimeout (20s). A mutant that removes a settle call (or an abort - * listener) leaves an awaited promise pending forever; without this guard the test - * would outlive the cutoff and the mutant would be reported as Timeout (inconclusive). - * Settling the guard at 500ms turns those mutants into fast failures (KILLED). - */ -function withSettleGuard(promise: Promise, ms = 500): Promise { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - reject(new Error(`settle guard timed out after ${ms}ms`)) - }, ms) - void promise.then( - (value) => { - clearTimeout(timer) - resolve(value) - }, - (error) => { - clearTimeout(timer) - reject(error) - }, - ) - }) -} +import { withSettleGuard } from "../../../test-utils/settle-guard" const mockCreate = vitest.fn() @@ -765,13 +739,31 @@ describe("RequestyHandler", () => { 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) }), - expect.objectContaining({ - timeout: 5000, - }), - ) + // Capture the exact timeout signal instance the provider creates so the + // test can assert identity, not just type, for the signal it forwards. + const timeoutSignalSpy = vitest.spyOn(AbortSignal, "timeout") + try { + await handler.completePrompt("test prompt", { timeoutMs: 5000 }) + // Assert the factory argument and call count too: the identity check + // below would also pass for a signal created with the wrong duration. + expect(timeoutSignalSpy).toHaveBeenCalledTimes(1) + expect(timeoutSignalSpy).toHaveBeenCalledWith(5000) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 5000 }), + ) + // Without a caller signal the merged signal is exactly the + // AbortSignal.timeout instance; the SDK relies on it to reject with a + // DOM-standard AbortError when the timeout fires. + const clientOptions = mockCreate.mock.calls[0]?.[1] as { signal?: AbortSignal } | undefined + const expectedSignal = timeoutSignalSpy.mock.results[0]?.value + expect(expectedSignal).toBeInstanceOf(AbortSignal) + expect(clientOptions?.signal).toBe(expectedSignal) + } finally { + // Restore on every exit path: a failed assertion above must not leave + // the static spy installed for the remaining tests in this file. + timeoutSignalSpy.mockRestore() + } }) it("should work without options (backward compatible)", async () => { @@ -795,6 +787,7 @@ describe("RequestyHandler", () => { name: "AbortError", message: "This operation was aborted", }) + expect(mockCreate).not.toHaveBeenCalled() }) it("rejects with AbortError when the signal aborts during model lookup", async () => { diff --git a/src/api/providers/requesty.ts b/src/api/providers/requesty.ts index f77beba9cd..2249ca21b2 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -263,14 +263,14 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan // 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") + throw createAbortError(this.providerName) } 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") + throw createAbortError(this.providerName) } return response.choices[0]?.message.content || "" } diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index c95b800d8c..04d657b564 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -6,33 +6,7 @@ import { rejectOnAbort, throwIfAborted, } from "../abort-signal" - -/** - * Stryker guard: fails fast if `promise` does not settle within `ms`. - * - * Stryker's per-mutant cutoff (timeoutMS 5s x timeoutFactor 1.5 ~= 7.5s) is shorter - * than vitest's testTimeout (20s). A mutant that removes a settle call (or an abort - * listener) leaves an awaited promise pending forever; without this guard the test - * would outlive the cutoff and the mutant would be reported as Timeout (inconclusive). - * Settling the guard at 500ms turns those mutants into fast failures (KILLED). - */ -function withSettleGuard(promise: Promise, ms = 500): Promise { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - reject(new Error(`settle guard timed out after ${ms}ms`)) - }, ms) - void promise.then( - (value) => { - clearTimeout(timer) - resolve(value) - }, - (error) => { - clearTimeout(timer) - reject(error) - }, - ) - }) -} +import { withSettleGuard } from "../../../../test-utils/settle-guard" describe("rejectOnAbort", () => { it("resolves with the pending value when it settles before the signal aborts", async () => { @@ -64,6 +38,7 @@ describe("rejectOnAbort", () => { await expect(withSettleGuard(rejectOnAbort(pending, controller.signal, "TestProvider"))).rejects.toMatchObject({ name: "AbortError", + message: "The TestProvider request was aborted", }) }) @@ -78,18 +53,28 @@ describe("rejectOnAbort", () => { it("detaches the abort listener once the pending settles", async () => { const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, "addEventListener") const removeSpy = vi.spyOn(controller.signal, "removeEventListener") await expect( withSettleGuard(rejectOnAbort(Promise.resolve("done"), controller.signal, "TestProvider")), ).resolves.toBe("done") - expect(removeSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + // The settle path must remove the exact listener that was registered, not just any + // function: removing a different reference would leave the original abort listener + // attached to the signal. First require the registration to have happened at all, + // so a missing registration cannot silently degrade to an undefined comparison. + expect(addSpy).toHaveBeenCalledTimes(1) + const registeredListener = addSpy.mock.calls[0]?.[1] as EventListener | undefined + expect(typeof registeredListener).toBe("function") + expect(removeSpy).toHaveBeenCalledWith("abort", registeredListener) + addSpy.mockRestore() removeSpy.mockRestore() }) it("detaches the abort listener when the pending rejects", async () => { const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, "addEventListener") const removeSpy = vi.spyOn(controller.signal, "removeEventListener") const lookupError = new Error("lookup failed") @@ -97,7 +82,15 @@ describe("rejectOnAbort", () => { withSettleGuard(rejectOnAbort(Promise.reject(lookupError), controller.signal, "TestProvider")), ).rejects.toBe(lookupError) - expect(removeSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + // The settle path must remove the exact listener that was registered, not just any + // function: removing a different reference would leave the original abort listener + // attached to the signal. First require the registration to have happened at all, + // so a missing registration cannot silently degrade to an undefined comparison. + expect(addSpy).toHaveBeenCalledTimes(1) + const registeredListener = addSpy.mock.calls[0]?.[1] as EventListener | undefined + expect(typeof registeredListener).toBe("function") + expect(removeSpy).toHaveBeenCalledWith("abort", registeredListener) + addSpy.mockRestore() removeSpy.mockRestore() }) }) diff --git a/src/test-utils/settle-guard.ts b/src/test-utils/settle-guard.ts new file mode 100644 index 0000000000..1efe5286c8 --- /dev/null +++ b/src/test-utils/settle-guard.ts @@ -0,0 +1,26 @@ +/** + * Stryker guard: fails fast if `promise` does not settle within `ms`. + * + * Stryker's per-mutant cutoff (timeoutMS 5s x timeoutFactor 1.5 ~= 7.5s) is shorter + * than vitest's testTimeout (20s). A mutant that removes a settle call (or an abort + * listener) leaves an awaited promise pending forever; without this guard the test + * would outlive the cutoff and the mutant would be reported as Timeout (inconclusive). + * Settling the guard at 500ms turns those mutants into fast failures (KILLED). + */ +export function withSettleGuard(promise: Promise, ms = 500): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new Error(`settle guard timed out after ${ms}ms`)) + }, ms) + void promise.then( + (value) => { + clearTimeout(timer) + resolve(value) + }, + (error) => { + clearTimeout(timer) + reject(error) + }, + ) + }) +} From f64950885415e52745af9cf91c2ccd5e84810a51 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 22:47:53 +0800 Subject: [PATCH 03/10] feat(api): abort signal support for requesty (createMessage + kill tests) --- src/api/providers/__tests__/requesty.spec.ts | 423 ++++++++++++++++++- src/api/providers/requesty.ts | 189 ++++++--- 2 files changed, 549 insertions(+), 63 deletions(-) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index 4a3faeb34a..e37f6b7f24 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -14,7 +14,7 @@ import type { ModelRecord } from "@roo-code/types" 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" import { withSettleGuard } from "../../../test-utils/settle-guard" @@ -262,9 +262,158 @@ describe("RequestyHandler", () => { stream_options: { include_usage: true }, temperature: 0, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) + it("forwards the settings reasoningEffort to the Requesty request", async () => { + const { getModels } = await import("../fetchers/modelCache") + vitest.mocked(getModels).mockImplementationOnce(async () => ({ + "coding/claude-4-sonnet": { + maxTokens: 8192, + contextWindow: 200000, + supportsImages: true, + supportsPromptCache: true, + supportsReasoningEffort: true, + inputPrice: 3, + outputPrice: 15, + description: "Claude 4 Sonnet", + }, + })) + mockCreate.mockResolvedValue(asyncStreamFrom([{ id: "c1", choices: [{ delta: { content: "ok" } }] }])) + const handler = new RequestyHandler( + makeApiHandlerOptions({ + requestyApiKey: "test-key", + requestyModelId: "coding/claude-4-sonnet", + reasoningEffort: "high", + }), + ) + + const chunks = await collectStream( + handler.createMessage("test system", [{ role: "user", content: "test" }]), + ) + expect(chunks).toHaveLength(1) + expect(mockCreate).toHaveBeenCalledTimes(1) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: "coding/claude-4-sonnet", reasoning_effort: "high" }), + expect.anything(), + ) + }) + + it("omits reasoning_effort when the settings effort is outside the model's supported set", async () => { + const { getModels } = await import("../fetchers/modelCache") + vitest.mocked(getModels).mockImplementationOnce(async () => ({ + "coding/claude-4-sonnet": { + maxTokens: 8192, + contextWindow: 200000, + supportsImages: true, + supportsPromptCache: true, + supportsReasoningEffort: ["low", "medium", "high"], + inputPrice: 3, + outputPrice: 15, + description: "Claude 4 Sonnet", + }, + })) + mockCreate.mockResolvedValue(asyncStreamFrom([{ id: "c1", choices: [{ delta: { content: "ok" } }] }])) + const handler = new RequestyHandler( + makeApiHandlerOptions({ + requestyApiKey: "test-key", + requestyModelId: "coding/claude-4-sonnet", + reasoningEffort: "minimal", + }), + ) + + const chunks = await collectStream( + handler.createMessage("test system", [{ role: "user", content: "test" }]), + ) + expect(chunks).toHaveLength(1) + expect(mockCreate).toHaveBeenCalledTimes(1) + // "minimal" is outside ["low", "medium", "high"], so the key must be absent entirely. + expect(mockCreate.mock.calls[0][0]).not.toHaveProperty("reasoning_effort") + }) + + it("forwards the task metadata into the requesty-specific request block", async () => { + mockCreate.mockResolvedValue(asyncStreamFrom([{ id: "c1", choices: [{ delta: { content: "ok" } }] }])) + const handler = new RequestyHandler(mockOptions) + + const chunks = await collectStream( + handler.createMessage("test system", [{ role: "user", content: "test" }], { + taskId: "task-123", + mode: "plan", + }), + ) + expect(chunks).toHaveLength(1) + expect(mockCreate).toHaveBeenCalledTimes(1) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ + requesty: { trace_id: "task-123", extra: { mode: "plan" } }, + }), + expect.anything(), + ) + }) + + it("tolerates chunks with empty choices before the first delta", async () => { + mockCreate.mockResolvedValue( + asyncStreamFrom([ + { id: "c1", choices: [] }, + { id: "c2", choices: [{ delta: { content: "after empty" } }] }, + ]), + ) + const handler = new RequestyHandler(mockOptions) + + const chunks = await collectStream( + handler.createMessage("test system", [{ role: "user", content: "test" }]), + ) + expect(chunks).toEqual([{ type: "text", text: "after empty" }]) + }) + + it("streams tool_call_partial chunks when the tool call has no function payload", async () => { + mockCreate.mockResolvedValue( + asyncStreamFrom([ + { + id: "c1", + choices: [{ delta: { tool_calls: [{ index: 0, id: "call_123" }] } }], + }, + ]), + ) + const handler = new RequestyHandler(mockOptions) + + const chunks = await collectStream( + handler.createMessage("test system", [{ role: "user", content: "test" }]), + ) + expect(chunks).toEqual([{ type: "tool_call_partial", index: 0, id: "call_123" }]) + }) + + it("emits the usage chunk once when a final chunk carries no usage", async () => { + mockCreate.mockResolvedValue( + asyncStreamFrom([ + { id: "c1", choices: [{ delta: { content: "text" } }] }, + { id: "c2", choices: [{ delta: {} }], usage: { prompt_tokens: 3, completion_tokens: 4 } }, + { id: "c3", choices: [{ delta: {} }] }, + ]), + ) + const handler = new RequestyHandler(mockOptions) + + const chunks = await collectStream( + handler.createMessage("test system", [{ role: "user", content: "test" }]), + ) + expect(chunks).toHaveLength(2) + expect(chunks[0]).toEqual({ type: "text", text: "text" }) + expect(chunks[1]).toMatchObject({ type: "usage", inputTokens: 3, outputTokens: 4 }) + }) + + it("does not emit a usage chunk when the stream reports no usage", async () => { + mockCreate.mockResolvedValue( + asyncStreamFrom([{ id: "c1", choices: [{ delta: { content: "only text" } }] }]), + ) + const handler = new RequestyHandler(mockOptions) + + const chunks = await collectStream( + handler.createMessage("test system", [{ role: "user", content: "test" }]), + ) + expect(chunks).toEqual([{ type: "text", text: "only text" }]) + }) + it("uses adaptive thinking for Claude Fable 5 when reasoning is enabled", async () => { const handler = new RequestyHandler( makeApiHandlerOptions({ @@ -295,6 +444,7 @@ describe("RequestyHandler", () => { thinking: { type: "adaptive" }, temperature: undefined, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -328,6 +478,7 @@ describe("RequestyHandler", () => { thinking: { type: "adaptive" }, temperature: undefined, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -361,6 +512,7 @@ describe("RequestyHandler", () => { thinking: { type: "adaptive" }, temperature: undefined, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -394,6 +546,7 @@ describe("RequestyHandler", () => { thinking: { type: "adaptive" }, temperature: undefined, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -532,6 +685,7 @@ describe("RequestyHandler", () => { ]), tool_choice: "auto", }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -613,6 +767,273 @@ 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", + message: "The Requesty request was aborted", + }) + // The fast-fail guard must reject before model discovery starts. + const { getModels } = await import("../fetchers/modelCache") + expect(vitest.mocked(getModels)).not.toHaveBeenCalled() + }) + + it("rejects with AbortError when the external signal aborts during deferred model discovery", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + // Model discovery is deferred: capture the resolver and settle it only at the end of + // the test, so the abort deterministically lands while the lookup is still pending. + // The barrier below (instead of a fixed sleep) synchronizes on the lookup starting. + let resolveModelLookup!: (models: ModelRecord) => void + const deferredModelLookup = new Promise((resolve) => { + resolveModelLookup = resolve + }) + let notifyLookupStarted!: () => void + const lookupStarted = new Promise((resolve) => { + notifyLookupStarted = resolve + }) + const { getModels } = await import("../fetchers/modelCache") + vitest.mocked(getModels).mockImplementationOnce(() => { + notifyLookupStarted() + return deferredModelLookup + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + const nextPromise = generator.next() + // Defensive: the fast-fail path may reject before the barrier below settles, + // which would otherwise surface as an unhandled rejection. + void nextPromise.catch(() => {}) + await withSettleGuard(lookupStarted) + controller.abort() + + await expect(withSettleGuard(nextPromise)).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + expect(mockCreate).not.toHaveBeenCalled() + + // Settle the abandoned lookup so it cannot outlive the test. + resolveModelLookup({}) + }) + + 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) => { + expect(requestSignal).toBeDefined() + 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(withSettleGuard(iteration)).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + expect(chunks).toContainEqual({ type: "text", text: "first" }) + }) + it("rejects with AbortError when the stream ends normally after a mid-stream abort (swallowed AbortError)", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + // Simulate openai@5.23.2: the SDK stream iterator swallows the mid-stream + // AbortError and returns normally instead of throwing, so the catch in + // createMessage never runs. The per-request signal (second argument) is the + // one the SDK observes. + let requestSignal: AbortSignal | undefined + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + requestSignal = options?.signal + return (async function* () { + yield { id: "1", choices: [{ delta: { content: "partial" } }] } + // Wait for the abort instead of polling: the iterator ends gracefully + // (no throw) once the request signal aborts. + await new Promise((resolve) => { + expect(requestSignal).toBeDefined() + if (requestSignal!.aborted) { + resolve() + } else { + requestSignal!.addEventListener("abort", () => resolve(), { once: true }) + } + }) + })() + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + const first = await generator.next() + expect(first.value).toEqual({ type: "text", text: "partial" }) + // Abort mid-stream, after the first chunk has been yielded. + controller.abort() + + // The stream ended normally, but createMessage must still reject with AbortError. + await expect(withSettleGuard(generator.next())).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + }) + it("does not emit buffered chunks after a mid-stream abort (iterator keeps delivering)", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + + // Simulate openai@5.23.2 delivering a buffered chunk after the abort has already + // fired, then ending the iterator normally (no throw). + let requestSignal: AbortSignal | undefined + mockCreate.mockImplementationOnce(async (_params: unknown, options?: { signal?: AbortSignal }) => { + requestSignal = options?.signal + return (async function* () { + yield { id: "1", choices: [{ delta: { content: "partial" } }] } + // Wait for the abort instead of polling: the buffered chunk is delivered + // once the request signal aborts. + await new Promise((resolve) => { + expect(requestSignal).toBeDefined() + if (requestSignal!.aborted) { + resolve() + } else { + requestSignal!.addEventListener("abort", () => resolve(), { once: true }) + } + }) + yield { id: "2", choices: [{ delta: { content: "after-abort" } }] } + })() + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + const first = await generator.next() + expect(first.value).toEqual({ type: "text", text: "partial" }) + // Abort mid-stream, after the first chunk has been yielded. + controller.abort() + + // The buffered second chunk must not be emitted, and the generator must reject + // with the provider AbortError. + await expect(withSettleGuard(generator.next())).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + }) + it("rejects with AbortError when the external signal aborts during request creation", async () => { + 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) { + 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() + // Defensive: the fast-fail path may reject before the barrier below settles, + // which would otherwise surface as an unhandled rejection. + void nextPromise.catch(() => {}) + await withSettleGuard(createStarted) + controller.abort() + + await expect(withSettleGuard(nextPromise)).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + }) + + it("rethrows non-abort creation errors from createMessage", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockImplementationOnce(async () => { + throw new Error("boom") + }) + + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }]) + + await expect(collectStream(generator)).rejects.toThrow("boom") + }) + + it("removes the external abort listener when the stream completes", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + mockCreate.mockImplementationOnce(async () => { + return asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "done" } }] }]) + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + await collectStream(generator) + + expect(removeSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + removeSpy.mockRestore() + }) + + 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", () => { diff --git a/src/api/providers/requesty.ts b/src/api/providers/requesty.ts index 2249ca21b2..a917306d30 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -140,80 +140,145 @@ 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() + // Stryker disable next-line ObjectLiteral,BooleanLiteral: a signal fires its abort event exactly once, and the finally block removes this listener explicitly, so the once flag is unobservable + 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") + } - const reasoningText = extractReasoningFromDelta(delta) - if (reasoningText) { - yield { type: "reasoning", text: reasoningText } + // Model discovery is not signal-aware: race it against the per-request signal so an + // abort during the lookup rejects with AbortError instead of calling the API with an + // already-aborted signal. + const { + id: model, + info, + maxTokens: max_tokens, + temperature, + reasoningEffort: reasoning_effort, + reasoning: thinking, + } = await rejectOnAbort(this.fetchModel(), controller.signal, this.providerName) + + 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, } - if (delta?.content) { - yield { type: "text", text: delta.content } + 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") + } + throw handleOpenAIError(error, this.providerName) } + try { + let lastUsage: any = undefined + + for await (const chunk of stream) { + // The iterator can keep delivering buffered chunks after the abort has already + // fired (openai@5.23.2 swallows the mid-stream AbortError), so re-check the + // signal before processing each chunk. The yields below are synchronous (there + // is no await between this check and them), so nothing is emitted once the + // signal aborts. + if (controller.signal.aborted) { + break + } + + const delta = chunk.choices[0]?.delta - // 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, + // Yield reasoning chunks before content chunks so consumers see them in model order. + const reasoningText = extractReasoningFromDelta(delta) + if (reasoningText) { + yield { type: "reasoning", text: reasoningText } + } + + 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, + } + } + } + + if (chunk.usage) { + lastUsage = chunk.usage } } - } - if (chunk.usage) { - lastUsage = chunk.usage - } - } + // openai@5.23.2's stream iterator swallows a mid-stream AbortError and returns + // normally instead of throwing, so the catch below would never run: without this + // check, createMessage completes silently after yielding partial output. + if (controller.signal.aborted) { + throw createAbortError(this.providerName) + } - if (lastUsage) { - yield this.processUsageMetrics(lastUsage, info) + 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") + } + throw error + } + } finally { + removeExternalAbortListener?.() } } From 316214451bf98723612d3dd04b7606b7b387f2b9 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sun, 6 Sep 2026 01:10:55 +0800 Subject: [PATCH 04/10] test(api): assert requesty createMessage listener removal identity --- src/api/providers/__tests__/requesty.spec.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index e37f6b7f24..1b255ae454 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -1007,6 +1007,7 @@ describe("RequestyHandler", () => { it("removes the external abort listener when the stream completes", async () => { const handler = new RequestyHandler(mockOptions) const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, "addEventListener") const removeSpy = vi.spyOn(controller.signal, "removeEventListener") mockCreate.mockImplementationOnce(async () => { return asyncStreamFrom([{ id: "1", choices: [{ delta: { content: "done" } }] }]) @@ -1017,7 +1018,12 @@ describe("RequestyHandler", () => { await collectStream(generator) - expect(removeSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + // The cleanup path must remove the exact listener that was registered, not just any + // function: removing a different reference would leave the original abort listener + // attached to the signal. + const registeredListener = addSpy.mock.calls[0]?.[1] as EventListener | undefined + expect(removeSpy).toHaveBeenCalledWith("abort", registeredListener) + addSpy.mockRestore() removeSpy.mockRestore() }) From 64ccc46eba259677da782a04ac667d08891beae0 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sun, 6 Sep 2026 02:12:26 +0800 Subject: [PATCH 05/10] refactor(api): type requesty createMessage usage and strengthen listener tests --- src/api/providers/__tests__/requesty.spec.ts | 5 ++++- src/api/providers/requesty.ts | 6 ++---- src/eslint-suppressions.json | 2 +- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index 1b255ae454..c7745bc317 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -1020,8 +1020,11 @@ describe("RequestyHandler", () => { // The cleanup path must remove the exact listener that was registered, not just any // function: removing a different reference would leave the original abort listener - // attached to the signal. + // attached to the signal. First require the registration to have happened at all, + // so a missing registration cannot silently degrade to an undefined comparison. + expect(addSpy).toHaveBeenCalledTimes(1) const registeredListener = addSpy.mock.calls[0]?.[1] as EventListener | undefined + expect(typeof registeredListener).toBe("function") expect(removeSpy).toHaveBeenCalledWith("abort", registeredListener) addSpy.mockRestore() removeSpy.mockRestore() diff --git a/src/api/providers/requesty.ts b/src/api/providers/requesty.ts index a917306d30..2d985bbe98 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -186,9 +186,7 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan ] // 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 allowedEffort = (["low", "medium", "high"] as const).find((effort) => effort === reasoning_effort) const completionParams: RequestyChatCompletionParamsStreaming = { messages: openAiMessages, @@ -217,7 +215,7 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan throw handleOpenAIError(error, this.providerName) } try { - let lastUsage: any = undefined + let lastUsage: RequestyUsage | undefined = undefined for await (const chunk of stream) { // The iterator can keep delivering buffered chunks after the abort has already diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index 393e108645..2cd198b77f 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -421,7 +421,7 @@ }, "api/providers/requesty.ts": { "@typescript-eslint/no-explicit-any": { - "count": 3 + "count": 1 } }, "api/providers/unbound.ts": { From 7c8e729ba84d0577ae47bddde496f659c192a6ab Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sun, 6 Sep 2026 02:17:22 +0800 Subject: [PATCH 06/10] test(api): assert requesty per-request signal identity and failure-path listener cleanup --- src/api/providers/__tests__/requesty.spec.ts | 28 ++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index c7745bc317..db0f1bc92a 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -871,6 +871,10 @@ describe("RequestyHandler", () => { message: "The Requesty request was aborted", }) expect(chunks).toContainEqual({ type: "text", text: "first" }) + // The provider must forward its own per-request signal, not the caller's signal: + // the per-request controller isolates this request from the external signal. + expect(requestSignal).toBeInstanceOf(AbortSignal) + expect(requestSignal).not.toBe(controller.signal) }) it("rejects with AbortError when the stream ends normally after a mid-stream abort (swallowed AbortError)", async () => { const handler = new RequestyHandler(mockOptions) @@ -1030,6 +1034,30 @@ describe("RequestyHandler", () => { removeSpy.mockRestore() }) + it("removes the external abort listener when the stream fails without an abort", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, "addEventListener") + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + mockCreate.mockImplementationOnce(async () => { + throw new Error("boom") + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + await expect(collectStream(generator)).rejects.toThrow("boom") + + // The failure path must clean up just like the success path: the exact registered + // listener is removed even when the request rejects without any abort firing. + expect(addSpy).toHaveBeenCalledTimes(1) + const registeredListener = addSpy.mock.calls[0]?.[1] as EventListener | undefined + expect(typeof registeredListener).toBe("function") + expect(removeSpy).toHaveBeenCalledWith("abort", registeredListener) + addSpy.mockRestore() + removeSpy.mockRestore() + }) + it("rethrows non-abort stream errors from createMessage", async () => { const handler = new RequestyHandler(mockOptions) mockCreate.mockImplementationOnce(async () => { From 23f386e3461014bbdd65cd39b3f3a29a7e9de01e Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Mon, 7 Sep 2026 22:46:24 +0800 Subject: [PATCH 07/10] fix(api): thread cancellation signal and bounded timeout into Requesty model discovery CodeRabbit finding: the changed cancellation path could abandon an unbounded model-discovery request. rejectOnAbort() is cooperative at that boundary - the underlying fetch keeps running after abort and has no timeout. Thread the per-request signal from createMessage and the merged signal from completePrompt through fetchModel -> getModels -> getRequestyModels, and bound the fetcher's axios models request with a 10_000 ms timeout (matching the in-tree axios fetchers). The rejectOnAbort races remain as a second line of defence for the window in which the fetcher swallows the cancellation and resolves with an empty list. Covered at the narrowest layers: fetcher spec (signal + timeout reach the axios call), modelCache spec (options.signal forwards to the fetcher), provider spec (the exact per-request/merged signal instance reaches discovery). --- src/api/providers/__tests__/requesty.spec.ts | 63 +++++++++++++++++++ .../fetchers/__tests__/modelCache.spec.ts | 30 ++++++++- .../fetchers/__tests__/requesty.spec.ts | 25 ++++++++ src/api/providers/fetchers/modelCache.ts | 5 +- src/api/providers/fetchers/requesty.ts | 17 ++++- src/api/providers/requesty.ts | 15 ++--- src/shared/api.ts | 6 +- 7 files changed, 147 insertions(+), 14 deletions(-) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index db0f1bc92a..0914509eac 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -827,6 +827,40 @@ describe("RequestyHandler", () => { resolveModelLookup({}) }) + it("threads the per-request signal into model discovery", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockResolvedValue(asyncStreamFrom([{ id: "c1", choices: [{ delta: { content: "ok" } }] }])) + + let discoverySignal: AbortSignal | undefined + const { getModels } = await import("../fetchers/modelCache") + vitest.mocked(getModels).mockImplementationOnce((options) => { + discoverySignal = options.signal + return Promise.resolve({ + "coding/claude-4-sonnet": { + maxTokens: 8192, + contextWindow: 200000, + supportsImages: true, + supportsPromptCache: true, + inputPrice: 3, + outputPrice: 15, + description: "Claude 4 Sonnet", + }, + }) + }) + + const controller = new AbortController() + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + + await collectStream(handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata)) + + expect(discoverySignal).toBeInstanceOf(AbortSignal) + // Discovery must receive the same per-request signal that is forwarded to the + // SDK, so a single abort cancels both the lookup and the completion request. + const sdkOptions = mockCreate.mock.calls[0]?.[1] as { signal?: AbortSignal } | undefined + expect(sdkOptions?.signal).toBeInstanceOf(AbortSignal) + expect(discoverySignal).toBe(sdkOptions?.signal) + }) + 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() @@ -1193,6 +1227,35 @@ describe("RequestyHandler", () => { }) }) + it("threads the merged abort signal into model discovery", async () => { + const handler = new RequestyHandler(mockOptions) + mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) + + let discoverySignal: AbortSignal | undefined + const { getModels } = await import("../fetchers/modelCache") + vitest.mocked(getModels).mockImplementationOnce((options) => { + discoverySignal = options.signal + return Promise.resolve({ + "coding/claude-4-sonnet": { + maxTokens: 8192, + contextWindow: 200000, + supportsImages: true, + supportsPromptCache: true, + inputPrice: 3, + outputPrice: 15, + description: "Claude 4 Sonnet", + }, + }) + }) + + const controller = new AbortController() + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + + // Without a timeout the merged signal is the external signal itself, so the + // lookup receives exactly the caller's signal (identity, not just type). + expect(discoverySignal).toBe(controller.signal) + }) + it("should pass timeout through to client", async () => { const handler = new RequestyHandler(mockOptions) mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) diff --git a/src/api/providers/fetchers/__tests__/modelCache.spec.ts b/src/api/providers/fetchers/__tests__/modelCache.spec.ts index 108aa1827b..3d69642d97 100644 --- a/src/api/providers/fetchers/__tests__/modelCache.spec.ts +++ b/src/api/providers/fetchers/__tests__/modelCache.spec.ts @@ -158,7 +158,29 @@ describe("getModels with new GetModelsOptions", () => { const result = await getModels({ provider: providerIdentifiers.requesty, apiKey: DUMMY_REQUESTY_KEY }) - expect(mockGetRequestyModels).toHaveBeenCalledWith(undefined, DUMMY_REQUESTY_KEY) + expect(mockGetRequestyModels).toHaveBeenCalledWith(undefined, DUMMY_REQUESTY_KEY, undefined) + expect(result).toEqual(mockModels) + }) + + it("forwards the caller's cancellation signal to Requesty model discovery", async () => { + const mockModels = { + "requesty/model": { + maxTokens: 4096, + contextWindow: 8192, + supportsPromptCache: false, + description: "Requesty model", + }, + } + mockGetRequestyModels.mockResolvedValue(mockModels) + + const controller = new AbortController() + const result = await getModels({ + provider: providerIdentifiers.requesty, + apiKey: DUMMY_REQUESTY_KEY, + signal: controller.signal, + }) + + expect(mockGetRequestyModels).toHaveBeenCalledWith(undefined, DUMMY_REQUESTY_KEY, controller.signal) expect(result).toEqual(mockModels) }) @@ -179,7 +201,11 @@ describe("getModels with new GetModelsOptions", () => { baseUrl: "https://router.requesty.ai/v1", }) - expect(mockGetRequestyModels).toHaveBeenCalledWith("https://router.requesty.ai/v1", DUMMY_REQUESTY_KEY) + expect(mockGetRequestyModels).toHaveBeenCalledWith( + "https://router.requesty.ai/v1", + DUMMY_REQUESTY_KEY, + undefined, + ) expect(result).toEqual(mockModels) }) diff --git a/src/api/providers/fetchers/__tests__/requesty.spec.ts b/src/api/providers/fetchers/__tests__/requesty.spec.ts index 53891273a9..58bccb67af 100644 --- a/src/api/providers/fetchers/__tests__/requesty.spec.ts +++ b/src/api/providers/fetchers/__tests__/requesty.spec.ts @@ -140,4 +140,29 @@ describe("getRequestyModels", () => { expect(sonnet.supportsReasoningBinary).toBeUndefined() expect(sonnet.supportsTemperature).toBeUndefined() }) + + it("threads the cancellation signal and the bounded timeout into the models request", async () => { + const controller = new AbortController() + mockAxiosGet.mockResolvedValueOnce({ data: { data: [] } }) + + await getRequestyModels(undefined, undefined, controller.signal) + + // The shared axios mock accumulates calls across this file's tests, so assert on + // the call this test just made (the last one) rather than a global call count. + const calls = mockAxiosGet.mock.calls + const config = calls[calls.length - 1]?.[1] + expect(config?.signal).toBe(controller.signal) + expect(config?.timeout).toBe(10_000) + }) + + it("applies the bounded timeout without a signal key when no signal is provided", async () => { + mockAxiosGet.mockResolvedValueOnce({ data: { data: [] } }) + + await getRequestyModels() + + const calls = mockAxiosGet.mock.calls + const config = calls[calls.length - 1]?.[1] + expect(config?.signal).toBeUndefined() + expect(config?.timeout).toBe(10_000) + }) }) diff --git a/src/api/providers/fetchers/modelCache.ts b/src/api/providers/fetchers/modelCache.ts index 50dbe12f6e..73e912d120 100644 --- a/src/api/providers/fetchers/modelCache.ts +++ b/src/api/providers/fetchers/modelCache.ts @@ -232,7 +232,10 @@ async function fetchModelsFromProvider(options: GetModelsOptions): Promise> { +// Bounded wall-clock limit for the models discovery request. Without it a hung connection +// would keep the request alive indefinitely; 10_000 ms matches the other in-tree axios +// fetchers (kenari, opencode-go, nanogpt). +const REQUESTY_MODELS_TIMEOUT_MS = 10_000 + +export async function getRequestyModels( + baseUrl?: string, + apiKey?: string, + signal?: AbortSignal, +): Promise> { const models: Record = {} try { @@ -18,7 +27,11 @@ export async function getRequestyModels(baseUrl?: string, apiKey?: string): Prom const resolvedBaseUrl = toRequestyServiceUrl(baseUrl) const modelsUrl = new URL("v1/models", resolvedBaseUrl) - const response = await axios.get(modelsUrl.toString(), { headers }) + const response = await axios.get(modelsUrl.toString(), { + headers, + ...(signal && { signal }), + timeout: REQUESTY_MODELS_TIMEOUT_MS, + }) const rawModels = response.data.data for (const rawModel of rawModels) { diff --git a/src/api/providers/requesty.ts b/src/api/providers/requesty.ts index 2d985bbe98..f71b217afb 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -86,8 +86,8 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan }) } - public async fetchModel() { - this.models = await getModels({ provider: providerIdentifiers.requesty, baseUrl: this.baseURL }) + public async fetchModel(signal?: AbortSignal) { + this.models = await getModels({ provider: providerIdentifiers.requesty, baseUrl: this.baseURL, signal }) return this.getModel() } @@ -168,9 +168,10 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan throw createAbortError("Requesty") } - // Model discovery is not signal-aware: race it against the per-request signal so an - // abort during the lookup rejects with AbortError instead of calling the API with an - // already-aborted signal. + // Model discovery is signal-aware: the per-request signal is threaded into the lookup so + // the underlying models request is cancelled on abort. The race below remains as a + // second line of defence for the window in which the fetcher swallows the cancellation + // and resolves with an empty model list. const { id: model, info, @@ -178,7 +179,7 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan temperature, reasoningEffort: reasoning_effort, reasoning: thinking, - } = await rejectOnAbort(this.fetchModel(), controller.signal, this.providerName) + } = await rejectOnAbort(this.fetchModel(controller.signal), controller.signal, this.providerName) const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ { role: "system", content: systemPrompt }, @@ -291,7 +292,7 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan let modelData: Awaited> try { modelData = requestAbortSignal - ? await rejectOnAbort(this.fetchModel(), requestAbortSignal, this.providerName) + ? await rejectOnAbort(this.fetchModel(requestAbortSignal), requestAbortSignal, this.providerName) : await this.fetchModel() } catch (error) { if (isRequestAborted(error, requestAbortSignal)) { diff --git a/src/shared/api.ts b/src/shared/api.ts index d1cc21cad0..6c3618ef94 100644 --- a/src/shared/api.ts +++ b/src/shared/api.ts @@ -166,11 +166,13 @@ export const getModelMaxOutputTokens = ({ // GetModelsOptions -// Allow callers to always pass apiKey/baseUrl without excess property errors, -// while still enforcing required fields per provider where applicable. +// Allow callers to always pass apiKey/baseUrl (plus an optional cancellation signal for +// signal-aware fetchers) without excess property errors, while still enforcing required +// fields per provider where applicable. type CommonFetchParams = { apiKey?: string baseUrl?: string + signal?: AbortSignal } // Exhaustive, value-level map for all dynamic providers. From 70172909a832d622b2e7e4f66f93687aa7eadcf6 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Fri, 11 Sep 2026 13:13:15 +0800 Subject: [PATCH 08/10] chore: re-trigger CodeRabbit review From d70c2e7d290386e078bdd1ccc376af0f58678ce0 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Fri, 11 Sep 2026 14:26:45 +0800 Subject: [PATCH 09/10] fix(api): stop Requesty stream yields after a mid-chunk abort MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Class h of the abort-signal contract (streaming yield granularity): the createMessage loop re-checked the per-request signal only at the top of the iteration, but a yield is a suspension point, so an abort can land between two yields of the same chunk and the remaining parts of that chunk leak after the request was already aborted. The text and tool-call yields now re-check via throwIfAborted before emitting; the reasoning yield needs no guard because it is the first yield of the iteration (no suspension point between the top-of-loop check and it). The loop comment claiming "the yields below are synchronous" is corrected. Tests: a mid-chunk abort after the reasoning yield of a reasoning+content+tool_call chunk (kills the text-yield guard), an abort between the content and tool-call yields of one chunk (kills the tool-call guard), and a structural case: buffered chunks after the abort point with a pull counter — the buffered chunk is reasoning-shaped (unguarded first yield of its iteration) so only the top-of-loop break can prevent it from leaking, and the assertion that no chunk beyond it is pulled. All assert no further yields and the normalized "The Requesty request was aborted". --- src/api/providers/__tests__/requesty.spec.ts | 118 +++++++++++++++++++ src/api/providers/requesty.ts | 13 +- 2 files changed, 128 insertions(+), 3 deletions(-) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index 0914509eac..2dff8dc843 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -910,6 +910,124 @@ describe("RequestyHandler", () => { expect(requestSignal).toBeInstanceOf(AbortSignal) expect(requestSignal).not.toBe(controller.signal) }) + it("does not yield text or tool calls after the signal aborts mid-chunk (pre-yield guards)", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + // A single chunk carrying reasoning + content + a tool call: each part's yield + // is a suspension point, so an abort can land between any two of them within + // the same chunk. + mockCreate.mockResolvedValue( + asyncStreamFrom([ + { + id: "c1", + choices: [ + { + delta: { + reasoning_content: "reasoning", + content: "content", + tool_calls: [ + { + index: 0, + id: "1", + type: "function", + function: { name: "f", arguments: "{}" }, + }, + ], + }, + }, + ], + }, + ]), + ) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + const first = await generator.next() + expect(first.value?.type).toBe("reasoning") + // Abort after the reasoning yield: the text and tool-call yields of the same + // chunk must not be emitted. + controller.abort() + + await expect(generator.next()).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + }) + it("does not yield tool calls after the signal aborts between the content and tool-call yields of one chunk", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + // Content + tool calls without reasoning: the content yield comes first, so + // the abort lands between it and the tool-call yield. + mockCreate.mockResolvedValue( + asyncStreamFrom([ + { + id: "c1", + choices: [ + { + delta: { + content: "content", + tool_calls: [ + { + index: 0, + id: "1", + type: "function", + function: { name: "f", arguments: "{}" }, + }, + ], + }, + }, + ], + }, + ]), + ) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + const first = await generator.next() + expect(first.value?.type).toBe("text") + controller.abort() + + await expect(generator.next()).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + }) + it("does not process buffered chunks after an abort and surfaces AbortError instead of completing the stream", async () => { + const handler = new RequestyHandler(mockOptions) + const controller = new AbortController() + let pulls = 0 + mockCreate.mockImplementationOnce(async (_params: unknown, _options?: { signal?: AbortSignal }) => { + return (async function* () { + pulls++ + yield { id: "c1", choices: [{ delta: { reasoning_content: "first" } }] } + pulls++ + // Reasoning shape: its yield is the first yield of the iteration and + // carries no pre-yield guard, so only the top-of-loop break prevents + // it from leaking. + yield { id: "c2", choices: [{ delta: { reasoning_content: "buffered" } }] } + pulls++ + yield { id: "c3", choices: [{ delta: { content: "third" } }] } + })() + }) + + const metadata = makeCreateMessageMetadata({ abortSignal: controller.signal }) + const generator = handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata) + + const first = await generator.next() + expect(first.value?.type).toBe("reasoning") + controller.abort() + + await expect(generator.next()).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + // The for-await mechanism pulls the already-buffered chunk before the + // top-of-loop check runs; the break must ensure it is not processed and + // that no chunk beyond it is pulled. + expect(pulls).toBe(2) + }) it("rejects with AbortError when the stream ends normally after a mid-stream abort (swallowed AbortError)", async () => { const handler = new RequestyHandler(mockOptions) const controller = new AbortController() diff --git a/src/api/providers/requesty.ts b/src/api/providers/requesty.ts index f71b217afb..04634a493c 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -221,9 +221,10 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan for await (const chunk of stream) { // The iterator can keep delivering buffered chunks after the abort has already // fired (openai@5.23.2 swallows the mid-stream AbortError), so re-check the - // signal before processing each chunk. The yields below are synchronous (there - // is no await between this check and them), so nothing is emitted once the - // signal aborts. + // signal before processing each chunk. A yield is a suspension point, so an + // abort can land between two yields of the same chunk: the guarded yields + // below re-check before emitting (the first yield of a chunk is covered by + // this check alone — no suspension point separates them). if (controller.signal.aborted) { break } @@ -231,18 +232,24 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan const delta = chunk.choices[0]?.delta // Yield reasoning chunks before content chunks so consumers see them in model order. + // No pre-yield guard: this is the first yield of the iteration — the + // top-of-loop check runs without a suspension point before it. const reasoningText = extractReasoningFromDelta(delta) if (reasoningText) { yield { type: "reasoning", text: reasoningText } } if (delta?.content) { + // Re-check before emitting: the consumer may have aborted while + // processing a previously yielded part of this chunk. + throwIfAborted(controller.signal) 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) { + throwIfAborted(controller.signal) yield { type: "tool_call_partial", index: toolCall.index, From 28ded4a3d957ecb657d01a76b989633342436a47 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Fri, 11 Sep 2026 18:54:05 +0800 Subject: [PATCH 10/10] fix(api): isolate Requesty model lookup from per-request abort signals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The model discovery fetch is shared single-flight (dedupedFetch in modelCache): concurrent createMessage/completePrompt callers join one in-flight fetch. Passing the per-request signal into the lookup let one caller's abort or timeout reject the shared fetch for every other waiter. - fetchModel no longer threads a signal into getModels - per-request cancellation remains the rejectOnAbort race (abort + timeout) - regression: two-signal cache-miss test — aborting one waiter mid-lookup must not reject the shared fetch for the other - old signal-threading tests inverted to the new contract --- src/api/providers/__tests__/requesty.spec.ts | 112 +++++++++++++++++-- src/api/providers/requesty.ts | 21 ++-- 2 files changed, 116 insertions(+), 17 deletions(-) diff --git a/src/api/providers/__tests__/requesty.spec.ts b/src/api/providers/__tests__/requesty.spec.ts index 2dff8dc843..6775d0594e 100644 --- a/src/api/providers/__tests__/requesty.spec.ts +++ b/src/api/providers/__tests__/requesty.spec.ts @@ -201,6 +201,96 @@ describe("RequestyHandler", () => { }) }) + describe("model lookup signal isolation", () => { + // The model fetch is a shared single-flight (dedupedFetch in modelCache): concurrent + // callers join one in-flight fetch. The lookup must stay independent of any waiter's + // signal, or one caller's abort would reject the shared fetch for every other waiter. + it("aborting one waiter mid-lookup does not reject the shared fetch for the other", async () => { + const handler = new RequestyHandler(mockOptions) + const { getModels } = await import("../fetchers/modelCache") + const getModelsMock = vitest.mocked(getModels) + + // A single in-flight lookup, joined by both waiters (single-flight semantics). + let resolveLookup!: (models: ModelRecord) => void + let joins = 0 + let notifyJoined!: () => void + const bothJoined = new Promise((resolve) => { + notifyJoined = resolve + }) + const sharedLookup = new Promise((resolve) => { + resolveLookup = resolve + }) + const joinOnce = () => { + joins++ + if (joins === 2) notifyJoined() + return sharedLookup + } + getModelsMock.mockImplementationOnce(joinOnce) + getModelsMock.mockImplementationOnce(joinOnce) + + const controllerA = new AbortController() + const controllerB = new AbortController() + const metadataA = makeCreateMessageMetadata({ abortSignal: controllerA.signal }) + const metadataB = makeCreateMessageMetadata({ abortSignal: controllerB.signal }) + + const pA = collectStream(handler.createMessage("sys", [{ role: "user", content: "a" }], metadataA)) + const pB = collectStream(handler.createMessage("sys", [{ role: "user", content: "b" }], metadataB)) + // Defensive: A is expected to reject below; park both so a mid-test failure + // cannot surface as an unhandled rejection. + void pA.catch(() => {}) + void pB.catch(() => {}) + + // Both waiters must have joined the same in-flight lookup before anyone settles. + await withSettleGuard(bothJoined) + // The lookup carries no waiter signal: the shared fetch is independent of the callers. + for (const [options] of getModelsMock.mock.calls) { + expect(options.signal).toBeUndefined() + } + + // Abort waiter A mid-lookup: A rejects with the AbortError contract, and the + // shared fetch stays pending for B (no cross-request rejection). + controllerA.abort() + await expect(pA).rejects.toMatchObject({ + name: "AbortError", + message: "The Requesty request was aborted", + }) + + // B: the shared fetch resolves for B; B proceeds normally to the stream phase. + mockCreate.mockResolvedValue( + asyncStreamFrom([ + { + id: mockOptions.requestyModelId, + choices: [{ delta: { content: "b-response" } }], + }, + { + id: "test-id", + choices: [{ delta: {} }], + usage: { + prompt_tokens: 1, + completion_tokens: 1, + total_tokens: 2, + }, + }, + ]), + ) + resolveLookup({ + "coding/claude-4-sonnet": { + maxTokens: 8192, + contextWindow: 200000, + supportsImages: true, + supportsPromptCache: true, + inputPrice: 3, + outputPrice: 15, + cacheWritesPrice: 3.75, + cacheReadsPrice: 0.3, + description: "Claude 4 Sonnet", + }, + }) + const chunksB = await pB + expect(chunksB).toContainEqual({ type: "text", text: "b-response" }) + }) + }) + describe("createMessage", () => { it("generates correct stream chunks", async () => { const handler = new RequestyHandler(mockOptions) @@ -827,7 +917,7 @@ describe("RequestyHandler", () => { resolveModelLookup({}) }) - it("threads the per-request signal into model discovery", async () => { + it("keeps model discovery independent of the per-request signal (shared single-flight fetch)", async () => { const handler = new RequestyHandler(mockOptions) mockCreate.mockResolvedValue(asyncStreamFrom([{ id: "c1", choices: [{ delta: { content: "ok" } }] }])) @@ -853,12 +943,14 @@ describe("RequestyHandler", () => { await collectStream(handler.createMessage("sys", [{ role: "user", content: "hi" }], metadata)) - expect(discoverySignal).toBeInstanceOf(AbortSignal) - // Discovery must receive the same per-request signal that is forwarded to the - // SDK, so a single abort cancels both the lookup and the completion request. + // The model lookup is a shared single-flight fetch joined by concurrent callers, so it + // must not carry this request's signal: one caller's abort would otherwise reject the + // shared fetch for every other waiter. Per-request cancellation is the rejectOnAbort + // race (covered by the mid-lookup abort tests above). + expect(discoverySignal).toBeUndefined() + // The completion request still receives a per-request signal forwarded to the SDK. const sdkOptions = mockCreate.mock.calls[0]?.[1] as { signal?: AbortSignal } | undefined expect(sdkOptions?.signal).toBeInstanceOf(AbortSignal) - expect(discoverySignal).toBe(sdkOptions?.signal) }) it("aborts the in-flight stream and rejects with AbortError when the external signal aborts", async () => { @@ -1345,7 +1437,7 @@ describe("RequestyHandler", () => { }) }) - it("threads the merged abort signal into model discovery", async () => { + it("keeps model discovery independent of the merged abort signal (shared single-flight fetch)", async () => { const handler = new RequestyHandler(mockOptions) mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "response" } }] }) @@ -1369,9 +1461,11 @@ describe("RequestyHandler", () => { const controller = new AbortController() await handler.completePrompt("test prompt", { abortSignal: controller.signal }) - // Without a timeout the merged signal is the external signal itself, so the - // lookup receives exactly the caller's signal (identity, not just type). - expect(discoverySignal).toBe(controller.signal) + // The model lookup is a shared single-flight fetch: it must not carry the caller's + // merged signal, or one caller's abort or timeout would reject the shared fetch for + // every other waiter. Per-request cancellation is the rejectOnAbort race (covered + // by the mid-lookup abort tests). + expect(discoverySignal).toBeUndefined() }) it("should pass timeout through to client", async () => { diff --git a/src/api/providers/requesty.ts b/src/api/providers/requesty.ts index 04634a493c..ada554915a 100644 --- a/src/api/providers/requesty.ts +++ b/src/api/providers/requesty.ts @@ -86,8 +86,12 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan }) } - public async fetchModel(signal?: AbortSignal) { - this.models = await getModels({ provider: providerIdentifiers.requesty, baseUrl: this.baseURL, signal }) + // Model discovery is a shared single-flight fetch (see dedupedFetch in modelCache): the lookup + // carries no signal, because concurrent createMessage/completePrompt callers join one + // in-flight fetch and a per-request signal would let one caller's abort or timeout reject the + // shared fetch for every other waiter. Each caller handles its own cancellation (rejectOnAbort). + public async fetchModel() { + this.models = await getModels({ provider: providerIdentifiers.requesty, baseUrl: this.baseURL }) return this.getModel() } @@ -168,10 +172,11 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan throw createAbortError("Requesty") } - // Model discovery is signal-aware: the per-request signal is threaded into the lookup so - // the underlying models request is cancelled on abort. The race below remains as a - // second line of defence for the window in which the fetcher swallows the cancellation - // and resolves with an empty model list. + // Model discovery is shared single-flight: the lookup carries no signal (see fetchModel), + // so this request's abort or timeout cannot reject the shared fetch for concurrent + // callers. Per-request cancellation is the rejectOnAbort race below: controller.signal + // bridges the external abort, and the race settles this waiter promptly while the + // underlying fetch keeps running to serve the other waiters. const { id: model, info, @@ -179,7 +184,7 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan temperature, reasoningEffort: reasoning_effort, reasoning: thinking, - } = await rejectOnAbort(this.fetchModel(controller.signal), controller.signal, this.providerName) + } = await rejectOnAbort(this.fetchModel(), controller.signal, this.providerName) const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ { role: "system", content: systemPrompt }, @@ -299,7 +304,7 @@ export class RequestyHandler extends BaseProvider implements SingleCompletionHan let modelData: Awaited> try { modelData = requestAbortSignal - ? await rejectOnAbort(this.fetchModel(requestAbortSignal), requestAbortSignal, this.providerName) + ? await rejectOnAbort(this.fetchModel(), requestAbortSignal, this.providerName) : await this.fetchModel() } catch (error) { if (isRequestAborted(error, requestAbortSignal)) {