fix(ai): report mid-stream connection loss instead of a decode error (#52634)

packages/ai/src/route/executor.ts

@@ -159,6 +159,15 @@ const nativeTransportFailure = (error: unknown) => {
   return failure
 }
 
+// HTTP hooks re-wrap response bodies, so a read failure can arrive as an HttpClientError caused by
+// another HttpClientError. The innermost cause carries the native failure.
+const rootCause = (error: unknown): unknown =>
+  HttpClientError.isHttpClientError(error) && "cause" in error.reason && error.reason.cause !== undefined
+    ? rootCause(error.reason.cause)
+    : error
+
+const CONNECTION_LOST = "Connection lost while reading the response"
+
 const httpError = (input: {
   readonly error: unknown
   readonly request: HttpClientRequest.HttpClientRequest
@@ -179,20 +188,21 @@ const httpError = (input: {
       }),
     })
 
-  const source =
-    HttpClientError.isHttpClientError(input.error) && "cause" in input.error.reason
-      ? (input.error.reason.cause ?? input.error)
-      : input.error
+  const source = rootCause(input.error)
   const native = nativeTransportFailure(source)
   const code = native?.code
-  const raw = native?.message ?? (input.error instanceof Error ? input.error.message : undefined)
-  const detail = raw
-  const message = code && detail && !detail.includes(code) ? `${code}: ${detail}` : detail
+  const detail =
+    code && native?.message && !native.message.includes(code) ? `${code}: ${native.message}` : native?.message
+  const message = detail ?? (input.error instanceof Error ? input.error.message : undefined)
 
   if (Cause.isTimeoutError(input.error) || Cause.isTimeoutError(source))
     return transportError({ message: message ?? "HTTP transport timed out", code: code ?? "Timeout" })
   if (!HttpClientError.isHttpClientError(input.error))
     return transportError({ message: message ?? "HTTP transport failed", code })
+  // Effect reports every response body read failure as a DecodeError, but the raw byte stream decodes
+  // nothing: provider output parsing happens later and fails as InvalidProviderOutput.
+  if (input.operation === "read" && input.error.reason._tag === "DecodeError")
+    return transportError({ message: detail ? `${CONNECTION_LOST}: ${detail}` : CONNECTION_LOST, code })
   if (input.error.reason._tag === "TransportError") {
     return transportError({
       message: message ?? input.error.reason.description ?? "HTTP transport failed",

packages/ai/test/executor.test.ts

@@ -1,11 +1,17 @@
 import { describe, expect } from "bun:test"
 import { Deferred, Effect, Fiber, Layer, Ref, Stream } from "effect"
-import { Headers, HttpClientError, HttpClientRequest } from "effect/unstable/http"
-import { LLM, AIError, HttpContext, InvalidProviderOutputError, TransportError } from "../src/index.js"
-import { LLMClient, RequestExecutor, WebSocketTransport, type WebSocketChannelExecutor } from "../src/route.js"
+import { Headers, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
+import { LLM, AIError, HttpContext, InvalidProviderOutputError, TransportError, isRetryable } from "../src/index.js"
+import {
+  LLMClient,
+  RequestExecutor,
+  WebSocketTransport,
+  type HttpMiddleware,
+  type WebSocketChannelExecutor,
+} from "../src/route.js"
 import { route } from "../src/protocols/openai-chat.js"
 import { configure } from "../src/providers/openai.js"
-import { dynamicResponse, fixedResponse, handlerLayer, systemError } from "./lib/http.js"
+import { dynamicResponse, fixedResponse, handlerLayer, systemError, truncatedStream } from "./lib/http.js"
 import { deltaChunk } from "./lib/openai-chunks.js"
 import { sseEvents, sseRaw } from "./lib/sse.js"
 import { it } from "./lib/effect.js"
@@ -18,6 +24,8 @@ const secretRequest = HttpClientRequest.post("https://provider.test/v1/chat?api_
   HttpClientRequest.setHeaders(Headers.fromInput({ authorization: "Bearer header-secret-456" })),
 )
 
+const sseRequest = HttpClientRequest.post("https://provider.test/v1/messages")
+
 const expectAIError = (error: unknown) => {
   expect(error).toBeInstanceOf(AIError)
   if (!(error instanceof AIError)) throw new Error("expected AIError")
@@ -93,7 +101,9 @@ describe("RequestExecutor", () => {
       const error = yield* RequestExecutor.stream(executor, secretRequest).pipe(Stream.runDrain, Effect.flip)
 
       expectAIError(error)
-      expect(error.message).toBe("ECONNRESET: disconnected query-secret-123 header-secret-456")
+      expect(error.message).toBe(
+        "Connection lost while reading the response: ECONNRESET: disconnected query-secret-123 header-secret-456",
+      )
       expect(error.reason.http).toMatchObject({ status: 200, url: secretRequest.url })
       expect(error.reason.cause).toMatchObject({ code: "ECONNRESET" })
       expect(error.reason).toMatchObject({
@@ -123,7 +133,7 @@ describe("RequestExecutor", () => {
       const error = yield* RequestExecutor.stream(executor, secretRequest).pipe(Stream.runDrain, Effect.flip)
 
       expectAIError(error)
-      expect(error.message).toBe("ECONNRESET: socket closed")
+      expect(error.message).toBe("Connection lost while reading the response: ECONNRESET: socket closed")
       expect(error.reason.cause).toBeInstanceOf(TypeError)
       expect(error.reason).toMatchObject({
         _tag: "Transport",
@@ -144,6 +154,54 @@ describe("RequestExecutor", () => {
     ),
   )
 
+  it.effect("reports a connection lost mid-stream through middleware that re-wraps the body", () =>
+    Effect.gen(function* () {
+      const executor = yield* RequestExecutor.Service
+      const chunks: Array<Uint8Array> = []
+      // Session HTTP hooks hand plugins a web Response, so the body is re-wrapped around the original stream.
+      const rewrap: HttpMiddleware = (input, handler) =>
+        Effect.gen(function* () {
+          const response = yield* handler(input)
+          const body = yield* Stream.toReadableStreamEffect(response.stream)
+          return HttpClientResponse.fromWeb(
+            input,
+            new Response(body, { status: response.status, headers: response.headers }),
+          )
+        })
+      const error = yield* RequestExecutor.stream(executor, sseRequest, rewrap).pipe(
+        Stream.runForEach((chunk) => Effect.sync(() => chunks.push(chunk))),
+        Effect.flip,
+      )
+
+      expectAIError(error)
+      expect(new TextDecoder().decode(chunks[0])).toBe('data: {"type":"ping"}\n\n')
+      expect(error.message).toBe("Connection lost while reading the response: ECONNRESET: other side closed")
+      expect(error.reason.cause).toBeInstanceOf(TypeError)
+      expect(error.reason.http).toMatchObject({ status: 200 })
+      expect(error.reason).toMatchObject({ _tag: "Transport", operation: "read", code: "ECONNRESET" })
+      expect(isRetryable(error)).toBeTrue()
+    }).pipe(
+      Effect.provide(
+        truncatedStream(
+          ['data: {"type":"ping"}\n\n'],
+          new TypeError("terminated", { cause: systemError("ECONNRESET", "other side closed") }),
+        ),
+      ),
+    ),
+  )
+
+  it.effect("does not report a body read failure without a native cause as a decode error", () =>
+    Effect.gen(function* () {
+      const executor = yield* RequestExecutor.Service
+      const error = yield* RequestExecutor.stream(executor, sseRequest).pipe(Stream.runDrain, Effect.flip)
+
+      expectAIError(error)
+      expect(error.message).toBe("Connection lost while reading the response")
+      expect(error.reason).toMatchObject({ _tag: "Transport", operation: "read", code: undefined })
+      expect(isRetryable(error)).toBeTrue()
+    }).pipe(Effect.provide(truncatedStream(['data: {"type":"ping"}\n\n'], new DOMException("aborted", "AbortError")))),
+  )
+
   it.effect("preserves middleware error messages", () =>
     Effect.gen(function* () {
       const executor = yield* RequestExecutor.Service

packages/ai/test/provider/openai-chat.test.ts

@@ -2205,7 +2205,7 @@ describe("OpenAI Chat route", () => {
       )
 
       expect((yield* Ref.get(events)).some((event) => event.type === "text-delta")).toBeTrue()
-      expect(error.message).toBe("ECONNRESET: socket closed unexpectedly")
+      expect(error.message).toBe("Connection lost while reading the response: ECONNRESET: socket closed unexpectedly")
       expect(error.reason).toMatchObject({
         _tag: "Transport",
         transport: "http",
@@ -2223,7 +2223,7 @@ describe("OpenAI Chat route", () => {
         Effect.flip,
       )
 
-      expect(error.message).toBe("ECONNRESET: socket closed before output")
+      expect(error.message).toBe("Connection lost while reading the response: ECONNRESET: socket closed before output")
       expect(error.reason).toMatchObject({
         _tag: "Transport",
         transport: "http",