diff --git a/src/api/providers/__tests__/complete-prompt-options.spec.ts b/src/api/providers/__tests__/complete-prompt-options.spec.ts new file mode 100644 index 0000000000..f9925cd119 --- /dev/null +++ b/src/api/providers/__tests__/complete-prompt-options.spec.ts @@ -0,0 +1,29 @@ +import { describe, it, expect } from "vitest" + +import type { CompletePromptOptions } from "../../index" + +describe("CompletePromptOptions", () => { + it("should allow abortSignal property", () => { + const controller = new AbortController() + const options: CompletePromptOptions = { abortSignal: controller.signal } + expect(options.abortSignal).toBe(controller.signal) + }) + + it("should allow timeoutMs property", () => { + const options: CompletePromptOptions = { timeoutMs: 5000 } + expect(options.timeoutMs).toBe(5000) + }) + + it("should allow both abortSignal and timeoutMs together", () => { + const controller = new AbortController() + const options: CompletePromptOptions = { abortSignal: controller.signal, timeoutMs: 10000 } + expect(options.abortSignal).toBe(controller.signal) + expect(options.timeoutMs).toBe(10000) + }) + + it("should allow empty options object", () => { + const options: CompletePromptOptions = {} + expect(options.abortSignal).toBeUndefined() + expect(options.timeoutMs).toBeUndefined() + }) +}) diff --git a/src/api/providers/__tests__/lm-studio-timeout.spec.ts b/src/api/providers/__tests__/lm-studio-timeout.spec.ts index f661d9092e..2f84457cb6 100644 --- a/src/api/providers/__tests__/lm-studio-timeout.spec.ts +++ b/src/api/providers/__tests__/lm-studio-timeout.spec.ts @@ -11,21 +11,33 @@ vitest.mock("../utils/timeout-config", () => ({ import { getApiRequestTimeout } from "../utils/timeout-config" import { clearAllMocks } from "../../../test-utils/reset" +import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" -// Mock OpenAI +interface MockOpenAiClient { + chat: { + completions: { + create: ReturnType + } + } +} + +// Mock OpenAI (records each created client so tests can drive its create call) const mockOpenAIConstructor = vitest.fn() +const createdClients: MockOpenAiClient[] = [] vitest.mock("openai", () => { return { __esModule: true, default: vitest.fn().mockImplementation(function (config) { - mockOpenAIConstructor(config) - return { + const client: MockOpenAiClient = { chat: { completions: { create: vitest.fn(), }, }, } + createdClients.push(client) + mockOpenAIConstructor(config) + return client }), } }) @@ -36,7 +48,7 @@ describe("LmStudioHandler timeout configuration", () => { }) it("should use default timeout of 600 seconds when no configuration is set", () => { - ;(getApiRequestTimeout as any).mockReturnValue(600000) + vitest.mocked(getApiRequestTimeout).mockReturnValue(600000) const options: ApiHandlerOptions = { apiModelId: "llama2", @@ -57,7 +69,7 @@ describe("LmStudioHandler timeout configuration", () => { }) it("should use custom timeout when configuration is set", () => { - ;(getApiRequestTimeout as any).mockReturnValue(1200000) // 20 minutes + vitest.mocked(getApiRequestTimeout).mockReturnValue(1200000) // 20 minutes const options: ApiHandlerOptions = { apiModelId: "llama2", @@ -75,7 +87,7 @@ describe("LmStudioHandler timeout configuration", () => { }) it("should handle zero timeout (no timeout)", () => { - ;(getApiRequestTimeout as any).mockReturnValue(0) + vitest.mocked(getApiRequestTimeout).mockReturnValue(0) const options: ApiHandlerOptions = { apiModelId: "llama2", @@ -91,3 +103,278 @@ describe("LmStudioHandler timeout configuration", () => { ) }) }) + +describe("LmStudioHandler abort signal wiring", () => { + let options: ApiHandlerOptions + + // Mirror the OpenAI SDK's APIUserAbortError shape: name "Error", message + // "Request was aborted." It does not satisfy the Task.ts abort contract + // (message must end in "aborted"), so the provider must normalize it. + const sdkAbortError = (): Error => { + const err = new Error("Request was aborted.") + err.name = "Error" + return err + } + + const waitForCreateCall = async (create: { mock: { calls: unknown[][] } }, timeoutMs = 5000): Promise => { + const start = Date.now() + while (create.mock.calls.length === 0) { + if (Date.now() - start > timeoutMs) { + throw new Error("timed out waiting for the SDK create call") + } + await new Promise((resolve) => setTimeout(resolve, 5)) + } + } + + const waitForSignalAbort = (signal: AbortSignal | undefined): Promise => { + return new Promise((resolve, reject) => { + if (!signal) { + reject(new Error("SDK create was called without a signal")) + return + } + if (signal.aborted) { + resolve() + return + } + signal.addEventListener("abort", () => resolve(), { once: true }) + }) + } + + const lastCreate = (): MockOpenAiClient["chat"]["completions"]["create"] => { + const client = createdClients[createdClients.length - 1] + if (!client) { + throw new Error("no OpenAI client was created") + } + return client.chat.completions.create + } + + beforeEach(() => { + clearAllMocks() + vitest.mocked(getApiRequestTimeout).mockReturnValue(600000) + options = { + apiModelId: "llama2", + lmStudioModelId: "llama2", + lmStudioBaseUrl: "http://localhost:1234", + } + }) + + describe("createMessage", () => { + it("should pass a request-local AbortSignal to the SDK and bridge the external signal", async () => { + const handler = new LmStudioHandler(options) + vitest.spyOn(handler, "countTokens").mockResolvedValue(1) + const create = lastCreate() + create.mockResolvedValue(asyncStreamFrom([])) + + const external = new AbortController() + const stream = handler.createMessage("system", [], { taskId: "t1", abortSignal: external.signal }) + await stream.next() + + const opts = create.mock.calls[0][1] + expect(opts?.signal).toBeInstanceOf(AbortSignal) + expect(opts.signal).not.toBe(external.signal) // request-local, not the external signal + expect(opts.signal.aborted).toBe(false) + + external.abort() + expect(opts.signal.aborted).toBe(true) // the external abort is bridged to the SDK signal + + await stream.next() // drain the generator + }) + + it("should fast-fail with a normalized AbortError when the signal is pre-aborted", async () => { + const handler = new LmStudioHandler(options) + const create = lastCreate() + const external = new AbortController() + external.abort() + + const stream = handler.createMessage("system", [], { taskId: "t1", abortSignal: external.signal }) + let caught: unknown + try { + await stream.next() + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + expect(create).not.toHaveBeenCalled() + }) + + it("should abort the in-flight SDK request when the external signal fires", async () => { + const handler = new LmStudioHandler(options) + vitest.spyOn(handler, "countTokens").mockResolvedValue(1) + const create = lastCreate() + // Simulate the OpenAI SDK: reject with its abort error when the signal aborts. + create.mockImplementation((_params: unknown, opts?: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + if (!opts?.signal) { + reject(new Error("SDK create was called without a signal")) + return + } + opts.signal.addEventListener("abort", () => reject(sdkAbortError()), { once: true }) + }) + }) + + const external = new AbortController() + const stream = handler.createMessage("system", [], { taskId: "t1", abortSignal: external.signal }) + const pending = stream.next() + await waitForCreateCall(create) + external.abort() + + let caught: unknown + try { + await pending + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + }) + + it("should normalize an abort error thrown mid-stream", async () => { + const handler = new LmStudioHandler(options) + vitest.spyOn(handler, "countTokens").mockResolvedValue(1) + const create = lastCreate() + const external = new AbortController() + // Simulate the OpenAI SDK stream: yield once, then reject with its + // abort error once the request-local signal is aborted. + create.mockImplementation((_params: unknown, opts?: { signal?: AbortSignal }) => { + return (async function* () { + yield { choices: [{ delta: { content: "partial" } }] } + await waitForSignalAbort(opts?.signal) + throw sdkAbortError() + })() + }) + + const stream = handler.createMessage("system", [], { taskId: "t1", abortSignal: external.signal }) + const chunks: { type: string; text?: string }[] = [] + let caught: unknown + try { + for await (const chunk of stream) { + chunks.push(chunk) + if (chunk.type === "text") { + external.abort() + } + } + } catch (error) { + caught = error + } + + expect(chunks).toContainEqual({ type: "text", text: "partial" }) + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + }) + + it("should stream reasoning chunks from a reasoning_content delta", async () => { + // Changed-line coverage regression: reasoning models served by LM Studio + // stream thinking via delta.reasoning_content, and createMessage must yield + // a reasoning chunk from that dedicated field. + const handler = new LmStudioHandler(options) + vitest.spyOn(handler, "countTokens").mockResolvedValue(1) + const create = lastCreate() + create.mockResolvedValue( + asyncStreamFrom([ + { choices: [{ delta: { reasoning_content: "thinking..." }, index: 0 }] }, + { choices: [{ delta: { content: "answer" }, index: 0 }] }, + { + choices: [{ delta: {}, index: 0 }], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }, + ]), + ) + + const chunks = await collectStream(handler.createMessage("system", [])) + + expect(chunks).toContainEqual({ type: "reasoning", text: "thinking..." }) + expect(chunks).toContainEqual({ type: "text", text: "answer" }) + }) + }) + + describe("completePrompt", () => { + it("should pass the external signal through, and nothing without a signal or with a zero timeout", async () => { + const handler = new LmStudioHandler(options) + const create = lastCreate() + create.mockResolvedValue({ choices: [{ message: { content: "ok" } }] }) + const external = new AbortController() + + expect(await handler.completePrompt("hi")).toBe("ok") + expect(create.mock.calls[0][1]).toBeUndefined() // no signal, no timeout: nothing reaches the SDK + + expect(await handler.completePrompt("hi", { abortSignal: external.signal })).toBe("ok") + // no timeout: the merged signal is the external signal itself + expect(create.mock.calls[1][1]?.signal).toBe(external.signal) + + // timeoutMs <= 0 means "no explicit timeout": nothing may reach the SDK + expect(await handler.completePrompt("hi", { timeoutMs: 0 })).toBe("ok") + expect(create.mock.calls[2][1]).toBeUndefined() + }) + + it("should merge the external signal with a positive timeoutMs", async () => { + const handler = new LmStudioHandler(options) + const create = lastCreate() + create.mockImplementation((_params: unknown, opts?: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + const signal = opts?.signal + if (!signal) { + reject(new Error("SDK create was called without a signal")) + return + } + signal.addEventListener("abort", () => reject(sdkAbortError()), { once: true }) + }) + }) + const external = new AbortController() + + const pending = handler.completePrompt("hi", { abortSignal: external.signal, timeoutMs: 60_000 }) + const opts = create.mock.calls[0][1] + expect(opts?.signal).toBeInstanceOf(AbortSignal) + expect(opts.signal).not.toBe(external.signal) // merged via AbortSignal.any + expect(opts.signal.aborted).toBe(false) + + external.abort() + let caught: unknown + try { + await pending + } catch (error) { + caught = error + } + + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + }) + + it("should normalize SDK abort errors instead of wrapping them", async () => { + const handler = new LmStudioHandler(options) + const create = lastCreate() + create.mockRejectedValue(sdkAbortError()) + + let caught: unknown + try { + await handler.completePrompt("hi") + } catch (error) { + caught = error + } + + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + }) + + it("should keep wrapping non-abort errors in the LM Studio debug message", async () => { + const handler = new LmStudioHandler(options) + const create = lastCreate() + create.mockRejectedValue(new Error("boom")) + + let caught: unknown + try { + await handler.completePrompt("hi") + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).message).toContain("Please check the LM Studio developer logs") + }) + }) +}) diff --git a/src/api/providers/__tests__/lmstudio-native-tools.spec.ts b/src/api/providers/__tests__/lmstudio-native-tools.spec.ts index c6a63902a1..7bd28cf8ef 100644 --- a/src/api/providers/__tests__/lmstudio-native-tools.spec.ts +++ b/src/api/providers/__tests__/lmstudio-native-tools.spec.ts @@ -81,6 +81,7 @@ describe("LmStudioHandler Native Tools", () => { }), ]), }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) // parallel_tool_calls should be true by default when not explicitly set const callArgs = mockCreate.mock.calls[0][0] @@ -103,6 +104,7 @@ describe("LmStudioHandler Native Tools", () => { expect.objectContaining({ tool_choice: "auto", }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -204,6 +206,7 @@ describe("LmStudioHandler Native Tools", () => { expect.objectContaining({ parallel_tool_calls: true, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) diff --git a/src/api/providers/__tests__/lmstudio.spec.ts b/src/api/providers/__tests__/lmstudio.spec.ts index 7ab674a0a9..6aa354b8c4 100644 --- a/src/api/providers/__tests__/lmstudio.spec.ts +++ b/src/api/providers/__tests__/lmstudio.spec.ts @@ -204,18 +204,58 @@ describe("LmStudioHandler", () => { "Please check the LM Studio developer logs to debug what went wrong. You may need to load the model with a larger context length to work with Zoo Code's prompts.", ) }) + + it("should not issue the request when the caller aborts while input token counting is pending", async () => { + const controller = new AbortController() + let releaseCount!: () => void + const countGate = new Promise((resolve) => { + releaseCount = resolve + }) + const countSpy = vi.spyOn(handler, "countTokens").mockImplementation(async () => { + await countGate + return 10 + }) + + const stream = handler.createMessage(systemPrompt, messages, { + taskId: "test-task-id", + abortSignal: controller.signal, + }) + const pending = collectStream(stream).catch((error: unknown) => error) + + // Let the generator reach the token count, then abort while it is pending. + const start = Date.now() + while (countSpy.mock.calls.length === 0) { + if (Date.now() - start > 5000) { + throw new Error("timed out waiting for the token count") + } + await new Promise((resolve) => setTimeout(resolve, 5)) + } + controller.abort() + releaseCount() + + const caught = (await pending) as Error + countSpy.mockRestore() + + expect(caught).toBeInstanceOf(Error) + expect(caught.name).toBe("AbortError") + expect(caught.message).toBe("The LM Studio request was aborted") + expect(mockCreate).not.toHaveBeenCalled() + }) }) describe("completePrompt", () => { it("should complete prompt successfully", async () => { const result = await handler.completePrompt("Test prompt") expect(result).toBe("Test response") - expect(mockCreate).toHaveBeenCalledWith({ - model: mockOptions.lmStudioModelId, - messages: [{ role: "user", content: "Test prompt" }], - temperature: 0, - stream: false, - }) + expect(mockCreate).toHaveBeenCalledWith( + { + model: mockOptions.lmStudioModelId, + messages: [{ role: "user", content: "Test prompt" }], + temperature: 0, + stream: false, + }, + undefined, // no abort signal or timeout: no request options reach the SDK + ) }) it("should handle API errors", async () => { diff --git a/src/api/providers/__tests__/qwen-code-native-tools.spec.ts b/src/api/providers/__tests__/qwen-code-native-tools.spec.ts index 54df551d4e..9c716927b8 100644 --- a/src/api/providers/__tests__/qwen-code-native-tools.spec.ts +++ b/src/api/providers/__tests__/qwen-code-native-tools.spec.ts @@ -101,6 +101,7 @@ describe("QwenCodeHandler Native Tools", () => { ]), parallel_tool_calls: true, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -120,6 +121,7 @@ describe("QwenCodeHandler Native Tools", () => { expect.objectContaining({ tool_choice: "auto", }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -220,6 +222,7 @@ describe("QwenCodeHandler Native Tools", () => { expect.objectContaining({ parallel_tool_calls: true, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -418,4 +421,424 @@ describe("QwenCodeHandler Native Tools", () => { expect(endChunks).toHaveLength(1) }) }) + + describe("abort signal wiring", () => { + afterEach(() => { + vi.unstubAllGlobals() + }) + + // Mirror the OpenAI SDK's APIUserAbortError shape: name "Error", message + // "Request was aborted." It does not satisfy the Task.ts abort contract + // (message must end in "aborted"), so the provider must normalize it. + const sdkAbortError = (): Error => { + const err = new Error("Request was aborted.") + err.name = "Error" + return err + } + + const unauthorizedError = (): Error & { status: number } => + Object.assign(new Error("unauthorized"), { status: 401 }) + + const tokenResponse = (): { ok: boolean; json: () => Promise> } => ({ + ok: true, + json: async () => ({ + access_token: "new-access-token", + refresh_token: "new-refresh-token", + token_type: "Bearer", + expires_in: 3600, + }), + }) + + const waitForCreateCall = async (create: { mock: { calls: unknown[][] } }, timeoutMs = 5000): Promise => { + const start = Date.now() + while (create.mock.calls.length === 0) { + if (Date.now() - start > timeoutMs) { + throw new Error("timed out waiting for the SDK create call") + } + await new Promise((resolve) => setTimeout(resolve, 5)) + } + } + + const waitForSignalAbort = (signal: AbortSignal | undefined): Promise => { + return new Promise((resolve, reject) => { + if (!signal) { + reject(new Error("SDK create was called without a signal")) + return + } + if (signal.aborted) { + resolve() + return + } + signal.addEventListener("abort", () => resolve(), { once: true }) + }) + } + + describe("createMessage", () => { + it("should pass a request-local AbortSignal to the SDK and bridge the external signal", async () => { + const external = new AbortController() + let sdkSignal: AbortSignal | undefined + // A live stream: yield one chunk, then stay open until the SDK signal aborts. + mockCreate.mockImplementationOnce((_params: unknown, opts?: { signal?: AbortSignal }) => { + sdkSignal = opts?.signal + return (async function* () { + yield { choices: [{ delta: { content: "x" } }] } + await waitForSignalAbort(sdkSignal) + })() + }) + + const stream = handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal }) + const first = await stream.next() + expect(first.value?.type).toBe("text") + + const opts = mockCreate.mock.calls[0][1] + expect(opts?.signal).toBeInstanceOf(AbortSignal) + expect(opts.signal).not.toBe(external.signal) // request-local, not the external signal + expect(opts.signal.aborted).toBe(false) + + external.abort() + expect(opts.signal.aborted).toBe(true) // the external abort is bridged to the SDK signal + + await stream.next() // resume; the live stream ends once the signal aborts + }) + + it("should fast-fail with a normalized AbortError for a pre-aborted signal", async () => { + const external = new AbortController() + external.abort() + + const stream = handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal }) + let caught: unknown + try { + await stream.next() + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + expect(mockCreate).not.toHaveBeenCalled() + }) + + it("should abort the in-flight SDK request when the external signal fires", async () => { + const external = new AbortController() + // Simulate the OpenAI SDK: reject with its abort error when the signal aborts. + mockCreate.mockImplementationOnce((_params: unknown, opts?: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + if (!opts?.signal) { + reject(new Error("SDK create was called without a signal")) + return + } + opts.signal.addEventListener("abort", () => reject(sdkAbortError()), { once: true }) + }) + }) + + const stream = handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal }) + const pending = stream.next() + await waitForCreateCall(mockCreate) + external.abort() + + let caught: unknown + try { + await pending + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + }) + + it("should normalize an abort error thrown mid-stream", async () => { + const external = new AbortController() + // Simulate the OpenAI SDK stream: yield once, then reject with its + // abort error once the request-local signal is aborted. + mockCreate.mockImplementationOnce((_params: unknown, opts?: { signal?: AbortSignal }) => { + return (async function* () { + yield { choices: [{ delta: { content: "partial" } }] } + await waitForSignalAbort(opts?.signal) + throw sdkAbortError() + })() + }) + + const stream = handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal }) + const chunks: { type: string; text?: string }[] = [] + let caught: unknown + try { + for await (const chunk of stream) { + chunks.push(chunk) + if (chunk.type === "text") { + external.abort() + } + } + } catch (error) { + caught = error + } + + expect(chunks).toContainEqual({ type: "text", text: "partial" }) + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + }) + + it("should rethrow non-abort stream errors unchanged", async () => { + const boom = new Error("boom") + mockCreate.mockImplementationOnce(() => { + return (async function* () { + yield { choices: [{ delta: { content: "x" } }] } + throw boom + })() + }) + + const stream = handler.createMessage("test prompt", [], { taskId: "t1" }) + let caught: unknown + try { + await collectStream(stream) + } catch (error) { + caught = error + } + + expect(caught).toBe(boom) + }) + + it("should split tag boundaries across chunks into reasoning and text", async () => { + // Exercises the incremental think-tag parser: one chunk opens a + // thinking block (odd segment), the next one closes it (even + // segment) and continues as visible text. + mockCreate.mockImplementationOnce(() => + asyncStreamFrom([ + { choices: [{ delta: { content: "ab" } }] }, + { choices: [{ delta: { content: "c" } }] }, + ]), + ) + + const stream = handler.createMessage("test prompt", [], { taskId: "t1" }) + const chunks = await collectStream(stream) + + expect(chunks).toContainEqual({ type: "reasoning", text: "b" }) + expect(chunks).toContainEqual({ type: "text", text: "c" }) + expect(chunks).not.toContainEqual(expect.objectContaining({ type: "text", text: "b" })) + }) + + it("should tolerate degenerate stream shapes (empty choice, repeated content, zero usage)", async () => { + // Changed-line coverage: exercises the defensive branches of the stream + // loop — a chunk with no choice, a delta that repeats the previous full + // content (empty after trimming), a think block that starts the text so the + // split yields an empty leading segment, and a usage payload of zeros. + mockCreate.mockImplementationOnce(() => + asyncStreamFrom([ + { choices: [] }, + { choices: [{ delta: { content: "hi" }, index: 0 }] }, + { choices: [{ delta: { content: "hi" }, index: 0 }] }, + { choices: [{ delta: { content: "thoughtout" }, index: 0 }] }, + { + choices: [{ delta: {}, index: 0, finish_reason: "stop" }], + usage: { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }, + }, + ]), + ) + + const stream = handler.createMessage("test prompt", [], { taskId: "t1" }) + const chunks = await collectStream(stream) + + const hiChunks = chunks.filter((chunk) => chunk.type === "text" && chunk.text === "hi") + expect(hiChunks).toHaveLength(1) // the repeated content chunk yields nothing + expect(chunks).toContainEqual({ type: "reasoning", text: "thought" }) + expect(chunks).toContainEqual({ type: "text", text: "out" }) + expect(chunks).toContainEqual({ type: "usage", inputTokens: 0, outputTokens: 0 }) + }) + + it("should not retry after 401 when the abort signal fires during the refresh", async () => { + const external = new AbortController() + const fetchMock = vi.fn().mockImplementation(async () => { + external.abort() // simulate Stop pressed while the token refresh is in flight + return tokenResponse() + }) + vi.stubGlobal("fetch", fetchMock) + mockCreate.mockRejectedValueOnce(unauthorizedError()) + + const stream = handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal }) + let caught: unknown + try { + await collectStream(stream) + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + expect(mockCreate).toHaveBeenCalledTimes(1) // the retried request was never sent + expect(fetchMock).toHaveBeenCalledTimes(1) + }) + + it("should normalize an abort error from the 401 retry instead of exposing the raw SDK error", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(tokenResponse())) + const external = new AbortController() + mockCreate + .mockRejectedValueOnce(unauthorizedError()) + .mockImplementationOnce((_params: unknown, opts?: { signal?: AbortSignal }) => { + external.abort() // Stop pressed while the retried request is in flight + return Promise.reject(sdkAbortError()) + }) + + const stream = handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal }) + let caught: unknown + try { + await collectStream(stream) + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toBe("The Qwen Code request was aborted") + expect((caught as Error).message).not.toBe("Request was aborted.") // not the raw SDK error + expect(mockCreate).toHaveBeenCalledTimes(2) // first attempt 401, then the aborted retry + }) + }) + + describe("completePrompt", () => { + it("should pass the external signal through, and nothing without a signal or with a zero timeout", async () => { + mockCreate + .mockResolvedValueOnce({ choices: [{ message: { content: "ok" } }] }) + .mockResolvedValueOnce({ choices: [{ message: { content: "ok" } }] }) + .mockResolvedValueOnce({ choices: [{ message: { content: "ok" } }] }) + const external = new AbortController() + + expect(await handler.completePrompt("hi")).toBe("ok") + expect(mockCreate.mock.calls[0][1]).toBeUndefined() // no signal, no timeout: nothing reaches the SDK + + expect(await handler.completePrompt("hi", { abortSignal: external.signal })).toBe("ok") + // no timeout: the merged signal is the external signal itself + expect(mockCreate.mock.calls[1][1]?.signal).toBe(external.signal) + + // timeoutMs <= 0 means "no explicit timeout": nothing may reach the SDK + expect(await handler.completePrompt("hi", { timeoutMs: 0 })).toBe("ok") + expect(mockCreate.mock.calls[2][1]).toBeUndefined() + }) + + it("should merge the external signal with a positive timeoutMs", async () => { + const external = new AbortController() + mockCreate.mockImplementationOnce((_params: unknown, opts?: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + const signal = opts?.signal + if (!signal) { + reject(new Error("SDK create was called without a signal")) + return + } + signal.addEventListener("abort", () => reject(sdkAbortError()), { once: true }) + }) + }) + + const pending = handler.completePrompt("hi", { abortSignal: external.signal, timeoutMs: 60_000 }) + await waitForCreateCall(mockCreate) + const opts = mockCreate.mock.calls[0][1] + expect(opts?.signal).toBeInstanceOf(AbortSignal) + expect(opts.signal).not.toBe(external.signal) // merged via AbortSignal.any + expect(opts.signal.aborted).toBe(false) + + external.abort() + let caught: unknown + try { + await pending + } catch (error) { + caught = error + } + + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + }) + + it("should fast-fail with a normalized AbortError for a pre-aborted signal", async () => { + const external = new AbortController() + external.abort() + + let caught: unknown + try { + await handler.completePrompt("hi", { abortSignal: external.signal }) + } catch (error) { + caught = error + } + + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + expect(fs.readFile).not.toHaveBeenCalled() // no work starts after a pre-aborted signal + expect(mockCreate).not.toHaveBeenCalled() + }) + + it("should retry after 401 and pass the same abort signal to the retry", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(tokenResponse())) + mockCreate + .mockRejectedValueOnce(unauthorizedError()) + .mockResolvedValueOnce({ choices: [{ message: { content: "retried" } }] }) + + const result = await handler.completePrompt("hi", { abortSignal: new AbortController().signal }) + + expect(result).toBe("retried") + expect(mockCreate).toHaveBeenCalledTimes(2) + expect(mockCreate.mock.calls[1][1]?.signal).toBe(mockCreate.mock.calls[0][1]?.signal) + }) + + it("should not retry after 401 when the abort signal fires during the refresh", async () => { + const external = new AbortController() + const fetchMock = vi.fn().mockImplementation(async () => { + external.abort() // simulate Stop pressed while the token refresh is in flight + return tokenResponse() + }) + vi.stubGlobal("fetch", fetchMock) + mockCreate.mockRejectedValueOnce(unauthorizedError()) + + let caught: unknown + try { + await handler.completePrompt("hi", { abortSignal: external.signal }) + } catch (error) { + caught = error + } + + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + expect(mockCreate).toHaveBeenCalledTimes(1) // the retried request was never sent + expect(fetchMock).toHaveBeenCalledTimes(1) + }) + + it("should normalize SDK abort errors instead of rethrowing them", async () => { + mockCreate.mockRejectedValueOnce(sdkAbortError()) + + let caught: unknown + try { + await handler.completePrompt("hi") + } catch (error) { + caught = error + } + + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toMatch(/aborted$/) + }) + + it("should normalize an abort error from the 401 retry instead of exposing the raw SDK error", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(tokenResponse())) + const external = new AbortController() + mockCreate + .mockRejectedValueOnce(unauthorizedError()) + .mockImplementationOnce((_params: unknown, opts?: { signal?: AbortSignal }) => { + external.abort() // Stop pressed while the retried request is in flight + return Promise.reject(sdkAbortError()) + }) + + let caught: unknown + try { + await handler.completePrompt("hi", { abortSignal: external.signal }) + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + expect((caught as Error).message).toBe("The Qwen Code request was aborted") + expect((caught as Error).message).not.toBe("Request was aborted.") // not the raw SDK error + expect(mockCreate).toHaveBeenCalledTimes(2) // first attempt 401, then the aborted retry + }) + }) + }) }) diff --git a/src/api/providers/lm-studio.ts b/src/api/providers/lm-studio.ts index 0c828984bc..de3a0110f5 100644 --- a/src/api/providers/lm-studio.ts +++ b/src/api/providers/lm-studio.ts @@ -18,8 +18,16 @@ import { convertToOpenAiMessages } from "../transform/openai-format" import { ApiStream } from "../transform/stream" import { BaseProvider } from "./base-provider" +import { RequestConfigBuilder } from "./config-builder/request-config-builder" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { getModelsFromCache } from "./fetchers/modelCache" +import { + mergeAbortSignalAndTimeout, + throwIfAborted, + createAbortError, + isRequestAborted, + type OpenAiRequestOptions, +} from "./utils/abort-signal" import { handleOpenAIError } from "./utils/error-handler" import { extractReasoningFromDelta } from "./utils/extract-reasoning" @@ -47,6 +55,9 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { + // Fast-fail if the caller’s stop signal already fired before we started. + throwIfAborted(metadata?.abortSignal) + const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ { role: "system", content: systemPrompt }, ...convertToOpenAiMessages(messages), @@ -88,6 +99,24 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan let assistantText = "" let reasoningOutput = "" + // Request-local abort controller — a class field would outlive this + // request and let concurrent requests abort each other. + const requestController = new AbortController() + const onExternalAbort = () => { + requestController.abort() + } + const externalSignal = metadata?.abortSignal + if (externalSignal) { + externalSignal.addEventListener("abort", onExternalAbort) + // An abort can land while the input token count above is still + // pending: a listener registered after the signal already aborted + // never fires, so bridge the aborted state into the request-local + // controller. + if (externalSignal.aborted) { + requestController.abort() + } + } + try { const params: OpenAI.Chat.ChatCompletionCreateParamsStreaming & { draft_model?: string } = { model: this.getModel().id, @@ -103,10 +132,24 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan params.draft_model = this.options.lmStudioDraftModelId } + // Bridge the request-local signal into the SDK request options so the + // in-flight request can be cancelled. + const createOptions = new RequestConfigBuilder() + .setOption("signal", requestController.signal) + .build() + + // Fast-fail if the caller aborted while countTokens() above was + // pending — the request must not be issued once the request-local + // signal has aborted. + throwIfAborted(requestController.signal) + let results try { - results = await this.client.chat.completions.create(params) + results = await this.client.chat.completions.create(params, createOptions) } catch (error) { + if (isRequestAborted(error, externalSignal)) { + throw createAbortError("LM Studio") + } throw handleOpenAIError(error, this.providerName) } @@ -181,9 +224,20 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan outputTokens, } as const } catch (error) { + if (isRequestAborted(error, externalSignal)) { + throw createAbortError("LM Studio") + } throw new Error( "Please check the LM Studio developer logs to debug what went wrong. You may need to load the model with a larger context length to work with Zoo Code's prompts.", ) + } finally { + // Cancel the in-flight SDK request if the consumer stopped iterating + // early (break or return): the external signal may never fire in that + // case, and only the request-local signal reaches the SDK. + requestController.abort() + if (externalSignal) { + externalSignal.removeEventListener("abort", onExternalAbort) + } } } @@ -206,6 +260,14 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan } async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { + // Fast-fail if the caller’s stop signal already fired before we started. + throwIfAborted(options?.abortSignal) + + // Merge the external stop signal with an optional per-call timeout. A + // timeoutMs <= 0 means "no explicit timeout" inside the util, so zero + // never reaches the SDK as an explicit timeout. + const requestSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) + try { // Create params object with optional draft model const params: any = { @@ -220,14 +282,27 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan params.draft_model = this.options.lmStudioDraftModelId } + // CompletePromptOptions is not createMessage metadata (no taskId), so + // the generic builder takes the merged signal via setOption instead of + // setAbortSignal(metadata). + const createOptions = new RequestConfigBuilder() + .setOption("signal", requestSignal) + .build() + let response try { - response = await this.client.chat.completions.create(params) + response = await this.client.chat.completions.create(params, createOptions) } catch (error) { + if (isRequestAborted(error, requestSignal)) { + throw createAbortError("LM Studio") + } throw handleOpenAIError(error, this.providerName) } return response.choices[0]?.message.content || "" } catch (error) { + if (isRequestAborted(error, requestSignal)) { + throw createAbortError("LM Studio") + } throw new Error( "Please check the LM Studio developer logs to debug what went wrong. You may need to load the model with a larger context length to work with Zoo Code's prompts.", ) diff --git a/src/api/providers/qwen-code.ts b/src/api/providers/qwen-code.ts index 5001b4c8ed..b7b06bfb11 100644 --- a/src/api/providers/qwen-code.ts +++ b/src/api/providers/qwen-code.ts @@ -14,7 +14,15 @@ import { convertToOpenAiMessages } from "../transform/openai-format" import { ApiStream } from "../transform/stream" import { BaseProvider } from "./base-provider" +import { RequestConfigBuilder } from "./config-builder/request-config-builder" import { extractReasoningFromDelta } from "./utils/extract-reasoning" +import { + mergeAbortSignalAndTimeout, + throwIfAborted, + createAbortError, + isRequestAborted, + type OpenAiRequestOptions, +} from "./utils/abort-signal" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" const QWEN_OAUTH_BASE_URL = "https://chat.qwen.ai" @@ -194,17 +202,42 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan return baseUrl.endsWith("/v1") ? baseUrl : `${baseUrl}/v1` } - private async callApiWithRetry(apiCall: () => Promise): Promise { + private async callApiWithRetry(apiCall: () => Promise, externalSignal?: AbortSignal): Promise { try { return await apiCall() } catch (error: any) { + // An aborted request must never be retried: normalize it to the + // Task.ts abort contract (name "AbortError", message ending in + // "aborted") instead of rethrowing the raw SDK abort error. + if (isRequestAborted(error, externalSignal)) { + throw createAbortError("Qwen Code") + } if (error.status === 401) { - // Token expired, refresh and retry + // Token expired, refresh and retry. The retry reuses apiCall’s + // captured request options, so it carries the same abort signal. + // (An already-aborted request is normalized above and never + // reaches this branch.) this.credentials = await this.refreshAccessToken(this.credentials!) + // A stop can land while the refresh await is in flight — re-check + // before the retried request goes out so it is not sent. + if (externalSignal?.aborted) { + throw createAbortError("Qwen Code") + } const client = this.ensureClient() client.apiKey = this.credentials.access_token client.baseURL = this.getBaseUrl(this.credentials) - return await apiCall() + // A stop can also land while the retried request itself is in + // flight; that rejection must go through the same abort + // normalization as the first attempt instead of escaping as the + // raw SDK abort error. + try { + return await apiCall() + } catch (retryError) { + if (isRequestAborted(retryError, externalSignal)) { + throw createAbortError("Qwen Code") + } + throw retryError + } } else { throw error } @@ -216,107 +249,145 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { - await this.ensureAuthenticated() - const client = this.ensureClient() - const model = this.getModel() - - const systemMessage: OpenAI.Chat.ChatCompletionSystemMessageParam = { - role: "system", - content: systemPrompt, + // Fast-fail if the caller’s stop signal already fired before we started. + throwIfAborted(metadata?.abortSignal) + + // Request-local abort controller — a class field would outlive this + // request and let concurrent requests abort each other. + const requestController = new AbortController() + const onExternalAbort = () => { + requestController.abort() } + const externalSignal = metadata?.abortSignal + if (externalSignal) { + externalSignal.addEventListener("abort", onExternalAbort) + } + + try { + await this.ensureAuthenticated() + const client = this.ensureClient() + const model = this.getModel() - const convertedMessages = [systemMessage, ...convertToOpenAiMessages(messages)] + const systemMessage: OpenAI.Chat.ChatCompletionSystemMessageParam = { + role: "system", + content: systemPrompt, + } - const requestOptions: OpenAI.Chat.Completions.ChatCompletionCreateParamsStreaming = { - model: model.id, - temperature: 0, - messages: convertedMessages, - stream: true, - stream_options: { include_usage: true }, - max_completion_tokens: model.info.maxTokens, - tools: this.convertToolsForOpenAI(metadata?.tools), - tool_choice: metadata?.tool_choice, - parallel_tool_calls: metadata?.parallelToolCalls ?? true, - } + const convertedMessages = [systemMessage, ...convertToOpenAiMessages(messages)] + + const requestOptions: OpenAI.Chat.Completions.ChatCompletionCreateParamsStreaming = { + model: model.id, + temperature: 0, + messages: convertedMessages, + stream: true, + stream_options: { include_usage: true }, + max_completion_tokens: model.info.maxTokens, + tools: this.convertToolsForOpenAI(metadata?.tools), + tool_choice: metadata?.tool_choice, + parallel_tool_calls: metadata?.parallelToolCalls ?? true, + } - const stream = await this.callApiWithRetry(() => client.chat.completions.create(requestOptions)) + // Bridge the request-local signal into the SDK request options so + // the in-flight request (and any 401 retry) can be cancelled. + const createOptions = new RequestConfigBuilder() + .setOption("signal", requestController.signal) + .build() - let fullContent = "" + const stream = await this.callApiWithRetry( + () => client.chat.completions.create(requestOptions, createOptions), + externalSignal, + ) - for await (const apiChunk of stream) { - const delta = apiChunk.choices[0]?.delta ?? {} - const finishReason = apiChunk.choices[0]?.finish_reason + let fullContent = "" - if (delta.content) { - let newText = delta.content - if (newText.startsWith(fullContent)) { - newText = newText.substring(fullContent.length) - } - fullContent = delta.content - - if (newText) { - // Check for thinking blocks - if (newText.includes("") || newText.includes("")) { - // Simple parsing for thinking blocks - const parts = newText.split(/<\/?think>/g) - for (let i = 0; i < parts.length; i++) { - if (parts[i]) { - if (i % 2 === 0) { - // Outside thinking block - yield { - type: "text", - text: parts[i], - } - } else { - // Inside thinking block - yield { - type: "reasoning", - text: parts[i], + for await (const apiChunk of stream) { + const delta = apiChunk.choices[0]?.delta ?? {} + const finishReason = apiChunk.choices[0]?.finish_reason + + if (delta.content) { + let newText = delta.content + if (newText.startsWith(fullContent)) { + newText = newText.substring(fullContent.length) + } + fullContent = delta.content + + if (newText) { + // Check for thinking blocks + if (newText.includes("") || newText.includes("")) { + // Simple parsing for thinking blocks + const parts = newText.split(/<\/?think>/g) + for (let i = 0; i < parts.length; i++) { + if (parts[i]) { + if (i % 2 === 0) { + // Outside thinking block + yield { + type: "text", + text: parts[i], + } + } else { + // Inside thinking block + yield { + type: "reasoning", + text: parts[i], + } } } } + } else { + yield { + type: "text", + text: newText, + } } - } else { + } + } + + const reasoningText = extractReasoningFromDelta(delta) + if (reasoningText) { + yield { type: "reasoning", text: reasoningText } + } + + // Handle tool calls in stream - emit partial chunks for NativeToolCallParser + if (delta.tool_calls) { + for (const toolCall of delta.tool_calls) { yield { - type: "text", - text: newText, + type: "tool_call_partial", + index: toolCall.index, + id: toolCall.id, + name: toolCall.function?.name, + arguments: toolCall.function?.arguments, } } } - } - const reasoningText = extractReasoningFromDelta(delta) - if (reasoningText) { - yield { type: "reasoning", text: reasoningText } - } + // Process finish_reason to emit tool_call_end events + if (finishReason) { + const endEvents = NativeToolCallParser.processFinishReason(finishReason) + for (const event of endEvents) { + yield event + } + } - // Handle tool calls in stream - emit partial chunks for NativeToolCallParser - if (delta.tool_calls) { - for (const toolCall of delta.tool_calls) { + if (apiChunk.usage) { yield { - type: "tool_call_partial", - index: toolCall.index, - id: toolCall.id, - name: toolCall.function?.name, - arguments: toolCall.function?.arguments, + type: "usage", + inputTokens: apiChunk.usage.prompt_tokens || 0, + outputTokens: apiChunk.usage.completion_tokens || 0, } } } - - // Process finish_reason to emit tool_call_end events - if (finishReason) { - const endEvents = NativeToolCallParser.processFinishReason(finishReason) - for (const event of endEvents) { - yield event - } + } catch (error) { + if (isRequestAborted(error, externalSignal)) { + throw createAbortError("Qwen Code") } - - if (apiChunk.usage) { - yield { - type: "usage", - inputTokens: apiChunk.usage.prompt_tokens || 0, - outputTokens: apiChunk.usage.completion_tokens || 0, - } + throw error + } finally { + // Cancel the in-flight SDK request if the consumer stopped iterating + // early (break or return): the external signal may never fire in that + // case, and only the request-local signal reaches the SDK. + requestController.abort() + if (externalSignal) { + externalSignal.removeEventListener("abort", onExternalAbort) } } } @@ -328,6 +399,21 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan } async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { + // Fast-fail if the caller’s stop signal already fired before we started. + throwIfAborted(options?.abortSignal) + + // Merge the external stop signal with an optional per-call timeout. A + // timeoutMs <= 0 means "no explicit timeout" inside the util, so zero + // never reaches the SDK as an explicit timeout. + const requestSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) + + // CompletePromptOptions is not createMessage metadata (no taskId), so + // the generic builder takes the merged signal via setOption instead of + // setAbortSignal(metadata). + const createOptions = new RequestConfigBuilder() + .setOption("signal", requestSignal) + .build() + await this.ensureAuthenticated() const client = this.ensureClient() const model = this.getModel() @@ -338,7 +424,13 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan max_completion_tokens: model.info.maxTokens, } - const response = await this.callApiWithRetry(() => client.chat.completions.create(requestOptions)) + // The retry reuses the captured request options, so it carries the same + // merged signal — and the guard in callApiWithRetry refuses to retry + // once this signal has aborted. + const response = await this.callApiWithRetry( + () => client.chat.completions.create(requestOptions, createOptions), + requestSignal, + ) 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 ebc7edf3d3..aba72c181f 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -1,4 +1,10 @@ -import { mergeAbortSignalAndTimeout, mergeAbortSignals } from "../abort-signal" +import { + createAbortError, + isRequestAborted, + mergeAbortSignalAndTimeout, + mergeAbortSignals, + throwIfAborted, +} from "../abort-signal" describe("abort-signal utilities", () => { describe("mergeAbortSignalAndTimeout", () => { @@ -99,4 +105,84 @@ describe("abort-signal utilities", () => { expect(result.aborted).toBe(true) }) }) + + describe("throwIfAborted", () => { + it("does not throw when signal is undefined", () => { + expect(() => throwIfAborted()).not.toThrow() + }) + + it("does not throw when signal is not aborted", () => { + const controller = new AbortController() + + expect(() => throwIfAborted(controller.signal)).not.toThrow() + }) + + it("throws an AbortError when signal is already aborted", () => { + const controller = new AbortController() + controller.abort() + + let caught: unknown + try { + throwIfAborted(controller.signal) + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(Error) + expect((caught as Error).name).toBe("AbortError") + }) + }) + + describe("isRequestAborted", () => { + it("returns true when the caller signal is aborted", () => { + const controller = new AbortController() + controller.abort() + + expect(isRequestAborted(new Error("boom"), controller.signal)).toBe(true) + expect(isRequestAborted(undefined, controller.signal)).toBe(true) + }) + + it("returns true for a native AbortError or the OpenAI SDK APIUserAbortError", () => { + const native = new Error("This operation was aborted") + native.name = "AbortError" + expect(isRequestAborted(native)).toBe(true) + + const sdk = new Error("whatever") + sdk.name = "APIUserAbortError" + expect(isRequestAborted(sdk)).toBe(true) + }) + + it("matches the OpenAI SDK abort message exactly, not as a substring", () => { + expect(isRequestAborted(new Error("Request was aborted."))).toBe(true) + expect(isRequestAborted(new Error("Request was aborted"))).toBe(false) + expect(isRequestAborted(new Error("Request was aborted. Please retry"))).toBe(false) + }) + + it("returns false for unrelated errors, nullish errors, and live signals", () => { + expect(isRequestAborted(new Error("the abort failed"))).toBe(false) + expect(isRequestAborted(undefined)).toBe(false) + expect(isRequestAborted(null)).toBe(false) + + const controller = new AbortController() + expect(isRequestAborted(new Error("boom"), controller.signal)).toBe(false) + }) + }) + + describe("createAbortError", () => { + it("builds an error satisfying the Task.ts abort contract", () => { + const error = createAbortError("LM Studio") + + expect(error).toBeInstanceOf(Error) + expect(error.name).toBe("AbortError") + expect(error.message).toBe("The LM Studio request was aborted") + }) + + it("interpolates the provider name", () => { + expect(createAbortError("Qwen Code").message).toBe("The Qwen Code request was aborted") + }) + + it("returns a fresh error on each call", () => { + expect(createAbortError("X")).not.toBe(createAbortError("X")) + }) + }) }) diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index 73e0356f7b..26f57c3e9a 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -35,3 +35,61 @@ export function mergeAbortSignals(primarySignal: AbortSignal, secondarySignal?: return AbortSignal.any([primarySignal, secondarySignal]) } + +/** + * Throw an AbortError if the given signal is already aborted. + * + * Use as a fast-fail guard at the top of request-building code paths so + * callers receive a consistent `name === "AbortError"` when the operation + * was cancelled before it started, without building or issuing the request. + */ +export function throwIfAborted(signal?: AbortSignal): void { + if (!signal?.aborted) { + return + } + + const abortError = new Error("This operation was aborted") + abortError.name = "AbortError" + throw abortError +} + +/** + * Request options this series passes to the OpenAI SDK call. The SDK's + * `RequestOptions` declares `signal` as `AbortSignal | null | undefined`, + * which does not satisfy the builder's base constraint, so the builder is + * typed with only the options this series sets. The built config is still + * assignable to the SDK's `RequestOptions`. + */ +export type OpenAiRequestOptions = { + signal?: AbortSignal +} + +/** + * Whether a failure indicates an aborted request: the caller's signal fired, + * the SDK raised a native abort error, or the error carries the OpenAI SDK + * abort error message (exactly "Request was aborted."). The message check + * is an exact match on purpose: a substring match would misclassify + * unrelated errors that merely mention aborting. + */ +export function isRequestAborted(error: unknown, signal?: AbortSignal): boolean { + const candidate = error as { name?: string; message?: string } + return ( + Boolean(signal?.aborted) || + candidate?.name === "AbortError" || + candidate?.name === "APIUserAbortError" || + candidate?.message === "Request was aborted." + ) +} + +/** + * Fresh error satisfying the Task.ts abort contract: `name === + * "AbortError"` and a message ending in "aborted" (no trailing period). The + * OpenAI SDK's own abort error does not satisfy this contract (name "Error", + * message "Request was aborted."), so raw SDK abort errors must be + * normalized instead of rethrown. + */ +export function createAbortError(providerName: string): Error { + const abortError = new Error(`The ${providerName} request was aborted`) + abortError.name = "AbortError" + return abortError +} diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index f790fba436..897d93d0af 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -174,11 +174,6 @@ "count": 36 } }, - "api/providers/__tests__/lm-studio-timeout.spec.ts": { - "@typescript-eslint/no-explicit-any": { - "count": 3 - } - }, "api/providers/__tests__/mimo.spec.ts": { "@typescript-eslint/no-explicit-any": { "count": 18