diff --git a/packages/ai/src/route/transport/http.ts b/packages/ai/src/route/transport/http.ts index c87f0d75c2d5..771df58b51a5 100644 --- a/packages/ai/src/route/transport/http.ts +++ b/packages/ai/src/route/transport/http.ts @@ -119,20 +119,27 @@ export const httpJson =
(input: HttpJsonInput): HttpJs ) const http = RequestExecutor.responseHttp(response) const remaining = Duration.subtract(total, Duration.millis((yield* Clock.currentTimeMillis) - started)) + // SSE comment heartbeats (`: keepalive`) arrive as bytes but produce no frame, so for + // SSE the stall limit rides the framed stream — only real events re-arm it. Every other + // framing keeps byte-level progress. + const sse = prepared.framing.id === "sse" + const stall = Stream.timeoutOrElse({ + duration: timeoutDuration(request.http?.chunkTimeout), + orElse: () => Stream.fail(timeout("read", "Timed out waiting for response data", http)), + }) return { - frames: prepared.framing.frame( - RequestExecutor.responseStream(response).pipe( - Stream.timeoutOrElse({ - duration: timeoutDuration(request.http?.chunkTimeout), - orElse: () => Stream.fail(timeout("read", "Timed out waiting for response data", http)), - }), - Stream.interruptWhen( - Effect.sleep(remaining).pipe( - Effect.andThen(Effect.fail(timeout("read", "Timed out waiting for the response to complete", http))), + frames: prepared.framing + .frame( + RequestExecutor.responseStream(response).pipe( + sse ? (stream) => stream : stall, + Stream.interruptWhen( + Effect.sleep(remaining).pipe( + Effect.andThen(Effect.fail(timeout("read", "Timed out waiting for the response to complete", http))), + ), ), ), - ), - ), + ) + .pipe(sse ? stall : (stream) => stream), http, body: prepared.framing.body, } diff --git a/packages/ai/test/http-timeout.test.ts b/packages/ai/test/http-timeout.test.ts index 6b45a8070079..03c70796d159 100644 --- a/packages/ai/test/http-timeout.test.ts +++ b/packages/ai/test/http-timeout.test.ts @@ -87,6 +87,38 @@ describe("HTTP transport timeouts", () => { }), ) + // SSE comment heartbeats keep a stalled generation "warm" at the byte level without + // ever dispatching an event, so they must not extend the chunk timeout. Runs on the + // live clock so heartbeats arriving faster than the chunk timeout would keep a + // byte-level stall timer alive forever. + it.live("fails while a stalled stream only emits SSE comment heartbeats", () => + Effect.gen(function* () { + const encoder = new TextEncoder() + const layer = dynamicResponse((input) => + Effect.sync(() => + input.respond( + new ReadableStream