fix: retry empty incomplete streams (#40535)

packages/ai/src/route/client.ts

@@ -20,6 +20,7 @@ import {
   LanguageModel,
   LanguageModelLimits,
   LLMEvent,
+  InvalidProviderOutputReason,
   ProviderID,
   mergeGenerationOptions,
   mergeHttpOptions,
@@ -231,6 +232,17 @@ const streamError = (route: string, message: string, cause: Cause.Cause<unknown>
   return ProviderShared.eventError(route, message, Cause.pretty(cause))
 }
 
+const incompleteStreamError = (route: string) =>
+  new AIError({
+    module: "LLMClient",
+    method: "stream",
+    reason: new InvalidProviderOutputReason({
+      classification: "incomplete-stream",
+      message: "The provider response ended unexpectedly.",
+      route,
+    }),
+  })
+
 const requireTerminalEvent = (route: string) => (events: Stream.Stream<LLMEvent, AIError>) =>
   Stream.suspend(() => {
     let terminal = false
@@ -247,7 +259,7 @@ const requireTerminalEvent = (route: string) => (events: Stream.Stream<LLMEvent,
         Effect.suspend(() =>
           terminal
             ? Effect.void
-            : Effect.fail(ProviderShared.eventError(route, "Provider stream ended without a terminal finish event")),
+            : Effect.fail(incompleteStreamError(route)),
         ),
       ),
     )
@@ -416,10 +428,7 @@ const generateWith = (stream: Interface["stream"]) =>
     const state = yield* stream(request, options).pipe(Stream.runFold(LLMResponse.empty, LLMResponse.reduce))
     const response = LLMResponse.complete(state)
     if (response) return response
-    return yield* ProviderShared.eventError(
-      `${request.model.provider}/${request.model.route.id}`,
-      "Provider stream ended without a terminal finish event",
-    )
+    return yield* incompleteStreamError(`${request.model.provider}/${request.model.route.id}`)
   })
 
 export function stream(request: LLMRequest, options?: StreamOptions): Stream.Stream<LLMEvent, AIError, Service> {

packages/ai/src/schema/errors.ts

@@ -105,6 +105,7 @@ export class InvalidProviderOutputReason extends Schema.Class<InvalidProviderOut
 )({
   _tag: Schema.tag("InvalidProviderOutput"),
   message: Schema.String,
+  classification: Schema.optional(Schema.Literals(["incomplete-stream"])),
   route: Schema.optional(Schema.String),
   raw: Schema.optional(Schema.String),
   providerMetadata: Schema.optional(ProviderMetadata),

packages/ai/test/adapter.test.ts

@@ -133,8 +133,8 @@ describe("llm route", () => {
     Effect.gen(function* () {
       const error = yield* (yield* LLMClient.Service).stream(request).pipe(Stream.runDrain, Effect.flip)
 
-      expect(error.reason).toMatchObject({ _tag: "InvalidProviderOutput" })
-      expect(error.message).toContain("Provider stream ended without a terminal finish event")
+      expect(error.reason).toMatchObject({ _tag: "InvalidProviderOutput", classification: "incomplete-stream" })
+      expect(error.message).toContain("The provider response ended unexpectedly.")
     }),
   )
 

packages/ai/test/provider/anthropic-messages.test.ts

@@ -538,7 +538,8 @@ describe("Anthropic Messages route", () => {
 
       expect(error.reason).toMatchObject({
         _tag: "InvalidProviderOutput",
-        message: "Provider stream ended without a terminal finish event",
+        classification: "incomplete-stream",
+        message: "The provider response ended unexpectedly.",
       })
     }),
   )

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

@@ -1136,9 +1136,12 @@ describe("OpenAI Chat route", () => {
         { type: "tool-input-delta", id: "call_1", name: "lookup", text: ':"weather"}' },
       ])
       expect(events.filter(LLMEvent.is.toolCall)).toEqual([])
-      expect(streamError.reason).toMatchObject({ _tag: "InvalidProviderOutput" })
-      expect(streamError.message).toContain("Provider stream ended without a terminal finish event")
-      expect(error.message).toContain("Provider stream ended without a terminal finish event")
+      expect(streamError.reason).toMatchObject({
+        _tag: "InvalidProviderOutput",
+        classification: "incomplete-stream",
+      })
+      expect(streamError.message).toContain("The provider response ended unexpectedly.")
+      expect(error.message).toContain("The provider response ended unexpectedly.")
     }),
   )
 

packages/cli/src/run/noninteractive.ts

@@ -470,7 +470,7 @@ export async function runNonInteractivePrompt(input: Input) {
       if (event.type === "session.step.failed") {
         if (
           input.compatibility === "v1" &&
-          event.data.error.message === "Provider stream ended without a terminal finish event"
+          event.data.error.message === "The provider response ended unexpectedly."
         ) {
           pendingStep = undefined
           v1InvalidOutput = true

packages/cli/test/run/noninteractive.test.ts

@@ -503,8 +503,8 @@ describe("runNonInteractivePrompt", () => {
       turn: (messageID) => [
         prompted(messageID),
         stepStarted(),
-        stepFailed("Provider stream ended without a terminal finish event"),
-        executionFailed("Provider stream ended without a terminal finish event"),
+        stepFailed("The provider response ended unexpectedly."),
+        executionFailed("The provider response ended unexpectedly."),
       ],
     })
 

packages/core/src/session/runner/retry.ts

@@ -20,10 +20,11 @@ export function isRetryable(error: AIError) {
     case "ProviderInternal":
     case "Transport":
       return true
+    case "InvalidProviderOutput":
+      return error.reason.classification === "incomplete-stream"
     case "Authentication":
     case "QuotaExceeded":
     case "ContentPolicy":
-    case "InvalidProviderOutput":
     case "InvalidRequest":
     case "NoRoute":
     case "UnknownProvider":

packages/core/test/session-runner.test.ts

@@ -513,6 +513,16 @@ const providerUnavailable = () =>
     reason: new TransportReason({ message: "Provider unavailable" }),
   })
 
+const incompleteStream = () =>
+  new AIError({
+    module: "test",
+    method: "stream",
+    reason: new InvalidProviderOutputReason({
+      classification: "incomplete-stream",
+      message: "The provider response ended unexpectedly.",
+    }),
+  })
+
 const invalidRequest = () =>
   new AIError({
     module: "test",
@@ -3949,6 +3959,26 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("retries an incomplete stream before output", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      yield* admit(session, "Retry incomplete stream")
+      yield* TestLLM.push(Stream.fail(incompleteStream()))
+      yield* TestLLM.push(TestLLM.text("Recovered", "incomplete-stream-success"))
+
+      const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
+      yield* TestLLM.wait(1)
+      yield* TestClock.adjust("2 seconds")
+      yield* Fiber.join(run)
+
+      expect(requests).toHaveLength(2)
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user" },
+        { type: "assistant", finish: "stop", content: [{ type: "text", text: "Recovered" }] },
+      ])
+    }),
+  )
+
   it.effect("uses a larger provider retry-after delay", () =>
     Effect.gen(function* () {
       const session = yield* setup
@@ -3969,7 +3999,7 @@ describe("SessionRunnerLLM", () => {
   it.effect("does not retry eligible failures after observable output", () =>
     Effect.gen(function* () {
       const session = yield* setup
-      const failure = rateLimited()
+      const failure = incompleteStream()
       yield* TestLLM.push(
         TestLLM.failAfter(
           failure,
@@ -3987,7 +4017,7 @@ describe("SessionRunnerLLM", () => {
         {
           type: "assistant",
           finish: "error",
-          error: { type: "provider.rate-limit" },
+          error: { type: "provider.invalid-output" },
           content: [{ type: "text", text: "Partial" }],
         },
       ])