diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index cc3a65b7a0..3de3d01097 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -6287,6 +6287,598 @@ describe('AiSdkBackend model history', () => { assert.doesNotMatch(prompt, /CODEX_WRONG_MODEL_STATE|cmp_wrong_model/); }); + test('replays a checkpoint whose covered prefix carries a stale tool-result transition (#4842)', async () => { + // The standalone compaction path pins the checkpoint's coverage digest on + // RAW RuntimeEvents, while pre-turn replay used to match it against the + // transition-folded view: any durable stale-result archive inside the + // covered prefix then failed the digest and the turn silently fell back to + // full-history replay. Replay now matches the raw view and folds the + // projected [block, tail] afterwards. + const model = completionModel(); + const transitions: ModelProjectionTransition[] = []; + const recorded: HistoryCompactCheckpoint[] = []; + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { + name: 'checkpoint-transition-replay-test', + charsPerToken: 1, + staleToolResultPrune: { + enabled: true, + maxResultEstimatedTokens: 1, + minRecentTurnsFull: 0, + }, + historyCompact: { enabled: true }, + }, + summarizeHistoryCompact: async () => structuredSummary('FOLDED_PREFIX_COMPACT_SENTINEL'), + recordHistoryCompactCheckpoint: (checkpoint) => { + recorded.push(checkpoint); + }, + loadHistoryCompactCheckpoint: () => recorded.at(-1), + toolResultArchive: testToolResultArchive({ + archiveToolResult: async (event) => ({ artifactId: `artifact-${event.runtimeEventId}` }), + }), + loadModelProjectionTransitions: async () => ({ + transitions: [...transitions], + unreadableTargets: new Set(), + unscopedUnreadable: 0, + }), + recordModelProjectionTransition: async (transition) => { + transitions.push(transition); + }, + }); + const priorEvents = [ + runtimeTextEvent({ + id: 'fold-old-user', + turnId: 'turn-old', + role: 'user', + author: 'user', + text: 'FOLD_OLD_USER_ALPHA '.repeat(60), + }), + runtimeEvent({ + id: 'fold-call', + turnId: 'turn-old', + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'tool-fold-1', + name: 'Read', + args: { path: 'a.ts' }, + }, + }), + runtimeEvent({ + id: 'fold-result', + turnId: 'turn-old', + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-fold-1', + name: 'Read', + result: { body: 'y'.repeat(400) }, + isError: false, + }, + }), + runtimeTextEvent({ + id: 'fold-recent-user', + turnId: 'turn-recent', + role: 'user', + author: 'user', + text: 'FOLD_RECENT_RETAINED_CONTEXT', + }), + ]; + + // Turn 1 commits the durable archive transition for the stale result. Each + // phase gets fresh clones: production readers deserialize their own event + // objects from the ledger, so no in-memory mutation can alias across them. + await drain( + backend.send({ + turnId: 'turn-seed', + text: 'seed the archive transition', + context: [], + runtimeContext: structuredClone(priorEvents), + }), + ); + assert.equal(transitions.length, 1); + + // Standalone compaction creates the checkpoint over the raw prefix, exactly + // like the production path whose input is begin.runtimeContext. + const compact = await backend.compactHistory({ + turnId: 'turn-compact', + runId: 'run-compact', + runtimeContext: structuredClone(priorEvents), + }); + assert.equal(compact.outcome.kind, 'compacted'); + assert.equal(recorded.length, 1); + + // The next turn must replay through the checkpoint, not fail open. A fresh + // backend mirrors production: the compaction operation and the next send + // run as separate runs with separate backend instances. + const replayBackend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { + name: 'checkpoint-transition-replay-test', + charsPerToken: 1, + staleToolResultPrune: { + enabled: true, + maxResultEstimatedTokens: 1, + minRecentTurnsFull: 0, + }, + historyCompact: { enabled: true }, + }, + loadHistoryCompactCheckpoint: () => recorded.at(-1), + toolResultArchive: testToolResultArchive({ + archiveToolResult: async (event) => ({ artifactId: `artifact-${event.runtimeEventId}` }), + }), + loadModelProjectionTransitions: async () => ({ + transitions: [...transitions], + unreadableTargets: new Set(), + unscopedUnreadable: 0, + }), + recordModelProjectionTransition: async (transition) => { + transitions.push(transition); + }, + }); + await drain( + replayBackend.send({ + turnId: 'turn-after-compact', + text: 'after compact', + context: [], + runtimeContext: structuredClone(priorEvents), + }), + ); + + const lastCall = model.doStreamCalls.at(-1); + const prompt = JSON.stringify( + lastCall?.prompt.map((message) => ({ role: message.role, content: message.content })), + ); + assert.match(prompt, /FOLDED_PREFIX_COMPACT_SENTINEL/); + assert.doesNotMatch(prompt, /FOLD_OLD_USER_ALPHA/); + }); + + test('a checkpoint summary cannot echo a body a durable transition removed (#4845)', async () => { + // Coverage identity is pinned on raw events, but the summarizer must read + // the EFFECTIVE (transition-folded) prefix: an echoing summarizer fed raw + // events would quote the archived body into the checkpoint block and every + // later replay would restore what the transition removed. + const model = completionModel(); + const transitions: ModelProjectionTransition[] = []; + const recorded: HistoryCompactCheckpoint[] = []; + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { + name: 'checkpoint-effective-summary-test', + charsPerToken: 1, + staleToolResultPrune: { + enabled: true, + maxResultEstimatedTokens: 1, + minRecentTurnsFull: 0, + }, + historyCompact: { enabled: true }, + }, + summarizeHistoryCompact: async (input) => + // Echo the covered span verbatim into a structurally valid summary. + structuredSummary( + `ECHO ${input.source.foldedRuntimeEvents + .map((event) => JSON.stringify(event.content)) + .join(' ')}`, + ), + recordHistoryCompactCheckpoint: (checkpoint) => { + recorded.push(checkpoint); + }, + loadHistoryCompactCheckpoint: () => recorded.at(-1), + toolResultArchive: testToolResultArchive({ + archiveToolResult: async (event) => ({ artifactId: `artifact-${event.runtimeEventId}` }), + }), + loadModelProjectionTransitions: async () => ({ + transitions: [...transitions], + unreadableTargets: new Set(), + unscopedUnreadable: 0, + }), + recordModelProjectionTransition: async (transition) => { + transitions.push(transition); + }, + }); + const priorEvents = [ + runtimeTextEvent({ + id: 'echo-old-user', + turnId: 'turn-old', + role: 'user', + author: 'user', + text: 'ECHO_OLD_USER '.repeat(60), + }), + runtimeEvent({ + id: 'echo-call', + turnId: 'turn-old', + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'tool-echo-1', + name: 'Read', + args: { path: 'secret.ts' }, + }, + }), + runtimeEvent({ + id: 'echo-result', + turnId: 'turn-old', + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-echo-1', + name: 'Read', + result: { body: 'RAW_TRANSITIONED_TOOL_BODY '.repeat(40) }, + isError: false, + }, + }), + ]; + + await drain( + backend.send({ + turnId: 'turn-seed', + text: 'seed the archive transition', + context: [], + runtimeContext: structuredClone(priorEvents), + }), + ); + assert.equal(transitions.length, 1); + + const compact = await backend.compactHistory({ + turnId: 'turn-compact', + runId: 'run-compact', + runtimeContext: structuredClone(priorEvents), + }); + assert.equal(compact.outcome.kind, 'compacted'); + assert.equal(recorded.length, 1); + + // The summary was written from the effective view: it carries the archive + // placeholder's identity, not the transitioned body. + const summary = recorded[0]?.version === 2 ? recorded[0].summary : ''; + assert.match(summary, /artifact-echo-result/); + assert.doesNotMatch(summary, /RAW_TRANSITIONED_TOOL_BODY/); + + const replayBackend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { + name: 'checkpoint-effective-summary-test', + charsPerToken: 1, + staleToolResultPrune: { + enabled: true, + maxResultEstimatedTokens: 1, + minRecentTurnsFull: 0, + }, + historyCompact: { enabled: true }, + }, + loadHistoryCompactCheckpoint: () => recorded.at(-1), + toolResultArchive: testToolResultArchive({}), + loadModelProjectionTransitions: async () => ({ + transitions: [...transitions], + unreadableTargets: new Set(), + unscopedUnreadable: 0, + }), + }); + await drain( + replayBackend.send({ + turnId: 'turn-after-compact', + text: 'after compact', + context: [], + runtimeContext: structuredClone(priorEvents), + }), + ); + + const lastCall = model.doStreamCalls.at(-1); + const prompt = JSON.stringify( + lastCall?.prompt.map((message) => ({ role: message.role, content: message.content })), + ); + assert.match(prompt, /ECHO /); + assert.doesNotMatch(prompt, /RAW_TRANSITIONED_TOOL_BODY/); + }); + + test('a transition committed after creation invalidates the checkpoint at pre-turn replay (#4845 review)', async () => { + // The checkpoint pins the EFFECTIVE digest of its covered prefix. A + // projection transition committed AFTER the fold (here: a later turn's + // stale-result prune) leaves the raw ledger untouched, so the identity + // match still passes — but the summary describes a view that no longer + // exists. Replay must reject the checkpoint and fail open to the + // effective history rather than restore the transitioned body. + const model = completionModel(); + const transitions: ModelProjectionTransition[] = []; + const recorded: HistoryCompactCheckpoint[] = []; + const echoSummarizer: Parameters[0]['summarizeHistoryCompact'] = + async (input) => + structuredSummary( + `ECHO ${input.source.foldedRuntimeEvents + .map((event) => JSON.stringify(event.content)) + .join(' ')}`, + ); + const loadTransitions = async () => ({ + transitions: [...transitions], + unreadableTargets: new Set(), + unscopedUnreadable: 0, + }); + const priorEvents = [ + runtimeTextEvent({ + id: 'echo-old-user', + turnId: 'turn-old', + role: 'user', + author: 'user', + text: 'ECHO_OLD_USER '.repeat(60), + }), + runtimeEvent({ + id: 'echo-call', + turnId: 'turn-old', + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'tool-echo-1', + name: 'Read', + args: { path: 'secret.ts' }, + }, + }), + runtimeEvent({ + id: 'echo-result', + turnId: 'turn-old', + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-echo-1', + name: 'Read', + result: { body: 'RAW_TRANSITIONED_TOOL_BODY '.repeat(40) }, + isError: false, + }, + }), + ]; + + // 1. Creation: no transition exists yet, so the effective view IS the raw + // view and the echo summary legitimately quotes the body into the block. + const creationBackend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: '[redacted]', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { + name: 'checkpoint-effective-drift-test', + charsPerToken: 1, + staleToolResultPrune: { enabled: false }, + historyCompact: { enabled: true }, + }, + summarizeHistoryCompact: echoSummarizer, + recordHistoryCompactCheckpoint: (checkpoint) => { + recorded.push(checkpoint); + }, + loadModelProjectionTransitions: loadTransitions, + }); + const compact = await creationBackend.compactHistory({ + turnId: 'turn-compact', + runId: 'run-compact', + runtimeContext: structuredClone(priorEvents), + }); + assert.equal(compact.outcome.kind, 'compacted'); + assert.equal(recorded.length, 1); + assert.ok(recorded[0]!.coverage.effectiveSourceDigest); + const summary = recorded[0]!.version === 2 ? recorded[0]!.summary : ''; + assert.match(summary, /RAW_TRANSITIONED_TOOL_BODY/); + + // 2. A later turn commits a stale-result transition over the covered span. + const pruneModel = completionModel(); + const pruneBackend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: '[redacted]', + modelId: 'mock-model-id', + modelFactory: () => pruneModel, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { + name: 'checkpoint-effective-drift-test', + charsPerToken: 1, + staleToolResultPrune: { + enabled: true, + maxResultEstimatedTokens: 1, + minRecentTurnsFull: 0, + }, + historyCompact: { enabled: true }, + }, + loadHistoryCompactCheckpoint: () => recorded.at(-1), + toolResultArchive: testToolResultArchive({ + archiveToolResult: async (event) => ({ artifactId: `artifact-${event.runtimeEventId}` }), + }), + loadModelProjectionTransitions: loadTransitions, + recordModelProjectionTransition: async (transition) => { + transitions.push(transition); + }, + }); + await drain( + pruneBackend.send({ + turnId: 'turn-seed', + text: 'seed the archive transition', + context: [], + runtimeContext: structuredClone(priorEvents), + }), + ); + assert.equal(transitions.length, 1); + + // 3. Replay: the raw identity still matches, the effective digest does + // not. The stale block must never reach the provider. + const replayModel = completionModel(); + const replayBackend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: '[redacted]', + modelId: 'mock-model-id', + modelFactory: () => replayModel, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { + name: 'checkpoint-effective-drift-test', + charsPerToken: 1, + staleToolResultPrune: { enabled: false }, + historyCompact: { enabled: true }, + }, + loadHistoryCompactCheckpoint: () => recorded.at(-1), + toolResultArchive: testToolResultArchive({}), + loadModelProjectionTransitions: loadTransitions, + }); + const events: unknown[] = []; + for await (const event of replayBackend.send({ + turnId: 'turn-after-transition', + text: 'after transition', + context: [], + runtimeContext: structuredClone(priorEvents), + })) { + events.push(event); + } + + const lastCall = replayModel.doStreamCalls.at(-1); + const prompt = JSON.stringify( + lastCall?.prompt.map((message) => ({ role: message.role, content: message.content })), + ); + // Fail-open replayed the effective history: the archive placeholder is + // visible, the stale summary and the transitioned body are not. + assert.match(prompt, /artifact-echo-result/); + assert.doesNotMatch(prompt, /ECHO /); + assert.doesNotMatch(prompt, /RAW_TRANSITIONED_TOOL_BODY/); + // The rejection is diagnosed so the fail-open note can name it. + const usageEvent = events.find( + (event) => (event as { type?: string }).type === 'token_usage', + ) as + | { + contextBudget?: { + compactionDecisions?: Array<{ decision?: string; failOpenReason?: string }>; + }; + } + | undefined; + const decisions = usageEvent?.contextBudget?.compactionDecisions ?? []; + assert.ok( + decisions.some( + (decision) => + decision.decision === 'failedOpen' && + decision.failOpenReason === 'effective_history_changed', + ), + ); + }); + + test('pre-turn replay withholds a tool result whose transition record is unreadable (#4845)', async () => { + // prepareContextBudgetPolicy folds with the unreadable-target set so an + // undecodable record withholds the body behind the failure sentinel; the + // post-match fold of the projected [block, tail] must do the same or it + // becomes the one consumer that replays the removed body. + const model = completionModel(); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { + name: 'unreadable-target-replay-test', + charsPerToken: 1, + staleToolResultPrune: { enabled: false }, + historyCompact: { enabled: true }, + }, + loadModelProjectionTransitions: async () => ({ + transitions: [], + unreadableTargets: new Set(['unreadable-result::tool_result']), + unscopedUnreadable: 0, + }), + }); + + await drain( + backend.send({ + turnId: 'turn-current', + text: 'current user', + context: [], + runtimeContext: [ + runtimeEvent({ + id: 'unreadable-call', + turnId: 'turn-prev', + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'tool-unreadable-1', + name: 'Read', + args: { path: 'a.ts' }, + }, + }), + runtimeEvent({ + id: 'unreadable-result', + turnId: 'turn-prev', + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-unreadable-1', + name: 'Read', + result: { body: 'RAW_UNREADABLE_TARGET_BODY' }, + isError: false, + }, + }), + ], + }), + ); + + const prompt = JSON.stringify(compactPrompt(model)); + assert.match(prompt, /could not be projected safely/); + assert.doesNotMatch(prompt, /RAW_UNREADABLE_TARGET_BODY/); + }); + test('keeps RuntimeEvent replay when a tool result is unmatched (orphan dropped, rest replayed)', async () => { // `unmatched_tool_result` is a non-blocking diagnostic: the materializer // drops the orphan itself (a standalone tool message is an Anthropic 400) diff --git a/packages/runtime/src/__tests__/history-compaction.test.ts b/packages/runtime/src/__tests__/history-compaction.test.ts index e4d66c4f10..ab11cc3545 100644 --- a/packages/runtime/src/__tests__/history-compaction.test.ts +++ b/packages/runtime/src/__tests__/history-compaction.test.ts @@ -534,6 +534,123 @@ describe('plan context compaction', () => { assert.deepEqual(seenNewlyFolded, ['call-a', 'res-a']); assert.equal(second.checkpoint.previousCheckpointId, first.checkpoint.checkpointId); }); + + test('keeps rolling forward when the effective coverage is unchanged', async () => { + const events = longTurnEvents(); + const longerEvents = [ + ...events, + call('call-c', 'cc', 'turn-1'), + result('res-c', 'cc', 'turn-1'), + ]; + const identityFold = async (covered: readonly RuntimeEvent[]) => [...covered]; + // Coverage ends at `res-a`: the first fold's covered span contains it. + const first = await planHistoryCompaction( + planInput({ orderedEvents: events, projectEffectiveCoverage: identityFold }), + ); + assert.equal(first.decision, 'compacted'); + if (first.decision !== 'compacted') return; + + let seenNewlyFolded: string[] = []; + const second = await planHistoryCompaction( + planInput({ + orderedEvents: longerEvents, + previousCheckpoint: first.checkpoint, + projectEffectiveCoverage: identityFold, + summarize: ({ newlyFoldedRuntimeEvents, previousCheckpoint }) => { + seenNewlyFolded = newlyFoldedRuntimeEvents.map((event) => event.id); + assert.equal(previousCheckpoint?.checkpointId, first.checkpoint.checkpointId); + return structuredSummary('rolled-forward summary'); + }, + }), + ); + assert.equal(second.decision, 'compacted'); + if (second.decision !== 'compacted') return; + assert.deepEqual(seenNewlyFolded, ['call-b', 'res-b']); + assert.equal(second.checkpoint.previousCheckpointId, first.checkpoint.checkpointId); + }); + + test('discards a previous checkpoint whose effective coverage drifted before roll-forward (#4845 review)', async () => { + const events = longTurnEvents(); + const longerEvents = [ + ...events, + call('call-c', 'cc', 'turn-1'), + result('res-c', 'cc', 'turn-1'), + ]; + const identityFold = async (covered: readonly RuntimeEvent[]) => [...covered]; + // First fold: no transition exists, so the effective view IS the raw view + // and the checkpoint (covering through `res-a`) pins that digest. + const first = await planHistoryCompaction( + planInput({ orderedEvents: events, projectEffectiveCoverage: identityFold }), + ); + assert.equal(first.decision, 'compacted'); + if (first.decision !== 'compacted') return; + + // A cross-turn projection transition then archives `res-a` — INSIDE the + // first checkpoint's coverage — leaving the raw prefix untouched. The + // inherited summary still quotes the raw body, but the view it describes + // no longer exists. + const foldWithArchive = async (covered: readonly RuntimeEvent[]): Promise => + covered.map((event) => { + if (event.id !== 'res-a') return event; + const content = event.content as Extract< + RuntimeEvent['content'], + { kind: 'function_response' } + >; + return { + ...event, + content: { + ...content, + modelProjection: { + version: 1 as const, + kind: 'text' as const, + text: '[archived: artifact-res-a]', + }, + }, + }; + }); + + let summarizeSawPrevious: string | undefined; + let seenCovered: string[] = []; + let seenNewlyFolded: string[] = []; + const second = await planHistoryCompaction( + planInput({ + orderedEvents: longerEvents, + previousCheckpoint: first.checkpoint, + projectEffectiveCoverage: foldWithArchive, + summarize: ({ coveredRuntimeEvents, newlyFoldedRuntimeEvents, previousCheckpoint }) => { + summarizeSawPrevious = previousCheckpoint?.checkpointId; + seenCovered = coveredRuntimeEvents.map((event) => event.id); + seenNewlyFolded = newlyFoldedRuntimeEvents.map((event) => event.id); + return structuredSummary('re-summarized effective span'); + }, + }), + ); + assert.equal(second.decision, 'compacted'); + if (second.decision !== 'compacted') return; + + // The drifted checkpoint is discarded: nothing is inherited (the text + // summarizer cannot prepend the stale summary, the codex path receives no + // stale provider state), the whole effective span is re-summarized, and + // the new checkpoint records no lineage to the stale one. + assert.equal(summarizeSawPrevious, undefined); + assert.deepEqual(seenNewlyFolded, seenCovered); + assert.deepEqual(seenCovered, [ + 'prior-0', + 'prior-1', + 'anchor', + 'call-a', + 'res-a', + 'call-b', + 'res-b', + ]); + assert.equal(second.checkpoint.previousCheckpointId, undefined); + // Coverage identity stays raw: the new checkpoint still matches the ledger + // prefix, now with the drifted effective digest pinned. + assert.equal( + matchHistoryCompactCheckpointPrefix(second.checkpoint, longerEvents.slice(0, 7)).reason, + undefined, + ); + }); }); function base(id: string, turnId: string): Omit { diff --git a/packages/runtime/src/ai-sdk-compaction.ts b/packages/runtime/src/ai-sdk-compaction.ts index 4199e4541a..aa781d6c78 100644 --- a/packages/runtime/src/ai-sdk-compaction.ts +++ b/packages/runtime/src/ai-sdk-compaction.ts @@ -48,12 +48,14 @@ import { buildContextBudgetDiagnosticShell, estimateRuntimeEventsTokens, mergeContextBudgetDiagnostic, + mergeContextBudgetDiagnosticPatches, type ContextBudgetPolicy, } from './context-budget.js'; import { isHistoryCompactContentEvent } from './history-compaction.js'; import { canContinueHistoryCompactCheckpointForModel, canReplayHistoryCompactCheckpointForModel, + historyCompactSourceDigest, matchHistoryCompactCheckpointPrefix, projectHistoryCompactCheckpointReplay, type HistoryCompactCheckpoint, @@ -383,8 +385,17 @@ export class AiSdkCompaction { : {}), ...(automaticMemoryBoundary ? { memoryExtractionBoundary: automaticMemoryBoundary } : {}), ...(previousCheckpoint ? { previousCheckpoint } : {}), - summarize: async ({ coveredRuntimeEvents, newlyFoldedRuntimeEvents, previousCheckpoint }) => - await this.summarizeWithFailureCircuit(summarizer, { + // The planner projects the covered span to its effective view before + // summarizing and pins its digest as coverage.effectiveSourceDigest: + // the summary can never quote a body a durable transition removed, and + // a later transition invalidates the checkpoint at replay (#4845). + projectEffectiveCoverage: (covered) => this.foldEffectiveModelHistory(covered), + summarize: async ({ + coveredRuntimeEvents, + newlyFoldedRuntimeEvents, + previousCheckpoint, + }) => { + return await this.summarizeWithFailureCircuit(summarizer, { sessionId: this.sessionId, turnId: input.turnId, runId: input.runId, @@ -398,7 +409,8 @@ export class AiSdkCompaction { ...(previousCheckpoint ? { previousCheckpoint } : {}), abortSignal: historyCompactAbortController.signal, ...(tracker ? { providerRequestTracker: tracker } : {}), - }), + }); + }, }); if (historyCompactAbortController.signal.aborted) { return { outcome: { kind: 'failed', reason: 'aborted' } }; @@ -543,8 +555,36 @@ export class AiSdkCompaction { */ public async foldEffectiveModelHistory(events: readonly RuntimeEvent[]): Promise { const loaded = await this.loadModelProjectionTransitions(); - if (loaded.transitions.length === 0) return [...events]; - return reduceEffectiveModelProjections(events, loaded.transitions).events; + if (loaded.transitions.length === 0 && loaded.unreadableTargets.size === 0) { + return [...events]; + } + // Forward the unreadable set: a target whose transition record this build + // cannot decode must fold to the withholding sentinel here too, or this + // path becomes the one consumer that replays the body a record removed. + return reduceEffectiveModelProjections(events, loaded.transitions, loaded.unreadableTargets) + .events; + } + + /** + * Whether the checkpoint's pinned effective view still is the covered + * prefix's effective view. The raw identity match happens later in the + * replay authority; this gate is about content currency: a projection + * transition committed after the fold changes what the model may see of the + * covered span without touching the raw ledger, and the checkpoint's summary + * or provider state must not survive that drift (#4845 review). + */ + private checkpointEffectiveCoverageMatches( + checkpoint: HistoryCompactCheckpoint, + effectiveEvents: readonly RuntimeEvent[], + ): boolean { + const pinned = checkpoint.coverage.effectiveSourceDigest; + if (pinned === undefined) return false; + const covered = effectiveEvents + .filter(isHistoryCompactContentEvent) + .slice(0, checkpoint.coverage.eventCount); + if (covered.length !== checkpoint.coverage.eventCount) return false; + if (covered.at(-1)?.id !== checkpoint.coverage.through.runtimeEventId) return false; + return historyCompactSourceDigest(covered) === pinned; } /** @@ -662,10 +702,43 @@ export class AiSdkCompaction { this.input.modelId, ) ) { - nextPolicy = { - ...nextPolicy, - historyCompact: { ...nextPolicy.historyCompact!, checkpoint: loadedCheckpoint }, - }; + // Raw identity first: a coverage miss keeps the apply-stage path, whose + // matcher reports the real identity reason (coverage_miss and friends). + // This gate is only for the case where the raw identity matches but a + // projection transition committed after the fold changed the effective + // view the summary (or provider state) was built from (#4845 review). + const rawIdentityMatch = matchHistoryCompactCheckpointPrefix( + loadedCheckpoint, + runtimeContext.filter(isHistoryCompactContentEvent), + ); + if (rawIdentityMatch.reason) { + nextPolicy = { + ...nextPolicy, + historyCompact: { ...nextPolicy.historyCompact!, checkpoint: loadedCheckpoint }, + }; + } else if (this.checkpointEffectiveCoverageMatches(loadedCheckpoint, effective.events)) { + nextPolicy = { + ...nextPolicy, + historyCompact: { ...nextPolicy.historyCompact!, checkpoint: loadedCheckpoint }, + }; + } else { + // Rejecting keeps the stale block from restoring what the transition + // removed; the next fold re-summarizes (#4845 review). + diagnosticPatch = mergeContextBudgetDiagnosticPatches( + diagnosticPatch, + compactionDecisionDiagnosticPatch({ + stage: 'priorReplay', + sourceKind: 'runtimeEvents', + decision: 'failedOpen', + phase: 'pre_turn', + boundaryKind: 'historyCompact', + ...(loadedCheckpoint.coverage.effectiveSourceDigest !== undefined + ? { boundaryIds: [loadedCheckpoint.checkpointId] } + : {}), + failOpenReason: 'effective_history_changed', + }), + ); + } } return { policy: nextPolicy, @@ -1104,7 +1177,12 @@ export class AiSdkCompaction { }, } : {}), + projectEffectiveCoverage: (covered) => this.foldEffectiveModelHistory(covered), summarize: async ({ coveredRuntimeEvents, newlyFoldedRuntimeEvents, previousCheckpoint }) => { + // Same contract as the standalone path: the planner hands the + // effective (transition-folded) view to the summarizer and pins its + // digest, so a summary can never quote a body a durable transition + // removed (#4845). return await this.summarizeWithFailureCircuit(summarizer, { sessionId: this.sessionId, turnId, diff --git a/packages/runtime/src/ai-sdk-turn.ts b/packages/runtime/src/ai-sdk-turn.ts index 8ae5270101..b3e5400614 100644 --- a/packages/runtime/src/ai-sdk-turn.ts +++ b/packages/runtime/src/ai-sdk-turn.ts @@ -166,6 +166,7 @@ import { import { isHistoryCompactContentEvent } from './history-compaction.js'; import { canContinueHistoryCompactCheckpointForModel, + historyCompactSourceDigest, isProviderHistoryCompactCheckpoint, matchHistoryCompactCheckpointPrefix, projectHistoryCompactCheckpointReplay, @@ -1357,6 +1358,7 @@ export class AiSdkTurn { ] : turnEvents; let replayEvents = rawProjectionEvents; + let effectiveProjectionCheckpoint = projectionCheckpoint; if (projectionCheckpoint) { const checkpointMatch = matchHistoryCompactCheckpointPrefix( projectionCheckpoint, @@ -1365,11 +1367,27 @@ export class AiSdkTurn { if (checkpointMatch.reason) { throw new Error(`durable checkpoint projection mismatch: ${checkpointMatch.reason}`); } - replayEvents = projectHistoryCompactCheckpointReplay( - projectionCheckpoint, + // Content-currency guard: the raw identity still matches, but a + // transition committed after this fold (e.g. an active-turn prune + // in this very send) changed the effective view the block was + // built from. Replay without the stale block — the provider + // decides fit and overflow recovery re-folds (#4845 review). + const pinnedEffectiveDigest = projectionCheckpoint.coverage.effectiveSourceDigest; + const coveredEffective = await this.deps.compaction.foldEffectiveModelHistory( checkpointMatch.coveredRuntimeEvents, - checkpointMatch.successorRuntimeEvents, ); + if ( + pinnedEffectiveDigest === undefined || + historyCompactSourceDigest(coveredEffective) !== pinnedEffectiveDigest + ) { + effectiveProjectionCheckpoint = undefined; + } else { + replayEvents = projectHistoryCompactCheckpointReplay( + projectionCheckpoint, + checkpointMatch.coveredRuntimeEvents, + checkpointMatch.successorRuntimeEvents, + ); + } // The checkpoint was capacity-validated before it was persisted. // Do not re-run that gate against a later, larger successor tail: // the active-step shaper must see that growth so it can roll the @@ -1398,7 +1416,7 @@ export class AiSdkTurn { await this.deps.messageProjection.materializeRuntimeReplayPlan( replayPlan, this.imageBudget, - projectionCheckpoint, + effectiveProjectionCheckpoint, compatibleProviderReasoningReplayEventIds( replayEvents, input.runtimeContextInvocations, @@ -1407,7 +1425,7 @@ export class AiSdkTurn { this.runId, ), ); - return projectionCheckpoint + return effectiveProjectionCheckpoint ? currentTurnMessages : [...priorReplay.messages, ...currentTurnMessages]; }; @@ -2866,8 +2884,18 @@ export class AiSdkTurn { this.deps.backend.modelId, ); let contextBudget = preparedContextBudget.policy; - const budgeted = applyRuntimeEventContextBudget(priorRuntimeContext, contextBudget); - let runtimeContext = budgeted?.events ?? priorRuntimeContext; + // Match the durable checkpoint against the RAW ledger prefix: every + // creation path (standalone compactHistory and the mid-turn state) pins + // its coverage digest on raw events, so matching the folded view here + // lets any durable projection transition inside the covered prefix orphan + // the checkpoint and silently fail open into a full-history replay + // (#4842). The projected [block, tail] is then folded through the + // transition reducer before it becomes messages, so a committed + // transition still cannot resurrect content for the model (#4283). + const budgeted = applyRuntimeEventContextBudget(rawPriorRuntimeContext, contextBudget); + let runtimeContext = await this.deps.compaction.foldEffectiveModelHistory( + budgeted?.events ?? rawPriorRuntimeContext, + ); let contextBudgetDiagnostic = budgeted?.diagnostic; let projectedHistoryCompactCheckpoint = budgeted?.historyCompactCheckpoint; if (preparedContextBudget.diagnosticPatch) { diff --git a/packages/runtime/src/history-compact-checkpoint.ts b/packages/runtime/src/history-compact-checkpoint.ts index e19f64d4f2..9d78e31682 100644 --- a/packages/runtime/src/history-compact-checkpoint.ts +++ b/packages/runtime/src/history-compact-checkpoint.ts @@ -29,13 +29,16 @@ import { type SectionedSummaryFormat, } from './history-compact-summary-validation.js'; -// v2: coverage and source digest are taken over EFFECTIVE model history (the -// durable Tool Result projection), not raw RuntimeEvent evidence. A v1 -// checkpoint's digest was computed over a different source, so it fails the -// shape check and its session re-summarizes rather than replaying a -// coverage claim this policy never made. +// v3: coverage identity and the source digest stay pinned on the RAW ledger +// prefix — the immutable view every creation path (standalone and mid-turn) +// and both match sites (pre-turn replay and the mid-turn durable projection) +// share — while the model-visible summary is produced from the EFFECTIVE, +// transition-folded prefix. A v2 checkpoint's summary could quote a Tool +// Result body a durable projection transition had already removed, so it is +// superseded on load and the session re-summarizes rather than replaying a +// coverage claim whose summary this policy never audited (#4845). export const HISTORY_COMPACT_SOURCE_POLICY_VERSION = - 'maka.compactable_runtime_event_projection.v2' as const; + 'maka.compactable_runtime_event_projection.v3' as const; export interface HistoryCompactCheckpointSource { schemaVersion: 1; kind: 'runtime_event_projection'; @@ -53,6 +56,15 @@ export interface HistoryCompactCheckpointCoverage { runtimeEventId: string; }; sourceDigest: string; + /** + * Digest of the EFFECTIVE (transition-folded) view of the covered prefix at + * creation time — the exact content the summary or provider state describes. + * Replay must reject the checkpoint when the current effective view no + * longer matches: a projection transition committed after creation changes + * what the model may see without touching the raw ledger, and a stale block + * would restore what the transition removed (#4845 review). + */ + effectiveSourceDigest?: string; } /** @@ -134,6 +146,13 @@ export type HistoryCompactCheckpoint = interface BuildHistoryCompactCheckpointBaseInput { sessionId: string; coveredRuntimeEvents: readonly RuntimeEvent[]; + /** + * The effective (transition-folded) view of `coveredRuntimeEvents`, when the + * caller folded it. Its digest is pinned as `coverage.effectiveSourceDigest`; + * without it the raw coverage doubles as the effective view, which is only + * true when no projection transition touches the span. + */ + effectiveCoveredRuntimeEvents?: readonly RuntimeEvent[]; highWaterName?: string; highWaterSeq?: number; previousCheckpointId?: string; @@ -271,6 +290,9 @@ export function buildHistoryCompactCheckpoint( runtimeEventId: lastEvent.id, }, sourceDigest: historyCompactSourceDigest(input.coveredRuntimeEvents), + effectiveSourceDigest: historyCompactSourceDigest( + input.effectiveCoveredRuntimeEvents ?? input.coveredRuntimeEvents, + ), }; const highWaterName = input.highWaterName ?? 'history-compact-high-water'; const highWaterSeq = input.highWaterSeq ?? createdAt; @@ -455,6 +477,11 @@ export function validateHistoryCompactCheckpointShape( nonEmpty(through?.turnId) && nonEmpty(through?.runtimeEventId) && nonEmpty(coverage?.sourceDigest) && + // A checkpoint minted under the current source policy pins the effective + // (transition-folded) view its summary or provider state describes; + // without the digest there is no content-currency binding at replay, so + // the record must not validate (#4845 review). + (checkpoint.source === undefined || nonEmpty(coverage?.effectiveSourceDigest)) && (checkpoint.source === undefined || validHistoryCompactCheckpointSource(checkpoint.source, checkpoint.sessionId, coverage)) && (checkpoint.phase === undefined || @@ -737,7 +764,7 @@ function sameHistoryCompactSourceCoverage( * response's effective projection does change the digest, so a checkpoint can * never be replayed over content it never covered. */ -function historyCompactSourceDigest(events: readonly RuntimeEvent[]): string { +export function historyCompactSourceDigest(events: readonly RuntimeEvent[]): string { const hash = createHash('sha256'); for (const event of events) { const serialized = stableStringify(effectiveDigestEvent(event)); diff --git a/packages/runtime/src/history-compaction.ts b/packages/runtime/src/history-compaction.ts index de33cc1280..69210d4893 100644 --- a/packages/runtime/src/history-compaction.ts +++ b/packages/runtime/src/history-compaction.ts @@ -31,6 +31,7 @@ import { findCheckpointSummaryDefect } from './history-compact-summary-validatio import { buildHistoryCompactCheckpoint, historyCompactCheckpointToRuntimeEvent, + historyCompactSourceDigest, matchHistoryCompactCheckpointPrefix, midTurnHeadAnchorEvent, projectHistoryCompactCheckpointReplay, @@ -194,6 +195,15 @@ export interface PlanHistoryCompactionInput { acceptedRoute?: { modelId: string; connectionId?: string }; /** Present only when this automatic Compaction should create a Memory task. */ memoryExtractionBoundary?: HistoryCompactMemoryExtractionBoundary; + /** + * Projects the covered span to its effective (transition-folded) view. When + * present, the summary is written from that view and its digest is pinned as + * `coverage.effectiveSourceDigest`, so a later projection transition + * invalidates the checkpoint instead of being restored by it (#4845 review). + */ + projectEffectiveCoverage?: ( + coveredRuntimeEvents: readonly RuntimeEvent[], + ) => Promise; summarize: HistoryCompactionSummarizer; } @@ -308,18 +318,47 @@ export async function planHistoryCompaction( const checkpointMatch = input.previousCheckpoint ? matchHistoryCompactCheckpointPrefix(input.previousCheckpoint, coveredRuntimeEvents) : undefined; - const previousCheckpoint = + let previousCheckpoint = checkpointMatch && !checkpointMatch.reason ? input.previousCheckpoint : undefined; + // Content-currency gate for roll-forward: the inherited summary or + // provider state describes the EFFECTIVE view of the previous coverage at + // its own creation. A projection transition committed since rewrites that + // view without touching the raw prefix, and reusing the stale content here + // would launder it into the new checkpoint under the current effective + // digest — later replay guards would then pass it (#4845 review). On + // drift, discard the checkpoint and re-summarize the whole effective span. + if (previousCheckpoint && checkpointMatch && input.projectEffectiveCoverage) { + const pinned = previousCheckpoint.coverage.effectiveSourceDigest; + const previousEffectiveCovered = await input.projectEffectiveCoverage( + checkpointMatch.coveredRuntimeEvents, + ); + if (pinned === undefined || historyCompactSourceDigest(previousEffectiveCovered) !== pinned) { + previousCheckpoint = undefined; + } + } const newlyFoldedRuntimeEvents = previousCheckpoint ? checkpointMatch!.successorRuntimeEvents : coveredRuntimeEvents; + // The model-visible summary reads the effective (transition-folded) view + // of the covered span; the raw events keep the coverage identity. + const effectiveCoveredRuntimeEvents = input.projectEffectiveCoverage + ? [...(await input.projectEffectiveCoverage(coveredRuntimeEvents))] + : undefined; + const effectiveNewlyFoldedRuntimeEvents = effectiveCoveredRuntimeEvents + ? newlyFoldedRuntimeEvents.length === coveredRuntimeEvents.length + ? effectiveCoveredRuntimeEvents + : effectiveCoveredRuntimeEvents.slice( + effectiveCoveredRuntimeEvents.length - newlyFoldedRuntimeEvents.length, + ) + : undefined; + let compacted: string | HistoryCompactProviderState | undefined; try { compacted = await Promise.resolve( input.summarize({ - coveredRuntimeEvents, - newlyFoldedRuntimeEvents, + coveredRuntimeEvents: effectiveCoveredRuntimeEvents ?? coveredRuntimeEvents, + newlyFoldedRuntimeEvents: effectiveNewlyFoldedRuntimeEvents ?? newlyFoldedRuntimeEvents, ...(previousCheckpoint ? { previousCheckpoint } : {}), }), ); @@ -378,6 +417,7 @@ export async function planHistoryCompaction( const checkpoint = buildHistoryCompactCheckpoint({ sessionId: input.sessionId, coveredRuntimeEvents, + ...(effectiveCoveredRuntimeEvents ? { effectiveCoveredRuntimeEvents } : {}), ...(typeof compacted === 'string' ? { summary: compacted } : { providerState: compacted }), ...(phase === 'mid_turn' ? { phase: 'mid_turn' as const, headAnchor: input.headAnchor! }