mirrors/opencode
1
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-11-13 19:46:16 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(core): retry acknowledged websocket read failures (#49183)

Browse Source
This commit is contained in:
Daniel Polito
2026-09-16 07:45:19 -03:00
committed by GitHub
Unverified
parent eb052f70b4
commit ba753b383f
3 changed files with 98 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/session/runner/retry.ts
+6 -3
View File
@@ -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
packages/core/test/session-error.test.ts
+13 -4
View File
@@ -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])
})
packages/core/test/session-runner.test.ts
+79 -62
View File
@@ -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")