diff --git a/packages/ai/src/route/client.ts b/packages/ai/src/route/client.ts index f4873f347ef7..89000c15dbdf 100644 --- a/packages/ai/src/route/client.ts +++ b/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 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) => Stream.suspend(() => { let terminal = false @@ -247,7 +259,7 @@ const requireTerminalEvent = (route: string) => (events: Stream.Stream 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 { diff --git a/packages/ai/src/schema/errors.ts b/packages/ai/src/schema/errors.ts index a7ca7f75ec1e..4c239a9966a8 100644 --- a/packages/ai/src/schema/errors.ts +++ b/packages/ai/src/schema/errors.ts @@ -105,6 +105,7 @@ export class InvalidProviderOutputReason extends Schema.Class { 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.") }), ) diff --git a/packages/ai/test/provider/anthropic-messages.test.ts b/packages/ai/test/provider/anthropic-messages.test.ts index 5765d2b1b9e3..3e5d39937797 100644 --- a/packages/ai/test/provider/anthropic-messages.test.ts +++ b/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.", }) }), ) diff --git a/packages/ai/test/provider/openai-chat.test.ts b/packages/ai/test/provider/openai-chat.test.ts index 340abf3f5d81..f9f226f75b38 100644 --- a/packages/ai/test/provider/openai-chat.test.ts +++ b/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.") }), ) diff --git a/packages/cli/src/run/noninteractive.ts b/packages/cli/src/run/noninteractive.ts index 7d18fb7af067..83c479db16f6 100644 --- a/packages/cli/src/run/noninteractive.ts +++ b/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 diff --git a/packages/cli/test/run/noninteractive.test.ts b/packages/cli/test/run/noninteractive.test.ts index 2fabd599c279..24991fd42970 100644 --- a/packages/cli/test/run/noninteractive.test.ts +++ b/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."), ], }) diff --git a/packages/core/src/session/runner/retry.ts b/packages/core/src/session/runner/retry.ts index ff09fbc67698..21dcf9bc62d6 100644 --- a/packages/core/src/session/runner/retry.ts +++ b/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": diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index eb83723d5fc8..2567aa34c1b7 100644 --- a/packages/core/test/session-runner.test.ts +++ b/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" }], }, ])