From 664ed37b0c36ec74a9d431383b9e1a0eadc73858 Mon Sep 17 00:00:00 2001 From: DanielWalnut <45447813+hetaoBackend@users.noreply.github.com> Date: Fri, 9 Oct 2026 13:29:32 +0800 Subject: [PATCH 1/2] fix: stop retrying oversized requests and recover with one forced compaction A request the provider rejects as too large (HTTP 413, request_too_large, the local request-body check, or a context-window overflow such as MiniMax 'context window exceeds limit (2013)') fails identically on every resend. The API-key and custom-provider paths retried it five times. - Classify these rejections as deterministic (isLLMRequestOversized), so withLLMRetry sends them once on every path. - Add contextOverflowRecoveryHook to PiTurnHooks. When the main-agent request is rejected as too large before any output, the runner runs only these hooks once per turn with trigger 'context_overflow_recovery', persists the replacement and resends once; a second rejection is final. - local-runtime-v2 registers the automatic compaction hook there; the trigger forces compaction and targets at most half of the rejected request, since the local estimate had admitted it. --- .../context-overflow-recovery.ts | 122 ++++++++ .../agent-core/src/pi-turn-runner/hooks.ts | 15 + packages/agent-core/src/pi-turn-runner/llm.ts | 128 ++++++--- .../agent-core/src/pi-turn-runner/turn.ts | 28 +- .../context-overflow-recovery.test.ts | 267 ++++++++++++++++++ .../unit/pi-turn-runner/llm-retry.test.ts | 69 +++++ .../compaction/context-compaction.ts | 3 + .../agent-host/compaction/contracts.ts | 7 + .../agent-host/execution/executor.ts | 8 + .../automatic-context-compactor.test.ts | 66 +++++ .../compaction/automatic-context-compactor.ts | 41 ++- packages/shared/src/llm-error-classifier.ts | 44 ++- release/public-source.json | 2 + test/vitest-suites.json | 1 + 14 files changed, 757 insertions(+), 44 deletions(-) create mode 100644 packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts create mode 100644 packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts diff --git a/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts b/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts new file mode 100644 index 000000000..38a0af37f --- /dev/null +++ b/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts @@ -0,0 +1,122 @@ +import type { StreamFn } from '@earendil-works/pi-agent-core'; +import { + isContextOverflow, + type AssistantMessage, + type AssistantMessageEvent, + type AssistantMessageEventStream, + type Message, +} from '@earendil-works/pi-ai'; +import { isLLMRequestOversized, normalizeLLMError } from '@mavis/shared/llm-error-classifier'; + +/** + * Per-turn slot connecting the provider stream (built before the Agent exists) + * with the Agent-owned context lifecycle that can shrink the context. + */ +export interface ContextOverflowRecoveryState { + /** + * Shrinks the context of the request that just failed and returns the + * provider-bound messages to resend, or `undefined` when nothing changed. + * Installed by `setLLMHook` only when recovery hooks exist. + */ + recover?: (signal: AbortSignal | undefined) => Promise; + /** Recovery runs at most once per turn, so a second rejection is terminal. */ + attempted: boolean; +} + +export function newContextOverflowRecoveryState(): ContextOverflowRecoveryState { + return { attempted: false }; +} + +/** + * True when a failed provider response means the request was too large to be + * accepted as sent: HTTP 413 / `request_too_large`, the local request-body + * check, or a context-window overflow recognized by pi-ai. + */ +export function isRecoverableContextOverflow( + message: AssistantMessage, + contextWindow: number | undefined, +): boolean { + if (message.stopReason !== 'error') return false; + if (isContextOverflow(message, contextWindow)) return true; + return isLLMRequestOversized( + normalizeLLMError({ + finishReason: message.stopReason, + ...(message.errorMessage ? { errorMessage: message.errorMessage } : {}), + }), + ); +} + +/** + * When the provider rejects a main-agent request as too large before producing + * any output, compact once and resend once. Successful streams only pay for + * buffering their first non-`start` event. A second rejection, a recovery that + * changes nothing, or any other failure is surfaced exactly as received. + */ +export function withContextOverflowRecovery( + inner: StreamFn, + state: ContextOverflowRecoveryState, +): StreamFn { + return (async (model, context, options) => { + const stream = await inner(model, context, options); + if (!state.recover || state.attempted) return stream; + const iterator = stream[Symbol.asyncIterator](); + const buffered: AssistantMessageEvent[] = []; + try { + for (;;) { + const next = await iterator.next(); + if (next.done) return replay(stream, buffered, undefined); + buffered.push(next.value); + if (next.value.type !== 'start') break; + } + } catch (error) { + return replay(stream, buffered, undefined, { error }); + } + const first = buffered.at(-1); + if ( + first?.type !== 'error' || + options?.signal?.aborted || + !isRecoverableContextOverflow(first.error, model.contextWindow) + ) { + return replay(stream, buffered, iterator); + } + state.attempted = true; + const messages = await state.recover(options?.signal); + if (!messages || options?.signal?.aborted) return replay(stream, buffered, iterator); + try { + void Promise.resolve(iterator.return?.()).catch(() => undefined); + } catch { + // Best-effort cleanup of the rejected attempt. + } + return inner(model, { ...context, messages }, options); + }) as StreamFn; +} + +function replay( + stream: AssistantMessageEventStream, + buffered: readonly AssistantMessageEvent[], + iterator: AsyncIterator | undefined, + pending?: { readonly error: unknown }, +): AssistantMessageEventStream { + return { + [Symbol.asyncIterator]() { + let index = 0; + let thrown = false; + return { + async next(): Promise> { + if (index < buffered.length) return { value: buffered[index++]!, done: false }; + if (pending && !thrown) { + thrown = true; + throw pending.error; + } + if (!iterator) return { value: undefined, done: true }; + return iterator.next(); + }, + async return(value?: unknown): Promise> { + await iterator?.return?.(value); + return { value: undefined, done: true }; + }, + }; + }, + result: () => stream.result(), + } as unknown as AssistantMessageEventStream; +} diff --git a/packages/agent-core/src/pi-turn-runner/hooks.ts b/packages/agent-core/src/pi-turn-runner/hooks.ts index 9702ff9fb..4e0e51c8e 100644 --- a/packages/agent-core/src/pi-turn-runner/hooks.ts +++ b/packages/agent-core/src/pi-turn-runner/hooks.ts @@ -26,6 +26,12 @@ export interface PiBeforeLlmCallHookInput { sessionId: string; turnId: string; phase: PiBeforeLlmCallPhase; + /** + * Set only when the runner re-runs `contextOverflowRecoveryHook` after the + * provider rejected the request as too large (HTTP 413 / context overflow). + * Compaction hooks must compact regardless of their own size estimate. + */ + trigger?: 'context_overflow_recovery'; /** Current provider request view after request-only hook transforms. */ messages: AgentMessage[]; /** Current durable history, updated only by canonical replacements. */ @@ -251,6 +257,15 @@ export type PiOnTurnEndHook = PiOnStepEndHook; export interface PiTurnHooks { beforeLlmCallHook?: readonly PiBeforeLlmCallHook[]; + /** + * Hooks that may shrink the context after the provider rejected a main-agent + * request as too large (HTTP 413, `request_too_large`, the local request-body + * check, or a context-window overflow) before producing any output. The + * runner runs them once per turn with `trigger: 'context_overflow_recovery'` + * and, when they replace the context, resends the request once. Only + * compaction belongs here: reminder or append hooks would run twice. + */ + contextOverflowRecoveryHook?: readonly PiBeforeLlmCallHook[]; /** Best-effort observers of actual logical main-agent provider calls. */ onLlmCallPreparedHook?: readonly PiOnLlmCallPreparedHook[]; afterLlmCallHook?: readonly PiAfterLlmCallHook[]; diff --git a/packages/agent-core/src/pi-turn-runner/llm.ts b/packages/agent-core/src/pi-turn-runner/llm.ts index d9bcfcb8c..5206dd847 100644 --- a/packages/agent-core/src/pi-turn-runner/llm.ts +++ b/packages/agent-core/src/pi-turn-runner/llm.ts @@ -5,6 +5,7 @@ import { resolveLLMStreamTimeouts, withLLMStreamTimeouts } from './llm-stream-ti import type { PiAfterLlmReplacementCommit, PiBeforeLlmCallAppendMessage, + PiBeforeLlmCallHook, PiBeforeLlmCallReplaceMetadata, } from './hooks.js'; import type { LLMModelConfig } from './types.js'; @@ -41,40 +42,24 @@ export function composeStreamFn(resolved: LLMModelConfig): StreamFn { } export function setLLMHook(agent: Agent, turn: turnState, history: turnHistory): void { - if (turn.hooks.beforeLLM.length === 0 && turn.hooks.afterLLM.length === 0) return; + const recoveryHooks = turn.hooks.contextOverflowRecovery; + if ( + turn.hooks.beforeLLM.length === 0 && + turn.hooks.afterLLM.length === 0 && + recoveryHooks.length === 0 + ) { + return; + } let checkpointCallCount = 0; - agent.transformContext = async (messages, signal) => { - const rejectedMessages = messages.filter((message) => - turn.rejectedAssistantMessages.has(message), - ); - if (rejectedMessages.length > 0) { - messages.splice( - 0, - messages.length, - ...messages.filter((message) => !turn.rejectedAssistantMessages.has(message)), - ); - for (const message of rejectedMessages) turn.rejectedAssistantMessages.delete(message); - } + // Pi's live loop context of the latest main-agent request; overflow recovery + // compacts exactly this context and keeps the loop in sync through it. + let lastRequest: { messages: AgentMessage[]; phase: 'initial' | 'iteration' } | undefined; - if (turn.hooks.beforeLLM.length === 0) return messages; - const phase = checkpointCallCount === 0 ? 'initial' : 'iteration'; - checkpointCallCount += 1; - const checkpoint = await runBeforeLLM(turn, messages, phase, signal); - if (checkpoint.type === 'abort') { - turn.metrics.tryRecordTerminalFailure({ - errorSource: 'before_llm_hook', - errorKind: 'hook_error', - }); - throw new Error(`before_llm_checkpoint_aborted: ${checkpoint.reason}`); - } - if (checkpoint.type === 'respond') { - turn.syntheticResponse.pending = { - text: checkpoint.text, - reason: checkpoint.reason, - }; - return messages; - } + const applyCheckpoint = async ( + messages: AgentMessage[], + checkpoint: Exclude, + ): Promise => { if (checkpoint.type === 'requestOnly') { return checkpoint.messages; } @@ -130,13 +115,89 @@ export function setLLMHook(agent: Agent, turn: turnState, history: turnHistory): } return checkpoint.messages; }; + + agent.transformContext = async (messages, signal) => { + const rejectedMessages = messages.filter((message) => + turn.rejectedAssistantMessages.has(message), + ); + if (rejectedMessages.length > 0) { + messages.splice( + 0, + messages.length, + ...messages.filter((message) => !turn.rejectedAssistantMessages.has(message)), + ); + for (const message of rejectedMessages) turn.rejectedAssistantMessages.delete(message); + } + + const phase = checkpointCallCount === 0 ? 'initial' : 'iteration'; + lastRequest = { messages, phase }; + if (turn.hooks.beforeLLM.length === 0) return messages; + checkpointCallCount += 1; + const checkpoint = await runBeforeLLM(turn, messages, phase, signal, turn.hooks.beforeLLM); + if (checkpoint.type === 'abort') { + turn.metrics.tryRecordTerminalFailure({ + errorSource: 'before_llm_hook', + errorKind: 'hook_error', + }); + throw new Error(`before_llm_checkpoint_aborted: ${checkpoint.reason}`); + } + if (checkpoint.type === 'respond') { + turn.syntheticResponse.pending = { + text: checkpoint.text, + reason: checkpoint.reason, + }; + return messages; + } + return applyCheckpoint(messages, checkpoint); + }; + + if (recoveryHooks.length === 0) return; + turn.contextOverflowRecovery.recover = async (signal) => { + const request = lastRequest; + if (!request) return undefined; + const logContext = { + session_id: turn.input.sessionId, + turn_id: turn.input.turnId, + phase: request.phase, + provider: turn.llm.model.provider, + }; + const checkpoint = await runBeforeLLM( + turn, + request.messages, + request.phase, + signal, + recoveryHooks, + 'context_overflow_recovery', + ); + if ( + checkpoint.type === 'abort' || + checkpoint.type === 'respond' || + checkpoint.type === 'unchanged' + ) { + turn.logger.warn( + { ...logContext, outcome: checkpoint.type }, + '[pi-turn-runner] context overflow recovery left the request unchanged; surfacing the provider rejection', + ); + return undefined; + } + const messages = await applyCheckpoint(request.messages, checkpoint); + turn.logger.info( + { ...logContext, outcome: checkpoint.type, message_count: messages.length }, + '[pi-turn-runner] context overflow recovery compacted the request; resending once', + ); + return await agent.convertToLlm(messages); + }; } +type BeforeLLMResult = Awaited>; + async function runBeforeLLM( turn: turnState, messages: AgentMessage[], phase: 'initial' | 'iteration', - signal?: AbortSignal, + signal: AbortSignal | undefined, + hooks: readonly PiBeforeLlmCallHook[], + trigger?: 'context_overflow_recovery', ): Promise< | { type: 'unchanged'; messages: AgentMessage[] } | { type: 'requestOnly'; messages: AgentMessage[] } @@ -172,13 +233,14 @@ async function runBeforeLLM( afterCommit?: PiAfterLlmReplacementCommit; } | undefined; - for (const hook of turn.hooks.beforeLLM) { + for (const hook of hooks) { let decision: Awaited>; try { decision = await hook({ sessionId: turn.input.sessionId, turnId: turn.input.turnId, phase, + ...(trigger ? { trigger } : {}), messages: [...currentMessages], canonicalMessages: [...canonicalMessages], model: turn.llm.model, diff --git a/packages/agent-core/src/pi-turn-runner/turn.ts b/packages/agent-core/src/pi-turn-runner/turn.ts index 95fe531da..15d235362 100644 --- a/packages/agent-core/src/pi-turn-runner/turn.ts +++ b/packages/agent-core/src/pi-turn-runner/turn.ts @@ -19,6 +19,11 @@ import type { PiStepEndHookInput, PiToolExecutionStartHook, } from './hooks.js'; +import { + newContextOverflowRecoveryState, + withContextOverflowRecovery, + type ContextOverflowRecoveryState, +} from './context-overflow-recovery.js'; import { composeStreamFn } from './llm.js'; import { withLLMRetry, type LLMCallScope, type LLMCallSettledEvent } from './llm-retry.js'; import type { TurnMetricsRecorder } from './metrics.js'; @@ -86,6 +91,7 @@ export interface LlmCaptureAgentEventSource { export interface turnHooks { beforeLLM: readonly PiBeforeLlmCallHook[]; + contextOverflowRecovery: readonly PiBeforeLlmCallHook[]; onLLMPrepared: readonly PiOnLlmCallPreparedHook[]; afterLLM: readonly PiAfterLlmCallHook[]; beforeTool: readonly PiBeforeToolCallHook[]; @@ -120,6 +126,8 @@ export interface turnState; + /** One-shot compact-and-resend after a main-agent 413 / context overflow. */ + contextOverflowRecovery: ContextOverflowRecoveryState; nextEventId(kind: string): string; nextRuntimeSeq(): number; } @@ -130,6 +138,7 @@ export function newTurn( ): turnState { const resolved = input.llm; const composedStreamFn = composeStreamFn(resolved); + const contextOverflowRecovery = newContextOverflowRecoveryState(); const retryStream = (scope: LLMCallScope): StreamFn => { const metricsStreamFn = deps.metrics ? deps.metrics.wrapStreamFn(composedStreamFn, { recordTerminalFailure: false }) @@ -151,11 +160,17 @@ export function newTurn( return callerOnCallSettled?.(event); }, }); + // Overflow recovery sits inside the logical-call observers so a recovered + // request settles as one successful logical call. + const recoveringStreamFn = + scope === 'agent' + ? withContextOverflowRecovery(retryingStreamFn, contextOverflowRecovery) + : retryingStreamFn; // One recorder spans the whole logical call. Physical retries overwrite its // in-memory candidate; only the final successful settlement can persist. const capturedStreamFn = deps.llmCapture - ? deps.llmCapture.wrapLogicalStreamFn(retryingStreamFn, { scope }) - : retryingStreamFn; + ? deps.llmCapture.wrapLogicalStreamFn(recoveringStreamFn, { scope }) + : recoveringStreamFn; return deps.metrics ? deps.metrics.wrapLogicalStreamFn(capturedStreamFn, { // Auto-compaction is an auxiliary degradation: local context lifecycle @@ -169,9 +184,10 @@ export function newTurn( const mainStreamFn = input.llmRetry || deps.llmCapture ? retryStream('agent') - : deps.metrics - ? deps.metrics.wrapStreamFn(composedStreamFn) - : composedStreamFn; + : withContextOverflowRecovery( + deps.metrics ? deps.metrics.wrapStreamFn(composedStreamFn) : composedStreamFn, + contextOverflowRecovery, + ); const auxiliaryStreamFn = input.llmRetry ? retryStream('compaction') : undefined; const syntheticResponse: turnState['syntheticResponse'] = {}; let eventSeq = 0; @@ -223,6 +239,7 @@ export function newTurn( ); const hooks: turnHooks = { beforeLLM: input.hooks?.beforeLlmCallHook ?? [], + contextOverflowRecovery: input.hooks?.contextOverflowRecoveryHook ?? [], onLLMPrepared: input.hooks?.onLlmCallPreparedHook ?? [], afterLLM: input.hooks?.afterLlmCallHook ?? [], beforeTool: input.hooks?.beforeToolCallHook ?? [], @@ -265,6 +282,7 @@ export function newTurn( blockedToolCalls: [], syntheticResponse, rejectedAssistantMessages: new Set(), + contextOverflowRecovery, nextEventId, nextRuntimeSeq, }; diff --git a/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts b/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts new file mode 100644 index 000000000..78fd1b0ba --- /dev/null +++ b/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts @@ -0,0 +1,267 @@ +import { describe, expect, it, vi } from "vitest"; +import type { AgentMessage, StreamFn } from "@earendil-works/pi-agent-core"; +import type { + AssistantMessage, + AssistantMessageEvent, + AssistantMessageEventStream, + Context, + Model, +} from "@earendil-works/pi-ai"; +import { PiTurnRunner } from "../../../src/pi-turn-runner/pi-turn-runner.js"; +import type { + PiBeforeLlmCallHook, + PiBeforeLlmCallHookInput, + PiHistoryChangedHookInput, +} from "../../../src/pi-turn-runner/hooks.js"; +import type { RuntimeEvent } from "../../../src/pi-turn-runner/types.js"; +import { RuntimeEventStatus } from "../../../src/protocol/runtime-event.js"; + +function fakeModel(provider = "minimax_api"): Model { + return { + id: "fake-model", + name: "fake-model", + api: "test-messages", + provider, + baseUrl: "https://example.invalid", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 200_000, + maxTokens: 8_192, + } as Model; +} + +function assistant( + stopReason: AssistantMessage["stopReason"], + content: AssistantMessage["content"], + errorMessage?: string, +): AssistantMessage { + return { + role: "assistant", + content, + api: "test-messages", + provider: "minimax_api", + model: "fake-model", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason, + ...(errorMessage ? { errorMessage } : {}), + timestamp: 0, + } as AssistantMessage; +} + +function stream(events: AssistantMessageEvent[], final: AssistantMessage): AssistantMessageEventStream { + return { + [Symbol.asyncIterator]() { + let index = 0; + return { + async next() { + if (index < events.length) return { value: events[index++], done: false }; + return { value: undefined, done: true }; + }, + } as AsyncIterableIterator; + }, + async result() { + return final; + }, + } as unknown as AssistantMessageEventStream; +} + +function successStream(text: string): AssistantMessageEventStream { + const empty = assistant("stop", []); + const final = assistant("stop", [{ type: "text", text }]); + return stream( + [ + { type: "start", partial: empty }, + { type: "text_start", contentIndex: 0, partial: empty }, + { type: "text_delta", contentIndex: 0, delta: text, partial: final }, + { type: "text_end", contentIndex: 0, content: text, partial: final }, + { type: "done", reason: "stop", message: final }, + ], + final, + ); +} + +function errorStream(message: string): AssistantMessageEventStream { + const empty = assistant("error", []); + const final = assistant("error", [], message); + return stream( + [ + { type: "start", partial: empty }, + { type: "error", reason: "error", error: final }, + ], + final, + ); +} + +const OVERSIZED = "413 Request Entity Too Large"; + +function textOf(message: { content: unknown }): string { + const content = message.content; + if (typeof content === "string") return content; + return (content as { type: string; text?: string }[]) + .map((part) => (part.type === "text" ? (part.text ?? "") : "")) + .join(""); +} + +function summaryMessage(): AgentMessage { + return { role: "user", content: [{ type: "text", text: "SUMMARY" }], timestamp: 1 } as AgentMessage; +} + +function compactingHook(calls: PiBeforeLlmCallHookInput[]): PiBeforeLlmCallHook { + return (input) => { + calls.push(input); + const kept = input.messages.slice(-1); + return { + type: "replaceMessages", + messages: [summaryMessage(), ...kept], + metadata: { + replacementId: `r${calls.length}`, + strategyVersion: "test", + summary: "SUMMARY", + compactedMessages: input.messages.slice(0, -1), + keptMessages: kept, + firstKeptIndex: input.messages.length - 1, + }, + }; + }; +} + +async function runTurn(options: { + responses: Array<() => AssistantMessageEventStream>; + recoveryHooks?: PiBeforeLlmCallHook[]; + beforeLlmCallHook?: PiBeforeLlmCallHook[]; + provider?: string; +}) { + const contexts: Context[] = []; + let index = 0; + const streamFn = (async (_model: Model, context: Context) => { + contexts.push({ ...context, messages: structuredClone(context.messages) }); + const next = options.responses[Math.min(index, options.responses.length - 1)]!; + index += 1; + return next(); + }) as unknown as StreamFn; + const events: RuntimeEvent[] = []; + const historyChanges: PiHistoryChangedHookInput[] = []; + const runner = new PiTurnRunner(); + await runner.runTurn({ + sessionId: "session-1", + turnId: "turn-1", + workspaceDir: "/tmp", + systemPrompt: "system", + userMessage: { text: "current question", timestamp: 2 }, + history: [ + { role: "user", content: [{ type: "text", text: "old question" }], timestamp: 0 } as AgentMessage, + assistant("stop", [{ type: "text", text: "old answer ".repeat(100) }]) as AgentMessage, + ], + llm: { model: fakeModel(options.provider), apiKey: "k", streamFn }, + eventWriter: { + pushRuntime: (event) => void events.push(event), + appendEvents: (batch) => void events.push(...batch), + }, + toolConfig: { tools: [], context: {} as never }, + llmRetry: { sleep: vi.fn(async () => {}), random: () => 0 }, + hooks: { + beforeLlmCallHook: options.beforeLlmCallHook ?? [() => ({ type: "continue" })], + ...(options.recoveryHooks ? { contextOverflowRecoveryHook: options.recoveryHooks } : {}), + onHistoryChangedHook: [async (change) => void historyChanges.push(change)], + }, + }); + const terminal = events + .map((event) => event as unknown as { type?: string; payload?: { status?: string } }) + .filter((event) => event.payload?.status !== undefined) + .at(-1); + return { contexts, events, historyChanges, terminal }; +} + +describe("context overflow recovery", () => { + it("compacts once and resends once after a BYOK 413", async () => { + const recoveryCalls: PiBeforeLlmCallHookInput[] = []; + const result = await runTurn({ + responses: [() => errorStream(OVERSIZED), () => successStream("recovered")], + recoveryHooks: [compactingHook(recoveryCalls)], + }); + + // One rejected request, one resend: the 413 is not retried five times. + expect(result.contexts).toHaveLength(2); + expect(recoveryCalls).toHaveLength(1); + expect(recoveryCalls[0]!.trigger).toBe("context_overflow_recovery"); + expect(result.contexts[0]!.messages).toHaveLength(3); + expect(result.contexts[1]!.messages.map(textOf)).toEqual(["SUMMARY", "current question"]); + // The compaction is persisted as a durable replacement before the resend. + const replace = result.historyChanges.find((change) => change.reason === "replaceMessages"); + expect(replace?.messages.map((message) => textOf(message as never))).toEqual([ + "SUMMARY", + "current question", + ]); + // The rejected attempt never reaches history; only the recovered answer does. + const deltas = result.historyChanges + .filter((change) => change.reason === "messageDelta") + .flatMap((change) => change.messages); + const assistants = deltas.filter((message) => message.role === "assistant"); + expect(assistants.map((message) => (message as AssistantMessage).stopReason)).toEqual(["stop"]); + expect(result.terminal?.payload?.status).toBe(RuntimeEventStatus.COMPLETED); + }); + + it("surfaces the rejection when the resent request is still too large", async () => { + const recoveryCalls: PiBeforeLlmCallHookInput[] = []; + const result = await runTurn({ + responses: [() => errorStream(OVERSIZED)], + recoveryHooks: [compactingHook(recoveryCalls)], + }); + + expect(result.contexts).toHaveLength(2); + expect(recoveryCalls).toHaveLength(1); + const deltas = result.historyChanges + .filter((change) => change.reason === "messageDelta") + .flatMap((change) => change.messages); + expect((deltas.at(-1) as AssistantMessage).errorMessage).toBe(OVERSIZED); + expect(result.terminal?.payload?.status).toBe(RuntimeEventStatus.FAILED); + }); + + it("does not resend when recovery leaves the context unchanged", async () => { + const recovery = vi.fn(() => ({ type: "skip", reason: "context_compaction_failed" })); + const result = await runTurn({ + responses: [() => errorStream(OVERSIZED)], + recoveryHooks: [recovery], + }); + + expect(result.contexts).toHaveLength(1); + expect(recovery).toHaveBeenCalledOnce(); + }); + + it("does not run recovery for errors that are not about request size", async () => { + const recoveryCalls: PiBeforeLlmCallHookInput[] = []; + const result = await runTurn({ + responses: [() => errorStream('400 {"type":"invalid_request_error","message":"messages: text content blocks must be non-empty"}')], + recoveryHooks: [compactingHook(recoveryCalls)], + }); + + expect(result.contexts).toHaveLength(1); + expect(recoveryCalls).toHaveLength(0); + }); + + it("does not re-run ordinary before-LLM hooks during recovery", async () => { + const ordinary = vi.fn(() => ({ type: "continue" })); + const recoveryCalls: PiBeforeLlmCallHookInput[] = []; + const result = await runTurn({ + responses: [() => errorStream(OVERSIZED), () => successStream("recovered")], + recoveryHooks: [compactingHook(recoveryCalls)], + beforeLlmCallHook: [ordinary], + }); + + expect(result.contexts).toHaveLength(2); + expect(ordinary).toHaveBeenCalledOnce(); + }); + + it("keeps the previous single-failure behaviour when no recovery hook is installed", async () => { + const result = await runTurn({ responses: [() => errorStream(OVERSIZED)] }); + expect(result.contexts).toHaveLength(1); + }); +}); diff --git a/packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts b/packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts index ccfbf42f9..69292f1f2 100644 --- a/packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts +++ b/packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts @@ -12,6 +12,7 @@ import { classifyLLMRequestRejectionMessage, isLLMDeterministicRequestRejection, isLLMImageLimitMessage, + isLLMRequestOversized, normalizeLLMError, } from "@mavis/shared/llm-error-classifier"; import { @@ -605,6 +606,36 @@ describe("withLLMRetry", () => { expect(error?.type === "error" ? error.error.errorMessage : undefined).toBe(message); }); + // A request that is too large (HTTP 413, request_too_large, the local + // pre-send body check, or a context-window overflow) is rejected again on + // every resend; the BYOK retry-all budget used to send it six times. + it.each([ + ["minimax_api", "413 Request Entity Too Large"], + ["minimax_api", '413 {"type":"error","error":{"type":"request_too_large","message":"Request exceeds the maximum allowed number of bytes."}}'], + ["custom_provider:work", "413 Payload Too Large"], + ["custom_provider:work", "413 status code (no body)"], + ["minimax_api", "Request body too large: 70000000 bytes exceeds client limit 67108864 bytes."], + ["minimax_api", "400 invalid params, context window exceeds limit (2013)"], + ["custom_provider:work", '400 {"error":{"message":"prompt is too long: 600000 tokens > 512000 maximum"}}'], + ["minimax", "413 Request Entity Too Large"], + ["minimax", "Request body too large: 70000000 bytes exceeds client limit 67108864 bytes."], + ])("sends an oversized request from %s only once: %s", async (provider, message) => { + let attempts = 0; + const sleep = vi.fn(async () => {}); + const inner = (async () => { + attempts += 1; + return errorStream(message); + }) as StreamFn; + const wrapped = withLLMRetry(inner, retryOptions({ sleep })); + + const events = await collectEvents(await wrapped(fakeModel(provider), CONTEXT, {})); + + expect(attempts).toBe(1); + expect(sleep).not.toHaveBeenCalled(); + const error = events.find((event) => event.type === "error"); + expect(error?.type === "error" ? error.error.errorMessage : undefined).toBe(message); + }); + it.each([ "[Error] 400 Bad Request", "503 Service Unavailable: too many images queued, try again", @@ -1181,4 +1212,42 @@ describe("deterministic request rejection classification", () => { expect(rejection("401 invalid api key")).toBe(false); expect(classifyLLMRequestRejectionMessage("invalid_request_error without a status")).toBeUndefined(); }); + + it("treats HTTP 413, byte-limit and context-overflow rejections as oversized", () => { + const oversized = (errorMessage: string, statusCode?: number) => + isLLMRequestOversized(normalizeLLMError({ errorMessage, statusCode })); + expect(oversized("413 Request Entity Too Large")).toBe(true); + expect(oversized("anything", 413)).toBe(true); + expect(oversized('{"type":"error","error":{"type":"request_too_large","message":"too big"}}')).toBe(true); + expect(oversized("Request body too large: 9 bytes exceeds client limit 8 bytes.")).toBe(true); + expect(oversized("400 invalid params, context window exceeds limit (2013)")).toBe(true); + expect(oversized("prompt is too long: 213462 tokens > 200000 maximum")).toBe(true); + expect(rejection("413 Payload Too Large")).toBe(true); + }); + + it("does not treat transient or unrelated errors as oversized", () => { + const oversized = (errorMessage: string, statusCode?: number) => + isLLMRequestOversized(normalizeLLMError({ errorMessage, statusCode })); + // The bare MiniMax 2013 code is a generic invalid-params code. + expect(oversized("400 invalid params (2013)")).toBe(false); + expect(oversized("400 Bad Request")).toBe(false); + expect(oversized("prompt is too long", 503)).toBe(false); + expect(oversized("429 rate limited: request too large for TPM")).toBe(false); + expect(oversized("Request body too large: timeout while uploading")).toBe(false); + expect(oversized("401 invalid api key")).toBe(false); + }); + + it("classifies the BYOK-attributed form of an oversized request", () => { + const byok = (detail: string) => + `BYOK upstream error: ${JSON.stringify({ + errorCode: LLM_ERROR_CODES.LLM_UPSTREAM_ERROR, + message: `BYOK provider custom_provider:fixture upstream error: ${detail}`, + errorSource: "byok_upstream", + errorDetail: detail, + errorProviderId: "custom_provider:fixture", + })}`; + expect(rejection(byok("413 Payload Too Large"))).toBe(true); + expect(rejection(byok("400 invalid params, context window exceeds limit (2013)"))).toBe(true); + expect(rejection(byok("503 Service Unavailable"))).toBe(false); + }); }); diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts index 6ab3e3db7..e23785b56 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts @@ -238,6 +238,9 @@ function automaticContextCompactionInput( tools: input.tools, thinkingLevel: input.thinkingLevel, signal: input.signal, + ...(input.trigger === 'context_overflow_recovery' + ? { force: 'context_overflow_recovery' as const } + : {}), }; } diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts index 592d20ba3..7b0c6557a 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts @@ -206,6 +206,13 @@ export type AutomaticContextCompactionInput = Readonly< | 'signal' > & { readonly maxSerializedInputBytes?: number; + /** + * The provider rejected the previous request as too large (HTTP 413 or a + * context-window overflow). Compaction starts regardless of the local + * estimate, and the result must be meaningfully smaller than the request + * the provider just rejected, because that estimate evidently said it fit. + */ + readonly force?: 'context_overflow_recovery'; } >; diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.ts index 7911e6339..4fbab0a30 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.ts @@ -875,6 +875,14 @@ function mergeHooks { await input.onHistoryChanged(change); diff --git a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.test.ts b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.test.ts index 6cd430fca..c331732f5 100644 --- a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.test.ts +++ b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.test.ts @@ -988,6 +988,72 @@ describe('AutomaticContextCompactor request boundaries', () => { ).rejects.toThrow(/streamFn/i); }); + it('starts below every trigger when the Provider rejected the request as too large', async () => { + const messages: AgentMessage[] = [ + { role: 'user', content: [{ type: 'text', text: 'small but rejected' }], timestamp: 1 }, + ]; + const model: Model<'anthropic-messages'> = { + id: 'model-1', + name: 'Model 1', + api: 'anthropic-messages', + provider: 'test', + baseUrl: 'https://example.invalid', + reasoning: false, + input: ['text'], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 100_000, + maxTokens: 4_096, + }; + const input = { + sessionId: 'session-1', + phase: 'initial' as const, + messages, + model, + thinkingLevel: 'off' as const, + maxSerializedInputBytes: 16 * 1024 * 1024, + }; + + expect(automaticCompactor().probeBeforeLlm(input)).toMatchObject({ shouldStart: false }); + expect( + automaticCompactor().probeBeforeLlm({ ...input, force: 'context_overflow_recovery' }), + ).toMatchObject({ shouldStart: true }); + // Forced recovery goes on to a checkpoint (which needs the Provider transport). + await expect( + automaticCompactor().compactBeforeLlm({ ...input, force: 'context_overflow_recovery' }), + ).rejects.toThrow(/streamFn/i); + }); + + it('maps the runner overflow-recovery trigger to a forced automatic compaction', async () => { + const probeBeforeLlm = vi.fn(() => ({ shouldStart: false })); + const hook = createAutomaticContextCompactionHook( + { + automatic: { probeBeforeLlm, compactBeforeLlm: vi.fn() }, + lifecycle: { + completeCommittedHistory: vi.fn(async () => undefined), + failCommittedHistory: vi.fn(async () => undefined), + }, + } as never, + { sessionId: 'session-1', turnId: 'turn-1', turnSequence: 1 }, + 'lease-overflow', + ); + const hookInput = { + sessionId: 'session-1', + turnId: 'turn-1', + phase: 'iteration' as const, + messages: [], + canonicalMessages: [], + model: { contextWindow: 100_000, maxTokens: 4_096 } as never, + thinkingLevel: 'off' as const, + }; + await hook(hookInput); + await hook({ ...hookInput, trigger: 'context_overflow_recovery' }); + + expect(probeBeforeLlm.mock.calls.map(([input]) => (input as { force?: string }).force)).toEqual([ + undefined, + 'context_overflow_recovery', + ]); + }); + it('uses the requested Messages API maxTokens at the trigger boundary', async () => { const messages: AgentMessage[] = [ { role: 'user', content: [{ type: 'text', text: 'small context' }], timestamp: 1 }, diff --git a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts index 885483207..a6f925afc 100644 --- a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts +++ b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts @@ -116,7 +116,9 @@ function prepareAutomaticCompaction( sessionId: input.sessionId, messages: input.messages, }); + const forced = input.force === 'context_overflow_recovery'; const normalTriggerMatched = + forced || footprint.inputTokens > automaticTriggerAt || (input.maxSerializedInputBytes !== undefined && footprint.serializedBytes > input.maxSerializedInputBytes); @@ -128,6 +130,31 @@ function prepareAutomaticCompaction( normalTriggerMatched, shouldStart: toolResultCompactionPlan !== undefined || normalTriggerMatched, providerInputLimit, + forcedLimits: forced ? overflowRecoveryLimits(footprint, input.maxSerializedInputBytes) : {}, + }; +} + +/** + * Share of the rejected request a forced overflow-recovery compaction may keep. + * The provider rejected a request the local estimate admitted, so admitting a + * candidate against the configured limits alone could resend the same size. + */ +const OVERFLOW_RECOVERY_RETAINED_SHARE = 0.5; + +function overflowRecoveryLimits( + footprint: { readonly inputTokens: number; readonly serializedBytes: number }, + maxSerializedInputBytes: number | undefined, +): { readonly providerInputLimit?: number; readonly maxSerializedInputBytes?: number } { + const tokens = Math.floor(footprint.inputTokens * OVERFLOW_RECOVERY_RETAINED_SHARE); + const bytes = Math.floor(footprint.serializedBytes * OVERFLOW_RECOVERY_RETAINED_SHARE); + return { + ...(tokens > 0 ? { providerInputLimit: tokens } : {}), + ...(bytes > 0 + ? { + maxSerializedInputBytes: + maxSerializedInputBytes === undefined ? bytes : Math.min(bytes, maxSerializedInputBytes), + } + : {}), }; } @@ -144,7 +171,7 @@ async function runAutomaticPolicy(options: { const { input, hooks, preparation } = options; return compactContext({ history: input.messages, - limits: automaticLimits(input, preparation.providerInputLimit), + limits: automaticLimits(input, preparation.providerInputLimit, preparation.forcedLimits), measurePair: async (pair) => preparation.measurer.measurePair(pair), ...(options.toolResultCompactionCandidate ? { toolResultCompactionCandidate: options.toolResultCompactionCandidate } @@ -165,12 +192,16 @@ async function runAutomaticPolicy(options: { function automaticLimits( input: AutomaticContextCompactionInput, providerInputLimit: number, + forcedLimits: ReturnType, ): CompactContextInput['limits'] { + const maxSerializedInputBytes = + forcedLimits.maxSerializedInputBytes ?? input.maxSerializedInputBytes; return { - providerInputLimit, - ...(input.maxSerializedInputBytes === undefined - ? {} - : { maxSerializedInputBytes: input.maxSerializedInputBytes }), + providerInputLimit: Math.min( + providerInputLimit, + forcedLimits.providerInputLimit ?? providerInputLimit, + ), + ...(maxSerializedInputBytes === undefined ? {} : { maxSerializedInputBytes }), }; } diff --git a/packages/shared/src/llm-error-classifier.ts b/packages/shared/src/llm-error-classifier.ts index 3aa02f9f2..3582dbc75 100644 --- a/packages/shared/src/llm-error-classifier.ts +++ b/packages/shared/src/llm-error-classifier.ts @@ -62,7 +62,9 @@ export type LLMErrorSignal = | 'empty_response' | 'length' | 'image_limit' - | 'invalid_request'; + | 'invalid_request' + | 'request_too_large' + | 'context_overflow'; /** * Model-side safety classifier decline (Messages API `stop_reason: "refusal"`). Providers surface it @@ -153,6 +155,21 @@ const LLM_IMAGE_LIMIT_MESSAGE_RE = /\btoo many images\b|\b(?:max|maximum) (?:number of |of \d+ )?images\b|\bnumber of images\b[^.]{0,40}\bexceed|\bimages? (?:count |limit )?exceed(?:s|ed)?\b|\bat most \d+ images?\b/i; /** OpenAI/Anthropic-style error type for a request the provider will never accept as sent. */ const LLM_INVALID_REQUEST_MESSAGE_RE = /\binvalid_request_error\b/i; +/** + * The request body is over a provider or client byte limit: HTTP 413 bodies + * (`request_too_large`, "Request Entity Too Large", "Payload Too Large") and the + * local pre-send check (`Request body too large: N bytes exceeds client limit`). + * The same body is rejected again on every resend. + */ +const LLM_REQUEST_TOO_LARGE_MESSAGE_RE = + /\brequest_too_large\b|\bREQUEST_BODY_TOO_LARGE\b|\brequest body too large\b|\brequest entity too large\b|\bpayload too large\b/i; +/** + * The prompt is over the model context window. Kept to unambiguous provider + * shapes; MiniMax reports it as `invalid params, context window exceeds limit + * (2013)`, so the bare 2013 code (a generic invalid-params code) is not used. + */ +const LLM_CONTEXT_OVERFLOW_MESSAGE_RE = + /\bprompt is too long\b|\bcontext window exceeds limit\b|\bexceeds the context window\b|\bmaximum context length\b|\binput is too long for requested model\b|\bcontext[_ ]length[_ ]exceeded\b|\bmodel_context_window_exceeded\b/i; const LLM_USAGE_LIMIT_UPSTREAM_STATUS_CODES = [2056, 2067] as const; const LLM_USAGE_LIMIT_UPSTREAM_STATUS_CODE_SET = new Set( LLM_USAGE_LIMIT_UPSTREAM_STATUS_CODES, @@ -201,6 +218,8 @@ export function normalizeLLMError(input: LLMErrorInput): NormalizedLLMError { const boundedVisible = sanitizeMessage(visibleMessage); if (LLM_IMAGE_LIMIT_MESSAGE_RE.test(boundedVisible)) signals.add('image_limit'); if (LLM_INVALID_REQUEST_MESSAGE_RE.test(boundedVisible)) signals.add('invalid_request'); + if (LLM_REQUEST_TOO_LARGE_MESSAGE_RE.test(boundedVisible)) signals.add('request_too_large'); + if (LLM_CONTEXT_OVERFLOW_MESSAGE_RE.test(boundedVisible)) signals.add('context_overflow'); } applyNumericSignals(facts, signals); return { facts, ...(message ? { sanitizedMessage: message } : {}) }; @@ -252,9 +271,30 @@ export function isLLMDeterministicRequestRejection(normalized: NormalizedLLMErro if (facts.signals.has('image_limit')) { return status === undefined || status === 400 || status === 413 || status === 422; } + if (isLLMRequestOversized(normalized)) return true; return status === 400 && facts.signals.has('invalid_request'); } +/** + * The request is too big to be accepted as sent: an HTTP 413, a byte-limit + * rejection (`request_too_large`, the local pre-send body check) or a context + * window overflow. Resending the same request fails again, so it must not be + * retried; the only recovery is to make the request smaller (compaction). + * Anything that also looks transient (timeout, network, 408/429/5xx) is excluded. + */ +export function isLLMRequestOversized(normalized: NormalizedLLMError): boolean { + const { facts } = normalized; + if (facts.explicitAbort || facts.timeout || facts.signals.has('network')) return false; + if (facts.signals.has('tpm_rate_limit')) return false; + const status = effectiveHttpStatus(facts); + if (status !== undefined && (status === 408 || status === 429 || status >= 500)) return false; + if (status === 413) return true; + if (!facts.signals.has('request_too_large') && !facts.signals.has('context_overflow')) { + return false; + } + return status === undefined || status === 400 || status === 413 || status === 422; +} + export function toLLMMetricErrorKind(facts: LLMErrorFacts): LLMMetricErrorKind { try { if (facts.explicitAbort) return 'abort'; @@ -554,6 +594,8 @@ function applyMessageSignals( if (/\bempty.?response/i.test(message)) signals.add('empty_response'); if (LLM_IMAGE_LIMIT_MESSAGE_RE.test(message)) signals.add('image_limit'); if (LLM_INVALID_REQUEST_MESSAGE_RE.test(message)) signals.add('invalid_request'); + if (LLM_REQUEST_TOO_LARGE_MESSAGE_RE.test(message)) signals.add('request_too_large'); + if (LLM_CONTEXT_OVERFLOW_MESSAGE_RE.test(message)) signals.add('context_overflow'); const hasStructuredError = facts.existingProtocolCode !== undefined || effectiveHttpStatus(facts) !== undefined; if (!hasStructuredError && /\b(content.?filter|content.?policy|safety|blocked)\b/i.test(message)) diff --git a/release/public-source.json b/release/public-source.json index 7539f8cd2..0133a5f0e 100644 --- a/release/public-source.json +++ b/release/public-source.json @@ -101,6 +101,7 @@ "packages/agent-core/src/index.ts", "packages/agent-core/src/pi-turn-runner/agent.ts", "packages/agent-core/src/pi-turn-runner/assembly-fingerprint.ts", + "packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts", "packages/agent-core/src/pi-turn-runner/defaults.ts", "packages/agent-core/src/pi-turn-runner/events.ts", "packages/agent-core/src/pi-turn-runner/history.ts", @@ -137,6 +138,7 @@ "packages/agent-core/test/unit/event-bridge/decode-timing.test.ts", "packages/agent-core/test/unit/image-dimensions.test.ts", "packages/agent-core/test/unit/image-fixtures.ts", + "packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts", "packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts", "packages/agent-core/test/unit/pi-turn-runner/llm-stream-timeout.test.ts", "packages/agent-core/test/unit/pi-turn-runner/request-image-limit.test.ts", diff --git a/test/vitest-suites.json b/test/vitest-suites.json index a30ece859..208cb70e1 100644 --- a/test/vitest-suites.json +++ b/test/vitest-suites.json @@ -110,6 +110,7 @@ "packages/tui/test/unit/tui/features/composer/attachments.test.ts", "packages/tui/test/unit/tui/features/settings/status-line-picker.test.ts", "packages/tui/test/unit/tui/widgets/editor/editor-behavior.test.ts", + "packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts", "packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts", "packages/agent-core/test/unit/pi-turn-runner/llm-stream-timeout.test.ts", "packages/agent-core/test/unit/pi-turn-runner/request-image-limit.test.ts", From 681c1516d8a767adca943e4b75850ab79c522c40 Mon Sep 17 00:00:00 2001 From: DanielWalnut <45447813+hetaoBackend@users.noreply.github.com> Date: Fri, 9 Oct 2026 13:55:41 +0800 Subject: [PATCH 2/2] fix: surface the original rejection when overflow recovery throws --- .../context-overflow-recovery.ts | 8 +++++- .../context-overflow-recovery.test.ts | 25 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts b/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts index 38a0af37f..ed09da1d1 100644 --- a/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts +++ b/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts @@ -80,7 +80,13 @@ export function withContextOverflowRecovery( return replay(stream, buffered, iterator); } state.attempted = true; - const messages = await state.recover(options?.signal); + let messages: Message[] | undefined; + try { + messages = await state.recover(options?.signal); + } catch { + // A failing recovery must not mask the provider rejection it tried to fix. + return replay(stream, buffered, iterator); + } if (!messages || options?.signal?.aborted) return replay(stream, buffered, iterator); try { void Promise.resolve(iterator.return?.()).catch(() => undefined); diff --git a/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts b/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts index 78fd1b0ba..4d71f7c3c 100644 --- a/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts +++ b/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts @@ -7,6 +7,10 @@ import type { Context, Model, } from "@earendil-works/pi-ai"; +import { + newContextOverflowRecoveryState, + withContextOverflowRecovery, +} from "../../../src/pi-turn-runner/context-overflow-recovery.js"; import { PiTurnRunner } from "../../../src/pi-turn-runner/pi-turn-runner.js"; import type { PiBeforeLlmCallHook, @@ -265,3 +269,24 @@ describe("context overflow recovery", () => { expect(result.contexts).toHaveLength(1); }); }); + +describe("withContextOverflowRecovery", () => { + it("surfaces the original rejection when recovery itself throws", async () => { + const calls: number[] = []; + const inner = (async () => { + calls.push(1); + return errorStream(OVERSIZED); + }) as unknown as StreamFn; + const state = newContextOverflowRecoveryState(); + state.recover = async () => { + throw new Error("compaction exploded"); + }; + const wrapped = withContextOverflowRecovery(inner, state); + const out = await wrapped(fakeModel(), { messages: [] } as unknown as Context, {}); + const types: string[] = []; + for await (const event of out) types.push(event.type); + expect(types).toContain("error"); + expect(calls).toHaveLength(1); + expect(state.attempted).toBe(true); + }); +});