diff --git a/packages/ai/src/provider-package.ts b/packages/ai/src/provider-package.ts index 79f290848332..bd42a4ddf9b8 100644 --- a/packages/ai/src/provider-package.ts +++ b/packages/ai/src/provider-package.ts @@ -5,6 +5,7 @@ export interface Settings extends Readonly> { readonly baseURL?: string readonly headers?: Readonly> readonly body?: Readonly> + readonly chunkTimeout?: number } export interface Definition< diff --git a/packages/ai/src/providers/azure.ts b/packages/ai/src/providers/azure.ts index c1ae110ea64f..92402f81373b 100644 --- a/packages/ai/src/providers/azure.ts +++ b/packages/ai/src/providers/azure.ts @@ -35,7 +35,6 @@ export type Settings = ProviderPackage.Settings & readonly useDeploymentBasedUrls?: boolean readonly providerOptions?: OpenAIProviderOptionsInput } - const resourceBaseURL = (resourceName: string) => `https://${resourceName.trim()}.openai.azure.com/openai` const responsesRoute = OpenAIResponses.route.with({ @@ -152,11 +151,18 @@ export const provider = { } const config = (settings: Settings): Config => { + const http = + settings.body === undefined && settings.chunkTimeout === undefined + ? undefined + : { + body: settings.body === undefined ? undefined : { ...settings.body }, + chunkTimeout: settings.chunkTimeout, + } const common = { apiKey: settings.apiKey, apiVersion: settings.apiVersion, headers: settings.headers === undefined ? undefined : { ...settings.headers }, - http: settings.body === undefined ? undefined : { body: { ...settings.body } }, + http, providerOptions: settings.providerOptions, queryParams: settings.queryParams === undefined ? undefined : { ...settings.queryParams }, useDeploymentBasedUrls: settings.useDeploymentBasedUrls, diff --git a/packages/ai/src/route/transport/http.ts b/packages/ai/src/route/transport/http.ts index c8f94d1a7a1a..d6eb6c2c100c 100644 --- a/packages/ai/src/route/transport/http.ts +++ b/packages/ai/src/route/transport/http.ts @@ -1,11 +1,11 @@ -import { Effect } from "effect" +import { Duration, Effect, Stream } from "effect" import { Headers, HttpClientRequest } from "effect/unstable/http" import { Auth } from "../auth.js" import { render as renderEndpoint } from "../endpoint.js" import { Framing } from "../framing.js" import type { HttpMiddleware, Transport, TransportPrepareInput } from "./index.js" import * as ProviderShared from "../../protocols/shared.js" -import { mergeJsonRecords, type LLMRequest } from "../../schema/index.js" +import { AIError, mergeJsonRecords, TransportError, type LLMRequest } from "../../schema/index.js" import { RequestExecutor } from "../executor.js" export type JsonRequestInput = TransportPrepareInput @@ -87,11 +87,36 @@ export const httpJson = (input: HttpJsonInput): HttpJs middleware: prepareInput.middleware, } }), - execute: (prepared, _request, runtime) => + execute: (prepared, request, runtime) => Effect.gen(function* () { const response = yield* runtime.http.execute(prepared.request, prepared.middleware) + const chunkTimeout = request.http?.chunkTimeout + const bytes = RequestExecutor.responseStream(response) + const guarded = + chunkTimeout === undefined || chunkTimeout <= 0 + ? bytes + : bytes.pipe( + // A stalled provider stream that sends nothing is otherwise bounded only by the + // OS socket; abort once no chunk arrives within the configured window. + Stream.timeoutOrElse({ + duration: Duration.millis(chunkTimeout), + orElse: () => + Stream.fail( + new AIError({ + reason: new TransportError({ + message: `No data received from ${response.request.url} within the ${chunkTimeout}ms chunkTimeout; the stream may be stalled`, + transport: "http", + operation: "read", + code: "chunk-timeout", + url: response.request.url, + phase: "receive", + }), + }), + ), + }), + ) return { - frames: prepared.framing.frame(RequestExecutor.responseStream(response)), + frames: prepared.framing.frame(guarded), http: RequestExecutor.responseHttp(response), body: prepared.framing.body, } diff --git a/packages/ai/src/schema/options.ts b/packages/ai/src/schema/options.ts index 27a16fbe3c83..ff04b0f9e6f8 100644 --- a/packages/ai/src/schema/options.ts +++ b/packages/ai/src/schema/options.ts @@ -47,6 +47,9 @@ export class HttpOptions extends Schema.Class("AI.HttpOptions")({ body: Schema.optional(JsonSchema), headers: Schema.optional(Schema.Record(Schema.String, Schema.String)), query: Schema.optional(Schema.Record(Schema.String, Schema.String)), + chunkTimeout: Schema.optional(Schema.Number).annotate({ + description: "Abort the stream when no data chunk arrives within this many milliseconds.", + }), }) {} export namespace HttpOptions { @@ -60,8 +63,9 @@ export const mergeHttpOptions = (...items: ReadonlyArray item?.body)) const headers = mergeStringRecords(...items.map((item) => item?.headers)) const query = mergeStringRecords(...items.map((item) => item?.query)) - if (!body && !headers && !query) return undefined - return new HttpOptions({ body, headers, query }) + const chunkTimeout = items.reduceRight((found, item) => found ?? item?.chunkTimeout, undefined as number | undefined) + if (!body && !headers && !query && chunkTimeout === undefined) return undefined + return new HttpOptions({ body, headers, query, chunkTimeout }) } export class GenerationOptions extends Schema.Class("LLM.GenerationOptions")({ diff --git a/packages/ai/test/chunk-timeout.test.ts b/packages/ai/test/chunk-timeout.test.ts new file mode 100644 index 000000000000..a8d87a778a6e --- /dev/null +++ b/packages/ai/test/chunk-timeout.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, test } from "bun:test" +import { Effect } from "effect" +import { AIError, LLM } from "../src/index.js" +import { HttpOptions, mergeHttpOptions } from "../src/schema/options.js" +import * as OpenAIChat from "../src/protocols/openai-chat.js" +import { LLMClient } from "../src/route.js" +import { dynamicResponse } from "./lib/http.js" +import { it } from "./lib/effect.js" + +const headers = { "content-type": "text/event-stream" } + +// Sends one SSE frame, then holds the connection open without further data. +// Only a configured chunkTimeout can terminate the read. +const stalledResponse = dynamicResponse((input) => + Effect.sync(() => { + const encoder = new TextEncoder() + let sent = false + const stream = new ReadableStream({ + pull(controller) { + if (!sent) { + sent = true + controller.enqueue(encoder.encode('data: {"choices":[]}\n\n')) + } + }, + }) + return input.respond(stream, { headers }) + }), +) + +describe("chunkTimeout", () => { + it.live("aborts a stalled HTTP SSE stream with a typed transport error", () => + Effect.gen(function* () { + const route = OpenAIChat.route.with({ http: { chunkTimeout: 100 } }) + const request = LLM.request({ model: route.model({ id: "test" }), prompt: "Hello" }) + const error = yield* LLMClient.generate(request).pipe(Effect.provide(stalledResponse), Effect.flip) + + expect(error).toBeInstanceOf(AIError) + expect(error.reason).toMatchObject({ + _tag: "Transport", + operation: "read", + phase: "receive", + code: "chunk-timeout", + }) + expect(error.message).toContain("chunkTimeout") + }), + ) + + test("mergeHttpOptions keeps the rightmost configured chunkTimeout", () => { + const merged = mergeHttpOptions(new HttpOptions({ chunkTimeout: 1000 }), new HttpOptions({ chunkTimeout: 250 })) + expect(merged?.chunkTimeout).toBe(250) + expect(mergeHttpOptions(new HttpOptions({ chunkTimeout: 250 }))?.chunkTimeout).toBe(250) + expect(mergeHttpOptions(new HttpOptions({ body: { a: 1 } }))?.chunkTimeout).toBeUndefined() + expect(mergeHttpOptions(new HttpOptions({ chunkTimeout: 250 }), new HttpOptions())?.chunkTimeout).toBe(250) + }) +}) diff --git a/services/www/src/docs/content/providers.mdx b/services/www/src/docs/content/providers.mdx index 52fd0af18260..beced324b4c5 100644 --- a/services/www/src/docs/content/providers.mdx +++ b/services/www/src/docs/content/providers.mdx @@ -61,7 +61,7 @@ The `providers` object is keyed by the provider ID used in model references, suc | `env` | Ordered environment variable names that can provide the credential. | | `package` | Runtime provider package. | | `canonical` | Built-in provider ID whose catalog defaults this provider inherits. | -| `settings` | JSON options passed to the runtime package. | +| `settings` | JSON options passed to the runtime package, such as `baseURL` and `chunkTimeout`. | | `headers` | String-valued HTTP headers added to requests. | | `body` | JSON fields merged into request bodies. | | `models` | Models to add or override, keyed by the OpenCode model ID. | @@ -86,7 +86,9 @@ models, and connection still apply. } ``` -`settings` is package-specific. A field only has an effect when the selected package supports it. +`settings` is package-specific. A field only has an effect when the selected package supports it. Packages that +support `chunkTimeout` abort a stream when no data chunk arrives within the given number of milliseconds, guarding +against a stalled provider. ## Requests