vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-09-07 17:08:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(core): retry transient compaction failures (#47159)

Browse source
This commit is contained in:
Aiden Cline 2026-09-03 20:45:47 -05:00 • committed by GitHub
parent c9d240704d
commit 4bf5269c4c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 250 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

168
packages/core/src/session/compaction.ts
View file

@ -1,6 +1,16 @@
export * as SessionCompaction from "./compaction.js"
import { LLMClient, LLMEvent, LLMRequest, Message, type ContentPart } from "@opencode-ai/ai"
import {
AIError,
InvalidProviderOutputError,
UnknownProviderError,
isContextOverflowFailure,
LLMClient,
LLMEvent,
LLMRequest,
Message,
type ContentPart,
} from "@opencode-ai/ai"
import { Agent } from "@opencode-ai/schema/agent"
import { SessionError } from "@opencode-ai/schema/session-error"
import { Context, Effect, Layer, Stream } from "effect"
@ -12,6 +22,7 @@ import type { SessionContext } from "./context.js"
import type { SessionMessage } from "./message.js"
import { SessionModelRequest } from "./model-request.js"
import type { SessionRunnerModel } from "./runner/model.js"
import { SessionRunnerRetry } from "./runner/retry.js"
import { SessionSchema } from "./schema.js"
import { toSessionError } from "./to-session-error.js"
import { Token } from "../util/token.js"
@ -405,68 +416,105 @@ export const layer = Layer.effect(
],
},
})
// Ignored tool calls never enter the follow-up history or need fabricated results.
for (let attempt = 0; attempt < 2; attempt++) {
chunks.length = 0
providerState = undefined
yield* llm
.stream(
attempt === 0
? prepared.request
: LLMRequest.update(prepared.request, {
messages: [
...prepared.request.messages,
Message.user(
"The previous response did not fill in the required summary template. Do not call tools. Return the summary as text using the exact section headings from the template.",
),
],
}),
prepared.options,
)
.pipe(
Stream.runForEach((event) => {
if (LLMEvent.is.providerError(event))
failure = {
type: event.classification === "context-overflow" ? "provider.invalid-request" : "provider.error",
message: event.message,
}
if (LLMEvent.is.textDelta(event)) {
chunks.push(event.text)
return bus.publish(SessionEvent.Compaction.Delta, {
sessionID: context.session.id,
text: event.text,
})
}
if (LLMEvent.is.stepFinish(event)) {
providerState =
event.providerMetadata?.[
context.model.model.route.providerMetadataKey ?? context.model.model.provider
]
const step = SessionUsage.record(event.usage, context.model.cost)
usage = usage ? SessionUsage.add(usage, step) : step
}
return Effect.void
}),
Effect.catchTag("AI.Error", (error) =>
Effect.sync(() => {
failure = toSessionError(error)
}),
const retry = yield* SessionRunnerRetry.policy(context.session.id)
// Both requests share the retry allowance; rejected output never enters the reminder request.
for (const request of [
prepared.request,
LLMRequest.update(prepared.request, {
messages: [
...prepared.request.messages,
Message.user(
"The previous response did not fill in the required summary template. Do not call tools. Return the summary as text using the exact section headings from the template.",
),
Effect.onInterrupt(() =>
recordUsage.pipe(
Effect.andThen(
input.reason === "auto"
? failed({
sessionID: context.session.id,
reason: input.reason,
error: { type: "compaction.interrupted", message: "Compaction was interrupted" },
inputID: input.inputID,
}).pipe(Effect.asVoid)
: Effect.void,
),
],
}),
]) {
yield* Stream.suspend(() => {
chunks.length = 0
providerState = undefined
failure = undefined
return llm.stream(request, prepared.options)
}).pipe(
Stream.runForEach((event) => {
if (LLMEvent.is.providerError(event))
failure = {
type: event.classification === "context-overflow" ? "provider.invalid-request" : "provider.error",
message: event.message,
}
if (LLMEvent.is.textDelta(event)) {
chunks.push(event.text)
return bus.publish(SessionEvent.Compaction.Delta, {
sessionID: context.session.id,
text: event.text,
})
}
if (LLMEvent.is.stepFinish(event)) {
providerState =
event.providerMetadata?.[context.model.model.route.providerMetadataKey ?? context.model.model.provider]
const step = SessionUsage.record(event.usage, context.model.cost)
usage = usage ? SessionUsage.add(usage, step) : step
}
if (LLMEvent.is.finish(event)) {
if (event.reason.normalized === "length")
failure = { type: "compaction.failed", message: "Compaction summary reached the output token limit" }
if (event.reason.normalized === "content-filter")
failure = {
type: "provider.content-filter",
message: "Compaction summary was blocked by the provider",
}
if (event.reason.normalized === "unknown")
return Effect.fail(
new AIError({
reason: new InvalidProviderOutputError({
message: "The provider response ended with an unknown finish reason.",
classification: "incomplete-stream",
}),
}),
)
if (event.reason.normalized === "error")
return Effect.fail(
new AIError({ reason: new UnknownProviderError({ message: "Compaction generation failed" }) }),
)
}
return Effect.void
}),
Effect.retry({
while: (cause) =>
Effect.gen(function* () {
if (isContextOverflowFailure(cause)) return false
const decision = yield* retry({
cause,
error: toSessionError(cause),
agent: Agent.ID.make("compaction"),
model: context.model.ref,
hook: prepared.retry,
retry: SessionRunnerRetry.isRetryable(cause),
})
if (!decision.retry) return false
yield* Effect.sleep(decision.delay)
return true
}),
}),
Effect.catchTag("AI.Error", (error) =>
Effect.sync(() => {
failure = toSessionError(error)
}),
),
Effect.onInterrupt(() =>
recordUsage.pipe(
Effect.andThen(
input.reason === "auto"
? failed({
sessionID: context.session.id,
reason: input.reason,
error: { type: "compaction.interrupted", message: "Compaction was interrupted" },
inputID: input.inputID,
}).pipe(Effect.asVoid)
: Effect.void,
),
),
)
),
)
if (failure || hasSummarySection(chunks.join(""))) break
}
yield* recordUsage

9
packages/core/src/session/runner/retry.ts
View file

@ -78,11 +78,11 @@ const schedule = Schedule.max([Schedule.exponential("2 seconds"), Schedule.recur
}),
)
export const make = (bus: Bus.Interface, sessionID: SessionSchema.ID) =>
export const policy = (sessionID: SessionSchema.ID) =>
Effect.gen(function* () {
const step = yield* Schedule.toStep(schedule)
let attempt = 1
const decide = (input: Input) =>
return (input: Input) =>
Effect.gen(function* () {
const now = yield* Clock.currentTimeMillis
const next = yield* step(now, input).pipe(Pull.catchDone(() => Effect.succeed(undefined)))
@ -104,6 +104,11 @@ export const make = (bus: Bus.Interface, sessionID: SessionSchema.ID) =>
Number.isFinite(event.decision.delay) && event.decision.delay >= 0 ? Math.ceil(event.decision.delay) : delay
return { retry: true as const, attempt, delay: normalized }
})
})
export const make = (bus: Bus.Interface, sessionID: SessionSchema.ID) =>
Effect.gen(function* () {
const decide = yield* policy(sessionID)
const wait = (input: {
readonly decision: Decision
readonly assistantMessageID: SessionMessage.ID

139
packages/core/test/session-runner.test.ts
View file

@ -1,6 +1,7 @@
import { describe, expect, test } from "bun:test"
import {
AIError,
HttpContext,
LLMEvent,
LLMRequest,
Message,
@ -2260,21 +2261,151 @@ describe("SessionRunnerLLM", () => {
}
}
scenario("preserves typed provider failures from manual compaction", function* (s) {
scenario("restarts compaction drafts after transient failures and unsuccessful finishes", function* (s) {
yield* s.llm.push(TestLLM.text("Earlier answer", "text-manual-failure-history"))
yield* s.runPrompt("Earlier question")
yield* s.llm.push(Stream.fail(providerUnavailable()))
s.requests.length = 0
const hooks = yield* PluginHooks.Service
const retries: PluginHooks.Domains["session"]["retry"][] = []
yield* hooks.register("session", "retry", (event) =>
Effect.sync(() => {
retries.push({ ...event })
event.decision = { retry: true, delay: 0 }
}),
)
const draft = TestLLM.complete(
{
reason: { normalized: "unknown" },
usage: { nonCachedInputTokens: 10 },
providerMetadata: { openai: { responseId: "discarded-draft" } },
},
LLMEvent.textDelta({ id: "draft", text: "## Objective\n- Partial draft" }),
)
yield* s.llm.push(
TestLLM.failAfter(streamDisconnected(), ...draft.slice(0, -1)),
draft,
TestLLM.complete(
{ reason: { normalized: "error" } },
LLMEvent.textDelta({ id: "failed", text: "## Objective\n- Failed draft" }),
),
Stream.fail(rateLimited(60_000)),
TestLLM.textWithUsage("## Objective\n- Accepted summary", "accepted", 30),
)
const compaction = yield* s.session.compact({ sessionID })
yield* s.resume
expect(s.requests).toHaveLength(5)
for (const request of s.requests) expect(request).toEqual(s.requests[0])
expect(retries.map((event) => event.attempt)).toEqual([2, 3, 4, 5])
expect(retries.every((event) => event.sessionID === sessionID && event.agent === "compaction")).toBe(true)
expect(retries[3].decision).toEqual({ retry: true, delay: 60_000 })
expect((yield* s.messages).find((message) => message.id === compaction.id)).toMatchObject({
status: "completed",
summary: "## Objective\n- Accepted summary",
})
expect(JSON.stringify(yield* s.messages)).not.toContain("discarded-draft")
expect((yield* s.session.get(sessionID))?.tokens.input).toBe(50)
})
scenario("bounds compaction network retries across a template correction", function* (s) {
yield* s.llm.push(TestLLM.text("Earlier answer", "history"))
yield* s.runPrompt("Earlier question")
s.requests.length = 0
const hooks = yield* PluginHooks.Service
const attempts: number[] = []
yield* hooks.register("session", "retry", (event) =>
Effect.sync(() => {
attempts.push(event.attempt)
expect(event.decision).toMatchObject({ retry: true })
event.decision = { retry: true, delay: 0 }
}),
)
yield* s.llm.push(Stream.fail(providerUnavailable()), TestLLM.text("Not a summary", "invalid"))
yield* s.llm.always(Stream.fail(providerUnavailable()))
const compaction = yield* s.session.compact({ sessionID })
yield* s.resume
expect(attempts).toEqual([2, 3, 4, 5])
expect(s.requests).toHaveLength(6)
expect((yield* s.messages).find((message) => message.id === compaction.id)).toMatchObject({
type: "compaction",
status: "failed",
error: { type: "provider.transport", message: "Provider unavailable" },
})
expect((yield* s.context).some((message) => message.type === "user" && message.text === "Earlier question")).toBe(
true,
)
})
for (const header of [false, true]) {
scenario(`stops compaction retries through the ${header ? "provider header" : "retry hook"}`, function* (s) {
yield* s.llm.push(TestLLM.text("Earlier answer", "history"))
yield* s.runPrompt("Earlier question")
s.requests.length = 0
const hooks = yield* PluginHooks.Service
yield* hooks.register("session", "retry", (event) =>
Effect.sync(() => {
expect(event.decision.retry).toBe(!header)
event.decision = { retry: false }
}),
)
yield* s.llm.push(
Stream.fail(
header
? new AIError({
reason: new TransportError({
message: "Connection closed",
transport: "http",
operation: "read",
http: new HttpContext({
url: "https://example.com",
status: 200,
headers: { "x-should-retry": "false" },
}),
}),
})
: incompleteStream(),
),
)
const compaction = yield* s.session.compact({ sessionID })
yield* s.resume
expect(s.requests).toHaveLength(1)
expect((yield* s.messages).find((message) => message.id === compaction.id)).toMatchObject({
status: "failed",
error: { type: header ? "provider.transport" : "provider.invalid-output" },
})
})
}
for (const response of ["length", "content-filter", "context overflow"] as const) {
scenario(`rejects compaction ${response} without retrying or committing its draft`, function* (s) {
yield* s.llm.push(TestLLM.text("Earlier answer", "history"))
yield* s.runPrompt("Earlier question")
s.requests.length = 0
yield* s.llm.push(
response === "context overflow"
? Stream.fail(
new AIError({
reason: new InvalidRequestError({ message: "Too long", classification: "context-overflow" }),
}),
)
: TestLLM.complete(
{ reason: { normalized: response } },
LLMEvent.textDelta({ id: "truncated", text: "## Objective\n- Incomplete summary" }),
),
)
const compaction = yield* s.session.compact({ sessionID })
yield* s.resume
expect(s.requests).toHaveLength(1)
expect((yield* s.messages).find((message) => message.id === compaction.id)).toMatchObject({ status: "failed" })
yield* s.llm.push(TestLLM.text("Continued", "continued"))
yield* s.runPrompt("Continue")
expect(userTexts(s.requests[1])).toContain("Earlier question")
expect(JSON.stringify(s.requests[1])).not.toContain("Incomplete summary")
})
}
scenario("records cancelled manual compaction without surfacing an internal failure", function* (s) {
yield* s.llm.push(TestLLM.text("Earlier answer", "text-manual-interrupt-history"))
yield* s.runPrompt("Earlier question")