diff --git a/packages/core/src/session/compaction.ts b/packages/core/src/session/compaction.ts index 912a9ac498e..1cad39a7dcf 100644 --- a/packages/core/src/session/compaction.ts +++ b/packages/core/src/session/compaction.ts @@ -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 diff --git a/packages/core/src/session/runner/retry.ts b/packages/core/src/session/runner/retry.ts index b67cb23de0c..828904c2824 100644 --- a/packages/core/src/session/runner/retry.ts +++ b/packages/core/src/session/runner/retry.ts @@ -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 diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 26e4494bb30..f77849a2ecc 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -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")