fix(core): continue interrupted model streams (#43757)

This commit is contained in:
Aiden Cline
2026-08-20 23:22:21 -05:00
committed by GitHub
Unverified
parent 212139ff95
commit 5c4f6ef1e3
2 changed files with 74 additions and 5 deletions
+15 -5
View File
@@ -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,
+59
View File
@@ -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