diff --git a/package.json b/package.json index 8431467918..0a914fb148 100644 --- a/package.json +++ b/package.json @@ -43,7 +43,8 @@ "rimraf": "6.0.1", "tsx": "4.22.4", "turbo": "2.10.0", - "typescript": "5.9.3" + "typescript": "5.9.3", + "vitest": "4.1.9" }, "lint-staged": { "*.{js,jsx,ts,tsx,json,css,md}": [ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 183a5e02d1..a8972fd426 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -74,6 +74,9 @@ importers: typescript: specifier: 5.9.3 version: 5.9.3 + vitest: + specifier: 4.1.9 + version: 4.1.9(@opentelemetry/api@1.9.0)(@types/node@22.20.1)(@vitest/coverage-v8@4.1.9)(@vitest/ui@4.1.9)(jsdom@26.1.0)(vite@8.1.0(@types/node@22.20.1)(esbuild@0.28.1)(jiti@2.7.0)(tsx@4.22.4)(yaml@2.9.0)) apps/cli: dependencies: @@ -5037,6 +5040,7 @@ packages: eslint@9.39.4: resolution: {integrity: sha512-XoMjdBOwe/esVgEvLmNsD3IRHkm7fbKIUGvrleloJXUZgDHig2IPWNniv+GwjyJXzuNqVjlr5+4yVUZjycJwfQ==} engines: {node: ^18.18.0 || ^20.9.0 || >=21.1.0} + deprecated: This version is no longer supported. Please see https://eslint.org/version-support for other options. hasBin: true peerDependencies: jiti: '*' 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__/opencode-go.spec.ts b/src/api/providers/__tests__/opencode-go.spec.ts index 828ffb8655..373afc973e 100644 --- a/src/api/providers/__tests__/opencode-go.spec.ts +++ b/src/api/providers/__tests__/opencode-go.spec.ts @@ -9,8 +9,12 @@ vitest.mock("vscode", () => ({ }, })) -import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI from "openai" +import { + Anthropic, + APIConnectionTimeoutError as AnthropicTimeoutError, + APIUserAbortError as AnthropicAbortError, +} from "@anthropic-ai/sdk" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { opencodeGoDefaultModelId, @@ -24,6 +28,7 @@ import { getModels } from "../fetchers/modelCache" import { ApiHandlerOptions } from "../../../shared/api" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { clearAllMocks } from "../../../test-utils/reset" +import { makeCreateMessageMetadata } from "../../../test-utils/api" vitest.mock("openai") vitest.mock("delay", () => ({ @@ -64,15 +69,22 @@ const mockResponsesCreate = vitest.fn() } }) -vitest.mock("@anthropic-ai/sdk", () => ({ - Anthropic: vitest.fn(function () { - return { - messages: { - create: mockAnthropicCreate, - }, - } - }), -})) +// The real SDK error classes are re-exported alongside the mocked client so +// tests can emulate the SDK's abort/timeout rejections and the provider's +// instanceof checks resolve against the same class identity. +vitest.mock("@anthropic-ai/sdk", async () => { + const actual = await vi.importActual("@anthropic-ai/sdk") + return { + ...actual, + Anthropic: vitest.fn(function () { + return { + messages: { + create: mockAnthropicCreate, + }, + } + }), + } +}) describe("OpencodeGoHandler", () => { const mockOptions: ApiHandlerOptions = { @@ -205,6 +217,7 @@ describe("OpencodeGoHandler", () => { max_completion_tokens: 40_960, temperature: expect.any(Number), }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -219,6 +232,7 @@ describe("OpencodeGoHandler", () => { model: "glm-5.1", reasoning_effort: "medium", }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -369,10 +383,241 @@ describe("OpencodeGoHandler", () => { await collectStream(handler.createMessage("sys", messages)) - expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ max_completion_tokens: 999 })) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ max_completion_tokens: 999 }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), + ) + }) + + it("rethrows non-abort errors from the OpenAI stream unchanged", async () => { + // A mid-stream failure that is not an abort (e.g. a connection + // reset) must propagate unchanged — the catch only normalizes + // aborts to a DOM-standard AbortError. + const streamError = new Error("connection reset") + mockCreate.mockImplementation(async () => + (async function* () { + yield { choices: [{ delta: { content: "partial" }, index: 0 }], index: 0 } + throw streamError + })(), + ) + + const handler = new OpencodeGoHandler(mockOptions) + + const error = await collectStream(handler.createMessage("sys", [{ role: "user", content: "hi" }])).then( + () => undefined, + (e: unknown) => e, + ) + + expect(error).toBe(streamError) + }) + + it("skips empty choices, empty deltas and tool calls without a function field", async () => { + // Full-list assertions: a frame without choices[0], a delta without + // content/tool_calls, and a tool call missing `function` must each + // contribute nothing except the partial tool call with undefined + // name/arguments (toEqual ignores undefined-valued keys). + mockCreate.mockImplementationOnce(async () => + asyncStreamFrom([ + { choices: [], index: 0 }, + { choices: [{ delta: {}, index: 0 }], index: 0 }, + { + choices: [{ delta: { tool_calls: [{ index: 0, id: "call_1" }] }, index: 0 }], + index: 0, + }, + { + choices: [{ delta: {}, index: 0 }], + index: 0, + usage: { prompt_tokens: 1, completion_tokens: 2 }, + }, + ]), + ) + + const handler = new OpencodeGoHandler(mockOptions) + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "hi" }] + + const chunks = await collectStream(handler.createMessage("sys", messages)) + + expect(chunks).toEqual([ + { type: "tool_call_partial", index: 0, id: "call_1" }, + { type: "usage", inputTokens: 1, outputTokens: 2 }, + ]) + }) + + it("emits the text delta and the usage chunk with cached tokens from the final frame", async () => { + mockCreate.mockImplementationOnce(async () => + asyncStreamFrom([ + { choices: [{ delta: { content: "hi" }, index: 0 }], index: 0 }, + { + choices: [{ delta: {}, index: 0 }], + index: 0, + usage: { + prompt_tokens: 4, + completion_tokens: 9, + prompt_tokens_details: { cached_tokens: 6 }, + }, + }, + ]), + ) + + const handler = new OpencodeGoHandler(mockOptions) + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "hi" }] + + const chunks = await collectStream(handler.createMessage("sys", messages)) + + expect(chunks).toEqual([ + { type: "text", text: "hi" }, + { type: "usage", inputTokens: 4, outputTokens: 9, cacheReadTokens: 6 }, + ]) }) }) + describe("createMessage abort signal bridging", () => { + it("rejects with an AbortError when the external signal is already aborted", async () => { + // Capture the INTERNAL controller signal the SDK call receives: an + // already-aborted external signal must abort the controller before + // the request starts so the catch path can normalize the rejection + // to the DOM-standard AbortError. + let capturedSignal: AbortSignal | undefined + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + throw new DOMException("The operation was aborted.", "AbortError") + }) + + const handler = new OpencodeGoHandler(mockOptions) + const controller = new AbortController() + controller.abort() + + const stream = handler.createMessage( + "sys", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const error = await collectStream(stream).then( + () => undefined, + (e: unknown) => e, + ) + expect(capturedSignal?.aborted).toBe(true) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("aborts the in-flight request when the external signal fires mid-stream", async () => { + // The mock polls the INTERNAL controller signal instead of waiting + // for an "abort" event: bounded polling means the test can never + // hang if the bridge stops forwarding aborts, and it rejects as + // soon as the bridge aborts the controller. + let capturedSignal: AbortSignal | undefined + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + return (async function* () { + yield { choices: [{ delta: { content: "partial" }, index: 0 }], index: 0 } + for (let i = 0; i < 40 && !capturedSignal?.aborted; i++) { + await new Promise((resolve) => setTimeout(resolve, 5)) + } + if (capturedSignal?.aborted) { + throw new DOMException("The operation was aborted.", "AbortError") + } + })() + }) + + const handler = new OpencodeGoHandler(mockOptions) + const controller = new AbortController() + + const consumed = collectStream( + handler.createMessage( + "sys", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + + // Let the request start and the first chunk be yielded before aborting. + await new Promise((resolve) => setTimeout(resolve, 25)) + controller.abort() + + const error = await consumed.then( + () => undefined, + (e: unknown) => e, + ) + expect(capturedSignal?.aborted).toBe(true) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("detaches the bridged abort listener when the request completes normally", async () => { + // The listener is added with { once: true }, so it only detaches on + // abort. A task-scoped signal spanning many requests must not + // accumulate a listener per request: assert explicit removal after a + // normal (non-aborted) completion. + mockCreate.mockImplementation(async () => + asyncStreamFrom([ + { + choices: [{ delta: { content: "ok" }, index: 0 }], + index: 0, + }, + { + choices: [{ delta: {}, index: 0 }], + index: 0, + usage: { prompt_tokens: 2, completion_tokens: 3 }, + }, + ]), + ) + + const handler = new OpencodeGoHandler(mockOptions) + const controller = new AbortController() + const removeListenerSpy = vi.spyOn(controller.signal, "removeEventListener") + const addEventListenerSpy = vi.spyOn(controller.signal, "addEventListener") + + const stream = handler.createMessage( + "sys", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const chunks = await collectStream(stream) + expect(chunks).toContainEqual({ type: "text", text: "ok" }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + // The listener is registered with { once: true } — assert the exact + // options so a bridge that drops them (and relies on the finally + // block alone for single-shot semantics) is caught. + expect(addEventListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function), { once: true }) + expect(controller.signal.aborted).toBe(false) + }) + + it("detaches the bridged abort listener from the Anthropic-format path", async () => { + // The Anthropic branch (streamAnthropicMessage) has its own finally + // block that removes the bridged listener; assert explicit removal + // after a normal (non-aborted) completion on that path too. + mockAnthropicCreate.mockImplementationOnce(async () => + asyncStreamFrom([ + { type: "message_start", message: { usage: { input_tokens: 1, output_tokens: 0 } } }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "ok" } }, + { type: "message_delta", usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]), + ) + + const handler = new OpencodeGoHandler({ + opencodeGoApiKey: "test-key", + opencodeGoModelId: "qwen3.7-max", + }) + const controller = new AbortController() + const removeListenerSpy = vi.spyOn(controller.signal, "removeEventListener") + + const stream = handler.createMessage( + "sys", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const chunks = await collectStream(stream) + expect(chunks).toContainEqual({ type: "text", text: "ok" }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + expect(controller.signal.aborted).toBe(false) + }) + }) describe("completePrompt", () => { it("returns the message content for a non-streaming completion", async () => { mockCreate.mockResolvedValue({ choices: [{ message: { content: "the answer" } }] }) @@ -386,6 +631,7 @@ describe("OpencodeGoHandler", () => { max_completion_tokens: 40_960, reasoning_effort: "medium", }), + {}, ) }) @@ -411,7 +657,7 @@ describe("OpencodeGoHandler", () => { mockCreate.mockResolvedValue({ choices: [{ message: { content: "ok" } }] }) const handler = new OpencodeGoHandler({ ...mockOptions, includeMaxTokens: true, modelMaxTokens: 4321 }) await handler.completePrompt("ping") - expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ max_completion_tokens: 4321 })) + expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ max_completion_tokens: 4321 }), {}) }) }) @@ -473,6 +719,7 @@ describe("OpencodeGoHandler", () => { stream: true, system: expect.arrayContaining([expect.objectContaining({ type: "text", text: "sys" })]), }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) // The OpenAI chat completions endpoint must NOT be used for this model. expect(mockCreate).not.toHaveBeenCalled() @@ -518,6 +765,23 @@ describe("OpencodeGoHandler", () => { ) }) + it("preserves abort identity when the Anthropic request rejects with a name-based AbortError", async () => { + // No SDK abort class and no aborted signal: only the DOM-standard + // name === "AbortError" check marks a cancelled pre-stream request. + const rawAbort = Object.assign(new Error("raw"), { name: "AbortError" }) + mockAnthropicCreate.mockRejectedValueOnce(rawAbort) + + const handler = new OpencodeGoHandler(anthropicOptions) + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "hi" }] + + const error = await collectStream(handler.createMessage("sys", messages)).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + it("applies cache-control breakpoints when the model supports prompt caching", async () => { const handler = new OpencodeGoHandler(anthropicOptions) const messages: Anthropic.Messages.MessageParam[] = [ @@ -554,6 +818,7 @@ describe("OpencodeGoHandler", () => { // so the model default is used. max_tokens: 65_536, }), + undefined, ) expect(mockCreate).not.toHaveBeenCalled() }) @@ -569,7 +834,7 @@ describe("OpencodeGoHandler", () => { modelMaxTokens: 2048, }) await handler.completePrompt("ping") - expect(mockAnthropicCreate).toHaveBeenCalledWith(expect.objectContaining({ max_tokens: 2048 })) + expect(mockAnthropicCreate).toHaveBeenCalledWith(expect.objectContaining({ max_tokens: 2048 }), undefined) }) it("completePrompt rethrows non-Error values unchanged from the Anthropic path", async () => { @@ -584,6 +849,320 @@ describe("OpencodeGoHandler", () => { expect(await handler.completePrompt("ping")).toBe("") }) + it("completePrompt passes abort signal through to Anthropic client", async () => { + mockAnthropicCreate.mockResolvedValue({ content: [{ type: "text", text: "response" }] }) + const controller = new AbortController() + const handler = new OpencodeGoHandler(anthropicOptions) + await handler.completePrompt("ping", { abortSignal: controller.signal }) + expect(mockAnthropicCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + signal: controller.signal, + }) + }) + + it("completePrompt passes both signal and timeoutMs through to Anthropic client", async () => { + mockAnthropicCreate.mockResolvedValue({ content: [{ type: "text", text: "response" }] }) + const controller = new AbortController() + const handler = new OpencodeGoHandler(anthropicOptions) + await handler.completePrompt("ping", { abortSignal: controller.signal, timeoutMs: 10000 }) + expect(mockAnthropicCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + signal: controller.signal, + timeout: 10000, + }) + }) + + it("completePrompt passes only timeoutMs when no signal is provided", async () => { + mockAnthropicCreate.mockResolvedValue({ content: [{ type: "text", text: "response" }] }) + const handler = new OpencodeGoHandler(anthropicOptions) + await handler.completePrompt("ping", { timeoutMs: 5000 }) + expect(mockAnthropicCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + timeout: 5000, + }) + }) + + it("completePrompt omits the timeout option when timeoutMs is 0 (Anthropic path)", async () => { + // The SDK treats timeout: 0 as an immediate abort, so the "disabled" + // value must never be forwarded — assert the absence of the option. + mockAnthropicCreate.mockResolvedValue({ content: [{ type: "text", text: "response" }] }) + const handler = new OpencodeGoHandler(anthropicOptions) + await handler.completePrompt("ping", { timeoutMs: 0 }) + const call = mockAnthropicCreate.mock.calls[mockAnthropicCreate.mock.calls.length - 1] + const requestOptions = call[1] as { timeout?: number } | undefined + // When no option is forwarded the provider omits the SDK options + // argument entirely, so absence means: undefined arg OR an arg + // without a timeout key. + expect(Object.keys(requestOptions ?? {})).not.toContain("timeout") + }) + + it("completePrompt preserves abort identity when the caller aborts (Anthropic path)", async () => { + // Emulate the Anthropic SDK: an aborted request signal rejects with + // APIUserAbortError ("Request was aborted." — the trailing period would + // fail task-level abort detection, so the provider must normalize it). + mockAnthropicCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + throw new AnthropicAbortError() + } + throw new Error("boom") + }) + const handler = new OpencodeGoHandler(anthropicOptions) + const controller = new AbortController() + controller.abort() + + const error = await handler.completePrompt("ping", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("completePrompt surfaces request timeouts as an AbortError (Anthropic path)", async () => { + // Emulate the Anthropic SDK: when the request timeout fires, the SDK + // surfaces APIConnectionTimeoutError ("Request timed out.") once retries + // are exhausted — verified against @anthropic-ai/sdk against a hung server. + mockAnthropicCreate.mockImplementation(async (_params: unknown, options: { timeout?: number }) => { + await new Promise((resolve) => setTimeout(resolve, options?.timeout ?? 50)) + throw new AnthropicTimeoutError() + }) + const handler = new OpencodeGoHandler(anthropicOptions) + + const error = await handler.completePrompt("ping", { timeoutMs: 50 }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("completePrompt works without options (backward compatible, Anthropic path)", async () => { + mockAnthropicCreate.mockResolvedValue({ content: [{ type: "text", text: "response" }] }) + const handler = new OpencodeGoHandler(anthropicOptions) + const result = await handler.completePrompt("ping") + expect(result).toBe("response") + expect(mockAnthropicCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + undefined, + ) + }) + + it("completePrompt keeps the model max_tokens when includeMaxTokens is off (Anthropic path)", async () => { + // includeMaxTokens unset: modelMaxTokens must NOT replace the model + // default — only the explicit includeMaxTokens flag opts into the + // user override. + mockAnthropicCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "ok" }] }) + const handler = new OpencodeGoHandler({ ...anthropicOptions, modelMaxTokens: 2048 }) + await handler.completePrompt("ping") + expect(mockAnthropicCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: "qwen3.7-max", max_tokens: 65_536 }), + undefined, + ) + }) + + it("completePrompt forwards an explicit model temperature (Anthropic path)", async () => { + mockAnthropicCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "ok" }] }) + const handler = new OpencodeGoHandler({ ...anthropicOptions, modelTemperature: 0.7 }) + await handler.completePrompt("ping") + expect(mockAnthropicCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: "qwen3.7-max", temperature: 0.7 }), + undefined, + ) + }) + + it("completePrompt preserves abort identity when the signal is pre-aborted with a plain error", async () => { + // The aborted-signal disjunct alone must normalize a plain + // rejection (not just SDK abort classes) to the DOM-standard + // AbortError. + mockAnthropicCreate.mockRejectedValueOnce(new Error("boom")) + const controller = new AbortController() + controller.abort() + const handler = new OpencodeGoHandler(anthropicOptions) + + const error = await handler.completePrompt("ping", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("completePrompt preserves abort identity for a name-based AbortError rejection (Anthropic path)", async () => { + // No aborted signal and no SDK abort class: only the DOM-standard + // name === "AbortError" check marks a cancelled request. + const rawAbort = Object.assign(new Error("raw"), { name: "AbortError" }) + mockAnthropicCreate.mockRejectedValueOnce(rawAbort) + const handler = new OpencodeGoHandler(anthropicOptions) + + const error = await handler.completePrompt("ping").then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + describe("completePrompt (OpenAI path)", () => { + const openaiOptions: ApiHandlerOptions = { + opencodeGoApiKey: "test-key", + apiModelId: "glm-5.1", // OpenAI-format model + } + + beforeEach(() => { + vitest.clearAllMocks() + }) + + it("completePrompt returns text for OpenAI path", async () => { + mockCreate.mockResolvedValueOnce({ + choices: [{ message: { content: "response" } }], + }) + + const handler = new OpencodeGoHandler(openaiOptions) + expect(await handler.completePrompt("ping")).toBe("response") + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String), stream: false }), + {}, // empty object when no options + ) + }) + + it("completePrompt passes abort signal through to OpenAI client", async () => { + mockCreate.mockResolvedValueOnce({ + choices: [{ message: { content: "response" } }], + }) + const controller = new AbortController() + const handler = new OpencodeGoHandler(openaiOptions) + + await handler.completePrompt("ping", { abortSignal: controller.signal }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String), stream: false }), + { signal: controller.signal }, + ) + }) + + it("completePrompt passes both signal and timeoutMs through to OpenAI client", async () => { + mockCreate.mockResolvedValueOnce({ + choices: [{ message: { content: "response" } }], + }) + const controller = new AbortController() + const handler = new OpencodeGoHandler(openaiOptions) + + await handler.completePrompt("ping", { abortSignal: controller.signal, timeoutMs: 10000 }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String), stream: false }), + { signal: controller.signal, timeout: 10000 }, + ) + }) + + it("completePrompt passes only timeoutMs when no signal is provided", async () => { + mockCreate.mockResolvedValueOnce({ + choices: [{ message: { content: "response" } }], + }) + const handler = new OpencodeGoHandler(openaiOptions) + + await handler.completePrompt("ping", { timeoutMs: 5000 }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String), stream: false }), + { timeout: 5000 }, + ) + }) + + it("completePrompt omits the timeout option when timeoutMs is 0 (OpenAI path)", async () => { + // The OpenAI SDK treats timeout: 0 as an immediate abort, so the + // "disabled" value must never be forwarded — assert the absence of + // the option. + mockCreate.mockResolvedValueOnce({ + choices: [{ message: { content: "response" } }], + }) + const handler = new OpencodeGoHandler(openaiOptions) + await handler.completePrompt("ping", { timeoutMs: 0 }) + const call = mockCreate.mock.calls[mockCreate.mock.calls.length - 1] + const requestOptions = call[1] as { timeout?: number } | undefined + expect(requestOptions).not.toHaveProperty("timeout") + }) + + it("completePrompt preserves abort identity when the caller aborts (OpenAI path)", async () => { + // Emulate the OpenAI SDK: an aborted request signal rejects with + // APIUserAbortError ("Request was aborted." — the trailing period would + // fail task-level abort detection, so the provider must normalize it). + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + throw new APIUserAbortError() + } + throw new Error("boom") + }) + const handler = new OpencodeGoHandler(openaiOptions) + const controller = new AbortController() + controller.abort() + + const error = await handler.completePrompt("ping", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("completePrompt surfaces request timeouts as an AbortError (OpenAI path)", async () => { + // Emulate the OpenAI SDK: when the request timeout fires, the SDK + // surfaces APIConnectionTimeoutError ("Request timed out.") once retries + // are exhausted — verified against openai v5.23.2 against a hung server. + mockCreate.mockImplementation(async (_params: unknown, options: { timeout?: number }) => { + await new Promise((resolve) => setTimeout(resolve, options?.timeout ?? 50)) + throw new APIConnectionTimeoutError() + }) + const handler = new OpencodeGoHandler(openaiOptions) + + const error = await handler.completePrompt("ping", { timeoutMs: 50 }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("completePrompt works without options (backward compatible, OpenAI path)", async () => { + mockCreate.mockResolvedValueOnce({ + choices: [{ message: { content: "response" } }], + }) + const handler = new OpencodeGoHandler(openaiOptions) + + const result = await handler.completePrompt("ping") + expect(result).toBe("response") + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String), stream: false }), + {}, // empty object when no options + ) + }) + + it("completePrompt preserves abort identity when the signal is pre-aborted with a plain error", async () => { + // The aborted-signal disjunct alone must normalize a plain + // rejection (not just SDK abort classes) to the DOM-standard + // AbortError. + mockCreate.mockRejectedValueOnce(new Error("boom")) + const controller = new AbortController() + controller.abort() + const handler = new OpencodeGoHandler(openaiOptions) + + const error = await handler.completePrompt("ping", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("completePrompt preserves abort identity for a name-based AbortError rejection (OpenAI path)", async () => { + // No aborted signal and no SDK abort class: only the DOM-standard + // name === "AbortError" check marks a cancelled request. + const rawAbort = Object.assign(new Error("raw"), { name: "AbortError" }) + mockCreate.mockRejectedValueOnce(rawAbort) + const handler = new OpencodeGoHandler(openaiOptions) + + const error = await handler.completePrompt("ping").then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + }) it("omits tools and tool_choice from the Anthropic request when no tools are provided", async () => { const handler = new OpencodeGoHandler(anthropicOptions) const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "Hi" }] @@ -729,7 +1308,10 @@ describe("OpencodeGoHandler", () => { await collectStream(handler.createMessage("sys", messages)) - expect(mockAnthropicCreate).toHaveBeenCalledWith(expect.objectContaining({ max_tokens: 8192 })) + expect(mockAnthropicCreate).toHaveBeenCalledWith( + expect.objectContaining({ max_tokens: 8192 }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), + ) }) it("falls back to the model max_tokens when includeMaxTokens is on but modelMaxTokens is unset", async () => { @@ -739,7 +1321,10 @@ describe("OpencodeGoHandler", () => { await collectStream(handler.createMessage("sys", messages)) // qwen3.7-max maxTokens (65_536) clamped to 20% of 1M context => 65_536. - expect(mockAnthropicCreate).toHaveBeenCalledWith(expect.objectContaining({ max_tokens: 65_536 })) + expect(mockAnthropicCreate).toHaveBeenCalledWith( + expect.objectContaining({ max_tokens: 65_536 }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), + ) }) it("accumulates output tokens across message_delta events into the final cost", async () => { @@ -806,6 +1391,21 @@ describe("OpencodeGoHandler", () => { await collectStream(handler.createMessage("sys", messages)) }).rejects.toThrow("Opencode Go completion error: rate limited") }) + + it("preserves abort identity for aborted Anthropic requests from createMessage", async () => { + // A cancelled /v1/messages request (the SDK rejects with + // APIUserAbortError) must surface as a DOM-standard AbortError, not + // the wrapped "completion error" reserved for other failures. + mockAnthropicCreate.mockRejectedValue(new AnthropicAbortError()) + const handler = new OpencodeGoHandler(anthropicOptions) + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "Hi" }] + await expect(async () => { + await collectStream(handler.createMessage("sys", messages)) + }).rejects.toMatchObject({ + name: "AbortError", + message: "The Opencode Go request was aborted", + }) + }) }) describe("Responses-format models (gpt-5.6-luna)", () => { diff --git a/src/api/providers/__tests__/unbound.spec.ts b/src/api/providers/__tests__/unbound.spec.ts index 99f11d7b6f..d4bcb912d7 100644 --- a/src/api/providers/__tests__/unbound.spec.ts +++ b/src/api/providers/__tests__/unbound.spec.ts @@ -1,18 +1,27 @@ import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI from "openai" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { UnboundHandler } from "../unbound" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { clearAllMocks } from "../../../test-utils/reset" +import { makeCreateMessageMetadata } from "../../../test-utils/api" -vi.mock("openai", () => { - const createMock = vi.fn() +// Single hoisted mock shared by the `openai` factory and every test so tests +// can configure the SDK `create` call without untyped access casts. +const sharedMockCreate = vi.hoisted(() => vi.fn()) + +// The real SDK error classes are re-exported alongside the mocked client so +// tests can emulate the SDK's abort/timeout rejections (APIUserAbortError, +// APIConnectionTimeoutError) and the provider's instanceof checks resolve. +vi.mock("openai", async () => { + const actual = await vi.importActual("openai") return { + ...actual, default: vi.fn(function () { return { chat: { completions: { - create: createMock, + create: sharedMockCreate, }, }, } @@ -179,7 +188,169 @@ describe("UnboundHandler", () => { mode: "architect", }, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), + ) + }) + + it("wraps non-abort pre-stream failures via handleOpenAIError", async () => { + // A non-abort rejection from create() (e.g. an upstream 500) must be + // routed through handleOpenAIError, not the AbortError normalization + // path: assert the wrapped identity and the preserved message. + sharedMockCreate.mockRejectedValue(new Error("upstream 500")) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const stream = handler.createMessage("system", [{ role: "user", content: "hi" }], { + taskId: "t", + tools: [], + }) + + const error = await collectStream(stream).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toBeInstanceOf(Error) + expect((error as Error).message).toBe("Unbound completion error: upstream 500") + expect((error as Error).name).not.toBe("AbortError") + }) + + it("emits tool_call_partial chunks for native tool calls in the stream", async () => { + // Native tool calls arrive on delta.tool_calls and must be re-emitted + // as raw tool_call_partial chunks for NativeToolCallParser to assemble. + sharedMockCreate.mockResolvedValue( + asyncStreamFrom([ + { + choices: [ + { + delta: { + tool_calls: [ + { + index: 0, + id: "call_1", + type: "function", + function: { name: "get_weather", arguments: '{"city": "NYC"}' }, + }, + ], + }, + }, + ], + }, + { choices: [{ delta: { content: "done" } }], usage: { prompt_tokens: 1, completion_tokens: 1 } }, + ]), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const chunks = await collectStream( + handler.createMessage("system", [{ role: "user", content: "hi" }], { + taskId: "t", + tools: [], + }), ) + + expect(chunks).toContainEqual({ + type: "tool_call_partial", + index: 0, + id: "call_1", + name: "get_weather", + arguments: '{"city": "NYC"}', + }) + expect(chunks).toContainEqual({ type: "text", text: "done" }) + }) + + it("skips frames without a first choice", async () => { + sharedMockCreate.mockResolvedValue( + asyncStreamFrom([{ choices: [] }, { choices: [{ delta: { content: "hi" } }] }]), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const chunks = await collectStream( + handler.createMessage("system", [{ role: "user", content: "hi" }], { taskId: "t", tools: [] }), + ) + + expect(chunks).toEqual([{ type: "text", text: "hi" }]) + }) + + it("ignores a non-array tool_calls field on the delta", async () => { + sharedMockCreate.mockResolvedValue( + asyncStreamFrom([{ choices: [{ delta: { content: "hi", tool_calls: null } }] }]), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const chunks = await collectStream( + handler.createMessage("system", [{ role: "user", content: "hi" }], { taskId: "t", tools: [] }), + ) + + expect(chunks).toEqual([{ type: "text", text: "hi" }]) + }) + + it("emits a partial tool call with undefined name and arguments when function is absent", async () => { + sharedMockCreate.mockResolvedValue( + asyncStreamFrom([{ choices: [{ delta: { tool_calls: [{ index: 0, id: "call_1" }] } }] }]), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const chunks = await collectStream( + handler.createMessage("system", [{ role: "user", content: "hi" }], { taskId: "t", tools: [] }), + ) + + expect(chunks).toEqual([{ type: "tool_call_partial", index: 0, id: "call_1" }]) + }) + + it("keeps the last reported usage when a later frame carries none", async () => { + sharedMockCreate.mockResolvedValue( + asyncStreamFrom([ + { choices: [{ delta: { content: "hi" } }], usage: { prompt_tokens: 1, completion_tokens: 2 } }, + { choices: [{ delta: {} }] }, + ]), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const chunks = await collectStream( + handler.createMessage("system", [{ role: "user", content: "hi" }], { taskId: "t", tools: [] }), + ) + + expect(chunks).toEqual([ + { type: "text", text: "hi" }, + expect.objectContaining({ type: "usage", inputTokens: 1, outputTokens: 2 }), + ]) + }) + + it("emits no usage chunk when the stream reports none", async () => { + sharedMockCreate.mockResolvedValue(asyncStreamFrom([{ choices: [{ delta: { content: "hi" } }] }])) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const chunks = await collectStream( + handler.createMessage("system", [{ role: "user", content: "hi" }], { taskId: "t", tools: [] }), + ) + + expect(chunks).toEqual([{ type: "text", text: "hi" }]) }) it("completePrompt returns the response text", async () => { @@ -199,6 +370,371 @@ describe("UnboundHandler", () => { expect.objectContaining({ messages: [{ role: "system", content: "Write a haiku" }], }), + {}, ) }) + + it("completePrompt should pass abort signal through to client", async () => { + const controller = new AbortController() + sharedMockCreate.mockResolvedValue({ + choices: [{ message: { content: "completed text" } }], + }) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + await handler.completePrompt("Write a haiku", { abortSignal: controller.signal }) + expect(sharedMockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ signal: controller.signal }), + ) + }) + + it("completePrompt should pass timeout through to client", async () => { + sharedMockCreate.mockResolvedValue({ + choices: [{ message: { content: "completed text" } }], + }) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + await handler.completePrompt("Write a haiku", { timeoutMs: 5000 }) + expect(sharedMockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 5000 }), + ) + }) + + it("completePrompt should omit the timeout option when timeoutMs is 0", async () => { + // The OpenAI SDK treats timeout: 0 as an immediate abort, so the + // "disabled" value must never be forwarded — assert the absence of + // the option (a forwarded timeout: 0 would fail this assertion). + sharedMockCreate.mockResolvedValue({ + choices: [{ message: { content: "completed text" } }], + }) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + await handler.completePrompt("Write a haiku", { timeoutMs: 0 }) + const call = sharedMockCreate.mock.calls[sharedMockCreate.mock.calls.length - 1] + const requestOptions = call[1] as { timeout?: number } | undefined + expect(requestOptions).not.toHaveProperty("timeout") + }) + + it("completePrompt should preserve abort identity when the caller aborts", async () => { + // Emulate the OpenAI SDK: an aborted request signal rejects with + // APIUserAbortError ("Request was aborted." — the trailing period would + // fail task-level abort detection, so the provider must normalize it). + sharedMockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + throw new APIUserAbortError() + } + throw new Error("boom") + }) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + const controller = new AbortController() + controller.abort() + + const error = await handler.completePrompt("Write a haiku", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Unbound request was aborted") + }) + + it("completePrompt should surface request timeouts as an AbortError", async () => { + // Emulate the OpenAI SDK: when the request timeout fires, the SDK + // surfaces APIConnectionTimeoutError ("Request timed out.") once retries + // are exhausted — verified against openai v5.23.2 against a hung server. + sharedMockCreate.mockImplementation(async (_params: unknown, options: { timeout?: number }) => { + await new Promise((resolve) => setTimeout(resolve, options?.timeout ?? 50)) + throw new APIConnectionTimeoutError() + }) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await handler.completePrompt("Write a haiku", { timeoutMs: 50 }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Unbound request was aborted") + }) + + it("completePrompt should preserve abort identity when the signal is pre-aborted with a plain error", async () => { + // The aborted-signal disjunct alone must normalize a plain rejection + // (not just SDK abort classes) to the DOM-standard AbortError. + sharedMockCreate.mockRejectedValueOnce(new Error("boom")) + const controller = new AbortController() + controller.abort() + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await handler.completePrompt("Write a haiku", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Unbound request was aborted") + }) + + it("completePrompt should preserve abort identity for a name-based AbortError rejection", async () => { + // No aborted signal and no SDK abort class: only the DOM-standard + // name === "AbortError" check marks a cancelled request. + sharedMockCreate.mockRejectedValueOnce(Object.assign(new Error("raw"), { name: "AbortError" })) + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await handler.completePrompt("Write a haiku").then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Unbound request was aborted") + }) + + it("completePrompt should wrap a plain rejection when no options are provided", async () => { + // No options at all: options?.abortSignal must tolerate an undefined + // options argument and the rejection must surface as the wrapped + // completion error. + sharedMockCreate.mockRejectedValueOnce(new Error("boom")) + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await handler.completePrompt("Write a haiku").then( + () => undefined, + (e: unknown) => e, + ) + expect((error as Error).message).toBe("Unbound completion error: boom") + }) + + it("completePrompt should wrap a non-Error rejection with its object string", async () => { + // The 4th disjunct must require an actual Error instance: a plain + // object with name === "AbortError" is not a cancelled request and + // must go through the completion-error wrapping path. + sharedMockCreate.mockRejectedValueOnce({ name: "AbortError" }) + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await handler.completePrompt("Write a haiku").then( + () => undefined, + (e: unknown) => e, + ) + expect((error as Error).message).toBe("Unbound completion error: [object Object]") + }) + it("completePrompt should work without options (backward compatible)", async () => { + sharedMockCreate.mockResolvedValue({ + choices: [{ message: { content: "completed text" } }], + }) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const result = await handler.completePrompt("Write a haiku") + expect(result).toBe("completed text") + }) + + describe("createMessage abort signal bridging", () => { + it("rejects the request with an AbortError when the external signal is already aborted", async () => { + let requestError: unknown + let capturedSignal: AbortSignal | undefined + sharedMockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + // The real SDK rejects with an AbortError when its request signal is aborted. + capturedSignal = options?.signal + requestError = new DOMException("The operation was aborted.", "AbortError") + throw requestError + }) + + const controller = new AbortController() + controller.abort() + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const stream = handler.createMessage( + "system", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + // The bridge surfaces a DOM-standard AbortError (series standard) + // instead of the wrapped completion error. + await expect(collectStream(stream)).rejects.toMatchObject({ + name: "AbortError", + message: "The Unbound request was aborted", + }) + // An already-aborted external signal must abort the INTERNAL + // controller before the request starts. + expect(capturedSignal?.aborted).toBe(true) + expect(requestError).toMatchObject({ name: "AbortError" }) + }) + + it("aborts the in-flight request when the external signal fires mid-stream", async () => { + // The mock polls the INTERNAL controller signal (bounded 40x5ms) + // instead of waiting for an "abort" event, so the test can never + // hang if the bridge stops forwarding aborts. + let capturedSignal: AbortSignal | undefined + sharedMockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + return (async function* () { + yield { choices: [{ delta: { content: "partial" } }] } + for (let i = 0; i < 40 && !capturedSignal?.aborted; i++) { + await new Promise((resolve) => setTimeout(resolve, 5)) + } + if (capturedSignal?.aborted) { + throw new Error("boom") + } + })() + }) + + const controller = new AbortController() + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const consumed = collectStream( + handler.createMessage( + "system", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + + // Let the request start and the first chunk be yielded before aborting. + await new Promise((resolve) => setTimeout(resolve, 25)) + controller.abort() + + // createMessage's stream loop has no catch: the raw SDK rejection + // propagates once the bridge aborts the in-flight request. + const error = await consumed.then( + () => undefined, + (e: unknown) => e, + ) + expect(capturedSignal?.aborted).toBe(true) + expect((error as Error).message).toBe("boom") + }) + + it("detaches the bridged abort listener when the request completes normally", async () => { + // The listener is added with { once: true }, so it only detaches on + // abort. A task-scoped signal spanning many requests must not + // accumulate a listener per request: assert explicit removal after a + // normal (non-aborted) completion. + sharedMockCreate.mockImplementation(async () => + asyncStreamFrom([ + { choices: [{ delta: { content: "ok" } }] }, + { choices: [{ delta: {} }], usage: { prompt_tokens: 1, completion_tokens: 1 } }, + ]), + ) + + const controller = new AbortController() + const removeListenerSpy = vi.spyOn(controller.signal, "removeEventListener") + const addEventListenerSpy = vi.spyOn(controller.signal, "addEventListener") + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const chunks = await collectStream( + handler.createMessage( + "system", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + + expect(chunks).toContainEqual({ type: "text", text: "ok" }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + // The listener is registered with { once: true } — assert the exact + // options so a bridge that drops them (and relies on the finally + // block alone for single-shot semantics) is caught. + expect(addEventListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function), { once: true }) + expect(controller.signal.aborted).toBe(false) + }) + + it("streams normally when called without metadata", async () => { + // metadata?.abortSignal must tolerate a missing metadata argument. + sharedMockCreate.mockImplementation(async () => + asyncStreamFrom([{ choices: [{ delta: { content: "hi" } }] }]), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const chunks = await collectStream(handler.createMessage("system", [{ role: "user", content: "hi" }])) + + expect(chunks).toEqual([{ type: "text", text: "hi" }]) + }) + + it("preserves abort identity when the SDK rejects with APIUserAbortError and no signal is aborted", async () => { + // No external signal: the aborted-controller disjunct is false, so + // the APIUserAbortError disjunct alone must normalize the rejection + // to the DOM-standard AbortError. + sharedMockCreate.mockRejectedValueOnce(new APIUserAbortError()) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await collectStream( + handler.createMessage("system", [{ role: "user", content: "hi" }], { taskId: "t", tools: [] }), + ).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Unbound request was aborted") + }) + + it("preserves abort identity when the SDK rejects with a name-based AbortError", async () => { + // No SDK abort class and no aborted signal: only the DOM-standard + // name === "AbortError" check marks a cancelled pre-stream request. + sharedMockCreate.mockRejectedValueOnce(Object.assign(new Error("raw"), { name: "AbortError" })) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await collectStream( + handler.createMessage("system", [{ role: "user", content: "hi" }], { taskId: "t", tools: [] }), + ).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Unbound request was aborted") + }) + }) }) diff --git a/src/api/providers/__tests__/vercel-ai-gateway.spec.ts b/src/api/providers/__tests__/vercel-ai-gateway.spec.ts index b238d72381..bf7433338a 100644 --- a/src/api/providers/__tests__/vercel-ai-gateway.spec.ts +++ b/src/api/providers/__tests__/vercel-ai-gateway.spec.ts @@ -10,10 +10,10 @@ vitest.mock("vscode", () => ({ })) import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI from "openai" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { VercelAiGatewayHandler } from "../vercel-ai-gateway" -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 { vercelAiGatewayDefaultModelId, VERCEL_AI_GATEWAY_DEFAULT_TEMPERATURE } from "@roo-code/types" @@ -273,6 +273,55 @@ describe("VercelAiGatewayHandler", () => { }).rejects.toThrow("Vercel AI Gateway stream error") }) + it("throws the default message when an in-stream error chunk has an empty message", async () => { + // An empty message must not be forwarded — it would become + // Error("") with no diagnostic at all. + mockCreate.mockImplementation(async () => asyncStreamFrom([{ error: { message: "" } }])) + + const handler = new VercelAiGatewayHandler(mockOptions) + const stream = handler.createMessage("You are a helpful assistant.", [{ role: "user", content: "Hello" }]) + + await expect(async () => { + await collectStream(stream) + }).rejects.toThrow("Vercel AI Gateway stream error") + }) + + it("treats a present-but-undefined error key as no error", async () => { + mockCreate.mockImplementation(async () => + asyncStreamFrom([{ error: undefined, choices: [{ delta: { content: "hi" }, index: 0 }], index: 0 }]), + ) + + const handler = new VercelAiGatewayHandler(mockOptions) + const stream = handler.createMessage("You are a helpful assistant.", [{ role: "user", content: "Hello" }]) + + const chunks = await collectStream(stream) + expect(chunks).toEqual([{ type: "text", text: "hi" }]) + }) + + it("skips frames without a delta and tool calls without a function field", async () => { + // Full-list assertion: a frame without choices[0], a choice without + // a delta, and a tool call missing function must each contribute + // nothing except the partial tool call with undefined name/arguments + // (toEqual ignores undefined-valued keys). + mockCreate.mockImplementation(async () => + asyncStreamFrom([ + { choices: [], index: 0 }, + { choices: [{}], index: 0 }, + { choices: [{ delta: { content: "hi" }, index: 0 }], index: 0 }, + { choices: [{ delta: { tool_calls: [{ index: 0, id: "call_1" }] }, index: 0 }], index: 0 }, + ]), + ) + + const handler = new VercelAiGatewayHandler(mockOptions) + const stream = handler.createMessage("You are a helpful assistant.", [{ role: "user", content: "Hello" }]) + + const chunks = await collectStream(stream) + expect(chunks).toEqual([ + { type: "text", text: "hi" }, + { type: "tool_call_partial", index: 0, id: "call_1" }, + ]) + }) + it("uses correct temperature from options", async () => { const customTemp = 0.5 const handler = new VercelAiGatewayHandler( @@ -291,6 +340,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ temperature: customTemp, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -306,6 +356,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ temperature: VERCEL_AI_GATEWAY_DEFAULT_TEMPERATURE, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -325,6 +376,7 @@ describe("VercelAiGatewayHandler", () => { temperature: undefined, max_completion_tokens: 128000, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -397,6 +449,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ max_completion_tokens: 64000, // max tokens for sonnet 4 }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -472,6 +525,7 @@ describe("VercelAiGatewayHandler", () => { }), ]), }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -489,6 +543,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ tool_choice: "auto", }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -506,6 +561,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ parallel_tool_calls: true, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -523,6 +579,7 @@ describe("VercelAiGatewayHandler", () => { tools: expect.any(Array), parallel_tool_calls: true, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -619,6 +676,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ stream_options: { include_usage: true }, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) }) @@ -657,6 +715,7 @@ describe("VercelAiGatewayHandler", () => { temperature: VERCEL_AI_GATEWAY_DEFAULT_TEMPERATURE, max_completion_tokens: 64000, }), + undefined, ) }) @@ -675,6 +734,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ temperature: customTemp, }), + undefined, ) }) @@ -707,10 +767,290 @@ describe("VercelAiGatewayHandler", () => { const result = await handler.completePrompt("Test") expect(result).toBe("") }) + + it("should pass abort signal through to client", async () => { + const handler = new VercelAiGatewayHandler(mockOptions) + const controller = new AbortController() + mockCreate.mockImplementation(async () => ({ + choices: [ + { + message: { role: "assistant", content: "response" }, + finish_reason: "stop", + index: 0, + }, + ], + })) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ signal: controller.signal }), + ) + }) + + it("should pass timeout through to client", async () => { + const handler = new VercelAiGatewayHandler(mockOptions) + mockCreate.mockImplementation(async () => ({ + choices: [ + { + message: { role: "assistant", content: "response" }, + finish_reason: "stop", + index: 0, + }, + ], + })) + + await handler.completePrompt("test prompt", { timeoutMs: 5000 }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 5000 }), + ) + }) + + it("should omit the timeout option when timeoutMs is 0", async () => { + // The OpenAI SDK treats timeout: 0 as an immediate abort, so the + // "disabled" value must never be forwarded — assert the absence of + // the option (a forwarded timeout: 0 would fail this assertion). + const handler = new VercelAiGatewayHandler(mockOptions) + mockCreate.mockImplementation(async () => ({ + choices: [ + { + message: { role: "assistant", content: "response" }, + finish_reason: "stop", + index: 0, + }, + ], + })) + + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + const call = mockCreate.mock.calls[mockCreate.mock.calls.length - 1] + const requestOptions = call[1] as { timeout?: number } | undefined + // When no option is forwarded the provider omits the SDK options + // argument entirely, so absence means: undefined arg OR an arg + // without a timeout key. + expect(Object.keys(requestOptions ?? {})).not.toContain("timeout") + }) + + it("should preserve abort identity when the caller aborts", async () => { + // Emulate the OpenAI SDK: an aborted request signal rejects with + // APIUserAbortError ("Request was aborted." — the trailing period would + // fail task-level abort detection, so the provider must normalize it). + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + throw new APIUserAbortError() + } + throw new Error("boom") + }) + + const handler = new VercelAiGatewayHandler(mockOptions) + const controller = new AbortController() + controller.abort() + + const error = await handler.completePrompt("test prompt", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Vercel AI Gateway request was aborted") + }) + + it("should surface request timeouts as an AbortError", async () => { + // Emulate the OpenAI SDK: when the request timeout fires, the SDK + // surfaces APIConnectionTimeoutError ("Request timed out.") once retries + // are exhausted — verified against openai v5.23.2 against a hung server. + mockCreate.mockImplementation(async (_params: unknown, options: { timeout?: number }) => { + await new Promise((resolve) => setTimeout(resolve, options?.timeout ?? 50)) + throw new APIConnectionTimeoutError() + }) + + const handler = new VercelAiGatewayHandler(mockOptions) + + const error = await handler.completePrompt("test prompt", { timeoutMs: 50 }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Vercel AI Gateway request was aborted") + }) + + it("should preserve abort identity when the signal is pre-aborted with a plain error", async () => { + // The aborted-signal disjunct alone must normalize a plain + // rejection (not just SDK abort classes) to the DOM-standard + // AbortError. + mockCreate.mockRejectedValueOnce(new Error("boom")) + const controller = new AbortController() + controller.abort() + const handler = new VercelAiGatewayHandler(mockOptions) + + const error = await handler.completePrompt("test prompt", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Vercel AI Gateway request was aborted") + }) + + it("should preserve abort identity for a name-based AbortError rejection", async () => { + // No aborted signal and no SDK abort class: only the DOM-standard + // name === "AbortError" check marks a cancelled request. + mockCreate.mockRejectedValueOnce(Object.assign(new Error("raw"), { name: "AbortError" })) + const handler = new VercelAiGatewayHandler(mockOptions) + + const error = await handler.completePrompt("test prompt").then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Vercel AI Gateway request was aborted") + }) + it("should work without options (backward compatible)", async () => { + const handler = new VercelAiGatewayHandler(mockOptions) + mockCreate.mockImplementation(async () => ({ + choices: [ + { + message: { role: "assistant", content: "response" }, + finish_reason: "stop", + index: 0, + }, + ], + })) + + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + }) + }) + + describe("createMessage abort signal bridging", () => { + it("rejects with an AbortError when the external signal is already aborted", async () => { + let capturedSignal: AbortSignal | undefined + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + throw new DOMException("The operation was aborted.", "AbortError") + }) + + const handler = new VercelAiGatewayHandler(mockOptions) + const controller = new AbortController() + controller.abort() + + const stream = handler.createMessage( + "test prompt", + [{ role: "user", content: "hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + // An already-aborted external signal must abort the INTERNAL + // controller before the request starts. + const error = await collectStream(stream).then( + () => undefined, + (e: unknown) => e, + ) + expect(capturedSignal?.aborted).toBe(true) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Vercel AI Gateway request was aborted") + }) + + it("aborts the in-flight request when the external signal fires mid-stream", async () => { + // The mock polls the INTERNAL controller signal (bounded 40x5ms) + // instead of waiting for an "abort" event, so the test can never + // hang if the bridge stops forwarding aborts. + let capturedSignal: AbortSignal | undefined + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + return (async function* () { + yield { choices: [{ delta: { content: "partial" }, index: 0 }], index: 0 } + for (let i = 0; i < 40 && !capturedSignal?.aborted; i++) { + await new Promise((resolve) => setTimeout(resolve, 5)) + } + if (capturedSignal?.aborted) { + throw new DOMException("The operation was aborted.", "AbortError") + } + })() + }) + + const handler = new VercelAiGatewayHandler(mockOptions) + const controller = new AbortController() + + const consumed = collectStream( + handler.createMessage( + "test prompt", + [{ role: "user", content: "hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + + // Let the request start and the first chunk be yielded before aborting. + await new Promise((resolve) => setTimeout(resolve, 25)) + controller.abort() + + const error = await consumed.then( + () => undefined, + (e: unknown) => e, + ) + expect(capturedSignal?.aborted).toBe(true) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Vercel AI Gateway request was aborted") + }) + + it("detaches the bridged abort listener when the request completes normally", async () => { + // The listener is added with { once: true }, so it only detaches on + // abort. A task-scoped signal spanning many requests must not + // accumulate a listener per request: assert explicit removal after a + // normal (non-aborted) completion. + mockCreate.mockImplementation(async () => + asyncStreamFrom([ + { + choices: [{ delta: { content: "ok" }, index: 0 }], + index: 0, + }, + { + choices: [{ delta: {}, index: 0 }], + index: 0, + usage: { prompt_tokens: 2, completion_tokens: 3 }, + }, + ]), + ) + + const handler = new VercelAiGatewayHandler(mockOptions) + const controller = new AbortController() + const removeListenerSpy = vi.spyOn(controller.signal, "removeEventListener") + const addEventListenerSpy = vi.spyOn(controller.signal, "addEventListener") + + const stream = handler.createMessage( + "test prompt", + [{ role: "user", content: "hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const chunks = await collectStream(stream) + expect(chunks).toContainEqual({ type: "text", text: "ok" }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + // The listener is registered with { once: true } — assert the exact + // options so a bridge that drops them (and relies on the finally + // block alone for single-shot semantics) is caught. + expect(addEventListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function), { once: true }) + expect(controller.signal.aborted).toBe(false) + }) }) describe("temperature support", () => { it("applies temperature for supported models", async () => { + // Pin the response: a later describe's mock implementation may have + // left the shared mock in a state this test does not expect. + mockCreate.mockResolvedValueOnce({ + choices: [ + { + message: { role: "assistant", content: "Test completion response" }, + finish_reason: "stop", + index: 0, + }, + ], + usage: { + prompt_tokens: 8, + completion_tokens: 4, + total_tokens: 12, + }, + }) + const handler = new VercelAiGatewayHandler( makeApiHandlerOptions({ ...mockOptions, @@ -725,6 +1065,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ temperature: 0.9, }), + undefined, ) }) }) diff --git a/src/api/providers/__tests__/zoo-gateway.spec.ts b/src/api/providers/__tests__/zoo-gateway.spec.ts index c6f4c15c1e..7042dd3110 100644 --- a/src/api/providers/__tests__/zoo-gateway.spec.ts +++ b/src/api/providers/__tests__/zoo-gateway.spec.ts @@ -26,7 +26,7 @@ vitest.mock("../../../i18n", () => ({ t: (key: string) => key, })) -import OpenAI from "openai" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { zooGatewayDefaultModelId, ZOO_GATEWAY_DEFAULT_TEMPERATURE } from "@roo-code/types" @@ -36,6 +36,7 @@ import { Package } from "../../../shared/package" import { clearZooCodeToken } from "../../../services/zoo-code-auth" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { clearAllMocks } from "../../../test-utils/reset" +import { makeCreateMessageMetadata } from "../../../test-utils/api" vitest.mock("openai") vitest.mock("delay", () => ({ @@ -375,6 +376,7 @@ describe("ZooGatewayHandler", () => { "X-Zoo-Task-ID": "task-123", "X-Zoo-Mode": "code", }, + signal: expect.any(AbortSignal), }), ) }) @@ -520,6 +522,7 @@ describe("ZooGatewayHandler", () => { temperature: ZOO_GATEWAY_DEFAULT_TEMPERATURE, max_completion_tokens: 64000, }), + {}, ) }) @@ -542,8 +545,245 @@ describe("ZooGatewayHandler", () => { await expect(handler.completePrompt("Test")).resolves.toBe("") }) + + it("should pass abort signal through to client", async () => { + const handler = new ZooGatewayHandler(mockOptions) + const controller = new AbortController() + mockCreate.mockImplementation(async () => ({ + choices: [{ message: { role: "assistant", content: "response" } }], + })) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ signal: controller.signal }), + ) + }) + + it("should pass timeout through to client", async () => { + const handler = new ZooGatewayHandler(mockOptions) + mockCreate.mockImplementation(async () => ({ + choices: [{ message: { role: "assistant", content: "response" } }], + })) + + await handler.completePrompt("test prompt", { timeoutMs: 5000 }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 5000 }), + ) + }) + + it("should omit the timeout option when timeoutMs is 0", async () => { + // The OpenAI SDK treats timeout: 0 as an immediate abort, so the + // "disabled" value must never be forwarded — assert the absence of + // the option (a forwarded timeout: 0 would fail this assertion). + const handler = new ZooGatewayHandler(mockOptions) + mockCreate.mockImplementation(async () => ({ + choices: [{ message: { role: "assistant", content: "response" } }], + })) + + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + const call = mockCreate.mock.calls[mockCreate.mock.calls.length - 1] + const requestOptions = call[1] as { timeout?: number } | undefined + expect(requestOptions).not.toHaveProperty("timeout") + }) + + it("should preserve abort identity when the caller aborts", async () => { + // Emulate the OpenAI SDK: an aborted request signal rejects with + // APIUserAbortError ("Request was aborted." — the trailing period would + // fail task-level abort detection, so the provider must normalize it). + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + throw new APIUserAbortError() + } + throw new Error("boom") + }) + + const handler = new ZooGatewayHandler(mockOptions) + const controller = new AbortController() + controller.abort() + + const error = await handler.completePrompt("test prompt", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + }) + + it("should surface request timeouts as an AbortError", async () => { + // Emulate the OpenAI SDK: when the request timeout fires, the SDK + // surfaces APIConnectionTimeoutError ("Request timed out.") once retries + // are exhausted — verified against openai v5.23.2 against a hung server. + mockCreate.mockImplementation(async (_params: unknown, options: { timeout?: number }) => { + await new Promise((resolve) => setTimeout(resolve, options?.timeout ?? 50)) + throw new APIConnectionTimeoutError() + }) + + const handler = new ZooGatewayHandler(mockOptions) + + const error = await handler.completePrompt("test prompt", { timeoutMs: 50 }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + }) + + it("should preserve abort identity when the signal is pre-aborted with a plain error", async () => { + // The aborted-signal disjunct alone must normalize a plain + // rejection (not just SDK abort classes) to the DOM-standard + // AbortError. + mockCreate.mockRejectedValueOnce(new Error("boom")) + const controller = new AbortController() + controller.abort() + const handler = new ZooGatewayHandler(mockOptions) + + const error = await handler.completePrompt("test prompt", { abortSignal: controller.signal }).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + }) + + it("should preserve abort identity for a name-based AbortError rejection", async () => { + // No aborted signal and no SDK abort class: only the DOM-standard + // name === "AbortError" check marks a cancelled request. + mockCreate.mockRejectedValueOnce(Object.assign(new Error("raw"), { name: "AbortError" })) + const handler = new ZooGatewayHandler(mockOptions) + + const error = await handler.completePrompt("test prompt").then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + }) + it("should work without options (backward compatible)", async () => { + const handler = new ZooGatewayHandler(mockOptions) + mockCreate.mockImplementation(async () => ({ + choices: [{ message: { role: "assistant", content: "response" } }], + })) + + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + }) }) + describe("createMessage abort signal bridging", () => { + it("rejects with an AbortError when the external signal is already aborted", async () => { + let capturedSignal: AbortSignal | undefined + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + throw new Error("boom") + }) + + const handler = new ZooGatewayHandler(mockOptions) + const controller = new AbortController() + controller.abort() + + const stream = handler.createMessage( + "prompt", + [{ role: "user", content: "hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + // An already-aborted external signal must abort the INTERNAL + // controller before the request starts; the catch path then + // normalizes the rejection to the DOM-standard AbortError before + // the gateway error surfacing path. + const error = await collectStream(stream).then( + () => undefined, + (e: unknown) => e, + ) + expect(capturedSignal?.aborted).toBe(true) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + }) + + it("aborts the in-flight request when the external signal fires mid-stream", async () => { + // The mock polls the INTERNAL controller signal (bounded 40x5ms) + // instead of waiting for an "abort" event, so the test can never + // hang if the bridge stops forwarding aborts. + let capturedSignal: AbortSignal | undefined + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + return (async function* () { + yield { choices: [{ delta: { content: "partial" }, index: 0 }], index: 0 } + for (let i = 0; i < 40 && !capturedSignal?.aborted; i++) { + await new Promise((resolve) => setTimeout(resolve, 5)) + } + if (capturedSignal?.aborted) { + throw new DOMException("The operation was aborted.", "AbortError") + } + })() + }) + + const handler = new ZooGatewayHandler(mockOptions) + const controller = new AbortController() + + const consumed = collectStream( + handler.createMessage( + "prompt", + [{ role: "user", content: "hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + + // Let the request start and the first chunk be yielded before aborting. + await new Promise((resolve) => setTimeout(resolve, 25)) + controller.abort() + + const error = await consumed.then( + () => undefined, + (e: unknown) => e, + ) + expect(capturedSignal?.aborted).toBe(true) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + }) + + it("detaches the bridged abort listener when the request completes normally", async () => { + // The listener is added with { once: true }, so it only detaches on + // abort. A task-scoped signal spanning many requests must not + // accumulate a listener per request: assert explicit removal after a + // normal (non-aborted) completion. + mockCreate.mockImplementation(async () => + asyncStreamFrom([ + { + choices: [{ delta: { content: "ok" }, index: 0 }], + index: 0, + }, + { + choices: [{ delta: {}, index: 0 }], + index: 0, + usage: { prompt_tokens: 2, completion_tokens: 3 }, + }, + ]), + ) + + const handler = new ZooGatewayHandler(mockOptions) + const controller = new AbortController() + const removeListenerSpy = vi.spyOn(controller.signal, "removeEventListener") + const addEventListenerSpy = vi.spyOn(controller.signal, "addEventListener") + + const stream = handler.createMessage( + "prompt", + [{ role: "user", content: "hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const chunks = await collectStream(stream) + expect(chunks).toContainEqual({ type: "text", text: "ok" }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function)) + // The listener is registered with { once: true } — assert the exact + // options so a bridge that drops them (and relies on the finally + // block alone for single-shot semantics) is caught. + expect(addEventListenerSpy).toHaveBeenCalledWith("abort", expect.any(Function), { once: true }) + expect(controller.signal.aborted).toBe(false) + }) + }) describe("classifyGatewayApiError", () => { it("returns sign_in on 401", () => { expect(classifyGatewayApiError(makeApiError(401))).toEqual({ kind: "sign_in" }) diff --git a/src/api/providers/opencode-go.ts b/src/api/providers/opencode-go.ts index 1f4faa45d6..c0a7a30246 100644 --- a/src/api/providers/opencode-go.ts +++ b/src/api/providers/opencode-go.ts @@ -1,6 +1,10 @@ -import { Anthropic } from "@anthropic-ai/sdk" +import { + Anthropic, + APIConnectionTimeoutError as AnthropicTimeoutError, + APIUserAbortError as AnthropicAbortError, +} from "@anthropic-ai/sdk" import { CacheControlEphemeral } from "@anthropic-ai/sdk/resources" -import OpenAI from "openai" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { type ModelInfo, @@ -32,6 +36,7 @@ import { convertOpenAIToolsToAnthropic, convertOpenAIToolChoiceToAnthropic, } from "../../core/prompts/tools/native-tools/converters" +import { createAbortError } from "./utils/abort-signal" /** * The wire formats exposed by the Opencode Go gateway: @@ -200,8 +205,40 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio ): ApiStream { const { id: modelId, info, format, temperature, reasoningEffort, maxTokens } = await this.resolveModel() + // Per-request controller so an external abort signal (e.g. task + // cancellation) can interrupt the in-flight streaming request. + // Bridge it to our controller using the Bedrock pattern: + // - pre-aborted guard: check if already aborted before adding listener + // - { once: true }: remove listener after first abort to avoid leaks + // The listener is stored so it can be detached when the request ends: + // { once: true } only removes it on abort, so a task-scoped signal + // would otherwise accumulate one listener per request. + const controller = new AbortController() + const externalAbortSignal = metadata?.abortSignal + const abortListener = () => controller.abort() + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + controller.abort() + } else { + externalAbortSignal.addEventListener("abort", abortListener, { once: true }) + } + } + if (format === "anthropic") { - yield* this.streamAnthropicMessage(modelId, info, temperature, maxTokens, systemPrompt, messages, metadata) + try { + yield* this.streamAnthropicMessage( + modelId, + info, + temperature, + maxTokens, + systemPrompt, + messages, + controller.signal, + metadata, + ) + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) + } return } @@ -247,42 +284,54 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio }), } - const completion = await this.client.chat.completions.create(body) + try { + const completion = await this.client.chat.completions.create(body, { signal: controller.signal }) - for await (const chunk of completion) { - const delta = chunk.choices[0]?.delta + for await (const chunk of completion) { + const delta = chunk.choices[0]?.delta - // Several Go-plan models (GLM, DeepSeek) stream reasoning via this field. - const reasoningText = extractReasoningFromDelta(delta) - if (reasoningText) { - yield { type: "reasoning", text: reasoningText } - } + // Several Go-plan models (GLM, DeepSeek) stream reasoning via this field. + const reasoningText = extractReasoningFromDelta(delta) + if (reasoningText) { + yield { type: "reasoning", text: reasoningText } + } - if (delta?.content) { - yield { type: "text", text: delta.content } - } + if (delta?.content) { + yield { type: "text", text: delta.content } + } - // Emit raw tool call chunks - NativeToolCallParser handles state management. - if (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, + // Emit raw tool call chunks - NativeToolCallParser handles state management. + if (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) { - yield { - type: "usage", - inputTokens: chunk.usage.prompt_tokens || 0, - outputTokens: chunk.usage.completion_tokens || 0, - cacheReadTokens: chunk.usage.prompt_tokens_details?.cached_tokens || undefined, + if (chunk.usage) { + yield { + type: "usage", + inputTokens: chunk.usage.prompt_tokens || 0, + outputTokens: chunk.usage.completion_tokens || 0, + cacheReadTokens: chunk.usage.prompt_tokens_details?.cached_tokens || undefined, + } } } + } catch (error) { + // Preserve abort identity (series standard): surface a cancelled + // request as a DOM-standard AbortError rather than leaking the + // raw SDK abort error. + if (controller.signal.aborted) { + throw createAbortError("Opencode Go") + } + throw error + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) } } @@ -455,6 +504,7 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio maxTokens: number | undefined, systemPrompt: string, messages: Anthropic.Messages.MessageParam[], + abortSignal: AbortSignal, metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { const cacheControl: CacheControlEphemeral = { type: "ephemeral" } @@ -505,8 +555,18 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio // errors propagate unchanged, matching the OpenAI streaming path. let stream try { - stream = await this.anthropicClient.messages.create(requestParams) + stream = await this.anthropicClient.messages.create(requestParams, { signal: abortSignal }) } catch (error) { + // Preserve abort identity (series standard): a cancelled request + // must surface as a DOM-standard AbortError, not a wrapped + // completion error. + if ( + abortSignal.aborted || + error instanceof AnthropicAbortError || + (error instanceof Error && error.name === "AbortError") + ) { + throw createAbortError("Opencode Go") + } if (error instanceof Error) { throw new Error(`Opencode Go completion error: ${error.message}`) } @@ -691,24 +751,53 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio if (format === "anthropic") { try { - const message = await this.anthropicClient.messages.create({ - model: modelId, - // Honour the same includeMaxTokens/modelMaxTokens override - // logic as the streaming path so non-streaming completions - // respect the user's max-output slider instead of always - // falling back to the model default. - max_tokens: - this.options.includeMaxTokens === true - ? this.options.modelMaxTokens || maxTokens || 16_384 - : (maxTokens ?? 16_384), - temperature: this.supportsTemperature(modelId) ? (temperature ?? 1.0) : undefined, - messages: [{ role: "user", content: prompt }], - stream: false, - }) + // Build request options with abortSignal and/or timeout handling. + // timeoutMs <= 0 means "no explicit timeout": omit the SDK timeout + // option entirely — the SDKs treat timeout: 0 as an immediate + // abort, which would cancel the request right away. + const requestOptions: Anthropic.RequestOptions = {} + if (options?.abortSignal) { + requestOptions.signal = options.abortSignal + } + if (options?.timeoutMs !== undefined && options.timeoutMs > 0) { + requestOptions.timeout = options.timeoutMs + } + + const message = await this.anthropicClient.messages.create( + { + model: modelId, + // Honour the same includeMaxTokens/modelMaxTokens override + // logic as the streaming path so non-streaming completions + // respect the user's max-output slider instead of always + // falling back to the model default. + max_tokens: + this.options.includeMaxTokens === true + ? this.options.modelMaxTokens || maxTokens || 16_384 + : (maxTokens ?? 16_384), + temperature: this.supportsTemperature(modelId) ? (temperature ?? 1.0) : undefined, + messages: [{ role: "user", content: prompt }], + stream: false, + }, + Object.keys(requestOptions).length > 0 ? requestOptions : undefined, + ) const content = message.content.find(({ type }) => type === "text") return content?.type === "text" ? content.text : "" } catch (error) { + // Preserve abort identity (series standard): caller-initiated + // cancellations and request timeouts must surface as a + // DOM-standard AbortError, not a wrapped completion error. The + // Anthropic SDK reports both with messages ending in a period + // ("Request was aborted.", "Request timed out."), which would not + // match task-level abort detection (message ending in "aborted"). + if ( + options?.abortSignal?.aborted || + error instanceof AnthropicAbortError || + error instanceof AnthropicTimeoutError || + (error instanceof Error && error.name === "AbortError") + ) { + throw createAbortError("Opencode Go") + } if (error instanceof Error) { throw new Error(`Opencode Go completion error: ${error.message}`) } @@ -775,9 +864,35 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio reasoningEffort as OpenAI.Chat.ChatCompletionCreateParams["reasoning_effort"] } - const response = await this.client.chat.completions.create(requestOptions) + // Build request options with abortSignal and/or timeout for OpenAI path. + // timeoutMs <= 0 means "no explicit timeout": omit the SDK timeout + // option entirely — the OpenAI SDK treats timeout: 0 as an immediate + // abort, which would cancel the request right away. + const createOptions: OpenAI.RequestOptions = {} + if (options?.abortSignal) { + createOptions.signal = options.abortSignal + } + if (options?.timeoutMs !== undefined && options.timeoutMs > 0) { + createOptions.timeout = options.timeoutMs + } + + const response = await this.client.chat.completions.create(requestOptions, createOptions) return response.choices[0]?.message.content || "" } catch (error) { + // Preserve abort identity (series standard): caller-initiated + // cancellations and request timeouts must surface as a + // DOM-standard AbortError, not a wrapped completion error. The + // OpenAI SDK reports both with messages ending in a period + // ("Request was aborted.", "Request timed out."), which would not + // match task-level abort detection (message ending in "aborted"). + if ( + options?.abortSignal?.aborted || + error instanceof APIUserAbortError || + error instanceof APIConnectionTimeoutError || + (error instanceof Error && error.name === "AbortError") + ) { + throw createAbortError("Opencode Go") + } if (error instanceof Error) { throw new Error(`Opencode Go completion error: ${error.message}`) } diff --git a/src/api/providers/unbound.ts b/src/api/providers/unbound.ts index 61a2d1ae38..de1b0b0737 100644 --- a/src/api/providers/unbound.ts +++ b/src/api/providers/unbound.ts @@ -1,5 +1,5 @@ import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI from "openai" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { type ModelInfo, @@ -23,6 +23,7 @@ import { BaseProvider } from "./base-provider" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { handleOpenAIError } from "./utils/error-handler" import { applyRouterToolPreferences } from "./utils/router-tool-preferences" +import { createAbortError } from "./utils/abort-signal" import { extractReasoningFromDelta } from "./utils/extract-reasoning" // Unbound usage includes extra fields for Anthropic cache tokens. @@ -158,46 +159,79 @@ export class UnboundHandler extends BaseProvider implements SingleCompletionHand tool_choice: metadata?.tool_choice, } - let stream - try { - stream = await this.client.chat.completions.create(completionParams) - } catch (error) { - throw handleOpenAIError(error, this.providerName) + // Per-request controller so an external abort signal (e.g. task + // cancellation) can interrupt the in-flight streaming request. + // Bridge it to our controller using the Bedrock pattern: + // - pre-aborted guard: check if already aborted before adding listener + // - { once: true }: remove listener after first abort to avoid leaks + // The listener is stored so it can be detached when the request ends: + // { once: true } only removes it on abort, so a task-scoped signal + // would otherwise accumulate one listener per request. + const controller = new AbortController() + const externalAbortSignal = metadata?.abortSignal + const abortListener = () => controller.abort() + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + controller.abort() + } else { + externalAbortSignal.addEventListener("abort", abortListener, { once: true }) + } } - let lastUsage: any = undefined - - for await (const chunk of stream) { - const delta = chunk.choices[0]?.delta - const reasoningText = extractReasoningFromDelta(delta) - if (reasoningText) { - yield { type: "reasoning", text: reasoningText } + try { + let stream + try { + stream = await this.client.chat.completions.create(completionParams, { signal: controller.signal }) + } catch (error) { + // Preserve abort identity (series standard): a cancelled request + // must surface as a DOM-standard AbortError, not a wrapped + // completion error. + if ( + controller.signal.aborted || + error instanceof APIUserAbortError || + (error instanceof Error && error.name === "AbortError") + ) { + throw createAbortError("Unbound") + } + throw handleOpenAIError(error, this.providerName) } + let lastUsage: any = undefined - if (delta?.content) { - yield { type: "text", text: delta.content } - } + for await (const chunk of stream) { + const delta = chunk.choices[0]?.delta + + const reasoningText = extractReasoningFromDelta(delta) + if (reasoningText) { + yield { type: "reasoning", text: reasoningText } + } - // Handle native tool calls - if (delta && "tool_calls" in delta && Array.isArray(delta.tool_calls)) { - for (const toolCall of delta.tool_calls) { - yield { - type: "tool_call_partial", - index: toolCall.index, - id: toolCall.id, - name: toolCall.function?.name, - arguments: toolCall.function?.arguments, + if (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 + } } - } - if (lastUsage) { - yield this.processUsageMetrics(lastUsage, info) + if (lastUsage) { + yield this.processUsageMetrics(lastUsage, info) + } + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) } } @@ -212,11 +246,36 @@ export class UnboundHandler extends BaseProvider implements SingleCompletionHand messages: openAiMessages, temperature: temperature, } + // Build request options with abortSignal and/or timeout. + // timeoutMs <= 0 means "no explicit timeout": omit the SDK timeout + // option entirely — the OpenAI SDK treats timeout: 0 as an immediate + // abort, which would cancel the request right away. + const createOptions: OpenAI.RequestOptions = {} + if (options?.abortSignal) { + createOptions.signal = options.abortSignal + } + if (options?.timeoutMs !== undefined && options.timeoutMs > 0) { + createOptions.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) { + // Preserve abort identity (series standard): caller-initiated + // cancellations and request timeouts must surface as a + // DOM-standard AbortError, not a wrapped completion error. The + // OpenAI SDK reports both with messages ending in a period + // ("Request was aborted.", "Request timed out."), which would not + // match task-level abort detection (message ending in "aborted"). + if ( + options?.abortSignal?.aborted || + error instanceof APIUserAbortError || + error instanceof APIConnectionTimeoutError || + (error instanceof Error && error.name === "AbortError") + ) { + throw createAbortError("Unbound") + } throw handleOpenAIError(error, 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 ebc7edf3d3..1692f71e63 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,85 @@ 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") + expect((caught as Error).message).toBe("This operation was aborted") + }) + }) + + 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/api/providers/vercel-ai-gateway.ts b/src/api/providers/vercel-ai-gateway.ts index bf434e5a00..4139ce9561 100644 --- a/src/api/providers/vercel-ai-gateway.ts +++ b/src/api/providers/vercel-ai-gateway.ts @@ -1,5 +1,5 @@ import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI from "openai" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { vercelAiGatewayDefaultModelId, @@ -17,6 +17,7 @@ import { addCacheBreakpoints } from "../transform/caching/vercel-ai-gateway" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { RouterProvider } from "./router-provider" +import { createAbortError } from "./utils/abort-signal" // Extend OpenAI's CompletionUsage to include Vercel AI Gateway specific fields interface VercelAiGatewayUsage extends OpenAI.CompletionUsage { @@ -69,52 +70,83 @@ export class VercelAiGatewayHandler extends RouterProvider implements SingleComp parallel_tool_calls: metadata?.parallelToolCalls ?? true, } - const completion = await this.client.chat.completions.create(body) - - for await (const chunk of completion) { - // Vercel AI Gateway reports mid-stream failures as an in-band error chunk - // rather than throwing, so surface it instead of returning an empty response. - if ("error" in chunk && chunk.error) { - const raw = chunk.error as { message?: unknown } - const message = - typeof raw.message === "string" && raw.message.length > 0 - ? raw.message - : "Vercel AI Gateway stream error" - throw new Error(message) + // Per-request controller so an external abort signal (e.g. task + // cancellation) can interrupt the in-flight streaming request. + // Bridge it to our controller using the Bedrock pattern: + // - pre-aborted guard: check if already aborted before adding listener + // - { once: true }: remove listener after first abort to avoid leaks + // The listener is stored so it can be detached when the request ends: + // { once: true } only removes it on abort, so a task-scoped signal + // would otherwise accumulate one listener per request. + const controller = new AbortController() + const externalAbortSignal = metadata?.abortSignal + const abortListener = () => controller.abort() + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + controller.abort() + } else { + externalAbortSignal.addEventListener("abort", abortListener, { once: true }) } + } - const delta = chunk.choices[0]?.delta - if (delta?.content) { - yield { - type: "text", - text: delta.content, + try { + const completion = await this.client.chat.completions.create(body, { signal: controller.signal }) + + for await (const chunk of completion) { + // Vercel AI Gateway reports mid-stream failures as an in-band error chunk + // rather than throwing, so surface it instead of returning an empty response. + if ("error" in chunk && chunk.error) { + const raw = chunk.error as { message?: unknown } + const message = + typeof raw.message === "string" && raw.message.length > 0 + ? raw.message + : "Vercel AI Gateway stream error" + throw new Error(message) } - } - // Emit raw tool call chunks - NativeToolCallParser handles state management - if (delta?.tool_calls) { - for (const toolCall of delta.tool_calls) { + const delta = chunk.choices[0]?.delta + if (delta?.content) { yield { - type: "tool_call_partial", - index: toolCall.index, - id: toolCall.id, - name: toolCall.function?.name, - arguments: toolCall.function?.arguments, + type: "text", + text: delta.content, } } - } - if (chunk.usage) { - const usage = chunk.usage as VercelAiGatewayUsage - yield { - type: "usage", - inputTokens: usage.prompt_tokens || 0, - outputTokens: usage.completion_tokens || 0, - cacheWriteTokens: usage.cache_creation_input_tokens || undefined, - cacheReadTokens: usage.prompt_tokens_details?.cached_tokens || undefined, - totalCost: usage.cost ?? 0, + // Emit raw tool call chunks - NativeToolCallParser handles state management + if (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) { + const usage = chunk.usage as VercelAiGatewayUsage + yield { + type: "usage", + inputTokens: usage.prompt_tokens || 0, + outputTokens: usage.completion_tokens || 0, + cacheWriteTokens: usage.cache_creation_input_tokens || undefined, + cacheReadTokens: usage.prompt_tokens_details?.cached_tokens || undefined, + totalCost: usage.cost ?? 0, + } + } + } + } catch (error) { + // Preserve abort identity (series standard): surface a cancelled + // request as a DOM-standard AbortError rather than leaking the + // raw SDK abort error. + if (controller.signal.aborted) { + throw createAbortError("Vercel AI Gateway") } + throw error + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) } } @@ -133,10 +165,38 @@ export class VercelAiGatewayHandler extends RouterProvider implements SingleComp } requestOptions.max_completion_tokens = info.maxTokens + // Build request options with abortSignal and/or timeout. + // timeoutMs <= 0 means "no explicit timeout": omit the SDK timeout + // option entirely — the OpenAI SDK treats timeout: 0 as an immediate + // abort, which would cancel the request right away. + const createOptions: OpenAI.RequestOptions = {} + if (options?.abortSignal) { + createOptions.signal = options.abortSignal + } + if (options?.timeoutMs !== undefined && options.timeoutMs > 0) { + createOptions.timeout = options.timeoutMs + } - const response = await this.client.chat.completions.create(requestOptions) + const response = await this.client.chat.completions.create( + requestOptions, + Object.keys(createOptions).length > 0 ? createOptions : undefined, + ) return response.choices[0]?.message.content || "" } catch (error) { + // Preserve abort identity (series standard): caller-initiated + // cancellations and request timeouts must surface as a + // DOM-standard AbortError, not a wrapped completion error. The + // OpenAI SDK reports both with messages ending in a period + // ("Request was aborted.", "Request timed out."), which would not + // match task-level abort detection (message ending in "aborted"). + if ( + options?.abortSignal?.aborted || + error instanceof APIUserAbortError || + error instanceof APIConnectionTimeoutError || + (error instanceof Error && error.name === "AbortError") + ) { + throw createAbortError("Vercel AI Gateway") + } if (error instanceof Error) { throw new Error(`Vercel AI Gateway completion error: ${error.message}`) } diff --git a/src/api/providers/zoo-gateway.ts b/src/api/providers/zoo-gateway.ts index 4ff059df61..3f96b69de8 100644 --- a/src/api/providers/zoo-gateway.ts +++ b/src/api/providers/zoo-gateway.ts @@ -1,6 +1,6 @@ import * as vscode from "vscode" import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI from "openai" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { zooGatewayDefaultModelId, @@ -22,6 +22,7 @@ import { addCacheBreakpoints } from "../transform/caching/vercel-ai-gateway" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { NOT_PROVIDED } from "./constants" import { RouterProvider } from "./router-provider" +import { createAbortError } from "./utils/abort-signal" function getApiErrorStatus(error: unknown): number | undefined { if (typeof error === "object" && error !== null && "status" in error) { @@ -219,9 +220,29 @@ export class ZooGatewayHandler extends RouterProvider implements SingleCompletio parallel_tool_calls: metadata?.parallelToolCalls ?? true, } + // Per-request controller so an external abort signal (e.g. task + // cancellation) can interrupt the in-flight streaming request. + // Bridge it to our controller using the Bedrock pattern: + // - pre-aborted guard: check if already aborted before adding listener + // - { once: true }: remove listener after first abort to avoid leaks + // The listener is stored so it can be detached when the request ends: + // { once: true } only removes it on abort, so a task-scoped signal + // would otherwise accumulate one listener per request. + const controller = new AbortController() + const externalAbortSignal = metadata?.abortSignal + const abortListener = () => controller.abort() + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + controller.abort() + } else { + externalAbortSignal.addEventListener("abort", abortListener, { once: true }) + } + } + try { const completion = await this.client.chat.completions.create(body, { headers: requestHeaders, + signal: controller.signal, }) for await (const chunk of completion) { @@ -266,6 +287,12 @@ export class ZooGatewayHandler extends RouterProvider implements SingleCompletio } } } catch (error) { + // Preserve abort identity (series standard): surface a cancelled + // request as a DOM-standard AbortError before the gateway error + // surfacing/telemetry path. + if (controller.signal.aborted) { + throw createAbortError("Zoo Gateway") + } try { await surfaceGatewayApiError(error) } catch (surfaceError) { @@ -275,6 +302,8 @@ export class ZooGatewayHandler extends RouterProvider implements SingleCompletio ) } throw error + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) } } @@ -295,10 +324,35 @@ export class ZooGatewayHandler extends RouterProvider implements SingleCompletio } requestOptions.max_completion_tokens = info.maxTokens + // Build request options with abortSignal and/or timeout. + // timeoutMs <= 0 means "no explicit timeout": omit the SDK timeout + // option entirely — the OpenAI SDK treats timeout: 0 as an immediate + // abort, which would cancel the request right away. + const createOptions: OpenAI.RequestOptions = {} + if (options?.abortSignal) { + createOptions.signal = options.abortSignal + } + if (options?.timeoutMs !== undefined && options.timeoutMs > 0) { + createOptions.timeout = options.timeoutMs + } - const response = await this.client.chat.completions.create(requestOptions) + const response = await this.client.chat.completions.create(requestOptions, createOptions) return response.choices[0]?.message.content || "" } catch (error) { + // Preserve abort identity (series standard): caller-initiated + // cancellations and request timeouts must surface as a + // DOM-standard AbortError, not a wrapped completion error. The + // OpenAI SDK reports both with messages ending in a period + // ("Request was aborted.", "Request timed out."), which would not + // match task-level abort detection (message ending in "aborted"). + if ( + options?.abortSignal?.aborted || + error instanceof APIUserAbortError || + error instanceof APIConnectionTimeoutError || + (error instanceof Error && error.name === "AbortError") + ) { + throw createAbortError("Zoo Gateway") + } try { await surfaceGatewayApiError(error) } catch (surfaceError) {