diff --git a/packages/core/src/session/runner/retry.ts b/packages/core/src/session/runner/retry.ts index ca5f1b59d04..860c97bd155 100644 --- a/packages/core/src/session/runner/retry.ts +++ b/packages/core/src/session/runner/retry.ts @@ -35,10 +35,13 @@ export function isRetryable(error: AIError) { case "RateLimit": case "ProviderInternal": return true - // HTTP transport errors carry no delivery and always retry. WebSocket marks accepted and rejected - // requests as final; not-sent and ambiguous (no frame observed) are still pre-output. + // A WebSocket acknowledgment marks delivery accepted before model output may exist. + // Read failures can still recover; the Step chooses retry versus continuation from durable output. case "Transport": - return error.reason.delivery !== "accepted" && error.reason.delivery !== "rejected" + return ( + error.reason.delivery !== "rejected" && + (error.reason.delivery !== "accepted" || error.reason.operation === "read") + ) case "InvalidProviderOutput": return error.reason.classification === "incomplete-stream" // Unrecognized failures retry: classification records affirmative diff --git a/packages/core/test/session-error.test.ts b/packages/core/test/session-error.test.ts index 2c280ba6d04..34b09cf96a8 100644 --- a/packages/core/test/session-error.test.ts +++ b/packages/core/test/session-error.test.ts @@ -192,7 +192,7 @@ describe("toSessionError", () => { expect(ineligible.map(SessionRunnerRetry.isRetryable)).toEqual([false, false, false, false, false, false, false]) }) - test("retries transport failures unless the provider accepted or rejected the request", () => { + test("retries accepted transport reads but not accepted writes or rejected requests", () => { const retryable = [ llm(new TransportError({ message: "http transport", transport: "http", operation: "request" })), llm( @@ -213,8 +213,6 @@ describe("toSessionError", () => { phase: "send", }), ), - ] - const ineligible = [ llm( new TransportError({ message: "response interrupted", @@ -224,6 +222,17 @@ describe("toSessionError", () => { phase: "receive", }), ), + ] + const ineligible = [ + llm( + new TransportError({ + message: "accepted write failed", + transport: "websocket", + operation: "write", + delivery: "accepted", + phase: "send", + }), + ), llm( new TransportError({ message: "continuation rejected", @@ -236,7 +245,7 @@ describe("toSessionError", () => { ), ] - expect(retryable.map(SessionRunnerRetry.isRetryable)).toEqual([true, true, true]) + expect(retryable.map(SessionRunnerRetry.isRetryable)).toEqual([true, true, true, true]) expect(ineligible.map(SessionRunnerRetry.isRetryable)).toEqual([false, false]) }) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 9f58161ce1f..a9acd54d273 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -622,6 +622,18 @@ const continuationRejected = (recovery: "retry-full" | "rotate-and-retry-full") }), }) +const websocketDisconnected = () => + new AIError({ + reason: new TransportError({ + message: "WebSocket closed with code 1006", + transport: "websocket", + operation: "read", + phase: "close", + code: "1006", + delivery: "accepted", + }), + }) + const incompleteStream = () => new AIError({ reason: new InvalidProviderOutputError({ @@ -5011,30 +5023,33 @@ describe("SessionRunnerLLM", () => { ]) }) - scenario("bounds jittered exponential backoff for eligible pre-output failures", function* (s) { - yield* s.admit("Retry transport") - yield* s.llm.push(Stream.fail(providerUnavailable())) - yield* s.llm.push(TestLLM.text("Recovered", "retry-success")) + for (const failure of [providerUnavailable, websocketDisconnected]) { + scenario(`bounds jittered exponential backoff before output for ${failure.name}`, function* (s) { + yield* s.admit("Retry transport") + yield* s.llm.push(TestLLM.failAfter(failure(), LLMEvent.stepStart({ index: 0 }))) + yield* s.llm.push(TestLLM.text("Recovered", "retry-success")) - const scheduled = yield* subscribeRetries(s) - const run = yield* s.resume.pipe(Effect.forkChild) - yield* Queue.take(scheduled) - yield* TestClock.adjust("1599 millis") - expect(s.requests).toHaveLength(1) - yield* TestClock.adjust("801 millis") - yield* Fiber.join(run) + const scheduled = yield* subscribeRetries(s) + const run = yield* s.resume.pipe(Effect.forkChild) + yield* Queue.take(scheduled) + yield* TestClock.adjust("1599 millis") + expect(s.requests).toHaveLength(1) + yield* TestClock.adjust("801 millis") + yield* Fiber.join(run) - expect(s.requests).toHaveLength(2) - const eventTypes = yield* recordedEventTypes(sessionID) - expect(eventTypes).toContain("session.retry.scheduled.1") - expect(eventTypes.filter((type) => type === "session.step.started.1")).toHaveLength(2) - expect(yield* s.context).toMatchObject([ - { type: "user" }, - Expected.assistant({ finish: "stop" }, [Expected.text("Recovered")]), - ]) - yield* replaySessionProjection(sessionID) - expect((yield* s.context).filter((message) => message.type === "assistant")).toHaveLength(1) - }) + expect(s.requests).toHaveLength(2) + expect(s.requests[1]?.messages).toEqual(s.requests[0]?.messages) + const eventTypes = yield* recordedEventTypes(sessionID) + expect(eventTypes).toContain("session.retry.scheduled.1") + expect(eventTypes.filter((type) => type === "session.step.started.1")).toHaveLength(2) + expect(yield* s.context).toMatchObject([ + { type: "user" }, + Expected.assistant({ finish: "stop" }, [Expected.text("Recovered")]), + ]) + yield* replaySessionProjection(sessionID) + expect((yield* s.context).filter((message) => message.type === "assistant")).toHaveLength(1) + }) + } scenario("does not start another physical attempt after interruption during retry backoff", function* (s) { yield* s.admit("Interrupt retry backoff") @@ -5338,47 +5353,49 @@ describe("SessionRunnerLLM", () => { ]) }) - scenario("continues after a transport read failure with durable reasoning state", function* (s) { - yield* s.admit("Recover disconnected reasoning") - yield* s.llm.push( - TestLLM.failAfter( - streamDisconnected(), - LLMEvent.stepStart({ index: 0 }), - LLMEvent.reasoningStart({ - id: "disconnected-reasoning", - providerMetadata: { - openai: { itemId: "rs_disconnected", reasoningEncryptedContent: "encrypted-state" }, + for (const failure of [streamDisconnected, websocketDisconnected]) { + scenario(`continues after ${failure.name} with durable reasoning state`, function* (s) { + yield* s.admit("Recover disconnected reasoning") + yield* s.llm.push( + TestLLM.failAfter( + failure(), + LLMEvent.stepStart({ index: 0 }), + LLMEvent.reasoningStart({ + id: "disconnected-reasoning", + providerMetadata: { + openai: { itemId: "rs_disconnected", reasoningEncryptedContent: "encrypted-state" }, + }, + }), + ), + ) + yield* s.llm.push(TestLLM.text("Recovered", "reasoning-transport-recovery")) + + const scheduled = yield* subscribeRetries(s) + const run = yield* s.resume.pipe(Effect.forkChild) + yield* Queue.take(scheduled) + yield* TestClock.adjust("2400 millis") + yield* Fiber.join(run) + + expect(s.requests).toHaveLength(2) + expect(yield* recordedEventTypes(sessionID)).toContain("session.retry.scheduled.1") + expect(s.requests[1]?.messages.slice(-2)).toMatchObject([ + { role: "user", content: [{ type: "text", text: "Recover disconnected reasoning" }] }, + { role: "user", content: [{ type: "text", text: INCOMPLETE_STREAM_CONTINUATION }] }, + ]) + expect(yield* s.context).toMatchObject([ + { type: "user" }, + Expected.assistant({ finish: "error" }, [ + { + type: "reasoning", + text: "", + state: { itemId: "rs_disconnected", reasoningEncryptedContent: "encrypted-state" }, }, - }), - ), - ) - yield* s.llm.push(TestLLM.text("Recovered", "reasoning-transport-recovery")) - - const scheduled = yield* subscribeRetries(s) - const run = yield* s.resume.pipe(Effect.forkChild) - yield* Queue.take(scheduled) - yield* TestClock.adjust("2400 millis") - yield* Fiber.join(run) - - expect(s.requests).toHaveLength(2) - expect(yield* recordedEventTypes(sessionID)).toContain("session.retry.scheduled.1") - expect(s.requests[1]?.messages.slice(-2)).toMatchObject([ - { role: "user", content: [{ type: "text", text: "Recover disconnected reasoning" }] }, - { role: "user", content: [{ type: "text", text: INCOMPLETE_STREAM_CONTINUATION }] }, - ]) - expect(yield* s.context).toMatchObject([ - { type: "user" }, - Expected.assistant({ finish: "error" }, [ - { - type: "reasoning", - text: "", - state: { itemId: "rs_disconnected", reasoningEncryptedContent: "encrypted-state" }, - }, - ]), - { type: "synthetic", text: INCOMPLETE_STREAM_CONTINUATION }, - Expected.assistant({ finish: "stop" }, [Expected.text("Recovered")]), - ]) - }) + ]), + { type: "synthetic", text: INCOMPLETE_STREAM_CONTINUATION }, + Expected.assistant({ finish: "stop" }, [Expected.text("Recovered")]), + ]) + }) + } scenario("continues an incomplete stream after settling a local tool", function* (s) { yield* s.admit("Continue after tool")