From a3e5edbdbf662cc0bef5f6ebf8fb6f6835f90212 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Sun, 6 Sep 2026 02:08:48 +0800 Subject: [PATCH] perf(session-persistence): prepare historical reads before publication --- benchmarks/session-open/session-open.bench.ts | 2 +- .../src/generation.ts | 407 ++++++++---------- .../session-persistence-jsonl/src/index.ts | 380 +++++++++++++--- .../session-persistence-jsonl/src/storage.ts | 78 +++- .../tests/built-migration-worker.e2e.ts | 2 +- .../tests/generation.spec.ts | 393 +++++++++-------- .../tests/jsonl.spec.ts | 355 +++++++++++++-- .../tests/lease.spec.ts | 2 +- .../tests/lease.two-process.e2e.ts | 4 +- .../tests/migration-verifier.spec.ts | 4 +- .../tests/zstd.spec.ts | 12 +- 11 files changed, 1082 insertions(+), 557 deletions(-) diff --git a/benchmarks/session-open/session-open.bench.ts b/benchmarks/session-open/session-open.bench.ts index 11e90cc293..91dde3529d 100644 --- a/benchmarks/session-open/session-open.bench.ts +++ b/benchmarks/session-open/session-open.bench.ts @@ -173,7 +173,7 @@ class SessionOpenBenchmarkSuite { this.legacySourcePath = this.facts.path // Produce one real post-upgrade directory outside every measured interval. const templateRoot = await this.createRoot('first-open', 'post-upgrade-template') - requireReport(await runWorker(templateRoot, 'phase-migrate'), 'phase-migrate') + requireReport(await runWorker(templateRoot, 'agent-resume'), 'agent-resume') this.currentSourcePath = join( templateRoot, SYNTHETIC_SESSION_DIRECTORY, diff --git a/packages/session/session-persistence-jsonl/src/generation.ts b/packages/session/session-persistence-jsonl/src/generation.ts index 0f2ba4df38..1e1fd9aa0f 100644 --- a/packages/session/session-persistence-jsonl/src/generation.ts +++ b/packages/session/session-persistence-jsonl/src/generation.ts @@ -22,8 +22,11 @@ import { basename, dirname, join } from 'node:path' import { performance } from 'node:perf_hooks' import { pipeline, Readable } from 'node:stream' import { scheduler } from 'node:timers/promises' +import { isDeepStrictEqual } from 'node:util' import { constants, createZstdCompress } from 'node:zlib' import { Session } from '@deepseek-ai/dsh-session' +import type { SessionEvent } from '@deepseek-ai/dsh-session' +import { BlockAssembler, expandAssistantStream } from '@deepseek-ai/dsh-llm' import type { SessionFormatArtifact, SessionFormatJsonValue, @@ -62,8 +65,8 @@ export interface JsonlGenerationFormatAdapter { isUnsupportedMigrationError?(error: unknown): error is Error } -/** Inputs for ensuring one already-resolved generation has a current successor. */ -export interface EnsureJsonlGenerationOptions { +/** Inputs for preparing one historical generation and publishing its current successor later. */ +export interface PrepareJsonlMigrationOptions { /** Immutable generation selected by the backend resolver. */ readonly sourcePath: string /** Version selected from the source filename and independently checked against its header. */ @@ -101,36 +104,24 @@ export interface JsonlExpectedPrefix { readonly digest: string } -/** Result of current classification or exclusive publication. */ -export type EnsureJsonlGenerationResult = - | { - readonly status: 'current' - readonly version: number - readonly path: string - readonly snapshot: JsonlPhysicalSnapshot - } - | { - readonly status: 'migrated' - readonly fromVersion: number - readonly toVersion: number - readonly path: string - readonly sourcePath: string - readonly snapshot: JsonlPhysicalSnapshot - } +/** A historical source changed after its single decode and migration pass. */ +export class JsonlGenerationSourceChangedError extends Error { + override readonly name = 'JsonlGenerationSourceChangedError' -/** A future physical header was readable, but this writer cannot interpret it. */ -export class JsonlGenerationNewerVersionError extends Error { - override readonly name = 'JsonlGenerationNewerVersionError' - - constructor( - readonly storedVersion: number, - readonly currentVersion: number, - readonly storedId: string, - ) { - super(`session log format v${storedVersion} is newer than current v${currentVersion}`) + /** @param path - historical generation whose revision changed. */ + constructor(readonly path: string) { + super(`historical session generation changed during migration: "${path}"`) } } +/** Current logical state prepared independently from durable publication. */ +export interface PreparedJsonlMigration { + readonly sourceIdentity: JsonlPhysicalIdentity + readonly artifact: SessionFormatArtifact + /** Encode, verify, and exclusively publish once; every call shares the same success or failure. */ + publish(): Promise +} + /** A historical artifact is intact, but the format edge refuses its contents. */ export class JsonlGenerationUnsupportedMigrationError extends Error { override readonly name = 'JsonlGenerationUnsupportedMigrationError' @@ -172,23 +163,12 @@ export interface JsonlPhysicalIdentity { readonly ctimeNs: bigint } -/** One revision-stable physical artifact returned to the immediate backend decoder. */ -export interface JsonlPhysicalSnapshot extends StablePhysicalFile { - readonly headerValue: Record - readonly headerRecord: Buffer -} - /** Exact bytes of one stable file revision together with the stat identity that proved it stable. */ export interface StablePhysicalFile { readonly bytes: Buffer readonly identity: JsonlPhysicalIdentity } -interface JsonlPhysicalHeader { - readonly value: Record - readonly record: Buffer -} - interface GenerationFileSystem { open(path: string, flags: string, mode?: number): Promise readFile(path: string, signal?: AbortSignal): Promise @@ -219,7 +199,7 @@ export type JsonlGenerationRuntimeOverrides = Partial - ensure(options: EnsureJsonlGenerationOptions): Promise + prepare(options: PrepareJsonlMigrationOptions): Promise verify( path: string, compression: JsonlCompression, @@ -260,10 +240,6 @@ function identity(value: JsonlPhysicalIdentity): string { return [value.dev, value.ino, value.size, value.mtimeNs, value.ctimeNs].join(':') } -function fingerprint(value: JsonlPhysicalIdentity, bytes: Buffer): string { - return `${identity(value)}:${createHash('sha256').update(bytes).digest('hex')}` -} - /** * Read one stable revision of a JSONL file with a single retry. If an append * overlaps both reads, return the second read's committed pre-read prefix @@ -313,10 +289,6 @@ function storedVersion(header: unknown): number { return version as number } -function storedId(header: unknown): string { - return String((header as { id?: unknown }).id) -} - function parseJson(text: string, subject: string): unknown { try { return JSON.parse(text) @@ -398,10 +370,15 @@ interface StartedMigrationStream { async function startMigrationStream( headerRecord: Buffer, + sourceVersion: number, format: JsonlGenerationFormatAdapter, - validateHistoricalHeader?: EnsureJsonlGenerationOptions['validateHistoricalHeader'], + validateHistoricalHeader?: PrepareJsonlMigrationOptions['validateHistoricalHeader'], ): Promise { const value = parseJson(headerRecord.subarray(0, -1).toString('utf8'), 'header line') + const version = storedVersion(value) + if (version !== sourceVersion) { + throw new Error(`resolved JSONL source filename identifies v${sourceVersion}, but its header identifies v${version}`) + } const header = value as Record const validation = validateHistoricalHeader?.(header) if (validation !== undefined) await validation @@ -430,17 +407,18 @@ async function consumeMigrationBytes( async function decodeStreamingMigration( bytes: Buffer, compression: JsonlCompression, + sourceVersion: number, format: JsonlGenerationFormatAdapter, - validateHistoricalHeader: EnsureJsonlGenerationOptions['validateHistoricalHeader'], + validateHistoricalHeader: PrepareJsonlMigrationOptions['validateHistoricalHeader'], signal?: AbortSignal, ): Promise { signal?.throwIfAborted() if (compression === 'none') { const headerEnd = bytes.indexOf(0x0A) - /* v8 ignore next -- ensureCurrent's physical-header preflight already requires this newline. */ if (headerEnd === -1) throw new Error('empty or header-less session log') const stream = await startMigrationStream( bytes.subarray(0, headerEnd + 1), + sourceVersion, format, validateHistoricalHeader, ) @@ -457,7 +435,6 @@ async function decodeStreamingMigration( } const { frames, tornStart } = scanZstdFrames(bytes) - /* v8 ignore next -- ensureCurrent's physical-header preflight already requires a complete header frame. */ if (frames.length === 0) throw new Error('empty or header-less Zstandard session log') const decoder = createZstdFrameDecoder() try { @@ -468,6 +445,7 @@ async function decodeStreamingMigration( assertIndependentHeaderFrame(first.value) const stream = await startMigrationStream( first.value, + sourceVersion, format, validateHistoricalHeader, ) @@ -557,7 +535,9 @@ async function verifyCurrentGeneration( generation.events, generation.meta, generation.inheritedEventCount, + 'detached', ) + assertCurrentAssistantStreams(generation.events) return { identity: snapshot.identity, bytes: snapshot.bytes.length, @@ -565,6 +545,32 @@ async function verifyCurrentGeneration( } } +/** Fully replay embedded streams only inside isolated current-generation verification. */ +function assertCurrentAssistantStreams(events: readonly SessionEvent[]): void { + for (const [index, event] of events.entries()) { + if (event.type !== 'assistant/message' && event.type !== 'assistant/attempt') continue + const assembler = new BlockAssembler() + let timed: ReturnType + try { + timed = expandAssistantStream(event.data.stream) + for (const member of timed) assembler.push(member.chunk) + } catch (error: unknown) { + throw new Error(`seed ${event.type} at index ${index} has an invalid embedded stream`, { cause: error }) + } + if (event.type === 'assistant/attempt' || timed.length === 0) continue + const content = event.data.interrupted === true ? assembler.interruptedBlocks() : assembler.blocks() + if (!isDeepStrictEqual(event.data.message.content, content)) { + throw new Error(`seed assistant/message at index ${index} content disagrees with its embedded stream`) + } + if (!isDeepStrictEqual(event.data.usage, assembler.usage)) { + throw new Error(`seed assistant/message at index ${index} usage disagrees with its embedded stream`) + } + if (!isDeepStrictEqual(event.data.message.source.replayState, assembler.replayState)) { + throw new Error(`seed assistant/message at index ${index} replay state disagrees with its embedded stream`) + } + } +} + function decodeCurrentGeneration( bytes: Buffer, compression: JsonlCompression, @@ -620,45 +626,6 @@ function assertIndependentHeaderFrame(plaintext: Buffer): void { } } -function readRawHeader(bytes: Buffer): JsonlPhysicalHeader { - const newline = bytes.indexOf(0x0A) - if (newline === -1) throw new Error('empty or header-less session log') - const record = bytes.subarray(0, newline + 1) - const value = parseJson(record.subarray(0, -1).toString('utf8'), 'header line') - storedVersion(value) - return { value: value as Record, record } -} - -function readZstdHeader(bytes: Buffer, signal?: AbortSignal): JsonlPhysicalHeader { - signal?.throwIfAborted() - const first = scanZstdFrames(bytes, 1).frames[0] - if (first === undefined) throw new Error('empty or header-less Zstandard session log') - const decoder = createZstdFrameDecoder() - const decodedFrames = decoder.decode(bytes, [first]) - try { - const decoded = decodedFrames.next() - /* v8 ignore next -- one complete frame yields once or the decoder throws. */ - if (decoded.done) throw new Error('empty or header-less Zstandard session log') - signal?.throwIfAborted() - assertIndependentHeaderFrame(decoded.value) - const record = Buffer.from(decoded.value) - const value = parseJson(record.subarray(0, -1).toString('utf8'), 'header line') - storedVersion(value) - return { value: value as Record, record } - } finally { - decodedFrames.return() - decoder.close() - } -} - -function readPhysicalHeader( - bytes: Buffer, - compression: JsonlCompression, - signal: AbortSignal | undefined, -): JsonlPhysicalHeader { - return compression === 'zstd' ? readZstdHeader(bytes, signal) : readRawHeader(bytes) -} - function assertGenerationPaths( sourcePath: string, sourceVersion: number, @@ -823,7 +790,6 @@ async function removeTemporary( try { await internals.fs.rm(path) } catch (cleanupFailure: unknown) { - if (primaryFailure === undefined) throw cleanupFailure throw new AggregateError( [primaryFailure, cleanupFailure], `failed to clean migration temporary "${path}" after an earlier failure`, @@ -879,19 +845,16 @@ function asError(error: unknown): Error { async function inspectExpectedCurrent( currentPath: string, - checkCanonicalTargetName: boolean, internals: JsonlGenerationInternals, inspect: () => Promise, ): Promise { try { - if (checkCanonicalTargetName) { - const expectedName = basename(currentPath) - const names = await internals.fs.readdir(dirname(currentPath)) - if (!names.includes(expectedName)) { - const noncanonical = names.find(name => name.toLowerCase() === expectedName.toLowerCase()) - if (noncanonical !== undefined) { - throw new Error(`target resolves to noncanonical directory entry "${noncanonical}"`) - } + const expectedName = basename(currentPath) + const names = await internals.fs.readdir(dirname(currentPath)) + if (!names.includes(expectedName)) { + const noncanonical = names.find(name => name.toLowerCase() === expectedName.toLowerCase()) + if (noncanonical !== undefined) { + throw new Error(`target resolves to noncanonical directory entry "${noncanonical}"`) } } const info = await internals.fs.lstat(currentPath) @@ -913,39 +876,71 @@ function withOverrides(overrides: JsonlGenerationRuntimeOverrides): JsonlGenerat } } -async function reopenExpectedCurrent( - currentPath: string, - staged: StreamedMigrationStage, - compression: JsonlCompression, - expectedId: string, - expectedEventCount: number, - verifyCurrentFile: EnsureJsonlGenerationOptions['verifyCurrentFile'], - signal: AbortSignal | undefined, - checkCanonicalTargetName: boolean, +async function publishPreparedMigration( + options: PrepareJsonlMigrationOptions, + suffix: string, + artifact: SessionFormatArtifact, + sourceIdentity: JsonlPhysicalIdentity, internals: JsonlGenerationInternals, -): Promise { - return inspectExpectedCurrent(currentPath, checkCanonicalTargetName, internals, async () => { - const verified = await verifyCurrentFile( - currentPath, +): Promise { + await scheduler.yield() + const { sourcePath, currentPath, compression, verifyCurrentFile } = options + const eventCount = artifact.events.length + let staged = await writeSyncedTemp(currentPath, suffix, compression, artifact, options.format, undefined, internals) + try { + const verifiedStage = await verifyCurrentFile( + staged.path, compression, - expectedId, - expectedEventCount, - staged, - signal, + artifact.header.id, + eventCount, ) - if (verified.bytes !== staged.bytes || verified.digest !== staged.digest) { - throw new Error('target bytes differ from the migrated generation') + if (verifiedStage.bytes !== staged.bytes || verifiedStage.digest !== staged.digest) { + throw new Error('staged session generation changed during verification') } - const snapshot = await readStableSnapshot(currentPath, signal, internals.fs) - const header = readPhysicalHeader(snapshot.bytes, compression, signal) - return { ...snapshot, headerValue: header.value, headerRecord: header.record } - }) + await internals.barrier('before-source-check', 1) + const beforePublish = await internals.fs.stat(sourcePath) + if (identity(beforePublish) !== identity(sourceIdentity)) { + throw new JsonlGenerationSourceChangedError(sourcePath) + } + const published = await publishCurrentExclusive(staged.path, currentPath, internals) + if (published && internals.platform === 'win32') staged = { ...staged, path: '' } + await internals.barrier('after-publication', 1) + let currentIdentity: JsonlPhysicalIdentity + if (published) { + if (staged.path !== '') { + await removeCommittedTemporary(staged.path, internals) + staged = { ...staged, path: '' } + } + currentIdentity = await internals.fs.stat(currentPath) + } else { + const winner = await inspectExpectedCurrent(currentPath, internals, async () => { + const candidate = await verifyCurrentFile( + currentPath, + compression, + artifact.header.id, + eventCount, + staged, + ) + if (candidate.bytes !== staged.bytes || candidate.digest !== staged.digest) { + throw new Error('target bytes differ from the migrated generation') + } + return candidate + }) + currentIdentity = winner.identity + await removeCommittedTemporary(staged.path, internals) + staged = { ...staged, path: '' } + } + return currentIdentity + } catch (error: unknown) { + if (staged.path !== '') await removeTemporary(staged.path, error, internals) + throw error + } } -async function ensureCurrent( - options: EnsureJsonlGenerationOptions, +async function prepareMigration( + options: PrepareJsonlMigrationOptions, internals: JsonlGenerationInternals, -): Promise { +): Promise { const { sourcePath, sourceVersion, currentPath, compression, format, signal } = options const suffix = assertGenerationPaths( sourcePath, @@ -954,124 +949,58 @@ async function ensureCurrent( format.currentVersion, compression, ) - let attempt = 0 - for (;;) { - attempt += 1 - signal?.throwIfAborted() - const source = await readStableSnapshot(sourcePath, signal, internals.fs) - const quickHeader = readPhysicalHeader(source.bytes, compression, signal) - const quickVersion = storedVersion(quickHeader.value) - if (quickVersion !== sourceVersion) { - throw new Error( - `resolved JSONL source filename identifies v${sourceVersion}, but its header identifies v${quickVersion}: ` - + sourcePath, - ) + if (sourceVersion >= format.currentVersion) { + throw new Error(`migration preparation requires a historical source, got v${sourceVersion}`) + } + const source = await readStableSnapshot(sourcePath, signal, internals.fs) + let artifact: SessionFormatArtifact + try { + artifact = await decodeStreamingMigration( + source.bytes, + compression, + sourceVersion, + format, + options.validateHistoricalHeader, + signal, + ) + } catch (error: unknown) { + if (format.isUnsupportedMigrationError?.(error) === true) { + throw new JsonlGenerationUnsupportedMigrationError(sourceVersion, error) } - if (sourceVersion > format.currentVersion) { - throw new JsonlGenerationNewerVersionError( - sourceVersion, - format.currentVersion, - storedId(quickHeader.value), - ) - } - if (sourceVersion === format.currentVersion) { - return { - status: 'current', - version: quickVersion, - path: sourcePath, - snapshot: { - ...source, - headerValue: quickHeader.value, - headerRecord: quickHeader.record, - }, + throw error + } + if (artifact.header.version !== format.currentVersion) { + throw new Error(`format migration returned v${artifact.header.version}, expected v${format.currentVersion}`) + } + const sourceIdentity = source.identity + let publication: Promise | undefined + return { + sourceIdentity, + artifact, + publish() { + if (publication === undefined) { + publication = publishPreparedMigration( + options, + suffix, + artifact, + sourceIdentity, + internals, + ) } - } - let artifact: SessionFormatArtifact - try { - artifact = await decodeStreamingMigration( - source.bytes, - compression, - format, - options.validateHistoricalHeader, - signal, - ) - } catch (error: unknown) { - if (format.isUnsupportedMigrationError?.(error) === true) { - throw new JsonlGenerationUnsupportedMigrationError(sourceVersion, error) - } - throw error - } - if (artifact.header.version !== format.currentVersion) { - throw new Error(`format migration returned v${artifact.header.version}, expected v${format.currentVersion}`) - } - - await scheduler.yield() - signal?.throwIfAborted() - const sourceFingerprint = fingerprint(source.identity, source.bytes) - const eventCount = artifact.events.length - let staged = await writeSyncedTemp(currentPath, suffix, compression, artifact, format, signal, internals) - let failure: unknown - try { - const verifiedStage = await options.verifyCurrentFile( - staged.path, - compression, - artifact.header.id, - eventCount, - undefined, - signal, - ) - if (verifiedStage.bytes !== staged.bytes || verifiedStage.digest !== staged.digest) { - throw new Error('staged session generation changed during verification') - } - await internals.barrier('before-source-check', attempt) - const beforePublish = await readStableSnapshot(sourcePath, signal, internals.fs) - if (fingerprint(beforePublish.identity, beforePublish.bytes) !== sourceFingerprint) continue - - const published = await publishCurrentExclusive(staged.path, currentPath, internals) - if (published && internals.platform === 'win32') staged = { ...staged, path: '' } - await internals.barrier('after-publication', attempt) - signal?.throwIfAborted() - const committed = await reopenExpectedCurrent( - currentPath, - staged, - compression, - artifact.header.id, - eventCount, - options.verifyCurrentFile, - signal, - !published, - internals, - ) - if (staged.path !== '') { - await removeCommittedTemporary(staged.path, internals) - staged = { ...staged, path: '' } - } - return { - status: 'migrated', - fromVersion: sourceVersion, - toVersion: format.currentVersion, - path: currentPath, - sourcePath, - snapshot: committed, - } - } catch (error: unknown) { - failure = error - throw error - } finally { - if (staged.path !== '') await removeTemporary(staged.path, failure, internals) - } + return publication + }, } } /** - * Ensure one resolved generation has a current-format successor before returning. - * @param options - resolved source, current target, format adapter, verification, and cancellation. - * @returns the current source or the verified and reopened migrated successor. + * Decode and migrate one historical generation without writing its successor. + * @param options - resolved source, current target, format adapter, and load cancellation. + * @returns the current artifact and an idempotent explicit publication operation. */ -export function ensureJsonlGenerationCurrent( - options: EnsureJsonlGenerationOptions, -): Promise { - return defaultGenerationRuntime.ensure(options) +export function prepareJsonlMigration( + options: PrepareJsonlMigrationOptions, +): Promise { + return defaultGenerationRuntime.prepare(options) } /** @@ -1085,7 +1014,7 @@ export function createJsonlGenerationRuntime( const internals = withOverrides(overrides) return { readStable: (path, signal) => readStableSnapshot(path, signal, internals.fs), - ensure: options => ensureCurrent(options, internals), + prepare: options => prepareMigration(options, internals), verify: (path, compression, expectedId, expectedEventCount, expectedPrefix) => verifyCurrentGeneration( path, compression, expectedId, expectedEventCount, internals.fs, expectedPrefix, ), diff --git a/packages/session/session-persistence-jsonl/src/index.ts b/packages/session/session-persistence-jsonl/src/index.ts index 2020acaff8..ba4f8996f0 100644 --- a/packages/session/session-persistence-jsonl/src/index.ts +++ b/packages/session/session-persistence-jsonl/src/index.ts @@ -24,12 +24,13 @@ import { SessionAlreadyExistsError, SessionPersistenceNotFoundError, assertStoredId, materializeCreateHeader, sessionFormatVersionRefusal, validateStoredEvents, type SessionAccess, type SessionHandle, + type SessionHandleReadResult, type SessionLocation, type SessionPersistenceCreateOptions, type SessionPersistenceListOptions, type SessionPersistenceOpenOptions, type SessionPersistenceSnapshot, type SessionPersistenceStatOptions, type SessionPersistenceRevision as PersistenceRevision, } from '@deepseek-ai/dsh-session-persistence' -import { JsonlBackendTracker, JsonlSessionHandle } from './storage.ts' +import { JsonlBackendTracker, JsonlSessionHandle, type StorageHandleState } from './storage.ts' import { SessionWriteLease } from './lease.ts' import { SESSION_FORMAT_VERSION, SessionId as makeSessionId, SessionLogOffset } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionId, SessionHeader, SessionLogOffset as SessionLogOffsetType } from '@deepseek-ai/dsh-session' @@ -44,13 +45,13 @@ import { import { ensureDurableDirectoryWin32, publishNewFileWin32 } from './win32.ts' import { verifyCurrentGenerationInWorker } from './migration-verifier.ts' import { - ensureJsonlGenerationCurrent, - JsonlGenerationNewerVersionError, + JsonlGenerationSourceChangedError, JsonlGenerationUnsupportedMigrationError, + prepareJsonlMigration, readStableJsonlFile, - type EnsureJsonlGenerationResult, type JsonlGenerationFormatAdapter, type JsonlPhysicalIdentity, + type PreparedJsonlMigration, } from './generation.ts' export type { JsonlCompression } from './format.ts' @@ -97,11 +98,14 @@ export interface Config { compression?: JsonlCompression } -/** A parsed, validated stored log: header, logical events, and any torn-tail repair state. */ -interface StoredLog { +/** One stored event graph whose producer has established immutable sharing. */ +interface FrozenStoredEvents extends SessionHandleReadResult { + readonly eventState: 'shared-frozen' +} + +/** State shared by prepared historical and published current logs. */ +interface StoredLogBase extends FrozenStoredEvents { readonly meta: SessionHeader - /** The logical log, including any events recovered from a torn final frame. */ - readonly events: SessionEvent[] readonly tornTruncateTo: number | undefined /** Complete events recovered from the torn final frame; the write path rewrites them durably. */ readonly recoveredTail: SessionEvent[] @@ -110,6 +114,45 @@ interface StoredLog { readonly revision: PersistenceRevision } +/** A decoded current generation that is already durable. */ +interface CurrentStoredLog extends StoredLogBase { + readonly status: 'current' +} + +/** A migrated historical generation retained until an explicit write open publishes it. */ +interface PreparedStoredLog extends StoredLogBase { + readonly status: 'prepared' + readonly publication: { + readonly source: ResolvedJsonlGeneration + readonly value: PreparedJsonlMigration + } +} + +/** A validated logical log, either durable current state or prepared historical state. */ +type StoredLog = CurrentStoredLog | PreparedStoredLog + +/** Deep-freeze one acyclic stored JSON event without recursive calls. */ +function freezeStoredEvent(event: SessionEvent): void { + const pending: object[] = [event] + while (pending.length > 0) { + // The non-empty check proves an object remains to visit. + // oxlint-disable-next-line typescript/no-non-null-assertion + const current = pending.pop()! + Object.freeze(current) + for (const key in current) { + const child = (current as Record)[key] + if (child !== null && typeof child === 'object') pending.push(child) + } + } +} + +/** Establish immutable sharing for one decoded event graph and report that state. */ +function freezeStoredEvents(events: SessionEvent[]): FrozenStoredEvents { + for (const event of events) freezeStoredEvent(event) + Object.freeze(events) + return { eventState: 'shared-frozen', events } +} + /** One authoritative immutable generation selected from a Session directory. */ interface ResolvedJsonlGeneration { readonly sourcePath: string @@ -117,6 +160,16 @@ interface ResolvedJsonlGeneration { readonly currentPath: string } +/** One backend-owned historical preparation shared by its current callers. */ +interface MigrationPreparation { + readonly sourcePath: string + readonly sourceRevision: PersistenceRevision + readonly controller: AbortController + readonly promise: Promise + settled: boolean + waiters: number +} + /** Build the stat-derived best-effort change token shared by full and lightweight reads. */ function fileRevision(identity: JsonlPhysicalIdentity): PersistenceRevision { return SessionPersistenceRevision([ @@ -138,6 +191,41 @@ function isErrnoException(error: unknown): error is NodeJS.ErrnoException { return typeof (error as NodeJS.ErrnoException | null)?.code === 'string' } +/** Preserve an Error abort reason and normalize hostile non-Error reasons. */ +function abortError(signal: AbortSignal): Error { + return signal.reason instanceof Error + ? signal.reason + : new Error('session migration preparation aborted', { cause: signal.reason }) +} + +/** Let one caller stop waiting without transferring cancellation ownership to shared work. */ +function waitWithAbort(operation: Promise, signal?: AbortSignal): Promise { + if (signal === undefined) return operation + /* v8 ignore next -- requireStoredLog synchronously rechecks the signal immediately before waiting. */ + if (signal.aborted) return Promise.reject(abortError(signal)) + return new Promise((resolve, reject) => { + const stopWaiting = (): void => { + reject(abortError(signal)) + } + signal.addEventListener('abort', stopWaiting, { once: true }) + void operation.then( + (value) => { + signal.removeEventListener('abort', stopWaiting) + resolve(value) + }, + (error: unknown) => { + signal.removeEventListener('abort', stopWaiting) + /* v8 ignore else -- the preparation owner normalizes every rejection before this waiter sees it. */ + if (error instanceof Error) { + reject(error) + } else { + reject(new Error('session migration preparation failed', { cause: error })) + } + }, + ) + }) +} + /** * The JSONL persistence backend. Load as a plugin; it registers as * `ctx.sessionPersistence`. Sessions materialize lazily: a created session is @@ -166,6 +254,8 @@ class JsonlSessionPersistence extends SessionPersistence { * revision guard. */ private readonly coldLogMemo = new Map() + /** One joinable decode/migration operation per selected historical Session file revision. */ + private readonly migrationPreparations = new Map() constructor(ctx: Context, public config: Config) { super(ctx) @@ -253,11 +343,22 @@ class JsonlSessionPersistence extends SessionPersistence { return this.tracker.adopt(new JsonlSessionHandle(this, id, pending.header, 'read', { cursor: 0, materialized: false, inheritedEventCount: pending.inheritedEventCount })) } const stored = await this.requireStoredLog(id, options?.signal) - return this.tracker.adopt(new JsonlSessionHandle(this, id, stored.meta, 'read', { - cursor: 0, - materialized: true, - inheritedEventCount: stored.inheritedEventCount, - })) + let state: StorageHandleState + if (stored.status === 'prepared') { + state = { + cursor: 0, + materialized: true, + inheritedEventCount: stored.inheritedEventCount, + primed: stored, + } + } else { + state = { + cursor: 0, + materialized: true, + inheritedEventCount: stored.inheritedEventCount, + } + } + return this.tracker.adopt(new JsonlSessionHandle(this, id, stored.meta, 'read', state)) } // A pending entry always belongs to an ACTIVE creator handle (close erases // it), so the claim below rejects that case as already owned. @@ -267,14 +368,22 @@ class JsonlSessionPersistence extends SessionPersistence { const resolved = await this.findLog(id, options?.signal) if (resolved === undefined) throw new SessionPersistenceNotFoundError(id) lease = await this.acquireLease(id, undefined, dirname(resolved.currentPath)) - const stored = await this.requireStoredLog(id, options?.signal) + const prepared = await this.requireStoredLog(id, options?.signal) + options?.signal?.throwIfAborted() + let stored: CurrentStoredLog + if (prepared.status === 'prepared') { + stored = await this.publishStoredMigration(id, prepared) + } else { + stored = prepared + } + options?.signal?.throwIfAborted() return this.tracker.adopt(new JsonlSessionHandle(this, id, stored.meta, 'write', { cursor: stored.events.length, materialized: true, tornTruncateTo: stored.tornTruncateTo, recoveredTail: stored.recoveredTail, inheritedEventCount: stored.inheritedEventCount, - primed: stored.events, + primed: stored, }, lease)) } catch (error) { // Free the in-process claim no matter how the kernel-lock release @@ -386,34 +495,115 @@ class JsonlSessionPersistence extends SessionPersistence { private async requireStoredLog(id: SessionId, signal?: AbortSignal): Promise { const selected = await this.findLog(id, signal) if (selected === undefined) throw new SessionPersistenceNotFoundError(id) - if (selected.sourceVersion === SESSION_FORMAT_VERSION) { - const probe = fileRevision(await stat(selected.sourcePath, { bigint: true })) - const memoized = this.coldLogMemo.get(id) - if (memoized !== undefined && memoized.revision === probe) { - this.coldLogMemo.delete(id) - this.coldLogMemo.set(id, memoized) - return memoized + if (selected.sourceVersion < SESSION_FORMAT_VERSION) { + const sourceRevision = fileRevision(await stat(selected.sourcePath, { bigint: true })) + signal?.throwIfAborted() + let preparation = this.migrationPreparations.get(id) + if (preparation === undefined + || preparation.sourcePath !== selected.sourcePath + || preparation.sourceRevision !== sourceRevision) { + const controller = new AbortController() + const promise = this.loadStoredMigration(id, selected, sourceRevision, controller.signal) + preparation = { + sourcePath: selected.sourcePath, + sourceRevision, + controller, + promise, + settled: false, + waiters: 0, + } + this.migrationPreparations.set(id, preparation) + const created = preparation + const release = (): void => { + created.settled = true + if (this.migrationPreparations.get(id) === created) { + this.migrationPreparations.delete(id) + } + } + void promise.then(release, release) } + signal?.throwIfAborted() + return this.waitForPreparation(id, preparation, signal) } - const current = await this.ensureCurrentLog(id, signal, selected) + if (selected.sourceVersion > SESSION_FORMAT_VERSION) { + const header = await this.readGenerationHeader(selected, id, signal) + /* v8 ignore else -- a readable future header is rejected inside readGenerationHeader. */ + if (header === undefined) { + throw new SessionPersistenceCorruptionError( + `session "${id}": stored log has a malformed header (raw log: ${selected.sourcePath})`, + { cause: new Error('malformed Session header') }, + ) + } + /* v8 ignore next -- readGenerationHeader rejects every future version. */ + throw new SessionFormatUnsupportedError( + `${sessionFormatVersionRefusal(id, selected.sourceVersion)} (raw log: ${selected.sourcePath})`, + { kind: 'jsonl', path: selected.sourcePath }, + ) + } + const probe = fileRevision(await stat(selected.sourcePath, { bigint: true })) + const memoized = this.coldLogMemo.get(id) + if (memoized?.status === 'current' && memoized.revision === probe) { + this.coldLogMemo.delete(id) + this.coldLogMemo.set(id, memoized) + return memoized + } + const current = await readStableJsonlFile(selected.sourcePath, signal) return this.decodeStoredLog( - current.path, + selected.sourcePath, id, - current.snapshot.bytes, - fileRevision(current.snapshot.identity), + current.bytes, + fileRevision(current.identity), signal, ) } - /** Select and, when required, publish one immutable current generation. */ - private async ensureCurrentLog( + /** Probe the memo and otherwise decode one historical generation under backend cancellation. */ + private async loadStoredMigration( id: SessionId, - signal: AbortSignal | undefined, selected: ResolvedJsonlGeneration, - ): Promise { - signal?.throwIfAborted() + sourceRevision: PersistenceRevision, + signal: AbortSignal, + ): Promise { + signal.throwIfAborted() + const memoized = this.coldLogMemo.get(id) + if (memoized?.status === 'prepared' && memoized.revision === sourceRevision) { + this.coldLogMemo.delete(id) + this.coldLogMemo.set(id, memoized) + return memoized + } + return this.prepareStoredMigration(id, selected, signal) + } + + /** Await shared preparation for one caller and abort it only after its last waiter leaves. */ + private async waitForPreparation( + id: SessionId, + preparation: MigrationPreparation, + signal?: AbortSignal, + ): Promise { + preparation.waiters += 1 try { - return await ensureJsonlGenerationCurrent({ + return await waitWithAbort(preparation.promise, signal) + } finally { + preparation.waiters -= 1 + if (preparation.waiters === 0 && !preparation.settled) { + /* v8 ignore else -- a newer selected source may already own this id's preparation slot. */ + if (this.migrationPreparations.get(id) === preparation) { + this.migrationPreparations.delete(id) + } + preparation.controller.abort() + } + } + } + + /** Decode one historical generation without publishing a successor. */ + private async prepareStoredMigration( + id: SessionId, + selected: ResolvedJsonlGeneration, + signal: AbortSignal, + ): Promise { + let prepared: Awaited> + try { + prepared = await prepareJsonlMigration({ sourcePath: selected.sourcePath, sourceVersion: selected.sourceVersion, currentPath: selected.currentPath, @@ -426,32 +616,75 @@ class JsonlSessionPersistence extends SessionPersistence { id, signal, ), - ...(signal === undefined ? {} : { signal }), + signal, }) } catch (error: unknown) { - signal?.throwIfAborted() - if (error instanceof JsonlGenerationNewerVersionError) { - const reason = sessionFormatVersionRefusal(error.storedId, error.storedVersion) - throw new SessionFormatUnsupportedError( - `${reason} (raw log: ${selected.sourcePath})`, - { kind: 'jsonl', path: selected.sourcePath }, - ) - } - if (error instanceof JsonlGenerationUnsupportedMigrationError) { - throw new SessionFormatUnsupportedError( - `${error.message}; source v${error.fromVersion} artifact remains unchanged (raw log: ${selected.sourcePath})`, - { kind: 'jsonl', path: selected.sourcePath }, - ) - } - if (error instanceof SessionFormatUnsupportedError - || error instanceof SessionPersistenceCorruptionError - || isErrnoException(error) - || error instanceof DOMException && error.name === 'AbortError') throw error - throw new SessionPersistenceCorruptionError( - `session "${id}": stored log is corrupt: ${String(error)} (raw log: ${selected.sourcePath})`, - { cause: error }, + throw this.generationFailure(id, selected, error) + } + const meta = this.currentHeader(prepared.artifact.header) + assertStoredId(id, meta) + const events = prepared.artifact.events as SessionEvent[] + validateStoredEvents(meta, events, { kind: 'jsonl', path: selected.sourcePath }) + const stored: PreparedStoredLog = { + status: 'prepared', + meta, + ...freezeStoredEvents(events), + tornTruncateTo: undefined, + recoveredTail: [], + inheritedEventCount: SessionLogOffset(prepared.artifact.inheritedEventCount), + revision: fileRevision(prepared.sourceIdentity), + publication: { source: selected, value: prepared }, + } + this.memoizeStoredLog(id, stored) + return stored + } + + /** Publish a prepared historical log before granting write access. */ + private async publishStoredMigration(id: SessionId, stored: PreparedStoredLog): Promise { + const migration = stored.publication + let identity: JsonlPhysicalIdentity + try { + identity = await migration.value.publish() + } catch (error: unknown) { + /* v8 ignore else -- a newer preparation may have replaced this stale cache entry. */ + if (this.coldLogMemo.get(id) === stored) this.coldLogMemo.delete(id) + throw this.generationFailure(id, migration.source, error) + } + const published: CurrentStoredLog = { + status: 'current', + meta: stored.meta, + eventState: stored.eventState, + events: stored.events, + tornTruncateTo: stored.tornTruncateTo, + recoveredTail: stored.recoveredTail, + inheritedEventCount: stored.inheritedEventCount, + revision: fileRevision(identity), + } + this.memoizeStoredLog(id, published) + return published + } + + /** Translate generation-layer failures into the persistence seam's error vocabulary. */ + private generationFailure( + id: SessionId, + selected: ResolvedJsonlGeneration, + error: unknown, + ): Error { + if (error instanceof JsonlGenerationUnsupportedMigrationError) { + return new SessionFormatUnsupportedError( + `${error.message}; source v${error.fromVersion} artifact remains unchanged (raw log: ${selected.sourcePath})`, + { kind: 'jsonl', path: selected.sourcePath }, ) } + if (error instanceof JsonlGenerationSourceChangedError) return error + if (error instanceof SessionFormatUnsupportedError + || error instanceof SessionPersistenceCorruptionError + || isErrnoException(error) + || error instanceof DOMException && error.name === 'AbortError') return error + return new SessionPersistenceCorruptionError( + `session "${id}": stored log is corrupt: ${String(error)} (raw log: ${selected.sourcePath})`, + { cause: error }, + ) } /** @@ -461,11 +694,11 @@ class JsonlSessionPersistence extends SessionPersistence { * @param signal - optional cancellation for the stat/read/decode work. * @returns the validated stored log with any torn-tail truncation point. */ - async readStoredLog(path: string, expectedId: SessionId, signal?: AbortSignal): Promise { + async readStoredLog(path: string, expectedId: SessionId, signal?: AbortSignal): Promise { signal?.throwIfAborted() const probe = fileRevision(await stat(path, { bigint: true })) const memoized = this.coldLogMemo.get(expectedId) - if (memoized !== undefined && memoized.revision === probe) { + if (memoized?.status === 'current' && memoized.revision === probe) { this.coldLogMemo.delete(expectedId) this.coldLogMemo.set(expectedId, memoized) return memoized @@ -481,7 +714,7 @@ class JsonlSessionPersistence extends SessionPersistence { buffer: Buffer, revision: PersistenceRevision, signal?: AbortSignal, - ): Promise { + ): Promise { let parsed: { meta: SessionHeader inheritedEventCount: SessionLogOffsetType @@ -523,30 +756,45 @@ class JsonlSessionPersistence extends SessionPersistence { assertStoredId(expectedId, parsed.meta) const location = this.locate(parsed.meta) validateStoredEvents(parsed.meta, parsed.events, location) - const stored: StoredLog = { ...parsed, revision } - this.coldLogMemo.delete(expectedId) - this.coldLogMemo.set(expectedId, stored) + const { events, ...rest } = parsed + const stored: CurrentStoredLog = { + status: 'current', + ...rest, + ...freezeStoredEvents(events), + revision, + } + this.memoizeStoredLog(expectedId, stored) + return stored + } + + /** Insert one parsed log into the bounded handoff cache. */ + private memoizeStoredLog(id: SessionId, stored: StoredLog): void { + this.coldLogMemo.delete(id) + this.coldLogMemo.set(id, stored) for (const oldest of this.coldLogMemo.keys()) { if (this.coldLogMemo.size <= COLD_LOG_MEMO_MAX_ENTRIES) break this.coldLogMemo.delete(oldest) } - return stored } /** - * Resolve a session's unique log path. + * Resolve a session's current-generation log path. * @param id - the stored session to locate. * @param signal - optional cancellation for the directory scans. - * @returns the artifact path, or `undefined` when absent. + * @returns the current artifact path, or `undefined` while only a historical generation exists. */ - async resolveLog(id: SessionId, signal?: AbortSignal): Promise { + async resolveCurrentLog(id: SessionId, signal?: AbortSignal): Promise { await this.ensureRootEncoding() signal?.throwIfAborted() const selected = await this.findLog(id, signal) if (selected === undefined) return undefined if (selected.sourceVersion === SESSION_FORMAT_VERSION) return selected.sourcePath - const current = await this.ensureCurrentLog(id, signal, selected) - return current.path + if (selected.sourceVersion < SESSION_FORMAT_VERSION) return undefined + const reason = sessionFormatVersionRefusal(id, selected.sourceVersion) + throw new SessionFormatUnsupportedError( + `${reason} (raw log: ${selected.sourcePath})`, + { kind: 'jsonl', path: selected.sourcePath }, + ) } /** diff --git a/packages/session/session-persistence-jsonl/src/storage.ts b/packages/session/session-persistence-jsonl/src/storage.ts index 7cb4132cc9..029bb4839d 100644 --- a/packages/session/session-persistence-jsonl/src/storage.ts +++ b/packages/session/session-persistence-jsonl/src/storage.ts @@ -28,6 +28,7 @@ import type { SessionHandleAppendOptions, SessionHandleFlushOptions, SessionHandleReadOptions, + SessionHandleReadResult, } from '@deepseek-ai/dsh-session-persistence' import type { SessionWriteLease } from './lease.ts' @@ -47,10 +48,10 @@ export interface JsonlHandleStorage { persistHeader(header: SessionHeader, inheritedEventCount: SessionLogOffset): Promise /** Truncate a torn physical tail before the first new append lands. */ truncateTornTail(header: SessionHeader, truncateTo: number): Promise - /** Resolve the session's artifact path, or `undefined` before materialization. */ - resolveLog(id: SessionId, signal?: AbortSignal): Promise - /** Read and validate the stored log at `path`. */ - readStoredLog(path: string, expectedId: SessionId, signal?: AbortSignal): Promise<{ events: SessionEvent[] }> + /** Resolve the current-generation artifact path, or `undefined` when absent. */ + resolveCurrentLog(id: SessionId, signal?: AbortSignal): Promise + /** Read and validate the stored log at `path`, including its established event aliasing state. */ + readStoredLog(path: string, expectedId: SessionId, signal?: AbortSignal): Promise /** Whether the id is still a created-but-unmaterialized session here. */ hasPendingSession(id: SessionId): boolean /** Acquire the session's cross-process write lock in its artifact directory. */ @@ -72,7 +73,7 @@ export interface StorageHandleState { /** Exact fork-inherited prefix length stored with the log; `0` when unseeded. */ inheritedEventCount: SessionLogOffset /** The validated stored prefix from a write open, served to reads until the first append. */ - primed?: SessionEvent[] | undefined + primed?: SessionHandleReadResult | undefined } /** @@ -112,9 +113,9 @@ export class JsonlSessionHandle implements SessionHandle { * @param offset - first logical seq to include (default 0). * @param length - maximum events returned (default: the rest). * @param options - optional cancellation. - * @returns the requested slice. + * @returns a slice carrying the aliasing state established by its producer. */ - async read(offset = 0, length = Number.MAX_SAFE_INTEGER, options?: SessionHandleReadOptions): Promise { + async read(offset = 0, length = Number.MAX_SAFE_INTEGER, options?: SessionHandleReadOptions): Promise { // Closed-handle refusal precedes argument validation: a closed handle // rejects SessionHandleClosedError regardless of the arguments. this.assertOpen('read') @@ -125,24 +126,57 @@ export class JsonlSessionHandle implements SessionHandle { throw new TypeError(`read length must be a non-negative safe integer, got ${String(length)}`) } options?.signal?.throwIfAborted() - if (this.state.primed !== undefined) { - this.observedLength = Math.max(this.observedLength, this.state.primed.length) - return this.state.primed.slice(offset, offset + length) + let result: SessionHandleReadResult + const primed = this.state.primed + if (primed !== undefined) { + if (this.access === 'write') { + result = this.readPrimed(primed, offset, length) + } else { + const currentPath = await this.storage.resolveCurrentLog(this.id, options?.signal) + if (currentPath === undefined) { + result = this.readPrimed(primed, offset, length) + } else { + this.state.primed = undefined + result = await this.readCurrent(currentPath, offset, length, options?.signal) + } + } + } else if (this.access === 'write' && !this.state.materialized) { + result = { eventState: 'detached', events: [] } + } else { + const currentPath = await this.storage.resolveCurrentLog(this.id, options?.signal) + if (currentPath !== undefined) { + result = await this.readCurrent(currentPath, offset, length, options?.signal) + } else if (this.storage.hasPendingSession(this.id)) { + result = { eventState: 'detached', events: [] } + } else { + throw new SessionPersistenceNotFoundError(this.id) + } } - // A write handle knows its own materialization; a read handle asks the - // backend so a writer's later materialization becomes visible here. - if (this.access === 'write' && !this.state.materialized) return [] - const path = await this.storage.resolveLog(this.id, options?.signal) - if (path === undefined) { - if (this.storage.hasPendingSession(this.id)) return [] - throw new SessionPersistenceNotFoundError(this.id) + return result + } + + /** Read one slice from the prepared historical prefix retained by this handle. */ + private readPrimed(source: SessionHandleReadResult, offset: number, length: number): SessionHandleReadResult { + this.observedLength = Math.max(this.observedLength, source.events.length) + return { eventState: source.eventState, events: source.events.slice(offset, offset + length) } + } + + /** Read one current physical generation and enforce this handle's monotonic view. */ + private async readCurrent( + path: string, + offset: number, + length: number, + signal?: AbortSignal, + ): Promise { + const source = await this.storage.readStoredLog(path, this.id, signal) + if (source.events.length < this.observedLength) { + throw new Error(`session "${this.id}": stored log shrank below a previously observed prefix (${source.events.length} < ${this.observedLength})`) } - const { events } = await this.storage.readStoredLog(path, this.id, options?.signal) - if (events.length < this.observedLength) { - throw new Error(`session "${this.id}": stored log shrank below a previously observed prefix (${events.length} < ${this.observedLength})`) + this.observedLength = source.events.length + return { + eventState: source.eventState, + events: source.events.slice(offset, offset + length), } - this.observedLength = events.length - return events.slice(offset, offset + length) } /** diff --git a/packages/session/session-persistence-jsonl/tests/built-migration-worker.e2e.ts b/packages/session/session-persistence-jsonl/tests/built-migration-worker.e2e.ts index c02d875a60..555b2d48b8 100644 --- a/packages/session/session-persistence-jsonl/tests/built-migration-worker.e2e.ts +++ b/packages/session/session-persistence-jsonl/tests/built-migration-worker.e2e.ts @@ -27,7 +27,7 @@ describe.skipIf(!built)('built migration verifier (plain node)', () => { type: 'session', version: 0, id, createdAt: 1, delegationDepth: 0, }) + '\\n') await ctx.plugin(JsonlSessionPersistence, { root, compression: 'none' }) - const handle = await ctx.sessionPersistence.open(id, 'read') + const handle = await ctx.sessionPersistence.open(id, 'write') await handle.close() await ctx.sessionPersistence.flush() const header = JSON.parse((await readFile(join(directory, 'session.v2.jsonl'), 'utf8')).trim()) diff --git a/packages/session/session-persistence-jsonl/tests/generation.spec.ts b/packages/session/session-persistence-jsonl/tests/generation.spec.ts index a83fa020ec..4938a47037 100644 --- a/packages/session/session-persistence-jsonl/tests/generation.spec.ts +++ b/packages/session/session-persistence-jsonl/tests/generation.spec.ts @@ -17,19 +17,24 @@ import { import { tmpdir } from 'node:os' import { basename, join } from 'node:path' import { performance } from 'node:perf_hooks' -import { scheduler } from 'node:timers/promises' import { - ensureJsonlGenerationCurrent as ensureJsonlGenerationCurrentProduction, + JsonlGenerationSourceChangedError, JsonlGenerationTargetConflictError, JsonlGenerationUnsupportedMigrationError, + prepareJsonlMigration, verifyJsonlCurrentGeneration, - type EnsureJsonlGenerationOptions, type JsonlGenerationFormatAdapter, + type PrepareJsonlMigrationOptions, } from '../src/generation.ts' import { createJsonlGenerationTestRuntime } from '../src/testing/generation.ts' import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts' import type { JsonlCompression } from '../src/format.ts' -import type { SessionFormatArtifact, SessionFormatRestore } from '@deepseek-ai/dsh-session-format' +import type { + SessionFormatArtifact, + SessionFormatEvent, + SessionFormatJsonValue, + SessionFormatRestore, +} from '@deepseek-ai/dsh-session-format' const roots: string[] = [] @@ -78,6 +83,61 @@ function header(version: number, id = 'generation-test'): Record): SessionFormatRestore @@ -120,12 +180,12 @@ function streamingAdapter(): JsonlGenerationFormatAdapter & { return adapter() } -function verifier(): EnsureJsonlGenerationOptions['verifyCurrentFile'] { +function verifier(): PrepareJsonlMigrationOptions['verifyCurrentFile'] { return (path, compression, expectedId, expectedEventCount, expectedPrefix) => verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount, expectedPrefix) } -const byteVerifier: EnsureJsonlGenerationOptions['verifyCurrentFile'] = async (path) => { +const byteVerifier: PrepareJsonlMigrationOptions['verifyCurrentFile'] = async (path) => { const [bytes, identity] = await Promise.all([readFile(path), stat(path, { bigint: true })]) return { identity, @@ -144,7 +204,7 @@ function options( compression: JsonlCompression = 'none', format: JsonlGenerationFormatAdapter = adapter(), sourceVersion = 0, -): Omit { +): Omit { return { sourcePath: generationPath(root, sourceVersion, compression), sourceVersion, @@ -156,7 +216,7 @@ function options( type TestMigrationOptions = ReturnType & { readonly signal?: AbortSignal - readonly verifyCurrentFile?: EnsureJsonlGenerationOptions['verifyCurrentFile'] + readonly verifyCurrentFile?: PrepareJsonlMigrationOptions['verifyCurrentFile'] } type TestGenerationOverrides = Parameters[0] @@ -175,17 +235,24 @@ async function ensureWithOverrides( expectedPrefix, ) ) - return runtime.ensure({ + const prepared = await runtime.prepare({ ...request, verifyCurrentFile, }) + const identity = await prepared.publish() + const bytes = await readFile(request.currentPath) + return { + status: 'migrated' as const, + fromVersion: request.sourceVersion, + toVersion: request.format.currentVersion, + path: request.currentPath, + sourcePath: request.sourcePath, + snapshot: { identity, bytes }, + } } function ensureJsonlGenerationCurrent(request: TestMigrationOptions) { - return ensureJsonlGenerationCurrentProduction({ - ...request, - verifyCurrentFile: request.verifyCurrentFile ?? verifier(), - }) + return ensureWithOverrides(request, {}) } async function encodeZstd(version: number, rows: readonly unknown[]): Promise { @@ -207,7 +274,7 @@ async function decodeZstdJsonl(path: string): Promise { } describe('JSONL immutable generation publication', () => { - it('does not return until verification and publication complete', async () => { + it('returns migrated events while publication is still waiting for verification', async () => { const root = await tempRoot() const request = options(root, 'none', streamingAdapter()) const boundaryBase = { ...event0, data: { turn: 1, text: '' } } @@ -223,7 +290,7 @@ describe('JSONL immutable generation publication', () => { const entered = Promise.withResolvers() const release = Promise.withResolvers() - const migration = ensureJsonlGenerationCurrent({ + const prepared = await prepareJsonlMigration({ ...request, verifyCurrentFile: async (path, compression, expectedId, expectedEventCount) => { entered.resolve(undefined) @@ -231,11 +298,14 @@ describe('JSONL immutable generation publication', () => { return verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount) }, }) + expect(prepared.artifact.events).toEqual([boundaryEvent, largeEvent, finalEvent]) + const publication = prepared.publish() + expect(prepared.publish()).toBe(publication) await entered.promise await expect(readFile(request.currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) release.resolve(undefined) - await migration + await publication const [writtenHeader, ...writtenEvents] = (await readFile(request.currentPath, 'utf8')).trimEnd().split('\n') expect(JSON.parse(writtenHeader as string)).toEqual({ ...header(2), isSeeded: false }) expect(writtenEvents.map(row => JSON.parse(row) as unknown)).toEqual([boundaryEvent, largeEvent, finalEvent]) @@ -252,15 +322,16 @@ describe('JSONL immutable generation publication', () => { const events = widths.map((_, seq) => ({ ...event0, seq })) await writeFile(request.sourcePath, line(header(0)) + events.map(line).join('')) - await ensureJsonlGenerationCurrent({ + const prepared = await prepareJsonlMigration({ ...request, verifyCurrentFile: byteVerifier, }) + await prepared.publish() expect((await stat(request.currentPath)).size).toBeGreaterThan(8 * mib) }) - it('retries migration when the source changes before publication', async () => { + it('fails publication without rerunning migration when the source changes', async () => { const root = await tempRoot() const base = streamingAdapter() const sourceStreams = vi.fn() @@ -274,21 +345,18 @@ describe('JSONL immutable generation publication', () => { const source = line(header(0)) + line(event0) await writeFile(request.sourcePath, source) - let verifications = 0 - await ensureJsonlGenerationCurrent({ + const prepared = await prepareJsonlMigration({ ...request, verifyCurrentFile: async (path, compression, expectedId, expectedEventCount) => { const verified = await verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount) - if (++verifications === 1) await writeFile(request.sourcePath, source + line(event1)) + await writeFile(request.sourcePath, source + line(event1)) return verified }, }) - expect(sourceStreams).toHaveBeenCalledTimes(2) - expect(verifications).toBe(3) - expect(await readFile(request.currentPath, 'utf8')).toBe( - line(header(2)) + line(event0) + line(event1), - ) + await expect(prepared.publish()).rejects.toBeInstanceOf(JsonlGenerationSourceChangedError) + expect(sourceStreams).toHaveBeenCalledOnce() + await expect(readFile(request.currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) }) it('refuses malformed streaming inputs before publication', async () => { @@ -296,23 +364,24 @@ describe('JSONL immutable generation publication', () => { const request = options(root, 'none', streamingAdapter()) await writeFile(request.sourcePath, '') - await expect(ensureJsonlGenerationCurrent({ ...request, verifyCurrentFile: vi.fn() })) + await expect(prepareJsonlMigration({ ...request, verifyCurrentFile: vi.fn() })) .rejects.toThrow('empty or header-less') await writeFile(request.sourcePath, line(header(1))) - await expect(ensureJsonlGenerationCurrent({ ...request, verifyCurrentFile: vi.fn() })) + await expect(prepareJsonlMigration({ ...request, verifyCurrentFile: vi.fn() })) .rejects.toThrow(/filename identifies v0.*header identifies v1/) await writeFile(request.sourcePath, line(header(0)) + '{bad json}\n' + line(event1)) - await expect(ensureJsonlGenerationCurrent({ ...request, verifyCurrentFile: vi.fn() })) + await expect(prepareJsonlMigration({ ...request, verifyCurrentFile: vi.fn() })) .rejects.toThrow('row 1 is not valid JSON') await writeFile(request.sourcePath, line(header(0)) + '{bad json}\n' + line(event0)) - await ensureJsonlGenerationCurrent({ + const dropped = await prepareJsonlMigration({ ...request, verifyCurrentFile: verifier(), }) - expect((await readFile(request.currentPath, 'utf8')).trimEnd().split('\n')).toHaveLength(1) + expect(dropped.artifact.events).toEqual([]) + await dropped.publish() }) @@ -346,11 +415,52 @@ describe('JSONL immutable generation publication', () => { .rejects.toThrow('torn physical tail') }) + it('keeps complete Assistant stream checks in current-generation verification', async () => { + const root = await tempRoot() + const path = generationPath(root, 2, 'none') + const verify = async (events: readonly SessionFormatEvent[]) => { + await writeFile(path, line(header(2)) + events.map(line).join('')) + return verifyJsonlCurrentGeneration(path, 'none', 'generation-test', events.length) + } + + const valid = [ + assistantLifecycle('assistant/message', assistantData()), + assistantLifecycle('assistant/message', assistantData({ + interrupted: true, + stream: [{ type: 'text-chunks', time0: 3, index: 0, dt: [], texts: ['hello'] }], + usage: null, + replayState: null, + })), + assistantLifecycle('assistant/message', assistantData({ + content: [], stream: [], usage: null, replayState: null, + })), + assistantLifecycle('assistant/attempt', { + turn: 1, + step: 1, + stream: [{ type: 'text-chunks', time0: 3, index: 0, dt: [1], texts: ['a', 'b'] }], + }), + ] + for (const events of valid) expect((await verify(events)).bytes).toBeGreaterThan(0) + + await expect(verify(assistantLifecycle('assistant/attempt', { + turn: 1, step: 1, stream: [{ type: 'future' }], + }))).rejects.toThrow(/invalid embedded stream/) + await expect(verify(assistantLifecycle('assistant/message', assistantData({ + content: [{ type: 'text', text: 'different' }], + })))).rejects.toThrow(/content disagrees/) + await expect(verify(assistantLifecycle('assistant/message', assistantData({ + usage: { inputTokens: 9, outputTokens: 2 }, + })))).rejects.toThrow(/usage disagrees/) + await expect(verify(assistantLifecycle('assistant/message', assistantData({ + replayState: { response: { id: 'different' } }, + })))).rejects.toThrow(/replay state disagrees/) + }) + it('accepts an identical publication winner and rejects different bytes', async () => { const identicalRoot = await tempRoot() const identical = options(identicalRoot, 'none', streamingAdapter()) await writeFile(identical.sourcePath, line(header(0)) + line(event0)) - await ensureJsonlGenerationCurrent({ + const prepared = await prepareJsonlMigration({ ...identical, verifyCurrentFile: async (path, compression, expectedId, expectedEventCount) => { const verified = await verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount) @@ -358,16 +468,17 @@ describe('JSONL immutable generation publication', () => { return verified }, }) - expect((await stat(identical.currentPath, { bigint: true })).size).toBeGreaterThan(0n) + expect((await prepared.publish()).size).toBeGreaterThan(0n) const differentRoot = await tempRoot() const different = options(differentRoot, 'none', streamingAdapter()) await writeFile(different.sourcePath, line(header(0)) + line(event0)) await writeFile(different.currentPath, line({ ...header(2), isSeeded: false }) + line({ ...event0, time: 99 })) - await expect(ensureJsonlGenerationCurrent({ + const conflicted = await prepareJsonlMigration({ ...different, verifyCurrentFile: verifier(), - })).rejects.toBeInstanceOf(JsonlGenerationTargetConflictError) + }) + await expect(conflicted.publish()).rejects.toBeInstanceOf(JsonlGenerationTargetConflictError) const uncheckedRoot = await tempRoot() const unchecked = options(uncheckedRoot, 'none', streamingAdapter()) @@ -376,17 +487,19 @@ describe('JSONL immutable generation publication', () => { unchecked.currentPath, line({ ...header(2), isSeeded: false }) + line({ ...event0, time: 99 }), ) - await expect(ensureJsonlGenerationCurrent({ + const uncheckedPublication = await prepareJsonlMigration({ ...unchecked, verifyCurrentFile: byteVerifier, - })).rejects.toThrow(/target bytes differ from the migrated generation/) + }) + await expect(uncheckedPublication.publish()) + .rejects.toThrow(/target bytes differ from the migrated generation/) }) it('handles empty, incomplete-record, and torn Zstandard migration sources', async () => { const emptyRoot = await tempRoot() const empty = options(emptyRoot, 'zstd', streamingAdapter()) await writeFile(empty.sourcePath, Buffer.alloc(0)) - await expect(ensureJsonlGenerationCurrent({ ...empty, verifyCurrentFile: vi.fn() })) + await expect(prepareJsonlMigration({ ...empty, verifyCurrentFile: vi.fn() })) .rejects.toThrow('empty or header-less Zstandard') const incompleteRoot = await tempRoot() @@ -395,7 +508,7 @@ describe('JSONL immutable generation publication', () => { await compressZstdFrame(line(header(0))), await compressZstdFrame(JSON.stringify(event0)), ])) - await expect(ensureJsonlGenerationCurrent({ ...incomplete, verifyCurrentFile: vi.fn() })) + await expect(prepareJsonlMigration({ ...incomplete, verifyCurrentFile: vi.fn() })) .rejects.toThrow('complete frame contains a torn JSONL record') const tornRoot = await tempRoot() @@ -405,11 +518,12 @@ describe('JSONL immutable generation publication', () => { await compressZstdFrame(line(header(0))), tornBody.subarray(0, -3), ])) - await ensureJsonlGenerationCurrent({ + const recovered = await prepareJsonlMigration({ ...torn, verifyCurrentFile: verifier(), }) - expect((await decodeZstdJsonl(torn.currentPath)).trimEnd().split('\n')).toHaveLength(3) + expect(recovered.artifact.events).toEqual([event0, event1]) + await recovered.publish() const emptyTailRoot = await tempRoot() const emptyTail = options(emptyTailRoot, 'zstd', streamingAdapter()) @@ -417,11 +531,12 @@ describe('JSONL immutable generation publication', () => { await compressZstdFrame(line(header(0))), tornBody.subarray(0, 8), ])) - await ensureJsonlGenerationCurrent({ + const withoutTail = await prepareJsonlMigration({ ...emptyTail, verifyCurrentFile: verifier(), }) - expect((await decodeZstdJsonl(emptyTail.currentPath)).trimEnd().split('\n')).toHaveLength(1) + expect(withoutTail.artifact.events).toEqual([]) + await withoutTail.publish() }) it('checks migration and verification identities exactly', async () => { @@ -441,13 +556,18 @@ describe('JSONL immutable generation publication', () => { const mismatchRoot = await tempRoot() const mismatch = options(mismatchRoot, 'none', streamingAdapter()) await writeFile(mismatch.sourcePath, line(header(0))) - await expect(ensureJsonlGenerationCurrent({ + const mismatched = await prepareJsonlMigration({ ...mismatch, verifyCurrentFile: async (path, compression, expectedId, expectedEventCount) => ({ ...await verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount), digest: 'different', }), - })).rejects.toThrow('changed during verification') + }) + await expect(mismatched.publish()).rejects.toThrow('changed during verification') + + const current = options(await tempRoot(), 'none', streamingAdapter(), 2) + await expect(prepareJsonlMigration({ ...current, verifyCurrentFile: vi.fn() })) + .rejects.toThrow('requires a historical source') const wrongRoot = await tempRoot() const wrongFormat = streamingAdapter() @@ -464,7 +584,7 @@ describe('JSONL immutable generation publication', () => { }), }) await writeFile(wrong.sourcePath, line(header(0))) - await expect(ensureJsonlGenerationCurrent({ ...wrong, verifyCurrentFile: vi.fn() })) + await expect(prepareJsonlMigration({ ...wrong, verifyCurrentFile: vi.fn() })) .rejects.toThrow('migration returned v0') }) @@ -478,12 +598,13 @@ describe('JSONL immutable generation publication', () => { sync: async () => {}, close: async () => { throw new Error('close failed') }, } as unknown as FileHandle - await expect(createJsonlGenerationTestRuntime({ + const failedPreparation = await createJsonlGenerationTestRuntime({ fs: { open: async () => failedHandle }, - }).ensure({ + }).prepare({ ...failed, verifyCurrentFile: vi.fn(), - })).rejects.toBeInstanceOf(AggregateError) + }) + await expect(failedPreparation.publish()).rejects.toBeInstanceOf(AggregateError) }) it('propagates a streamed encoder failure through the Zstandard pipeline', async () => { @@ -494,77 +615,26 @@ describe('JSONL immutable generation publication', () => { })) await writeFile(request.sourcePath, await encodeZstd(0, [event0])) - await expect(ensureJsonlGenerationCurrent({ + const prepared = await prepareJsonlMigration({ ...request, verifyCurrentFile: vi.fn(), - })).rejects.toBe(failure) + }) + await expect(prepared.publish()).rejects.toBe(failure) expect(await readdir(root)).toEqual(['session.jsonl.zstd']) }) - it('observes cancellation at the existing encode yield boundary', async () => { - const root = await tempRoot() - const controller = new AbortController() - const reason = new Error('cancelled during encoding') - const request = { ...options(root), signal: controller.signal } - const payload = 'x'.repeat(600 * 1024) - await writeFile(request.sourcePath, line(header(0)) + line({ - ...event0, data: { turn: 1, payload }, - }) + line({ - ...event1, data: { turn: 1, reason: { kind: 'completed' }, payload }, - })) - vi.spyOn(performance, 'now').mockReturnValue(0) - let yields = 0 - vi.spyOn(scheduler, 'yield').mockImplementation(async () => { - yields += 1 - if (yields === 2) controller.abort(reason) - }) - - await expect(ensureJsonlGenerationCurrent({ - ...request, - verifyCurrentFile: vi.fn(), - })).rejects.toBe(reason) - expect(yields).toBe(2) - expect(await readdir(root)).toEqual(['session.jsonl']) - }) - - it('forwards cancellation to staged verification', async () => { - const root = await tempRoot() - const controller = new AbortController() - const reason = new Error('cancelled during verification') - const request = { ...options(root), signal: controller.signal } - await writeFile(request.sourcePath, line(header(0)) + line(event0)) - const verifyCurrentFile: EnsureJsonlGenerationOptions['verifyCurrentFile'] = async ( - _path, - _compression, - _expectedId, - _expectedEventCount, - _expectedPrefix, - signal, - ) => { - expect(signal).toBe(controller.signal) - controller.abort(reason) - signal?.throwIfAborted() - throw new Error('unreachable') - } - - await expect(ensureJsonlGenerationCurrent({ - ...request, - verifyCurrentFile, - })).rejects.toBe(reason) - expect(await readdir(root)).toEqual(['session.jsonl']) - }) - - it('publishes through the Windows no-overwrite path', async () => { + it('publishes a prepared stage through the Windows no-overwrite path', async () => { const winRoot = await tempRoot() const win = options(winRoot, 'none', streamingAdapter()) await writeFile(win.sourcePath, line(header(0))) - await createJsonlGenerationTestRuntime({ + const winPrepared = await createJsonlGenerationTestRuntime({ platform: 'win32', publishNewWin32: rename, - }).ensure({ + }).prepare({ ...win, verifyCurrentFile: verifier(), }) + await winPrepared.publish() expect(await readFile(win.currentPath, 'utf8')).toContain('"version":2') }) @@ -588,56 +658,6 @@ describe('JSONL immutable generation publication', () => { expect((await readdir(root)).sort()).toEqual(['session.jsonl', 'session.v2.jsonl']) }) - it('takes the current fast path with one read and no format callback', async () => { - const root = await tempRoot() - const base = adapter() - const createRestore = vi.fn((value: Record) => base.createRestore(value)) - const encodeHeader = vi.fn((value: SessionFormatArtifact['header'], cut: number) => - base.encodeHeader(value, cut)) - const encodeEvent = vi.fn((value: SessionFormatArtifact['events'][number]) => base.encodeEvent(value)) - const validateHistoricalHeader = vi.fn() - const request = { - ...options(root, 'none', { ...base, createRestore, encodeHeader, encodeEvent }, 2), - validateHistoricalHeader, - } - const contents = line({ ...header(2), isSeeded: false }) + line(event0) - await writeFile(request.sourcePath, contents) - const readStableFile = vi.fn(async (path: string, signal?: AbortSignal) => - readFile(path, signal === undefined ? undefined : { signal })) - - const result = await ensureWithOverrides(request, { fs: { readFile: readStableFile } }) - - expect(result).toMatchObject({ status: 'current', version: 2, path: request.sourcePath }) - expect(readStableFile).toHaveBeenCalledOnce() - expect(createRestore).not.toHaveBeenCalled() - expect(encodeHeader).not.toHaveBeenCalled() - expect(encodeEvent).not.toHaveBeenCalled() - expect(validateHistoricalHeader).not.toHaveBeenCalled() - expect(await readFile(request.sourcePath, 'utf8')).toBe(contents) - }) - - it('bounds current snapshot retries under continuous revision churn', async () => { - const root = await tempRoot() - const request = options(root, 'none', adapter(), 2) - const contents = line(header(2)) + line(event0) - await writeFile(request.sourcePath, contents) - let revision = 0n - const statFile = vi.fn(async (path: string) => { - const value = await stat(path, { bigint: true }) - revision += 1n - return { ...value, mtimeNs: value.mtimeNs + revision } - }) - const readChangingFile = vi.fn(async () => Buffer.from(contents + line(event1))) - - const result = await ensureWithOverrides(request, { - fs: { stat: statFile, readFile: readChangingFile }, - }) - - expect(result.snapshot.bytes.toString('utf8')).toBe(contents) - expect(readChangingFile).toHaveBeenCalledTimes(2) - expect(statFile).toHaveBeenCalledTimes(3) - }) - it.each(['none', 'zstd'] as const)( 'validates the selected %s historical header before invoking migration', async (compression) => { @@ -700,24 +720,15 @@ describe('JSONL immutable generation publication', () => { expect(await readdir(root)).toEqual(['session.jsonl']) }) - it('rejects malformed and future version discriminators before migration', async () => { + it('rejects a malformed version discriminator before migration', async () => { const root = await tempRoot() const malformed = options(join(root, 'malformed')) - const future = options(join(root, 'future'), 'none', adapter(), 3) await mkdir(join(root, 'malformed')) - await mkdir(join(root, 'future')) await writeFile(malformed.sourcePath, line(header(-1))) - await writeFile(future.sourcePath, line(header(3, 'future-id'))) await expect(ensureJsonlGenerationCurrent(malformed)).rejects.toThrow( 'header version is not a non-negative safe integer', ) - await expect(ensureJsonlGenerationCurrent(future)).rejects.toMatchObject({ - name: 'JsonlGenerationNewerVersionError', - storedVersion: 3, - currentVersion: 2, - storedId: 'future-id', - }) }) it.each([ @@ -976,7 +987,7 @@ describe('JSONL immutable generation publication', () => { } }) - it('retries a bracketed physical read and a source changed before publication', async () => { + it('bounds a bracketed physical read and does not rerun migration after a publication race', async () => { const root = await tempRoot() const request = options(root) const first = Buffer.from(line(header(0)) + line(event0)) @@ -995,17 +1006,15 @@ describe('JSONL immutable generation publication', () => { if (phase === 'before-source-check' && attempt === 1) await writeFile(request.sourcePath, second) }) - await ensureWithOverrides( + await expect(ensureWithOverrides( { ...request, format: { ...base, createRestore } }, { fs: { stat: statFile }, barrier }, - ) + )).rejects.toBeInstanceOf(JsonlGenerationSourceChangedError) expect(stats).toBeGreaterThan(2) - expect(createRestore).toHaveBeenCalledTimes(2) + expect(createRestore).toHaveBeenCalledOnce() expect(await readFile(request.sourcePath)).toEqual(second) - expect(await readFile(request.currentPath, 'utf8')).toBe( - line(header(2)) + line(event0) + line(event1), - ) + await expect(readFile(request.currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) expect((await readdir(root)).every(name => !name.includes('.tmp'))).toBe(true) }) @@ -1020,7 +1029,7 @@ describe('JSONL immutable generation publication', () => { if (phase === 'before-source-check' && attempt === 1) await writeFile(request.sourcePath, second) } - await expect(ensureWithOverrides(request, { + const failure = await ensureWithOverrides(request, { barrier, fs: { rm: async (path: string) => { @@ -1028,7 +1037,10 @@ describe('JSONL immutable generation publication', () => { await rm(path, { force: true }) }, }, - })).rejects.toBe(cleanup) + }).then(() => undefined, (error: unknown) => error) + if (!(failure instanceof AggregateError)) throw new Error('expected source and cleanup failures') + expect(failure.errors[0]).toBeInstanceOf(JsonlGenerationSourceChangedError) + expect(failure.errors[1]).toBe(cleanup) expect(await readFile(request.sourcePath)).toEqual(second) await expect(readFile(request.currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) }) @@ -1133,7 +1145,7 @@ describe('JSONL immutable generation publication', () => { const format = adapter() const request = { ...options(root, 'none', format), - verifyCurrentFile: async (...args: Parameters) => { + verifyCurrentFile: async (...args: Parameters) => { validations += 1 if (validations === 2) throw 'non-error rejection' return verifier()(...args) @@ -1175,7 +1187,26 @@ describe('JSONL immutable generation publication', () => { expect(await readFile(request.currentPath, 'utf8')).toBe(line(header(2)) + line(event0)) }) - it('reports cancellation during committed reopen and leaves the target', async () => { + it('retains a committed generation when its post-publication stat fails', async () => { + const root = await tempRoot() + const request = options(root) + const statFailure = new Error('published target stat failed') + await writeFile(request.sourcePath, line(header(0)) + line(event0)) + let targetStats = 0 + + await expect(ensureWithOverrides(request, { + fs: { + stat: async (path) => { + if (path === request.currentPath && ++targetStats === 1) throw statFailure + return stat(path, { bigint: true }) + }, + }, + })).rejects.toBe(statFailure) + expect(await readFile(request.currentPath, 'utf8')).toBe(line(header(2)) + line(event0)) + expect((await readdir(root)).every(name => !name.includes('.tmp'))).toBe(true) + }) + + it('finishes a committed publication despite later caller cancellation', async () => { const root = await tempRoot() const controller = new AbortController() const reason = new Error('stop after publication') @@ -1186,7 +1217,7 @@ describe('JSONL immutable generation publication', () => { barrier: (phase) => { if (phase === 'after-publication') controller.abort(reason) }, - })).rejects.toBe(reason) + })).resolves.toMatchObject({ status: 'migrated', path: request.currentPath }) expect(await readFile(request.currentPath, 'utf8')).toBe(line(header(2)) + line(event0)) }) @@ -1219,7 +1250,7 @@ describe('JSONL immutable generation publication', () => { })).rejects.toMatchObject({ code: 'ENOENT', path: request.currentPath }) }) - it('reopens a target after exclusive publication', async () => { + it('does not reopen a target after exclusive publication', async () => { const root = await tempRoot() const request = options(root) await writeFile(request.sourcePath, line(header(0)) + line(event0)) @@ -1233,7 +1264,7 @@ describe('JSONL immutable generation publication', () => { }, }, })).resolves.toMatchObject({ status: 'migrated', path: request.currentPath }) - expect(reads).toContain(request.currentPath) + expect(reads).not.toContain(request.currentPath) expect(await readFile(request.currentPath, 'utf8')).toBe(line(header(2)) + line(event0)) }) diff --git a/packages/session/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session/session-persistence-jsonl/tests/jsonl.spec.ts index 1117430ff2..544935a76b 100644 --- a/packages/session/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session/session-persistence-jsonl/tests/jsonl.spec.ts @@ -4,6 +4,7 @@ import { Context } from '@deepseek-ai/cordis' import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat, symlink } from 'node:fs/promises' import { tmpdir } from 'node:os' import { dirname, join, relative, resolve } from 'node:path' +import { scheduler } from 'node:timers/promises' import { SESSION_FORMAT_VERSION, SessionLogOffset, SessionSeq, SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' @@ -18,6 +19,7 @@ import { } from '../../session-persistence/tests/contract.ts' import { runLiveWritePathContract } from '../../session-persistence/tests/live-write-contract.ts' import { LIVE_WRITE_BATCH_MAX_DELAY_MS, type JsonlSessionHandle } from '../src/storage.ts' +import { JsonlGenerationSourceChangedError } from '../src/generation.ts' import SessionStore from '@deepseek-ai/dsh-session' const statRace = vi.hoisted(() => ({ @@ -43,6 +45,21 @@ const readTally = vi.hoisted(() => ({ enabled: false, })) +const readFailure = vi.hoisted(() => ({ + path: undefined as string | undefined, + error: undefined as Error | undefined, +})) + +const pausedRead = vi.hoisted(() => ({ + path: undefined as string | undefined, + active: false, + entered: undefined as (() => void) | undefined, + resume: undefined as Promise | undefined, + release: undefined as (() => void) | undefined, + done: undefined as Promise | undefined, + finished: undefined as (() => void) | undefined, +})) + vi.mock('node:fs/promises', async (importOriginal) => { const actual = await importOriginal() return { @@ -57,10 +74,25 @@ vi.mock('node:fs/promises', async (importOriginal) => { return { ...identity, mtimeNs: identity.mtimeNs + 1n } }) as typeof actual.stat, readFile: (async (...args: Parameters) => { - if (readTally.enabled && typeof args[0] === 'string') { - readTally.bySuffix.set(args[0], (readTally.bySuffix.get(args[0]) ?? 0) + 1) + const path = typeof args[0] === 'string' ? args[0] : undefined + if (path === readFailure.path && readFailure.error !== undefined) throw readFailure.error + if (readTally.enabled && path !== undefined) { + readTally.bySuffix.set(path, (readTally.bySuffix.get(path) ?? 0) + 1) + } + if (path !== pausedRead.path || pausedRead.resume === undefined) { + return actual.readFile(...args) + } + const resume = pausedRead.resume + const finished = pausedRead.finished + pausedRead.active = true + pausedRead.entered?.() + await resume + try { + return await actual.readFile(...args) + } finally { + pausedRead.active = false + finished?.() } - return actual.readFile(...args) }) as typeof actual.readFile, readdir: (async (...args: Parameters) => { if (String(args[0]) === readdirFailure.path && readdirFailure.error !== undefined) { @@ -107,6 +139,27 @@ async function freshRoot(): Promise { return dir } +function pausePhysicalRead(path: string): { + readonly entered: Promise + readonly finished: Promise + release(): void +} { + const entered = Promise.withResolvers() + const resume = Promise.withResolvers() + const finished = Promise.withResolvers() + pausedRead.path = path + pausedRead.entered = () => { entered.resolve(undefined) } + pausedRead.resume = resume.promise + pausedRead.release = () => { resume.resolve(undefined) } + pausedRead.done = finished.promise + pausedRead.finished = () => { finished.resolve(undefined) } + return { + entered: entered.promise, + finished: finished.promise, + release: () => { resume.resolve(undefined) }, + } +} + function rawLogPath(root: string, cwd: string | undefined, id: SessionId): string { return logPath(root, cwd, id, 'none') } @@ -178,7 +231,7 @@ async function writeLog(persistence: SessionPersistence, m: SessionHeader, event async function readAll(persistence: SessionPersistence, id: SessionId): Promise<{ meta: SessionHeader; events: readonly SessionEvent[] }> { const handle = await persistence.open(id, 'read') try { - return { meta: handle.header, events: await handle.read() } + return { meta: handle.header, events: (await handle.read()).events } } finally { await handle.close() } @@ -200,6 +253,18 @@ afterEach(async () => { statRace.mode = 'settle' readTally.bySuffix.clear() readTally.enabled = false + const pausedReadDone = pausedRead.active ? pausedRead.done : undefined + readFailure.path = undefined + readFailure.error = undefined + pausedRead.release?.() + await pausedReadDone + pausedRead.path = undefined + pausedRead.active = false + pausedRead.entered = undefined + pausedRead.resume = undefined + pausedRead.release = undefined + pausedRead.done = undefined + pausedRead.finished = undefined statFailure.path = undefined statFailure.error = undefined readdirFailure.path = undefined @@ -451,6 +516,17 @@ describe('JsonlSessionPersistence: stored-format refusals', () => { expect(await ctx.sessionPersistence.list()).toEqual([]) }) + it('classifies an unparsable future generation header as corruption', async () => { + const id = SessionId('future-malformed') + const path = generationLogPath(root, '/work', id, 42, 'none') + await mkdir(dirname(path), { recursive: true }) + await writeFile(path, '{not-json}\n') + + await expect(ctx.sessionPersistence.open(id, 'read')).rejects.toMatchObject({ + name: 'SessionPersistenceCorruptionError', + }) + }) + it('refuses a well-shaped newer-version header at read open with the upgrade direction', async () => { // A header that satisfies the current shape but carries a future version: // stat can parse it, and the open still refuses before handing out a @@ -524,7 +600,14 @@ describe('JsonlSessionPersistence: stored-format refusals', () => { const handle = await ctx.sessionPersistence.open(m.id, 'read', { signal: new AbortController().signal }) try { expect(handle.header).toMatchObject({ id: m.id, cwd: '/work' }) - expect(await handle.read()).toEqual(oneTurnLog()) + const read = await handle.read() + expect(read.eventState).toBe('shared-frozen') + expect(read.events).toEqual(oneTurnLog()) + expect(read.events.every(event => Object.isFrozen(event) && Object.isFrozen(event.data))).toBe(true) + const reread = await handle.read() + expect(reread.events).not.toBe(read.events) + expect(reread.events[0]).toBe(read.events[0]) + expect((await handle.read(read.events.length)).eventState).toBe('shared-frozen') } finally { await handle.close() } @@ -597,7 +680,7 @@ describe('JsonlSessionPersistence: immutable format generations', () => { await expect(stat(currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) }) - it('publishes v2 beside an unchanged v0 source before returning a read handle', async () => { + it('serves a migrated v0 read without publishing at the service durability barrier', async () => { const header = meta('released-v0-read', '/work') const sourcePath = historicalLogPath(root, header.cwd, header.id) const currentPath = rawLogPath(root, header.cwd, header.id) @@ -607,18 +690,147 @@ describe('JsonlSessionPersistence: immutable format generations', () => { await mkdir(dirname(sourcePath), { recursive: true }) await writeFile(sourcePath, source) - await expect(readAll(ctx.sessionPersistence, header.id)).resolves.toEqual({ + const restored = await readAll(ctx.sessionPersistence, header.id) + expect(restored).toEqual({ meta: { ...header, delegationDepth: 0 }, events: oneTurnLog(), }) + const userMessage = restored.events.find(event => event.type === 'user/message') + expect(userMessage).toBeDefined() + expect(Object.isFrozen(userMessage?.data)).toBe(true) expect(await readFile(sourcePath)).toEqual(source) - const current = (await readFile(currentPath, 'utf8')).trimEnd().split('\n') - expect(JSON.parse(current[0] as string)).toMatchObject({ - id: header.id, - version: SESSION_FORMAT_VERSION, - }) + await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) expect((await readdir(dirname(sourcePath))).filter(name => name.startsWith('session')).sort()) - .toEqual(['session.jsonl', 'session.v2.jsonl']) + .toEqual(['session.jsonl']) + }) + + it('resolves absent, current, and historical current-generation paths', async () => { + const persistence = ctx.sessionPersistence as JsonlSessionPersistence + expect(await persistence.resolveCurrentLog(SessionId('missing-generation'))).toBeUndefined() + + const current = meta('resolved-current', '/work') + const currentHandle = await ctx.sessionPersistence.create(current) + await currentHandle.flush() + await currentHandle.close() + await expect(persistence.resolveCurrentLog(current.id)).resolves.toBe(rawLogPath(root, current.cwd, current.id)) + + const historical = meta('resolved-historical', '/work') + const sourcePath = historicalLogPath(root, historical.cwd, historical.id) + await mkdir(dirname(sourcePath), { recursive: true }) + await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(historical))}\n`) + await expect(persistence.resolveCurrentLog(historical.id)).resolves.toBeUndefined() + + const future = meta('resolved-future', '/work') + const futurePath = generationLogPath(root, future.cwd, future.id, SESSION_FORMAT_VERSION + 1, 'none') + await mkdir(dirname(futurePath), { recursive: true }) + await writeFile(futurePath, `${JSON.stringify({ ...toHeaderLine(future), version: SESSION_FORMAT_VERSION + 1 })}\n`) + await expect(persistence.resolveCurrentLog(future.id)).rejects.toMatchObject({ + name: 'SessionFormatUnsupportedError', + }) + }) + + it('singleflights concurrent historical reads and keeps service flush read-only', async () => { + const header = meta('released-v0-source-drift', '/work') + const sourcePath = historicalLogPath(root, header.cwd, header.id) + await mkdir(dirname(sourcePath), { recursive: true }) + await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`) + readTally.enabled = true + + const [first, second] = await Promise.all([ + ctx.sessionPersistence.open(header.id, 'read'), + ctx.sessionPersistence.open(header.id, 'read'), + ]) + expect((await first.read()).events).toEqual([]) + expect((await second.read()).events).toEqual([]) + expect(readTally.bySuffix.get(sourcePath)).toBe(1) + await appendFile(sourcePath, '\n') + + await expect(ctx.sessionPersistence.flush()).resolves.toBeUndefined() + await expect(stat(rawLogPath(root, header.cwd, header.id))).rejects.toMatchObject({ code: 'ENOENT' }) + await Promise.all([first.close(), second.close()]) + await ctx.fiber.dispose() + ctx = new Context() + }) + + it('does not join an in-flight historical preparation for an older source revision', async () => { + const header = meta('released-v0-revision-singleflight', '/work') + const sourcePath = historicalLogPath(root, header.cwd, header.id) + await mkdir(dirname(sourcePath), { recursive: true }) + await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`) + const pause = pausePhysicalRead(sourcePath) + readTally.enabled = true + + const firstOpening = ctx.sessionPersistence.open(header.id, 'read') + await pause.entered + await appendFile(sourcePath, `${eventLines(releasedV1OneTurnLog())}\n`) + const secondOpening = ctx.sessionPersistence.open(header.id, 'read') + let tallyFailure: unknown + try { + await vi.waitFor(() => { expect(readTally.bySuffix.get(sourcePath)).toBe(2) }) + } catch (error: unknown) { + tallyFailure = error + } finally { + pause.release() + } + + const [first, second] = await Promise.all([firstOpening, secondOpening]) + try { + if (tallyFailure !== undefined) throw tallyFailure + expect((await first.read()).events).toEqual(oneTurnLog()) + expect((await second.read()).events).toEqual(oneTurnLog()) + } finally { + await Promise.all([first.close(), second.close()]) + } + }) + + it('lets one historical-open caller abort without cancelling another waiter', async () => { + const header = meta('released-v0-shared-cancellation', '/work') + const sourcePath = historicalLogPath(root, header.cwd, header.id) + await mkdir(dirname(sourcePath), { recursive: true }) + await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`) + const pause = pausePhysicalRead(sourcePath) + readTally.enabled = true + const controller = new AbortController() + const reason = new Error('first historical waiter cancelled') + + const first = ctx.sessionPersistence.open(header.id, 'read', { signal: controller.signal }) + const second = ctx.sessionPersistence.open(header.id, 'read') + await pause.entered + await scheduler.yield() + controller.abort(reason) + await expect(first).rejects.toBe(reason) + pause.release() + const handle = await second + expect((await handle.read()).events).toEqual([]) + expect(readTally.bySuffix.get(sourcePath)).toBe(1) + await handle.close() + }) + + it('cancels shared historical preparation after its last waiter leaves', async () => { + const header = meta('released-v0-last-waiter-cancellation', '/work') + const sourcePath = historicalLogPath(root, header.cwd, header.id) + await mkdir(dirname(sourcePath), { recursive: true }) + await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`) + const pause = pausePhysicalRead(sourcePath) + readTally.enabled = true + const controller = new AbortController() + const reason = 'last historical waiter cancelled' + + const opening = ctx.sessionPersistence.open(header.id, 'read', { signal: controller.signal }) + await pause.entered + controller.abort(reason) + await expect(opening).rejects.toMatchObject({ + message: 'session migration preparation aborted', + cause: reason, + }) + pause.release() + await pause.finished + await scheduler.yield() + + const retried = await ctx.sessionPersistence.open(header.id, 'read') + expect((await retried.read()).events).toEqual([]) + expect(readTally.bySuffix.get(sourcePath)).toBe(2) + await retried.close() }) it('migrates released-v0 retry, repeated-compaction, provenance, and late-title shapes', async () => { @@ -648,16 +860,15 @@ describe('JsonlSessionPersistence: immutable format generations', () => { expect(titleBlock).toMatchObject({ type: 'text' }) if (titleBlock?.type !== 'text') throw new Error('fixture title request lacks its text block') expect(titleBlock.text).toContain('{"seq":21,"text":"late"}') - const currentRows = (await readFile(currentPath, 'utf8')).trimEnd().split('\n') - .map(line => JSON.parse(line) as Record) - expect(currentRows.find(row => row['type'] === 'user/message' - && (row['data'] as { source?: { plugin?: string } }).source?.plugin === 'compact')) + expect(restored.events.find(event => event.type === 'user/message' + && (event.data as { source?: { plugin?: string } }).source?.plugin === 'compact')) .toMatchObject({ seq: 14, sourceEventSeqs: [12, 13, 11, 2, 3, 4], surfaceOp: { op: 'replace', start: 11, end: 4 }, }) expect(await readFile(sourcePath)).toEqual(source) + await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) }) it('publishes v2 beside an unchanged physical v1 source with packed chunk rows', async () => { @@ -677,10 +888,9 @@ describe('JsonlSessionPersistence: immutable format generations', () => { .toMatchObject({ data: { message: { content: [{ type: 'text', text: 'hello' }] } } }) expect(await readFile(sourcePath)).toEqual(source) - expect(JSON.parse((await readFile(currentPath, 'utf8')).split('\n')[0] as string)) - .toMatchObject({ id: header.id, version: SESSION_FORMAT_VERSION }) + await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) expect((await readdir(dirname(sourcePath))).filter(name => name.startsWith('session')).sort()) - .toEqual(['session.v1.jsonl', 'session.v2.jsonl']) + .toEqual(['session.v1.jsonl']) }) it('selects v1 from a v0/v1 directory, then v2 from the retained three-generation set', async () => { @@ -695,8 +905,12 @@ describe('JsonlSessionPersistence: immutable format generations', () => { const migrated = await readAll(ctx.sessionPersistence, header.id) expect(migrated.events.map(event => event.type)).toContain('assistant/message') + const writer = await ctx.sessionPersistence.open(header.id, 'write') + await writer.close() expect((await readdir(directory)).filter(name => name.startsWith('session')).sort()) - .toEqual(['session.jsonl', 'session.v1.jsonl', 'session.v2.jsonl']) + .toEqual(process.platform === 'win32' + ? ['session.jsonl', 'session.v1.jsonl', 'session.v2.jsonl'] + : ['session.jsonl', 'session.lock', 'session.v1.jsonl', 'session.v2.jsonl']) await writeFile(v0Path, 'corrupt lower v0\n') await writeFile(v1Path, 'corrupt lower v1\n') @@ -704,18 +918,17 @@ describe('JsonlSessionPersistence: immutable format generations', () => { expect(await readFile(v2Path, 'utf8')).toContain('"version":2') }) - it('uses the same migration path for a handle storage resolution', async () => { + it('does not publish a historical generation through handle storage resolution', async () => { const header = meta('released-v0-handle-read', '/work') const sourcePath = historicalLogPath(root, header.cwd, header.id) const currentPath = rawLogPath(root, header.cwd, header.id) await mkdir(dirname(sourcePath), { recursive: true }) await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`) - const storage = ctx.sessionPersistence as unknown as { - resolveLog(id: SessionId, signal?: AbortSignal): Promise - } + const persistence = ctx.sessionPersistence as JsonlSessionPersistence - await expect(storage.resolveLog(header.id, new AbortController().signal)).resolves.toBe(currentPath) + await expect(persistence.resolveCurrentLog(header.id, new AbortController().signal)).resolves.toBeUndefined() expect(await readFile(sourcePath, 'utf8')).toBe(`${JSON.stringify(releasedV0Header(header))}\n`) + await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) }) it('opens the migrated successor for append while retaining the historical source', async () => { @@ -740,6 +953,68 @@ describe('JsonlSessionPersistence: immutable format generations', () => { ]) }) + it('switches an existing prepared read handle to the published append tail', async () => { + const header = meta('released-v0-read-handoff', '/work') + const sourcePath = historicalLogPath(root, header.cwd, header.id) + await mkdir(dirname(sourcePath), { recursive: true }) + await writeFile( + sourcePath, + `${JSON.stringify(releasedV0Header(header))}\n${eventLines(releasedV1OneTurnLog())}\n`, + ) + const reader = await ctx.sessionPersistence.open(header.id, 'read') + const suffix: SessionEvent[] = [ + { type: 'turn/start', seq: SessionSeq(6), time: 9, data: { turn: 2 } }, + { type: 'turn/end', seq: SessionSeq(7), time: 10, data: { turn: 2, reason: { kind: 'completed' } } }, + ] + try { + expect((await reader.read()).events).toEqual(oneTurnLog()) + await appendBatch(ctx.sessionPersistence, header.id, suffix) + expect((await reader.read()).events).toEqual([...oneTurnLog(), ...suffix]) + } finally { + await reader.close() + } + }) + + it('fails a stale prepared publication once and re-prepares on the next write open', async () => { + const header = meta('released-v0-write-source-drift', '/work') + const sourcePath = historicalLogPath(root, header.cwd, header.id) + const currentPath = rawLogPath(root, header.cwd, header.id) + const source = `${JSON.stringify(releasedV0Header(header))}\n` + await mkdir(dirname(sourcePath), { recursive: true }) + await writeFile(sourcePath, source) + await readAll(ctx.sessionPersistence, header.id) + vi.spyOn(scheduler, 'yield').mockImplementationOnce(async () => { + await appendFile(sourcePath, '\n') + }) + + await expect(ctx.sessionPersistence.open(header.id, 'write')) + .rejects.toBeInstanceOf(JsonlGenerationSourceChangedError) + await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) + + const writer = await ctx.sessionPersistence.open(header.id, 'write') + await writer.close() + expect(await readFile(sourcePath, 'utf8')).toBe(`${source}\n`) + expect(await readFile(currentPath, 'utf8')).toContain('"version":2') + }) + + it('finishes publication before rejecting a write open cancelled during publication', async () => { + const header = meta('released-v0-publication-cancellation', '/work') + const sourcePath = historicalLogPath(root, header.cwd, header.id) + const currentPath = rawLogPath(root, header.cwd, header.id) + await mkdir(dirname(sourcePath), { recursive: true }) + await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`) + await readAll(ctx.sessionPersistence, header.id) + const controller = new AbortController() + const reason = new Error('write open cancelled during publication') + vi.spyOn(scheduler, 'yield').mockImplementationOnce(async () => { controller.abort(reason) }) + + await expect(ctx.sessionPersistence.open(header.id, 'write', { signal: controller.signal })) + .rejects.toBe(reason) + expect(await readFile(currentPath, 'utf8')).toContain('"version":2') + const writer = await ctx.sessionPersistence.open(header.id, 'write') + await writer.close() + }) + it('treats a historical generation as an existing id at create', async () => { const header = meta('released-v0-collision', '/work') const sourcePath = historicalLogPath(root, header.cwd, header.id) @@ -822,6 +1097,12 @@ describe('JsonlSessionPersistence: immutable format generations', () => { await expect(ctx.sessionPersistence.open(header.id, 'read')).rejects.toMatchObject({ code: 'EACCES' }) statFailure.error = new DOMException('source read aborted', 'AbortError') await expect(ctx.sessionPersistence.open(header.id, 'read')).rejects.toMatchObject({ name: 'AbortError' }) + statFailure.error = undefined + readFailure.path = sourcePath + readFailure.error = new DOMException('source read failed', 'InvalidStateError') + await expect(ctx.sessionPersistence.open(header.id, 'read')).rejects.toMatchObject({ + name: 'SessionPersistenceCorruptionError', + }) }) it('selects the highest opposite-encoding generation for its refusal', async () => { @@ -930,6 +1211,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { it('lazy materialization: create() writes no file until the first append', async () => { const m = meta('lazy', '/work') const handle = await ctx.sessionPersistence.create(m) + expect((await handle.read()).eventState).toBe('detached') // create() materializes no file before the first append — while the // created session is already visible to this process. const dir = sessionDir(root, '/work', m.id) @@ -951,7 +1233,10 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { await handle.close() expect(await readFile(rawLogPath(root, '/work', m.id), 'utf8')).toBe(`${JSON.stringify(toHeaderLine(m))}\n`) - await expect(readAll(ctx.sessionPersistence, m.id)).resolves.toMatchObject({ events: [] }) + const reader = await ctx.sessionPersistence.open(m.id, 'read') + const read = await reader.read() + expect(read).toEqual({ eventState: 'shared-frozen', events: [] }) + await reader.close() }) it('close drains a routed event that arrives while it waits for an in-flight append', async () => { @@ -1062,7 +1347,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { // ...and the immediate write-open (resume) reuses the parsed log through // the revision guard instead of re-reading the file. const writer = await ctx.sessionPersistence.open(m.id, 'write') - expect((await writer.read()).length).toBe(oneTurnLog().length) + expect((await writer.read()).events.length).toBe(oneTurnLog().length) expect(readTally.bySuffix.get(path)).toBe(1) // A local append invalidates the memo: the next cold read re-parses and @@ -1122,7 +1407,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { const internals = ctx.sessionPersistence as unknown as { coldLogMemo: Map } internals.coldLogMemo.clear() statRace.path = rawLogPath(root, '/work', m.id) - expect(await handle.read()).toEqual(oneTurnLog()) + expect((await handle.read()).events).toEqual(oneTurnLog()) // The memo probe, the initial identity, the mismatching post-read stat // (reused as the retry's pre-read identity), and the retry's matching // post-read stat. @@ -1146,7 +1431,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { // serves the retry's pre-read committed prefix — here the whole log. // Four stats: the memo probe, the initial identity, and one mismatching // post-read stat per bounded attempt. - expect(await handle.read()).toEqual(oneTurnLog()) + expect((await handle.read()).events).toEqual(oneTurnLog()) expect(statRace.reads).toBe(4) } finally { statRace.path = undefined @@ -1300,7 +1585,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { // The retry now succeeds with NO seq gap — the log is contiguous 0..7. await handle.append(turn2) - expect((await handle.read()).map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7]) + expect((await handle.read()).events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7]) } finally { await handle.close() } @@ -1412,7 +1697,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { ], { signal })).rejects.toBe(reason) await expect(handle.flush({ signal })).rejects.toBe(reason) // The aborted mutations left the log untouched. - expect((await handle.read()).map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5]) + expect((await handle.read()).events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5]) } finally { await handle.close() } @@ -1473,7 +1758,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { await handle.append([]) await expect(stat(rawLogPath(root, '/work', m.id))).rejects.toThrow() await handle.append(oneTurnLog()) - expect((await handle.read()).map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5]) + expect((await handle.read()).events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5]) await handle.close() }) @@ -1498,7 +1783,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { const m = meta('erased-pending') const creator = await ctx.sessionPersistence.create(m) const reader = await ctx.sessionPersistence.open(m.id, 'read') - expect(await reader.read()).toEqual([]) + expect((await reader.read()).events).toEqual([]) // The creator closes without ever appending: the session never existed. await creator.close() await expect(reader.read()).rejects.toThrow(/not found/) @@ -1510,7 +1795,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => { await writeLog(ctx.sessionPersistence, m, oneTurnLog()) const reader = await ctx.sessionPersistence.open(m.id, 'read') try { - expect(await reader.read()).toHaveLength(6) + expect((await reader.read()).events).toHaveLength(6) // Committed events are never rewritten; a shorter file is damage, not a // legal state, and a handle must not silently backtrack. await writeFile(rawLogPath(root, '/work', m.id), [ diff --git a/packages/session/session-persistence-jsonl/tests/lease.spec.ts b/packages/session/session-persistence-jsonl/tests/lease.spec.ts index b792b937f2..3933c865fd 100644 --- a/packages/session/session-persistence-jsonl/tests/lease.spec.ts +++ b/packages/session/session-persistence-jsonl/tests/lease.spec.ts @@ -181,7 +181,7 @@ describe('cross-process write lock', () => { await pendingWinner.close() // Reads never touch the lock. const reader = await second.open(SessionId('excluded'), 'read') - expect((await reader.read()).map(event => event.seq)).toEqual([0, 1]) + expect((await reader.read()).events.map(event => event.seq)).toEqual([0, 1]) await reader.close() await holder.close() diff --git a/packages/session/session-persistence-jsonl/tests/lease.two-process.e2e.ts b/packages/session/session-persistence-jsonl/tests/lease.two-process.e2e.ts index 9b6bfe8f68..2cf0865787 100644 --- a/packages/session/session-persistence-jsonl/tests/lease.two-process.e2e.ts +++ b/packages/session/session-persistence-jsonl/tests/lease.two-process.e2e.ts @@ -51,7 +51,7 @@ describe('two-process write lock (built lib)', () => { await expect(mine.open(SessionId(SESSION), 'write')).rejects.toBeInstanceOf(SessionAlreadyOwnedError) // Reads are unaffected across processes. const reader = await mine.open(SessionId(SESSION), 'read') - expect((await reader.read()).map(event => event.seq)).toEqual([0, 1]) + expect((await reader.read()).events.map(event => event.seq)).toEqual([0, 1]) await reader.close() // Crash the holder: no release runs, but the kernel drops the lock with @@ -60,7 +60,7 @@ describe('two-process write lock (built lib)', () => { await exited const taken = await mine.open(SessionId(SESSION), 'write') await taken.append([{ type: 'turn/start', seq: SessionSeq(2), time: 3, data: { turn: 2 } }]) - expect((await taken.read()).map(event => event.seq)).toEqual([0, 1, 2]) + expect((await taken.read()).events.map(event => event.seq)).toEqual([0, 1, 2]) await taken.close() } finally { if (holder.exitCode === null) holder.kill('SIGKILL') diff --git a/packages/session/session-persistence-jsonl/tests/migration-verifier.spec.ts b/packages/session/session-persistence-jsonl/tests/migration-verifier.spec.ts index eb3b0089fc..54004522a2 100644 --- a/packages/session/session-persistence-jsonl/tests/migration-verifier.spec.ts +++ b/packages/session/session-persistence-jsonl/tests/migration-verifier.spec.ts @@ -48,10 +48,12 @@ afterEach(() => { describe('migration verifier Worker lifecycle', () => { it('resolves only after terminating a successful Worker', async () => { - const verification = verifyCurrentGenerationInWorker('/stage', 'none', 'session', 2) + const expectedPrefix = { bytes: 3, digest: 'a'.repeat(64) } + const verification = verifyCurrentGenerationInWorker('/stage', 'none', 'session', 2, expectedPrefix) const instance = worker() expect(instance.options.workerData).toEqual({ path: '/stage', compression: 'none', expectedId: 'session', expectedEventCount: 2, + expectedPrefix, }) instance.emit('message', { ok: true, result }) diff --git a/packages/session/session-persistence-jsonl/tests/zstd.spec.ts b/packages/session/session-persistence-jsonl/tests/zstd.spec.ts index b4d58c3630..de12d4dcf0 100644 --- a/packages/session/session-persistence-jsonl/tests/zstd.spec.ts +++ b/packages/session/session-persistence-jsonl/tests/zstd.spec.ts @@ -5,7 +5,7 @@ import type { FileHandle } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { performance } from 'node:perf_hooks' -import { SESSION_FORMAT_VERSION, SessionSeq, SessionId } from '@deepseek-ai/dsh-session' +import { SessionSeq, SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl' @@ -68,7 +68,7 @@ async function writeLog(persistence: SessionPersistence, m: SessionHeader, event async function readAll(persistence: SessionPersistence, id: SessionId): Promise<{ meta: SessionHeader; events: readonly SessionEvent[] }> { const handle = await persistence.open(id, 'read') try { - return { meta: handle.header, events: await handle.read() } + return { meta: handle.header, events: (await handle.read()).events } } finally { await handle.close() } @@ -411,7 +411,7 @@ describe('JsonlSessionPersistence: default Zstandard encoding', () => { expect((await readAll(ctx.sessionPersistence, header.id)).events).toEqual(oneTurnLog()) }) - it('publishes v2 beside an unchanged compressed v0 source before returning a read handle', async () => { + it('serves a migrated compressed v0 read without publishing a successor', async () => { const root = await freshRoot() const ctx = await mount(root) const header = meta('zstd-v0-read', '/work') @@ -429,11 +429,7 @@ describe('JsonlSessionPersistence: default Zstandard encoding', () => { events: oneTurnLog(), }) expect(await readFile(sourcePath)).toEqual(source) - const current = (await decodeCompleteFrames(await readFile(currentPath))).toString().split('\n') - expect(JSON.parse(current[0] as string)).toMatchObject({ - id: header.id, - version: SESSION_FORMAT_VERSION, - }) + await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' }) })