/** * Surface retention selection and the shared log-recorded compaction * transaction for automatic open-turn and manual idle-session compaction. * * @module @deepseek-ai/dsh-compact-basic/region */ import { isDeepStrictEqual } from 'node:util' import { COMPACT_CHECKPOINT_SOURCE, ManualCompactionError, toolPairingBalancedAfter, toolPairingBalancedBefore, } from '@deepseek-ai/dsh-compact' import type { CompactionResult } from '@deepseek-ai/dsh-compact' import { createUserMessage, errorChain } from '@deepseek-ai/dsh-llm' import type { Message, UserMessage } from '@deepseek-ai/dsh-llm' import type { TokenMeasurement, TokenMeterService } from '@deepseek-ai/dsh-token-meter' import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' import type { Agent } from '@deepseek-ai/dsh-agent' import { frameSummary } from './summarizer.ts' import type { SummarizationInput, SummaryResult } from './summarizer.ts' interface RegionDependencies { readonly meter: TokenMeterService summarize(input: SummarizationInput, agent: Agent, signal?: AbortSignal): Promise } /** One validated inclusive span of current surface positions. */ interface SurfaceSelection { readonly start: number readonly end: number readonly startIdx: number readonly endIdx: number readonly shadowedSeqs: readonly number[] } /** A selection with its priced snapshot and the replay input built from it. */ interface PreparedCompaction extends SurfaceSelection { readonly measurement: TokenMeasurement readonly selectedNodes: TokenMeasurement['nodes'] readonly shadowedTokenCount: number readonly input: SummarizationInput } interface SummarizedCompaction extends PreparedCompaction, SummaryResult { readonly checkpointMessage: UserMessage } interface CompactionTransactionOptions { /** `current-turn` derives a numbered owner; `null` writes a standalone bracket. */ readonly owner: 'current-turn' | null /** Surface relationship that must survive asynchronous summarization. */ readonly stability: 'whole-surface' | 'selected-span' /** Optional durability checkpoint after a successfully closed bracket. */ readonly flush?: () => Promise } interface CompactionEntryState { readonly openTurn: number | null readonly unmatchedCompactionStart: SessionEvent<'compact/start'> | undefined readonly latestEndSeedSeq: number | undefined } /** * Rejects a summary whose replacement boundaries are no longer the ones it was * built from, distinguished from summarizer and shrink failures so a manual * caller can report the two causes differently. */ class SurfaceChangedError extends Error {} /** Whether the summary may still replace the span it was built from. */ type StabilityCheck = ( dependencies: RegionDependencies, session: Session, prepared: PreparedCompaction, ) => void /** Failure captured after `compact/start` has committed. */ interface TransactionFailure { readonly error: unknown readonly stage: 'summary' | 'commit' } /** * Resolve the next head-anchored range while retaining a priced recent tail * and never splitting an assistant tool-call/result pair. * @param session - session supplying authoritative current surface positions. * @param measurement - unified pressure and surface measurement from the conversation meter. * @param retainTokens - minimum recent tail budget retained verbatim. * @returns the inclusive positional seq range to compact, or `null`. */ export function selectCompactableRange( session: Session, measurement: TokenMeasurement, retainTokens: number, ): { start: number; end: number } | null { const pricedNodes = measurement.nodes if (pricedNodes.length === 0) return null const surfaceNodes = session.surface.nodes if (surfaceNodes.length !== pricedNodes.length || surfaceNodes.some((seq, index) => seq !== pricedNodes[index]?.seq)) { throw new Error('compaction: token-meter surface does not match the current session surface') } let accumulated = 0 let keepFromIdx = pricedNodes.length for (let index = pricedNodes.length - 1; index >= 0; index -= 1) { // oxlint-disable-next-line typescript/no-non-null-assertion accumulated += pricedNodes[index]!.tokens keepFromIdx = index if (accumulated >= retainTokens) break } if (keepFromIdx === 0) return null while (keepFromIdx > 0) { // oxlint-disable-next-line typescript/no-non-null-assertion if (toolPairingBalancedBefore(session, surfaceNodes[keepFromIdx]!)) break keepFromIdx -= 1 } if (keepFromIdx === 0) return null // oxlint-disable-next-line typescript/no-non-null-assertion const first = surfaceNodes[0]! // oxlint-disable-next-line typescript/no-non-null-assertion const cutoff = surfaceNodes[keepFromIdx - 1]! return { start: first, end: cutoff } } /** * Run the single compaction transaction over one selected positional span. * Selection and validation are read-only. Idle/log validation and * `compact/start` are synchronously adjacent, so the durable opening marker is * the compaction lock before summarization yields. Every later failure makes * exactly one `compact/end` attempt; a failed close deliberately leaves the * unmatched start detectable. * @param dependencies - conversation meter and dynamically dispatched summarizer hook. * @param session - session whose surface is mutated. * @param start - inclusive first surface-node seq. * @param end - inclusive last surface-node seq. * @param agent - agent used by the summarizer. * @param options - bracket owner, stability rule, and optional durability checkpoint. * @param signal - optional summarization cancellation signal. * @returns the successful durable compaction result. */ export async function compactSurfaceRegion( dependencies: RegionDependencies, session: Session, start: number, end: number, agent: Agent, options: CompactionTransactionOptions, signal?: AbortSignal, ): Promise { if (options.owner === null) signal?.throwIfAborted() const selection = validateSurfaceRegion(session, start, end) const entryState = inspectCompactionEntryState(session.events) assertCompactionInactive( entryState.unmatchedCompactionStart, entryState.latestEndSeedSeq, 'compaction', ) let owner: number | null if (options.owner === null) { if (entryState.openTurn !== null) { throw new ManualCompactionError('busy', 'manual compaction: the session already has an open turn') } owner = null } else { if (entryState.openTurn === null) { throw new Error('compactRegion: no open turn — automatic compaction events must be enclosed in a turn') } owner = entryState.openTurn } const startEvent = session.append('compact/start', { turn: owner }) const assertStable: StabilityCheck = options.stability === 'whole-surface' ? assertWholeSurfaceUnchanged : assertSelectedSpanStable let failure: TransactionFailure | undefined let flushFailure: unknown let result: CompactionResult | undefined let closed = false let closing = false let stage: TransactionFailure['stage'] = 'summary' try { const prepared = prepareCompaction(dependencies, session, selection) const summarized = await summarizeCompaction(dependencies, prepared, agent, signal) if (options.owner === null) signal?.throwIfAborted() assertStable(dependencies, session, summarized) stage = 'commit' const pending = commitCompactionBody(session, startEvent, summarized) closing = true const endEvent = session.append('compact/end', { turn: owner }) closed = true result = completeCompaction(pending, endEvent) } catch (error: unknown) { failure = { error, stage: closing ? 'commit' : stage } if (!closing) { closing = true try { session.append('compact/end', { turn: owner, error: errorChain(error) }) closed = true } catch (closeError: unknown) { failure = { error: closeError, stage: 'commit' } } } } if (closed && options.flush !== undefined) { try { await options.flush() } catch (error: unknown) { flushFailure = error } } if (options.owner === null) signal?.throwIfAborted() if (failure !== undefined) { if (options.owner === null) throwManualFailure(failure) throw failure.error } if (flushFailure !== undefined) { throw new ManualCompactionError( 'persistence', 'manual compaction durability checkpoint failed', { cause: flushFailure }, ) } /* v8 ignore next -- every path without a result records and throws a failure above. */ if (result === undefined) throw new Error('compaction committed without a result') return result } /** Classify one closed manual attempt without weakening cancellation precedence. */ function throwManualFailure(failure: TransactionFailure): never { if (failure.stage === 'commit') { throw new ManualCompactionError( 'commit', 'manual compaction did not commit cleanly', { cause: failure.error }, ) } if (failure.error instanceof SurfaceChangedError) { throw new ManualCompactionError( 'changed', 'the compacted history changed during manual compaction', { cause: failure.error }, ) } throw new ManualCompactionError( 'summary', 'manual compaction could not produce a smaller summary', { cause: failure.error }, ) } /** * Reject a durable unmatched compaction marker unless a later constructor-seed * boundary proves that its owner belongs to an earlier session lifecycle. * @param unmatchedCompactionStart - latest unmatched opening marker, if any. * @param latestEndSeedSeq - newest constructor-seed boundary, if any. * @param stage - operation label included in the busy diagnostic. */ function assertCompactionInactive( unmatchedCompactionStart: SessionEvent<'compact/start'> | undefined, latestEndSeedSeq: number | undefined, stage: string, ): void { if (unmatchedCompactionStart === undefined || (latestEndSeedSeq !== undefined && latestEndSeedSeq > unmatchedCompactionStart.seq)) return throw new ManualCompactionError( 'busy', `${stage}: compaction already in progress; the session compaction lock is already active`, ) } /** * Recheck the durable compaction lock after an asynchronous policy decision. * @param session - session whose latest marker state is inspected. * @param stage - operation label included in the busy diagnostic. */ export function assertNoActiveCompaction(session: Session, stage: string): void { const entryState = inspectCompactionEntryState(session.events) assertCompactionInactive( entryState.unmatchedCompactionStart, entryState.latestEndSeedSeq, stage, ) } /** Validate one requested surface-position span before asynchronous work begins. */ function validateSurfaceRegion(session: Session, start: number, end: number): SurfaceSelection { const nodes = session.surface.nodes const startIdx = nodes.indexOf(start) const endIdx = nodes.indexOf(end) if (startIdx === -1) throw new Error(`compactRegion: start seq ${start} not found in surface`) if (endIdx === -1) throw new Error(`compactRegion: end seq ${end} not found in surface`) if (startIdx > endIdx) { throw new Error( `compactRegion: start seq ${start} (position ${startIdx}) is after end seq ${end} (position ${endIdx}) on the surface`, ) } // oxlint-disable-next-line typescript/no-non-null-assertion if (!toolPairingBalancedBefore(session, nodes[startIdx]!)) { throw new Error(`compactRegion: start seq ${start} is not a balanced boundary (would split a step's tool-call/result pair)`) } // oxlint-disable-next-line typescript/no-non-null-assertion if (!toolPairingBalancedAfter(session, nodes[endIdx]!)) { throw new Error(`compactRegion: end seq ${end} is not a balanced boundary (would split a step, or the step is still open)`) } return { start, end, startIdx, endIdx, shadowedSeqs: nodes.slice(startIdx, endIdx + 1) } } /** Snapshot pricing and replay input for a validated surface range. */ function prepareCompaction( dependencies: RegionDependencies, session: Session, selection: SurfaceSelection, ): PreparedCompaction { const measurement = dependencies.meter.measure(session) const selectedNodes = measurement.nodes.slice(selection.startIdx, selection.endIdx + 1) if (selectedNodes.length !== selection.shadowedSeqs.length || selectedNodes.some((node, index) => node.seq !== selection.shadowedSeqs[index])) { throw new SurfaceChangedError('compaction: selected surface changed before summarization began') } return { ...selection, measurement, selectedNodes, shadowedTokenCount: selectedNodes.reduce((total, node) => total + node.tokens, 0), input: buildSummarizationInput(session, selection.shadowedSeqs), } } /** Run the summarizer and frame its replacement checkpoint. */ async function summarizeCompaction( dependencies: RegionDependencies, prepared: PreparedCompaction, agent: Agent, signal?: AbortSignal, ): Promise { const summaryResult = await dependencies.summarize(prepared.input, agent, signal) const checkpointMessage = createUserMessage({ content: frameSummary(summaryResult.summary), source: COMPACT_CHECKPOINT_SOURCE, }) const framedSummaryTokenCount = dependencies.meter.estimateMessage(checkpointMessage) if (framedSummaryTokenCount >= prepared.shadowedTokenCount) { throw new Error( `summary is not smaller than the shadowed content (${framedSummaryTokenCount} estimated framed tokens >= ${prepared.shadowedTokenCount})`, ) } return { ...prepared, ...summaryResult, checkpointMessage, } } /** Reject a summary prepared against any earlier surface generation. */ function assertWholeSurfaceUnchanged( dependencies: RegionDependencies, session: Session, prepared: PreparedCompaction, ): void { const current = dependencies.meter.measure(session) if (!isDeepStrictEqual(current.nodes, prepared.measurement.nodes)) { throw new SurfaceChangedError('compaction: session surface changed during summarization') } } /** * Require only that the selected span remain the same present, contiguous, * equally priced, balanced replacement target. Nodes added outside it remain * visible and do not invalidate the summary. */ function assertSelectedSpanStable( dependencies: RegionDependencies, session: Session, prepared: PreparedCompaction, ): void { let current: SurfaceSelection try { current = validateSurfaceRegion(session, prepared.start, prepared.end) } catch (error: unknown) { throw new SurfaceChangedError( 'compaction: the selected span is no longer a valid replacement target', { cause: error }, ) } if (!isDeepStrictEqual([...current.shadowedSeqs], [...prepared.shadowedSeqs])) { throw new SurfaceChangedError('compaction: the selected span changed during summarization') } const measured = dependencies.meter.measure(session).nodes.slice(current.startIdx, current.endIdx + 1) if (!isDeepStrictEqual(measured, prepared.selectedNodes)) { throw new SurfaceChangedError('compaction: the selected span was rewritten during summarization') } } /** Append one already-summarized provenance and replacement body without yielding. */ function commitCompactionBody( session: Session, startEvent: SessionEvent<'compact/start'>, summarized: SummarizedCompaction, ): Omit { const { start, end, shadowedSeqs, shadowedTokenCount, summary, rawOutput, provider, model, maxTokens, usage, checkpointMessage, } = summarized const summaryEvent = session.append('compact/summary', { summary, ...rawOutput === undefined ? {} : { rawOutput }, shadowedRange: { start, end }, shadowedSeqs: [...shadowedSeqs], shadowedTokenCount, provider, model, ...maxTokens === undefined ? {} : { maxTokens }, ...usage === undefined ? {} : { usage }, }) session.append('user/message', checkpointMessage, { surfaceOp: { op: 'replace', start, end }, sourceEventSeqs: [startEvent.seq, summaryEvent.seq, ...shadowedSeqs], }) return { startSeq: startEvent.seq, summarySeq: summaryEvent.seq, summary, shadowedRange: { start, end }, shadowedSeqs: [...shadowedSeqs], shadowedTokenCount, } } /** Attach the successfully appended close event to a pending result. */ function completeCompaction( pending: Omit, endEvent: SessionEvent<'compact/end'>, ): CompactionResult { return { ...pending, endSeq: endEvent.seq } } /** * Reconstruct the last routed request's cacheable prefix for the shadowed * region: its system prompt and tool schemas, then the region's own derived * messages in surface order. The summarizer appends only the compaction * instruction after this, so the call is a genuine prefix of the conversation * and reuses the provider's KV cache. * @param session - session supplying the request header and per-node projection. * @param shadowedSeqs - the surface-node seqs, in order, being compacted. * @returns the replayed conversation prefix to condense. */ function buildSummarizationInput( session: Session, shadowedSeqs: readonly number[], ): SummarizationInput { const header = session.requestHeader() const events = session.events const regionMessages = shadowedSeqs // shadowedSeqs are current surface seqs, so each is a valid log index. // oxlint-disable-next-line typescript/no-non-null-assertion .map(seq => session.deriveEventMessage(events[seq]!)) .filter((message): message is Message => message !== null) return { ...header?.system === undefined ? {} : { system: header.system }, ...header?.tools === undefined ? {} : { tools: header.tools }, messages: regionMessages, } } /** Inspect open-turn, unmatched-compaction, and latest seed-boundary state independently. */ function inspectCompactionEntryState(events: readonly SessionEvent[]): CompactionEntryState { let openTurn: number | null = null let openTurnStateKnown = false let unmatchedCompactionStart: SessionEvent<'compact/start'> | undefined let compactionEntryStateKnown = false let latestEndSeedSeq: number | undefined for (let index = events.length - 1; index >= 0; index -= 1) { // oxlint-disable-next-line typescript/no-non-null-assertion const event = events[index]! if (latestEndSeedSeq === undefined && event.type === 'session/end-seed') { latestEndSeedSeq = event.seq } if (!compactionEntryStateKnown) { if (event.type === 'compact/start') { unmatchedCompactionStart = event compactionEntryStateKnown = true } else if (event.type === 'compact/end') { compactionEntryStateKnown = true } } if (!openTurnStateKnown) { if (event.type === 'turn/start') { openTurn = event.data.turn openTurnStateKnown = true } else if (event.type === 'turn/end') { openTurnStateKnown = true } } if (openTurnStateKnown && compactionEntryStateKnown && latestEndSeedSeq !== undefined) break } return { openTurn, unmatchedCompactionStart, latestEndSeedSeq } }