From a3fb7622a6ee692988fce20cf96ff6d30bea8bb4 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 16 Sep 2026 10:32:12 +0800 Subject: [PATCH 1/3] feat(api): add abort-signal cancellation-scope helpers to the shared utils Extend src/api/providers/utils/abort-signal.ts with the abort-signal series helpers used by the gateway providers: - isRequestAborted(error, signal): wider abort detection - an aborted signal, a DOM AbortError, the OpenAI/Anthropic SDK APIUserAbortError (name check), or the exact SDK abort message "Request was aborted." - trusting name/message only on real Error instances so a plain object that merely looks like an abort propagates unchanged - createAbortError(providerName): fresh error satisfying the Task.ts abort contract (name "AbortError", message ending in "aborted") - rejectOnAbort(pending, signal, providerName): settle a signal-less async phase (model discovery) on the provider AbortError when the signal fires first; the abort listener detaches when pending settles - resolveModelWithAbort(fetchModel, signal, providerName): run model resolution inside a cancellation scope - entry fast-fail for a pre-aborted signal, the rejectOnAbort race while the lookup is pending, and normalization of abort-flavored lookup failures; any other resolution failure propagates unchanged Includes direct unit tests for the resolveModelWithAbort cancellation scope (pre-aborted fast-fail, no-signal pass-through, mid-resolution race, abort normalization, non-abort propagation), the isRequestAborted instanceof tightening tests, and the settle-guard test utility. Unit 1/3 of the #1295 split (content source: 62f596c5d). Part of the abort-signal series (round 1). Builds on #674, #901, #1008. Addresses #404. --- .../utils/__tests__/abort-signal.spec.ts | 228 +++++++++++++++--- src/api/providers/utils/abort-signal.ts | 107 +++++--- src/test-utils/settle-guard.ts | 26 ++ 3 files changed, 304 insertions(+), 57 deletions(-) create mode 100644 src/test-utils/settle-guard.ts diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index 1692f71e63..8a9a63bb94 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -3,8 +3,196 @@ import { isRequestAborted, mergeAbortSignalAndTimeout, mergeAbortSignals, - throwIfAborted, + rejectOnAbort, + resolveModelWithAbort, } from "../abort-signal" +import { withSettleGuard } from "../../../../test-utils/settle-guard" + +describe("rejectOnAbort", () => { + it("resolves with the pending value when it settles before the signal aborts", async () => { + const controller = new AbortController() + + await expect( + withSettleGuard(rejectOnAbort(Promise.resolve("done"), controller.signal, "TestProvider")), + ).resolves.toBe("done") + expect(controller.signal.aborted).toBe(false) + }) + + it("rejects with the provider abort error when the signal aborts first", async () => { + const controller = new AbortController() + // Never settles: the race must end purely via the abort. + const pending = new Promise(() => {}) + const race = rejectOnAbort(pending, controller.signal, "TestProvider") + controller.abort() + + await expect(withSettleGuard(race)).rejects.toMatchObject({ + name: "AbortError", + message: "The TestProvider request was aborted", + }) + }) + + it("rejects immediately when the signal is already aborted", async () => { + const controller = new AbortController() + controller.abort() + const pending = new Promise(() => {}) + + await expect(withSettleGuard(rejectOnAbort(pending, controller.signal, "TestProvider"))).rejects.toMatchObject({ + name: "AbortError", + message: "The TestProvider request was aborted", + }) + }) + + it("propagates the pending rejection when the signal stays active", async () => { + const controller = new AbortController() + const boom = new Error("lookup failed") + + await expect( + withSettleGuard(rejectOnAbort(Promise.reject(boom), controller.signal, "TestProvider")), + ).rejects.toBe(boom) + }) + + it("detaches the abort listener once the pending settles", async () => { + const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, "addEventListener") + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + + await expect( + withSettleGuard(rejectOnAbort(Promise.resolve("done"), controller.signal, "TestProvider")), + ).resolves.toBe("done") + + // The settle path must remove the exact listener that was registered, not just any + // function: removing a different reference would leave the original abort listener + // attached to the signal. First require the registration to have happened at all, + // so a missing registration cannot silently degrade to an undefined comparison. + expect(addSpy).toHaveBeenCalledTimes(1) + const registeredListener = addSpy.mock.calls[0]?.[1] as EventListener | undefined + expect(typeof registeredListener).toBe("function") + expect(removeSpy).toHaveBeenCalledWith("abort", registeredListener) + addSpy.mockRestore() + removeSpy.mockRestore() + }) + + it("detaches the abort listener when the pending rejects", async () => { + const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, "addEventListener") + const removeSpy = vi.spyOn(controller.signal, "removeEventListener") + const lookupError = new Error("lookup failed") + + await expect( + withSettleGuard(rejectOnAbort(Promise.reject(lookupError), controller.signal, "TestProvider")), + ).rejects.toBe(lookupError) + + // The settle path must remove the exact listener that was registered, not just any + // function: removing a different reference would leave the original abort listener + // attached to the signal. First require the registration to have happened at all, + // so a missing registration cannot silently degrade to an undefined comparison. + expect(addSpy).toHaveBeenCalledTimes(1) + const registeredListener = addSpy.mock.calls[0]?.[1] as EventListener | undefined + expect(typeof registeredListener).toBe("function") + expect(removeSpy).toHaveBeenCalledWith("abort", registeredListener) + addSpy.mockRestore() + removeSpy.mockRestore() + }) +}) + +describe("resolveModelWithAbort", () => { + it("fast-fails with the provider abort error when the signal is already aborted", async () => { + const controller = new AbortController() + controller.abort() + let lookupRan = false + + await expect( + withSettleGuard( + resolveModelWithAbort( + async () => { + lookupRan = true + return "model" + }, + controller.signal, + "TestProvider", + ), + ), + ).rejects.toMatchObject({ + name: "AbortError", + message: "The TestProvider request was aborted", + }) + // The entry guard must run before the lookup starts at all. + expect(lookupRan).toBe(false) + }) + + it("resolves the model when no abort signal is present", async () => { + await expect(resolveModelWithAbort(async () => "model", undefined, "TestProvider")).resolves.toBe("model") + }) + + it("propagates a raw resolution failure unchanged when no abort signal is present", async () => { + const boom = new Error("lookup failed") + + await expect( + resolveModelWithAbort( + async () => { + throw boom + }, + undefined, + "TestProvider", + ), + ).rejects.toBe(boom) + }) + + it("rejects with the provider abort error when the signal aborts while resolution is pending", async () => { + const controller = new AbortController() + let release: (value: string) => void = () => {} + const pending = new Promise((resolve) => { + release = resolve + }) + const race = withSettleGuard(resolveModelWithAbort(async () => pending, controller.signal, "TestProvider")) + + setTimeout(() => controller.abort(), 10) + + await expect(race).rejects.toMatchObject({ + name: "AbortError", + message: "The TestProvider request was aborted", + }) + // A late resolution after the abort won the race must not surface. + release("late model") + }) + + it("normalizes an abort-flavored resolution failure to the provider abort error", async () => { + const controller = new AbortController() + const abortFlavored = new Error("The operation was aborted.") + abortFlavored.name = "AbortError" + + const error = await resolveModelWithAbort( + async () => { + throw abortFlavored + }, + controller.signal, + "TestProvider", + ).catch((caught: unknown) => caught) + + expect(error).not.toBe(abortFlavored) + expect(error).toMatchObject({ + name: "AbortError", + message: "The TestProvider request was aborted", + }) + }) + + it("propagates non-abort resolution failures unchanged when a signal is present", async () => { + const controller = new AbortController() + const boom = new Error("lookup failed") + + await expect( + withSettleGuard( + resolveModelWithAbort( + async () => { + throw boom + }, + controller.signal, + "TestProvider", + ), + ), + ).rejects.toBe(boom) + }) +}) describe("abort-signal utilities", () => { describe("mergeAbortSignalAndTimeout", () => { @@ -106,34 +294,6 @@ describe("abort-signal utilities", () => { }) }) - 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() @@ -167,6 +327,16 @@ describe("abort-signal utilities", () => { const controller = new AbortController() expect(isRequestAborted(new Error("boom"), controller.signal)).toBe(false) }) + + it("requires an Error instance for the name and message checks", () => { + // A plain object that merely looks like an abort must not be + // classified as one: the instanceof guard keeps such failures + // propagating unchanged so callers can inspect the real shape. + const fakeAbort = { name: "AbortError", message: "Request was aborted." } + expect(isRequestAborted(fakeAbort)).toBe(false) + expect(isRequestAborted(Object.assign(Object.create(null), { name: "APIUserAbortError" }))).toBe(false) + expect(isRequestAborted("Request was aborted.")).toBe(false) + }) }) describe("createAbortError", () => { diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index 26f57c3e9a..e95226d2a3 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -36,23 +36,6 @@ 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`, @@ -66,19 +49,20 @@ export type OpenAiRequestOptions = { /** * 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. + * an `Error` carries a native abort error name (`AbortError`, + * `APIUserAbortError`), or an `Error` carries the OpenAI SDK abort error + * message (exactly "Request was aborted."). + * + * The name and message checks require an `Error` instance on purpose: a + * plain object that merely looks like an abort must propagate unchanged so + * callers can inspect the real failure shape. 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." - ) + const hasAbortName = error instanceof Error && (error.name === "AbortError" || error.name === "APIUserAbortError") + const hasSdkAbortMessage = error instanceof Error && error.message === "Request was aborted." + return Boolean(signal?.aborted) || hasAbortName || hasSdkAbortMessage } /** @@ -93,3 +77,70 @@ export function createAbortError(providerName: string): Error { abortError.name = "AbortError" return abortError } + +/** + * Await `pending` but reject with the provider's abort error when `signal` + * aborts first. For async phases that have no native signal support (model + * discovery) yet must still settle promptly on cancellation. The underlying + * promise keeps running (its settlement is ignored) — cancellation is + * cooperative at this boundary. + * + * The abort listener is detached once `pending` settles (success or + * failure), so repeated calls on one signal do not accumulate listeners. + */ +export function rejectOnAbort(pending: Promise, signal: AbortSignal, providerName: string): Promise { + if (signal.aborted) { + return Promise.reject(createAbortError(providerName)) + } + + return new Promise((resolve, reject) => { + const onAbort = () => reject(createAbortError(providerName)) + // Stryker disable next-line ObjectLiteral,BooleanLiteral: a signal fires its abort event exactly once and the settle handler removes this listener, so the once flag is unobservable + signal.addEventListener("abort", onAbort, { once: true }) + void pending.then( + (value) => { + signal.removeEventListener("abort", onAbort) + resolve(value) + }, + (error) => { + signal.removeEventListener("abort", onAbort) + reject(error) + }, + ) + }) +} + +/** + * Resolve a provider's model metadata inside a cancellation scope. + * + * A pre-aborted signal rejects immediately with the standardized AbortError + * before any lookup starts; a signal that fires while the lookup is pending + * settles via {@link rejectOnAbort} instead of waiting for the catalog to + * resolve. Abort failures from the lookup itself are normalized to the + * provider AbortError; any other resolution failure propagates unchanged. + * + * Providers pass their own resolution step, so the entry guard, the race, and + * the normalization logic exist once and are exercised through every + * provider's spec. + */ +export async function resolveModelWithAbort( + fetchModel: () => Promise, + abortSignal: AbortSignal | undefined, + providerName: string, +): Promise { + if (abortSignal?.aborted) { + throw createAbortError(providerName) + } + + try { + if (abortSignal) { + return await rejectOnAbort(fetchModel(), abortSignal, providerName) + } + return await fetchModel() + } catch (error) { + if (isRequestAborted(error, abortSignal)) { + throw createAbortError(providerName) + } + throw error + } +} diff --git a/src/test-utils/settle-guard.ts b/src/test-utils/settle-guard.ts new file mode 100644 index 0000000000..1efe5286c8 --- /dev/null +++ b/src/test-utils/settle-guard.ts @@ -0,0 +1,26 @@ +/** + * Stryker guard: fails fast if `promise` does not settle within `ms`. + * + * Stryker's per-mutant cutoff (timeoutMS 5s x timeoutFactor 1.5 ~= 7.5s) is shorter + * than vitest's testTimeout (20s). A mutant that removes a settle call (or an abort + * listener) leaves an awaited promise pending forever; without this guard the test + * would outlive the cutoff and the mutant would be reported as Timeout (inconclusive). + * Settling the guard at 500ms turns those mutants into fast failures (KILLED). + */ +export function withSettleGuard(promise: Promise, ms = 500): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new Error(`settle guard timed out after ${ms}ms`)) + }, ms) + void promise.then( + (value) => { + clearTimeout(timer) + resolve(value) + }, + (error) => { + clearTimeout(timer) + reject(error) + }, + ) + }) +} From 8d8ee07d1fc6000d1b9c3ad5633196397f23a46f Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 16 Sep 2026 10:54:18 +0800 Subject: [PATCH 2/3] feat(api): abort signal support for opencode-go - createMessage: bridge metadata.abortSignal to a per-request AbortController (Bedrock pattern: pre-aborted guard, once-listener, detached on completion so a task-scoped signal does not accumulate listeners); model resolution runs inside the shared resolveModelWithAbort cancellation scope (pre-aborted fast-fail, mid-resolution race) - aborted/timeout requests normalize to the provider AbortError on all three wire formats (anthropic /v1/messages, responses /v1/responses, openai chat completions), both pre-stream and mid-stream; non-abort failures keep the wrapped "Opencode Go completion error:" identity - completePrompt: forwards abortSignal/timeoutMs to all three SDK paths (timeoutMs <= 0 omits the SDK timeout option, since the SDK treats timeout: 0 as an immediate abort); aborted completions and APIConnectionTimeoutError/APITimeoutError normalize to the provider AbortError (series standard) The two inner pre-stream guard mutants (the abort-normalization condition and its provider-name literal) are documented as provably equivalent with mutator-specific Stryker directives: createMessage's outer catch applies the identical isRequestAborted check to the same controller signal and re-standardizes, so the inner layer's only unique behavior is the non-abort completion-error wrap (stays kill-tested). Unit 2/3 of the #1295 split (content source: 62f596c5d); stacks on the shared-util unit. Part of the abort-signal series (round 1). Builds on #674, #901, #1008. Addresses #404. --- .../providers/__tests__/opencode-go.spec.ts | 984 +++++++++++++++++- src/api/providers/opencode-go.ts | 299 ++++-- 2 files changed, 1197 insertions(+), 86 deletions(-) diff --git a/src/api/providers/__tests__/opencode-go.spec.ts b/src/api/providers/__tests__/opencode-go.spec.ts index 2f859a8c43..c6dd9d98ad 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, @@ -25,6 +29,7 @@ import { ApiHandlerOptions } from "../../../shared/api" import { Package } from "../../../shared/package" 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", () => ({ @@ -65,15 +70,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 = { @@ -208,6 +220,7 @@ describe("OpencodeGoHandler", () => { max_completion_tokens: 40_960, temperature: expect.any(Number), }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -233,6 +246,7 @@ describe("OpencodeGoHandler", () => { model: "glm-5.1", reasoning_effort: "medium", }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -383,10 +397,350 @@ 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 () => { + // A pre-aborted request must fail fast before any model-catalog or + // SDK work starts: the standardized AbortError wins over any + // resolution failure, and the request itself never begins. + 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, + ) + // The fast-fail guard throws before the request starts. + expect(mockCreate).not.toHaveBeenCalled() + // Pre-flight cancellation must skip the model catalog entirely: + // the getModels mock must remain uncalled, not just the SDK create. + expect(vitest.mocked(getModels)).not.toHaveBeenCalled() + expect(capturedSignal).toBeUndefined() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("rejects with the standardized AbortError when the external signal aborts while model resolution is pending", async () => { + // The model catalog stays pending while the external signal aborts: + // the resolution race must settle with the standardized AbortError + // before the lookup is released, and the request itself must never + // start. A bridge-only fix would let the lookup finish and abort the + // internal controller after the fact; this gate proves the prompt + // settles on the abort itself. + let releaseResolution!: () => void + const resolutionGate = new Promise((resolve) => { + releaseResolution = resolve + }) + vitest.mocked(getModels).mockImplementationOnce(async () => { + await resolutionGate + return { "glm-5.1": { ...opencodeGoModels["glm-5.1"] } } + }) + + 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 lookup park on the gate, then abort while it is still pending. + await new Promise((resolve) => setTimeout(resolve, 10)) + controller.abort() + + const error = await consumed.then( + () => undefined, + (e: unknown) => e, + ) + // The catalog lookup was attempted but the prompt must have settled + // on the abort itself, before the lookup was released: no SDK call. + expect(vitest.mocked(getModels)).toHaveBeenCalledTimes(1) + expect(mockCreate).not.toHaveBeenCalled() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + + // Releasing the lookup afterwards must not start a late request. + releaseResolution() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(mockCreate).not.toHaveBeenCalled() + }) + + it("propagates a raw model-lookup failure unchanged when no abort signal is present", async () => { + // Without an abort signal the resolution must not be wrapped or + // normalized: a catalog failure keeps its own identity so callers + // can distinguish a lookup failure from a cancellation. + const lookupError = new Error("catalog offline") + vitest.mocked(getModels).mockRejectedValueOnce(lookupError) + + const handler = new OpencodeGoHandler(mockOptions) + + const error = await collectStream(handler.createMessage("sys", [{ role: "user", content: "hi" }])).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toBe(lookupError) + expect(mockCreate).not.toHaveBeenCalled() + }) + + it("normalizes an abort-flavored model-lookup failure to the provider AbortError", async () => { + // A lookup that fails with an AbortError-shaped error (e.g. the + // catalog fetch itself was aborted) must settle with the provider's + // standardized AbortError, not the raw error shape. + const abortFlavored = new Error("The operation was aborted.") + abortFlavored.name = "AbortError" + vitest.mocked(getModels).mockRejectedValueOnce(abortFlavored) + + const handler = new OpencodeGoHandler(mockOptions) + + const error = await collectStream(handler.createMessage("sys", [{ role: "user", content: "hi" }])).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).not.toBe(abortFlavored) + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + expect(mockCreate).not.toHaveBeenCalled() + }) + + 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" }) + // Assert the exact listener reference so a bridge that removes a + // different callback than the one it registered cannot pass. + // The rejectOnAbort race registers its own "abort" listener on the + // same external signal during model resolution, so the first "abort" + // registration is the race's, not the bridge's. Target the last + // registration so the options/removal assertions below cannot be + // satisfied by the race's listener. + const abortAddCalls = addEventListenerSpy.mock.calls.filter(([event]) => event === "abort") + const addedListener = abortAddCalls[abortAddCalls.length - 1]?.[1] + expect(typeof addedListener).toBe("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", addedListener, { once: true }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", addedListener) + 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 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" }) + // The rejectOnAbort race registers its own "abort" listener on the + // same external signal during model resolution, so the first "abort" + // registration is the race's, not the bridge's. Target the last + // registration so the options/removal assertions below cannot be + // satisfied by the race's listener. + const abortAddCalls = addEventListenerSpy.mock.calls.filter(([event]) => event === "abort") + const addedListener = abortAddCalls[abortAddCalls.length - 1]?.[1] + expect(typeof addedListener).toBe("function") + expect(addEventListenerSpy).toHaveBeenCalledWith("abort", addedListener, { once: true }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", addedListener) + 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" } }] }) @@ -400,6 +754,7 @@ describe("OpencodeGoHandler", () => { max_completion_tokens: 40_960, reasoning_effort: "medium", }), + {}, ) }) @@ -425,7 +780,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 }), {}) }) }) @@ -487,6 +842,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() @@ -543,6 +899,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[] = [ @@ -579,6 +952,7 @@ describe("OpencodeGoHandler", () => { // so the model default is used. max_tokens: 65_536, }), + undefined, ) expect(mockCreate).not.toHaveBeenCalled() }) @@ -594,7 +968,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 () => { @@ -609,6 +983,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" }] @@ -754,7 +1442,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 () => { @@ -764,7 +1455,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 () => { @@ -831,6 +1525,63 @@ 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", + }) + }) + + it("normalizes a mid-stream abort on the Anthropic path to the standardized AbortError", async () => { + // A cancellation that surfaces after the /v1/messages stream has + // started must normalize to the standardized AbortError like the + // other wire formats, not leak the raw SDK rejection. + let capturedSignal: AbortSignal | undefined + mockAnthropicCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + return (async function* () { + yield { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "partial" }, + } + for (let i = 0; i < 40 && !capturedSignal?.aborted; i++) { + await new Promise((resolve) => setTimeout(resolve, 5)) + } + if (capturedSignal?.aborted) { + throw new AnthropicAbortError() + } + })() + }) + const controller = new AbortController() + const handler = new OpencodeGoHandler(anthropicOptions) + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "Hi" }] + + const consumed = collectStream( + handler.createMessage("sys", messages, { taskId: "test-task", 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") + }) }) describe("Responses-format models (gpt-5.6-luna)", () => { @@ -861,7 +1612,11 @@ describe("OpencodeGoHandler", () => { ) }) - it("forwards the abort signal to the streaming Responses request", async () => { + it("forwards the per-request abort signal to the streaming Responses request", async () => { + // The /v1/responses request is wired to the per-request internal + // controller (the bridge target), not the caller's signal: the + // caller's signal is bridged into the internal controller, and the + // SDK watches only the internal one. const handler = new OpencodeGoHandler(lunaOptions) const controller = new AbortController() const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "Hi" }] @@ -870,10 +1625,118 @@ describe("OpencodeGoHandler", () => { handler.createMessage("sys", messages, { taskId: "test-task", abortSignal: controller.signal }), ) - expect(mockResponsesCreate.mock.calls[0][1]).toEqual({ - signal: controller.signal, - headers: { "x-opencode-session": "test-task" }, + const callOptions = mockResponsesCreate.mock.calls[0][1] as { signal?: AbortSignal; headers?: unknown } + expect(callOptions.headers).toEqual({ "x-opencode-session": "test-task" }) + expect(callOptions.signal).toBeInstanceOf(AbortSignal) + expect(callOptions.signal).not.toBe(controller.signal) + expect(callOptions.signal?.aborted).toBe(false) + }) + + it("normalizes an aborted Responses request to the standardized AbortError", async () => { + // Emulate the OpenAI SDK against the per-request internal signal: + // the caller's signal aborts just as the request starts, the bridge + // flips the internal controller, and the SDK rejects with + // APIUserAbortError. The Responses-path guard must normalize it to + // the DOM-standard AbortError instead of wrapping it. + const controller = new AbortController() + mockResponsesCreate.mockImplementation(async (_body: unknown, options: { signal?: AbortSignal }) => { + controller.abort() + if (options?.signal?.aborted) { + throw new APIUserAbortError() + } + throw new Error("unreachable: the bridge must have aborted the internal signal") }) + const handler = new OpencodeGoHandler(lunaOptions) + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "Hi" }] + + const error = await collectStream( + handler.createMessage("sys", messages, { taskId: "test-task", abortSignal: controller.signal }), + ).then( + () => undefined, + (e: unknown) => e, + ) + // A cancelled /v1/responses request surfaces as a DOM-standard + // AbortError, not the wrapped "completion error" reserved for other + // failures. + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Opencode Go request was aborted") + }) + + it("normalizes a mid-stream abort on the Responses path to the standardized AbortError", async () => { + // A cancellation that surfaces after the /v1/responses stream has + // started must normalize to the standardized AbortError like the + // other wire formats, not leak the raw SDK rejection. + let capturedSignal: AbortSignal | undefined + mockResponsesCreate.mockImplementation(async (_body: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + return (async function* () { + yield { type: "response.output_text.delta", delta: "partial" } + for (let i = 0; i < 40 && !capturedSignal?.aborted; i++) { + await new Promise((resolve) => setTimeout(resolve, 5)) + } + if (capturedSignal?.aborted) { + throw new APIUserAbortError() + } + })() + }) + const controller = new AbortController() + const handler = new OpencodeGoHandler(lunaOptions) + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "Hi" }] + + const consumed = collectStream( + handler.createMessage("sys", messages, { taskId: "test-task", 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("wraps non-abort Responses pre-stream failures with the Opencode Go prefix", async () => { + // A non-abort rejection from responses.create (e.g. an upstream 500) + // must be wrapped like the other wire formats, not normalized. + mockResponsesCreate.mockRejectedValue(new Error("boom")) + const handler = new OpencodeGoHandler(lunaOptions) + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "Hi" }] + + await expect(collectStream(handler.createMessage("sys", messages))).rejects.toThrow( + "Opencode Go completion error: boom", + ) + }) + + it("detaches the bridged abort listener from the Responses-format path", async () => { + // The Responses branch must detach the bridged listener on the way + // out like the other formats: a task-scoped signal must not + // accumulate one listener per request. + const handler = new OpencodeGoHandler(lunaOptions) + const controller = new AbortController() + const removeListenerSpy = vi.spyOn(controller.signal, "removeEventListener") + const addEventListenerSpy = vi.spyOn(controller.signal, "addEventListener") + const messages: Anthropic.Messages.MessageParam[] = [{ role: "user", content: "Hi" }] + + const chunks = await collectStream( + handler.createMessage("sys", messages, { taskId: "test-task", abortSignal: controller.signal }), + ) + expect(chunks).toContainEqual({ type: "text", text: "Hello" }) + // The rejectOnAbort race registers its own "abort" listener on the + // same external signal during model resolution, so the first "abort" + // registration is the race's, not the bridge's. Target the last + // registration so the options/removal assertions below cannot be + // satisfied by the race's listener. + const abortAddCalls = addEventListenerSpy.mock.calls.filter(([event]) => event === "abort") + const addedListener = abortAddCalls[abortAddCalls.length - 1]?.[1] + expect(typeof addedListener).toBe("function") + expect(addEventListenerSpy).toHaveBeenCalledWith("abort", addedListener, { once: true }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", addedListener) + expect(controller.signal.aborted).toBe(false) }) it("closes the Responses iterator when the consumer stops early", async () => { @@ -959,7 +1822,13 @@ describe("OpencodeGoHandler", () => { await vitest.waitFor(() => expect(mockResponsesCreate).toHaveBeenCalled()) controller.abort() - await expect(nextPromise).rejects.toThrow("request aborted") + // The read rejection lands on an aborted request signal, so the + // Responses branch normalizes it to the standardized AbortError + // (series standard) while still closing the in-flight iterator. + await expect(nextPromise).rejects.toMatchObject({ + name: "AbortError", + message: "The Opencode Go request was aborted", + }) expect(iterator.return).toHaveBeenCalled() }) @@ -1283,7 +2152,10 @@ describe("OpencodeGoHandler", () => { await expect(handler.completePrompt("ping")).rejects.toBe("completion failure") }) - it("rejects non-streaming Responses completion when the abort signal fires", async () => { + it("normalizes an aborted non-streaming Responses completion to the standardized AbortError", async () => { + // A caller-initiated cancellation of a /v1/responses completion must + // surface as a DOM-standard AbortError (series standard), not the raw + // rejection or a wrapped completion error. const controller = new AbortController() const request = new Promise((_resolve, reject) => { controller.signal.addEventListener("abort", () => reject(new Error("request aborted")), { once: true }) @@ -1295,7 +2167,10 @@ describe("OpencodeGoHandler", () => { await vitest.waitFor(() => expect(mockResponsesCreate).toHaveBeenCalled()) controller.abort() - await expect(completion).rejects.toThrow("request aborted") + await expect(completion).rejects.toMatchObject({ + name: "AbortError", + message: "The Opencode Go request was aborted", + }) }) it("completePrompt calls responses.create and returns output_text", async () => { @@ -1347,6 +2222,69 @@ describe("OpencodeGoHandler", () => { expect(mockCreate).not.toHaveBeenCalled() }) + it("forwards a positive timeoutMs to the non-streaming Responses request", async () => { + mockResponsesCreate.mockResolvedValue({ output_text: "Hello!" }) + const handler = new OpencodeGoHandler(lunaOptions) + + await handler.completePrompt("ping", { timeoutMs: 5_000 }) + + expect(mockResponsesCreate.mock.calls[0][1]).toEqual({ timeout: 5_000 }) + expect(mockCreate).not.toHaveBeenCalled() + }) + + it("omits the timeout option from the Responses request 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). + mockResponsesCreate.mockResolvedValue({ output_text: "Hello!" }) + const handler = new OpencodeGoHandler(lunaOptions) + + await handler.completePrompt("ping", { timeoutMs: 0 }) + + const callOptions = mockResponsesCreate.mock.calls[0][1] as Record + expect(callOptions).not.toHaveProperty("timeout") + expect(callOptions).not.toHaveProperty("signal") + expect(mockCreate).not.toHaveBeenCalled() + }) + + it("preserves abort identity on the Responses completion path", async () => { + // Emulate the OpenAI SDK: an aborted request signal rejects with + // APIUserAbortError; the Responses completion path must normalize it + // to the DOM-standard AbortError, not a wrapped completion error. + const controller = new AbortController() + controller.abort() + mockResponsesCreate.mockImplementation(async (_body: unknown, options: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + throw new APIUserAbortError() + } + throw new Error("unreachable") + }) + const handler = new OpencodeGoHandler(lunaOptions) + + 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("surfaces Responses request timeouts as an AbortError in completePrompt", async () => { + // Emulate the OpenAI SDK: when the request timeout fires, the SDK + // surfaces APIConnectionTimeoutError ("Request timed out.") — the + // series standard maps it to the same AbortError identity as caller + // cancellations. + mockResponsesCreate.mockRejectedValue(new APIConnectionTimeoutError()) + const handler = new OpencodeGoHandler(lunaOptions) + + 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("completePrompt wraps errors with an Opencode Go-specific message", async () => { mockResponsesCreate.mockRejectedValue(new Error("boom")) const handler = new OpencodeGoHandler(lunaOptions) diff --git a/src/api/providers/opencode-go.ts b/src/api/providers/opencode-go.ts index 46ea52a2be..0600ce5195 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, isRequestAborted, resolveModelWithAbort } from "./utils/abort-signal" /** * The wire formats exposed by the Opencode Go gateway: @@ -198,24 +203,85 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { - const { id: modelId, info, format, temperature, reasoningEffort, maxTokens } = await this.resolveModel() + // Establish the cancellation scope around model resolution: a + // pre-aborted signal rejects before the lookup starts, and a signal + // that fires while model metadata is loading settles on the + // standardized AbortError; any other resolution failure propagates + // unchanged. + const externalAbortSignal = metadata?.abortSignal + const resolved = await resolveModelWithAbort(() => this.resolveModel(), externalAbortSignal, "Opencode Go") + const { id: modelId, info, format, temperature, reasoningEffort, maxTokens } = resolved + + // 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 abortListener = () => controller.abort() + if (externalAbortSignal) { + // Stryker disable next-line ConditionalExpression: externalAbortSignal.aborted can never be true here - the entry guard rejects a pre-aborted signal and the rejectOnAbort race rejects an abort during model resolution, and no await sits between the race settling and this bridge, so the branch is unreachable + if (externalAbortSignal.aborted) { + // Stryker disable next-line CallExpression: unreachable branch body - a pre-aborted external signal is rejected by the entry guard (and a mid-resolution abort by the race) before this bridge registers + 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, + ) + } catch (error) { + // Preserve abort identity (series standard): a cancellation that + // surfaces mid-stream must normalize to the provider AbortError, + // matching the OpenAI streaming branch. + if (isRequestAborted(error, controller.signal)) { + throw createAbortError("Opencode Go") + } + throw error + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) + } return } if (format === "responses") { - yield* this.streamResponsesMessage( - modelId, - info, - temperature, - maxTokens, - reasoningEffort, - systemPrompt, - messages, - metadata, - ) + try { + yield* this.streamResponsesMessage( + modelId, + info, + temperature, + maxTokens, + reasoningEffort, + systemPrompt, + messages, + controller.signal, + metadata, + ) + } catch (error) { + // Preserve abort identity (series standard): a cancellation that + // surfaces mid-stream must normalize to the provider AbortError, + // matching the OpenAI streaming branch. + if (isRequestAborted(error, controller.signal)) { + throw createAbortError("Opencode Go") + } + throw error + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) + } return } @@ -247,46 +313,59 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio }), } - const completion = metadata?.taskId - ? await this.client.chat.completions.create(body, { - headers: { "x-opencode-session": metadata.taskId }, - }) - : await this.client.chat.completions.create(body) + try { + const completion = metadata?.taskId + ? await this.client.chat.completions.create(body, { + signal: controller.signal, + headers: { "x-opencode-session": metadata.taskId }, + }) + : 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 (isRequestAborted(error, controller.signal)) { + throw createAbortError("Opencode Go") + } + throw error + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) } } @@ -315,6 +394,7 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio reasoningEffort: ReasoningEffortExtended | undefined, systemPrompt: string, messages: Anthropic.Messages.MessageParam[], + abortSignal: AbortSignal, metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { const input = convertToResponsesApiInput(messages) @@ -380,13 +460,27 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio : {}), } + // Wrap pre-stream errors with the same "Opencode Go completion error:" + // prefix used by completePrompt so the Responses-format path surfaces + // failures consistently. Abort identity is preserved first (series + // standard): a cancelled request must surface as a DOM-standard + // AbortError, not a wrapped completion error. Mid-stream errors + // propagate unchanged, matching the other streaming paths. let stream: AsyncIterable try { stream = await this.client.responses.create(requestBody, { - signal: metadata?.abortSignal, + signal: abortSignal, headers: metadata?.taskId ? { "x-opencode-session": metadata.taskId } : undefined, }) } catch (error) { + // isRequestAborted covers the signal-aborted case plus the OpenAI + // SDK's APIUserAbortError (name "APIUserAbortError" / "Request was + // aborted."). + // Stryker disable next-line ConditionalExpression: abort normalization is subsumed by createMessage's outer catch, which applies the identical isRequestAborted check to the same controller signal and re-throws the standardized AbortError - this condition's mutation is unobservable (the inner layer's unique behavior is the non-abort wrap below) + if (isRequestAborted(error, abortSignal)) { + // Stryker disable next-line StringLiteral: the thrown error is re-caught by createMessage's outer catch, which re-standardizes with the correct provider name (the standardized error's name AbortError matches the outer isRequestAborted) - this literal is unobservable + throw createAbortError("Opencode Go") + } if (error instanceof Error) { throw new Error(`Opencode Go completion error: ${error.message}`) } @@ -462,6 +556,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" } @@ -514,10 +609,19 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio try { stream = metadata?.taskId ? await this.anthropicClient.messages.create(requestParams, { + signal: abortSignal, headers: { "x-opencode-session": metadata.taskId }, }) - : await this.anthropicClient.messages.create(requestParams) + : 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. isRequestAborted also covers the Anthropic + // SDK's APIUserAbortError (name "APIUserAbortError"). + if (isRequestAborted(error, abortSignal)) { + // Stryker disable next-line StringLiteral: the thrown error is re-caught by createMessage's outer catch, which re-standardizes with the correct provider name (the standardized error's name AbortError matches the outer isRequestAborted) - this literal is unobservable + throw createAbortError("Opencode Go") + } if (error instanceof Error) { throw new Error(`Opencode Go completion error: ${error.message}`) } @@ -702,24 +806,50 @@ 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 = { + ...(options?.abortSignal && { signal: options.abortSignal }), + ...(options?.timeoutMs !== undefined && options.timeoutMs > 0 && { 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}`) } @@ -729,6 +859,15 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio if (format === "responses") { try { + // Build request options with abortSignal and/or timeout (series + // standard): timeoutMs <= 0 means "no explicit timeout" — the + // OpenAI SDK treats timeout: 0 as an immediate abort, so the SDK + // timeout option is omitted unless the value is positive. + const createOptions: OpenAI.RequestOptions = { + ...(options?.abortSignal && { signal: options.abortSignal }), + ...(options?.timeoutMs !== undefined && options.timeoutMs > 0 && { timeout: options.timeoutMs }), + } + const response = await this.client.responses.create( { model: modelId, @@ -756,10 +895,21 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio } : {}), }, - { signal: options?.abortSignal }, + createOptions, ) return response.output_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 + // 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"). + // SDK request timeouts are not aborts, but the series standard + // maps them to the same AbortError identity as caller cancellations. + if (isRequestAborted(error, options?.abortSignal) || error instanceof APIConnectionTimeoutError) { + throw createAbortError("Opencode Go") + } if (error instanceof Error) { throw new Error(`Opencode Go completion error: ${error.message}`) } @@ -786,9 +936,32 @@ 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 = { + ...(options?.abortSignal && { signal: options.abortSignal }), + ...(options?.timeoutMs !== undefined && options.timeoutMs > 0 && { 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}`) } From 044e22bcb56467acc46807f299820118485d94bd Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 16 Sep 2026 10:36:52 +0800 Subject: [PATCH 3/3] feat(api): abort signal support for unbound, vercel-ai-gateway, zoo-gateway Each provider's createMessage bridges metadata.abortSignal to a per-request AbortController (Bedrock pattern: pre-aborted guard, { once: true } listener, detached on completion so a task-scoped signal does not accumulate listeners) and runs model resolution inside the shared resolveModelWithAbort cancellation scope; aborted requests (pre-aborted, mid-resolution, mid-stream) normalize to the provider AbortError while non-abort failures propagate or wrap unchanged. completePrompt forwards abortSignal/timeoutMs to the OpenAI SDK (timeoutMs <= 0 omits the SDK timeout option); aborted completions and APIConnectionTimeoutError normalize to the provider AbortError (series standard). Zoo Gateway places the entry fast-fail before ensureAuthenticated() so a pre-aborted task surfaces the AbortError instead of an auth failure (pinned by an unauthenticated pre-abort test). Unit 3/3 of the #1295 split (content source: 62f596c5d); stacks on the shared-util and opencode-go units. Three sibling OpenAI-SDK gateway providers with an identical abort-wiring shape are kept as one unit to avoid tripling the review surface for mechanically identical changes (a+d soft-cap rationale per the split budget; the mutation gate is measured on the unit delta and is green). Part of the abort-signal series (round 1). Builds on #674, #901, #1008. Addresses #404. --- src/api/providers/__tests__/unbound.spec.ts | 743 +++++++++++++++++- .../__tests__/vercel-ai-gateway.spec.ts | 421 +++++++++- .../providers/__tests__/zoo-gateway.spec.ts | 338 +++++++- src/api/providers/unbound.ts | 138 +++- src/api/providers/vercel-ai-gateway.ts | 143 +++- src/api/providers/zoo-gateway.ts | 71 +- 6 files changed, 1771 insertions(+), 83 deletions(-) diff --git a/src/api/providers/__tests__/unbound.spec.ts b/src/api/providers/__tests__/unbound.spec.ts index 99f11d7b6f..7adee93b3b 100644 --- a/src/api/providers/__tests__/unbound.spec.ts +++ b/src/api/providers/__tests__/unbound.spec.ts @@ -1,18 +1,28 @@ import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI from "openai" +import OpenAI, { APIConnectionTimeoutError, APIUserAbortError } from "openai" import { UnboundHandler } from "../unbound" +import { getModels } from "../fetchers/modelCache" 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 +189,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 +371,569 @@ 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 with the standardized AbortError before any request work when the external signal is already aborted", async () => { + // A pre-aborted request must fail fast before any model-catalog or + // SDK work starts: the standardized AbortError wins over any + // resolution failure, and the request itself never begins. + let capturedSignal: AbortSignal | undefined + sharedMockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + throw new DOMException("The operation was aborted.", "AbortError") + }) + + 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 }), + ) + + const error = await collectStream(stream).then( + () => undefined, + (e: unknown) => e, + ) + // The fast-fail guard rejects before the request starts. + expect(sharedMockCreate).not.toHaveBeenCalled() + // Pre-flight cancellation must skip the model catalog entirely: + // the getModels mock must remain uncalled, not just the SDK create. + expect(getModels).not.toHaveBeenCalled() + expect(capturedSignal).toBeUndefined() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Unbound request was aborted") + }) + + it("rejects with the standardized AbortError when the external signal aborts while model resolution is pending", async () => { + // The model catalog stays pending while the external signal aborts: + // the resolution race must settle with the standardized AbortError + // before the lookup is released, and the request itself must never + // start. A bridge-only fix would let the lookup finish and abort the + // internal controller after the fact; this gate proves the prompt + // settles on the abort itself. + let releaseResolution!: () => void + const resolutionGate = new Promise((resolve) => { + releaseResolution = resolve + }) + vi.mocked(getModels).mockImplementationOnce(async () => { + await resolutionGate + return { + "openai/gpt-4o": { + maxTokens: 4096, + contextWindow: 128000, + supportsImages: true, + supportsPromptCache: false, + inputPrice: 2.5, + outputPrice: 10, + description: "GPT-4o", + }, + } + }) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + const controller = new AbortController() + + const consumed = collectStream( + handler.createMessage( + "system", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ), + ) + + // Let the lookup park on the gate, then abort while it is still pending. + await new Promise((resolve) => setTimeout(resolve, 10)) + controller.abort() + + const error = await consumed.then( + () => undefined, + (e: unknown) => e, + ) + // The catalog lookup was attempted but the prompt must have settled + // on the abort itself, before the lookup was released: no SDK call. + expect(getModels).toHaveBeenCalledTimes(1) + expect(sharedMockCreate).not.toHaveBeenCalled() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Unbound request was aborted") + + // Releasing the lookup afterwards must not start a late request. + releaseResolution() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(sharedMockCreate).not.toHaveBeenCalled() + }) + + 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() + + // The stream loop normalizes failures that surface after the stream + // has started (series standard): the cancellation must surface as + // the standardized AbortError, not the raw rejection. + 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 Unbound request was aborted") + }) + + it("normalizes a mid-stream APIUserAbortError to the standardized AbortError", async () => { + // The SDK can raise its own user-abort error even when the per-request + // controller was not aborted (e.g. the SDK watched a different signal): + // the instanceof branch of the loop catch must normalize it. + sharedMockCreate.mockImplementation(async () => + (async function* () { + yield { choices: [{ delta: { content: "partial" } }] } + throw new APIUserAbortError() + })(), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + await expect( + collectStream( + handler.createMessage( + "system", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: new AbortController().signal }), + ), + ), + ).rejects.toMatchObject({ + name: "AbortError", + message: "The Unbound request was aborted", + }) + }) + + it("normalizes a name-based AbortError thrown from the stream to the standardized AbortError", async () => { + // The DOM-style name check must catch errors that are not SDK abort + // classes: a plain Error named AbortError is normalized too. + const abortError = new Error("The operation was aborted.") + abortError.name = "AbortError" + sharedMockCreate.mockImplementation(async () => + (async function* () { + yield { choices: [{ delta: { content: "partial" } }] } + throw abortError + })(), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + await expect( + collectStream( + handler.createMessage( + "system", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: new AbortController().signal }), + ), + ), + ).rejects.toMatchObject({ + name: "AbortError", + message: "The Unbound request was aborted", + }) + }) + + it("rethrows a non-Error mid-stream failure that merely looks like an abort", async () => { + // The instanceof guard is load-bearing: a plain object carrying an + // AbortError name is not an Error, so it must not be normalized and + // must propagate unchanged. + const fake = { name: "AbortError", message: "suspicious" } + sharedMockCreate.mockImplementation(async () => + (async function* () { + yield { choices: [{ delta: { content: "partial" } }] } + throw fake + })(), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await collectStream( + handler.createMessage( + "system", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: new AbortController().signal }), + ), + ).then( + () => undefined, + (e: unknown) => e, + ) + expect(error).toBe(fake) + }) + + it("rethrows a plain non-abort Error instance from the stream unchanged", async () => { + // The name check must not over-normalize: a genuine Error whose + // name is not "AbortError" must propagate unchanged — only real + // aborts surface as the standardized AbortError. + const boom = new Error("stream exploded") + sharedMockCreate.mockImplementation(async () => + (async function* () { + yield { choices: [{ delta: { content: "partial" } }] } + throw boom + })(), + ) + + const handler = new UnboundHandler({ + unboundApiKey: "test-key", + unboundModelId: "openai/gpt-4o", + }) + + const error = await collectStream( + handler.createMessage( + "system", + [{ role: "user", content: "hi" }], + makeCreateMessageMetadata({ abortSignal: new AbortController().signal }), + ), + ).then( + () => undefined, + (e: unknown) => e, + ) + // Identity, not shape: the rethrown error must be the very Error + // instance the stream raised, not a normalized abort error. + expect(error).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" }) + // Assert the exact listener reference so a bridge that removes a + // different callback than the one it registered cannot pass. + // The rejectOnAbort race registers its own "abort" listener on the + // same external signal during model resolution, so the first "abort" + // registration is the race's, not the bridge's. Target the last + // registration so the options/removal assertions below cannot be + // satisfied by the race's listener. + const abortAddCalls = addEventListenerSpy.mock.calls.filter(([event]) => event === "abort") + const addedListener = abortAddCalls[abortAddCalls.length - 1]?.[1] + expect(typeof addedListener).toBe("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", addedListener, { once: true }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", addedListener) + 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 268e42a7e2..c586bc236e 100644 --- a/src/api/providers/__tests__/vercel-ai-gateway.spec.ts +++ b/src/api/providers/__tests__/vercel-ai-gateway.spec.ts @@ -10,10 +10,11 @@ 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 { getModels } from "../fetchers/modelCache" +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" @@ -295,6 +296,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( @@ -313,6 +363,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ temperature: customTemp, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -328,6 +379,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ temperature: VERCEL_AI_GATEWAY_DEFAULT_TEMPERATURE, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -347,6 +399,7 @@ describe("VercelAiGatewayHandler", () => { temperature: undefined, max_completion_tokens: 128000, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -365,6 +418,7 @@ describe("VercelAiGatewayHandler", () => { model: "anthropic/claude-fable-5.1", temperature: undefined, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -461,6 +515,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ max_completion_tokens: 64000, // max tokens for sonnet 4 }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -536,6 +591,7 @@ describe("VercelAiGatewayHandler", () => { }), ]), }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -553,6 +609,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ tool_choice: "auto", }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -570,6 +627,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ parallel_tool_calls: true, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -587,6 +645,7 @@ describe("VercelAiGatewayHandler", () => { tools: expect.any(Array), parallel_tool_calls: true, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) @@ -683,6 +742,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ stream_options: { include_usage: true }, }), + expect.objectContaining({ signal: expect.any(AbortSignal) }), ) }) }) @@ -721,6 +781,7 @@ describe("VercelAiGatewayHandler", () => { temperature: VERCEL_AI_GATEWAY_DEFAULT_TEMPERATURE, max_completion_tokens: 64000, }), + undefined, ) }) @@ -739,6 +800,7 @@ describe("VercelAiGatewayHandler", () => { expect.objectContaining({ temperature: customTemp, }), + undefined, ) }) @@ -771,10 +833,364 @@ 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 the standardized AbortError before any request work when the external signal is already aborted", async () => { + // A pre-aborted request must fail fast before any model-catalog or + // SDK work starts: the standardized AbortError wins over any + // resolution failure, and the request itself never begins. + 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 }), + ) + + const error = await collectStream(stream).then( + () => undefined, + (e: unknown) => e, + ) + // The fast-fail guard rejects before the request starts. + expect(mockCreate).not.toHaveBeenCalled() + // Pre-flight cancellation must skip the model catalog entirely: + // the getModels mock must remain uncalled, not just the SDK create. + expect(getModels).not.toHaveBeenCalled() + expect(capturedSignal).toBeUndefined() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Vercel AI Gateway request was aborted") + }) + + it("rejects with the standardized AbortError when the external signal aborts while model resolution is pending", async () => { + // The model catalog stays pending while the external signal aborts: + // the resolution race must settle with the standardized AbortError + // before the lookup is released, and the request itself must never + // start. A bridge-only fix would let the lookup finish and abort the + // internal controller after the fact; this gate proves the prompt + // settles on the abort itself. + let releaseResolution!: () => void + const resolutionGate = new Promise((resolve) => { + releaseResolution = resolve + }) + vitest.mocked(getModels).mockImplementationOnce(async () => { + await resolutionGate + return { + "anthropic/claude-sonnet-4": { + maxTokens: 64000, + contextWindow: 200000, + supportsImages: true, + supportsPromptCache: true, + inputPrice: 3, + outputPrice: 15, + description: "Claude Sonnet 4", + }, + } + }) + + 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 lookup park on the gate, then abort while it is still pending. + await new Promise((resolve) => setTimeout(resolve, 10)) + controller.abort() + + const error = await consumed.then( + () => undefined, + (e: unknown) => e, + ) + // The catalog lookup was attempted but the prompt must have settled + // on the abort itself, before the lookup was released: no SDK call. + expect(getModels).toHaveBeenCalledTimes(1) + expect(mockCreate).not.toHaveBeenCalled() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Vercel AI Gateway request was aborted") + + // Releasing the lookup afterwards must not start a late request. + releaseResolution() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(mockCreate).not.toHaveBeenCalled() + }) + + 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" }) + // Assert the exact listener reference so a bridge that removes a + // different callback than the one it registered cannot pass. + // The rejectOnAbort race registers its own "abort" listener on the + // same external signal during model resolution, so the first "abort" + // registration is the race's, not the bridge's. Target the last + // registration so the options/removal assertions below cannot be + // satisfied by the race's listener. + const abortAddCalls = addEventListenerSpy.mock.calls.filter(([event]) => event === "abort") + const addedListener = abortAddCalls[abortAddCalls.length - 1]?.[1] + expect(typeof addedListener).toBe("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", addedListener, { once: true }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", addedListener) + 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, @@ -789,6 +1205,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..b8886f9e64 100644 --- a/src/api/providers/__tests__/zoo-gateway.spec.ts +++ b/src/api/providers/__tests__/zoo-gateway.spec.ts @@ -26,16 +26,18 @@ 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" import { ZooGatewayHandler, classifyGatewayApiError, toGatewayStreamError } from "../zoo-gateway" +import { getModels } from "../fetchers/modelCache" import { ApiHandlerOptions } from "../../../shared/api" 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 +377,7 @@ describe("ZooGatewayHandler", () => { "X-Zoo-Task-ID": "task-123", "X-Zoo-Mode": "code", }, + signal: expect.any(AbortSignal), }), ) }) @@ -520,6 +523,7 @@ describe("ZooGatewayHandler", () => { temperature: ZOO_GATEWAY_DEFAULT_TEMPERATURE, max_completion_tokens: 64000, }), + {}, ) }) @@ -542,8 +546,340 @@ 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 the standardized AbortError before any request work when the external signal is already aborted", async () => { + // A pre-aborted request must fail fast before any model-catalog or + // SDK work starts: the standardized AbortError wins over any + // resolution failure, and the request itself never begins. + 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 }), + ) + + const error = await collectStream(stream).then( + () => undefined, + (e: unknown) => e, + ) + // The fast-fail guard rejects before the request starts. + expect(mockCreate).not.toHaveBeenCalled() + // Pre-flight cancellation must skip the model catalog entirely: + // the getModels mock must remain uncalled, not just the SDK create. + expect(getModels).not.toHaveBeenCalled() + expect(capturedSignal).toBeUndefined() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + }) + + it("rejects with the standardized AbortError before the auth check when the external signal is already aborted and the session is unauthenticated", async () => { + // The entry fast-fail must win over the auth failure: with no + // session token a pre-aborted request surfaces the standardized + // AbortError, not the auth error the unauthenticated fall-through + // would raise. + let capturedSignal: AbortSignal | undefined + mockCreate.mockImplementation(async (_params: unknown, options: { signal?: AbortSignal }) => { + capturedSignal = options?.signal + throw new Error("boom") + }) + + const handler = new ZooGatewayHandler({ ...mockOptions, zooSessionToken: undefined }) + const controller = new AbortController() + controller.abort() + + const stream = handler.createMessage( + "prompt", + [{ role: "user", content: "hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const error = await collectStream(stream).then( + () => undefined, + (e: unknown) => e, + ) + // The fast-fail guard rejects before the auth check and the request. + expect(mockCreate).not.toHaveBeenCalled() + expect(getModels).not.toHaveBeenCalled() + expect(capturedSignal).toBeUndefined() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + }) + + it("rejects with the standardized AbortError when the external signal aborts while model resolution is pending", async () => { + // The model catalog stays pending while the external signal aborts: + // the resolution race must settle with the standardized AbortError + // before the lookup is released, and the request itself must never + // start. A bridge-only fix would let the lookup finish and abort the + // internal controller after the fact; this gate proves the prompt + // settles on the abort itself. + let releaseResolution!: () => void + const resolutionGate = new Promise((resolve) => { + releaseResolution = resolve + }) + vitest.mocked(getModels).mockImplementationOnce(async () => { + await resolutionGate + return DEFAULT_MODEL_CATALOG + }) + + 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 lookup park on the gate, then abort while it is still pending. + await new Promise((resolve) => setTimeout(resolve, 10)) + controller.abort() + + const error = await consumed.then( + () => undefined, + (e: unknown) => e, + ) + // The catalog lookup was attempted but the prompt must have settled + // on the abort itself, before the lookup was released: no SDK call. + expect(getModels).toHaveBeenCalledTimes(1) + expect(mockCreate).not.toHaveBeenCalled() + expect(error).toMatchObject({ name: "AbortError" }) + expect((error as Error).message).toBe("The Zoo Gateway request was aborted") + + // Releasing the lookup afterwards must not start a late request. + releaseResolution() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(mockCreate).not.toHaveBeenCalled() + }) + + 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" }) + // Assert the exact listener reference so a bridge that removes a + // different callback than the one it registered cannot pass. + // The rejectOnAbort race registers its own "abort" listener on the + // same external signal during model resolution, so the first "abort" + // registration is the race's, not the bridge's. Target the last + // registration so the options/removal assertions below cannot be + // satisfied by the race's listener. + const abortAddCalls = addEventListenerSpy.mock.calls.filter(([event]) => event === "abort") + const addedListener = abortAddCalls[abortAddCalls.length - 1]?.[1] + expect(typeof addedListener).toBe("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", addedListener, { once: true }) + expect(removeListenerSpy).toHaveBeenCalledWith("abort", addedListener) + 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/unbound.ts b/src/api/providers/unbound.ts index 61a2d1ae38..deb949d7c8 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 } 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, isRequestAborted, resolveModelWithAbort } from "./utils/abort-signal" import { extractReasoningFromDelta } from "./utils/extract-reasoning" // Unbound usage includes extra fields for Anthropic cache tokens. @@ -125,6 +126,13 @@ export class UnboundHandler extends BaseProvider implements SingleCompletionHand messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { + // Establish the cancellation scope around model resolution: a + // pre-aborted signal rejects before the lookup starts, and a signal + // that fires while model metadata is loading settles on the + // standardized AbortError; any other resolution failure propagates + // unchanged. + const externalAbortSignal = metadata?.abortSignal + const resolved = await resolveModelWithAbort(() => this.fetchModel(), externalAbortSignal, "Unbound") const { id: model, info, @@ -132,7 +140,7 @@ export class UnboundHandler extends BaseProvider implements SingleCompletionHand temperature, reasoningEffort: reasoning_effort, reasoning: thinking, - } = await this.fetchModel() + } = resolved const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ { role: "system", content: systemPrompt }, @@ -158,46 +166,86 @@ 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 abortListener = () => controller.abort() + if (externalAbortSignal) { + // Stryker disable next-line ConditionalExpression: externalAbortSignal.aborted can never be true here - the entry guard rejects a pre-aborted signal and the rejectOnAbort race rejects an abort during model resolution, and no await sits between the race settling and this bridge, so the branch is unreachable + if (externalAbortSignal.aborted) { + // Stryker disable next-line CallExpression: unreachable branch body - a pre-aborted external signal is rejected by the entry guard (and a mid-resolution abort by the race) before this bridge registers + 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 (isRequestAborted(error, externalAbortSignal)) { + throw createAbortError("Unbound") + } + throw handleOpenAIError(error, this.providerName) } + let lastUsage: any = undefined - if (delta?.content) { - yield { type: "text", text: delta.content } - } + try { + for await (const chunk of stream) { + const delta = chunk.choices[0]?.delta + + const reasoningText = extractReasoningFromDelta(delta) + if (reasoningText) { + yield { type: "reasoning", text: reasoningText } + } + + if (delta?.content) { + yield { type: "text", text: delta.content } + } + + // Handle native tool calls + if (delta && "tool_calls" in delta && Array.isArray(delta.tool_calls)) { + for (const toolCall of delta.tool_calls) { + yield { + type: "tool_call_partial", + index: toolCall.index, + id: toolCall.id, + name: toolCall.function?.name, + arguments: toolCall.function?.arguments, + } + } + } - // Handle native tool calls - if (delta && "tool_calls" in delta && Array.isArray(delta.tool_calls)) { - for (const toolCall of delta.tool_calls) { - yield { - type: "tool_call_partial", - index: toolCall.index, - id: toolCall.id, - name: toolCall.function?.name, - arguments: toolCall.function?.arguments, + if (chunk.usage) { + lastUsage = chunk.usage } } - } - if (chunk.usage) { - lastUsage = chunk.usage + if (lastUsage) { + yield this.processUsageMetrics(lastUsage, info) + } + } catch (error) { + // Preserve abort identity (series standard): a cancellation that + // surfaces after the stream has started must also normalize to + // the standardized AbortError, not the raw SDK rejection. + if (isRequestAborted(error, externalAbortSignal)) { + throw createAbortError("Unbound") + } + throw error } - } - - if (lastUsage) { - yield this.processUsageMetrics(lastUsage, info) + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) } } @@ -212,11 +260,33 @@ 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"). + // SDK request timeouts are not aborts, but the series standard maps + // them to the same AbortError identity as caller cancellations. + if (isRequestAborted(error, options?.abortSignal) || error instanceof APIConnectionTimeoutError) { + throw createAbortError("Unbound") + } throw handleOpenAIError(error, this.providerName) } return response.choices[0]?.message.content || "" diff --git a/src/api/providers/vercel-ai-gateway.ts b/src/api/providers/vercel-ai-gateway.ts index 3f4f3af26c..8bc9314d1a 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 } from "openai" import { vercelAiGatewayDefaultModelId, @@ -19,6 +19,7 @@ import { addCacheBreakpoints } from "../transform/caching/vercel-ai-gateway" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { RouterProvider } from "./router-provider" +import { createAbortError, isRequestAborted, resolveModelWithAbort } from "./utils/abort-signal" // Extend OpenAI's CompletionUsage to include Vercel AI Gateway specific fields interface VercelAiGatewayUsage extends OpenAI.CompletionUsage { @@ -58,7 +59,14 @@ export class VercelAiGatewayHandler extends RouterProvider implements SingleComp messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { - const { id: modelId, info } = await this.fetchModel() + // Establish the cancellation scope around model resolution: a + // pre-aborted signal rejects before the lookup starts, and a signal + // that fires while model metadata is loading settles on the + // standardized AbortError; any other resolution failure propagates + // unchanged. + const externalAbortSignal = metadata?.abortSignal + const resolved = await resolveModelWithAbort(() => this.fetchModel(), externalAbortSignal, "Vercel AI Gateway") + const { id: modelId, info } = resolved const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ { role: "system", content: systemPrompt }, @@ -89,52 +97,84 @@ export class VercelAiGatewayHandler extends RouterProvider implements SingleComp ;(body as { reasoning_effort?: ReasoningEffortExtended }).reasoning_effort = reasoningEffort } - 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 abortListener = () => controller.abort() + if (externalAbortSignal) { + // Stryker disable next-line ConditionalExpression: externalAbortSignal.aborted can never be true here - the entry guard rejects a pre-aborted signal and the rejectOnAbort race rejects an abort during model resolution, and no await sits between the race settling and this bridge, so the branch is unreachable + if (externalAbortSignal.aborted) { + // Stryker disable next-line CallExpression: unreachable branch body - a pre-aborted external signal is rejected by the entry guard (and a mid-resolution abort by the race) before this bridge registers + 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, + } + } + + // 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, + 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): a cancelled request + // must surface as a DOM-standard AbortError, not the raw SDK abort + // error or an in-band stream error raised while aborting. + if (isRequestAborted(error, externalAbortSignal)) { + throw createAbortError("Vercel AI Gateway") + } + throw error + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) } } @@ -157,10 +197,35 @@ 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"). + // SDK request timeouts are not aborts, but the series standard maps + // them to the same AbortError identity as caller cancellations. + if (isRequestAborted(error, options?.abortSignal) || error instanceof APIConnectionTimeoutError) { + 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..a786492e53 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 } 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, isRequestAborted, resolveModelWithAbort } from "./utils/abort-signal" function getApiErrorStatus(error: unknown): number | undefined { if (typeof error === "object" && error !== null && "status" in error) { @@ -181,9 +182,22 @@ export class ZooGatewayHandler extends RouterProvider implements SingleCompletio messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { + // Fail fast when the task is already cancelled before any model-catalog + // work starts: the standardized AbortError wins over any failure the + // fallible resolution could raise — including the auth check below. + const externalAbortSignal = metadata?.abortSignal + if (externalAbortSignal?.aborted) { + throw createAbortError("Zoo Gateway") + } + this.ensureAuthenticated() - const { id: modelId, info } = await this.fetchModel() + // Establish the cancellation scope around model resolution: a signal + // that fires while model metadata is loading settles on the + // standardized AbortError; any other resolution failure propagates + // unchanged. + const resolved = await resolveModelWithAbort(() => this.fetchModel(), externalAbortSignal, "Zoo Gateway") + const { id: modelId, info } = resolved const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [ { role: "system", content: systemPrompt }, @@ -219,9 +233,30 @@ 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 abortListener = () => controller.abort() + if (externalAbortSignal) { + // Stryker disable next-line ConditionalExpression: externalAbortSignal.aborted can never be true here - the entry guard rejects a pre-aborted signal and the rejectOnAbort race rejects an abort during model resolution, and no await sits between the race settling and this bridge, so the branch is unreachable + if (externalAbortSignal.aborted) { + // Stryker disable next-line CallExpression: unreachable branch body - a pre-aborted external signal is rejected by the entry guard (and a mid-resolution abort by the race) before this bridge registers + 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 +301,12 @@ export class ZooGatewayHandler extends RouterProvider implements SingleCompletio } } } catch (error) { + // Preserve abort identity (series standard): a cancelled request + // must surface as a DOM-standard AbortError before the gateway + // error surfacing/telemetry path, not the raw SDK abort error. + if (isRequestAborted(error, externalAbortSignal)) { + throw createAbortError("Zoo Gateway") + } try { await surfaceGatewayApiError(error) } catch (surfaceError) { @@ -275,6 +316,8 @@ export class ZooGatewayHandler extends RouterProvider implements SingleCompletio ) } throw error + } finally { + externalAbortSignal?.removeEventListener("abort", abortListener) } } @@ -295,10 +338,32 @@ 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"). + // SDK request timeouts are not aborts, but the series standard maps + // them to the same AbortError identity as caller cancellations. + if (isRequestAborted(error, options?.abortSignal) || error instanceof APIConnectionTimeoutError) { + throw createAbortError("Zoo Gateway") + } try { await surfaceGatewayApiError(error) } catch (surfaceError) {