| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| import { createHmac, randomBytes, timingSafeEqual } from 'node:crypto'; |
| import type { StoredMessage } from '@maka/core/session'; |
| import { |
| SESSION_TRANSCRIPT_PAGE_MAX_MESSAGES, |
| SESSION_TRANSCRIPT_RANGE_MAX_BYTES, |
| SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES, |
| type SessionTranscriptBootstrap, |
| type SessionTranscriptFragment, |
| type SessionTranscriptPage, |
| type SessionTranscriptPageDirection, |
| type SessionTranscriptPageInput, |
| type SessionTranscriptPageSource, |
| type TurnSnapshot, |
| } from '../protocol/index.js'; |
| import { |
| ACTIVE_TRANSCRIPT_OVERLAY_MAX_BYTES, |
| ACTIVE_TRANSCRIPT_OVERLAY_MAX_MESSAGES, |
| type SessionTranscriptReader, |
| } from './session-transcript-reader.js'; |
| import { projectSharedSessionTranscriptMessage } from './shared-session-transcript.js'; |
| |
| type SessionTranscriptProjection = 'owner' | 'shared'; |
| |
| interface TranscriptCursorState { |
| readonly version: 1; |
| readonly subscriptionId: string; |
| readonly sessionId: string; |
| readonly source: SessionTranscriptPageSource; |
| readonly direction: SessionTranscriptPageDirection; |
| readonly throughSequence: number | null; |
| readonly position: number; |
| readonly byteOffset: number | null; |
| readonly rangeBoundarySequence: number | null; |
| } |
| |
| export interface SubscriberTranscriptState { |
| readonly sessionId: string; |
| readonly subscriptionId: string; |
| readonly openedThroughSequence: number | null; |
| overlayMessages: readonly Buffer[] | undefined; |
| readonly cursorSecret: Buffer; |
| durableThroughSequence: number | null; |
| readonly projection: SessionTranscriptProjection; |
| } |
| |
| export interface ActiveTranscriptAssistantStream { |
| readonly turnId: string; |
| readonly messageId: string; |
| readonly kind: 'text' | 'thinking'; |
| readonly text: string; |
| } |
| |
| interface SelectedFragments { |
| readonly fragments: readonly SessionTranscriptFragment[]; |
| readonly rawBytes: number; |
| readonly next: { position: number; byteOffset: number | null } | null; |
| } |
| |
| export async function createSessionTranscriptBootstrap(input: { |
| reader: SessionTranscriptReader; |
| sessionId: string; |
| subscriptionId: string; |
| throughSequence: number | null; |
| rootTurn: TurnSnapshot | null; |
| activeAssistantStreams: Iterable<ActiveTranscriptAssistantStream>; |
| maxBytes: number; |
| maxEncodedBytes?: number; |
| preparedOverlayMessages?: readonly Buffer[]; |
| projection: SessionTranscriptProjection; |
| }): Promise<{ bootstrap: SessionTranscriptBootstrap; state: SubscriberTranscriptState }> { |
| const projection = input.projection; |
| const preparedOverlayMessages = |
| input.preparedOverlayMessages ?? (await prepareSessionTranscriptOverlay(input)); |
| const overlayMessages = |
| projection === 'shared' |
| ? preparedOverlayMessages.flatMap((message) => |
| projectEncodedSharedMessage(message, input.sessionId), |
| ) |
| : preparedOverlayMessages; |
| const cursorSecret = randomBytes(32); |
| let rawBudget = input.maxBytes; |
| for (;;) { |
| const overlayBudget = Math.min(8 * 1024, Math.max(1, Math.floor(rawBudget / 2))); |
| const selectedOverlay = selectOverlay( |
| overlayMessages, |
| 'older', |
| overlayMessages.length - 1, |
| null, |
| overlayBudget, |
| ); |
| const durableBudget = rawBudget - selectedOverlay.rawBytes; |
| const durableRequest = { |
| direction: 'older', |
| throughSequence: input.throughSequence, |
| maxBytes: durableBudget, |
| maxMessages: SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES, |
| } as const; |
| const durableStorage = |
| projection === 'shared' |
| ? await readSharedDurablePage(input.reader, input.sessionId, durableRequest) |
| : await input.reader.readDurablePage(input.sessionId, durableRequest); |
| if (durableStorage.throughSequence !== input.throughSequence) { |
| throw new Error('Session transcript durable watermark changed during bootstrap'); |
| } |
| const state: SubscriberTranscriptState = { |
| sessionId: input.sessionId, |
| subscriptionId: input.subscriptionId, |
| openedThroughSequence: input.throughSequence, |
| durableThroughSequence: input.throughSequence, |
| overlayMessages, |
| cursorSecret, |
| projection, |
| }; |
| const durableSelection = storageSelection(durableStorage); |
| const rangeEdges = await readRangeEdges({ |
| reader: input.reader, |
| state, |
| direction: 'older', |
| throughSequence: input.throughSequence, |
| selected: durableSelection, |
| }); |
| const bootstrap: SessionTranscriptBootstrap = { |
| throughSequence: input.throughSequence, |
| overlayMessageCount: overlayMessages.length, |
| durable: pageFromSelection( |
| state, |
| 'durable', |
| 'older', |
| rangeEdges.selected, |
| input.throughSequence, |
| rangeEdges.rangeBoundarySequence, |
| rangeEdges.protectedTurnSequence, |
| ), |
| overlay: pageFromSelection(state, 'overlay', 'older', selectedOverlay), |
| }; |
| const encodedBytes = Buffer.byteLength(JSON.stringify(bootstrap), 'utf8'); |
| if (input.maxEncodedBytes === undefined || encodedBytes <= input.maxEncodedBytes) { |
| return { state, bootstrap }; |
| } |
| if (rawBudget <= 2) { |
| throw new Error('Session transcript bootstrap cannot fit the subscription open result'); |
| } |
| const excess = encodedBytes - input.maxEncodedBytes; |
| rawBudget = Math.max(2, rawBudget - Math.max(1, Math.ceil((excess * 3) / 4))); |
| } |
| } |
| |
| export async function prepareSessionTranscriptOverlay(input: { |
| reader: SessionTranscriptReader; |
| sessionId: string; |
| throughSequence: number | null; |
| rootTurn: TurnSnapshot | null; |
| activeAssistantStreams: Iterable<ActiveTranscriptAssistantStream>; |
| }): Promise<readonly Buffer[]> { |
| const activeAssistantStreams = [...input.activeAssistantStreams]; |
| const activeMessageIds = [...new Set(activeAssistantStreams.map((stream) => stream.messageId))]; |
| const activeOverlay = await input.reader.readActiveOverlay(input.sessionId, input.rootTurn); |
| const durableActiveMessages = await input.reader.readDurableMessagesById(input.sessionId, { |
| messageIds: activeMessageIds, |
| throughSequence: input.throughSequence, |
| maxBytes: ACTIVE_TRANSCRIPT_OVERLAY_MAX_BYTES, |
| maxMessages: ACTIVE_TRANSCRIPT_OVERLAY_MAX_MESSAGES, |
| }); |
| const overlayMessages = mergeActiveAssistantStreams( |
| activeOverlay, |
| activeAssistantStreams, |
| durableActiveMessages, |
| ).map((message) => Buffer.from(JSON.stringify(message), 'utf8')); |
| assertOverlayRetainedBound(overlayMessages); |
| return overlayMessages; |
| } |
| |
| export async function readSessionTranscriptPage(input: { |
| reader: SessionTranscriptReader; |
| state: SubscriberTranscriptState; |
| request: SessionTranscriptPageInput; |
| }): Promise<SessionTranscriptPage> { |
| const { state, request } = input; |
| if ( |
| request.throughSequence !== null && |
| (state.durableThroughSequence === null || |
| request.throughSequence > state.durableThroughSequence) |
| ) { |
| throw new TranscriptPageRequestError('Transcript watermark is not known to this subscription'); |
| } |
| if (request.source === 'overlay' && request.throughSequence !== state.openedThroughSequence) { |
| throw new TranscriptPageRequestError('Transcript overlay watermark changed'); |
| } |
| if (request.source === 'overlay' && state.overlayMessages === undefined) { |
| throw new TranscriptPageRequestError('Transcript overlay has been released'); |
| } |
| const position = resolvePosition(state, request); |
| if (position === null) return emptyPage(state, request); |
| if (request.source === 'overlay') { |
| const overlayMessages = state.overlayMessages!; |
| const selected = selectOverlay( |
| overlayMessages, |
| request.direction, |
| position.position, |
| position.byteOffset, |
| request.maxBytes, |
| continuationMessageLimit(position), |
| ); |
| return pageFromSelection( |
| state, |
| 'overlay', |
| request.direction, |
| selected, |
| request.throughSequence, |
| ); |
| } |
| if (request.throughSequence === null) return emptyPage(state, request); |
| const durableRequest = { |
| direction: request.direction, |
| throughSequence: request.throughSequence, |
| position: position.position, |
| ...(position.byteOffset === null ? {} : { byteOffset: position.byteOffset }), |
| maxBytes: request.maxBytes, |
| maxMessages: continuationMessageLimit(position), |
| } as const; |
| const storage = |
| state.projection === 'shared' |
| ? await readSharedDurablePage( |
| input.reader, |
| state.sessionId, |
| durableRequest, |
| position.rangeBoundarySequence, |
| ) |
| : await input.reader.readDurablePage(state.sessionId, durableRequest); |
| const selected = selectionThroughRangeBoundary( |
| storageSelection(storage), |
| request.direction, |
| position.rangeBoundarySequence, |
| ); |
| const rangeEdges = await readRangeEdges({ |
| reader: input.reader, |
| state, |
| direction: request.direction, |
| throughSequence: request.throughSequence, |
| selected, |
| }); |
| return pageFromSelection( |
| state, |
| 'durable', |
| request.direction, |
| rangeEdges.selected, |
| request.throughSequence, |
| rangeEdges.rangeBoundarySequence, |
| rangeEdges.protectedTurnSequence, |
| ); |
| } |
| |
| async function readRangeEdges(input: { |
| reader: SessionTranscriptReader; |
| state: SubscriberTranscriptState; |
| direction: SessionTranscriptPageDirection; |
| throughSequence: number | null; |
| selected: SelectedFragments; |
| }): Promise<{ |
| readonly selected: SelectedFragments; |
| readonly rangeBoundarySequence: number | null; |
| readonly protectedTurnSequence: number | null; |
| }> { |
| if (input.throughSequence === null) { |
| return { |
| selected: input.selected, |
| rangeBoundarySequence: null, |
| protectedTurnSequence: null, |
| }; |
| } |
| const selectedSequences = input.selected.fragments.flatMap((fragment) => |
| fragment.kind === 'durable' ? [fragment.sequence] : [], |
| ); |
| if (selectedSequences.length === 0) { |
| return { |
| selected: input.selected, |
| rangeBoundarySequence: null, |
| protectedTurnSequence: null, |
| }; |
| } |
| const boundaryCandidate = |
| input.direction === 'older' ? Math.min(...selectedSequences) : Math.max(...selectedSequences); |
| const scanPosition = |
| input.direction === 'older' ? Math.max(...selectedSequences) : Math.min(...selectedSequences); |
| const rangeRecords: Array<{ |
| readonly sequence: number; |
| readonly turnId: string | undefined; |
| readonly bytes: number; |
| }> = []; |
| let targetTurnId: string | undefined; |
| let targetStart: number | null = null; |
| let candidateReached = false; |
| let hiddenBytes = 0; |
| let reachedFarEdge = false; |
| let position: number | null = scanPosition; |
| while (position !== null && !reachedFarEdge) { |
| const scanned = await input.reader.readDurableRecords(input.state.sessionId, { |
| direction: input.direction, |
| throughSequence: input.throughSequence, |
| position, |
| maxStoredBytes: SESSION_TRANSCRIPT_RANGE_MAX_BYTES, |
| maxMessages: SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES, |
| }); |
| for (const record of scanned.records) { |
| const message = |
| input.state.projection === 'shared' |
| ? projectSharedSessionTranscriptMessage(record.message, input.state.sessionId) |
| : record.message; |
| if (!message) { |
| hiddenBytes += Buffer.byteLength(JSON.stringify(record.message), 'utf8'); |
| if (hiddenBytes > SESSION_TRANSCRIPT_RANGE_MAX_BYTES) { |
| throw new RangeError('Session transcript projection scan exceeds its capacity limit'); |
| } |
| continue; |
| } |
| const turnId = messageTurnId(message); |
| if (targetStart !== null && turnId !== targetTurnId) { |
| reachedFarEdge = true; |
| break; |
| } |
| rangeRecords.push({ |
| sequence: record.sequence, |
| turnId, |
| bytes: Buffer.byteLength(JSON.stringify(message), 'utf8'), |
| }); |
| if (!candidateReached && record.sequence === boundaryCandidate) { |
| candidateReached = true; |
| if (turnId === undefined) { |
| reachedFarEdge = true; |
| break; |
| } |
| targetTurnId = turnId; |
| targetStart = rangeRecords.length - 1; |
| while (targetStart > 0 && rangeRecords[targetStart - 1]?.turnId === targetTurnId) { |
| targetStart -= 1; |
| } |
| } |
| if (targetStart !== null) { |
| const targetRecords = rangeRecords.slice(targetStart); |
| const targetBytes = targetRecords.reduce((sum, target) => sum + target.bytes, 0); |
| if ( |
| targetRecords.length > SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES || |
| targetBytes > SESSION_TRANSCRIPT_RANGE_MAX_BYTES |
| ) { |
| if (targetStart === 0) { |
| return { |
| selected: input.selected, |
| rangeBoundarySequence: null, |
| protectedTurnSequence: null, |
| }; |
| } |
| reachedFarEdge = true; |
| break; |
| } |
| } |
| } |
| if (reachedFarEdge || scanned.nextPosition === null) { |
| position = scanned.nextPosition; |
| break; |
| } |
| if (scanned.nextPosition === position) { |
| throw new Error('Session transcript projection scan did not advance'); |
| } |
| position = scanned.nextPosition; |
| } |
| if (!candidateReached || rangeRecords.length === 0) { |
| throw new Error('Session transcript range did not reach its authoritative Turn'); |
| } |
| let retainedEnd = 0; |
| let retainedMessages = 0; |
| let retainedBytes = 0; |
| while (retainedEnd < rangeRecords.length) { |
| const groupStart = retainedEnd; |
| const groupTurnId = rangeRecords[groupStart]!.turnId; |
| let groupEnd = groupStart + 1; |
| if (groupTurnId !== undefined) { |
| while (groupEnd < rangeRecords.length && rangeRecords[groupEnd]?.turnId === groupTurnId) { |
| groupEnd += 1; |
| } |
| } |
| const group = rangeRecords.slice(groupStart, groupEnd); |
| const groupBytes = group.reduce((sum, record) => sum + record.bytes, 0); |
| if ( |
| group.length > SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES || |
| groupBytes > SESSION_TRANSCRIPT_RANGE_MAX_BYTES |
| ) { |
| if (groupStart > 0) break; |
| return { |
| selected: input.selected, |
| rangeBoundarySequence: null, |
| protectedTurnSequence: null, |
| }; |
| } |
| if ( |
| retainedMessages + group.length > SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES || |
| retainedBytes + groupBytes > SESSION_TRANSCRIPT_RANGE_MAX_BYTES |
| ) { |
| break; |
| } |
| retainedMessages += group.length; |
| retainedBytes += groupBytes; |
| retainedEnd = groupEnd; |
| } |
| const retainedRecords = rangeRecords.slice(0, retainedEnd); |
| const boundary = retainedRecords.at(-1)?.sequence; |
| if (boundary === undefined) { |
| throw new RangeError('Session transcript Turn range exceeds its capacity limit'); |
| } |
| const selected = |
| retainedEnd === rangeRecords.length |
| ? input.selected |
| : (() => { |
| const retainedSequences = new Set(retainedRecords.map((record) => record.sequence)); |
| const fragments = input.selected.fragments.filter( |
| (fragment) => fragment.kind === 'durable' && retainedSequences.has(fragment.sequence), |
| ); |
| return { |
| fragments, |
| rawBytes: fragments.reduce( |
| (sum, fragment) => sum + Buffer.byteLength(fragment.data, 'base64'), |
| 0, |
| ), |
| next: { position: rangeRecords[retainedEnd]!.sequence, byteOffset: null }, |
| }; |
| })(); |
| const turnRecords = retainedRecords.filter((record) => record.turnId !== undefined); |
| const protectedTurnSequence = |
| input.direction === 'older' ? turnRecords[0]?.sequence : turnRecords.at(-1)?.sequence; |
| return { |
| selected, |
| rangeBoundarySequence: boundary, |
| protectedTurnSequence: protectedTurnSequence ?? boundary, |
| }; |
| } |
| |
| function messageTurnId(message: StoredMessage): string | undefined { |
| const turnId = 'turnId' in message ? message.turnId : undefined; |
| return typeof turnId === 'string' ? turnId : undefined; |
| } |
| |
| async function readSharedDurablePage( |
| reader: SessionTranscriptReader, |
| sessionId: string, |
| request: Parameters<SessionTranscriptReader['readDurablePage']>[1], |
| rangeBoundarySequence: number | null = null, |
| ): ReturnType<SessionTranscriptReader['readDurablePage']> { |
| const position = |
| request.position ?? |
| (request.direction === 'older' ? (request.throughSequence ?? undefined) : 0); |
| const fragments: Awaited< |
| ReturnType<SessionTranscriptReader['readDurablePage']> |
| >['fragments'][number][] = []; |
| let rawBytes = 0; |
| let hiddenBytes = 0; |
| let next: { position: number; byteOffset: number | null } | null = null; |
| let scanPosition = position; |
| let throughSequence = request.throughSequence ?? null; |
| while (scanPosition !== undefined && next === null) { |
| const scanned = await reader.readDurableRecords(sessionId, { |
| direction: request.direction, |
| ...(request.throughSequence === undefined |
| ? {} |
| : { throughSequence: request.throughSequence }), |
| position: scanPosition, |
| maxStoredBytes: ACTIVE_TRANSCRIPT_OVERLAY_MAX_BYTES, |
| maxMessages: SESSION_TRANSCRIPT_PAGE_MAX_MESSAGES, |
| }); |
| throughSequence = scanned.throughSequence; |
| let recordIndex = 0; |
| for (; recordIndex < scanned.records.length; recordIndex += 1) { |
| const record = scanned.records[recordIndex]!; |
| const projected = projectSharedSessionTranscriptMessage(record.message, sessionId); |
| if (!projected) { |
| hiddenBytes += Buffer.byteLength(JSON.stringify(record.message), 'utf8'); |
| if (hiddenBytes > SESSION_TRANSCRIPT_RANGE_MAX_BYTES) { |
| throw new RangeError('Session transcript projection scan exceeds its capacity limit'); |
| } |
| continue; |
| } |
| const bytes = Buffer.from(JSON.stringify(projected), 'utf8'); |
| const continuationOffset = |
| record.sequence === position && request.byteOffset !== undefined |
| ? request.byteOffset |
| : null; |
| const selected = selectBuffer( |
| bytes, |
| request.direction, |
| continuationOffset, |
| request.maxBytes - rawBytes, |
| ); |
| if (!selected) { |
| next = { position: record.sequence, byteOffset: null }; |
| break; |
| } |
| fragments.push({ |
| sequence: record.sequence, |
| byteOffset: selected.byteOffset, |
| totalBytes: bytes.byteLength, |
| payloadDigest: null, |
| data: selected.data, |
| }); |
| rawBytes += selected.data.byteLength; |
| if (!selected.complete) { |
| next = { position: record.sequence, byteOffset: selected.nextOffset }; |
| break; |
| } |
| if (record.sequence === rangeBoundarySequence) { |
| const following = scanned.records[recordIndex + 1]?.sequence ?? scanned.nextPosition; |
| next = following === null ? null : { position: following, byteOffset: null }; |
| break; |
| } |
| if (fragments.length === request.maxMessages || rawBytes === request.maxBytes) { |
| const following = scanned.records[recordIndex + 1]?.sequence ?? scanned.nextPosition; |
| next = following === null ? null : { position: following, byteOffset: null }; |
| break; |
| } |
| } |
| if (next !== null) break; |
| if (scanned.nextPosition === null) break; |
| if (scanned.nextPosition === scanPosition) { |
| throw new Error('Session transcript projection scan did not advance'); |
| } |
| scanPosition = scanned.nextPosition; |
| } |
| return { |
| throughSequence, |
| fragments, |
| rawBytes, |
| next, |
| }; |
| } |
| |
| function projectEncodedSharedMessage(bytes: Buffer, sessionId: string): Buffer[] { |
| const projected = projectSharedSessionTranscriptMessage( |
| JSON.parse(bytes.toString('utf8')), |
| sessionId, |
| ); |
| return projected ? [Buffer.from(JSON.stringify(projected), 'utf8')] : []; |
| } |
| |
| export function updateSubscriberTranscriptHighWater( |
| state: SubscriberTranscriptState, |
| throughSequence: number | null, |
| ): boolean { |
| if (throughSequence === null || throughSequence === state.durableThroughSequence) return false; |
| if (state.durableThroughSequence !== null && throughSequence < state.durableThroughSequence) { |
| throw new Error('Session transcript durable watermark moved backwards'); |
| } |
| state.durableThroughSequence = throughSequence; |
| return true; |
| } |
| |
| export class TranscriptPageRequestError extends Error { |
| readonly name = 'TranscriptPageRequestError'; |
| } |
| |
| function resolvePosition( |
| state: SubscriberTranscriptState, |
| request: SessionTranscriptPageInput, |
| ): { |
| position: number; |
| byteOffset: number | null; |
| rangeBoundarySequence: number | null; |
| } | null { |
| if (request.cursor !== null) { |
| const cursor = decodeCursor(request.cursor, state.cursorSecret); |
| if ( |
| cursor.subscriptionId !== state.subscriptionId || |
| cursor.sessionId !== state.sessionId || |
| cursor.source !== request.source || |
| cursor.direction !== request.direction || |
| cursor.throughSequence !== request.throughSequence |
| ) { |
| throw new TranscriptPageRequestError('Transcript cursor does not match request'); |
| } |
| return { |
| position: cursor.position, |
| byteOffset: cursor.byteOffset, |
| rangeBoundarySequence: cursor.rangeBoundarySequence, |
| }; |
| } |
| if (request.source === 'overlay') { |
| const overlayMessages = state.overlayMessages; |
| if (overlayMessages === undefined) return null; |
| const anchor = request.anchorSequence; |
| const position = |
| request.direction === 'older' ? (anchor ?? overlayMessages.length) - 1 : (anchor ?? -1) + 1; |
| return position < 0 || position >= overlayMessages.length |
| ? null |
| : { position, byteOffset: null, rangeBoundarySequence: null }; |
| } |
| if (request.throughSequence === null) return null; |
| const position = |
| request.direction === 'older' |
| ? (request.anchorSequence ?? request.throughSequence + 1) - 1 |
| : (request.anchorSequence ?? -1) + 1; |
| return position < 0 || position > request.throughSequence |
| ? null |
| : { position, byteOffset: null, rangeBoundarySequence: null }; |
| } |
| |
| function continuationMessageLimit(position: { |
| position: number; |
| rangeBoundarySequence: number | null; |
| }): number { |
| return position.rangeBoundarySequence === null |
| ? SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES |
| : Math.min( |
| SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES, |
| Math.abs(position.position - position.rangeBoundarySequence) + 1, |
| ); |
| } |
| |
| function storageSelection( |
| storage: Awaited<ReturnType<SessionTranscriptReader['readDurablePage']>>, |
| ): SelectedFragments { |
| return { |
| fragments: storage.fragments.map((fragment) => ({ |
| kind: 'durable' as const, |
| sequence: fragment.sequence, |
| byteOffset: fragment.byteOffset, |
| totalBytes: fragment.totalBytes, |
| payloadDigest: fragment.payloadDigest, |
| data: fragment.data.toString('base64'), |
| })), |
| rawBytes: storage.rawBytes, |
| next: storage.next, |
| }; |
| } |
| |
| function selectionThroughRangeBoundary( |
| selected: SelectedFragments, |
| direction: SessionTranscriptPageDirection, |
| rangeBoundarySequence: number | null, |
| ): SelectedFragments { |
| if (rangeBoundarySequence === null) return selected; |
| // RuntimeEvent-backed message sequences are sparse, so a continuation's |
| // message limit cannot infer how many records remain from sequence distance. |
| const firstOmittedIndex = selected.fragments.findIndex( |
| (fragment) => |
| fragment.kind === 'durable' && |
| (direction === 'older' |
| ? fragment.sequence < rangeBoundarySequence |
| : fragment.sequence > rangeBoundarySequence), |
| ); |
| if (firstOmittedIndex === -1) return selected; |
| const firstOmitted = selected.fragments[firstOmittedIndex]!; |
| if (firstOmitted.kind !== 'durable') { |
| throw new Error('Session transcript durable range contained an overlay fragment'); |
| } |
| const fragments = selected.fragments.slice(0, firstOmittedIndex); |
| return { |
| fragments, |
| rawBytes: fragments.reduce( |
| (sum, fragment) => sum + Buffer.from(fragment.data, 'base64').byteLength, |
| 0, |
| ), |
| next: { position: firstOmitted.sequence, byteOffset: null }, |
| }; |
| } |
| |
| function selectOverlay( |
| messages: readonly Buffer[], |
| direction: SessionTranscriptPageDirection, |
| position: number, |
| byteOffset: number | null, |
| maxBytes: number, |
| maxMessages = SESSION_TRANSCRIPT_PAGE_MAX_MESSAGES, |
| ): SelectedFragments { |
| const fragments: SessionTranscriptFragment[] = []; |
| let rawBytes = 0; |
| let index = position; |
| let offset = byteOffset; |
| while ( |
| index >= 0 && |
| index < messages.length && |
| rawBytes < maxBytes && |
| fragments.length < maxMessages |
| ) { |
| const message = messages[index]!; |
| const selected = selectBuffer(message, direction, offset, maxBytes - rawBytes); |
| if (!selected) break; |
| fragments.push({ |
| kind: 'overlay', |
| messageIndex: index, |
| byteOffset: selected.byteOffset, |
| totalBytes: message.byteLength, |
| data: selected.data.toString('base64'), |
| }); |
| rawBytes += selected.data.byteLength; |
| if (!selected.complete) { |
| return { |
| fragments, |
| rawBytes, |
| next: { position: index, byteOffset: selected.nextOffset }, |
| }; |
| } |
| index += direction === 'older' ? -1 : 1; |
| offset = null; |
| } |
| return { |
| fragments, |
| rawBytes, |
| next: index >= 0 && index < messages.length ? { position: index, byteOffset: null } : null, |
| }; |
| } |
| |
| function selectBuffer( |
| bytes: Buffer, |
| direction: SessionTranscriptPageDirection, |
| byteOffset: number | null, |
| budget: number, |
| ): { |
| byteOffset: number; |
| data: Buffer; |
| complete: boolean; |
| nextOffset: number; |
| } | null { |
| if (budget < 1) return null; |
| if (direction === 'older') { |
| const end = byteOffset ?? bytes.byteLength; |
| if (end < 1 || end > bytes.byteLength) |
| throw new TranscriptPageRequestError('Invalid cursor byte offset'); |
| const start = Math.max(0, end - budget); |
| return { |
| byteOffset: start, |
| data: bytes.subarray(start, end), |
| complete: start === 0, |
| nextOffset: start, |
| }; |
| } |
| const start = byteOffset ?? 0; |
| if (start < 0 || start >= bytes.byteLength) { |
| throw new TranscriptPageRequestError('Invalid cursor byte offset'); |
| } |
| const end = Math.min(bytes.byteLength, start + budget); |
| return { |
| byteOffset: start, |
| data: bytes.subarray(start, end), |
| complete: end === bytes.byteLength, |
| nextOffset: end, |
| }; |
| } |
| |
| function pageFromSelection( |
| state: SubscriberTranscriptState, |
| source: SessionTranscriptPageSource, |
| direction: SessionTranscriptPageDirection, |
| selected: SelectedFragments, |
| throughSequence: number | null = state.openedThroughSequence, |
| rangeBoundarySequence: number | null = null, |
| protectedTurnSequence: number | null = null, |
| ): SessionTranscriptPage { |
| const cursorRangeBoundarySequence = |
| selected.next !== null && |
| rangeBoundarySequence !== null && |
| (direction === 'older' |
| ? selected.next.position < rangeBoundarySequence |
| : selected.next.position > rangeBoundarySequence) |
| ? null |
| : rangeBoundarySequence; |
| return { |
| kind: 'page', |
| sessionId: state.sessionId, |
| source, |
| direction, |
| throughSequence, |
| rawBytes: selected.rawBytes, |
| fragments: selected.fragments, |
| rangeBoundarySequence, |
| protectedTurnSequence, |
| nextCursor: selected.next |
| ? encodeCursor( |
| { |
| version: 1, |
| subscriptionId: state.subscriptionId, |
| sessionId: state.sessionId, |
| source, |
| direction, |
| throughSequence, |
| rangeBoundarySequence: cursorRangeBoundarySequence, |
| ...selected.next, |
| }, |
| state.cursorSecret, |
| ) |
| : null, |
| }; |
| } |
| |
| function emptyPage( |
| state: SubscriberTranscriptState, |
| request: SessionTranscriptPageInput, |
| ): SessionTranscriptPage { |
| return { |
| kind: 'page', |
| sessionId: state.sessionId, |
| source: request.source, |
| direction: request.direction, |
| throughSequence: request.throughSequence, |
| rawBytes: 0, |
| fragments: [], |
| rangeBoundarySequence: null, |
| protectedTurnSequence: null, |
| nextCursor: null, |
| }; |
| } |
| |
| function encodeCursor(cursor: TranscriptCursorState, secret: Buffer): string { |
| const payload = Buffer.from(JSON.stringify(cursor), 'utf8').toString('base64url'); |
| return `${payload}.${signCursor(payload, secret).toString('base64url')}`; |
| } |
| |
| function decodeCursor(value: string, secret: Buffer): TranscriptCursorState { |
| let decoded: unknown; |
| try { |
| const parts = value.split('.'); |
| if (parts.length !== 2) throw new Error('invalid cursor envelope'); |
| const [payload, signatureValue] = parts as [string, string]; |
| const bytes = Buffer.from(payload, 'base64url'); |
| const signature = Buffer.from(signatureValue, 'base64url'); |
| const expected = signCursor(payload, secret); |
| if ( |
| bytes.toString('base64url') !== payload || |
| signature.toString('base64url') !== signatureValue || |
| signature.byteLength !== expected.byteLength || |
| !timingSafeEqual(signature, expected) |
| ) { |
| throw new Error('invalid cursor signature'); |
| } |
| decoded = JSON.parse(bytes.toString('utf8')) as unknown; |
| } catch (cause) { |
| throw new TranscriptPageRequestError('Invalid transcript cursor', { cause }); |
| } |
| if (!decoded || typeof decoded !== 'object' || Array.isArray(decoded)) { |
| throw new TranscriptPageRequestError('Invalid transcript cursor'); |
| } |
| const cursor = decoded as Record<string, unknown>; |
| const keys = [ |
| 'version', |
| 'subscriptionId', |
| 'sessionId', |
| 'source', |
| 'direction', |
| 'throughSequence', |
| 'position', |
| 'byteOffset', |
| 'rangeBoundarySequence', |
| ]; |
| if ( |
| Object.keys(cursor).length !== keys.length || |
| keys.some((key) => !Object.hasOwn(cursor, key)) |
| ) { |
| throw new TranscriptPageRequestError('Invalid transcript cursor fields'); |
| } |
| if ( |
| cursor.version !== 1 || |
| typeof cursor.subscriptionId !== 'string' || |
| typeof cursor.sessionId !== 'string' || |
| (cursor.source !== 'durable' && cursor.source !== 'overlay') || |
| (cursor.direction !== 'older' && cursor.direction !== 'newer') || |
| (cursor.throughSequence !== null && !isCount(cursor.throughSequence)) || |
| !isCount(cursor.position) || |
| (cursor.byteOffset !== null && !isCount(cursor.byteOffset)) || |
| (cursor.rangeBoundarySequence !== null && !isCount(cursor.rangeBoundarySequence)) |
| ) { |
| throw new TranscriptPageRequestError('Invalid transcript cursor values'); |
| } |
| return cursor as unknown as TranscriptCursorState; |
| } |
| |
| function signCursor(payload: string, secret: Buffer): Buffer { |
| return createHmac('sha256', secret).update(payload, 'utf8').digest(); |
| } |
| |
| function mergeActiveAssistantStreams( |
| overlay: readonly StoredMessage[], |
| prefixes: Iterable<ActiveTranscriptAssistantStream>, |
| durable: readonly StoredMessage[], |
| ): StoredMessage[] { |
| const merged = [...overlay]; |
| const indices = new Map(merged.map((message, index) => [message.id, index])); |
| const durableById = new Map<string, StoredMessage>(); |
| for (const message of durable) durableById.set(message.id, message); |
| for (const prefix of prefixes) { |
| let index = indices.get(prefix.messageId); |
| const durableMessage = durableById.get(prefix.messageId); |
| if (index === undefined) { |
| if (!durableMessage) { |
| throw new Error('Active assistant prefix has no matching transcript message'); |
| } |
| index = merged.length; |
| indices.set(prefix.messageId, index); |
| merged.push(durableMessage); |
| } else if (durableMessage) { |
| const projected = merged[index]; |
| if (projected?.type !== 'assistant' || durableMessage.type !== 'assistant') { |
| throw new Error('Active assistant prefix has no matching transcript message'); |
| } |
| merged[index] = reconcileAssistantMessage(durableMessage, projected); |
| } |
| const message = merged[index]; |
| if (message?.type !== 'assistant' || message.turnId !== prefix.turnId) { |
| throw new Error('Active assistant prefix has no matching transcript message'); |
| } |
| if (prefix.kind === 'text') { |
| merged[index] = { ...message, text: reconcileAssistantText(message.text, prefix.text) }; |
| continue; |
| } |
| if (!message.thinking) { |
| throw new Error('Active thinking prefix has no matching transcript content'); |
| } |
| merged[index] = { |
| ...message, |
| thinking: { |
| ...message.thinking, |
| text: reconcileAssistantText(message.thinking.text, prefix.text), |
| }, |
| }; |
| } |
| return merged; |
| } |
| |
| function assertOverlayRetainedBound(messages: readonly Buffer[]): void { |
| if (messages.length > ACTIVE_TRANSCRIPT_OVERLAY_MAX_MESSAGES) { |
| throw new Error('Active Session transcript overlay exceeds its message limit'); |
| } |
| let retainedBytes = 0; |
| for (const message of messages) { |
| retainedBytes += message.byteLength; |
| if (retainedBytes > ACTIVE_TRANSCRIPT_OVERLAY_MAX_BYTES) { |
| throw new Error('Active Session transcript overlay exceeds its byte limit'); |
| } |
| } |
| } |
| |
| function reconcileAssistantMessage( |
| durable: Extract<StoredMessage, { type: 'assistant' }>, |
| projected: Extract<StoredMessage, { type: 'assistant' }>, |
| ): Extract<StoredMessage, { type: 'assistant' }> { |
| const thinking = |
| durable.thinking && projected.thinking |
| ? { |
| ...projected.thinking, |
| text: reconcileAssistantText(durable.thinking.text, projected.thinking.text), |
| } |
| : (projected.thinking ?? durable.thinking); |
| return { |
| ...projected, |
| text: reconcileAssistantText(durable.text, projected.text), |
| ...(thinking ? { thinking } : {}), |
| }; |
| } |
| |
| function reconcileAssistantText(projected: string, active: string): string { |
| if (active.startsWith(projected)) return active; |
| if (projected.startsWith(active)) return projected; |
| return projected; |
| } |
| |
| function isCount(value: unknown): value is number { |
| return Number.isSafeInteger(value) && (value as number) >= 0; |
| } |
