From e22295a175b74090549115e0eb02b64af6aefb3e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Filip=20Kaln=C3=BD?= Date: Sun, 6 Sep 2026 13:21:11 +0000 Subject: [PATCH] fix(provider): enforce chunk timeout regardless of content type --- packages/opencode/src/provider/provider.ts | 5 +- packages/opencode/test/lib/llm-server.ts | 8 +- .../test/provider/header-timeout.test.ts | 738 +++++++++++++++++- .../test/session/processor-effect.test.ts | 67 +- 4 files changed, 789 insertions(+), 29 deletions(-) diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 72d5a7a59382..4eb0422b2778 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({ @@ -1821,7 +1820,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/lib/llm-server.ts b/packages/opencode/test/lib/llm-server.ts index 245acc7280f5..0c74f0805aed 100644 --- a/packages/opencode/test/lib/llm-server.ts +++ b/packages/opencode/test/lib/llm-server.ts @@ -36,6 +36,7 @@ type Sse = { type: "sse" head: unknown[] tail: unknown[] + contentType?: string | false wait?: PromiseLike hang?: boolean error?: unknown @@ -427,7 +428,10 @@ function send(item: Sse) { if (item.error) end = Stream.concat(empty, Stream.fail(item.error)) else if (item.hang) end = Stream.concat(empty, Stream.never) - return HttpServerResponse.stream(Stream.concat(body, end), { contentType: "text/event-stream" }) + return HttpServerResponse.stream( + Stream.concat(body, end), + item.contentType === false ? undefined : { contentType: item.contentType ?? "text/event-stream" }, + ) } const reset = Effect.fn("TestLLMServer.reset")(function* (item: Sse) { @@ -577,6 +581,7 @@ export function raw(input: { chunks?: unknown[] head?: unknown[] tail?: unknown[] + contentType?: string | false wait?: PromiseLike hang?: boolean error?: unknown @@ -586,6 +591,7 @@ export function raw(input: { type: "sse", head: input.head ?? input.chunks ?? [], tail: input.tail ?? [], + contentType: input.contentType, wait: input.wait, hang: input.hang, error: input.error, diff --git a/packages/opencode/test/provider/header-timeout.test.ts b/packages/opencode/test/provider/header-timeout.test.ts index 38b884dc7fc1..284bbb506e11 100644 --- a/packages/opencode/test/provider/header-timeout.test.ts +++ b/packages/opencode/test/provider/header-timeout.test.ts @@ -1,6 +1,6 @@ import { afterEach, expect } from "bun:test" import { createServer, type Server } from "node:http" -import { streamText } from "ai" +import { APICallError, streamText } from "ai" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" import { Effect } from "effect" @@ -48,6 +48,522 @@ it.live("headerTimeout does not abort delayed SSE body after headers arrive", () }), ) +it.live("chunkTimeout applies to a headerless stream before its first event", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => localBodyServer({ end: false })), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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 abort = yield* Effect.acquireRelease( + Effect.sync(() => new AbortController()), + (controller) => Effect.sync(() => controller.abort()), + ) + const result = streamText({ + model: yield* provider.getLanguage(model), + abortSignal: abort.signal, + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + + const error = yield* Effect.promise(() => readStreamError(result, abort)) + expect(error).toBeInstanceOf(ProviderError.ResponseStreamError) + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: providerConfig(server.url, { chunkTimeout: 50 }) }, + ) + }), +) + +it.live("chunkTimeout applies after reasoning begins without a content type", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + chunks: [{ delay: 0, body: event({ choices: [{ delta: { reasoning_content: "thinking" } }] }) }], + end: false, + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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 abort = yield* Effect.acquireRelease( + Effect.sync(() => new AbortController()), + (controller) => Effect.sync(() => controller.abort()), + ) + let sawReasoning = false + const result = streamText({ + model: yield* provider.getLanguage(model), + abortSignal: abort.signal, + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + + const error = yield* Effect.promise(async () => { + const timeout = setTimeout(() => abort.abort(new Error("bounded test timeout")), 2_000) + try { + for await (const part of result.fullStream) { + if (part.type === "reasoning-delta") sawReasoning = true + if (part.type === "error") return part.error + } + return new Error("stream completed without an error") + } catch (error) { + return error + } finally { + clearTimeout(timeout) + } + }) + + expect(sawReasoning).toBe(true) + expect(error).toBeInstanceOf(ProviderError.ResponseStreamError) + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: providerConfig(server.url, { chunkTimeout: 50 }) }, + ) + }), +) + +for (const contentType of ["application/json", "not-a-mime-type"]) { + it.live(`chunkTimeout applies to an SSE body with a misleading content type (${contentType})`, () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + contentType, + chunks: [{ delay: 0, body: event({ choices: [{ delta: { content: "partial" } }] }) }], + end: false, + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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 abort = yield* Effect.acquireRelease( + Effect.sync(() => new AbortController()), + (controller) => Effect.sync(() => controller.abort()), + ) + let sawText = false + const result = streamText({ + model: yield* provider.getLanguage(model), + abortSignal: abort.signal, + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + + const error = yield* Effect.promise(async () => { + const timeout = setTimeout(() => abort.abort(new Error("bounded test timeout")), 2_000) + try { + for await (const part of result.fullStream) { + if (part.type === "text-delta") { + expect(part.text).toBe("partial") + sawText = true + } + if (part.type === "error") { + expect(sawText).toBe(true) + return part.error + } + } + return new Error("stream completed without an error") + } catch (error) { + expect(sawText).toBe(true) + return error + } finally { + clearTimeout(timeout) + } + }) + + expect(sawText).toBe(true) + expect(error).toBeInstanceOf(ProviderError.ResponseStreamError) + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: providerConfig(server.url, { chunkTimeout: 50 }) }, + ) + }), + ) +} + +it.live("body progress resets chunkTimeout across multiple timeout intervals", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + contentType: "text/event-stream", + chunks: [ + { delay: 0, body: event({ choices: [{ delta: { role: "assistant" } }] }) }, + { delay: 200, body: event({ choices: [{ delta: { content: "a" } }] }) }, + { delay: 200, body: event({ choices: [{ delta: { content: "b" } }] }) }, + { delay: 200, body: "data: [DONE]\n\n" }, + ], + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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), + messages: [{ role: "user", content: "hello" }], + }) + + const text: string[] = [] + yield* Effect.promise(async () => { + for await (const part of result.fullStream) { + if (part.type === "text-delta") text.push(part.text) + if (part.type === "error") throw part.error + } + }) + expect(text.join("")).toBe("ab") + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: providerConfig(server.url, { chunkTimeout: 500 }) }, + ) + }), +) + +it.live("Bedrock binary EventStream progress survives multiple timeout intervals", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + contentType: "application/vnd.amazon.eventstream", + chunks: [ + { + delay: 0, + body: bedrockEvent("contentBlockDelta", { + contentBlockIndex: 0, + delta: { text: "a" }, + }), + }, + { + delay: 200, + body: bedrockEvent("contentBlockDelta", { + contentBlockIndex: 0, + delta: { text: "b" }, + }), + }, + { delay: 200, body: bedrockEvent("messageStop", { stopReason: "end_turn" }) }, + { + delay: 200, + body: bedrockEvent("metadata", { + usage: { inputTokens: 1, outputTokens: 2, totalTokens: 3 }, + }), + }, + ], + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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), + messages: [{ role: "user", content: "hello" }], + }) + + expect(yield* Effect.promise(() => result.text)).toBe("ab") + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: bedrockProviderConfig(server.url, { chunkTimeout: 500 }) }, + ) + }), +) + +it.live("chunkTimeout applies to a stalled binary EventStream body", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + contentType: "application/vnd.amazon.eventstream", + chunks: [ + { + delay: 0, + body: bedrockEvent("contentBlockDelta", { + contentBlockIndex: 0, + delta: { text: "partial" }, + }), + }, + ], + end: false, + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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 abort = yield* Effect.acquireRelease( + Effect.sync(() => new AbortController()), + (controller) => Effect.sync(() => controller.abort()), + ) + let sawText = false + const result = streamText({ + model: yield* provider.getLanguage(model), + abortSignal: abort.signal, + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + const error = yield* Effect.promise(async () => { + const timeout = setTimeout(() => abort.abort(new Error("bounded test timeout")), 2_000) + try { + for await (const part of result.fullStream) { + if (part.type === "text-delta") sawText = true + if (part.type === "error") return part.error + } + return new Error("stream completed without an error") + } catch (error) { + return error + } finally { + clearTimeout(timeout) + } + }) + + expect(sawText).toBe(true) + expect(error).toBeInstanceOf(ProviderError.ResponseStreamError) + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: bedrockProviderConfig(server.url, { chunkTimeout: 50 }) }, + ) + }), +) + +it.live("timeout false does not disable chunkTimeout", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => localBodyServer({ end: false })), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + yield* provideTmpdirInstance( + () => + Effect.gen(function* () { + const provider = yield* Provider.Service + const configured = yield* provider.getProvider(ProviderV2.ID.make("test")) + expect(configured.options.timeout).toBe(false) + const model = yield* provider.getModel(ProviderV2.ID.make("test"), ModelV2.ID.make("test-model")) + const abort = yield* Effect.acquireRelease( + Effect.sync(() => new AbortController()), + (controller) => Effect.sync(() => controller.abort()), + ) + const result = streamText({ + model: yield* provider.getLanguage(model), + abortSignal: abort.signal, + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + + const error = yield* Effect.promise(() => readStreamError(result, abort)) + expect(error).toBeInstanceOf(ProviderError.ResponseStreamError) + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: providerConfig(server.url, { timeout: false, chunkTimeout: 50 }) }, + ) + }), +) + +it.live("provider SDK preserves finite JSON response content and headers", () => + Effect.gen(function* () { + const body = JSON.stringify({ + id: "chatcmpl-json", + object: "chat.completion", + choices: [{ message: { role: "assistant", content: "json" }, finish_reason: "stop" }], + }) + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + contentType: "application/json", + status: 201, + statusMessage: "Created By Test", + headers: { "x-test-response": "preserved" }, + chunks: [{ delay: 0, body }], + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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 language = yield* provider.getLanguage(model) + + const result = yield* Effect.promise(() => + language.doGenerate({ prompt: [{ role: "user", content: [{ type: "text", text: "hello" }] }] }), + ) + expect(result.content).toEqual([{ type: "text", text: "json" }]) + expect(result.response?.headers?.["x-test-response"]).toBe("preserved") + expect(result.response?.body).toEqual(JSON.parse(body)) + }), + { config: providerConfig(server.url, { chunkTimeout: 50 }) }, + ) + }), +) + +it.live("chunkTimeout applies while a non-streaming JSON body is read", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + contentType: "application/json", + chunks: [{ delay: 0, body: '{"choices":[' }], + end: false, + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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 language = yield* provider.getLanguage(model) + const abort = yield* Effect.acquireRelease( + Effect.sync(() => new AbortController()), + (controller) => Effect.sync(() => controller.abort()), + ) + const error = yield* Effect.promise(async () => { + const timeout = setTimeout(() => abort.abort(new Error("bounded test timeout")), 2_000) + try { + return await language.doGenerate({ + abortSignal: abort.signal, + prompt: [{ role: "user", content: [{ type: "text", text: "hello" }] }], + }) + } catch (error) { + return error + } finally { + clearTimeout(timeout) + } + }) + + expect(APICallError.isInstance(error)).toBe(true) + if (APICallError.isInstance(error)) { + expect(error.cause).toBeInstanceOf(ProviderError.ResponseStreamError) + } + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: providerConfig(server.url, { chunkTimeout: 50 }) }, + ) + }), +) + +it.live("caller abort closes a stalled response after progress", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + chunks: [{ delay: 0, body: event({ choices: [{ delta: { content: "partial" } }] }) }], + end: false, + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + 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 abort = yield* Effect.acquireRelease( + Effect.sync(() => new AbortController()), + (controller) => Effect.sync(() => controller.abort()), + ) + const result = streamText({ + model: yield* provider.getLanguage(model), + abortSignal: abort.signal, + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + const consumed = (async () => { + for await (const part of result.fullStream) { + if (part.type !== "text-delta") continue + abort.abort(new Error("caller cancelled")) + return + } + })() + + expect(yield* Effect.promise(() => bounded(consumed, 2_000))).toBe(true) + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: providerConfig(server.url, { chunkTimeout: 10_000 }) }, + ) + }), +) + +it.live("consumer cancellation aborts the wrapped provider fetch", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + chunks: [{ delay: 0, body: event({ choices: [{ delta: { content: "partial" } }] }) }], + end: false, + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) + + yield* provideTmpdirInstance( + () => + Effect.gen(function* () { + const provider = yield* Provider.Service + const configured = yield* provider.getProvider(ProviderV2.ID.make("test")) + const model = yield* provider.getModel(ProviderV2.ID.make("test"), ModelV2.ID.make("test-model")) + let aborted = false + configured.options.fetch = async (input: RequestInfo | URL, init?: RequestInit) => { + init?.signal?.addEventListener("abort", () => { + aborted = true + }) + return fetch(input, init) + } + const language = yield* provider.getLanguage(model) + const result = yield* Effect.promise(() => + language.doStream({ prompt: [{ role: "user", content: [{ type: "text", text: "hello" }] }] }), + ) + const consumed = (async () => { + const reader = result.stream.getReader() + while (true) { + const part = await reader.read() + if (part.done) return + if (part.value.type !== "text-delta") continue + await reader.cancel("consumer cancelled") + return + } + })() + + expect(yield* Effect.promise(() => bounded(consumed, 2_000))).toBe(true) + expect(aborted).toBe(true) + expect(yield* Effect.promise(() => bounded(server.responseClosed, 2_000))).toBe(true) + }), + { config: providerConfig(server.url, { chunkTimeout: 10_000 }) }, + ) + }), +) + for (const timeout of ["chunkTimeout", "headerTimeout"] as const) { it.live(`default ${timeout} is applied at fetch without changing provider options`, () => Effect.gen(function* () { @@ -126,31 +642,43 @@ it.live("configured chunkTimeout raises a retryable response stream error when S }), ) -it.live("chunkTimeout can be disabled with false", () => - Effect.gen(function* () { - const server = yield* Effect.acquireRelease( - Effect.promise(() => delayedBodyServer(250)), - (server) => Effect.sync(() => server.server.close()), - ) +for (const contentType of ["text/event-stream", undefined] as const) { + it.live(`chunkTimeout can be disabled with false (${contentType ?? "absent content type"})`, () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => + localBodyServer({ + contentType, + chunks: [ + { + delay: 250, + body: `${event({ choices: [{ delta: { content: "late" } }] })}data: [DONE]\n\n`, + }, + ], + }), + ), + (server) => Effect.promise(() => closeServer(server.server)), + ) - yield* provideTmpdirInstance( - () => - Effect.gen(function* () { - const provider = yield* Provider.Service - const configured = yield* provider.getProvider(ProviderV2.ID.make("test")) - expect(configured.options.chunkTimeout).toBe(false) - const model = yield* provider.getModel(ProviderV2.ID.make("test"), ModelV2.ID.make("test-model")) - const result = streamText({ - model: yield* provider.getLanguage(model), - messages: [{ role: "user", content: "hello" }], - }) + yield* provideTmpdirInstance( + () => + Effect.gen(function* () { + const provider = yield* Provider.Service + const configured = yield* provider.getProvider(ProviderV2.ID.make("test")) + expect(configured.options.chunkTimeout).toBe(false) + const model = yield* provider.getModel(ProviderV2.ID.make("test"), ModelV2.ID.make("test-model")) + const result = streamText({ + model: yield* provider.getLanguage(model), + messages: [{ role: "user", content: "hello" }], + }) - expect(yield* Effect.promise(() => result.text)).toBe("late") - }), - { config: providerConfig(server.url, { chunkTimeout: false }) }, - ) - }), -) + expect(yield* Effect.promise(() => result.text)).toBe("late") + }), + { config: providerConfig(server.url, { chunkTimeout: false }) }, + ) + }), + ) +} it.live("headerTimeout aborts when response headers do not arrive", () => Effect.gen(function* () { @@ -257,6 +785,168 @@ function providerConfig(url: string, options: Record = {}) { } } +function bedrockProviderConfig(url: string, options: Record = {}) { + const config = testProviderConfig(url) + return { + ...config, + provider: { + test: { + ...config.provider.test, + npm: "@ai-sdk/amazon-bedrock", + options: { + ...config.provider.test.options, + apiKey: "test-key", + baseURL: url, + region: "us-east-1", + ...options, + }, + }, + }, + } +} + +type LocalBodyServer = { + server: Server + url: string + responseClosed: Promise +} + +async function localBodyServer(input: { + contentType?: string + status?: number + statusMessage?: string + headers?: Record + chunks?: { delay: number; body: string | Uint8Array }[] + end?: boolean +}): Promise { + let resolveClosed!: () => void + let closed = false + const responseClosed = new Promise((resolve) => { + resolveClosed = resolve + }) + const server = createServer((req, res) => { + const timers: ReturnType[] = [] + const markClosed = () => { + for (const timer of timers) clearTimeout(timer) + if (closed) return + closed = true + resolveClosed() + } + req.once("aborted", markClosed) + res.once("close", markClosed) + res.once("finish", markClosed) + + const headers = { ...input.headers } + if (input.contentType !== undefined) headers["content-type"] = input.contentType + if (input.statusMessage !== undefined) res.statusMessage = input.statusMessage + res.writeHead(input.status ?? 200, headers) + res.flushHeaders() + + let delay = 0 + for (const chunk of input.chunks ?? []) { + delay += chunk.delay + timers.push(setTimeout(() => res.write(chunk.body), delay)) + } + if (input.end !== false) timers.push(setTimeout(() => res.end(), delay)) + }) + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)) + const address = server.address() + if (!address || typeof address === "string") throw new Error("server did not bind to a TCP port") + return { server, url: `http://127.0.0.1:${address.port}`, responseClosed } +} + +async function closeServer(server: Server) { + if (!server.listening) return + await new Promise((resolve) => { + const timeout = setTimeout(resolve, 2_000) + server.close(() => { + clearTimeout(timeout) + resolve() + }) + server.closeIdleConnections() + server.closeAllConnections() + }) +} + +function event(input: unknown) { + return `data: ${JSON.stringify(input)}\n\n` +} + +function bedrockEvent(type: string, input: unknown) { + const encoder = new TextEncoder() + const headers = [ + eventStreamHeader(encoder, ":message-type", "event"), + eventStreamHeader(encoder, ":event-type", type), + ] + const headerLength = headers.reduce((total, header) => total + header.length, 0) + const body = encoder.encode(JSON.stringify(input)) + const totalLength = 16 + headerLength + body.length + const frame = new Uint8Array(totalLength) + const view = new DataView(frame.buffer) + view.setUint32(0, totalLength) + view.setUint32(4, headerLength) + view.setUint32(8, crc32(frame.subarray(0, 8))) + let offset = 12 + for (const header of headers) { + frame.set(header, offset) + offset += header.length + } + frame.set(body, offset) + view.setUint32(totalLength - 4, crc32(frame.subarray(0, totalLength - 4))) + return frame +} + +function eventStreamHeader(encoder: TextEncoder, name: string, value: string) { + const encodedName = encoder.encode(name) + const encodedValue = encoder.encode(value) + const header = new Uint8Array(1 + encodedName.length + 1 + 2 + encodedValue.length) + header[0] = encodedName.length + header.set(encodedName, 1) + header[1 + encodedName.length] = 7 + new DataView(header.buffer).setUint16(2 + encodedName.length, encodedValue.length) + header.set(encodedValue, 4 + encodedName.length) + return header +} + +function crc32(input: Uint8Array) { + let crc = 0xffffffff + for (const byte of input) { + crc ^= byte + for (let bit = 0; bit < 8; bit++) crc = (crc >>> 1) ^ (crc & 1 ? 0xedb88320 : 0) + } + return (crc ^ 0xffffffff) >>> 0 +} + +async function readStreamError(result: ReturnType, abort: AbortController) { + const timeout = setTimeout(() => abort.abort(new Error("bounded test timeout")), 2_000) + try { + for await (const part of result.fullStream) { + if (part.type === "error") return part.error + } + return new Error("stream completed without an error") + } catch (error) { + return error + } finally { + clearTimeout(timeout) + } +} + +function bounded(input: Promise, ms: number) { + return new Promise((resolve) => { + const timeout = setTimeout(() => resolve(false), ms) + input.then( + () => { + clearTimeout(timeout) + resolve(true) + }, + () => { + clearTimeout(timeout) + resolve(false) + }, + ) + }) +} + async function delayedHeaderServer(delay: number): Promise<{ server: Server; url: string }> { const server = createServer((_, res) => { setTimeout(() => { diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index c67f82d9c71b..1bdadd718a4d 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -70,7 +70,7 @@ const cfg = { }, } -function providerCfg(url: string) { +function providerCfg(url: string, options: Record = {}) { return { ...cfg, provider: { @@ -80,6 +80,7 @@ function providerCfg(url: string) { options: { ...cfg.provider.test.options, baseURL: url, + ...options, }, }, }, @@ -651,6 +652,70 @@ it.live("session.processor effect tests retry OpenAI-compatible midstream server ), ) +it.live("session.processor retries a headerless stalled stream to completion", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + + yield* llm.push( + raw({ + contentType: false, + head: [ + { + id: "chatcmpl-stalled", + object: "chat.completion.chunk", + choices: [{ delta: { role: "assistant" } }], + }, + { + id: "chatcmpl-stalled", + object: "chat.completion.chunk", + choices: [{ delta: { reasoning_content: "thinking" } }], + }, + ], + hang: true, + }), + reply().text("after").stop(), + ) + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "retry headerless stream") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "retry headerless stream" }], + tools: {}, + }) + + const parts = yield* MessageV2.parts(msg.id) + + expect(value).toBe("continue") + expect(yield* llm.calls).toBe(2) + expect(parts.some((part) => part.type === "text" && part.text === "after")).toBe(true) + expect(handle.message.error).toBeUndefined() + }), + { config: (url) => providerCfg(url, { chunkTimeout: 50 }) }, + ), +) + it.live("session.processor effect tests retry network_error finish reasons", () => provideTmpdirServer( ({ dir, llm }) =>