From d8e041175f490d13625f42e3b3cdad4154b8760b Mon Sep 17 00:00:00 2001 From: hetaoBackend Date: Fri, 9 Oct 2026 17:48:49 +0800 Subject: [PATCH 1/2] fix(runtime): drop forced compaction after oversized requests Remove the one-shot compact-and-resend recovery that ran after a provider rejected a main-agent request as too large. The resend bypassed the normal before-LLM pipeline, so request-level transforms (screenshot caps, session identity restoration) were not reapplied to it. Oversized requests (HTTP 413, request_too_large, the local request-body check, context-window overflow) are still classified as deterministic rejections and are not resent unchanged. Co-authored-by: Cursor --- .../context-overflow-recovery.ts | 128 -------- .../agent-core/src/pi-turn-runner/hooks.ts | 15 - packages/agent-core/src/pi-turn-runner/llm.ts | 128 ++------ .../agent-core/src/pi-turn-runner/turn.ts | 28 +- .../context-overflow-recovery.test.ts | 292 ------------------ .../unit/pi-turn-runner/llm-retry.test.ts | 11 +- .../compaction/context-compaction.ts | 3 - .../agent-host/compaction/contracts.ts | 7 - .../agent-host/execution/executor.ts | 8 - .../automatic-context-compactor.test.ts | 66 ---- .../compaction/automatic-context-compactor.ts | 41 +-- packages/shared/src/llm-error-classifier.ts | 4 +- release/public-source.json | 2 - test/vitest-suites.json | 1 - 14 files changed, 49 insertions(+), 685 deletions(-) delete mode 100644 packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts delete mode 100644 packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts diff --git a/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts b/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts deleted file mode 100644 index ed09da1d1..000000000 --- a/packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts +++ /dev/null @@ -1,128 +0,0 @@ -import type { StreamFn } from '@earendil-works/pi-agent-core'; -import { - isContextOverflow, - type AssistantMessage, - type AssistantMessageEvent, - type AssistantMessageEventStream, - type Message, -} from '@earendil-works/pi-ai'; -import { isLLMRequestOversized, normalizeLLMError } from '@mavis/shared/llm-error-classifier'; - -/** - * Per-turn slot connecting the provider stream (built before the Agent exists) - * with the Agent-owned context lifecycle that can shrink the context. - */ -export interface ContextOverflowRecoveryState { - /** - * Shrinks the context of the request that just failed and returns the - * provider-bound messages to resend, or `undefined` when nothing changed. - * Installed by `setLLMHook` only when recovery hooks exist. - */ - recover?: (signal: AbortSignal | undefined) => Promise; - /** Recovery runs at most once per turn, so a second rejection is terminal. */ - attempted: boolean; -} - -export function newContextOverflowRecoveryState(): ContextOverflowRecoveryState { - return { attempted: false }; -} - -/** - * True when a failed provider response means the request was too large to be - * accepted as sent: HTTP 413 / `request_too_large`, the local request-body - * check, or a context-window overflow recognized by pi-ai. - */ -export function isRecoverableContextOverflow( - message: AssistantMessage, - contextWindow: number | undefined, -): boolean { - if (message.stopReason !== 'error') return false; - if (isContextOverflow(message, contextWindow)) return true; - return isLLMRequestOversized( - normalizeLLMError({ - finishReason: message.stopReason, - ...(message.errorMessage ? { errorMessage: message.errorMessage } : {}), - }), - ); -} - -/** - * When the provider rejects a main-agent request as too large before producing - * any output, compact once and resend once. Successful streams only pay for - * buffering their first non-`start` event. A second rejection, a recovery that - * changes nothing, or any other failure is surfaced exactly as received. - */ -export function withContextOverflowRecovery( - inner: StreamFn, - state: ContextOverflowRecoveryState, -): StreamFn { - return (async (model, context, options) => { - const stream = await inner(model, context, options); - if (!state.recover || state.attempted) return stream; - const iterator = stream[Symbol.asyncIterator](); - const buffered: AssistantMessageEvent[] = []; - try { - for (;;) { - const next = await iterator.next(); - if (next.done) return replay(stream, buffered, undefined); - buffered.push(next.value); - if (next.value.type !== 'start') break; - } - } catch (error) { - return replay(stream, buffered, undefined, { error }); - } - const first = buffered.at(-1); - if ( - first?.type !== 'error' || - options?.signal?.aborted || - !isRecoverableContextOverflow(first.error, model.contextWindow) - ) { - return replay(stream, buffered, iterator); - } - state.attempted = true; - let messages: Message[] | undefined; - try { - messages = await state.recover(options?.signal); - } catch { - // A failing recovery must not mask the provider rejection it tried to fix. - return replay(stream, buffered, iterator); - } - if (!messages || options?.signal?.aborted) return replay(stream, buffered, iterator); - try { - void Promise.resolve(iterator.return?.()).catch(() => undefined); - } catch { - // Best-effort cleanup of the rejected attempt. - } - return inner(model, { ...context, messages }, options); - }) as StreamFn; -} - -function replay( - stream: AssistantMessageEventStream, - buffered: readonly AssistantMessageEvent[], - iterator: AsyncIterator | undefined, - pending?: { readonly error: unknown }, -): AssistantMessageEventStream { - return { - [Symbol.asyncIterator]() { - let index = 0; - let thrown = false; - return { - async next(): Promise> { - if (index < buffered.length) return { value: buffered[index++]!, done: false }; - if (pending && !thrown) { - thrown = true; - throw pending.error; - } - if (!iterator) return { value: undefined, done: true }; - return iterator.next(); - }, - async return(value?: unknown): Promise> { - await iterator?.return?.(value); - return { value: undefined, done: true }; - }, - }; - }, - result: () => stream.result(), - } as unknown as AssistantMessageEventStream; -} diff --git a/packages/agent-core/src/pi-turn-runner/hooks.ts b/packages/agent-core/src/pi-turn-runner/hooks.ts index 4e0e51c8e..9702ff9fb 100644 --- a/packages/agent-core/src/pi-turn-runner/hooks.ts +++ b/packages/agent-core/src/pi-turn-runner/hooks.ts @@ -26,12 +26,6 @@ export interface PiBeforeLlmCallHookInput { sessionId: string; turnId: string; phase: PiBeforeLlmCallPhase; - /** - * Set only when the runner re-runs `contextOverflowRecoveryHook` after the - * provider rejected the request as too large (HTTP 413 / context overflow). - * Compaction hooks must compact regardless of their own size estimate. - */ - trigger?: 'context_overflow_recovery'; /** Current provider request view after request-only hook transforms. */ messages: AgentMessage[]; /** Current durable history, updated only by canonical replacements. */ @@ -257,15 +251,6 @@ export type PiOnTurnEndHook = PiOnStepEndHook; export interface PiTurnHooks { beforeLlmCallHook?: readonly PiBeforeLlmCallHook[]; - /** - * Hooks that may shrink the context after the provider rejected a main-agent - * request as too large (HTTP 413, `request_too_large`, the local request-body - * check, or a context-window overflow) before producing any output. The - * runner runs them once per turn with `trigger: 'context_overflow_recovery'` - * and, when they replace the context, resends the request once. Only - * compaction belongs here: reminder or append hooks would run twice. - */ - contextOverflowRecoveryHook?: readonly PiBeforeLlmCallHook[]; /** Best-effort observers of actual logical main-agent provider calls. */ onLlmCallPreparedHook?: readonly PiOnLlmCallPreparedHook[]; afterLlmCallHook?: readonly PiAfterLlmCallHook[]; diff --git a/packages/agent-core/src/pi-turn-runner/llm.ts b/packages/agent-core/src/pi-turn-runner/llm.ts index 5206dd847..d9bcfcb8c 100644 --- a/packages/agent-core/src/pi-turn-runner/llm.ts +++ b/packages/agent-core/src/pi-turn-runner/llm.ts @@ -5,7 +5,6 @@ import { resolveLLMStreamTimeouts, withLLMStreamTimeouts } from './llm-stream-ti import type { PiAfterLlmReplacementCommit, PiBeforeLlmCallAppendMessage, - PiBeforeLlmCallHook, PiBeforeLlmCallReplaceMetadata, } from './hooks.js'; import type { LLMModelConfig } from './types.js'; @@ -42,24 +41,40 @@ export function composeStreamFn(resolved: LLMModelConfig): StreamFn { } export function setLLMHook(agent: Agent, turn: turnState, history: turnHistory): void { - const recoveryHooks = turn.hooks.contextOverflowRecovery; - if ( - turn.hooks.beforeLLM.length === 0 && - turn.hooks.afterLLM.length === 0 && - recoveryHooks.length === 0 - ) { - return; - } + if (turn.hooks.beforeLLM.length === 0 && turn.hooks.afterLLM.length === 0) return; let checkpointCallCount = 0; - // Pi's live loop context of the latest main-agent request; overflow recovery - // compacts exactly this context and keeps the loop in sync through it. - let lastRequest: { messages: AgentMessage[]; phase: 'initial' | 'iteration' } | undefined; + agent.transformContext = async (messages, signal) => { + const rejectedMessages = messages.filter((message) => + turn.rejectedAssistantMessages.has(message), + ); + if (rejectedMessages.length > 0) { + messages.splice( + 0, + messages.length, + ...messages.filter((message) => !turn.rejectedAssistantMessages.has(message)), + ); + for (const message of rejectedMessages) turn.rejectedAssistantMessages.delete(message); + } - const applyCheckpoint = async ( - messages: AgentMessage[], - checkpoint: Exclude, - ): Promise => { + if (turn.hooks.beforeLLM.length === 0) return messages; + const phase = checkpointCallCount === 0 ? 'initial' : 'iteration'; + checkpointCallCount += 1; + const checkpoint = await runBeforeLLM(turn, messages, phase, signal); + if (checkpoint.type === 'abort') { + turn.metrics.tryRecordTerminalFailure({ + errorSource: 'before_llm_hook', + errorKind: 'hook_error', + }); + throw new Error(`before_llm_checkpoint_aborted: ${checkpoint.reason}`); + } + if (checkpoint.type === 'respond') { + turn.syntheticResponse.pending = { + text: checkpoint.text, + reason: checkpoint.reason, + }; + return messages; + } if (checkpoint.type === 'requestOnly') { return checkpoint.messages; } @@ -115,89 +130,13 @@ export function setLLMHook(agent: Agent, turn: turnState, history: turnHistory): } return checkpoint.messages; }; - - agent.transformContext = async (messages, signal) => { - const rejectedMessages = messages.filter((message) => - turn.rejectedAssistantMessages.has(message), - ); - if (rejectedMessages.length > 0) { - messages.splice( - 0, - messages.length, - ...messages.filter((message) => !turn.rejectedAssistantMessages.has(message)), - ); - for (const message of rejectedMessages) turn.rejectedAssistantMessages.delete(message); - } - - const phase = checkpointCallCount === 0 ? 'initial' : 'iteration'; - lastRequest = { messages, phase }; - if (turn.hooks.beforeLLM.length === 0) return messages; - checkpointCallCount += 1; - const checkpoint = await runBeforeLLM(turn, messages, phase, signal, turn.hooks.beforeLLM); - if (checkpoint.type === 'abort') { - turn.metrics.tryRecordTerminalFailure({ - errorSource: 'before_llm_hook', - errorKind: 'hook_error', - }); - throw new Error(`before_llm_checkpoint_aborted: ${checkpoint.reason}`); - } - if (checkpoint.type === 'respond') { - turn.syntheticResponse.pending = { - text: checkpoint.text, - reason: checkpoint.reason, - }; - return messages; - } - return applyCheckpoint(messages, checkpoint); - }; - - if (recoveryHooks.length === 0) return; - turn.contextOverflowRecovery.recover = async (signal) => { - const request = lastRequest; - if (!request) return undefined; - const logContext = { - session_id: turn.input.sessionId, - turn_id: turn.input.turnId, - phase: request.phase, - provider: turn.llm.model.provider, - }; - const checkpoint = await runBeforeLLM( - turn, - request.messages, - request.phase, - signal, - recoveryHooks, - 'context_overflow_recovery', - ); - if ( - checkpoint.type === 'abort' || - checkpoint.type === 'respond' || - checkpoint.type === 'unchanged' - ) { - turn.logger.warn( - { ...logContext, outcome: checkpoint.type }, - '[pi-turn-runner] context overflow recovery left the request unchanged; surfacing the provider rejection', - ); - return undefined; - } - const messages = await applyCheckpoint(request.messages, checkpoint); - turn.logger.info( - { ...logContext, outcome: checkpoint.type, message_count: messages.length }, - '[pi-turn-runner] context overflow recovery compacted the request; resending once', - ); - return await agent.convertToLlm(messages); - }; } -type BeforeLLMResult = Awaited>; - async function runBeforeLLM( turn: turnState, messages: AgentMessage[], phase: 'initial' | 'iteration', - signal: AbortSignal | undefined, - hooks: readonly PiBeforeLlmCallHook[], - trigger?: 'context_overflow_recovery', + signal?: AbortSignal, ): Promise< | { type: 'unchanged'; messages: AgentMessage[] } | { type: 'requestOnly'; messages: AgentMessage[] } @@ -233,14 +172,13 @@ async function runBeforeLLM( afterCommit?: PiAfterLlmReplacementCommit; } | undefined; - for (const hook of hooks) { + for (const hook of turn.hooks.beforeLLM) { let decision: Awaited>; try { decision = await hook({ sessionId: turn.input.sessionId, turnId: turn.input.turnId, phase, - ...(trigger ? { trigger } : {}), messages: [...currentMessages], canonicalMessages: [...canonicalMessages], model: turn.llm.model, diff --git a/packages/agent-core/src/pi-turn-runner/turn.ts b/packages/agent-core/src/pi-turn-runner/turn.ts index 15d235362..95fe531da 100644 --- a/packages/agent-core/src/pi-turn-runner/turn.ts +++ b/packages/agent-core/src/pi-turn-runner/turn.ts @@ -19,11 +19,6 @@ import type { PiStepEndHookInput, PiToolExecutionStartHook, } from './hooks.js'; -import { - newContextOverflowRecoveryState, - withContextOverflowRecovery, - type ContextOverflowRecoveryState, -} from './context-overflow-recovery.js'; import { composeStreamFn } from './llm.js'; import { withLLMRetry, type LLMCallScope, type LLMCallSettledEvent } from './llm-retry.js'; import type { TurnMetricsRecorder } from './metrics.js'; @@ -91,7 +86,6 @@ export interface LlmCaptureAgentEventSource { export interface turnHooks { beforeLLM: readonly PiBeforeLlmCallHook[]; - contextOverflowRecovery: readonly PiBeforeLlmCallHook[]; onLLMPrepared: readonly PiOnLlmCallPreparedHook[]; afterLLM: readonly PiAfterLlmCallHook[]; beforeTool: readonly PiBeforeToolCallHook[]; @@ -126,8 +120,6 @@ export interface turnState; - /** One-shot compact-and-resend after a main-agent 413 / context overflow. */ - contextOverflowRecovery: ContextOverflowRecoveryState; nextEventId(kind: string): string; nextRuntimeSeq(): number; } @@ -138,7 +130,6 @@ export function newTurn( ): turnState { const resolved = input.llm; const composedStreamFn = composeStreamFn(resolved); - const contextOverflowRecovery = newContextOverflowRecoveryState(); const retryStream = (scope: LLMCallScope): StreamFn => { const metricsStreamFn = deps.metrics ? deps.metrics.wrapStreamFn(composedStreamFn, { recordTerminalFailure: false }) @@ -160,17 +151,11 @@ export function newTurn( return callerOnCallSettled?.(event); }, }); - // Overflow recovery sits inside the logical-call observers so a recovered - // request settles as one successful logical call. - const recoveringStreamFn = - scope === 'agent' - ? withContextOverflowRecovery(retryingStreamFn, contextOverflowRecovery) - : retryingStreamFn; // One recorder spans the whole logical call. Physical retries overwrite its // in-memory candidate; only the final successful settlement can persist. const capturedStreamFn = deps.llmCapture - ? deps.llmCapture.wrapLogicalStreamFn(recoveringStreamFn, { scope }) - : recoveringStreamFn; + ? deps.llmCapture.wrapLogicalStreamFn(retryingStreamFn, { scope }) + : retryingStreamFn; return deps.metrics ? deps.metrics.wrapLogicalStreamFn(capturedStreamFn, { // Auto-compaction is an auxiliary degradation: local context lifecycle @@ -184,10 +169,9 @@ export function newTurn( const mainStreamFn = input.llmRetry || deps.llmCapture ? retryStream('agent') - : withContextOverflowRecovery( - deps.metrics ? deps.metrics.wrapStreamFn(composedStreamFn) : composedStreamFn, - contextOverflowRecovery, - ); + : deps.metrics + ? deps.metrics.wrapStreamFn(composedStreamFn) + : composedStreamFn; const auxiliaryStreamFn = input.llmRetry ? retryStream('compaction') : undefined; const syntheticResponse: turnState['syntheticResponse'] = {}; let eventSeq = 0; @@ -239,7 +223,6 @@ export function newTurn( ); const hooks: turnHooks = { beforeLLM: input.hooks?.beforeLlmCallHook ?? [], - contextOverflowRecovery: input.hooks?.contextOverflowRecoveryHook ?? [], onLLMPrepared: input.hooks?.onLlmCallPreparedHook ?? [], afterLLM: input.hooks?.afterLlmCallHook ?? [], beforeTool: input.hooks?.beforeToolCallHook ?? [], @@ -282,7 +265,6 @@ export function newTurn( blockedToolCalls: [], syntheticResponse, rejectedAssistantMessages: new Set(), - contextOverflowRecovery, nextEventId, nextRuntimeSeq, }; diff --git a/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts b/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts deleted file mode 100644 index 4d71f7c3c..000000000 --- a/packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts +++ /dev/null @@ -1,292 +0,0 @@ -import { describe, expect, it, vi } from "vitest"; -import type { AgentMessage, StreamFn } from "@earendil-works/pi-agent-core"; -import type { - AssistantMessage, - AssistantMessageEvent, - AssistantMessageEventStream, - Context, - Model, -} from "@earendil-works/pi-ai"; -import { - newContextOverflowRecoveryState, - withContextOverflowRecovery, -} from "../../../src/pi-turn-runner/context-overflow-recovery.js"; -import { PiTurnRunner } from "../../../src/pi-turn-runner/pi-turn-runner.js"; -import type { - PiBeforeLlmCallHook, - PiBeforeLlmCallHookInput, - PiHistoryChangedHookInput, -} from "../../../src/pi-turn-runner/hooks.js"; -import type { RuntimeEvent } from "../../../src/pi-turn-runner/types.js"; -import { RuntimeEventStatus } from "../../../src/protocol/runtime-event.js"; - -function fakeModel(provider = "minimax_api"): Model { - return { - id: "fake-model", - name: "fake-model", - api: "test-messages", - provider, - baseUrl: "https://example.invalid", - reasoning: false, - input: ["text"], - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, - contextWindow: 200_000, - maxTokens: 8_192, - } as Model; -} - -function assistant( - stopReason: AssistantMessage["stopReason"], - content: AssistantMessage["content"], - errorMessage?: string, -): AssistantMessage { - return { - role: "assistant", - content, - api: "test-messages", - provider: "minimax_api", - model: "fake-model", - usage: { - input: 0, - output: 0, - cacheRead: 0, - cacheWrite: 0, - totalTokens: 0, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, - }, - stopReason, - ...(errorMessage ? { errorMessage } : {}), - timestamp: 0, - } as AssistantMessage; -} - -function stream(events: AssistantMessageEvent[], final: AssistantMessage): AssistantMessageEventStream { - return { - [Symbol.asyncIterator]() { - let index = 0; - return { - async next() { - if (index < events.length) return { value: events[index++], done: false }; - return { value: undefined, done: true }; - }, - } as AsyncIterableIterator; - }, - async result() { - return final; - }, - } as unknown as AssistantMessageEventStream; -} - -function successStream(text: string): AssistantMessageEventStream { - const empty = assistant("stop", []); - const final = assistant("stop", [{ type: "text", text }]); - return stream( - [ - { type: "start", partial: empty }, - { type: "text_start", contentIndex: 0, partial: empty }, - { type: "text_delta", contentIndex: 0, delta: text, partial: final }, - { type: "text_end", contentIndex: 0, content: text, partial: final }, - { type: "done", reason: "stop", message: final }, - ], - final, - ); -} - -function errorStream(message: string): AssistantMessageEventStream { - const empty = assistant("error", []); - const final = assistant("error", [], message); - return stream( - [ - { type: "start", partial: empty }, - { type: "error", reason: "error", error: final }, - ], - final, - ); -} - -const OVERSIZED = "413 Request Entity Too Large"; - -function textOf(message: { content: unknown }): string { - const content = message.content; - if (typeof content === "string") return content; - return (content as { type: string; text?: string }[]) - .map((part) => (part.type === "text" ? (part.text ?? "") : "")) - .join(""); -} - -function summaryMessage(): AgentMessage { - return { role: "user", content: [{ type: "text", text: "SUMMARY" }], timestamp: 1 } as AgentMessage; -} - -function compactingHook(calls: PiBeforeLlmCallHookInput[]): PiBeforeLlmCallHook { - return (input) => { - calls.push(input); - const kept = input.messages.slice(-1); - return { - type: "replaceMessages", - messages: [summaryMessage(), ...kept], - metadata: { - replacementId: `r${calls.length}`, - strategyVersion: "test", - summary: "SUMMARY", - compactedMessages: input.messages.slice(0, -1), - keptMessages: kept, - firstKeptIndex: input.messages.length - 1, - }, - }; - }; -} - -async function runTurn(options: { - responses: Array<() => AssistantMessageEventStream>; - recoveryHooks?: PiBeforeLlmCallHook[]; - beforeLlmCallHook?: PiBeforeLlmCallHook[]; - provider?: string; -}) { - const contexts: Context[] = []; - let index = 0; - const streamFn = (async (_model: Model, context: Context) => { - contexts.push({ ...context, messages: structuredClone(context.messages) }); - const next = options.responses[Math.min(index, options.responses.length - 1)]!; - index += 1; - return next(); - }) as unknown as StreamFn; - const events: RuntimeEvent[] = []; - const historyChanges: PiHistoryChangedHookInput[] = []; - const runner = new PiTurnRunner(); - await runner.runTurn({ - sessionId: "session-1", - turnId: "turn-1", - workspaceDir: "/tmp", - systemPrompt: "system", - userMessage: { text: "current question", timestamp: 2 }, - history: [ - { role: "user", content: [{ type: "text", text: "old question" }], timestamp: 0 } as AgentMessage, - assistant("stop", [{ type: "text", text: "old answer ".repeat(100) }]) as AgentMessage, - ], - llm: { model: fakeModel(options.provider), apiKey: "k", streamFn }, - eventWriter: { - pushRuntime: (event) => void events.push(event), - appendEvents: (batch) => void events.push(...batch), - }, - toolConfig: { tools: [], context: {} as never }, - llmRetry: { sleep: vi.fn(async () => {}), random: () => 0 }, - hooks: { - beforeLlmCallHook: options.beforeLlmCallHook ?? [() => ({ type: "continue" })], - ...(options.recoveryHooks ? { contextOverflowRecoveryHook: options.recoveryHooks } : {}), - onHistoryChangedHook: [async (change) => void historyChanges.push(change)], - }, - }); - const terminal = events - .map((event) => event as unknown as { type?: string; payload?: { status?: string } }) - .filter((event) => event.payload?.status !== undefined) - .at(-1); - return { contexts, events, historyChanges, terminal }; -} - -describe("context overflow recovery", () => { - it("compacts once and resends once after a BYOK 413", async () => { - const recoveryCalls: PiBeforeLlmCallHookInput[] = []; - const result = await runTurn({ - responses: [() => errorStream(OVERSIZED), () => successStream("recovered")], - recoveryHooks: [compactingHook(recoveryCalls)], - }); - - // One rejected request, one resend: the 413 is not retried five times. - expect(result.contexts).toHaveLength(2); - expect(recoveryCalls).toHaveLength(1); - expect(recoveryCalls[0]!.trigger).toBe("context_overflow_recovery"); - expect(result.contexts[0]!.messages).toHaveLength(3); - expect(result.contexts[1]!.messages.map(textOf)).toEqual(["SUMMARY", "current question"]); - // The compaction is persisted as a durable replacement before the resend. - const replace = result.historyChanges.find((change) => change.reason === "replaceMessages"); - expect(replace?.messages.map((message) => textOf(message as never))).toEqual([ - "SUMMARY", - "current question", - ]); - // The rejected attempt never reaches history; only the recovered answer does. - const deltas = result.historyChanges - .filter((change) => change.reason === "messageDelta") - .flatMap((change) => change.messages); - const assistants = deltas.filter((message) => message.role === "assistant"); - expect(assistants.map((message) => (message as AssistantMessage).stopReason)).toEqual(["stop"]); - expect(result.terminal?.payload?.status).toBe(RuntimeEventStatus.COMPLETED); - }); - - it("surfaces the rejection when the resent request is still too large", async () => { - const recoveryCalls: PiBeforeLlmCallHookInput[] = []; - const result = await runTurn({ - responses: [() => errorStream(OVERSIZED)], - recoveryHooks: [compactingHook(recoveryCalls)], - }); - - expect(result.contexts).toHaveLength(2); - expect(recoveryCalls).toHaveLength(1); - const deltas = result.historyChanges - .filter((change) => change.reason === "messageDelta") - .flatMap((change) => change.messages); - expect((deltas.at(-1) as AssistantMessage).errorMessage).toBe(OVERSIZED); - expect(result.terminal?.payload?.status).toBe(RuntimeEventStatus.FAILED); - }); - - it("does not resend when recovery leaves the context unchanged", async () => { - const recovery = vi.fn(() => ({ type: "skip", reason: "context_compaction_failed" })); - const result = await runTurn({ - responses: [() => errorStream(OVERSIZED)], - recoveryHooks: [recovery], - }); - - expect(result.contexts).toHaveLength(1); - expect(recovery).toHaveBeenCalledOnce(); - }); - - it("does not run recovery for errors that are not about request size", async () => { - const recoveryCalls: PiBeforeLlmCallHookInput[] = []; - const result = await runTurn({ - responses: [() => errorStream('400 {"type":"invalid_request_error","message":"messages: text content blocks must be non-empty"}')], - recoveryHooks: [compactingHook(recoveryCalls)], - }); - - expect(result.contexts).toHaveLength(1); - expect(recoveryCalls).toHaveLength(0); - }); - - it("does not re-run ordinary before-LLM hooks during recovery", async () => { - const ordinary = vi.fn(() => ({ type: "continue" })); - const recoveryCalls: PiBeforeLlmCallHookInput[] = []; - const result = await runTurn({ - responses: [() => errorStream(OVERSIZED), () => successStream("recovered")], - recoveryHooks: [compactingHook(recoveryCalls)], - beforeLlmCallHook: [ordinary], - }); - - expect(result.contexts).toHaveLength(2); - expect(ordinary).toHaveBeenCalledOnce(); - }); - - it("keeps the previous single-failure behaviour when no recovery hook is installed", async () => { - const result = await runTurn({ responses: [() => errorStream(OVERSIZED)] }); - expect(result.contexts).toHaveLength(1); - }); -}); - -describe("withContextOverflowRecovery", () => { - it("surfaces the original rejection when recovery itself throws", async () => { - const calls: number[] = []; - const inner = (async () => { - calls.push(1); - return errorStream(OVERSIZED); - }) as unknown as StreamFn; - const state = newContextOverflowRecoveryState(); - state.recover = async () => { - throw new Error("compaction exploded"); - }; - const wrapped = withContextOverflowRecovery(inner, state); - const out = await wrapped(fakeModel(), { messages: [] } as unknown as Context, {}); - const types: string[] = []; - for await (const event of out) types.push(event.type); - expect(types).toContain("error"); - expect(calls).toHaveLength(1); - expect(state.attempted).toBe(true); - }); -}); diff --git a/packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts b/packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts index 69292f1f2..ab0461d3f 100644 --- a/packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts +++ b/packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts @@ -12,7 +12,6 @@ import { classifyLLMRequestRejectionMessage, isLLMDeterministicRequestRejection, isLLMImageLimitMessage, - isLLMRequestOversized, normalizeLLMError, } from "@mavis/shared/llm-error-classifier"; import { @@ -1213,9 +1212,8 @@ describe("deterministic request rejection classification", () => { expect(classifyLLMRequestRejectionMessage("invalid_request_error without a status")).toBeUndefined(); }); - it("treats HTTP 413, byte-limit and context-overflow rejections as oversized", () => { - const oversized = (errorMessage: string, statusCode?: number) => - isLLMRequestOversized(normalizeLLMError({ errorMessage, statusCode })); + it("treats HTTP 413, byte-limit and context-overflow rejections as deterministic", () => { + const oversized = rejection; expect(oversized("413 Request Entity Too Large")).toBe(true); expect(oversized("anything", 413)).toBe(true); expect(oversized('{"type":"error","error":{"type":"request_too_large","message":"too big"}}')).toBe(true); @@ -1225,9 +1223,8 @@ describe("deterministic request rejection classification", () => { expect(rejection("413 Payload Too Large")).toBe(true); }); - it("does not treat transient or unrelated errors as oversized", () => { - const oversized = (errorMessage: string, statusCode?: number) => - isLLMRequestOversized(normalizeLLMError({ errorMessage, statusCode })); + it("does not treat transient or unrelated size-like errors as deterministic", () => { + const oversized = rejection; // The bare MiniMax 2013 code is a generic invalid-params code. expect(oversized("400 invalid params (2013)")).toBe(false); expect(oversized("400 Bad Request")).toBe(false); diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts index e23785b56..6ab3e3db7 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts @@ -238,9 +238,6 @@ function automaticContextCompactionInput( tools: input.tools, thinkingLevel: input.thinkingLevel, signal: input.signal, - ...(input.trigger === 'context_overflow_recovery' - ? { force: 'context_overflow_recovery' as const } - : {}), }; } diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts index 7b0c6557a..592d20ba3 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts @@ -206,13 +206,6 @@ export type AutomaticContextCompactionInput = Readonly< | 'signal' > & { readonly maxSerializedInputBytes?: number; - /** - * The provider rejected the previous request as too large (HTTP 413 or a - * context-window overflow). Compaction starts regardless of the local - * estimate, and the result must be meaningfully smaller than the request - * the provider just rejected, because that estimate evidently said it fit. - */ - readonly force?: 'context_overflow_recovery'; } >; diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.ts index 4fbab0a30..7911e6339 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.ts @@ -875,14 +875,6 @@ function mergeHooks { await input.onHistoryChanged(change); diff --git a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.test.ts b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.test.ts index c331732f5..6cd430fca 100644 --- a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.test.ts +++ b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.test.ts @@ -988,72 +988,6 @@ describe('AutomaticContextCompactor request boundaries', () => { ).rejects.toThrow(/streamFn/i); }); - it('starts below every trigger when the Provider rejected the request as too large', async () => { - const messages: AgentMessage[] = [ - { role: 'user', content: [{ type: 'text', text: 'small but rejected' }], timestamp: 1 }, - ]; - const model: Model<'anthropic-messages'> = { - id: 'model-1', - name: 'Model 1', - api: 'anthropic-messages', - provider: 'test', - baseUrl: 'https://example.invalid', - reasoning: false, - input: ['text'], - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, - contextWindow: 100_000, - maxTokens: 4_096, - }; - const input = { - sessionId: 'session-1', - phase: 'initial' as const, - messages, - model, - thinkingLevel: 'off' as const, - maxSerializedInputBytes: 16 * 1024 * 1024, - }; - - expect(automaticCompactor().probeBeforeLlm(input)).toMatchObject({ shouldStart: false }); - expect( - automaticCompactor().probeBeforeLlm({ ...input, force: 'context_overflow_recovery' }), - ).toMatchObject({ shouldStart: true }); - // Forced recovery goes on to a checkpoint (which needs the Provider transport). - await expect( - automaticCompactor().compactBeforeLlm({ ...input, force: 'context_overflow_recovery' }), - ).rejects.toThrow(/streamFn/i); - }); - - it('maps the runner overflow-recovery trigger to a forced automatic compaction', async () => { - const probeBeforeLlm = vi.fn(() => ({ shouldStart: false })); - const hook = createAutomaticContextCompactionHook( - { - automatic: { probeBeforeLlm, compactBeforeLlm: vi.fn() }, - lifecycle: { - completeCommittedHistory: vi.fn(async () => undefined), - failCommittedHistory: vi.fn(async () => undefined), - }, - } as never, - { sessionId: 'session-1', turnId: 'turn-1', turnSequence: 1 }, - 'lease-overflow', - ); - const hookInput = { - sessionId: 'session-1', - turnId: 'turn-1', - phase: 'iteration' as const, - messages: [], - canonicalMessages: [], - model: { contextWindow: 100_000, maxTokens: 4_096 } as never, - thinkingLevel: 'off' as const, - }; - await hook(hookInput); - await hook({ ...hookInput, trigger: 'context_overflow_recovery' }); - - expect(probeBeforeLlm.mock.calls.map(([input]) => (input as { force?: string }).force)).toEqual([ - undefined, - 'context_overflow_recovery', - ]); - }); - it('uses the requested Messages API maxTokens at the trigger boundary', async () => { const messages: AgentMessage[] = [ { role: 'user', content: [{ type: 'text', text: 'small context' }], timestamp: 1 }, diff --git a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts index a6f925afc..885483207 100644 --- a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts +++ b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts @@ -116,9 +116,7 @@ function prepareAutomaticCompaction( sessionId: input.sessionId, messages: input.messages, }); - const forced = input.force === 'context_overflow_recovery'; const normalTriggerMatched = - forced || footprint.inputTokens > automaticTriggerAt || (input.maxSerializedInputBytes !== undefined && footprint.serializedBytes > input.maxSerializedInputBytes); @@ -130,31 +128,6 @@ function prepareAutomaticCompaction( normalTriggerMatched, shouldStart: toolResultCompactionPlan !== undefined || normalTriggerMatched, providerInputLimit, - forcedLimits: forced ? overflowRecoveryLimits(footprint, input.maxSerializedInputBytes) : {}, - }; -} - -/** - * Share of the rejected request a forced overflow-recovery compaction may keep. - * The provider rejected a request the local estimate admitted, so admitting a - * candidate against the configured limits alone could resend the same size. - */ -const OVERFLOW_RECOVERY_RETAINED_SHARE = 0.5; - -function overflowRecoveryLimits( - footprint: { readonly inputTokens: number; readonly serializedBytes: number }, - maxSerializedInputBytes: number | undefined, -): { readonly providerInputLimit?: number; readonly maxSerializedInputBytes?: number } { - const tokens = Math.floor(footprint.inputTokens * OVERFLOW_RECOVERY_RETAINED_SHARE); - const bytes = Math.floor(footprint.serializedBytes * OVERFLOW_RECOVERY_RETAINED_SHARE); - return { - ...(tokens > 0 ? { providerInputLimit: tokens } : {}), - ...(bytes > 0 - ? { - maxSerializedInputBytes: - maxSerializedInputBytes === undefined ? bytes : Math.min(bytes, maxSerializedInputBytes), - } - : {}), }; } @@ -171,7 +144,7 @@ async function runAutomaticPolicy(options: { const { input, hooks, preparation } = options; return compactContext({ history: input.messages, - limits: automaticLimits(input, preparation.providerInputLimit, preparation.forcedLimits), + limits: automaticLimits(input, preparation.providerInputLimit), measurePair: async (pair) => preparation.measurer.measurePair(pair), ...(options.toolResultCompactionCandidate ? { toolResultCompactionCandidate: options.toolResultCompactionCandidate } @@ -192,16 +165,12 @@ async function runAutomaticPolicy(options: { function automaticLimits( input: AutomaticContextCompactionInput, providerInputLimit: number, - forcedLimits: ReturnType, ): CompactContextInput['limits'] { - const maxSerializedInputBytes = - forcedLimits.maxSerializedInputBytes ?? input.maxSerializedInputBytes; return { - providerInputLimit: Math.min( - providerInputLimit, - forcedLimits.providerInputLimit ?? providerInputLimit, - ), - ...(maxSerializedInputBytes === undefined ? {} : { maxSerializedInputBytes }), + providerInputLimit, + ...(input.maxSerializedInputBytes === undefined + ? {} + : { maxSerializedInputBytes: input.maxSerializedInputBytes }), }; } diff --git a/packages/shared/src/llm-error-classifier.ts b/packages/shared/src/llm-error-classifier.ts index 3582dbc75..2871537f4 100644 --- a/packages/shared/src/llm-error-classifier.ts +++ b/packages/shared/src/llm-error-classifier.ts @@ -279,10 +279,10 @@ export function isLLMDeterministicRequestRejection(normalized: NormalizedLLMErro * The request is too big to be accepted as sent: an HTTP 413, a byte-limit * rejection (`request_too_large`, the local pre-send body check) or a context * window overflow. Resending the same request fails again, so it must not be - * retried; the only recovery is to make the request smaller (compaction). + * retried. * Anything that also looks transient (timeout, network, 408/429/5xx) is excluded. */ -export function isLLMRequestOversized(normalized: NormalizedLLMError): boolean { +function isLLMRequestOversized(normalized: NormalizedLLMError): boolean { const { facts } = normalized; if (facts.explicitAbort || facts.timeout || facts.signals.has('network')) return false; if (facts.signals.has('tpm_rate_limit')) return false; diff --git a/release/public-source.json b/release/public-source.json index a74c8fc27..06dd1d251 100644 --- a/release/public-source.json +++ b/release/public-source.json @@ -101,7 +101,6 @@ "packages/agent-core/src/index.ts", "packages/agent-core/src/pi-turn-runner/agent.ts", "packages/agent-core/src/pi-turn-runner/assembly-fingerprint.ts", - "packages/agent-core/src/pi-turn-runner/context-overflow-recovery.ts", "packages/agent-core/src/pi-turn-runner/defaults.ts", "packages/agent-core/src/pi-turn-runner/events.ts", "packages/agent-core/src/pi-turn-runner/history.ts", @@ -138,7 +137,6 @@ "packages/agent-core/test/unit/event-bridge/decode-timing.test.ts", "packages/agent-core/test/unit/image-dimensions.test.ts", "packages/agent-core/test/unit/image-fixtures.ts", - "packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts", "packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts", "packages/agent-core/test/unit/pi-turn-runner/llm-stream-timeout.test.ts", "packages/agent-core/test/unit/pi-turn-runner/request-image-limit.test.ts", diff --git a/test/vitest-suites.json b/test/vitest-suites.json index 80c0ee07b..9e54f59d1 100644 --- a/test/vitest-suites.json +++ b/test/vitest-suites.json @@ -111,7 +111,6 @@ "packages/tui/test/unit/tui/features/composer/attachments.test.ts", "packages/tui/test/unit/tui/features/settings/status-line-picker.test.ts", "packages/tui/test/unit/tui/widgets/editor/editor-behavior.test.ts", - "packages/agent-core/test/unit/pi-turn-runner/context-overflow-recovery.test.ts", "packages/agent-core/test/unit/pi-turn-runner/llm-retry.test.ts", "packages/agent-core/test/unit/pi-turn-runner/llm-stream-timeout.test.ts", "packages/agent-core/test/unit/pi-turn-runner/request-image-limit.test.ts", From de02b4a904f370590a849ebff656dfde7b3a7234 Mon Sep 17 00:00:00 2001 From: hetaoBackend Date: Fri, 9 Oct 2026 17:48:50 +0800 Subject: [PATCH 2/2] fix(runtime): restore the session ID after compaction drops it The turn reminder omits `YOUR SESSION ID` while earlier history already shows it. When automatic compaction summarizes that history away, the model would lose its own session ID for the rest of the turn. A stateless post-compaction hook now appends a hidden session identity reminder when no visible user message or identity reminder after the latest compaction carries the ID (before the current user message on initial compaction, at the tail mid-turn). The preparation check counts that reminder too, so the next turn does not repeat the ID. Co-authored-by: Cursor --- .../assembly/local-turn-input-preparation.ts | 39 +----- .../agent-host/compaction/session-identity.ts | 37 ++++++ .../session-identity-reminder.test.ts | 116 ++++++++++++++++++ .../reminder/session-identity-reminder.ts | 36 ++++++ .../turn-system/production-composition.ts | 2 + release/public-source.json | 3 + test/vitest-suites.json | 1 + 7 files changed, 197 insertions(+), 37 deletions(-) create mode 100644 packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/session-identity.ts create mode 100644 packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.test.ts create mode 100644 packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.ts diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/assembly/local-turn-input-preparation.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/assembly/local-turn-input-preparation.ts index cac924e71..1d8a0b6c0 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/assembly/local-turn-input-preparation.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/assembly/local-turn-input-preparation.ts @@ -21,6 +21,7 @@ import { } from '../canonical-user-input.js'; import { createBackgroundTaskHostMetadata } from '../history/background/host-metadata.js'; import { assertAgentHostCapabilityAvailable } from '../empty-dependencies.js'; +import { messagesCarrySessionId } from '../compaction/session-identity.js'; import { prepareLocalInlineMedia, type LocalInlineMediaCandidate, @@ -211,7 +212,7 @@ export class LocalTurnInputPreparer { promptText, turnId: input.lease.turnId, ...(input.desktopCapabilities ? { desktopCapabilities: input.desktopCapabilities } : {}), - ...(input.history && historyCarriesSessionId(input.history, input.session.sessionId) + ...(input.history && messagesCarrySessionId(input.history, input.session.sessionId) ? { sessionIdInContext: true } : {}), }); @@ -555,42 +556,6 @@ function isImageAttachment(attachment: AgentHostInputAttachment): boolean { ); } -/** - * Whether a user message after the latest compaction summary (or anywhere when - * the session was never compacted) already shows the model this session's ID. - * Durable user messages carry the turn's ``, so the first - * turn's full agent-context keeps the ID visible until compaction drops it. - */ -export function historyCarriesSessionId(messages: readonly unknown[], sessionId: string): boolean { - if (!sessionId) return false; - for (let index = messages.length - 1; index >= 0; index -= 1) { - const message = messages[index]; - if (!isRecord(message)) continue; - // Native summaries and legacy (archon) compaction markers both start a - // fresh model-visible context; anything before them was summarized away. - if (message.role === 'compactionSummary' || Object.hasOwn(message, 'archonCompaction')) { - return false; - } - if (message.role !== 'user') continue; - if (userMessageText(message.content).includes(sessionId)) return true; - } - return false; -} - -function userMessageText(content: unknown): string { - if (typeof content === 'string') return content; - if (!Array.isArray(content)) return ''; - return content - .map((part: unknown) => - isRecord(part) && part.type === 'text' && typeof part.text === 'string' ? part.text : '', - ) - .join('\n'); -} - -function isRecord(value: unknown): value is Record { - return typeof value === 'object' && value !== null; -} - function validateBackgroundReminder( value: Awaited>, ): Awaited> { diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/session-identity.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/session-identity.ts new file mode 100644 index 000000000..72c3c32e1 --- /dev/null +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/session-identity.ts @@ -0,0 +1,37 @@ +export const SESSION_IDENTITY_REMINDER_CUSTOM_TYPE = 'session_identity_reminder'; + +/** + * Whether a user message or session identity reminder after the latest + * compaction summary (or anywhere when never compacted) shows `sessionId`. + */ +export function messagesCarrySessionId(messages: readonly unknown[], sessionId: string): boolean { + if (!sessionId) return false; + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]; + if (!isRecord(message)) continue; + // Native summaries and legacy (archon) compaction markers both start a + // fresh model-visible context; anything before them was summarized away. + if (message.role === 'compactionSummary' || Object.hasOwn(message, 'archonCompaction')) { + return false; + } + const carriesId = + message.role === 'user' || + (message.role === 'custom' && message.customType === SESSION_IDENTITY_REMINDER_CUSTOM_TYPE); + if (carriesId && messageText(message.content).includes(sessionId)) return true; + } + return false; +} + +function messageText(content: unknown): string { + if (typeof content === 'string') return content; + if (!Array.isArray(content)) return ''; + return content + .map((part: unknown) => + isRecord(part) && part.type === 'text' && typeof part.text === 'string' ? part.text : '', + ) + .join('\n'); +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null; +} diff --git a/packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.test.ts b/packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.test.ts new file mode 100644 index 000000000..e605177f7 --- /dev/null +++ b/packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.test.ts @@ -0,0 +1,116 @@ +import type { AgentMessage } from '@earendil-works/pi-agent-core'; +import type { Api, Model } from '@earendil-works/pi-ai'; +import type { PiBeforeLlmCallHookInput } from '@mavis/agent-core/pi-turn-runner'; +import { describe, expect, it } from 'vitest'; + +import { validateCanonicalHistoryMessages } from '../../agent-host/history/canonical-history-validation.js'; +import { ContextUsageAnchorState } from '../../compaction/execution/usage-anchor.js'; +import { messagesCarrySessionId } from '../../agent-host/compaction/session-identity.js'; +import { createSessionIdentityReminderHook } from './session-identity-reminder.js'; + +const SESSION_ID = 'ses_identity_1'; +const MODEL = { + id: 'test-model', + name: 'test-model', + api: 'anthropic-messages', + provider: 'test-provider', + baseUrl: 'https://example.invalid', + reasoning: false, + input: ['text'], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 200_000, + maxTokens: 8_192, +} as Model; + +const hook = createSessionIdentityReminderHook(new ContextUsageAnchorState()); + +function message(value: Record): AgentMessage { + return value as unknown as AgentMessage; +} + +function user(text: string, timestamp: number): AgentMessage { + return message({ role: 'user', content: [{ type: 'text', text }], timestamp }); +} + +const idBearingUser = user( + `\n YOUR SESSION ID: ${SESSION_ID}\n\nfirst ask`, + 1, +); +const checkpoint = message({ + role: 'compactionSummary', + summary: '## Goal\nKeep going', + tokensBefore: 100, + timestamp: 2, +}); +const currentUserWithoutId = user('\n date: today\n\nnext', 3); + +function hookInput( + messages: AgentMessage[], + phase: 'initial' | 'iteration' = 'iteration', +): PiBeforeLlmCallHookInput { + return { + sessionId: SESSION_ID, + turnId: 'turn-1', + phase, + messages, + canonicalMessages: messages, + model: MODEL, + thinkingLevel: 'off', + } as PiBeforeLlmCallHookInput; +} + +describe('Session identity reminder hook', () => { + it('stays silent while the model-visible context still shows the session ID', async () => { + expect(await hook(hookInput([idBearingUser, currentUserWithoutId]))).toBeUndefined(); + }); + + it('restores the ID before the current user when initial compaction summarized it away', async () => { + const decision = await hook(hookInput([checkpoint, currentUserWithoutId], 'initial')); + + expect(decision).toMatchObject({ + type: 'appendMessage', + reason: 'session_identity_reminder', + placement: 'before-current-user', + message: { + role: 'custom', + customType: 'session_identity_reminder', + display: false, + content: expect.stringContaining(`YOUR SESSION ID: ${SESSION_ID}`), + }, + }); + if (decision?.type !== 'appendMessage') throw new Error('Expected an append decision'); + expect(() => validateCanonicalHistoryMessages([decision.message])).not.toThrow(); + }); + + it('appends at the tail mid-turn and does not repeat once the marker is in context', async () => { + const decision = await hook(hookInput([checkpoint, currentUserWithoutId])); + if (decision?.type !== 'appendMessage') throw new Error('Expected an append decision'); + expect(decision).not.toHaveProperty('placement'); + + expect( + await hook(hookInput([checkpoint, currentUserWithoutId, decision.message])), + ).toBeUndefined(); + }); +}); + +describe('messagesCarrySessionId', () => { + it('only counts user messages or identity markers after the latest compaction', () => { + expect(messagesCarrySessionId([idBearingUser, currentUserWithoutId], SESSION_ID)).toBe(true); + expect( + messagesCarrySessionId([idBearingUser, checkpoint, currentUserWithoutId], SESSION_ID), + ).toBe(false); + expect( + messagesCarrySessionId( + [ + message({ + role: 'assistant', + content: [{ type: 'text', text: SESSION_ID }], + timestamp: 4, + }), + ], + SESSION_ID, + ), + ).toBe(false); + expect(messagesCarrySessionId([idBearingUser], '')).toBe(false); + }); +}); diff --git a/packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.ts b/packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.ts new file mode 100644 index 000000000..1baaf0618 --- /dev/null +++ b/packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.ts @@ -0,0 +1,36 @@ +import type { PiBeforeLlmCallHook } from '@mavis/agent-core/pi-turn-runner'; + +import { + SESSION_IDENTITY_REMINDER_CUSTOM_TYPE, + messagesCarrySessionId, +} from '../../agent-host/compaction/session-identity.js'; +import type { ContextUsageAnchorState } from '../../compaction/execution/usage-anchor.js'; +import { fitsReminderInFinalRequest } from './reminder-admission.js'; + +/** + * The turn reminder omits the session ID while earlier history still shows it, + * so compaction that summarizes that history away must restore the ID. + */ +export function createSessionIdentityReminderHook( + usageAnchor: ContextUsageAnchorState, +): PiBeforeLlmCallHook { + return (input) => { + if (!input.sessionId || messagesCarrySessionId(input.messages, input.sessionId)) { + return undefined; + } + const marker = { + role: 'custom' as const, + customType: SESSION_IDENTITY_REMINDER_CUSTOM_TYPE, + content: `\nYOUR SESSION ID: ${input.sessionId}\n`, + display: false as const, + timestamp: Date.now(), + }; + if (!fitsReminderInFinalRequest(input, marker, usageAnchor)) return undefined; + return { + type: 'appendMessage', + reason: SESSION_IDENTITY_REMINDER_CUSTOM_TYPE, + message: marker, + ...(input.phase === 'initial' ? { placement: 'before-current-user' as const } : {}), + }; + }; +} diff --git a/packages/local-runtime-v2/src/service/turn-system/production-composition.ts b/packages/local-runtime-v2/src/service/turn-system/production-composition.ts index 7c9db3dcd..4a418e787 100644 --- a/packages/local-runtime-v2/src/service/turn-system/production-composition.ts +++ b/packages/local-runtime-v2/src/service/turn-system/production-composition.ts @@ -84,6 +84,7 @@ import type { MessagesFileApiPatcherLogger } from './agent-host/assembly/message import { isProductionRuntimeErrorRetryable } from './runtime-error-retry-policy.js'; import { createTodoCadenceReminderHook } from './execution/reminder/todo-cadence-reminder.js'; import { createBackgroundCadenceReminder } from './execution/reminder/background-cadence-reminder.js'; +import { createSessionIdentityReminderHook } from './execution/reminder/session-identity-reminder.js'; import { fitsReminderInFinalRequest } from './execution/reminder/reminder-admission.js'; import { readSessionAgentExecutionSnapshot } from './agent-host/preparation/session-agent-execution-snapshot.js'; import { McpToolResultHistoryExternalizer } from './agent-host/execution/mcp-tool-result-history-externalizer.js'; @@ -444,6 +445,7 @@ export async function createLocalAgentHost< activatePluginHookSessionOwnership: options.pluginHookSessionOwnership.activate, afterCompactionBeforeLlmCallHooks: [ createAutomaticContextRequestFilterHook(), + createSessionIdentityReminderHook(contextUsageAnchor), createTodoCadenceReminderHook(contextUsageAnchor), ], onSteeringConsumed: options.onSteeringConsumed, diff --git a/release/public-source.json b/release/public-source.json index 06dd1d251..1a6a3ce09 100644 --- a/release/public-source.json +++ b/release/public-source.json @@ -1336,6 +1336,7 @@ "packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.test.ts", "packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/context-compaction.ts", "packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/contracts.ts", + "packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/session-identity.ts", "packages/local-runtime-v2/src/service/turn-system/agent-host/compaction/token-usage.ts", "packages/local-runtime-v2/src/service/turn-system/agent-host/contracts.ts", "packages/local-runtime-v2/src/service/turn-system/agent-host/empty-dependencies.ts", @@ -1441,6 +1442,8 @@ "packages/local-runtime-v2/src/service/turn-system/execution/queue-pause-on-abort.ts", "packages/local-runtime-v2/src/service/turn-system/execution/reminder/background-cadence-reminder.ts", "packages/local-runtime-v2/src/service/turn-system/execution/reminder/reminder-admission.ts", + "packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.test.ts", + "packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.ts", "packages/local-runtime-v2/src/service/turn-system/execution/reminder/todo-cadence-reminder.ts", "packages/local-runtime-v2/src/service/turn-system/execution/steering/steer-session.service.ts", "packages/local-runtime-v2/src/service/turn-system/execution/steering/steering-requeue.test.ts", diff --git a/test/vitest-suites.json b/test/vitest-suites.json index 9e54f59d1..1256adc80 100644 --- a/test/vitest-suites.json +++ b/test/vitest-suites.json @@ -117,6 +117,7 @@ "packages/tui/test/unit/runtime-error-presentation-rejection.test.ts", "packages/local-runtime-v2/src/service/turn-system/agent-host/assembly/local-turn-payload-transform.test.ts", "packages/local-runtime-v2/src/service/turn-system/agent-host/execution/executor.test.ts", + "packages/local-runtime-v2/src/service/turn-system/execution/reminder/session-identity-reminder.test.ts", "packages/local-runtime-v2/src/services.test.ts", "packages/local-runtime-v2/src/application/session/session-title-policy.test.ts", "packages/local-runtime-v2/test/unit/agent/builtin/catalog.test.ts",