diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 0200b212b21e..62a95060258b 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -40,15 +40,28 @@ function wrapSSE(res: Response, ms: number, ctl: AbortController) { if (!res.headers.get("content-type")?.includes("text/event-stream")) return res const reader = res.body.getReader() + const decoder = new TextDecoder() + let buffer = "" + let deadline: number | undefined + + function observedDataEvent(value: Uint8Array) { + buffer += decoder.decode(value, { stream: true }) + const events = buffer.split(/\r?\n\r?\n/) + buffer = events.pop() ?? "" + return events.some((event) => /^data\s*:/m.test(event)) + } + const body = new ReadableStream({ async pull(ctrl) { + if (deadline === undefined) deadline = Date.now() + ms const part = await new Promise>>((resolve, reject) => { + const remaining = Math.max(0, deadline - Date.now()) const id = setTimeout(() => { const err = new ProviderError.ResponseStreamError("SSE read timed out") ctl.abort(err) void reader.cancel(err) reject(err) - }, ms) + }, remaining) reader.read().then( (part) => { @@ -67,6 +80,7 @@ function wrapSSE(res: Response, ms: number, ctl: AbortController) { return } + if (observedDataEvent(part.value)) deadline = Date.now() + ms ctrl.enqueue(part.value) }, async cancel(reason) { diff --git a/packages/opencode/test/provider/header-timeout.test.ts b/packages/opencode/test/provider/header-timeout.test.ts index fc5ab04e108b..6dda537c7750 100644 --- a/packages/opencode/test/provider/header-timeout.test.ts +++ b/packages/opencode/test/provider/header-timeout.test.ts @@ -80,6 +80,36 @@ it.live("chunkTimeout raises a response stream error when SSE body stalls", () = }), ) +it.live("chunkTimeout ignores SSE comment heartbeats", () => + Effect.gen(function* () { + const server = yield* Effect.acquireRelease( + Effect.promise(() => keepaliveBodyServer(20)), + (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 () => { + for await (const part of result.fullStream) { + if (part.type === "error") return part.error + } + }) + expect(error).toBeInstanceOf(ProviderError.ResponseStreamError) + }), + { config: providerConfig(server.url, { chunkTimeout: 50 }) }, + ) + }), +) + it.live("headerTimeout aborts when response headers do not arrive", () => Effect.gen(function* () { const server = yield* Effect.acquireRelease( @@ -211,6 +241,19 @@ async function delayedBodyServer(delay: number): Promise<{ server: Server; url: return { server, url: `http://127.0.0.1:${address.port}` } } +async function keepaliveBodyServer(interval: number): Promise<{ server: Server; url: string }> { + const server = createServer((_, res) => { + res.writeHead(200, { "content-type": "text/event-stream" }) + res.write('data: {"choices":[{"delta":{"content":"partial"}}]}\n\n') + const keepalive = setInterval(() => res.write(": keepalive\n\n"), interval) + res.on("close", () => clearInterval(keepalive)) + }) + 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}` } +} + function withAuthContent(self: Effect.Effect, value: Record = defaultAuthContent()) { return Effect.acquireUseRelease( Effect.sync(() => {