diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index af6ed3e4555..c125f07bcbd 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -60,6 +60,13 @@ const isDecline = ( ): error is Permission.DeclinedError | QuestionTool.CancelledError => error._tag === "Permission.DeclinedError" || error._tag === "QuestionTool.CancelledError" +const isInterruptedStream = (failure: AIError) => { + if (failure.reason._tag === "InvalidProviderOutput") + return failure.reason.classification === "incomplete-stream" + if (failure.reason._tag === "Transport") return failure.reason.operation === "read" + return false +} + /** * Classifies how the owned tool fibers ended. Interrupts abort the step; a user decline * settles its own call and then aborts the step; a defect from a tool implementation @@ -571,11 +578,14 @@ const layer = Layer.effect( }) } - const incompleteStream = - llmFailure?.reason._tag === "InvalidProviderOutput" && - llmFailure.reason.classification === "incomplete-stream" - const toolsAllowContinuation = tools.declines.length === 0 && !tools.interrupted - if (llmError && incompleteStream && record.outputStarted && toolsAllowContinuation) + if ( + llmFailure && + llmError && + isInterruptedStream(llmFailure) && + record.outputStarted && + tools.declines.length === 0 && + !tools.interrupted + ) return CallOutcome.Continue({ cause: llmFailure, error: llmError, diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index a836edaa448..3d899c3be11 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -561,6 +561,17 @@ const providerUnavailable = () => }), }) +const streamDisconnected = () => + new AIError({ + module: "test", + method: "stream", + reason: new TransportReason({ + message: "The socket connection was closed unexpectedly", + transport: "http", + operation: "read", + }), + }) + const continuationRejected = (recovery: "retry-full" | "rotate-and-retry-full") => new AIError({ module: "test", @@ -4624,6 +4635,54 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("continues after a transport read failure with durable reasoning state", () => + Effect.gen(function* () { + const session = yield* setup + yield* admit(session, "Recover disconnected reasoning") + yield* TestLLM.push( + TestLLM.failAfter( + streamDisconnected(), + LLMEvent.stepStart({ index: 0 }), + LLMEvent.reasoningStart({ + id: "disconnected-reasoning", + providerMetadata: { + openai: { itemId: "rs_disconnected", reasoningEncryptedContent: "encrypted-state" }, + }, + }), + ), + ) + yield* TestLLM.push(TestLLM.text("Recovered", "reasoning-transport-recovery")) + + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* TestLLM.wait(1) + yield* TestClock.adjust("2400 millis") + yield* Fiber.join(run) + + expect(requests).toHaveLength(2) + expect(yield* recordedEventTypes(sessionID)).toContain("session.retry.scheduled.1") + expect(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* session.context(sessionID)).toMatchObject([ + { type: "user" }, + { + type: "assistant", + finish: "error", + content: [ + { + type: "reasoning", + text: "", + state: { itemId: "rs_disconnected", reasoningEncryptedContent: "encrypted-state" }, + }, + ], + }, + { type: "synthetic", text: INCOMPLETE_STREAM_CONTINUATION }, + { type: "assistant", finish: "stop", content: [{ type: "text", text: "Recovered" }] }, + ]) + }), + ) + it.effect("continues an incomplete stream after settling a local tool", () => Effect.gen(function* () { const session = yield* setup