diff --git a/packages/runtime/src/__tests__/model-adapter.test.ts b/packages/runtime/src/__tests__/model-adapter.test.ts index f0a517c551..1d5616a18e 100644 --- a/packages/runtime/src/__tests__/model-adapter.test.ts +++ b/packages/runtime/src/__tests__/model-adapter.test.ts @@ -1,6 +1,8 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; +import { createJsonErrorResponseHandler } from '@ai-sdk/provider-utils'; import type { SessionEvent } from '@maka/core/events'; +import { z } from 'zod/v4'; import { AsyncEventQueue } from '../async-queue.js'; import { @@ -211,6 +213,245 @@ describe('ModelAdapter stream and error normalization', () => { assert.equal(adapter.mapFinishReason('provider-new-reason'), 'end_turn'); }); + test('classifies provider context-length overflow errors as ContextLength', () => { + const adapter = newAdapter(); + const overflow = (message: string, extra: Record = {}) => + adapter.classifyError(Object.assign(new Error(message), { name: 'AI_APICallError', ...extra })); + + // A representative sample across the providers Maka supports. + assert.equal(overflow('prompt is too long: 213462 tokens > 200000 maximum', { statusCode: 400 }), 'ContextLength'); // Anthropic + assert.equal(overflow('413 request_too_large: Request exceeds the maximum size', { statusCode: 413 }), 'ContextLength'); // Anthropic 413 + assert.equal(overflow('Your input exceeds the context window of this model', { statusCode: 400 }), 'ContextLength'); // OpenAI + assert.equal(overflow("Requested token count exceeds the model's maximum context length of 131072 tokens", { statusCode: 400 }), 'ContextLength'); // LiteLLM + assert.equal(overflow('The input token count (1196265) exceeds the maximum number of tokens allowed (1048575)', { statusCode: 400 }), 'ContextLength'); // Google + assert.equal(overflow("This model's maximum prompt length is 131072 but the request contains 537812 tokens", { statusCode: 400 }), 'ContextLength'); // xAI + assert.equal(overflow('Please reduce the length of the messages or completion', { statusCode: 400 }), 'ContextLength'); // Groq + assert.equal(overflow("This endpoint's maximum context length is 262144 tokens", { statusCode: 400 }), 'ContextLength'); // OpenRouter + assert.equal(overflow('Prompt contains 5000 tokens; too large for model with 4096 maximum context length', { statusCode: 400 }), 'ContextLength'); // Mistral + assert.equal(overflow('invalid params, context window exceeds limit', { statusCode: 400 }), 'ContextLength'); // MiniMax + assert.equal(overflow('Your request exceeded model token limit: 200000 (requested: 260000)', { statusCode: 400 }), 'ContextLength'); // Kimi + assert.equal(overflow('prompt token count of 21000 exceeds the limit of 16384', { statusCode: 400 }), 'ContextLength'); // GitHub Copilot + assert.equal(overflow('the prompt contains too many tokens', { statusCode: 400 }), 'ContextLength'); // generic prompt-overflow wording + + // The classification covers the ORIGINAL error fields, not just the message. + // A real AI SDK APICallError carries the provider's structured error JSON in + // `data` (parsed by createJsonErrorResponseHandler) or `responseBody` — there + // is NO top-level `.code` — so a structured code with a generic HTTP message + // must classify from those fields (review round-7 P1-1). + assert.equal( + overflow('Bad Request', { + statusCode: 400, + data: { error: { message: 'Bad Request', type: 'invalid_request_error', code: 'context_length_exceeded' } }, + }), + 'ContextLength', + ); + // Same provider JSON reachable only through the raw response body. The + // body must be a shape the OpenAI errorSchema genuinely REJECTS (here: + // missing the required error.message), because that is the only way a + // real createJsonErrorResponseHandler leaves `data` absent while keeping + // `responseBody` — a schema-valid body always produces `data` (round-8 P3). + assert.equal( + overflow('Bad Request', { + statusCode: 400, + responseBody: '{"error":{"code":"context_length_exceeded"}}', + }), + 'ContextLength', + ); + // Anthropic puts the structured identifier in data.error.type. + assert.equal( + overflow('Request Entity Too Large', { + statusCode: 413, + data: { type: 'error', error: { type: 'request_too_large', message: 'Request Entity Too Large' } }, + }), + 'ContextLength', + ); + + // Stream error parts are NOT Error instances: each provider enqueues its + // parsed error value as `{type:'error', error}` on the fullStream, and the + // classifier must accept the real shapes (review round-8 P1-1): + // OpenAI Chat emits the INNER error object (openai-chat-language-model.ts:479)… + assert.equal( + adapter.classifyError({ + message: 'Bad Request', + type: 'invalid_request_error', + param: null, + code: 'context_length_exceeded', + }), + 'ContextLength', + ); + // …OpenAI Responses emits the WHOLE error chunk (openai-responses-language-model.ts:2105)… + assert.equal( + adapter.classifyError({ + type: 'error', + sequence_number: 3, + error: { type: 'invalid_request_error', code: 'context_length_exceeded', message: 'Bad Request', param: null }, + }), + 'ContextLength', + ); + // …Anthropic emits the inner {type, message} object (anthropic-messages-language-model.ts:2441)… + assert.equal( + adapter.classifyError({ type: 'invalid_request_error', message: 'prompt is too long: 213462 tokens > 200000 maximum' }), + 'ContextLength', + ); + assert.equal( + adapter.classifyError({ type: 'request_too_large', message: 'Request exceeds the maximum size' }), + 'ContextLength', + ); + // …and openai-compatible emits a bare message STRING (openai-compatible-chat-language-model.ts:466). + assert.equal( + adapter.classifyError("Requested token count exceeds the model's maximum context length of 131072 tokens."), + 'ContextLength', + ); + // Non-overflow object/string errors do not become ContextLength. + assert.equal( + adapter.classifyError({ type: 'invalid_request_error', message: 'missing required field' }), + 'Other', + ); + + // Specific overflow evidence outranks a generic 5xx (review round-8 P1-2): + // LiteLLM-style proxies surface a provider overflow through a 503 wrapper, + // both as a structured code and as message text (pi overflow fixture). + assert.equal( + overflow('Service Unavailable', { + statusCode: 503, + data: { error: { message: 'Service Unavailable', code: 'context_length_exceeded' } }, + }), + 'ContextLength', + ); + assert.equal( + overflow( + "503 litellm.ServiceUnavailableError: litellm.MidStreamFallbackError: litellm.APIConnectionError: APIConnectionError: OpenAIException - Requested token count exceeds the model's maximum context length of 131072 tokens.", + { statusCode: 503 }, + ), + 'ContextLength', + ); + // A bare 413 with no body is itself input-side evidence: HTTP request + // entity too large (Cerebras returns exactly this — review round-8 P1-3). + assert.equal(overflow('Request Entity Too Large', { statusCode: 413 }), 'ContextLength'); + assert.equal(overflow('Payload Too Large', { statusCode: 413 }), 'ContextLength'); + assert.equal(overflow('', { statusCode: 413 }), 'ContextLength'); + // A structured code embedded in free text must not be misread by a weaker + // substring heuristic checked earlier: "generate" contains "rate", and the + // rate/auth substring heuristics rank BELOW overflow evidence (round-7 P1-2). + assert.equal( + overflow('Failed to generate response: context_length_exceeded', { statusCode: 400 }), + 'ContextLength', + ); + // Explicit numeric statuses still outrank every text heuristic: a 5xx that + // happens to mention rate stays ProviderUnavailable. + assert.equal( + overflow('Please rate limit your requests', { statusCode: 503 }), + 'ProviderUnavailable', + ); + // The weak rate heuristic is word-shaped, not a substring: "generate" and + // "separate" are not rate limits (review round-8 P2)… + assert.notEqual(overflow('Failed to generate response', { statusCode: 400 }), 'RateLimit'); + assert.notEqual(overflow('Unable to separate response chunks', { statusCode: 400 }), 'RateLimit'); + // …while genuine rate wording without an explicit 429 still classifies. + assert.equal(overflow('Please rate limit your requests', {}), 'RateLimit'); + assert.equal(overflow('rate_limit_exceeded: slow down', {}), 'RateLimit'); + + // Exclusion-first: throttling/rate-limit wording must NOT be read as overflow + // even when it superficially mentions tokens. + assert.equal(overflow('Rate limit reached: too many tokens, please wait before trying again', { statusCode: 429 }), 'RateLimit'); + assert.notEqual(overflow('ThrottlingException: too many tokens, please wait before trying again', { statusCode: 400 }), 'ContextLength'); + // Unrelated 400s stay in their own buckets, never ContextLength: a token-free + // size limit and an output-parameter error merely mention limits/tokens, and + // misreading either would run (and persist) a pointless compaction + retry. + assert.notEqual(overflow('invalid request: missing required field', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('file size exceeds the limit of 10485760', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('max_tokens is too many tokens for this model', { statusCode: 400 }), 'ContextLength'); + // An OUTPUT token cap is not an input overflow: compacting the history + // cannot fix it, so it must never trigger a persisted compaction retry. + assert.notEqual(overflow('Output token limit exceeded', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('Maximum output token limit exceeded', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('output token count of 8192 exceeds the limit of 4096', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('completion token count of 8192 exceeds the limit of 4096', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('max output token count of 8192 exceeds the limit of 4096', { statusCode: 400 }), 'ContextLength'); + // A generic prefix must not smuggle an output cap past the input-subject + // constraints ("request" in "Invalid request:" is not the token subject): + // output caps are excluded at the exclusion-first owner, wording-wide. + assert.notEqual(overflow('Invalid request: output token count of 8192 exceeds the limit of 4096', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('Invalid request: completion token count of 8192 exceeds the limit of 4096', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('Invalid request: max output token count of 8192 exceeds the limit of 4096', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('Invalid request: max_tokens is too many tokens for this model', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('Invalid request: Maximum output token limit exceeded', { statusCode: 400 }), 'ContextLength'); + // Complete output-cap RELATIONS are excluded even when reworded — the + // veto is not a fixed word order. + assert.notEqual(overflow('Invalid request: completion has too many tokens for this model', { statusCode: 400 }), 'ContextLength'); + assert.notEqual(overflow('Invalid request: max_tokens token limit exceeded', { statusCode: 400 }), 'ContextLength'); + // ...including the passive voice, where the output subject FOLLOWS the + // token predicate (review round-7 P1-3). + assert.notEqual(overflow('Invalid input: too many tokens were requested for the completion', { statusCode: 400 }), 'ContextLength'); + // ...and the embedded-role permutation, where the output word sits INSIDE + // the token phrase — even when a capacity statement follows in the same + // message (review round-8 P1-4). + assert.notEqual( + overflow("Too many completion tokens were requested. This endpoint's maximum context length is 262144 tokens.", { statusCode: 400 }), + 'ContextLength', + ); + assert.notEqual(overflow('Too many output tokens requested for this model', { statusCode: 400 }), 'ContextLength'); + assert.notEqual( + overflow("Maximum completion tokens exceeded. This endpoint's maximum context length is 262144 tokens.", { statusCode: 400 }), + 'ContextLength', + ); + // A bare capacity STATEMENT inside an unrelated error is not an overflow + // relation: throttle/quota wording vetoes every free-text signal — only a + // structured provider code is unconditional (review round-7 P1-4). + assert.notEqual( + overflow("ThrottlingException: quota exceeded. This endpoint's maximum context length is 262144 tokens.", { statusCode: 400 }), + 'ContextLength', + ); + // ...while the input-side form of the same wording still classifies. + assert.equal(overflow('Input token limit exceeded: 250000 tokens > 200000 maximum', { statusCode: 400 }), 'ContextLength'); + // The output-cap exclusions stay adjacency-tight: OpenAI's classic input + // overflow mentions the completion and max_tokens without being an output + // cap, and must keep classifying. + assert.equal(overflow("This model's maximum context length is 8192 tokens. However, you requested 10240 tokens (10140 in the messages, 100 in the completion). Please reduce the length of the messages or completion.", { statusCode: 400 }), 'ContextLength'); + assert.equal(overflow("This model's maximum context length is 8192 tokens. However, you requested 10240 tokens (10140 in the messages, 100 in max_tokens). Please reduce the length of the messages or completion.", { statusCode: 400 }), 'ContextLength'); + // Structured provider evidence is the ONLY unconditional signal: a genuine + // input overflow may word its message as an output-cap relation the text + // vetoes would reject, and the context_length_exceeded code must still win. + assert.equal( + overflow('Invalid request: completion has too many tokens for this model', { + statusCode: 400, + data: { error: { message: 'Invalid request: completion has too many tokens for this model', code: 'context_length_exceeded' } }, + }), + 'ContextLength', + ); + assert.equal(adapter.classifyError(Object.assign(new Error('401 Authorization'), { statusCode: 401 })), 'Auth'); + }); + + test('classifies overflow wording that only survives in a schema-invalid responseBody (review round-9 P2)', async () => { + const adapter = newAdapter(); + // The REAL failed-response handler, with the OpenAI-family error schema + // (error must be an OBJECT with a message). When the provider body does + // not match — `{error: string}` genuinely exists among OpenAI-compatible + // providers — the handler degrades `message` to the statusText and keeps + // the provider's wording ONLY in `responseBody`. + const handler = createJsonErrorResponseHandler({ + errorSchema: z.object({ error: z.object({ message: z.string() }) }), + errorToMessage: (data) => data.error.message, + }); + const errorFromBody = async (body: string) => (await handler({ + response: new Response(body, { status: 400, statusText: 'Bad Request' }), + url: 'https://api.example.test/v1/chat/completions', + requestBodyValues: {}, + })).value; + + const overflowError = await errorFromBody('{"error":"Your input exceeds the context window of this model"}'); + // Prove the degradation is real before asserting on classification. + assert.equal(overflowError.message, 'Bad Request'); + assert.equal(overflowError.data, undefined); + assert.equal(adapter.classifyError(overflowError), 'ContextLength'); + // The veto layer runs on the same full text: an output-cap relation in the + // body must not classify even with a capacity statement next to it. + const outputCapError = await errorFromBody( + '{"error":"Too many completion tokens were requested. This endpoint\'s maximum context length is 262144 tokens."}', + ); + assert.notEqual(adapter.classifyError(outputCapError), 'ContextLength'); + }); + test('normalizes cache and reasoning usage variants in the adapter module', () => { assert.deepEqual( normalizeAiSdkUsage({ diff --git a/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts b/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts new file mode 100644 index 0000000000..c35f0c6993 --- /dev/null +++ b/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts @@ -0,0 +1,699 @@ +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import { setImmediate as flushMacrotask } from 'node:timers/promises'; +import { MockLanguageModelV3, simulateReadableStream } from 'ai/test'; +import type { LanguageModelV3StreamPart } from '@ai-sdk/provider'; +import type { LlmConnection, SessionHeader } from '@maka/core'; +import type { SessionEvent } from '@maka/core/events'; +import type { RuntimeEvent } from '@maka/core/runtime-event'; +import { z } from 'zod'; +import { AiSdkBackend } from '../ai-sdk-backend.js'; +import { createSessionEventMapMemory, mapSessionEventToRuntimeEvent } from '../ai-sdk-flow.js'; +import type { InvocationContext } from '../invocation-context.js'; +import { PermissionEngine } from '../permission-engine.js'; +import type { HistoryCompactCheckpoint } from '../history-compact-checkpoint.js'; + +const RAW_SPAN_ONE = 'RAW_SPAN_ONE_'.repeat(24); +const ANCHOR_TEXT = 'reactive overflow recovery keep my exact words'; +const OVERFLOW_MESSAGE = 'prompt is too long: 213462 tokens > 200000 maximum'; + +/** + * Per-provider-request script. Each entry drives one `doStream` invocation: + * - 'tool' → a Read tool call (completes a step, appends a durable pair) + * - 'bigtool' → assistant text (sentinel) + a Read with a huge result, so the + * proactive capacity trigger fires at this step's boundary + * - 'bigread' → a pure Read with a huge result and NO step text, so the + * durable pair is the trailing span a recovery fold must keep + * verbatim in the tail (the prune-resurrection shape) + * - 'load' → a `load_tools` call activating the gated 'big' group + * - 'gated' → a call to the gated `Big` tool + * - 'done' → final assistant text, finish stop + * - 'overflow' → the provider rejects with a context-length 400 (doStream + * throws; the SDK surfaces it as a fullStream error chunk and + * rejects finishReason — the fake-end_turn latent-bug path) + * - 'overflowPart' → OpenAI CHAT in-stream failure exactly as the locked + * provider transform produces it: the error part value is the + * INNER parsed error object (never an Error instance) and the + * flush trailer is a finish part with finishReason 'error' + * (openai-chat-language-model.ts:478-479 + flush) — the + * round-8 P1-1 end-to-end shape + * - 'overflowPartResponses' → OpenAI RESPONSES in-stream failure: the error + * part value is the WHOLE {type:'error', error:{...}} chunk + * and the flush trailer keeps finishReason 'other' (the + * isErrorChunk branch never reassigns it) — locks recovery + * against per-family trailer drift + * - 'error500' → a non-overflow provider failure (never a recovery trigger) + */ +type CallKind = + | 'tool' | 'bigtool' | 'bigread' | 'load' | 'gated' | 'done' + | 'overflow' | 'overflowPart' | 'overflowPartResponses' | 'error500'; + +const RETRY_STEP_TEXT_SENTINEL = 'RETRY_STEP_TEXT_SENTINEL reasoning before the big read'; +const BIG_RESULT = 'BIG_RESULT_'.repeat(200); + +interface ReactiveFixtureOptions { + script: CallKind[]; + contextWindow?: number; + reserveTokens?: number; + midTurnEnabled?: boolean; + withoutPriorTurns?: boolean; + bigPriors?: boolean; + summarize?: () => Promise | string | undefined; + /** Explicit send-level step budget forwarded to the backend. */ + maxSteps?: number; + /** The FIRST tool step reports an unusable usage object (no token counts). */ + firstStepUsageMissing?: boolean; + /** Economy tool availability with the gated `Big` tool behind `load_tools`. */ + gatedToolGroup?: boolean; + /** + * appendMessage yields several macrotasks before resolving, so the pump + * genuinely lags inside flushStep (text_complete not yet enqueued, + * flushedSteps not yet incremented) while the SDK's loop advances to the + * next prepareStep — the P1-A race window. + */ + slowAppendMessage?: boolean; + /** Enable the active tool-result prune with a small threshold + archive seam. */ + activeToolResultPrune?: boolean; +} + +interface ReactiveLlmCall { + status?: string; + errorClass?: string; + inputTokens?: number; + outputTokens?: number; + totalTokens?: number; +} + +interface ReactiveFixture { + backend: AiSdkBackend; + model: MockLanguageModelV3; + recorded: HistoryCompactCheckpoint[]; + toolExecutions: string[]; + summarizerCalls: () => number; + anchor: RuntimeEvent; + priorEvents: RuntimeEvent[]; + events: SessionEvent[]; + llmCalls: ReactiveLlmCall[]; + /** JSON of each summarizer call's folded runtime events (coverage evidence). */ + summarizedSources: string[]; + persist: (event: SessionEvent) => void; +} + +function buildReactiveFixture(options: ReactiveFixtureOptions): ReactiveFixture { + const contextWindow = options.contextWindow ?? 200_000; + const reserveTokens = options.reserveTokens ?? 1_000; + const recorded: HistoryCompactCheckpoint[] = []; + const toolExecutions: string[] = []; + const events: SessionEvent[] = []; + const llmCalls: ReactiveLlmCall[] = []; + const counters = { summarizerCalls: 0 }; + const usage = (input: number, output: number) => ({ + inputTokens: { total: input, noCache: input, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: output, text: output, reasoning: 0 }, + }); + const toolCallChunks = ( + call: number, + toolName: string, + args: object, + leadingText?: string, + ): LanguageModelV3StreamPart[] => [ + { type: 'stream-start', warnings: [] }, + ...(leadingText + ? ([ + { type: 'text-start', id: `step-text-${call}` }, + { type: 'text-delta', id: `step-text-${call}`, delta: leadingText }, + { type: 'text-end', id: `step-text-${call}` }, + ] satisfies LanguageModelV3StreamPart[]) + : []), + { type: 'tool-call', toolCallId: `tool-${call}`, toolName, input: JSON.stringify(args) }, + { + type: 'finish', + finishReason: { unified: 'tool-calls', raw: 'tool_calls' }, + // An unusable first-step usage: the SDK accepts the object but the + // adapter's normalization fails closed (undefined), the #972 shape. + usage: options.firstStepUsageMissing && call === 1 + ? ({ inputTokens: {}, outputTokens: {} } as ReturnType) + : usage(100, 20), + }, + ]; + const doneChunks = (): LanguageModelV3StreamPart[] => [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'done' }, + { type: 'text-end', id: 'text-1' }, + { type: 'finish', finishReason: { unified: 'stop', raw: 'stop' }, usage: usage(120, 10) }, + ]; + const streamForCall = (call: number): ReadableStream => { + const kind = options.script[call - 1]; + if (kind === 'overflow') { + throw Object.assign(new Error(OVERFLOW_MESSAGE), { name: 'AI_APICallError', statusCode: 400 }); + } + if (kind === 'error500') { + throw Object.assign(new Error('internal server error'), { name: 'AI_APICallError', statusCode: 500 }); + } + if (kind === 'overflowPart' || kind === 'overflowPartResponses') { + // The 200 response starts streaming, then the provider sends the error + // inside the SSE stream. Each shape below is exactly what the locked + // provider transform enqueues — no cross-family mixing: + // Chat forwards the INNER error object and its flush emits a finish + // part with finishReason 'error'; Responses forwards the WHOLE error + // chunk and its flush keeps the initial finishReason 'other'. + const errorValue = kind === 'overflowPart' + ? { message: 'Bad Request', type: 'invalid_request_error', param: null, code: 'context_length_exceeded' } + : { + type: 'error', + sequence_number: 1, + error: { type: 'invalid_request_error', code: 'context_length_exceeded', message: 'Bad Request', param: null }, + }; + const trailerReason = kind === 'overflowPart' + ? { unified: 'error' as const, raw: undefined } + : { unified: 'other' as const, raw: undefined }; + return simulateReadableStream({ + chunks: [ + { type: 'stream-start', warnings: [] }, + { type: 'error', error: errorValue }, + { type: 'finish', finishReason: trailerReason, usage: usage(0, 0) }, + ] satisfies LanguageModelV3StreamPart[], + initialDelayInMs: null, + chunkDelayInMs: null, + }); + } + const chunks = + kind === 'tool' ? toolCallChunks(call, 'Read', { path: 'one.md' }) + : kind === 'bigtool' ? toolCallChunks(call, 'Read', { path: 'big.md' }, RETRY_STEP_TEXT_SENTINEL) + : kind === 'bigread' ? toolCallChunks(call, 'Read', { path: 'big.md' }) + : kind === 'load' ? toolCallChunks(call, 'load_tools', { group: 'big' }) + : kind === 'gated' ? toolCallChunks(call, 'Big', { q: 'run' }) + : doneChunks(); + return simulateReadableStream({ chunks, initialDelayInMs: null, chunkDelayInMs: null }); + }; + const model = new MockLanguageModelV3({ + doStream: async ( + streamOptions: { abortSignal?: AbortSignal }, + ): Promise<{ stream: ReadableStream }> => { + if (streamOptions.abortSignal?.aborted) { + throw Object.assign(new Error('aborted'), { name: 'AbortError' }); + } + return { stream: streamForCall(model.doStreamCalls.length) }; + }, + }); + + const priorChars = options.bigPriors ? 4_000 : 120; + const priorEvents: RuntimeEvent[] = options.withoutPriorTurns ? [] : [ + runtimeTextEvent('prior-user', 'turn-0', 'user', `PRIOR_FACT question ${'p'.repeat(priorChars)}`), + runtimeTextEvent('prior-model', 'turn-0', 'model', `PRIOR_FACT answer ${'q'.repeat(priorChars)}`), + ]; + const anchor = runtimeTextEvent('anchor-1', 'turn-1', 'user', ANCHOR_TEXT); + + const ledger: RuntimeEvent[] = [anchor]; + const ledgerCtx: InvocationContext = { + sessionId: 'session-1', + invocationId: 'run-1', + runId: 'run-1', + turnId: 'turn-1', + source: 'desktop', + startedAt: 1, + request: { sessionId: 'session-1', turnId: 'turn-1', text: ANCHOR_TEXT, source: 'desktop' }, + newId: idGenerator(), + now: monotonicClock(), + }; + const ledgerMemory = createSessionEventMapMemory(); + const persist = (event: SessionEvent): void => { + const mapped = mapSessionEventToRuntimeEvent(event, ledgerCtx, ledgerMemory); + if (mapped.partial === true) return; + if (mapped.content?.kind === 'error') return; + ledger.push(mapped); + }; + + const midTurnEnabled = options.midTurnEnabled ?? true; + const summarizedSources: string[] = []; + const seams = midTurnEnabled + ? { + summarizeHistoryCompact: async (input: { source: { foldedRuntimeEvents: RuntimeEvent[] } }) => { + counters.summarizerCalls += 1; + summarizedSources.push(JSON.stringify(input.source.foldedRuntimeEvents)); + return options.summarize ? await options.summarize() : 'REACTIVE_SUMMARY_SENTINEL'; + }, + recordHistoryCompactCheckpoint: (checkpoint: HistoryCompactCheckpoint) => { recorded.push(checkpoint); }, + loadTurnRuntimeEvents: async (turnId: string) => { + await flushMacrotask(); + return ledger.filter((event) => event.turnId === turnId); + }, + } + : {}; + + const backend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => { + if (!options.slowAppendMessage) return; + for (let i = 0; i < 5; i += 1) await flushMacrotask(); + }, + connection: { ...connection(), models: [{ id: 'mock-model-id', contextWindow }] }, + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => model, + ...(options.maxSteps !== undefined ? { maxSteps: options.maxSteps } : {}), + tools: [ + { + name: 'Read', + description: 'Read description', + parameters: z.object({ path: z.string() }), + permissionRequired: false, + impl: async (args: { path: string }) => { + toolExecutions.push(args.path); + return { body: args.path === 'big.md' ? BIG_RESULT : RAW_SPAN_ONE }; + }, + }, + ...(options.gatedToolGroup + ? [{ + name: 'Big', + description: 'Gated capability behind the big group', + parameters: z.object({ q: z.string() }), + permissionRequired: false, + impl: async () => { + toolExecutions.push('BIG_EXEC'); + return { ok: true }; + }, + }] + : []), + ], + ...(options.gatedToolGroup + ? { toolAvailability: { economy: true, groups: [{ id: 'big', toolNames: ['Big'] }] } } + : {}), + contextBudget: { + name: 'reactive-test', + maxHistoryEstimatedTokens: 100_000, + minRecentTurns: 1, + historyCompact: { + enabled: true, + mode: 'read_write', + ...(midTurnEnabled ? { midTurn: { enabled: true, reserveTokens } } : {}), + }, + ...(options.activeToolResultPrune + ? { activeToolResultPrune: { enabled: true, maxCurrentResultEstimatedTokens: 100 } } + : {}), + }, + ...(options.activeToolResultPrune + ? { archiveToolResult: () => ({ artifactId: 'artifact-archived-1' }) } + : {}), + ...seams, + recordLlmCall: (record) => { llmCalls.push(record as (typeof llmCalls)[number]); }, + newId: idGenerator(), + now: monotonicClock(), + }); + + return { + backend, + model, + recorded, + toolExecutions, + summarizerCalls: () => counters.summarizerCalls, + anchor, + priorEvents, + events, + llmCalls, + summarizedSources, + persist, + }; +} + +async function runTurn(fixture: ReactiveFixture, consumer: 'immediate' | 'slow' = 'immediate'): Promise { + for await (const event of fixture.backend.send({ + runId: 'run-1', + turnId: 'turn-1', + headAnchorRuntimeEvent: fixture.anchor, + text: ANCHOR_TEXT, + context: [], + runtimeContext: [...fixture.priorEvents], + })) { + if (consumer === 'slow') { + // Scheduling perturbation (same as the mid-turn suite): hold the durable + // write back across several macrotasks so the ledger genuinely lags the + // SDK's step progression. + await flushMacrotask(); + await flushMacrotask(); + await flushMacrotask(); + } + // The consumer persists every non-partial event to the durable ledger, + // exactly like AgentRun, so the reactive compaction pool can span the + // completed steps. + fixture.persist(event); + fixture.events.push(event); + } +} + +function complete(fixture: ReactiveFixture): Extract | undefined { + return fixture.events.find((event) => event.type === 'complete') as + | Extract + | undefined; +} + +describe('reactive overflow recovery in the streaming backend', () => { + test('a request-level context-length 400 ends as a real error, never a fake end_turn', async () => { + // The latent bug: a provider that rejects the request (doStream throws) is + // surfaced as a fullStream error chunk while finishReason rejects. The old + // path caught that rejection as `stop` and emitted a CompleteEvent with + // end_turn plus success telemetry — a silent fabrication. Without the + // mid-turn seam there is nothing to recover, so the honest terminal is a + // real error carrying the provider's classification. + const fixture = buildReactiveFixture({ script: ['overflow'], midTurnEnabled: false }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 1); + // Real error terminal, never a fabricated end_turn success. + assert.equal(complete(fixture)?.stopReason, 'error'); + // A first-class error event carrying the overflow classification. + const errorEvent = fixture.events.find((event) => event.type === 'error') as + | Extract + | undefined; + assert.equal(errorEvent !== undefined, true); + assert.equal(errorEvent?.reason, 'context_overflow'); + // The old fake-success path emitted end_turn with success telemetry; the + // fixed path never records this dead request as a successful call. + assert.equal(fixture.llmCalls.some((call) => call.status === 'success'), false); + }); + + test('a non-overflow provider failure ends as a real error without any recovery attempt', async () => { + const fixture = buildReactiveFixture({ script: ['error500'], bigPriors: true }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 1); + assert.equal(complete(fixture)?.stopReason, 'error'); + assert.equal(fixture.events.some((event) => event.type === 'error'), true); + // Not a context-length error → no compaction, no retry. + assert.equal(fixture.recorded.length, 0); + assert.equal(fixture.summarizerCalls(), 0); + }); + + test('compacts once and retries after a mid-stream context-length overflow', async () => { + // A tool step completes, then the provider rejects the second request with + // a context-length 400 even though our proactive estimate stayed under the + // window. Reactive recovery folds a safe completed prefix into a durable + // mid_turn checkpoint and resends once; the retry succeeds and the turn + // completes normally on the compacted projection. + const fixture = buildReactiveFixture({ script: ['tool', 'overflow', 'done'], bigPriors: true }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 3); + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + assert.equal(fixture.events.some((event) => event.type === 'error'), false); + // Exactly one recovery compaction happened, tagged as an overflow trigger. + assert.equal(fixture.recorded.length, 1); + assert.equal(fixture.recorded[0]!.phase, 'mid_turn'); + assert.equal(fixture.summarizerCalls(), 1); + // The completed tool step was not re-executed on the retry. + assert.deepEqual(fixture.toolExecutions, ['one.md']); + // Send-level usage owner (review P1-2): the terminal record carries BOTH + // attempts' completed steps — the first attempt's tool step (100/20) plus + // the retry's final step (120/10) — not just the last attempt's totalUsage. + const lastCall = fixture.llmCalls.at(-1); + assert.equal(lastCall?.status, 'success'); + assert.equal(lastCall?.inputTokens, 220); + assert.equal(lastCall?.outputTokens, 30); + assert.equal(lastCall?.totalTokens, 250); + }); + + test('recovers from a plain-object in-stream error part, not just Error instances (review round-8 P1-1)', async () => { + // Providers deliver in-stream failures as parsed plain objects (or bare + // strings), never Error instances. The recovery decision must classify + // the real shape; an instanceof Error gate silently downgraded every + // genuine in-stream overflow to an unrecoverable terminal error. + const fixture = buildReactiveFixture({ script: ['tool', 'overflowPart', 'done'], bigPriors: true }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 3); + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + assert.equal(fixture.events.some((event) => event.type === 'error'), false); + assert.equal(fixture.recorded.length, 1); + assert.equal(fixture.recorded[0]!.phase, 'mid_turn'); + }); + + test('recovers from the Responses-family in-stream error shape with its non-error finish trailer (review round-9 P3)', async () => { + // Same failure, different family: the error part value is the WHOLE + // {type:'error', error:{...}} chunk and the trailer finish keeps + // finishReason 'other'. The recovery decision truncates at the error part, + // so trailer drift across provider families must not change the outcome. + const fixture = buildReactiveFixture({ script: ['tool', 'overflowPartResponses', 'done'], bigPriors: true }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 3); + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + assert.equal(fixture.events.some((event) => event.type === 'error'), false); + assert.equal(fixture.recorded.length, 1); + assert.equal(fixture.recorded[0]!.phase, 'mid_turn'); + }); + + test('the recovery baseline is the request the provider rejected, not the attempt-initial messages', async () => { + // Review P1-1 repro: four completed tool steps grow the provider-visible + // request far beyond the attempt's INITIAL messages. The fold shrinks the + // real rejected request but is larger than that initial request, so a + // baseline anchored to the initial messages refuses it as + // replacement_not_smaller and the turn dies on the exact scenario reactive + // recovery exists for — same-turn tool growth. The unique baseline owner + // is the verdict owner's per-request payload measure of the request that + // actually went out. + const fixture = buildReactiveFixture({ script: ['tool', 'tool', 'tool', 'tool', 'overflow', 'done'] }); + await runTurn(fixture); + + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + assert.equal(fixture.events.some((event) => event.type === 'error'), false); + assert.equal(fixture.recorded.length, 1); + assert.equal(fixture.model.doStreamCalls.length, 6); + // The four completed tool steps ran exactly once each. + assert.deepEqual(fixture.toolExecutions, ['one.md', 'one.md', 'one.md', 'one.md']); + }); + + test('an unusable first-attempt step usage fails the whole record closed even when the retry succeeds', async () => { + // Review P1-2, fail-closed direction: the first attempt's completed step + // has an unusable usage sample. The retry's totalUsage is valid but covers + // only the retry, so recording it as the whole send would fabricate a + // partial cost as complete (#972). No record at all is the truthful + // outcome; the turn itself still completes. + const fixture = buildReactiveFixture({ + script: ['tool', 'overflow', 'done'], + bigPriors: true, + firstStepUsageMissing: true, + }); + await runTurn(fixture); + + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + assert.equal(fixture.llmCalls.length, 0); + }); + + test('a retry only gets the remaining step budget under an explicit maxSteps (review P1-3)', async () => { + // maxSteps=2: one completed tool step before the overflow leaves a budget + // of exactly one step for the retry. The retry's tool step consumes it and + // the send ends at the explicit step limit — a fresh full budget would run + // a third step and a fourth provider request, breaching the send-level cap + // and its tool side effects. + const fixture = buildReactiveFixture({ + script: ['tool', 'overflow', 'tool', 'done'], + bigPriors: true, + maxSteps: 2, + }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 3); + assert.deepEqual(fixture.toolExecutions, ['one.md', 'one.md']); + assert.equal(complete(fixture)?.stopReason, 'step_limit'); + }); + + test('a completed retry step\'s assistant text is never dropped by a post-retry compaction (review P1-A)', async () => { + // Review round-2 P1-A repro: the SDK numbers prepareStep steps per + // streamText call, but flushedSteps / replacedStepNumber / lastShapeFailure + // are SEND-level. After a retry, attempt-local step 1 satisfies the + // durability wait with attempt 1's flushed boundary, so a capacity + // compaction at the retry's own step boundary can read the ledger BEFORE + // the pump has flushed the retry step's text_complete — and because the + // replacement projection replaces the whole message list, that streamed + // assistant text silently vanishes from both the covered span and the + // preserved tail. Same shape as PR 1's finding B, re-opened across the + // attempt boundary. The lag lever is a slow appendMessage: the pump gets + // stuck INSIDE flushStep (text_complete not yet enqueued, flushedSteps not + // yet incremented) while the immediate consumer has drained everything + // already pushed — so `consumed >= pushed` holds and only the send-global + // flushedSteps bound can still hold the ledger read back. + const fixture = buildReactiveFixture({ + // High water 400: the first attempt's boundary (~usage 100 + small + // delta) stays under it, the retry step's huge result (~BIG_RESULT/4) + // crosses it, so the capacity trigger fires exactly at the retry's own + // step boundary. + script: ['tool', 'overflow', 'bigtool', 'done'], + bigPriors: true, + contextWindow: 2_000, + reserveTokens: 1_600, + slowAppendMessage: true, + }); + await runTurn(fixture); + + // The turn completes on the compacted projections (recovery fold + the + // post-retry capacity fold), with the retry's step text streamed out. + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + assert.equal( + fixture.events.some((event) => event.type === 'text_complete' && event.text.includes('RETRY_STEP_TEXT_SENTINEL')), + true, + ); + // The projection accounts for that text: it survives either verbatim in + // the final request or inside a summarized covered span — never silently + // dropped from both. + const finalPrompt = JSON.stringify(fixture.model.doStreamCalls.at(-1)?.prompt); + const inTail = finalPrompt.includes('RETRY_STEP_TEXT_SENTINEL'); + const inCoveredSpan = fixture.summarizedSources.join('\n').includes('RETRY_STEP_TEXT_SENTINEL'); + assert.equal(inTail || inCoveredSpan, true); + }); + + test('a same-turn load_tools activation survives the retry (review P1-B)', async () => { + // Review round-2 P1-B repro: active tools were re-derived per streamText + // call from seed groups + that call's own steps. The retry's steps start + // empty, so a group loaded before the overflow was silently revoked — the + // gated tool disappeared from the provider request and the execute + // boundary rejected it. Activation must accumulate monotonically in the + // availability owner for the whole send. + const fixture = buildReactiveFixture({ + script: ['load', 'overflow', 'gated', 'done'], + bigPriors: true, + gatedToolGroup: true, + }); + await runTurn(fixture); + + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + assert.equal(fixture.events.some((event) => event.type === 'error'), false); + // The retry request still advertises the gated tool... + const retryRequestTools = JSON.stringify(fixture.model.doStreamCalls[2]?.tools ?? []); + assert.equal(retryRequestTools.includes('"Big"') || retryRequestTools.includes("'Big'"), true); + // ...and it executes for real after the retry. + assert.equal(fixture.toolExecutions.includes('BIG_EXEC'), true); + }); + + test('an actively pruned tool result stays a placeholder in the retry request (review round-3 P1)', async () => { + // Review round-3 P1 repro: the active tool-result prune derives its + // eligible tool-call IDs from `options.steps` and early-returns on an + // empty set. The retry's fresh streamText starts with empty steps, while + // the recovery projection is rebuilt from the durable ledger — which holds + // the ORIGINAL raw result, not the provider-only placeholder. The retry + // request therefore resurrected the archived raw body, breaking the + // active-prune invariant (an archived result never re-enters provider + // context) and inviting a second overflow. Third instance of the same + // disease: attempt-local `steps` consumed as send-level state. + // + // 'bigread' keeps the step text-free so the durable pair is the POOL'S + // trailing span: the safe boundary cannot split the pair, retreats before + // the call, and the fold re-materializes the pair verbatim in the tail — + // from the ledger, which holds the raw body, not the placeholder. + const fixture = buildReactiveFixture({ + script: ['bigread', 'overflow', 'done'], + bigPriors: true, + activeToolResultPrune: true, + }); + await runTurn(fixture); + + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + // The rejected request had already pruned the big result to a placeholder. + const overflowPrompt = JSON.stringify(fixture.model.doStreamCalls[1]?.prompt); + assert.equal(overflowPrompt.includes('BIG_RESULT_'), false); + assert.equal(overflowPrompt.includes('artifact-archived-1'), true); + // The retry request must keep the placeholder — never the raw body. + const retryPrompt = JSON.stringify(fixture.model.doStreamCalls[2]?.prompt); + assert.equal(retryPrompt.includes('BIG_RESULT_'), false); + assert.equal(retryPrompt.includes('artifact-archived-1'), true); + }); + + test('a second overflow after the single retry ends as a real error', async () => { + const fixture = buildReactiveFixture({ script: ['tool', 'overflow', 'overflow'], bigPriors: true }); + await runTurn(fixture); + + // The latch permits exactly one compact-and-retry; the retry's overflow is + // terminal, not a third attempt. + assert.equal(fixture.model.doStreamCalls.length, 3); + assert.equal(complete(fixture)?.stopReason, 'error'); + assert.equal(fixture.events.some((event) => event.type === 'error'), true); + assert.equal(fixture.recorded.length, 1); + assert.equal(fixture.llmCalls.at(-1)?.errorClass, 'ContextLength'); + }); + + test('no recovery seam means a context-length overflow ends as a real error', async () => { + const fixture = buildReactiveFixture({ script: ['tool', 'overflow'], midTurnEnabled: false, bigPriors: true }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 2); + assert.equal(complete(fixture)?.stopReason, 'error'); + assert.equal(fixture.recorded.length, 0); + assert.equal(fixture.summarizerCalls(), 0); + }); + + test('an overflow with no foldable completed span ends as a real error', async () => { + // First-request overflow with no prior turns: the pool is just the current + // user message, so there is no safe completed span to fold. Recovery is not + // possible, so the provider error is surfaced honestly (not a fake success, + // and not a synthesized context_budget_exhausted — the provider rejected). + const fixture = buildReactiveFixture({ script: ['overflow'], withoutPriorTurns: true }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 1); + assert.equal(complete(fixture)?.stopReason, 'error'); + assert.equal(fixture.events.some((event) => event.type === 'error'), true); + assert.equal(fixture.recorded.length, 0); + }); +}); + +function runtimeTextEvent(id: string, turnId: string, role: 'user' | 'model', text: string): RuntimeEvent { + return { + id, + sessionId: 'session-1', + runId: 'run-1', + turnId, + invocationId: 'run-1', + ts: 1_800_000_000_000, + partial: false, + role, + author: role === 'user' ? 'user' : 'agent', + content: { kind: 'text', text }, + }; +} + +function header(): SessionHeader { + return { + id: 'session-1', + workspaceRoot: '/tmp/maka', + cwd: '/tmp/maka', + createdAt: 1, + lastUsedAt: 1, + name: 'Test', + isFlagged: false, + labels: [], + isArchived: false, + status: 'active', + statusUpdatedAt: 1, + hasUnread: false, + backend: 'ai-sdk', + llmConnectionSlug: 'anthropic-main', + connectionLocked: true, + model: 'mock-model-id', + permissionMode: 'ask', + schemaVersion: 1, + }; +} + +function connection(): LlmConnection { + return { + slug: 'anthropic-main', + name: 'Anthropic', + providerType: 'anthropic', + defaultModel: 'mock-model-id', + enabled: true, + createdAt: 1, + updatedAt: 1, + }; +} + +function idGenerator(): () => string { + let index = 0; + return () => `id-${++index}`; +} + +function monotonicClock(): () => number { + let value = 1_000; + return () => ++value; +} diff --git a/packages/runtime/src/__tests__/tool-runtime-extraction-contract.test.ts b/packages/runtime/src/__tests__/tool-runtime-extraction-contract.test.ts index 43037b7610..c3c9662056 100644 --- a/packages/runtime/src/__tests__/tool-runtime-extraction-contract.test.ts +++ b/packages/runtime/src/__tests__/tool-runtime-extraction-contract.test.ts @@ -113,7 +113,10 @@ describe('ModelAdapter extraction contract', () => { assert.match(adapter, /startStream\(/); assert.match(adapter, /await import\('ai'\)/); assert.match(adapter, /streamText\(/); - assert.match(adapter, /stepCountIs\(this\.input\.maxSteps\)/); + // The step budget stays adapter-owned, with a per-call override so the + // backend's reactive overflow retry passes only the remaining budget. + assert.match(adapter, /input\.maxSteps \?\? this\.input\.maxSteps/); + assert.match(adapter, /stepCountIs\(maxSteps\)/); assert.match(adapter, /handleStreamChunk\(/); assert.match(adapter, /switch \(chunk\.type\)/); assert.match(adapter, /case 'reasoning-delta'/); diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 4355dc962b..45e1e13a97 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -107,6 +107,7 @@ import { type PrepareStepLike, type PrepareStepResultLike, type RepairableAiSdkToolCall, + type StreamTextResult, } from './model-adapter.js'; import { rewriteActiveToolResultsInMessages, @@ -406,6 +407,71 @@ function midTurnRequestPayloadChars( ); } +/** + * Outcome of folding the durable turn ledger into a replacement projection. + * Shared by the proactive prepareStep hook (which maps it to keepProjection / + * shapeFailure / a `context_limit` replacement) and the reactive overflow + * recovery (which maps it to a retry / a real error terminal, with an + * `overflow` reason). The verdict/diagnostic is the caller's; this only shapes. + */ +type MidTurnCompactionOutcome = + | { decision: 'skip' } + | { + decision: 'fail'; + detail: ContextBudgetExhaustedDetail; + diagnosticReason: string; + recorderCounters?: Partial; + } + | { + decision: 'compacted'; + checkpoint: HistoryCompactCheckpoint; + replacementMessages: ModelMessage[]; + estimatedTokensBefore: number; + estimatedTokensAfter: number; + }; + +/** + * The `decision: 'replaced'` diagnostic patch for a durable mid_turn fold, + * shared by the proactive (`reason: 'context_limit'`) and reactive + * (`reason: 'overflow'`) triggers so both report the fold identically. + */ +function buildMidTurnReplacedDiagnosticPatch(input: { + checkpoint: HistoryCompactCheckpoint; + estimatedTokensBefore: number; + estimatedTokensAfter: number; + reason: string; +}): Partial { + const { checkpoint, estimatedTokensBefore, estimatedTokensAfter, reason } = input; + return { + historyCompactEnabled: true, + historyCompactMode: 'read_write', + historyCompactWritesAttempted: 1, + historyCompactBlocksWritten: 1, + historyCompactWrittenBlockIds: [checkpoint.checkpointId], + historyCompactWriteEstimatedTokens: checkpoint.estimatedTokens, + historyCompactBlockIds: [checkpoint.checkpointId], + historyCompactedTurns: checkpoint.coverage.turnCount, + historyCompactedEvents: checkpoint.coverage.eventCount, + historyCompactedEstimatedTokensBefore: estimatedTokensBefore, + historyCompactedEstimatedTokensAfter: estimatedTokensAfter, + highWaterName: checkpoint.highWaterName, + highWaterSeq: checkpoint.highWaterSeq, + highWaterReason: 'history_compact', + ...compactionDecisionDiagnosticPatch({ + stage: 'activeStep', + sourceKind: 'runtimeEvents', + decision: 'replaced', + phase: 'mid_turn', + boundaryKind: 'historyCompact', + boundaryIds: [checkpoint.checkpointId], + coverage: { bodySha256: [checkpoint.coverage.sourceDigest] }, + reason, + estimatedTokensBefore, + estimatedTokensAfter, + }), + }; +} + /** * Event-driven wait for seq-ack progress: resolves when the queue reports any * push/ack/close/wake, or immediately on abort. The caller loops and re-checks @@ -1329,75 +1395,183 @@ export class AiSdkBackend implements AgentBackend { }) : shapedPrepareStep; - const result = await this.modelAdapter.startStream({ - model, - messages, - tools: aiSdkTools, - activeTools, - repairToolCall: async ( - { toolCall, error }: { toolCall: RepairableAiSdkToolCall; error: unknown }, - ) => { - return repairMakaToolCall({ - toolCall, - availableToolNames: currentRepairToolNames(), - error, - }); - }, - system: systemPrompt, - abortSignal: this.abortController!.signal, - ...(prepareStep ? { prepareStep } : {}), - }); - - for await (const chunk of result.fullStream) { - if (this.aborted) break; - watchdog.markActivity(); - // Step boundary, version-tolerant: AI SDK v6 delimits steps with - // `start-step` / `finish-step`, older releases said `step-finish`. - // Missing the boundary would silently degrade back to one message per - // turn, so match both names. A duplicate boundary is harmless: the - // second flush no-ops (accumulators already cleared) and one extra id - // rotation just discards an unused id. - const isStepFinishChunk = chunk.type === 'finish-step' || chunk.type === 'step-finish'; - if (isStepFinishChunk) { - runtimeSteps += 1; - const stepUsage = normalizeAiSdkUsage(chunk.usage, { rawFinishReason: chunk.finishReason }); - if (!stepUsage) sawUnusableStepUsage = true; - if (stepUsage) { - completedStepUsage = mergeNormalizedUsage(completedStepUsage, stepUsage); - this.cumulativeUsageCheckpoint = mergeNormalizedUsage(this.cumulativeUsageCheckpoint, stepUsage); - await this.input.recordUsageCheckpoint?.({ - ...this.cumulativeUsageCheckpoint, - costUsd: this.computeTokenUsageCostUsd(this.cumulativeUsageCheckpoint), + // Reactive overflow recovery (issue #882 PR 2). A request-level + // provider failure surfaces as a fullStream `error` chunk — both when + // the transport throws (finishReason then rejects with + // NoOutputGeneratedError) and when it streams an error part — after + // which this stream is dead. Capture it and, at most once, fold the + // durable ledger and resend on a context-length overflow; otherwise + // throw the real provider error so the terminal handler closes the turn + // as an error. Never fall through to the success path, which + // historically caught the rejected finishReason as `stop` and + // fabricated an end_turn completion with success telemetry. + // + // Attempt→send translation (reviews P1-A and round-3 P1): the SDK + // scopes BOTH `stepNumber` and `steps` to one streamText call, but + // every per-step consumer downstream works in SEND units — the + // capacity hook's durability wait (flushedSteps), replacedStepNumber, + // lastShapeFailure, the semantic-compact yield, the availability + // runtime's same-turn `load_tools` activations, and the active + // tool-result prune's eligible tool-call IDs. This single translation + // point (a) rebases each attempt's local step numbers onto the + // send-global clock (completed steps when the attempt started), and + // (b) presents the send-global steps view: completed steps archived + // from every prior attempt, then the current attempt's own. Without + // it a retry resets those clocks/views — an attempt-local wait bound + // already satisfied by a previous attempt let a post-retry compaction + // drop not-yet-durable step content, a fresh empty `steps` revoked + // same-turn tool activations, and the prune's empty eligible set + // resurrected archived raw tool results from the ledger-rebuilt + // recovery projection. Consumers stay untouched; any future steps + // consumer is send-correct by construction. Steps folded into a + // checkpoint stay in the view: ID-based consumers only act on + // messages actually present in the projection, so a folded step's + // entry is inert. + let attemptStepBase = 0; + const completedAttemptSteps: PrepareStepLike['steps'][number][] = []; + let attemptObservedSteps: PrepareStepLike['steps'] = []; + const sendScopedPrepareStep: PrepareStepFunctionLike | undefined = prepareStep + ? async (options) => { + // prepareStep sees every completed step of its own attempt, so + // the latest observation is the whole attempt when it dies (the + // rejected request completes no step after it). + attemptObservedSteps = options.steps; + return prepareStep({ + ...options, + stepNumber: attemptStepBase + options.stepNumber, + steps: [...completedAttemptSteps, ...options.steps], }); } - } - if (chunk.type === 'finish' || isStepFinishChunk) { - rawFinishReason = rawFinishReasonString(chunk.finishReason) ?? rawFinishReason; - } - this.modelAdapter.handleStreamChunk(chunk, turnId, this.currentStepMessageId!, queue, { - onText: (t) => { stepText += t; }, - onTextComplete: (t) => { stepText = t; }, - onThinking: (t) => { stepThinking += t; }, - onThinkingSignature: (sig) => { stepSignature = sig; }, + : undefined; + let attemptMessages: ModelMessage[] = messages; + let overflowRetryUsed = false; + let result!: StreamTextResult; + for (;;) { + // The step limit is a SEND-level cap: `runtimeSteps` (this send's + // completed steps across attempts) is its single counter, so a retry + // attempt gets only the remaining budget — never a fresh full one. + // It is also the attempt's step base: the pump has consumed every + // prior attempt's finish-step before the error chunk that ended it, + // so at this point the counter equals the send's completed steps. + attemptStepBase = runtimeSteps; + const remainingStepBudget = this.maxSteps === undefined + ? undefined + : Math.max(0, this.maxSteps - runtimeSteps); + result = await this.modelAdapter.startStream({ + model, + messages: attemptMessages, + tools: aiSdkTools, + activeTools, + repairToolCall: async ( + { toolCall, error }: { toolCall: RepairableAiSdkToolCall; error: unknown }, + ) => { + return repairMakaToolCall({ + toolCall, + availableToolNames: currentRepairToolNames(), + error, + }); + }, + system: systemPrompt, + abortSignal: this.abortController!.signal, + ...(sendScopedPrepareStep ? { prepareStep: sendScopedPrepareStep } : {}), + ...(remainingStepBudget !== undefined ? { maxSteps: remainingStepBudget } : {}), }); - // The step's text/thinking deltas are all in (the fullStream is - // drained in order), so flush this step's AssistantMessage and rotate - // to a fresh id for the next step. The step's tool calls (appended - // mid-step via execute()) already carry the pre-rotation id via - // `getCurrentStepId`, so replay can regroup them with this step's - // reasoning even though they land before this row in the ledger. - if (isStepFinishChunk) { - await flushStep(); - this.currentStepMessageId = this.newId(); - if (midTurnState) { - // Durability clock: step N's thinking/text completion events are - // enqueued by flushStep just above, so only after this boundary - // can a seq-ack wait for step N mean anything. Wake waiters AFTER - // the increment or they would re-check a stale count and sleep. - midTurnState.flushedSteps += 1; - queue.wake(); + + let streamErrorChunk: unknown; + let sawStreamError = false; + for await (const chunk of result.fullStream) { + if (this.aborted) break; + watchdog.markActivity(); + // A request-level error ends this stream; capture it and stop + // consuming (the synthesized trailer carries no real step) so the + // recovery decision runs on the outcome, not the trailer. + if (chunk.type === 'error') { + streamErrorChunk = chunk.error; + sawStreamError = true; + break; + } + // Step boundary, version-tolerant: AI SDK v6 delimits steps with + // `start-step` / `finish-step`, older releases said `step-finish`. + // Missing the boundary would silently degrade back to one message per + // turn, so match both names. A duplicate boundary is harmless: the + // second flush no-ops (accumulators already cleared) and one extra id + // rotation just discards an unused id. + const isStepFinishChunk = chunk.type === 'finish-step' || chunk.type === 'step-finish'; + if (isStepFinishChunk) { + runtimeSteps += 1; + const stepUsage = normalizeAiSdkUsage(chunk.usage, { rawFinishReason: chunk.finishReason }); + if (!stepUsage) sawUnusableStepUsage = true; + if (stepUsage) { + completedStepUsage = mergeNormalizedUsage(completedStepUsage, stepUsage); + this.cumulativeUsageCheckpoint = mergeNormalizedUsage(this.cumulativeUsageCheckpoint, stepUsage); + await this.input.recordUsageCheckpoint?.({ + ...this.cumulativeUsageCheckpoint, + costUsd: this.computeTokenUsageCostUsd(this.cumulativeUsageCheckpoint), + }); + } + } + if (chunk.type === 'finish' || isStepFinishChunk) { + rawFinishReason = rawFinishReasonString(chunk.finishReason) ?? rawFinishReason; + } + this.modelAdapter.handleStreamChunk(chunk, turnId, this.currentStepMessageId!, queue, { + onText: (t) => { stepText += t; }, + onTextComplete: (t) => { stepText = t; }, + onThinking: (t) => { stepThinking += t; }, + onThinkingSignature: (sig) => { stepSignature = sig; }, + }); + // The step's text/thinking deltas are all in (the fullStream is + // drained in order), so flush this step's AssistantMessage and rotate + // to a fresh id for the next step. The step's tool calls (appended + // mid-step via execute()) already carry the pre-rotation id via + // `getCurrentStepId`, so replay can regroup them with this step's + // reasoning even though they land before this row in the ledger. + if (isStepFinishChunk) { + await flushStep(); + this.currentStepMessageId = this.newId(); + if (midTurnState) { + // Durability clock: step N's thinking/text completion events are + // enqueued by flushStep just above, so only after this boundary + // can a seq-ack wait for step N mean anything. Wake waiters AFTER + // the increment or they would re-check a stale count and sleep. + midTurnState.flushedSteps += 1; + queue.wake(); + } + } + } + + if (sawStreamError && !this.aborted) { + // A retry is a fresh provider request that would run at least one + // more step; with the send-level budget already spent there is + // nothing left to grant it, so the error is terminal. + const stepBudgetRemains = this.maxSteps === undefined || runtimeSteps < this.maxSteps; + const recovered = stepBudgetRemains ? await this.recoverFromOverflowError({ + error: streamErrorChunk, + retryAlreadyUsed: overflowRetryUsed, + midTurnState, + turnId, + currentMessages: attemptMessages, + providerTools, + activeTools: currentRepairToolNames(), + systemPromptChars: midTurnSystemPromptChars, + turnTailPrompt, + queue, + onDiagnosticPatch: onMidTurnDiagnosticPatch, + }) : undefined; + if (recovered) { + overflowRetryUsed = true; + attemptMessages = recovered.messages; + // Archive the dead attempt's completed steps into the send view + // before the next attempt resets the SDK's local `steps`. + completedAttemptSteps.push(...attemptObservedSteps); + attemptObservedSteps = []; + continue; } + // Unrecoverable (not context-length, latch spent, no seam, or no + // safe fold): surface the real provider error via the terminal + // handler — never a fabricated success. + throw streamErrorChunk; } + break; } // If the stream loop exited because stop() flipped this.aborted while a @@ -1445,8 +1619,19 @@ export class AiSdkBackend implements AgentBackend { // Final usage event. AI SDK `usage` is the last step only; `totalUsage` // is the billing-relevant sum across all internal tool-loop steps. + // The send-level usage owner is `completedStepUsage`, the per-step + // accumulator that spans every attempt: after a reactive overflow + // retry, the last attempt's totalUsage covers only that attempt, so + // recording it would silently drop the first attempt's completed + // steps. totalUsage remains the authoritative shorthand only for the + // single-attempt send, and an unusable step sample in ANY attempt + // fails the whole record closed (#972) — a later attempt's valid + // totalUsage must not wash it back to "complete". try { - tokenUsage = normalizeAiSdkUsage(await (result.totalUsage ?? result.usage), { rawFinishReason }); + const attemptTotalUsage = normalizeAiSdkUsage(await (result.totalUsage ?? result.usage), { rawFinishReason }); + tokenUsage = overflowRetryUsed + ? (sawUnusableStepUsage ? undefined : completedStepUsage) + : attemptTotalUsage; if (tokenUsage) { const systemPromptHash = turnDiagnostics.requestShape.componentHashes.systemPromptHash; tokenUsageCostUsd = this.computeTokenUsageCostUsd(tokenUsage); @@ -2349,17 +2534,11 @@ export class AiSdkBackend implements AgentBackend { onDiagnosticPatch: (patch: Partial) => void, ): PrepareStepFunctionLike | undefined { if (!state) return undefined; - const summarizer = this.input.summarizeHistoryCompact!; - const recorder = this.input.recordHistoryCompactCheckpoint!; - const loadTurnRuntimeEvents = this.input.loadTurnRuntimeEvents!; const policy = this.input.contextBudget!; const compactPolicy = policy.historyCompact!; const midTurn = compactPolicy.midTurn!; const charsPerToken = policy.charsPerToken ?? 4; const reserveTokens = midTurn.reserveTokens ?? 16_384; - const maxSummaryEstimatedTokens = compactPolicy.maxBlockEstimatedTokens - ?? compactPolicy.maxSummaryEstimatedTokens - ?? 1_024; let acceptedProjection: ActiveFullCompactPrepareStepProjection | undefined; return async (options) => { @@ -2381,9 +2560,19 @@ export class AiSdkBackend implements AgentBackend { // non-positive input count is unusable for estimation, so clear the // baseline and let the estimate fall back to the whole-payload cold // start instead of "0 + delta". + // + // The usage anchor is only meaningful PAIRED with the payload baseline + // of the request it was reported for (`lastRequestPayloadChars`). A + // successful overflow recovery restructures the request and resets that + // baseline to undefined: the send-global steps view still carries the + // dead attempt's last usage, but anchoring on it against the rejected + // request's chars would under-estimate the retry by the whole previous + // step growth — so a missing baseline forces the whole-payload cold + // start, exactly like a missing usage sample. const lastStepInputTokens = normalizeAiSdkUsage(options.steps.at(-1)?.usage)?.inputTokens; state.lastRequestInputTokens = - lastStepInputTokens !== undefined && Number.isFinite(lastStepInputTokens) && lastStepInputTokens > 0 + state.lastRequestPayloadChars !== undefined + && lastStepInputTokens !== undefined && Number.isFinite(lastStepInputTokens) && lastStepInputTokens > 0 ? lastStepInputTokens : undefined; @@ -2452,220 +2641,352 @@ export class AiSdkBackend implements AgentBackend { return keepProjection(); } - // Coverage pool = the durable run ledger, read through the injected - // seam. Covered events are persisted by construction (no crash window - // between checkpoint and source), and their bytes are exactly what a - // recovery re-projection replays. - // - // Seq-ack durability boundary. The replacement projection REPLACES the - // whole message list, so any completed-step content event missing from - // the durable pool is silently dropped from the next request — a - // lagging ledger here is content loss (e.g. a step's already-emitted - // assistant text), not a conservative under-count. No event-kind - // predicate can close that: the wait counts the event stream itself. - // 1. The pump has flushed every finish-step boundary the SDK reports - // completed (state.flushedSteps), so ALL of the completed steps' - // session events — tool pairs AND thinking/text completions — are - // enqueued with producer-stamped sequence numbers. - // 2. The consumer has fully processed everything enqueued - // (consumedCount >= pushedCount). The consumer's pull is the ack - // (see drain()): it fires after processing, not after persisting, - // so deliberately-unpersisted events (non-terminal errors, - // partials) can never deadlock the wait. - // After both, ONE durable read (which itself re-awaits the run's - // serialized write queue) sees every event the projection may carry. - // Exits: the boundary, an abort, a detached consumer, or a read failure. - const abortSignal = this.abortController?.signal; - for (;;) { - if (abortSignal?.aborted) return shapeFailure('no_safe_completed_span', 'ledger_wait_aborted'); - if (queue.consumerDetached) return shapeFailure('no_safe_completed_span', 'ledger_wait_aborted'); - if (state.flushedSteps >= options.stepNumber && queue.consumedCount >= queue.pushedCount) break; - await waitForQueueProgressOrAbort(queue, abortSignal); - } - let turnLedger: RuntimeEvent[]; - try { - turnLedger = await loadTurnRuntimeEvents(turnId); - } catch { - return shapeFailure('no_safe_completed_span', 'ledger_read_failed'); - } - const currentTurnEvents = turnLedger - .filter((event) => event.turnId === turnId) - .filter(isHistoryCompactContentEvent); - // The head anchor is persisted before backend.send() is invoked, so - // its absence is a wiring error, not replication lag — fail open now. - if (!currentTurnEvents.some((event) => event.id === state.headAnchor.id)) { - return shapeFailure('no_safe_completed_span', 'head_anchor_not_durable'); - } - const orderedEvents = [...state.priorContentEvents, ...currentTurnEvents]; - - const plan = await planMidTurnCapacityCompaction({ - sessionId: this.sessionId, - orderedEvents, - headAnchor: { runtimeEventId: state.headAnchor.id, turnId }, + // Fold a safe completed prefix of the durable turn ledger into a + // replacement projection (validate → persist), shared with the reactive + // overflow path. This hook maps the outcome to the prepareStep contract: + // keep the raw projection on skip/fail, apply the fold on success. + const outcome = await this.computeMidTurnCompactionReplacement({ + turnId, + state, + queue, + minFlushedSteps: options.stepNumber, estimatedNextRequestTokens: estimate, - contextWindow: state.contextWindow, - reserveTokens, - reserveTailEvents: midTurn.reserveTailEvents ?? 1, - charsPerToken, - now: this.now(), - ...(compactPolicy.highWaterName !== undefined ? { highWaterName: compactPolicy.highWaterName } : {}), - maxSummaryEstimatedTokens, - ...(state.previousCheckpoint ? { previousCheckpoint: state.previousCheckpoint } : {}), - summarize: async ({ coveredRuntimeEvents, newlyFoldedRuntimeEvents, previousCheckpoint }) => { - const draftBlock = buildHistoryCompactBlockFromSummary({ - sessionId: this.sessionId, - foldedRuntimeEvents: coveredRuntimeEvents, - summary: 'Mid-turn capacity compaction draft.', - ...(compactPolicy.highWaterName !== undefined ? { highWaterName: compactPolicy.highWaterName } : {}), - maxSummaryEstimatedTokens, - charsPerToken, - now: this.now(), - }); - return await Promise.resolve(summarizer({ - sessionId: this.sessionId, - turnId, - source: { draftBlock, foldedRuntimeEvents: [...coveredRuntimeEvents] }, - limits: { - maxBlocks: 1, - maxBlockEstimatedTokens: maxSummaryEstimatedTokens, - maxEstimatedTokens: compactPolicy.maxEstimatedTokens ?? 2_048, - charsPerToken, - }, - ...(previousCheckpoint ? { previousCheckpoint } : {}), - newlyFoldedRuntimeEvents: [...newlyFoldedRuntimeEvents], - ...(this.abortController?.signal ? { abortSignal: this.abortController.signal } : {}), - })); - }, - }); - - if (plan.decision === 'skip') return keepProjection(); - if (plan.decision === 'fail_open') return shapeFailure(plan.reason, plan.reason); - - // Lifecycle order is validate → persist → apply, where validate = - // materializable ∧ smaller ∧ replay-admissible. Replay applies the - // session's latest checkpoint BEFORE any high-water check, so a - // checkpoint that fails ANY of the three must never be persisted — it - // would poison every later projection even though this step correctly - // refused it. - // Persistence still precedes application, so the crash-window property - // (never apply an unpersisted fold) is unchanged, and the recorder - // counters stay truthful: validation failures never reached the - // recorder, so they attach no write counters. - const replayPlan = buildRuntimeEventModelReplayPlan(plan.replacementEvents, { - toolActivityTurnIds: collectToolActivityTurnIds(orderedEvents), - }); - if ( - replayPlan.items.length === 0 - || hasBlockingReplayDiagnostics(replayPlan) - || (replayPlan.hasProviderNativeSemantics && !this.canReplayProviderNative(replayPlan)) - ) { - return shapeFailure('no_safe_completed_span', 'replacement_unmaterializable'); - } - // The head anchor must render exactly like the raw projection's current - // user message: the initial request decorates it with the volatile turn - // tail (cwd, shell context, task state — see send()), which is not part - // of the durable anchor bytes. Reuse the same decoration owner - // (appendTurnTailPrompt) on the anchor's replay item so a replacement - // never silently drops that context — and never counts the drop as - // shrinkage in the guard below. - const replayItemsWithAnchorTail = replayPlan.items.map((item) => - item.kind === 'text' && item.role === 'user' && item.eventId === state.headAnchor.id - ? { ...item, content: this.appendTurnTailPrompt(item.content, turnTailPrompt) as string } - : item, - ); - const replacementMessages = await this.materializeRuntimeReplayPlan({ - ...replayPlan, - items: replayItemsWithAnchorTail, - }); - // Apply the shape only when it actually shrinks the request: a - // materialized replacement that is not smaller than the current payload - // (e.g. a runaway summary block) would hand the verdict owner a WORSE - // request than the raw projection it just refused to improve. A - // non-shrinking fold proves the summarizer's OUTPUT is unusable — not - // that the irreducible remainder exceeds capacity — so over the window - // the owner reports it as summarizer_failed, with the precise - // replacement_not_smaller diagnostic reason. - const replacedPayloadChars = midTurnRequestPayloadChars( - replacementMessages, + referencePayloadChars: payloadChars, providerTools, activeToolsForStep, systemPromptChars, - ); - if (replacedPayloadChars >= payloadChars) { - return shapeFailure('summarizer_failed', 'replacement_not_smaller'); - } - // Replay admissibility, through the SAME single gate the recovery path - // runs (max block / max total / prefix budget) with the same policy — - // one acceptance standard, not two. A checkpoint accepted here but - // rejected at replay would still become the session's latest checkpoint - // and poison recovery: the next projection would refuse it and - // re-inject the covered raw span. Block-size rejections mean the - // summarizer's output is unusable (summarizer_failed); a prefix over - // the history budget means the irreducible remainder is too large. - const replayFit = evaluateHistoryCompactCheckpointReplay( - plan.checkpoint, - plan.replacementEvents.slice(1), - policy, - ); - if (!replayFit.fits) { - return shapeFailure( - replayFit.reason === 'prefix_over_budget' ? 'head_anchor_exceeds_capacity' : 'summarizer_failed', - `replay_rejected_${replayFit.reason}`, - ); - } - - // The replacement is valid: durably persist the checkpoint BEFORE - // applying the projection — the same order as the pre_turn path — so a - // recovery re-projection never re-injects the replaced raw span. A - // persistence failure keeps raw messages and records write_failed in - // the durable diagnostics; if the final payload is then over the - // window, the verdict owner maps this failure to the terminal - // summarizer_failed detail. - const writeFailedCounters: Partial = { - historyCompactWritesAttempted: 1, - historyCompactWriteFailures: 1, - }; - try { - await Promise.resolve(recorder(plan.checkpoint, turnId)); - } catch { - return shapeFailure('summarizer_failed', 'write_failed', writeFailedCounters); + turnTailPrompt, + }); + if (outcome.decision === 'skip') return keepProjection(); + if (outcome.decision === 'fail') { + return shapeFailure(outcome.detail, outcome.diagnosticReason, outcome.recorderCounters); } - state.previousCheckpoint = plan.checkpoint; acceptedProjection = { sourceSignatures: incomingMessages.map(modelMessageSignature), - projectedMessages: replacementMessages, + projectedMessages: outcome.replacementMessages, }; state.replacedStepNumber = options.stepNumber; - onDiagnosticPatch({ + onDiagnosticPatch(buildMidTurnReplacedDiagnosticPatch({ + checkpoint: outcome.checkpoint, + estimatedTokensBefore: outcome.estimatedTokensBefore, + estimatedTokensAfter: outcome.estimatedTokensAfter, + reason: 'context_limit', + })); + return { messages: outcome.replacementMessages }; + }; + } + + /** + * Fold a safe completed prefix of the durable turn ledger into a persisted + * mid_turn checkpoint and its `[block, verbatim anchor, tail]` replacement + * messages — the compaction core shared by the proactive prepareStep hook + * (issue #882 PR 1) and the reactive overflow recovery (PR 2). It waits for + * the seq-ack durability boundary, reads the ledger, plans the fold, then + * validates (materializable ∧ smaller than the reference request ∧ + * replay-admissible) and persists BEFORE returning the replacement, so a + * recovery re-projection never re-injects a covered raw span. It only shapes: + * the pass/terminate verdict and the diagnostic emission are the caller's. + */ + private async computeMidTurnCompactionReplacement(input: { + turnId: string; + state: MidTurnCapacityCompactState; + queue: AsyncEventQueue; + minFlushedSteps: number; + estimatedNextRequestTokens: number; + referencePayloadChars: number; + providerTools: readonly MakaTool[]; + activeToolsForStep: readonly string[]; + systemPromptChars: number; + turnTailPrompt: string | undefined; + }): Promise { + const { turnId, state, queue, providerTools, activeToolsForStep, systemPromptChars, turnTailPrompt } = input; + const summarizer = this.input.summarizeHistoryCompact!; + const recorder = this.input.recordHistoryCompactCheckpoint!; + const loadTurnRuntimeEvents = this.input.loadTurnRuntimeEvents!; + const policy = this.input.contextBudget!; + const compactPolicy = policy.historyCompact!; + const midTurn = compactPolicy.midTurn!; + const charsPerToken = policy.charsPerToken ?? 4; + const reserveTokens = midTurn.reserveTokens ?? 16_384; + const maxSummaryEstimatedTokens = compactPolicy.maxBlockEstimatedTokens + ?? compactPolicy.maxSummaryEstimatedTokens + ?? 1_024; + + // Coverage pool = the durable run ledger, read through the injected + // seam. Covered events are persisted by construction (no crash window + // between checkpoint and source), and their bytes are exactly what a + // recovery re-projection replays. + // + // Seq-ack durability boundary. The replacement projection REPLACES the + // whole message list, so any completed-step content event missing from + // the durable pool is silently dropped from the next request — a + // lagging ledger here is content loss (e.g. a step's already-emitted + // assistant text), not a conservative under-count. No event-kind + // predicate can close that: the wait counts the event stream itself. + // 1. The pump has flushed every finish-step boundary the SDK reports + // completed (state.flushedSteps), so ALL of the completed steps' + // session events — tool pairs AND thinking/text completions — are + // enqueued with producer-stamped sequence numbers. + // 2. The consumer has fully processed everything enqueued + // (consumedCount >= pushedCount). The consumer's pull is the ack + // (see drain()): it fires after processing, not after persisting, + // so deliberately-unpersisted events (non-terminal errors, + // partials) can never deadlock the wait. + // After both, ONE durable read (which itself re-awaits the run's + // serialized write queue) sees every event the projection may carry. + // Exits: the boundary, an abort, a detached consumer, or a read failure. + const abortSignal = this.abortController?.signal; + for (;;) { + if (abortSignal?.aborted) { + return { decision: 'fail', detail: 'no_safe_completed_span', diagnosticReason: 'ledger_wait_aborted' }; + } + if (queue.consumerDetached) { + return { decision: 'fail', detail: 'no_safe_completed_span', diagnosticReason: 'ledger_wait_aborted' }; + } + if (state.flushedSteps >= input.minFlushedSteps && queue.consumedCount >= queue.pushedCount) break; + await waitForQueueProgressOrAbort(queue, abortSignal); + } + let turnLedger: RuntimeEvent[]; + try { + turnLedger = await loadTurnRuntimeEvents(turnId); + } catch { + return { decision: 'fail', detail: 'no_safe_completed_span', diagnosticReason: 'ledger_read_failed' }; + } + const currentTurnEvents = turnLedger + .filter((event) => event.turnId === turnId) + .filter(isHistoryCompactContentEvent); + // The head anchor is persisted before backend.send() is invoked, so + // its absence is a wiring error, not replication lag — fail open now. + if (!currentTurnEvents.some((event) => event.id === state.headAnchor.id)) { + return { decision: 'fail', detail: 'no_safe_completed_span', diagnosticReason: 'head_anchor_not_durable' }; + } + const orderedEvents = [...state.priorContentEvents, ...currentTurnEvents]; + + const plan = await planMidTurnCapacityCompaction({ + sessionId: this.sessionId, + orderedEvents, + headAnchor: { runtimeEventId: state.headAnchor.id, turnId }, + estimatedNextRequestTokens: input.estimatedNextRequestTokens, + contextWindow: state.contextWindow, + reserveTokens, + reserveTailEvents: midTurn.reserveTailEvents ?? 1, + charsPerToken, + now: this.now(), + ...(compactPolicy.highWaterName !== undefined ? { highWaterName: compactPolicy.highWaterName } : {}), + maxSummaryEstimatedTokens, + ...(state.previousCheckpoint ? { previousCheckpoint: state.previousCheckpoint } : {}), + summarize: async ({ coveredRuntimeEvents, newlyFoldedRuntimeEvents, previousCheckpoint }) => { + const draftBlock = buildHistoryCompactBlockFromSummary({ + sessionId: this.sessionId, + foldedRuntimeEvents: coveredRuntimeEvents, + summary: 'Mid-turn capacity compaction draft.', + ...(compactPolicy.highWaterName !== undefined ? { highWaterName: compactPolicy.highWaterName } : {}), + maxSummaryEstimatedTokens, + charsPerToken, + now: this.now(), + }); + return await Promise.resolve(summarizer({ + sessionId: this.sessionId, + turnId, + source: { draftBlock, foldedRuntimeEvents: [...coveredRuntimeEvents] }, + limits: { + maxBlocks: 1, + maxBlockEstimatedTokens: maxSummaryEstimatedTokens, + maxEstimatedTokens: compactPolicy.maxEstimatedTokens ?? 2_048, + charsPerToken, + }, + ...(previousCheckpoint ? { previousCheckpoint } : {}), + newlyFoldedRuntimeEvents: [...newlyFoldedRuntimeEvents], + ...(this.abortController?.signal ? { abortSignal: this.abortController.signal } : {}), + })); + }, + }); + + if (plan.decision === 'skip') return { decision: 'skip' }; + if (plan.decision === 'fail_open') { + return { decision: 'fail', detail: plan.reason, diagnosticReason: plan.reason }; + } + + // Lifecycle order is validate → persist → apply, where validate = + // materializable ∧ smaller ∧ replay-admissible. Replay applies the + // session's latest checkpoint BEFORE any high-water check, so a + // checkpoint that fails ANY of the three must never be persisted — it + // would poison every later projection even though this step correctly + // refused it. + const replayPlan = buildRuntimeEventModelReplayPlan(plan.replacementEvents, { + toolActivityTurnIds: collectToolActivityTurnIds(orderedEvents), + }); + if ( + replayPlan.items.length === 0 + || hasBlockingReplayDiagnostics(replayPlan) + || (replayPlan.hasProviderNativeSemantics && !this.canReplayProviderNative(replayPlan)) + ) { + return { decision: 'fail', detail: 'no_safe_completed_span', diagnosticReason: 'replacement_unmaterializable' }; + } + // The head anchor must render exactly like the raw projection's current + // user message: the initial request decorates it with the volatile turn + // tail (cwd, shell context, task state — see send()), which is not part + // of the durable anchor bytes. Reuse the same decoration owner + // (appendTurnTailPrompt) on the anchor's replay item so a replacement + // never silently drops that context — and never counts the drop as + // shrinkage in the guard below. + const replayItemsWithAnchorTail = replayPlan.items.map((item) => + item.kind === 'text' && item.role === 'user' && item.eventId === state.headAnchor.id + ? { ...item, content: this.appendTurnTailPrompt(item.content, turnTailPrompt) as string } + : item, + ); + const replacementMessages = await this.materializeRuntimeReplayPlan({ + ...replayPlan, + items: replayItemsWithAnchorTail, + }); + // Apply the shape only when it actually shrinks the request versus the + // reference payload (the incoming request for the proactive hook, the + // request that overflowed for reactive recovery): a materialized + // replacement that is not smaller proves the summarizer's OUTPUT is + // unusable, reported as summarizer_failed via replacement_not_smaller. + const replacedPayloadChars = midTurnRequestPayloadChars( + replacementMessages, + providerTools, + activeToolsForStep, + systemPromptChars, + ); + if (replacedPayloadChars >= input.referencePayloadChars) { + return { decision: 'fail', detail: 'summarizer_failed', diagnosticReason: 'replacement_not_smaller' }; + } + // Replay admissibility, through the SAME single gate the recovery path + // runs (max block / max total / prefix budget) with the same policy — + // one acceptance standard, not two. A checkpoint accepted here but + // rejected at replay would still become the session's latest checkpoint + // and poison recovery. + const replayFit = evaluateHistoryCompactCheckpointReplay( + plan.checkpoint, + plan.replacementEvents.slice(1), + policy, + ); + if (!replayFit.fits) { + return { + decision: 'fail', + detail: replayFit.reason === 'prefix_over_budget' ? 'head_anchor_exceeds_capacity' : 'summarizer_failed', + diagnosticReason: `replay_rejected_${replayFit.reason}`, + }; + } + + // The replacement is valid: durably persist the checkpoint BEFORE + // applying the projection — the same order as the pre_turn path. A + // persistence failure keeps raw messages and records write_failed. + try { + await Promise.resolve(recorder(plan.checkpoint, turnId)); + } catch { + return { + decision: 'fail', + detail: 'summarizer_failed', + diagnosticReason: 'write_failed', + recorderCounters: { historyCompactWritesAttempted: 1, historyCompactWriteFailures: 1 }, + }; + } + state.previousCheckpoint = plan.checkpoint; + return { + decision: 'compacted', + checkpoint: plan.checkpoint, + replacementMessages, + estimatedTokensBefore: plan.estimatedTokensBefore, + estimatedTokensAfter: plan.estimatedTokensAfter, + }; + } + + /** + * Reactive overflow recovery (issue #882 PR 2): the second line of defense. + * When a provider rejects a request with a context-length error, fold the + * durable turn ledger once and resend once — a single compact-and-retry + * latch (pi's `_overflowRecoveryAttempted`). Returns the compacted messages + * to resend, or undefined when recovery is impossible or already spent, in + * which case the caller surfaces the real provider error rather than a + * fabricated success or a synthesized `context_budget_exhausted` (the + * provider — not the runtime — rejected the request). Non-context-length + * errors and turns without the mid-turn seam never reach compaction, so the + * default (no seam) behavior is already better than the old fake end_turn. + */ + private async recoverFromOverflowError(input: { + error: unknown; + retryAlreadyUsed: boolean; + midTurnState: MidTurnCapacityCompactState | undefined; + turnId: string; + currentMessages: readonly ModelMessage[]; + providerTools: readonly MakaTool[]; + activeTools: readonly string[]; + systemPromptChars: number; + turnTailPrompt: string | undefined; + queue: AsyncEventQueue; + onDiagnosticPatch: (patch: Partial) => void; + }): Promise<{ messages: ModelMessage[] } | undefined> { + const state = input.midTurnState; + if (input.retryAlreadyUsed || !state) return undefined; + if (this.modelAdapter.classifyError(input.error) !== 'ContextLength') return undefined; + + // The shrink baseline is the request the provider actually rejected. Its + // single owner is the verdict owner's per-request payload measure + // (state.lastRequestPayloadChars), recorded at the end of every + // prepareStep run — the attempt-INITIAL messages undercount the rejected + // request by every same-turn tool step, and a baseline anchored there + // refuses folds that genuinely shrink the real request (review P1-1). + // The cold-start fallback only covers a send whose verdict owner never + // ran a prepareStep (defensive; step 0 records the baseline too). + const referencePayloadChars = state.lastRequestPayloadChars ?? midTurnRequestPayloadChars( + input.currentMessages, + input.providerTools, + input.activeTools, + input.systemPromptChars, + ); + const outcome = await this.computeMidTurnCompactionReplacement({ + turnId: input.turnId, + state, + queue: input.queue, + // The stream has ended, so every completed step is already flushed; wait + // only for the consumer to drain the durable ledger up to date. + minFlushedSteps: state.flushedSteps, + // The provider rejected the request outright, so force the fold past the + // high water regardless of the (evidently under-counting) estimate. + estimatedNextRequestTokens: state.contextWindow + 1, + referencePayloadChars, + providerTools: input.providerTools, + activeToolsForStep: input.activeTools, + systemPromptChars: input.systemPromptChars, + turnTailPrompt: input.turnTailPrompt, + }); + if (outcome.decision !== 'compacted') { + // Recovery attempted but could not produce a smaller, admissible + // request; record the failed overflow attempt and let the caller surface + // the real provider error. + input.onDiagnosticPatch({ historyCompactEnabled: true, historyCompactMode: 'read_write', - historyCompactWritesAttempted: 1, - historyCompactBlocksWritten: 1, - historyCompactWrittenBlockIds: [plan.checkpoint.checkpointId], - historyCompactWriteEstimatedTokens: plan.checkpoint.estimatedTokens, - historyCompactBlockIds: [plan.checkpoint.checkpointId], - historyCompactedTurns: plan.checkpoint.coverage.turnCount, - historyCompactedEvents: plan.checkpoint.coverage.eventCount, - historyCompactedEstimatedTokensBefore: plan.estimatedTokensBefore, - historyCompactedEstimatedTokensAfter: plan.estimatedTokensAfter, - highWaterName: plan.checkpoint.highWaterName, - highWaterSeq: plan.checkpoint.highWaterSeq, - highWaterReason: 'history_compact', + ...(outcome.decision === 'fail' && outcome.recorderCounters ? outcome.recorderCounters : {}), ...compactionDecisionDiagnosticPatch({ stage: 'activeStep', sourceKind: 'runtimeEvents', - decision: 'replaced', + decision: 'failedOpen', phase: 'mid_turn', boundaryKind: 'historyCompact', - boundaryIds: [plan.checkpoint.checkpointId], - coverage: { bodySha256: [plan.checkpoint.coverage.sourceDigest] }, - reason: 'context_limit', - estimatedTokensBefore: plan.estimatedTokensBefore, - estimatedTokensAfter: plan.estimatedTokensAfter, + reason: 'overflow', + ...(outcome.decision === 'fail' ? { failOpenReason: outcome.diagnosticReason } : {}), }), }); - return { messages: replacementMessages }; - }; + return undefined; + } + input.onDiagnosticPatch(buildMidTurnReplacedDiagnosticPatch({ + checkpoint: outcome.checkpoint, + estimatedTokensBefore: outcome.estimatedTokensBefore, + estimatedTokensAfter: outcome.estimatedTokensAfter, + reason: 'overflow', + })); + // A successful recovery restructures the request, so the rejected + // request's payload measure no longer describes what the retry sends. + // Reset the baseline: the capacity hook's usage anchor is only coherent + // paired with the payload chars of the SAME request, and a missing + // baseline forces the whole-payload cold-start estimate instead of a + // stale pairing against the dead attempt. + state.lastRequestPayloadChars = undefined; + return { messages: outcome.replacementMessages }; } /** diff --git a/packages/runtime/src/model-adapter.ts b/packages/runtime/src/model-adapter.ts index 4cafa4c9e4..17bb11471c 100644 --- a/packages/runtime/src/model-adapter.ts +++ b/packages/runtime/src/model-adapter.ts @@ -113,6 +113,14 @@ export interface ModelAdapterStreamInput { * on the next step without mutating the cached tools prefix. */ prepareStep?: PrepareStepFunctionLike; + /** + * Per-call step budget override. The step limit is a SEND-level cap owned by + * the backend: a reactive overflow retry re-invokes startStream mid-send and + * must pass only the remaining budget (configured maxSteps minus the steps + * already completed), or every retry would silently reset the cap. Defaults + * to the adapter's configured maxSteps. + */ + maxSteps?: number; } export interface ModelAdapterStreamCallbacks { @@ -163,6 +171,7 @@ export class ModelAdapter { isLoopFinished: () => unknown; }; + const maxSteps = input.maxSteps ?? this.input.maxSteps; return streamText({ model: input.model, messages: input.messages, @@ -174,9 +183,9 @@ export class ModelAdapter { providerOptions: this.input.providerOptions, // streamText defaults to one step when stopWhen is omitted. Its exported // non-stopping condition is required for an unbounded tool loop. - stopWhen: this.input.maxSteps === undefined + stopWhen: maxSteps === undefined ? isLoopFinished() - : stepCountIs(this.input.maxSteps), + : stepCountIs(maxSteps), abortSignal: input.abortSignal, }); } diff --git a/packages/runtime/src/tool-availability.ts b/packages/runtime/src/tool-availability.ts index 86a6056849..f880ca70d9 100644 --- a/packages/runtime/src/tool-availability.ts +++ b/packages/runtime/src/tool-availability.ts @@ -160,7 +160,10 @@ export class ToolAvailabilityRuntime { const seedGroups = this.seedLoadedGroups(priorEvents); // Turn-local snapshot the guard / repair / diagnostics read; recomputed // before every step by `prepareStep`. No cross-turn mutable state — a load - // survives turns only via the ledger seed above (durable by construction). + // survives turns only via the ledger seed above (durable by construction), + // and within one send the backend's translation point hands every hook a + // send-global `steps` view spanning overflow-retry attempts, so activation + // stays monotonic per send without a bespoke set here. const turn = { active: new Set() }; const computeActive = (steps: ReadonlyArray | undefined): string[] => { const loaded = new Set([...seedGroups, ...this.loadedGroupsFromSteps(steps)]); diff --git a/packages/runtime/src/tool-runtime.ts b/packages/runtime/src/tool-runtime.ts index 3955185c80..39675aad67 100644 --- a/packages/runtime/src/tool-runtime.ts +++ b/packages/runtime/src/tool-runtime.ts @@ -1341,27 +1341,254 @@ export function formatSyntheticToolErrorText(error: unknown): string { return `${redacted.slice(0, TOOL_ERROR_RESULT_MAX_CHARS - 1)}…`; } +/** + * Structured provider error identifiers that mean the INPUT exceeded the + * model's context window. These come from the provider's error JSON and are + * the ONLY unconditional overflow evidence: free-text signals are vetoable. + */ +const CONTEXT_OVERFLOW_PROVIDER_CODES: ReadonlySet = new Set([ + 'context_length_exceeded', // OpenAI & OpenAI-compatible: error.code + 'model_context_window_exceeded', // z.ai: error.code + 'request_too_large', // Anthropic byte-size overflow (HTTP 413): error.type +]); + +/** + * A provider failure normalized into classification evidence. classifyError's + * real input domain is NOT just Error instances: a request-level failure is + * an AI SDK `APICallError` (provider JSON parsed in `data`, raw in + * `responseBody`; no top-level `.code`), while an in-stream error part + * carries the provider's parsed error VALUE — OpenAI Chat emits the inner + * `{message, type?, code?}` object, OpenAI Responses the whole + * `{type:'error', error:{type, code, message}}` chunk, Anthropic the inner + * `{type, message}` object, and openai-compatible a bare message string. + * Shapes read from the provider sources, never invented. + */ +interface ProviderErrorEvidence { + /** Lowercased composite of the textual fields, for pattern evidence. */ + text: string; + /** Explicit HTTP status from a field ('' when absent) — never a substring. */ + statusCode: string; + /** Top-level code field as a string ('' when absent). */ + code: string; + /** Structured provider identifiers (code/type), lowercased. */ + structuredCodes: string[]; +} + +/** Collects `code`/`type` strings from a payload and from its `error` wrapper. */ +function collectStructuredCodes(payload: unknown, out: string[]): void { + const fromRecord = (record: unknown) => { + if (typeof record !== 'object' || record === null) return; + for (const key of ['code', 'type'] as const) { + const value = (record as Record)[key]; + if (typeof value === 'string' && value) out.push(value.toLowerCase()); + } + }; + fromRecord(payload); + if (typeof payload === 'object' && payload !== null) { + fromRecord((payload as { error?: unknown }).error); + } +} + +function normalizeErrorEvidence(error: unknown): ProviderErrorEvidence | undefined { + if (error instanceof Error) { + const code = 'code' in error ? String((error as { code?: unknown }).code) : ''; + const statusCode = 'statusCode' in error + ? String((error as { statusCode?: unknown }).statusCode) + : 'status' in error + ? String((error as { status?: unknown }).status) + : ''; + const rawBody = (error as { responseBody?: unknown }).responseBody; + const body = typeof rawBody === 'string' ? rawBody : ''; + const structuredCodes: string[] = []; + collectStructuredCodes((error as { data?: unknown }).data, structuredCodes); + if (structuredCodes.length === 0 && body) { + // The failed-response handler keeps the raw body even when the provider + // JSON failed the schema (which is exactly when `data` is absent). + try { + collectStructuredCodes(JSON.parse(body), structuredCodes); + } catch { + // Not JSON — no structured evidence. + } + } + return { + // The raw body joins the text evidence: when the provider JSON fails + // the error schema, `message` degrades to the statusText and the body + // is the ONLY carrier of the provider's wording (e.g. an + // OpenAI-compatible `{error: string}` overflow). Positives and vetoes + // both run over the same full text. + text: `${error.name} ${code} ${statusCode} ${error.message}${body ? ` ${body}` : ''}`.toLowerCase(), + statusCode, + code, + structuredCodes, + }; + } + if (typeof error === 'string') { + const structuredCodes: string[] = []; + try { + collectStructuredCodes(JSON.parse(error), structuredCodes); + } catch { + // A plain message string — text evidence only. + } + return { text: error.toLowerCase(), statusCode: '', code: '', structuredCodes }; + } + if (typeof error === 'object' && error !== null) { + const record = error as Record; + const field = (key: string): string => { + const value = record[key]; + return typeof value === 'string' || typeof value === 'number' ? String(value) : ''; + }; + const structuredCodes: string[] = []; + collectStructuredCodes(record, structuredCodes); + let text: string; + try { + // Serialize the whole value so message/code text is evidence no matter + // which of the known provider shapes carried it. + text = JSON.stringify(error).toLowerCase(); + } catch { + text = String(error).toLowerCase(); + } + return { + text, + statusCode: field('statusCode') || field('status'), + code: field('code'), + structuredCodes, + }; + } + return undefined; +} + +/** + * Provider context-length overflow signatures. A request-level 400/413 whose + * message matches one of these means the input exceeded the model's context + * window — the reactive-recovery trigger (issue #882 PR 2). The set is ported + * from pi's battle-tested table and covers the providers Maka ships in its + * registry (Anthropic, OpenAI/-compatible, Google, xAI, Groq, OpenRouter, + * Mistral, MiniMax, Kimi/Moonshot, Together, llama.cpp/LM Studio/Ollama, …). + * Matched against the ORIGINAL error's composite fields (name, code, status, + * message), never the generalized string. All of these are free-text evidence + * and can be vetoed by NON_CONTEXT_OVERFLOW_PATTERNS: a capacity statement or + * overflow phrase quoted inside a throttling/quota error must not trigger + * recovery — only a structured provider code is unconditional. + */ +const CONTEXT_OVERFLOW_PATTERNS: readonly RegExp[] = [ + /prompt is too long/i, // Anthropic token overflow + /request_too_large/i, // Anthropic request byte-size overflow (HTTP 413) + /input is too long for requested model/i, // Amazon Bedrock + /exceeds the context window/i, // OpenAI (Completions & Responses) + /exceeds (?:the )?(?:model'?s )?maximum context length(?: of [\d,]+ tokens?|\s*\([\d,]+\))?/i, // OpenAI-compatible proxies (LiteLLM) + /input token count.*exceeds the maximum/i, // Google (Gemini) + /maximum prompt length is \d+/i, // xAI (Grok) + /reduce the length of the messages/i, // Groq + /maximum context length is \d+ tokens/i, // OpenRouter (most backends) + /exceeds (?:the )?maximum allowed input length of [\d,]+ tokens?/i, // OpenRouter/Poolside + /input \(\d+ tokens\) is longer than the model'?s context length \(\d+ tokens\)/i, // Together AI + // GitHub Copilot: "prompt token count of X exceeds the limit of Y". The INPUT + // subject is required — a bare "token count of N exceeds the limit of M" also + // matches output/completion caps, and a bare "exceeds the limit of N" matches + // file-size and other quota errors; neither is fixable by history compaction. + /(?:prompt|input|context|message)[^.]{0,80}token count of [\d,]+ exceeds the limit of [\d,]+/i, + /exceeds the available context size/i, // llama.cpp server + /greater than the context length/i, // LM Studio + /context window exceeds limit/i, // MiniMax + /exceeded model token limit/i, // Kimi For Coding + /too large for model with \d+ maximum context length/i, // Mistral + /prompt has [\d,]+ tokens?, but the configured context size is [\d,]+ tokens?/i, // DS4 server + /model_context_window_exceeded/i, // z.ai non-standard finish_reason surfaced as error text + /prompt too long; exceeded (?:max )?context length/i, // Ollama explicit overflow error + /context[_ ]length[_ ]exceeded/i, // OpenAI structured error code (also generic) + // Ambiguous token-limit wording that is an input overflow only when an + // input-like word is the subject. `request` is deliberately NOT in the + // subject list: it appears in generic prefixes ("Invalid request: ...") + // without saying anything about which side of the token budget overflowed. + /(?:prompt|input|context|message)[^.]{0,80}too many tokens/i, + /(?:prompt|input|context|message)[^.]{0,80}token limit exceeded/i, +]; + +/** + * Wording that looks token-shaped but is NOT an input overflow: throttling / + * quota / rate limiting, and complete OUTPUT-cap relations in every observed + * permutation of role word (output/completion/max_tokens) and token + * predicate — subject before predicate ("completion has too many tokens", + * "max_tokens token limit exceeded"), predicate before subject ("too many + * tokens were requested for the completion"), the count-of form ("output + * token count of N exceeds"), the role word embedded inside the phrase + * ("too many completion tokens were requested"), and the role-tokens-exceed + * form ("Maximum completion tokens exceeded"). Noun phrases alone (e.g. + * "completion token count") are not excluded: they also appear as usage + * breakdowns inside genuine input-overflow messages, and "(prompt + + * completion) exceed" combined-budget wording stays classifiable because the + * role word is not adjacent to "tokens". + */ +const NON_CONTEXT_OVERFLOW_PATTERNS: readonly RegExp[] = [ + /rate limit/i, + /too many requests/i, + /throttl/i, + /quota/i, + /(?:output|completion|max_tokens)\b[^.]{0,60}(?:too many tokens|token limit exceeded)/i, + /(?:too many tokens|token limit exceeded)[^.]{0,60}\b(?:output|completion|max_tokens)/i, + /(?:output|completion)\s+token\s+(?:count|limit)[^.]{0,40}exceed/i, + /too many (?:output|completion|max_tokens)[^.]{0,20}tokens/i, + /\b(?:output|completion|max_tokens)\s+tokens?\b[^.]{0,20}exceed/i, +]; + +/** + * Two-layer overflow detection on an error's raw text (the composite of its + * original name/code/status/message). Triggering recovery requires positive + * evidence of an INPUT overflow — the one class history compaction can fix: + * 1. Vetoes first: throttling/quota wording and complete output-cap relations + * disqualify every free-text signal. Free text is never unconditional — a + * capacity statement quoted inside a throttle error is not an overflow. + * 2. Positive overflow relations count only when nothing vetoed. Structured + * provider codes (the unconditional evidence) are classifyError's job, + * checked before this text layer ever runs. + */ +export function isContextOverflowErrorText(text: string): boolean { + if (!text) return false; + if (NON_CONTEXT_OVERFLOW_PATTERNS.some((pattern) => pattern.test(text))) return false; + return CONTEXT_OVERFLOW_PATTERNS.some((pattern) => pattern.test(text)); +} + +/** + * Classifies a provider error by DESCENDING evidence strength over the + * normalized evidence (Error, string, or plain stream-error-part object): + * abort → 402 → 429 → 401/403 (numeric fields, never substrings) → the + * provider's structured overflow code → bare 413 (HTTP: request entity too + * large — itself input-side evidence, Cerebras sends it with no body) → + * vetoable free-text overflow relations → generic 5xx → weak word + * heuristics. Specific overflow evidence outranks a generic 5xx because + * proxies (LiteLLM) wrap provider overflows in 503s; the weak heuristics + * rank last so "generate" can never become a rate limit. + */ export function classifyError(error: unknown): string { - if (!(error instanceof Error)) return 'Other'; - const code = 'code' in error ? String((error as { code?: unknown }).code) : ''; - const statusCode = 'statusCode' in error - ? String((error as { statusCode?: unknown }).statusCode) - : 'status' in error - ? String((error as { status?: unknown }).status) - : ''; - const text = `${error.name} ${code} ${statusCode} ${error.message}`.toLowerCase(); + const evidence = normalizeErrorEvidence(error); + if (!evidence) return 'Other'; + const { text, statusCode, code, structuredCodes } = evidence; if (text.includes('abort')) return 'Abort'; if (statusCode === '402' || code === '402') return 'ProviderBilling'; - if (text.includes('rate') || statusCode === '429' || code === '429') return 'RateLimit'; - if (text.includes('auth') || statusCode === '401' || statusCode === '403' || code === '401' || code === '403') return 'Auth'; + if (statusCode === '429' || code === '429') return 'RateLimit'; + if (statusCode === '401' || statusCode === '403' || code === '401' || code === '403') return 'Auth'; + // Structured provider evidence: the parsed error JSON's code/type is the + // only unconditional signal for a context overflow. + if (structuredCodes.some((c) => CONTEXT_OVERFLOW_PROVIDER_CODES.has(c))) return 'ContextLength'; + if (statusCode === '413' || code === '413') return 'ContextLength'; + // Free-text overflow relations on the composite text, veto-first inside. + if (isContextOverflowErrorText(text)) return 'ContextLength'; if (/^5\d\d$/.test(statusCode) || /^5\d\d$/.test(code)) return 'ProviderUnavailable'; + // Weak word heuristics, last: they only catch errors that carried no + // stronger evidence for any other class. `rate` must be word-shaped + // ("generate"/"separate" are not rate limits) while still matching the + // rate_limit/RateLimitError identifier spellings. + if (/\brate\b|rate[_-]?limit/.test(text)) return 'RateLimit'; + if (text.includes('auth')) return 'Auth'; if (text.includes('timeout')) return 'Timeout'; if (text.includes('network') || text.includes('fetch')) return 'Network'; - return error.name || 'Other'; + return error instanceof Error ? (error.name || 'Other') : 'Other'; } export function errorReasonFromClass(errorClass: string): string | undefined { switch (errorClass) { + case 'ContextLength': + return 'context_overflow'; case 'Timeout': return 'timeout'; case 'Auth':