diff --git a/packages/openai-adapters/src/apis/Gemini.ts b/packages/openai-adapters/src/apis/Gemini.ts index f7298267849..9e905f0963e 100644 --- a/packages/openai-adapters/src/apis/Gemini.ts +++ b/packages/openai-adapters/src/apis/Gemini.ts @@ -1,6 +1,7 @@ // IMPORTANT: Import nativeFetch FIRST to preserve native fetch before any pollution import { withNativeFetch } from "../util/nativeFetch.js"; -import { GoogleGenAI } from "@google/genai"; +import { withRequestOptionsFetch } from "../util/requestOptionsFetch.js"; +import { GoogleGenAI, type HttpOptions } from "@google/genai"; import { OpenAI } from "openai/index"; import { ChatCompletion, @@ -23,7 +24,6 @@ import { GeminiConfig } from "../types.js"; import { chatChunk, chatChunkFromDelta, - customFetch, embedding, usageChatChunk, } from "../util.js"; @@ -65,6 +65,126 @@ interface GeminiToolDelta }; } +/** + * Build the httpOptions handed to the GoogleGenAI SDK from Continue's config. + * + * The SDK constructs request URLs by joining `baseUrl` and `apiVersion` with + * a slash, skipping `apiVersion` when it is an empty string. Continue's + * `apiBase` already carries its version segment (e.g. `/v1beta/`), so + * `apiVersion` is blanked to keep the SDK from appending a second one. + * + * Returns undefined when nothing is configured so default-config construction + * is unchanged. + */ +function buildGoogleGenAIHttpOptions( + config: GeminiConfig, +): HttpOptions | undefined { + const httpOptions: HttpOptions = {}; + + if (config.apiBase) { + httpOptions.baseUrl = config.apiBase; + httpOptions.apiVersion = ""; + } + + if (config.requestOptions?.headers) { + httpOptions.headers = config.requestOptions.headers; + } + + if (config.requestOptions?.timeout !== undefined) { + httpOptions.timeout = config.requestOptions.timeout; + } + + return Object.keys(httpOptions).length > 0 ? httpOptions : undefined; +} + +type JsonObject = Record; + +function asJsonObject(value: unknown): JsonObject | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as JsonObject) + : undefined; +} + +/** + * Extract the human-readable message nested inside a Gemini API error blob. + * + * The SDK's ApiError.message is a JSON envelope ({ error: { message, code, + * status } }) whose error.message is often ITSELF a JSON string from Google + * (see upstream issue #12945) — so users see "Unknown error" while the real + * cause ("You exceeded your current quota...") sits two parse levels deep. + * Walks the nesting until the innermost message; returns undefined for + * non-JSON, malformed, or message-less input so callers can pass the + * original error through unchanged. + * + * Mirrored by extractNestedJsonMessage in + * extensions/cli/src/util/formatError.ts — kept as a small local mirror + * with shared test vectors rather than a new cross-package export. Keep the + * two in sync. + */ +export function extractNestedGeminiError( + raw: string, +): { message: string; code?: number } | undefined { + // Bound the unwrap depth so a gateway returning deeply nested error + // envelopes cannot force unbounded sequential parses. + const MAX_DEPTH = 8; + let node: unknown; + try { + node = JSON.parse(raw); + } catch { + return undefined; + } + + let message: string | undefined; + let code: number | undefined; + + for (let depth = 0; depth < MAX_DEPTH; depth++) { + const obj = asJsonObject(node); + if (!obj) { + break; + } + const target = asJsonObject(obj.error) ?? obj; + if (typeof target.code === "number") { + code = target.code; + } + if (typeof target.message !== "string") { + break; + } + message = target.message; + try { + node = JSON.parse(target.message.trim()); + } catch { + break; + } + } + + return message === undefined ? undefined : { message: message.trim(), code }; +} + +/** + * Rebuild an SDK error around its extracted nested message, preserving the + * HTTP status. Returns the original error untouched when there is nothing + * to extract — never masks or degrades an unrecognized error. + */ +function normalizeGeminiError(error: unknown): unknown { + if (!(error instanceof Error)) { + return error; + } + const extracted = extractNestedGeminiError(error.message); + if (extracted === undefined) { + return error; + } + const status = + "status" in error && typeof error.status === "number" + ? error.status + : extracted.code; + const normalized: Error & { status?: number; cause?: unknown } = new Error( + extracted.message, + ); + normalized.status = status; + normalized.cause = error; + return normalized; +} + export class GeminiApi implements BaseLlmApi { apiBase: string = "https://generativelanguage.googleapis.com/v1beta/"; private genAI: GoogleGenAI; @@ -79,6 +199,7 @@ export class GeminiApi implements BaseLlmApi { () => new GoogleGenAI({ apiKey: this.config.apiKey, + httpOptions: buildGoogleGenAIHttpOptions(this.config), }), ); } @@ -331,7 +452,11 @@ export class GeminiApi implements BaseLlmApi { if (chunk.usageMetadata) { usage = { prompt_tokens: chunk.usageMetadata.promptTokenCount || 0, - completion_tokens: chunk.usageMetadata.candidatesTokenCount || 0, + // OpenAI-compatible usage counts reasoning inside completion_tokens; + // thinking models report those separately as thoughtsTokenCount. + completion_tokens: + (chunk.usageMetadata.candidatesTokenCount || 0) + + (chunk.usageMetadata.thoughtsTokenCount || 0), total_tokens: chunk.usageMetadata.totalTokenCount || 0, }; } @@ -403,9 +528,11 @@ export class GeminiApi implements BaseLlmApi { model: string, convertedBody: ReturnType, ) { - // Use native fetch temporarily for stream operation to get proper ReadableStream - // The withNativeFetch wrapper restores native fetch, makes the call, then reverts - return withNativeFetch(() => + // Run the SDK call under a fetch matching this config: native fetch for + // default configs (proper ReadableStream, restored after the call), or a + // customFetch-backed fetch honoring proxy/TLS requestOptions with its + // node-fetch Response adapted back to a native one for SDK streaming. + return withRequestOptionsFetch(this.config.requestOptions, () => genAI.models.generateContentStream({ model, contents: convertedBody.contents, @@ -422,30 +549,42 @@ export class GeminiApi implements BaseLlmApi { body: ChatCompletionCreateParamsStreaming, _signal: AbortSignal, ): AsyncGenerator { - const convertedBody = this._convertBody( - body, - this.apiBase.includes("/v1/"), - true, - ); - const response = await this.generateStream( - this.genAI, - body.model, - convertedBody, - ); - yield* this.processStreamResponse(response, body.model); + try { + const convertedBody = this._convertBody( + body, + this.apiBase.includes("/v1/"), + true, + ); + const response = await this.generateStream( + this.genAI, + body.model, + convertedBody, + ); + yield* this.processStreamResponse(response, body.model); + } catch (error) { + // Surface the nested Gemini message (issue #12945 class of failures); + // unrecognized errors are rethrown untouched. + throw normalizeGeminiError(error); + } } async *streamWithGenAI( genAI: GoogleGenAI, body: ChatCompletionCreateParamsStreaming, ): AsyncGenerator { - const convertedBody = this._convertBody(body, false, true); - const response = await this.generateStream( - genAI, - body.model, - convertedBody, - ); - yield* this.processStreamResponse(response, body.model); + try { + const convertedBody = this._convertBody(body, false, true); + const response = await this.generateStream( + genAI, + body.model, + convertedBody, + ); + yield* this.processStreamResponse(response, body.model); + } catch (error) { + // Same normalization as chatCompletionStream — this entry point serves + // Vertex AI's Gemini models, which hit the identical error surface. + throw normalizeGeminiError(error); + } } completionNonStream( @@ -469,37 +608,42 @@ export class GeminiApi implements BaseLlmApi { async embed(body: EmbeddingCreateParams): Promise { const inputs = Array.isArray(body.input) ? body.input : [body.input]; - const response = await customFetch(this.config.requestOptions)( - new URL(`${body.model}:batchEmbedContents`, this.apiBase), - { - method: "POST", - body: JSON.stringify({ - requests: inputs.map((input) => ({ + try { + // The SDK owns the REST contract — URL construction, model-name + // normalization, and the response shape ({ embeddings: [{ values }] }). + // Same proxy/TLS wrapper and error normalization as the chat paths. + const response = await withRequestOptionsFetch( + this.config.requestOptions, + () => + this.genAI.models.embedContent({ model: body.model, - content: { - role: "user", - parts: [{ text: input }], - }, - })), - }), - headers: { - // eslint-disable-next-line @typescript-eslint/naming-convention - "x-goog-api-key": this.config.apiKey, - // eslint-disable-next-line @typescript-eslint/naming-convention - "Content-Type": "application/json", - }, - }, - ); + contents: inputs.map((input) => String(input)), + }), + ); + + const embeddings = response.embeddings; + if (!embeddings || embeddings.length === 0) { + throw new Error( + `Gemini returned no embeddings for model ${body.model}`, + ); + } - const data = (await response.json()) as any; - return embedding({ - model: body.model, - usage: { - total_tokens: data.total_tokens, - prompt_tokens: data.prompt_tokens, - }, - data: data.batchEmbedContents.map((embedding: any) => embedding.values), - }); + return embedding({ + model: body.model, + // Google's embeddings API reports no token counts — the shared + // helper applies its documented zero-usage default. + data: embeddings.map((entry, index) => { + if (!entry.values) { + throw new Error( + `Gemini returned no values for embedding at index ${index}`, + ); + } + return entry.values; + }), + }); + } catch (error) { + throw normalizeGeminiError(error); + } } list(): Promise { diff --git a/packages/openai-adapters/src/test/gemini-adapter.vitest.ts b/packages/openai-adapters/src/test/gemini-adapter.vitest.ts new file mode 100644 index 00000000000..9433982ac81 --- /dev/null +++ b/packages/openai-adapters/src/test/gemini-adapter.vitest.ts @@ -0,0 +1,554 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +import type { GeminiApi as GeminiApiType } from "../apis/Gemini.js"; + +const generateContentStream = vi.fn(); +const embedContent = vi.fn(); +const GoogleGenAIMock = vi.fn().mockImplementation(() => ({ + models: { + generateContentStream, + embedContent, + }, +})); + +vi.mock("@google/genai", () => ({ + GoogleGenAI: GoogleGenAIMock, +})); + +describe("GeminiApi GoogleGenAI construction", () => { + afterEach(() => { + vi.clearAllMocks(); + }); + + it("passes custom headers, timeout, and apiBase through httpOptions", async () => { + const { GeminiApi } = await import("../apis/Gemini.js"); + + new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + apiBase: + "https://gateway.example.com/v1/streaming-models/locations/europe-west4/publishers/google", + requestOptions: { + timeout: 10000, + headers: { + "x-api-key": "gateway-api-key", + "Content-Type": "application/json", + }, + }, + }); + + expect(GoogleGenAIMock).toHaveBeenCalledWith({ + apiKey: "primary-api-key", + httpOptions: { + baseUrl: + "https://gateway.example.com/v1/streaming-models/locations/europe-west4/publishers/google", + apiVersion: "", + timeout: 10000, + headers: { + "x-api-key": "gateway-api-key", + "Content-Type": "application/json", + }, + }, + }); + }); + + it("sets baseUrl with blank apiVersion for a custom apiBase alone", async () => { + const { GeminiApi } = await import("../apis/Gemini.js"); + + new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + apiBase: "https://gateway.example.com/v1beta/", + }); + + expect(GoogleGenAIMock).toHaveBeenCalledWith({ + apiKey: "primary-api-key", + httpOptions: { + baseUrl: "https://gateway.example.com/v1beta/", + apiVersion: "", + }, + }); + }); + + it("passes headers without apiBase when only requestOptions.headers is set", async () => { + const { GeminiApi } = await import("../apis/Gemini.js"); + + new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + requestOptions: { + headers: { "x-custom": "value" }, + }, + }); + + expect(GoogleGenAIMock).toHaveBeenCalledWith({ + apiKey: "primary-api-key", + httpOptions: { + headers: { "x-custom": "value" }, + }, + }); + }); + + it("omits httpOptions entirely for a default config — construction unchanged", async () => { + const { GeminiApi } = await import("../apis/Gemini.js"); + + new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + }); + + expect(GoogleGenAIMock).toHaveBeenCalledWith({ + apiKey: "primary-api-key", + httpOptions: undefined, + }); + }); +}); + +describe("GeminiApi error normalization", () => { + afterEach(() => { + vi.clearAllMocks(); + }); + + /** + * Real shape from upstream issue #12945 (and reproduced live in this + * session): the SDK's ApiError.message is a JSON envelope whose + * error.message is ITSELF a pretty-printed JSON string from Google. + */ + function quotaApiErrorMessage(): string { + const googleBody = JSON.stringify( + { + error: { + code: 429, + message: + "You exceeded your current quota, please check your plan and billing details. Please retry in 45.191226092s.", + status: "RESOURCE_EXHAUSTED", + }, + }, + null, + 2, + ); + return JSON.stringify({ + error: { + message: `${googleBody}\n`, + code: 429, + status: "Too Many Requests", + }, + }); + } + + class FakeApiError extends Error { + status: number; + constructor(message: string, status: number) { + super(message); + this.status = status; + } + } + + async function drainStreamError(api: GeminiApiType): Promise { + try { + for await (const _chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + // drain + } + return undefined; + } catch (error) { + return error; + } + } + + it("rethrows SDK errors with nested message and status extracted", async () => { + generateContentStream.mockRejectedValue( + new FakeApiError(quotaApiErrorMessage(), 429), + ); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + const thrown = (await drainStreamError(api)) as Error & { + status?: number; + }; + + expect(thrown).toBeInstanceOf(Error); + expect(thrown.message).toContain("You exceeded your current quota"); + expect(thrown.message.startsWith('{"error"')).toBe(false); + expect(thrown.status).toBe(429); + }); + + it("normalizes errors on the non-stream path too (delegation)", async () => { + generateContentStream.mockRejectedValue( + new FakeApiError(quotaApiErrorMessage(), 429), + ); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + let thrown: (Error & { status?: number }) | undefined; + try { + await api.chatCompletionNonStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + }, + new AbortController().signal, + ); + } catch (error) { + thrown = error as Error & { status?: number }; + } + + expect(thrown).toBeInstanceOf(Error); + expect(thrown!.message).toContain("You exceeded your current quota"); + // Raw JSON also contains the quota text — pin that the envelope is GONE + expect(thrown!.message.startsWith('{"error"')).toBe(false); + expect(thrown!.status).toBe(429); + }); + + it("passes through a plain non-JSON error unchanged", async () => { + const original = new Error("socket hang up"); + generateContentStream.mockRejectedValue(original); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + const thrown = await drainStreamError(api); + expect(thrown).toBe(original); + }); + + it("passes through malformed JSON messages unchanged", async () => { + const original = new Error("{invalid json"); + generateContentStream.mockRejectedValue(original); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + const thrown = await drainStreamError(api); + expect(thrown).toBe(original); + }); + + it("passes through JSON errors that carry no nested message unchanged", async () => { + const original = new Error( + JSON.stringify({ error: { status: "RESOURCE_EXHAUSTED" } }), + ); + generateContentStream.mockRejectedValue(original); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + const thrown = await drainStreamError(api); + expect(thrown).toBe(original); + }); + + it("passes through JSON errors whose error field is a primitive unchanged", async () => { + const original = new Error(JSON.stringify({ error: "Invalid API key" })); + generateContentStream.mockRejectedValue(original); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + const thrown = await drainStreamError(api); + expect(thrown).toBe(original); + }); + + it("prefers the SDK error's own HTTP status over the nested code when they differ", async () => { + generateContentStream.mockRejectedValue( + new FakeApiError(quotaApiErrorMessage(), 500), + ); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + const thrown = (await drainStreamError(api)) as Error & { status?: number }; + expect(thrown.status).toBe(500); + }); + + it("falls back to the nested code when the SDK error carries no status", async () => { + generateContentStream.mockRejectedValue(new Error(quotaApiErrorMessage())); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + const thrown = (await drainStreamError(api)) as Error & { status?: number }; + expect(thrown.message).toContain("You exceeded your current quota"); + expect(thrown.status).toBe(429); + }); + + it("normalizes errors on streamWithGenAI (the Vertex AI entry point) too", async () => { + const failingStream = vi + .fn() + .mockRejectedValue(new FakeApiError(quotaApiErrorMessage(), 429)); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + let thrown: (Error & { status?: number }) | undefined; + try { + for await (const _chunk of api.streamWithGenAI( + { + models: { generateContentStream: failingStream }, + } as unknown as Parameters[0], + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + )) { + // drain + } + } catch (error) { + thrown = error as Error & { status?: number }; + } + + expect(thrown).toBeInstanceOf(Error); + expect(thrown!.message).toContain("You exceeded your current quota"); + expect(thrown!.message.startsWith('{"error"')).toBe(false); + expect(thrown!.status).toBe(429); + }); +}); + +describe("extractNestedGeminiError (direct vectors)", () => { + it("extracts message and code from the double-nested #12945 shape", async () => { + const { extractNestedGeminiError } = await import("../apis/Gemini.js"); + const googleBody = JSON.stringify( + { + error: { + code: 429, + message: "You exceeded your current quota.", + status: "RESOURCE_EXHAUSTED", + }, + }, + null, + 2, + ); + const raw = JSON.stringify({ + error: { + message: `${googleBody}\n`, + code: 429, + status: "Too Many Requests", + }, + }); + + expect(extractNestedGeminiError(raw)).toEqual({ + message: "You exceeded your current quota.", + code: 429, + }); + }); + + it("extracts a single-level nested message", async () => { + const { extractNestedGeminiError } = await import("../apis/Gemini.js"); + expect( + extractNestedGeminiError( + JSON.stringify({ error: { message: "Quota exceeded", code: 429 } }), + ), + ).toEqual({ message: "Quota exceeded", code: 429 }); + }); + + it("returns undefined for message-less JSON", async () => { + const { extractNestedGeminiError } = await import("../apis/Gemini.js"); + expect( + extractNestedGeminiError( + JSON.stringify({ error: { status: "RESOURCE_EXHAUSTED" } }), + ), + ).toBeUndefined(); + }); + + it("returns undefined for a primitive error field", async () => { + const { extractNestedGeminiError } = await import("../apis/Gemini.js"); + expect( + extractNestedGeminiError(JSON.stringify({ error: "Invalid API key" })), + ).toBeUndefined(); + }); + + it("returns undefined for malformed JSON", async () => { + const { extractNestedGeminiError } = await import("../apis/Gemini.js"); + expect(extractNestedGeminiError("{invalid json")).toBeUndefined(); + }); + + it("returns undefined for plain non-JSON text", async () => { + const { extractNestedGeminiError } = await import("../apis/Gemini.js"); + expect(extractNestedGeminiError("socket hang up")).toBeUndefined(); + }); +}); + +describe("GeminiApi usage accounting", () => { + afterEach(() => { + vi.clearAllMocks(); + }); + + it("includes thinking tokens in completion_tokens per the OpenAI convention", async () => { + // Thinking models report thoughtsTokenCount separately; OpenAI-compatible + // usage counts reasoning inside completion_tokens, keeping the identity + // total_tokens === prompt_tokens + completion_tokens intact. + generateContentStream.mockResolvedValue( + (async function* () { + yield { + usageMetadata: { + promptTokenCount: 10, + candidatesTokenCount: 5, + thoughtsTokenCount: 20, + totalTokenCount: 35, + }, + candidates: [ + { + content: { role: "model", parts: [{ text: "hi" }] }, + finishReason: "STOP", + }, + ], + }; + })(), + ); + + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ provider: "gemini", apiKey: "k" }); + + let usage: + | { + prompt_tokens: number; + completion_tokens: number; + total_tokens: number; + } + | undefined; + for await (const chunk of api.chatCompletionStream( + { + model: "gemini-flash-latest", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + if (chunk.usage) { + usage = chunk.usage; + } + } + + expect(usage).toEqual({ + prompt_tokens: 10, + completion_tokens: 25, + total_tokens: 35, + }); + expect(usage!.total_tokens).toBe( + usage!.prompt_tokens + usage!.completion_tokens, + ); + }); +}); + +describe("GeminiApi embed (SDK-native)", () => { + afterEach(() => { + vi.clearAllMocks(); + }); + + it("calls SDK embedContent and maps embeddings to OpenAI format", async () => { + // Google's REAL response shape, live-verified 2026-07-28: + // { embeddings: [{ values: [...] }] } — no batchEmbedContents, no usage. + embedContent.mockResolvedValue({ + embeddings: [{ values: [0.1, 0.2] }, { values: [0.3, 0.4] }], + }); + + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + }); + + const result = await api.embed({ + model: "gemini-embedding-001", + input: ["hello", "world"], + }); + + expect(embedContent).toHaveBeenCalledWith({ + model: "gemini-embedding-001", + contents: ["hello", "world"], + }); + expect(result.data).toEqual([ + { index: 0, embedding: [0.1, 0.2], object: "embedding" }, + { index: 1, embedding: [0.3, 0.4], object: "embedding" }, + ]); + // Google reports no token counts for embeddings — shared helper default + expect(result.usage).toEqual({ prompt_tokens: 0, total_tokens: 0 }); + }); + + it("wraps a single string input into a one-element contents array", async () => { + embedContent.mockResolvedValue({ embeddings: [{ values: [0.5] }] }); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + }); + + await api.embed({ model: "gemini-embedding-001", input: "hi" }); + + expect(embedContent).toHaveBeenCalledWith({ + model: "gemini-embedding-001", + contents: ["hi"], + }); + }); + + it("normalizes SDK errors through the shared nested-message extraction", async () => { + const googleBody = JSON.stringify( + { + error: { + code: 429, + message: "You exceeded your current quota.", + status: "RESOURCE_EXHAUSTED", + }, + }, + null, + 2, + ); + class FakeApiError extends Error { + status = 429; + } + embedContent.mockRejectedValue( + new FakeApiError( + JSON.stringify({ + error: { + message: `${googleBody}\n`, + code: 429, + status: "Too Many Requests", + }, + }), + ), + ); + + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + }); + + let thrown: (Error & { status?: number }) | undefined; + try { + await api.embed({ model: "gemini-embedding-001", input: "hi" }); + } catch (error) { + thrown = error as Error & { status?: number }; + } + + expect(thrown).toBeInstanceOf(Error); + expect(thrown!.message).toContain("You exceeded your current quota"); + expect(thrown!.message.startsWith('{"error"')).toBe(false); + expect(thrown!.status).toBe(429); + }); + + it("throws when the SDK returns no embeddings", async () => { + embedContent.mockResolvedValue({ embeddings: undefined }); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + }); + + await expect( + api.embed({ model: "gemini-embedding-001", input: "hi" }), + ).rejects.toThrow(/no embeddings/i); + }); + + it("throws naming the index when an embedding entry has no values", async () => { + embedContent.mockResolvedValue({ + embeddings: [{ values: [0.1] }, {}], + }); + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ + provider: "gemini", + apiKey: "primary-api-key", + }); + + await expect( + api.embed({ model: "gemini-embedding-001", input: ["a", "b"] }), + ).rejects.toThrow(/no values for embedding at index 1/i); + }); +}); diff --git a/packages/openai-adapters/src/test/gemini-fetch-adaptation.vitest.ts b/packages/openai-adapters/src/test/gemini-fetch-adaptation.vitest.ts new file mode 100644 index 00000000000..4d76e5e46f2 --- /dev/null +++ b/packages/openai-adapters/src/test/gemini-fetch-adaptation.vitest.ts @@ -0,0 +1,355 @@ +import { Readable } from "node:stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { nativeFetch } from "../util/nativeFetch.js"; + +const fetchwithRequestOptionsMock = vi.fn(); +vi.mock("@continuedev/fetch", async () => { + const actual = await vi.importActual("@continuedev/fetch"); + return { + ...actual, + fetchwithRequestOptions: (...args: unknown[]) => + fetchwithRequestOptionsMock(...args), + }; +}); + +/** Minimal stand-in for a node-fetch Response: Node Readable body, no getReader. */ +function nodeFetchStyleResponse( + chunks: string[], + init: { status?: number; statusText?: string; headers?: [string, string][] }, +) { + return { + status: init.status ?? 200, + statusText: init.statusText ?? "OK", + headers: new Map(init.headers ?? []), + body: + chunks.length > 0 + ? Readable.from(chunks.map((chunk) => Buffer.from(chunk))) + : null, + }; +} + +async function readAll(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let out = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + out += decoder.decode(value, { stream: true }); + } + return out; +} + +describe("adaptToNativeResponse", () => { + it("adapts a node-fetch-style Response so its body supports getReader()", async () => { + const { adaptToNativeResponse } = await import( + "../util/requestOptionsFetch.js" + ); + + const adapted = adaptToNativeResponse( + nodeFetchStyleResponse( + ['data: {"text":"hel', 'lo"}\n\n', "data: [DONE]\n\n"], + { + status: 200, + statusText: "OK", + headers: [["content-type", "text/event-stream"]], + }, + ), + ); + + expect(adapted).toBeInstanceOf(Response); + expect(adapted.status).toBe(200); + expect(adapted.statusText).toBe("OK"); + expect(adapted.headers.get("content-type")).toBe("text/event-stream"); + expect(typeof adapted.body?.getReader).toBe("function"); + await expect(readAll(adapted.body!)).resolves.toBe( + 'data: {"text":"hello"}\n\ndata: [DONE]\n\n', + ); + }); + + it("produces a null body for null-body statuses", async () => { + const { adaptToNativeResponse } = await import( + "../util/requestOptionsFetch.js" + ); + + const adapted = adaptToNativeResponse( + nodeFetchStyleResponse([], { + status: 204, + statusText: "No Content", + }), + ); + + expect(adapted.status).toBe(204); + expect(adapted.body).toBeNull(); + }); +}); + +describe("withRequestOptionsFetch", () => { + afterEach(() => { + vi.unstubAllEnvs(); + vi.clearAllMocks(); + }); + + it("routes fetch through customFetch when proxy is configured, then restores", async () => { + const { withRequestOptionsFetch } = await import( + "../util/requestOptionsFetch.js" + ); + + fetchwithRequestOptionsMock.mockResolvedValue( + nodeFetchStyleResponse(["ok"], { status: 200, statusText: "OK" }), + ); + + const before = globalThis.fetch; + const requestOptions = { proxy: "http://proxy.example.com:8080" }; + + const result = await withRequestOptionsFetch(requestOptions, async () => { + const response = await globalThis.fetch( + "https://generativelanguage.googleapis.com/v1beta/models", + ); + return response; + }); + + expect(fetchwithRequestOptionsMock).toHaveBeenCalledTimes(1); + expect(fetchwithRequestOptionsMock.mock.calls[0][0].toString()).toBe( + "https://generativelanguage.googleapis.com/v1beta/models", + ); + expect(fetchwithRequestOptionsMock.mock.calls[0][2]).toEqual( + requestOptions, + ); + expect(result).toBeInstanceOf(Response); + expect(typeof result.body?.getReader).toBe("function"); + expect(globalThis.fetch).toBe(before); + }); + + function clearProxyEnv(): void { + vi.stubEnv("HTTPS_PROXY", ""); + vi.stubEnv("https_proxy", ""); + vi.stubEnv("HTTP_PROXY", ""); + vi.stubEnv("http_proxy", ""); + } + + it("uses the native-fetch fast path when no proxy/TLS options are set", async () => { + clearProxyEnv(); + const { withRequestOptionsFetch } = await import( + "../util/requestOptionsFetch.js" + ); + + // headers/timeout alone are handled via SDK httpOptions — not this wrapper + const seen: (typeof globalThis.fetch)[] = []; + await withRequestOptionsFetch( + { headers: { "x-api-key": "k" }, timeout: 5 }, + async () => { + seen.push(globalThis.fetch); + }, + ); + await withRequestOptionsFetch(undefined, async () => { + seen.push(globalThis.fetch); + }); + + expect(seen).toEqual([nativeFetch, nativeFetch]); + expect(fetchwithRequestOptionsMock).not.toHaveBeenCalled(); + }); + + it("engages the wrapper when only an environment proxy is set", async () => { + clearProxyEnv(); + vi.stubEnv("HTTPS_PROXY", "http://proxy.corp.example:8080"); + const { withRequestOptionsFetch } = await import( + "../util/requestOptionsFetch.js" + ); + + let observedFetch: typeof globalThis.fetch | undefined; + await withRequestOptionsFetch(undefined, async () => { + observedFetch = globalThis.fetch; + }); + + // Corporate env-var proxies (the reason a user's Python sample works + // while Continue fails) must route through customFetch like every + // other provider — not the native fast path. + expect(observedFetch).toBeDefined(); + expect(observedFetch).not.toBe(nativeFetch); + }); + + it("engages the wrapper for lowercase http_proxy too", async () => { + clearProxyEnv(); + vi.stubEnv("http_proxy", "http://proxy.corp.example:8080"); + const { withRequestOptionsFetch } = await import( + "../util/requestOptionsFetch.js" + ); + + let observedFetch: typeof globalThis.fetch | undefined; + await withRequestOptionsFetch(undefined, async () => { + observedFetch = globalThis.fetch; + }); + + expect(observedFetch).not.toBe(nativeFetch); + }); + + it("restores all four swapped globals even when the callback throws", async () => { + const { withRequestOptionsFetch } = await import( + "../util/requestOptionsFetch.js" + ); + + // Sentinels make every restore line falsifiable: the ambient globals + // already equal the native classes the wrapper installs, so asserting on + // the ambient values could never catch a deleted restore line. With + // sentinels installed first, the try block replaces them with natives and + // ONLY a working finally can bring each sentinel back. + const sentinelFetch: typeof globalThis.fetch = async () => new Response(); + class SentinelResponse extends Response {} + class SentinelRequest extends Request {} + class SentinelHeaders extends Headers {} + + const ambient = { + fetch: globalThis.fetch, + Response: globalThis.Response, + Request: globalThis.Request, + Headers: globalThis.Headers, + }; + try { + globalThis.fetch = sentinelFetch; + globalThis.Response = SentinelResponse; + globalThis.Request = SentinelRequest; + globalThis.Headers = SentinelHeaders; + + await expect( + withRequestOptionsFetch({ verifySsl: false }, async () => { + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + + expect(globalThis.fetch).toBe(sentinelFetch); + expect(globalThis.Response).toBe(SentinelResponse); + expect(globalThis.Request).toBe(SentinelRequest); + expect(globalThis.Headers).toBe(SentinelHeaders); + } finally { + globalThis.fetch = ambient.fetch; + globalThis.Response = ambient.Response; + globalThis.Request = ambient.Request; + globalThis.Headers = ambient.Headers; + } + }); + + it("serializes concurrent swaps so one config never observes another's fetch", async () => { + const { withRequestOptionsFetch } = await import( + "../util/requestOptionsFetch.js" + ); + fetchwithRequestOptionsMock.mockResolvedValue( + nodeFetchStyleResponse(["ok"], { status: 200, statusText: "OK" }), + ); + + // Each call records the fetch identity it observes across an await point + // (yielding to the event loop mid-window). Without serialization the two + // swaps interleave and each sees the OTHER's wrapped fetch installed. + async function run(tag: string): Promise<(typeof globalThis.fetch)[]> { + const seen: (typeof globalThis.fetch)[] = []; + await withRequestOptionsFetch( + { proxy: `http://proxy-${tag}.example:8080` }, + async () => { + seen.push(globalThis.fetch); + // Yield twice so a sibling swap has every chance to overwrite. + await Promise.resolve(); + await Promise.resolve(); + seen.push(globalThis.fetch); + }, + ); + return seen; + } + + const [a, b] = await Promise.all([run("a"), run("b")]); + + // Within each call, the fetch identity observed before and after the + // await must be stable — never swapped out from under it by the sibling. + expect(a[0]).toBe(a[1]); + expect(b[0]).toBe(b[1]); + // And the two calls must have observed DIFFERENT wrapped fetches + // (each bound to its own requestOptions), never a shared one. + expect(a[0]).not.toBe(b[0]); + }); + + it("releases the swap lock at establishment, not during stream consumption", async () => { + const { withRequestOptionsFetch } = await import( + "../util/requestOptionsFetch.js" + ); + fetchwithRequestOptionsMock.mockResolvedValue( + nodeFetchStyleResponse(["ok"], { status: 200, statusText: "OK" }), + ); + + // `fn` resolves when the SDK stream is ESTABLISHED; body iteration happens + // afterward, outside withRequestOptionsFetch. Model that: config A's + // establishment resolves immediately, then a pending "stream" continues. + // Config B must be able to acquire the lock and run WHILE A's stream is + // still in flight — proving the lock covers only establishment. + let bRanWhileAStreaming = false; + let resolveAStream!: () => void; + const aStreamDone = new Promise((r) => { + resolveAStream = r; + }); + + const aChain = withRequestOptionsFetch( + { proxy: "http://proxy-a.example:8080" }, + async () => { + /* establishment completes — lock should release here */ + }, + ).then(() => aStreamDone); // post-establishment stream consumption + + await withRequestOptionsFetch( + { proxy: "http://proxy-b.example:8080" }, + async () => { + bRanWhileAStreaming = true; + }, + ); + + // B completed while A's stream is still pending (aStreamDone unresolved). + expect(bRanWhileAStreaming).toBe(true); + + resolveAStream(); + await aChain; + }); +}); + +describe("GeminiApi SDK-call fetch routing", () => { + afterEach(() => { + vi.clearAllMocks(); + vi.resetModules(); + }); + + it("runs generateContentStream under the adapted fetch when proxy is set", async () => { + vi.doMock("@google/genai", () => { + const generateContentStream = vi.fn().mockImplementation(async () => { + // Capture the fetch active DURING the SDK call + capturedFetch = globalThis.fetch; + return (async function* () {})(); + }); + return { + GoogleGenAI: vi.fn().mockImplementation(() => ({ + models: { generateContentStream }, + })), + }; + }); + let capturedFetch: typeof globalThis.fetch | undefined; + + const { GeminiApi } = await import("../apis/Gemini.js"); + const api = new GeminiApi({ + provider: "gemini", + apiKey: "k", + requestOptions: { proxy: "http://proxy.example.com:8080" }, + }); + + const stream = api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + ); + for await (const _chunk of stream) { + // drain + } + + expect(capturedFetch).toBeDefined(); + expect(capturedFetch).not.toBe(nativeFetch); + expect(globalThis.fetch).not.toBe(capturedFetch); + }); +}); diff --git a/packages/openai-adapters/src/test/gemini-proxy-e2e.vitest.ts b/packages/openai-adapters/src/test/gemini-proxy-e2e.vitest.ts new file mode 100644 index 00000000000..95999ec8473 --- /dev/null +++ b/packages/openai-adapters/src/test/gemini-proxy-e2e.vitest.ts @@ -0,0 +1,449 @@ +import http from "node:http"; +import { AddressInfo } from "node:net"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; + +import { GeminiApi } from "../apis/Gemini.js"; +import { sseChunk } from "./gemini-test-helpers.js"; + +/** + * End-to-end proof for the proxy path: the REAL @google/genai SDK, the REAL + * customFetch/fetchwithRequestOptions stack (no mocks anywhere), a real local + * HTTP forward proxy, and a stub Gemini server emitting SSE. Verifies the + * node-fetch Response adaptation streams through the SDK's own parser. + */ + +let geminiServer: http.Server; +let proxyServer: http.Server; +let geminiPort: number; +let proxyPort: number; + +/** Requests the stub Gemini server actually received. */ +const geminiRequests: { + url: string; + apiKey: string | undefined; + customHeader: string | undefined; +}[] = []; +/** Number of requests that transited the proxy. */ +let proxiedRequests = 0; + +beforeAll(async () => { + geminiServer = http.createServer((req, res) => { + geminiRequests.push({ + url: req.url ?? "", + apiKey: req.headers["x-goog-api-key"] as string | undefined, + customHeader: req.headers["x-custom-gateway"] as string | undefined, + }); + if ((req.url ?? "").includes("mbedContent")) { + // embedContent / batchEmbedContents — Google's real response shape + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ embeddings: [{ values: [0.25, 0.75] }] })); + return; + } + res.writeHead(200, { "Content-Type": "text/event-stream" }); + res.write(sseChunk("Hello ")); + res.write(sseChunk("from the proxied stub", "STOP")); + res.end(); + }); + await new Promise((resolve) => + geminiServer.listen(0, "127.0.0.1", resolve), + ); + geminiPort = (geminiServer.address() as AddressInfo).port; + + proxyServer = makeForwardProxy(() => { + proxiedRequests += 1; + }); + await new Promise((resolve) => + proxyServer.listen(0, "127.0.0.1", resolve), + ); + proxyPort = (proxyServer.address() as AddressInfo).port; +}); + +/** + * Minimal HTTP forward proxy: receives absolute-URI requests, forwards them, + * pipes the response back. onRequest fires for every transit — the + * discriminating signal for which proxy (if any) a request dialed. + */ +function makeForwardProxy( + onRequest: (req: http.IncomingMessage) => void, +): http.Server { + return http.createServer((req, res) => { + onRequest(req); + const target = new URL(req.url ?? ""); + const upstream = http.request( + { + hostname: target.hostname, + port: target.port, + path: target.pathname + target.search, + method: req.method, + headers: { ...req.headers, host: target.host }, + }, + (upstreamRes) => { + res.writeHead(upstreamRes.statusCode ?? 502, upstreamRes.headers); + upstreamRes.pipe(res); + }, + ); + req.pipe(upstream); + }); +} + +afterAll(async () => { + await new Promise((resolve) => geminiServer.close(resolve)); + await new Promise((resolve) => proxyServer.close(resolve)); +}); + +describe("Gemini streaming through a real local proxy (no mocks)", () => { + beforeEach(() => { + // Deterministic regardless of the machine's ambient proxy environment; + // individual tests re-stub what they need. + vi.stubEnv("HTTP_PROXY", ""); + vi.stubEnv("http_proxy", ""); + vi.stubEnv("HTTPS_PROXY", ""); + vi.stubEnv("https_proxy", ""); + vi.stubEnv("NO_PROXY", ""); + vi.stubEnv("no_proxy", ""); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it("streams SSE chunks via the real SDK, customFetch, and proxy", async () => { + const proxiedBefore = proxiedRequests; + const requestsBefore = geminiRequests.length; + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${geminiPort}/v1beta/`, + requestOptions: { + proxy: `http://127.0.0.1:${proxyPort}`, + }, + }); + + let content = ""; + for await (const chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + content += chunk.choices[0]?.delta?.content ?? ""; + } + + expect(content).toBe("Hello from the proxied stub"); + expect(proxiedRequests).toBe(proxiedBefore + 1); + const observed = geminiRequests[requestsBefore]; + expect(observed.apiKey).toBe("stub-key"); + expect(observed.url).toContain(":streamGenerateContent"); + }); + + it("delivers custom requestOptions.headers to the wire", async () => { + const before = geminiRequests.length; + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${geminiPort}/v1beta/`, + requestOptions: { + proxy: `http://127.0.0.1:${proxyPort}`, + headers: { "x-custom-gateway": "gw-credential-123" }, + }, + }); + + for await (const _chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + // drain + } + + const observed = geminiRequests[before]; + expect(observed.customHeader).toBe("gw-credential-123"); + }); + + it("uses an environment proxy when no config proxy is set", async () => { + // The corporate case: proxy exists only as env vars (why a user's plain + // Python sample works). No requestOptions at all. + vi.stubEnv("HTTP_PROXY", `http://127.0.0.1:${proxyPort}`); + vi.stubEnv("NO_PROXY", ""); + vi.stubEnv("no_proxy", ""); + + const proxiedBefore = proxiedRequests; + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${geminiPort}/v1beta/`, + }); + + let content = ""; + for await (const chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + content += chunk.choices[0]?.delta?.content ?? ""; + } + + expect(content).toBe("Hello from the proxied stub"); + expect(proxiedRequests).toBe(proxiedBefore + 1); + }); + + it("delivers custom headers alongside an environment proxy", async () => { + vi.stubEnv("HTTP_PROXY", `http://127.0.0.1:${proxyPort}`); + const proxiedBefore = proxiedRequests; + const requestsBefore = geminiRequests.length; + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${geminiPort}/v1beta/`, + requestOptions: { + headers: { "x-custom-gateway": "gw-credential-456" }, + }, + }); + + for await (const _chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + // drain + } + + // Env proxy engages the wrapper AND custom headers still reach the wire. + expect(proxiedRequests).toBe(proxiedBefore + 1); + expect(geminiRequests[requestsBefore].customHeader).toBe( + "gw-credential-456", + ); + }); + + it("goes direct when NO_PROXY covers the target host", async () => { + vi.stubEnv("HTTP_PROXY", `http://127.0.0.1:${proxyPort}`); + vi.stubEnv("NO_PROXY", "127.0.0.1"); + + const proxiedBefore = proxiedRequests; + const requestsBefore = geminiRequests.length; + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${geminiPort}/v1beta/`, + }); + + let content = ""; + for await (const chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + content += chunk.choices[0]?.delta?.content ?? ""; + } + + expect(content).toBe("Hello from the proxied stub"); + expect(proxiedRequests).toBe(proxiedBefore); // proxy NOT used + expect(geminiRequests.length).toBe(requestsBefore + 1); // server reached directly + }); + + it("prefers the config proxy over an environment proxy", async () => { + // Two REAL proxies: env points at one, config at the other. The + // hit-counters discriminate which was actually dialed — directly + // exercising getProxy()'s config-first precedence at the wire. + let envProxyHits = 0; + const envProxy = makeForwardProxy(() => { + envProxyHits += 1; + }); + await new Promise((resolve) => + envProxy.listen(0, "127.0.0.1", resolve), + ); + const envProxyPort = (envProxy.address() as AddressInfo).port; + + try { + vi.stubEnv("HTTP_PROXY", `http://127.0.0.1:${envProxyPort}`); + + const configProxiedBefore = proxiedRequests; + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${geminiPort}/v1beta/`, + requestOptions: { + proxy: `http://127.0.0.1:${proxyPort}`, + }, + }); + + let content = ""; + for await (const chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + content += chunk.choices[0]?.delta?.content ?? ""; + } + + expect(content).toBe("Hello from the proxied stub"); + expect(proxiedRequests).toBe(configProxiedBefore + 1); // config proxy dialed + expect(envProxyHits).toBe(0); // env proxy never touched + } finally { + await new Promise((resolve) => envProxy.close(resolve)); + } + }); + + it("isolates proxy routing and credentials between two concurrent configs", async () => { + // The security-critical case: two GeminiApi configs, each behind its own + // proxy with its own gateway credential, run concurrently. Each request + // must stay on its own proxy carrying only its own credential — never + // config A's credential riding config B's request (the leak the swap + // mutex prevents). + const proxyAHeaders: (string | undefined)[] = []; + const proxyBHeaders: (string | undefined)[] = []; + const proxyA = makeForwardProxy((req) => + proxyAHeaders.push(req.headers["x-custom-gateway"] as string | undefined), + ); + const proxyB = makeForwardProxy((req) => + proxyBHeaders.push(req.headers["x-custom-gateway"] as string | undefined), + ); + await new Promise((r) => proxyA.listen(0, "127.0.0.1", r)); + await new Promise((r) => proxyB.listen(0, "127.0.0.1", r)); + const portA = (proxyA.address() as AddressInfo).port; + const portB = (proxyB.address() as AddressInfo).port; + + async function drain(port: number, credential: string): Promise { + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${geminiPort}/v1beta/`, + requestOptions: { + proxy: `http://127.0.0.1:${port}`, + headers: { "x-custom-gateway": credential }, + }, + }); + let content = ""; + for await (const chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + content += chunk.choices[0]?.delta?.content ?? ""; + } + return content; + } + + try { + const [ca, cb] = await Promise.all([ + drain(portA, "cred-A"), + drain(portB, "cred-B"), + ]); + + // Both streams completed concurrently (Promise.all) — serialization did + // not hold across body consumption. + expect(ca).toBe("Hello from the proxied stub"); + expect(cb).toBe("Hello from the proxied stub"); + // Each request transited ONLY its own proxy, carrying ONLY its own + // credential — zero cross-contamination. + expect(proxyAHeaders).toEqual(["cred-A"]); + expect(proxyBHeaders).toEqual(["cred-B"]); + } finally { + await new Promise((r) => proxyA.close(r)); + await new Promise((r) => proxyB.close(r)); + } + }); + + it("routes embed through the same proxy and parses Google's real shape", async () => { + const proxiedBefore = proxiedRequests; + const requestsBefore = geminiRequests.length; + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${geminiPort}/v1beta/`, + requestOptions: { + proxy: `http://127.0.0.1:${proxyPort}`, + }, + }); + + const result = await api.embed({ + model: "gemini-embedding-001", + input: ["hello"], + }); + + expect(proxiedRequests).toBe(proxiedBefore + 1); + expect(geminiRequests[requestsBefore].url).toContain("mbedContent"); + expect(result.data[0].embedding).toEqual([0.25, 0.75]); + }); +}); + +describe("Gemini requestOptions.timeout end-to-end (no mocks)", () => { + let slowServer: http.Server; + let slowPort: number; + + beforeAll(async () => { + // Never responds within the test's timeout window — forces the SDK's + // own timeout/abort path to fire. + slowServer = http.createServer((_req, res) => { + setTimeout(() => { + res.writeHead(200, { "Content-Type": "text/event-stream" }); + res.end(); + }, 5000); + }); + await new Promise((r) => slowServer.listen(0, "127.0.0.1", r)); + slowPort = (slowServer.address() as AddressInfo).port; + }); + + afterAll(async () => { + await new Promise((r) => slowServer.close(r)); + }); + + it("aborts a chat call when the configured timeout elapses", async () => { + const api = new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `http://127.0.0.1:${slowPort}/v1beta/`, + requestOptions: { timeout: 100 }, + }); + + const started = Date.now(); + let threw = false; + try { + for await (const _chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + // never arrives + } + } catch { + threw = true; + } + + // The call aborted from the timeout, not by waiting out the 5s server. + expect(threw).toBe(true); + expect(Date.now() - started).toBeLessThan(4000); + }); +}); diff --git a/packages/openai-adapters/src/test/gemini-test-helpers.ts b/packages/openai-adapters/src/test/gemini-test-helpers.ts new file mode 100644 index 00000000000..1cc4d67b77b --- /dev/null +++ b/packages/openai-adapters/src/test/gemini-test-helpers.ts @@ -0,0 +1,15 @@ +/** + * Shared helpers for the Gemini end-to-end test rigs (proxy and TLS stubs). + */ + +/** One SSE frame in the Gemini streaming response format. */ +export function sseChunk(text: string, finishReason?: string): string { + const candidate: Record = { + content: { role: "model", parts: [{ text }] }, + index: 0, + }; + if (finishReason) { + candidate.finishReason = finishReason; + } + return `data: ${JSON.stringify({ candidates: [candidate] })}\r\n\r\n`; +} diff --git a/packages/openai-adapters/src/test/gemini-tls-e2e.vitest.ts b/packages/openai-adapters/src/test/gemini-tls-e2e.vitest.ts new file mode 100644 index 00000000000..c9e995d0bf8 --- /dev/null +++ b/packages/openai-adapters/src/test/gemini-tls-e2e.vitest.ts @@ -0,0 +1,291 @@ +import { execSync } from "node:child_process"; +import fs from "node:fs"; +import https from "node:https"; +import { AddressInfo } from "node:net"; +import os from "node:os"; +import path from "node:path"; +import type { TLSSocket } from "node:tls"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; + +import { GeminiApi } from "../apis/Gemini.js"; +import { sseChunk } from "./gemini-test-helpers.js"; + +/** + * TLS and mutual-TLS behavior through the REAL @google/genai SDK and the REAL + * customFetch/fetchwithRequestOptions stack (no mocks): local HTTPS stub + * Gemini servers backed by an openssl mini-CA. Certificate choreography + * adapts packages/fetch/src/fetch.e2e.test.ts and the standard openssl + * CA-sign recipe (see also Node core's tls-client-verify tests). + */ + +let tempDir: string; +let tlsServer: https.Server; +let mtlsServer: https.Server; +let tlsPort: number; +let mtlsPort: number; +let caCertPath: string; +let clientCertPath: string; +let clientKeyPath: string; +let clientEncryptedKeyPath: string; + +/** Client identities the mTLS server actually observed, per request. */ +const mtlsObserved: { authorized: boolean; cn: string | undefined }[] = []; + +const CLIENT_CN = "continue-mtls-client"; +const CLIENT_KEY_PASSPHRASE = "test-passphrase"; + +/** + * Mini-CA: CA keypair, a CA-signed server certificate for 127.0.0.1, and a + * CA-signed client certificate (plus a passphrase-encrypted copy of the + * client key). Same openssl toolchain as packages/fetch. + */ +function generateMiniCa(dir: string): void { + const run = (cmd: string) => execSync(cmd, { stdio: "pipe" }); + + // CA + run(`openssl genrsa -out "${path.join(dir, "ca.key")}" 2048`); + run( + `openssl req -x509 -new -key "${path.join(dir, "ca.key")}" -out "${path.join(dir, "ca.crt")}" -days 365 -subj "/C=US/O=Continue Test CA/CN=Continue Test CA"`, + ); + + // Server certificate (SAN: 127.0.0.1 / localhost), signed by the CA + const serverConf = path.join(dir, "server.conf"); + fs.writeFileSync( + serverConf, + ` +[req] +distinguished_name = req_distinguished_name +req_extensions = v3_req +prompt = no + +[req_distinguished_name] +C = US +O = Continue TLS Test +CN = 127.0.0.1 + +[v3_req] +keyUsage = keyEncipherment, dataEncipherment +extendedKeyUsage = serverAuth +subjectAltName = @alt_names + +[alt_names] +IP.1 = 127.0.0.1 +DNS.1 = localhost +`, + ); + run(`openssl genrsa -out "${path.join(dir, "server.key")}" 2048`); + run( + `openssl req -new -key "${path.join(dir, "server.key")}" -out "${path.join(dir, "server.csr")}" -config "${serverConf}"`, + ); + run( + `openssl x509 -req -in "${path.join(dir, "server.csr")}" -CA "${path.join(dir, "ca.crt")}" -CAkey "${path.join(dir, "ca.key")}" -CAcreateserial -out "${path.join(dir, "server.crt")}" -days 365 -extensions v3_req -extfile "${serverConf}"`, + ); + + // Client certificate (clientAuth EKU), signed by the same CA + const clientExt = path.join(dir, "client.ext"); + fs.writeFileSync(clientExt, "extendedKeyUsage = clientAuth\n"); + run(`openssl genrsa -out "${path.join(dir, "client.key")}" 2048`); + run( + `openssl req -new -key "${path.join(dir, "client.key")}" -out "${path.join(dir, "client.csr")}" -subj "/C=US/O=Continue TLS Test/CN=${CLIENT_CN}"`, + ); + run( + `openssl x509 -req -in "${path.join(dir, "client.csr")}" -CA "${path.join(dir, "ca.crt")}" -CAkey "${path.join(dir, "ca.key")}" -CAcreateserial -out "${path.join(dir, "client.crt")}" -days 365 -extfile "${clientExt}"`, + ); + + // Passphrase-encrypted copy of the client key (schema's passphrase field) + run( + `openssl rsa -in "${path.join(dir, "client.key")}" -aes256 -passout pass:${CLIENT_KEY_PASSPHRASE} -out "${path.join(dir, "client-encrypted.key")}"`, + ); +} + +function sseHandler( + _req: unknown, + res: import("node:http").ServerResponse, +): void { + res.writeHead(200, { "Content-Type": "text/event-stream" }); + res.write(sseChunk("secure ", undefined)); + res.write(sseChunk("stream", "STOP")); + res.end(); +} + +beforeAll(async () => { + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "gemini-tls-test-")); + generateMiniCa(tempDir); + caCertPath = path.join(tempDir, "ca.crt"); + clientCertPath = path.join(tempDir, "client.crt"); + clientKeyPath = path.join(tempDir, "client.key"); + clientEncryptedKeyPath = path.join(tempDir, "client-encrypted.key"); + + // Plain TLS server (no client-cert requirement) + tlsServer = https.createServer( + { + cert: fs.readFileSync(path.join(tempDir, "server.crt")), + key: fs.readFileSync(path.join(tempDir, "server.key")), + }, + sseHandler, + ); + await new Promise((resolve) => + tlsServer.listen(0, "127.0.0.1", resolve), + ); + tlsPort = (tlsServer.address() as AddressInfo).port; + + // Mutual-TLS server: demands a client certificate signed by our CA + mtlsServer = https.createServer( + { + cert: fs.readFileSync(path.join(tempDir, "server.crt")), + key: fs.readFileSync(path.join(tempDir, "server.key")), + ca: fs.readFileSync(caCertPath), + requestCert: true, + rejectUnauthorized: true, + }, + (req, res) => { + const socket = req.socket as TLSSocket; + const cn = socket.getPeerCertificate()?.subject?.CN; + mtlsObserved.push({ + authorized: socket.authorized, + cn: Array.isArray(cn) ? cn[0] : cn, + }); + sseHandler(req, res); + }, + ); + await new Promise((resolve) => + mtlsServer.listen(0, "127.0.0.1", resolve), + ); + mtlsPort = (mtlsServer.address() as AddressInfo).port; + // Seven sequential openssl invocations — generous timeout for contended + // CI workers (runs in ~1s locally). +}, 30_000); + +afterAll(async () => { + await new Promise((resolve) => tlsServer.close(resolve)); + await new Promise((resolve) => mtlsServer.close(resolve)); + fs.rmSync(tempDir, { recursive: true, force: true }); +}); + +function makeApi( + port: number, + requestOptions: Record, +): GeminiApi { + return new GeminiApi({ + provider: "gemini", + apiKey: "stub-key", + apiBase: `https://127.0.0.1:${port}/v1beta/`, + requestOptions, + }); +} + +async function drainChat(api: GeminiApi): Promise { + let content = ""; + for await (const chunk of api.chatCompletionStream( + { + model: "gemini-2.5-flash", + messages: [{ role: "user", content: "hi" }], + stream: true, + }, + new AbortController().signal, + )) { + content += chunk.choices[0]?.delta?.content ?? ""; + } + return content; +} + +describe("Gemini TLS through the real SDK (no mocks)", () => { + beforeEach(() => { + // Deterministic under ambient corporate proxy environments — without + // this, a machine with HTTPS_PROXY set routes these stub-server calls + // into the proxy and the whole suite fails for non-TLS reasons. + vi.stubEnv("HTTP_PROXY", ""); + vi.stubEnv("http_proxy", ""); + vi.stubEnv("HTTPS_PROXY", ""); + vi.stubEnv("https_proxy", ""); + vi.stubEnv("NO_PROXY", ""); + vi.stubEnv("no_proxy", ""); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + }); + it("rejects a server signed by an unknown CA when verifySsl is on", async () => { + await expect( + drainChat(makeApi(tlsPort, { verifySsl: true })), + ).rejects.toThrow( + /self-signed|self signed|unable to verify|certificate|unknown ca/i, + ); + }); + + it("accepts the server when its CA is trusted via caBundlePath", async () => { + const content = await drainChat( + makeApi(tlsPort, { caBundlePath: caCertPath }), + ); + expect(content).toBe("secure stream"); + }); + + it("accepts the server when verifySsl is explicitly disabled", async () => { + const content = await drainChat(makeApi(tlsPort, { verifySsl: false })); + expect(content).toBe("secure stream"); + }); +}); + +describe("Gemini mutual TLS through the real SDK (no mocks)", () => { + beforeEach(() => { + vi.stubEnv("HTTP_PROXY", ""); + vi.stubEnv("http_proxy", ""); + vi.stubEnv("HTTPS_PROXY", ""); + vi.stubEnv("https_proxy", ""); + vi.stubEnv("NO_PROXY", ""); + vi.stubEnv("no_proxy", ""); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + }); + it("rejects the handshake when the server requires a client certificate and none is configured", async () => { + await expect( + drainChat(makeApi(mtlsPort, { caBundlePath: caCertPath })), + ).rejects.toThrow( + /certificate required|alert|socket hang up|ECONNRESET|EPROTO/i, + ); + expect(mtlsObserved).toHaveLength(0); + }); + + it("completes the handshake and streams when clientCertificate is configured", async () => { + const before = mtlsObserved.length; + const content = await drainChat( + makeApi(mtlsPort, { + caBundlePath: caCertPath, + clientCertificate: { cert: clientCertPath, key: clientKeyPath }, + }), + ); + + expect(content).toBe("secure stream"); + const observed = mtlsObserved[before]; + expect(observed.authorized).toBe(true); + expect(observed.cn).toBe(CLIENT_CN); + }); + + it("supports a passphrase-protected client key", async () => { + const before = mtlsObserved.length; + const content = await drainChat( + makeApi(mtlsPort, { + caBundlePath: caCertPath, + clientCertificate: { + cert: clientCertPath, + key: clientEncryptedKeyPath, + passphrase: CLIENT_KEY_PASSPHRASE, + }, + }), + ); + + expect(content).toBe("secure stream"); + expect(mtlsObserved[before].authorized).toBe(true); + }); +}); diff --git a/packages/openai-adapters/src/test/main.test.ts b/packages/openai-adapters/src/test/main.test.ts index 8a10c8b727f..8f3330f5067 100644 --- a/packages/openai-adapters/src/test/main.test.ts +++ b/packages/openai-adapters/src/test/main.test.ts @@ -80,7 +80,7 @@ const TESTS: Omit[] = [ }, { provider: "gemini", - model: "gemini-2.5-pro", + model: "gemini-pro-latest", apiKey: process.env.GEMINI_API_KEY!, roles: ["chat"], options: { @@ -90,7 +90,7 @@ const TESTS: Omit[] = [ }, { provider: "gemini", - model: "gemini-2.5-flash", + model: "gemini-flash-latest", apiKey: process.env.GEMINI_API_KEY!, roles: ["chat"], options: { diff --git a/packages/openai-adapters/src/util/requestOptionsFetch.ts b/packages/openai-adapters/src/util/requestOptionsFetch.ts new file mode 100644 index 00000000000..a73a16de923 --- /dev/null +++ b/packages/openai-adapters/src/util/requestOptionsFetch.ts @@ -0,0 +1,160 @@ +import { RequestOptions } from "@continuedev/config-types"; +import { Readable } from "node:stream"; + +import { customFetch } from "../util.js"; +import { + nativeFetch, + nativeHeaders, + nativeRequest, + nativeResponse, + withNativeFetch, +} from "./nativeFetch.js"; + +/** Statuses the WHATWG Response constructor rejects a non-null body for. */ +const NULL_BODY_STATUSES = new Set([101, 204, 205, 304]); + +/** + * True when requestOptions carry settings only customFetch can honor — + * proxying and TLS configuration. Headers and timeout are excluded on + * purpose: those already reach the @google/genai SDK via httpOptions. + */ +export function hasProxyOrTlsOptions( + requestOptions: RequestOptions | undefined, +): boolean { + return ( + !!requestOptions && + (requestOptions.proxy !== undefined || + requestOptions.verifySsl !== undefined || + requestOptions.caBundlePath !== undefined || + requestOptions.clientCertificate !== undefined) + ); +} + +/** + * True when an ambient HTTPS_PROXY/HTTP_PROXY environment proxy exists. + * Env-var proxies are the standard corporate setup and are honored by every + * customFetch-based provider; the Gemini SDK path must match. Resolution + * precedence (config over env) and NO_PROXY bypass stay entirely with + * customFetch/getProxy at request time — this predicate only decides whether + * to engage the wrapper, so it reads the same four variables + * @continuedev/fetch's getProxyFromEnv reads (that helper is not part of the + * package's public API). + */ +function hasEnvironmentProxy(): boolean { + const { HTTPS_PROXY, https_proxy, HTTP_PROXY, http_proxy } = process.env; + return !!(HTTPS_PROXY || https_proxy || HTTP_PROXY || http_proxy); +} + +/** + * Convert a node-fetch Response (Node Readable body, no getReader) into a + * native WHATWG Response the @google/genai SDK can stream from. Without this + * adaptation the SDK fails with "getReader is not a function" — the exact + * pollution problem documented in nativeFetch.ts. + */ +export function adaptToNativeResponse(response: { + status: number; + statusText: string; + headers: Iterable<[string, string]>; + body: Readable | ReadableStream | null; +}): Response { + const headers = new nativeHeaders([...response.headers]); + + let body: BodyInit | null = null; + if (response.body !== null && !NULL_BODY_STATUSES.has(response.status)) { + body = + response.body instanceof Readable + ? // Node's stream/web ReadableStream and the DOM lib type describe the + // same runtime object; the cast bridges the two type declarations. + (Readable.toWeb(response.body) as ReadableStream) + : response.body; + } + + return new nativeResponse(body, { + status: response.status, + statusText: response.statusText, + headers, + }); +} + +/** + * Serializes the global-fetch swap window across concurrent callers. + * + * The swapped `globalThis.fetch` is bound by closure to ONE config's + * requestOptions (its proxy Agent, client certificate, and headers). Without + * serialization, two concurrent Gemini calls with different configs race on + * the shared global — config A's gateway credential / mTLS identity could + * ride config B's request (chat behind a corporate gateway + concurrent + * embedding indexing is a normal Continue pattern). This mutex admits one + * swap window at a time. + * + * Only the swap window (call establishment) is held: `body` resolves once the + * SDK's stream is established, before body iteration, so streams still run + * concurrently after the lock is released. The tail promise is always + * released in `finally`, including when `body` throws. + */ +let fetchSwapChain: Promise = Promise.resolve(); + +async function withSerializedFetchSwap(body: () => Promise): Promise { + const prior = fetchSwapChain; + let release!: () => void; + fetchSwapChain = new Promise((resolve) => { + release = resolve; + }); + await prior; + try { + return await body(); + } finally { + release(); + } +} + +/** + * Run `fn` with globalThis.fetch honoring the given requestOptions. + * + * - No proxy/TLS options: delegates to withNativeFetch — byte-identical to + * the previous behavior for every existing config, and unlocked (the native + * fast path installs no config-bound credentials, so it needs no + * serialization). + * - Proxy/TLS options present: swaps globalThis.fetch for the duration of + * `fn` with a fetch that routes through customFetch(requestOptions) + * (proxy, CA bundles, client certs, verifySsl) and adapts its node-fetch + * Response to a native one so SDK streaming keeps working. The swap window + * is serialized (see withSerializedFetchSwap) so concurrent configs cannot + * leak credentials across each other. The previous globals are restored in + * a finally, including when `fn` throws. + */ +export async function withRequestOptionsFetch( + requestOptions: RequestOptions | undefined, + fn: () => Promise, +): Promise { + if (!hasProxyOrTlsOptions(requestOptions) && !hasEnvironmentProxy()) { + return withNativeFetch(fn); + } + + const optionsFetch = customFetch(requestOptions); + const wrappedFetch: typeof globalThis.fetch = async (input, init) => { + const response = await optionsFetch(input, init); + return adaptToNativeResponse(response); + }; + + return withSerializedFetchSwap(async () => { + const originalFetch = globalThis.fetch; + const originalResponse = globalThis.Response; + const originalRequest = globalThis.Request; + const originalHeaders = globalThis.Headers; + + try { + globalThis.fetch = wrappedFetch; + globalThis.Response = nativeResponse; + globalThis.Request = nativeRequest; + globalThis.Headers = nativeHeaders; + + return await fn(); + } finally { + globalThis.fetch = originalFetch; + globalThis.Response = originalResponse; + globalThis.Request = originalRequest; + globalThis.Headers = originalHeaders; + } + }); +}