diff --git a/packages/core/src/aisdk.ts b/packages/core/src/aisdk.ts index b604dac664b6..424f41f41cef 100644 --- a/packages/core/src/aisdk.ts +++ b/packages/core/src/aisdk.ts @@ -23,10 +23,9 @@ export interface LanguageEvent { language?: LanguageModelV3 } -function wrapSSE(res: Response, ms: number, ctl: AbortController) { +function wrapStream(res: Response, ms: number, ctl: AbortController) { if (typeof ms !== "number" || ms <= 0) return res if (!res.body) return res - if (!res.headers.get("content-type")?.includes("text/event-stream")) return res const reader = res.body.getReader() const body = new ReadableStream({ @@ -115,7 +114,7 @@ function prepareOptions(model: ModelV2.Info, pkg: string) { timeout: false, }) if (!chunkAbortCtl || typeof chunkTimeout !== "number") return res - return wrapSSE(res, chunkTimeout, chunkAbortCtl) + return wrapStream(res, chunkTimeout, chunkAbortCtl) } return options diff --git a/packages/core/src/v1/config/provider.ts b/packages/core/src/v1/config/provider.ts index d54a3f08f926..39e9c13a6cd0 100644 --- a/packages/core/src/v1/config/provider.ts +++ b/packages/core/src/v1/config/provider.ts @@ -110,7 +110,7 @@ export const Info = Schema.Struct({ }), chunkTimeout: Schema.optional(PositiveInt).annotate({ description: - "Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted.", + "Timeout in milliseconds between streamed response chunks for this provider. Disabled by default.", }), }), [Schema.Record(Schema.String, Schema.Any)], diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index a0ab5eb3474d..0a6f7a691349 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -34,10 +34,9 @@ import { ProviderError } from "./error" const OPENAI_HEADER_TIMEOUT_DEFAULT = 300_000 -function wrapSSE(res: Response, ms: number, ctl: AbortController) { +function wrapStream(res: Response, ms: number, ctl: AbortController) { if (typeof ms !== "number" || ms <= 0) return res if (!res.body) return res - if (!res.headers.get("content-type")?.includes("text/event-stream")) return res const reader = res.body.getReader() const body = new ReadableStream({ @@ -1759,7 +1758,7 @@ const layer = Layer.effect( }).finally(() => headerTimeoutCtl?.clear()) if (!chunkAbortCtl) return res - return wrapSSE(res, chunkTimeout, chunkAbortCtl) + return wrapStream(res, chunkTimeout, chunkAbortCtl) } const bundledLoader = BUNDLED_PROVIDERS[model.api.npm] diff --git a/packages/opencode/test/provider/header-timeout.test.ts b/packages/opencode/test/provider/header-timeout.test.ts index fc5ab04e108b..c6b39f77f8d9 100644 --- a/packages/opencode/test/provider/header-timeout.test.ts +++ b/packages/opencode/test/provider/header-timeout.test.ts @@ -69,6 +69,42 @@ it.live("chunkTimeout raises a response stream error when SSE body stalls", () = for await (const part of result.fullStream) { if (part.type === "error") return part.error } + return undefined + } catch (error) { + return error + } + }) + expect(error).toBeInstanceOf(ProviderError.ResponseStreamError) + }), + { config: providerConfig(server.url, { chunkTimeout: 50 }) }, + ) + }), +) + +it.live("chunkTimeout applies to non-SSE response streams", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => delayedBodyServer(250, "application/vnd.amazon.eventstream")), + (server) => Effect.sync(() => server.server.close()), + ) + + yield* provideTmpdirInstance( + () => + Effect.gen(function* () { + const provider = yield* Provider.Service + const model = yield* provider.getModel(ProviderV2.ID.make("test"), ModelV2.ID.make("test-model")) + const result = streamText({ + model: yield* provider.getLanguage(model), + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + + const error = yield* Effect.promise(async () => { + try { + for await (const part of result.fullStream) { + if (part.type === "error") return part.error + } + return undefined } catch (error) { return error } @@ -197,9 +233,12 @@ async function delayedHeaderServer(delay: number): Promise<{ server: Server; url return { server, url: `http://127.0.0.1:${address.port}` } } -async function delayedBodyServer(delay: number): Promise<{ server: Server; url: string }> { +async function delayedBodyServer( + delay: number, + contentType = "text/event-stream", +): Promise<{ server: Server; url: string }> { const server = createServer((_, res) => { - res.writeHead(200, { "content-type": "text/event-stream" }) + res.writeHead(200, { "content-type": contentType }) res.flushHeaders() setTimeout(() => { res.end('data: {"choices":[{"delta":{"content":"late"}}]}\n\ndata: [DONE]\n\n') diff --git a/packages/web/src/content/docs/config.mdx b/packages/web/src/content/docs/config.mdx index 8bf875b1f85f..aa4eb77d06bc 100644 --- a/packages/web/src/content/docs/config.mdx +++ b/packages/web/src/content/docs/config.mdx @@ -386,7 +386,7 @@ Provider options can include `timeout`, `chunkTimeout`, and `setCacheKey`: ``` - `timeout` - Request timeout in milliseconds (default: 300000). Set to `false` to disable. -- `chunkTimeout` - Timeout in milliseconds between streamed response chunks. If no chunk arrives in time, the request is aborted. +- `chunkTimeout` - Timeout in milliseconds between streamed response chunks. It is disabled by default because slow models can legitimately pause for several minutes. - `setCacheKey` - Ensure a cache key is always set for designated provider. You can also configure [local models](/docs/models#local). [Learn more](/docs/models).