diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index dc03aea7be..c440de253b 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -1168,7 +1168,13 @@ export interface SystemNoteMessage { | 'step_limit' | 'error' | 'abort'; - /** Shape depends on `kind`. */ + /** + * Shape depends on `kind`. `context_compaction_failed_open` carries + * `{ failOpenReason?: string }` — the reason the fold was refused (e.g. + * `coverage_miss`, `source_hash_mismatch`); when a turn is stopped before + * settlement, this note is the only durable record of the reason, because + * the `token_usage` diagnostic is never written (#4850). + */ data?: unknown; } diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 055875101e..cc3a65b7a0 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -5518,6 +5518,240 @@ describe('AiSdkBackend model history', () => { ); }); + test('persists the compaction fail-open note at decision time, before any settlement (#4850)', async () => { + // The replay fail-open decision is known at turn start; a stop before + // settlement skips usage persistence entirely, so a settlement-time note + // would never reach the transcript. + const gate = makeGate(); + const model = new MockLanguageModelV4({ + doStream: { + stream: new ReadableStream({ + async start(controller) { + controller.enqueue({ type: 'stream-start', warnings: [] }); + controller.enqueue({ type: 'text-start', id: 'text-1' }); + controller.enqueue({ type: 'text-delta', id: 'text-1', delta: 'PARTIAL' }); + // Hold the finish back so the send never reaches settlement until + // the test releases the gate. + await gate.promise; + controller.enqueue({ type: 'text-end', id: 'text-1' }); + controller.enqueue({ + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: { + inputTokens: { total: 1, noCache: 1, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 1, text: 1, reasoning: 0 }, + }, + }); + controller.close(); + }, + }), + }, + }); + // A checkpoint whose covered prefix does not match the replayed events: + // the pre-turn replay fails open with a coverage miss. + const checkpoint = buildHistoryCompactCheckpoint({ + sessionId: 'session-1', + coveredRuntimeEvents: [ + runtimeTextEvent({ + id: 'unrelated-covered', + turnId: 'turn-unrelated', + role: 'user', + author: 'user', + text: 'UNRELATED_COVERED '.repeat(50), + }), + ], + summary: structuredSummary('STALE_CHECKPOINT_SENTINEL'), + }); + const appended: Array<{ type: string; kind?: string; data?: unknown }> = []; + const isFailOpenNote = (message: { type: string; kind?: string }): boolean => + message.type === 'system_note' && message.kind === 'context_compaction_failed_open'; + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async (message: StoredMessage) => { + appended.push(message as unknown as { type: string; kind?: string; data?: unknown }); + }, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { historyCompact: { enabled: true } }, + loadHistoryCompactCheckpoint: () => checkpoint, + }); + + const sendPromise = drain( + backend.send({ + turnId: 'turn-1', + text: 'hi', + context: [], + runtimeContext: [ + runtimeTextEvent({ + id: 'rt-real-history', + turnId: 'turn-prev', + role: 'user', + author: 'user', + text: 'REAL_HISTORY '.repeat(60), + }), + ], + }), + ); + // The decision-time write precedes the provider stream's finish: wait for + // the note itself, not for any stream signal. On a settlement-only + // implementation this wait can only time out, which is the regression. + try { + await pollFor(() => appended.some(isFailOpenNote), { + timeoutMs: 10_000, + message: 'fail-open note was not written before settlement', + }); + } finally { + await backend.stop('user_stop'); + gate.release(); + } + await sendPromise; + + const note = appended.find(isFailOpenNote); + assert.ok(note, 'the fail-open note must be persisted even though the turn never settled'); + assert.equal( + (note?.data as { failOpenReason?: string } | undefined)?.failOpenReason, + 'coverage_miss', + ); + // Settlement never ran: no usage was persisted, and the note did not wait + // for it. + assert.equal( + appended.some((message) => message.type === 'token_usage'), + false, + ); + }); + + test('writes the compaction fail-open note exactly once when the send settles (#4850)', async () => { + const model = completionModel(); + const checkpoint = buildHistoryCompactCheckpoint({ + sessionId: 'session-1', + coveredRuntimeEvents: [ + runtimeTextEvent({ + id: 'settle-unrelated-covered', + turnId: 'turn-unrelated', + role: 'user', + author: 'user', + text: 'SETTLE_UNRELATED_COVERED '.repeat(50), + }), + ], + summary: structuredSummary('SETTLE_STALE_CHECKPOINT_SENTINEL'), + }); + const appended: Array<{ type: string; kind?: string }> = []; + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async (message: StoredMessage) => { + appended.push(message as unknown as { type: string; kind?: string }); + }, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { historyCompact: { enabled: true } }, + loadHistoryCompactCheckpoint: () => checkpoint, + }); + + await drain( + backend.send({ + turnId: 'turn-1', + text: 'hi', + context: [], + runtimeContext: [ + runtimeTextEvent({ + id: 'rt-settle-history', + turnId: 'turn-prev', + role: 'user', + author: 'user', + text: 'SETTLE_REAL_HISTORY '.repeat(60), + }), + ], + }), + ); + + const notes = appended.filter( + (message) => + message.type === 'system_note' && message.kind === 'context_compaction_failed_open', + ); + assert.equal(notes.length, 1, 'the settlement fallback must not duplicate the early note'); + }); + + test('a failed decision-time note write still leaves the settlement fallback armed (#4850)', async () => { + // The per-send flag must rise only after the append lands: a failed + // decision-time write falls through to settlement instead of losing the + // note for the whole send. + const model = completionModel(); + const checkpoint = buildHistoryCompactCheckpoint({ + sessionId: 'session-1', + coveredRuntimeEvents: [ + runtimeTextEvent({ + id: 'fallback-unrelated-covered', + turnId: 'turn-unrelated', + role: 'user', + author: 'user', + text: 'FALLBACK_UNRELATED_COVERED '.repeat(50), + }), + ], + summary: structuredSummary('FALLBACK_STALE_CHECKPOINT_SENTINEL'), + }); + const isFailOpenNote = (message: { type: string; kind?: string }): boolean => + message.type === 'system_note' && message.kind === 'context_compaction_failed_open'; + const persisted: Array<{ type: string; kind?: string }> = []; + let noteWriteAttempts = 0; + let failNextNoteWrite = true; + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async (message: StoredMessage) => { + const candidate = message as unknown as { type: string; kind?: string }; + if (isFailOpenNote(candidate)) { + noteWriteAttempts += 1; + if (failNextNoteWrite) { + failNextNoteWrite = false; + throw new Error('storage hiccup'); + } + } + persisted.push(candidate); + }, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + contextBudget: { historyCompact: { enabled: true } }, + loadHistoryCompactCheckpoint: () => checkpoint, + }); + + await drain( + backend.send({ + turnId: 'turn-1', + text: 'hi', + context: [], + runtimeContext: [ + runtimeTextEvent({ + id: 'rt-fallback-history', + turnId: 'turn-prev', + role: 'user', + author: 'user', + text: 'FALLBACK_REAL_HISTORY '.repeat(60), + }), + ], + }), + ); + + assert.equal(noteWriteAttempts, 2, 'the failed early write must be retried at settlement'); + assert.equal(persisted.filter(isFailOpenNote).length, 1); + }); + test('after-step stop preserves the current provider step usage and prevents another step', async () => { const loop = countingToolLoopModel(); const durable = durableTurnHarness('turn-1', 'hi'); diff --git a/packages/runtime/src/ai-sdk-turn.ts b/packages/runtime/src/ai-sdk-turn.ts index c501a66ac7..8ae5270101 100644 --- a/packages/runtime/src/ai-sdk-turn.ts +++ b/packages/runtime/src/ai-sdk-turn.ts @@ -1004,6 +1004,56 @@ export class AiSdkTurn { let contextReportedWindowNoteWritten = false; let contextOverflowAfterCompactionNoteWritten = false; let contextWindowSuggestionNoteWritten = false; + // A compaction decision is known the moment its stage reports it — the + // pre-turn replay resolves its fold before the first request goes out — + // while the settlement path is skipped entirely by a stop or a stream + // error. Write both notes when the decision is known, once per send, + // whichever stage reports first (#4850). + const appendCompactionDecisionNotes = async ( + contextBudget: ContextBudgetDiagnostic | undefined, + ): Promise => { + if ( + !contextCompactionFailedOpenNoteWritten && + shouldAppendContextCompactionFailedOpenNote(contextBudget) + ) { + // The most recent stage that refused the fold: a send can carry both a + // priorReplay and an activeStep refusal after a diagnostic merge, and + // array order would pin the stale one. + const failOpenReason = contextBudget?.compactionDecisions + ?.filter( + (decision) => + decision.boundaryKind === 'historyCompact' && decision.decision === 'failedOpen', + ) + .at(-1)?.failOpenReason; + const note: SystemNoteMessage = { + type: 'system_note', + id: this.deps.newId(), + turnId, + ts: this.deps.now(), + kind: 'context_compaction_failed_open', + ...(failOpenReason !== undefined ? { data: { failOpenReason } } : {}), + }; + // Mark written only after the append lands: a failed write must leave + // the flag down so the settlement fallback can still record the note. + contextCompactionFailedOpenNoteWritten = await this.deps.backend + .appendMessage(note) + .then(() => true) + .catch(() => false); + } + if (!contextCompactedNoteWritten && shouldAppendContextCompactedNote(contextBudget)) { + const note: SystemNoteMessage = { + type: 'system_note', + id: this.deps.newId(), + turnId, + ts: this.deps.now(), + kind: 'context_compacted', + }; + contextCompactedNoteWritten = await this.deps.backend + .appendMessage(note) + .then(() => true) + .catch(() => false); + } + }; // Request index (0-based) at which the active prune last rewrote the // request. A step Maka pruned is not append-only, so usage may legitimately // shrink. @@ -1189,6 +1239,10 @@ export class AiSdkTurn { yield* this.drain(queue); return; } + // The pre-turn replay's fold decision is final here: surface it now so a + // stop or stream error later in the send cannot keep it from the + // transcript (#4850). + await appendCompactionDecisionNotes(priorReplay.contextBudget); if (midTurnState) { // Roll-forward seed: the latest durable checkpoint (loaded or written at // turn start) so a mid-turn summary only re-reads the newly folded span. @@ -2489,34 +2543,10 @@ export class AiSdkTurn { ...usageFields, }; await this.deps.backend.appendMessage(tu).catch(() => {}); - if ( - !contextCompactionFailedOpenNoteWritten && - shouldAppendContextCompactionFailedOpenNote(contextBudgetForUsage) - ) { - contextCompactionFailedOpenNoteWritten = true; - const note: SystemNoteMessage = { - type: 'system_note', - id: this.deps.newId(), - turnId, - ts: this.deps.now(), - kind: 'context_compaction_failed_open', - }; - await this.deps.backend.appendMessage(note).catch(() => {}); - } - if ( - !contextCompactedNoteWritten && - shouldAppendContextCompactedNote(contextBudgetForUsage) - ) { - contextCompactedNoteWritten = true; - const note: SystemNoteMessage = { - type: 'system_note', - id: this.deps.newId(), - turnId, - ts: this.deps.now(), - kind: 'context_compacted', - }; - await this.deps.backend.appendMessage(note).catch(() => {}); - } + // Settlement fallback: a mid-turn or request-hook fold is only + // known here. Notes already written at decision time are skipped + // by the flags inside. + await appendCompactionDecisionNotes(contextBudgetForUsage); queue.push({ type: 'token_usage', id: this.deps.newId(),