diff --git a/packages/runtime/src/__tests__/mid-turn-capacity-compact.test.ts b/packages/runtime/src/__tests__/mid-turn-capacity-compact.test.ts index 78c10e17ac..50b1db6452 100644 --- a/packages/runtime/src/__tests__/mid-turn-capacity-compact.test.ts +++ b/packages/runtime/src/__tests__/mid-turn-capacity-compact.test.ts @@ -182,6 +182,26 @@ describe('plan mid-turn capacity compaction', () => { assert.equal(ids.at(-1), 'res-b'); }); + test('step-0 recovery folds prior history without covering the current user anchor', async () => { + const events = longTurnEvents(); + const result = await planMidTurnCapacityCompaction( + planInput({ phase: 'pre_turn', orderedEvents: events }), + ); + assert.equal(result.decision, 'compacted'); + if (result.decision !== 'compacted') return; + + assert.equal(result.checkpoint.phase, undefined); + assert.deepEqual( + result.coveredRuntimeEvents.map((event) => event.id), + ['prior-0', 'prior-1'], + ); + assert.deepEqual( + result.replacementEvents.slice(1).map((event) => event.id), + events.slice(2).map((event) => event.id), + ); + assert.deepEqual(result.replacementEvents[1], events[2]); + }); + test('persisted checkpoint replay-validates against the same ledger prefix (recovery)', async () => { const events = longTurnEvents(); const result = await planMidTurnCapacityCompaction(planInput({ orderedEvents: events })); diff --git a/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts b/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts index 5ee3e6a656..8827ba78a7 100644 --- a/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts +++ b/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts @@ -840,6 +840,36 @@ describe('reactive overflow recovery in the streaming backend', () => { assert.equal(lastCall?.totalTokens, 250); }); + test('keeps a step-0 overflow checkpoint projected after the retry returns a tool call', async () => { + // The first request overflows before any step completes. Recovery folds + // only prior history into a pre-turn checkpoint and retries with the + // current user anchor verbatim. When that retry returns a tool call, the + // next request must rebuild from raw durable events through the checkpoint + // boundary instead of resurrecting the cached raw prior replay. + const fixture = buildReactiveFixture({ script: ['overflow', 'tool', 'done'], bigPriors: true }); + await runTurn(fixture); + + assert.equal(fixture.model.doStreamCalls.length, 3); + assert.equal(complete(fixture)?.stopReason, 'end_turn'); + assert.equal( + fixture.events.some((event) => event.type === 'error'), + false, + ); + assert.equal(fixture.recorded.length, 1); + assert.equal(fixture.recorded[0]!.phase, undefined); + assert.equal(fixture.summarizerCalls(), 1); + assert.deepEqual(fixture.toolExecutions, ['one.md']); + + const retryPrompt = JSON.stringify(fixture.model.doStreamCalls[1]?.prompt); + const successorPrompt = JSON.stringify(fixture.model.doStreamCalls[2]?.prompt); + for (const prompt of [retryPrompt, successorPrompt]) { + assert.equal(prompt.includes('REACTIVE_SUMMARY_SENTINEL'), true); + assert.equal(prompt.includes(ANCHOR_TEXT), true); + assert.equal(prompt.includes('PRIOR_FACT'), false); + } + assert.equal(successorPrompt.includes(RAW_SPAN_ONE), true); + }); + test('does not retry an overflow after an after-step stop is requested', async () => { let signalSecondStream!: () => void; let releaseSecondStream!: () => void; diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 95fe39a9f7..bbb45787b0 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -232,6 +232,7 @@ import { } from './context-budget.js'; import { evaluateHistoryCompactCheckpointReplay, + isHistoryCompactContentEvent, replaceHistoryCompactReplayBlocks, } from './history-compact.js'; import { selectSynthesisCacheForReplay } from './synthesis-cache.js'; @@ -1672,7 +1673,33 @@ export class AiSdkBackend implements AgentBackend { }; const loadDurableTurnProjection = async (): Promise => { const turnEvents = await loadDurableTurnEvents(); - const replayPlan = buildRuntimeEventModelReplayPlan(turnEvents, { + const projectionCheckpoint = midTurnState?.projectionCheckpoint; + const rawProjectionEvents = projectionCheckpoint + ? [ + ...midTurnState.priorContentEvents, + ...turnEvents.filter(isHistoryCompactContentEvent), + ] + : turnEvents; + let replayEvents = rawProjectionEvents; + if (projectionCheckpoint) { + const checkpointMatch = matchHistoryCompactCheckpointPrefix( + projectionCheckpoint, + rawProjectionEvents, + ); + if (checkpointMatch.reason) { + throw new Error(`durable checkpoint projection mismatch: ${checkpointMatch.reason}`); + } + 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 + // checkpoint forward instead of resurrecting raw history. + } + const replayPlan = buildRuntimeEventModelReplayPlan(replayEvents, { toolActivityTurnIds: collectToolActivityTurnIds([ ...(input.runtimeContext ?? []), ...turnEvents, @@ -1706,7 +1733,9 @@ export class AiSdkBackend implements AgentBackend { scope.imageBudget, settledModelOutputs, ); - return [...priorReplay.messages, ...currentTurnMessages]; + return projectionCheckpoint + ? currentTurnMessages + : [...priorReplay.messages, ...currentTurnMessages]; }; const activeCompactionHeadAnchor = messages[messages.length - 1]?.role === 'user' @@ -2257,6 +2286,7 @@ export class AiSdkBackend implements AgentBackend { retryAlreadyUsed: overflowRetryUsed, midTurnState, turnId, + stepNumber: runtimeSteps, currentMessages: attemptMessages, providerTools, activeTools: activeToolsForRequest, diff --git a/packages/runtime/src/ai-sdk-compaction.ts b/packages/runtime/src/ai-sdk-compaction.ts index 17b3aa93d8..0d28fde01a 100644 --- a/packages/runtime/src/ai-sdk-compaction.ts +++ b/packages/runtime/src/ai-sdk-compaction.ts @@ -1496,6 +1496,7 @@ export class AiSdkCompaction { activeToolsForStep: readonly string[]; systemPromptChars: number; turnTailPrompt: string | undefined; + phase?: 'pre_turn' | 'mid_turn'; abortSignal?: AbortSignal; }): Promise { const { @@ -1591,6 +1592,7 @@ export class AiSdkCompaction { const plan = await planMidTurnCapacityCompaction({ sessionId: this.sessionId, + phase: input.phase ?? 'mid_turn', orderedEvents, headAnchor: { runtimeEventId: state.headAnchor.id, turnId }, estimatedNextRequestTokens: input.estimatedNextRequestTokens, @@ -1712,6 +1714,7 @@ export class AiSdkCompaction { }; } state.previousCheckpoint = plan.checkpoint; + state.projectionCheckpoint = plan.checkpoint; return { decision: 'compacted', checkpoint: plan.checkpoint, @@ -1738,6 +1741,7 @@ export class AiSdkCompaction { retryAlreadyUsed: boolean; midTurnState: MidTurnCapacityCompactState | undefined; turnId: string; + stepNumber: number; currentMessages: readonly ModelMessage[]; providerTools: readonly MakaTool[]; activeTools: readonly string[]; @@ -1768,8 +1772,10 @@ export class AiSdkCompaction { input.activeTools, input.systemPromptChars, ); + const phase = input.stepNumber === 0 ? 'pre_turn' : 'mid_turn'; const outcome = await this.computeMidTurnCompactionReplacement({ turnId: input.turnId, + phase, origin: input.origin, state, queue: input.queue, @@ -1800,7 +1806,7 @@ export class AiSdkCompaction { stage: 'activeStep', sourceKind: 'runtimeEvents', decision: 'failedOpen', - phase: 'mid_turn', + phase, boundaryKind: 'historyCompact', reason: 'overflow', ...(outcome.decision === 'fail' @@ -2174,6 +2180,8 @@ export class MidTurnCapacityCompactState { lastRequestInputTokens: number | undefined; /** Latest durable checkpoint (loaded or written) for roll-forward summaries. */ previousCheckpoint: HistoryCompactCheckpoint | undefined; + /** Checkpoint accepted during this send; pins every later durable projection. */ + projectionCheckpoint: HistoryCompactCheckpoint | undefined; /** Set when the turn must end with a context_budget_exhausted outcome. */ exhaustedDetail: ContextBudgetExhaustedDetail | undefined; /** @@ -2265,7 +2273,7 @@ type MidTurnCompactionOutcome = }; /** - * The `decision: 'replaced'` diagnostic patch for a durable mid_turn fold, + * The `decision: 'replaced'` diagnostic patch for a durable active-send fold, * shared by the proactive (`reason: 'context_limit'`) and reactive * (`reason: 'overflow'`) triggers so both report the fold identically. */ @@ -2295,7 +2303,7 @@ function buildMidTurnReplacedDiagnosticPatch(input: { stage: 'activeStep', sourceKind: 'runtimeEvents', decision: 'replaced', - phase: 'mid_turn', + phase: checkpoint.phase ?? 'pre_turn', boundaryKind: 'historyCompact', boundaryIds: [checkpoint.checkpointId], coverage: { bodySha256: [checkpoint.coverage.sourceDigest] }, diff --git a/packages/runtime/src/mid-turn-capacity-compact.ts b/packages/runtime/src/mid-turn-capacity-compact.ts index 0f6b50b17e..3e84b585b6 100644 --- a/packages/runtime/src/mid-turn-capacity-compact.ts +++ b/packages/runtime/src/mid-turn-capacity-compact.ts @@ -203,6 +203,8 @@ export type MidTurnSummarizer = (input: { export interface PlanMidTurnCapacityCompactionInput { sessionId: string; + /** Step-0 overflow folds only prior history; later steps also fold the head anchor. */ + phase?: 'pre_turn' | 'mid_turn'; /** * Full ordered content-event projection for the compaction pool: * `[...prior turns, head anchor, ...current-turn completed steps]`. @@ -233,7 +235,7 @@ export type PlanMidTurnCapacityCompactionResult = | { decision: 'compacted'; checkpoint: HistoryCompactCheckpoint; - /** Deterministic `[block, head anchor, tail]` replacement projection. */ + /** Deterministic checkpoint-block plus verbatim successor projection. */ replacementEvents: RuntimeEvent[]; coveredRuntimeEvents: RuntimeEvent[]; tailRuntimeEvents: RuntimeEvent[]; @@ -255,6 +257,7 @@ export type MidTurnFailReason = 'no_safe_completed_span' | 'summarizer_failed'; export async function planMidTurnCapacityCompaction( input: PlanMidTurnCapacityCompactionInput, ): Promise { + const phase = input.phase ?? 'mid_turn'; const charsPerToken = Math.max(1, input.charsPerToken ?? 4); const highWater = Math.max(1, input.contextWindow - Math.max(0, input.reserveTokens)); if (input.estimatedNextRequestTokens <= highWater) { @@ -268,24 +271,30 @@ export async function planMidTurnCapacityCompaction( const boundary = selectMidTurnSafeBoundary(input.orderedEvents, { reserveTailEvents: input.reserveTailEvents ?? 1, isPinned: (event) => - event.turnId === input.headAnchor.turnId && - event.content?.kind === 'text' && - event.content.steering === true, + (phase === 'pre_turn' && event.id === input.headAnchor.runtimeEventId) || + (event.turnId === input.headAnchor.turnId && + event.content?.kind === 'text' && + event.content.steering === true), }); const headAnchorIndex = input.orderedEvents.findIndex( (event) => event.id === input.headAnchor.runtimeEventId, ); - // Coverage must include the head anchor and at least one other event, since the - // anchor is re-rendered verbatim — folding only the anchor saves nothing. - if ( - !boundary.ok || - headAnchorIndex < 0 || - boundary.coveredCount <= headAnchorIndex || - boundary.coveredCount < 2 - ) { + // Mid-turn coverage includes the head anchor and at least one other event; + // the anchor is re-rendered verbatim, so folding only it saves nothing. + // Step-0 recovery is a pre-turn fold: the anchor is pinned in the successor + // tail and at least one prior event must be covered. + const hasSafeCoverage = + phase === 'mid_turn' + ? boundary.ok && boundary.coveredCount > headAnchorIndex && boundary.coveredCount >= 2 + : boundary.ok && boundary.coveredCount > 0 && boundary.coveredCount <= headAnchorIndex; + if (headAnchorIndex < 0 || !hasSafeCoverage) { return { decision: 'fail_open', reason: 'no_safe_completed_span' }; } + // Narrowing above proves the boundary is safe. + if (!boundary.ok) { + return { decision: 'fail_open', reason: 'no_safe_completed_span' }; + } const coveredRuntimeEvents = input.orderedEvents.slice(0, boundary.coveredCount); const tailRuntimeEvents = input.orderedEvents.slice(boundary.coveredCount); @@ -329,8 +338,7 @@ export async function planMidTurnCapacityCompaction( sessionId: input.sessionId, coveredRuntimeEvents, summary, - phase: 'mid_turn', - headAnchor: input.headAnchor, + ...(phase === 'mid_turn' ? { phase: 'mid_turn' as const, headAnchor: input.headAnchor } : {}), ...(input.highWaterName !== undefined ? { highWaterName: input.highWaterName } : {}), ...(input.highWaterSeq !== undefined ? { highWaterSeq: input.highWaterSeq } : {}), ...(previousCheckpoint ? { previousCheckpointId: previousCheckpoint.checkpointId } : {}),