mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-10-26 16:55:08 +00:00
imccyu
6aa157a05a
list(), the root encoding check, and historical corpus fingerprints read project directories, Session directories, headers, and stats one at a time. On a network filesystem every listing paid the summed round trips. These reads now run through one p-limit limiter per backend (listConcurrency, default 3), shared by concurrent listings; results and the first reported failure keep the sequential order.
1838 lines
74 KiB
TypeScript
1838 lines
74 KiB
TypeScript
/**
|
|
* JSONL durable session-persistence backend. It stores a header and contiguous
|
|
* events in immutable generation files under one directory per session and serves the handle-based
|
|
* `SessionPersistence` API: `create`/`open` return per-session handles, and
|
|
* every read validates the same fail-closed storage contract.
|
|
* @module @deepseek-ai/dsh-session-persistence-jsonl
|
|
*/
|
|
|
|
import { Context, type Fiber } from '@deepseek-ai/cordis'
|
|
import z from '@deepseek-ai/schemastery'
|
|
import {
|
|
createSessionFormatCatalogWithChildren,
|
|
SessionFormatUnsupportedMigrationError,
|
|
sessionFormatCatalog,
|
|
} from '@deepseek-ai/dsh-session-format-catalog'
|
|
import { readdirSync, type Dirent } from 'node:fs'
|
|
import { open, mkdir, readdir, realpath, link, rm, stat, truncate } from 'node:fs/promises'
|
|
import { dirname, join, resolve } from 'node:path'
|
|
import { performance } from 'node:perf_hooks'
|
|
import { scheduler } from 'node:timers/promises'
|
|
import { createHash, randomBytes } from 'node:crypto'
|
|
import pLimit, { type LimitFunction } from 'p-limit'
|
|
import {
|
|
SessionPersistence, SessionPersistenceRevision, SessionFormatUnsupportedError,
|
|
SessionPersistenceCorruptionError,
|
|
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, 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'
|
|
import {
|
|
assertNoRetiredHeaderFields, encodeSegment, eventLines, generationLogFilename, generationLogPath, logPath, logSuffix,
|
|
parseGenerationLogFilename, projectDir, scanLog, sessionDir, SessionLogScanner, toHeaderLine,
|
|
type JsonlCompression,
|
|
} from './format.ts'
|
|
import {
|
|
compressZstdFrame, createZstdFrameDecoder, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames,
|
|
} from './zstd.ts'
|
|
import { ensureDurableDirectoryWin32, publishNewFileWin32 } from './win32.ts'
|
|
import { verifyCurrentGenerationInWorker } from './migration-verifier.ts'
|
|
import { prepareCatalogFacts } from './catalog-migration.ts'
|
|
import {
|
|
JsonlGenerationSourceChangedError,
|
|
JsonlGenerationUnsupportedMigrationError,
|
|
prepareJsonlMigration,
|
|
readStableJsonlFile,
|
|
type JsonlGenerationFormatAdapter,
|
|
type JsonlPhysicalIdentity,
|
|
type PreparedJsonlMigration,
|
|
} from './generation.ts'
|
|
|
|
export type { JsonlCompression } from './format.ts'
|
|
|
|
/**
|
|
* Internal handoff-reuse policy, not deployment configuration: a cold
|
|
* observation and the resume that immediately follows it reuse one parsed
|
|
* log, so the memo only needs the sessions in flight between those steps.
|
|
*/
|
|
const COLD_LOG_MEMO_MAX_ENTRIES = 2
|
|
|
|
const DEFAULT_COMPRESSION: JsonlCompression = 'zstd'
|
|
/**
|
|
* Internal scheduling constant, not deployment configuration: balance
|
|
* frame-boundary event-loop yields against `setImmediate` overhead. One frame
|
|
* remains an indivisible synchronous decode.
|
|
*/
|
|
const ZSTD_DECODE_YIELD_INTERVAL_MS = 500
|
|
|
|
/** Assert that the independently decodable first frame contains only the header record. */
|
|
function assertZstdHeaderFrame(plaintext: Buffer): void {
|
|
if (plaintext.length === 0 || plaintext.indexOf(0x0A) !== plaintext.length - 1) {
|
|
throw new Error('corrupt Zstandard session log: first frame is not exactly one header line')
|
|
}
|
|
}
|
|
|
|
/** Loader schema for the JSONL artifact's physical encoding. */
|
|
export const JsonlCompressionSchema: z<JsonlCompression> = z.union([
|
|
z.const('zstd'),
|
|
z.const('none'),
|
|
]).default(DEFAULT_COMPRESSION)
|
|
|
|
/** Default maximum listing filesystem operations in flight per backend. */
|
|
const DEFAULT_LIST_CONCURRENCY = 3
|
|
|
|
/** Plugin config for the JSONL backend's root, physical encoding, and listing concurrency. */
|
|
export interface Config {
|
|
/**
|
|
* Root directory for all session files. Required (no default): a default of
|
|
* `process.cwd()` would scatter session files as the process's cwd changes
|
|
* (bash calls, subprocesses). Sessions group under human-readable project
|
|
* directories, then per-session directories. An existing root must be a
|
|
* readable directory; an absent root is created on first materialization.
|
|
*/
|
|
root: string
|
|
/** Physical encoding; defaults to checksummed Zstandard frames. */
|
|
compression?: JsonlCompression
|
|
/**
|
|
* Maximum directory, header, and stat reads in flight across all concurrent
|
|
* listings of this backend (`list()`, the root encoding check, and historical
|
|
* fingerprints). Defaults to 3.
|
|
*/
|
|
listConcurrency?: number
|
|
}
|
|
|
|
/** Logical header paired with the catalog's selected physical-generation classification. */
|
|
interface GenerationHeader {
|
|
readonly header: SessionHeader
|
|
readonly formatStatus: NonNullable<SessionPersistenceSnapshot['formatStatus']>
|
|
}
|
|
|
|
/** 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
|
|
readonly tornTruncateTo: number | undefined
|
|
/** Complete events recovered from the torn final frame; the write path rewrites them durably. */
|
|
readonly recoveredTail: SessionEvent[]
|
|
/** Exact fork-inherited prefix length stored in the header line. */
|
|
readonly inheritedEventCount: SessionLogOffsetType
|
|
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 validateRelatedSources: () => Promise<void>
|
|
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 acyclic stored JSON; its arrays contain only indexed JSON values. */
|
|
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)
|
|
if (Array.isArray(current)) {
|
|
for (let index = 0; index < current.length; index += 1) {
|
|
const child: unknown = current[index]
|
|
if (child !== null && typeof child === 'object') pending.push(child)
|
|
}
|
|
} else {
|
|
for (const key in current) {
|
|
const child = (current as Record<string, unknown>)[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
|
|
readonly sourceVersion: number
|
|
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<PreparedStoredLog>
|
|
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([
|
|
identity.dev,
|
|
identity.ino,
|
|
identity.size,
|
|
identity.mtimeNs,
|
|
identity.ctimeNs,
|
|
].join(':'))
|
|
}
|
|
|
|
/** Whether a filesystem error means absence; every non-ENOENT failure must surface. */
|
|
function isENOENT(error: unknown): boolean {
|
|
return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
|
|
}
|
|
|
|
/**
|
|
* Run `task` over `items` through the backend's shared listing limiter and
|
|
* settle every call. Callers consume the outcomes in input order, so the first
|
|
* failure they report matches a sequential loop whatever the completion order.
|
|
* A limited task must never wait on the same limiter.
|
|
* @param limit - the backend's shared listing limiter.
|
|
* @param items - inputs in their sequential order.
|
|
* @param task - per-item operation.
|
|
* @returns one outcome per item, in input order.
|
|
*/
|
|
function settleLimited<T, R>(
|
|
limit: LimitFunction,
|
|
items: readonly T[],
|
|
task: (item: T) => Promise<R>,
|
|
): Promise<PromiseSettledResult<R>[]> {
|
|
return Promise.allSettled(items.map(item => limit(() => task(item))))
|
|
}
|
|
|
|
/**
|
|
* Unwrap outcomes from {@link settleLimited}, rethrowing the first rejection in input order.
|
|
* @param outcomes - outcomes in input order.
|
|
* @returns the fulfilled values in input order.
|
|
*/
|
|
function settledValues<R>(outcomes: readonly PromiseSettledResult<R>[]): R[] {
|
|
return outcomes.map((outcome) => {
|
|
if (outcome.status === 'rejected') throw outcome.reason
|
|
return outcome.value
|
|
})
|
|
}
|
|
|
|
/** Whether a filesystem-owned failure should retain its original errno and path. */
|
|
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<T>(operation: Promise<T>, signal?: AbortSignal): Promise<T> {
|
|
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<T>((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 backend registered as `ctx.sessionPersistence` by the plugin.
|
|
* Sessions materialize lazily: a created session is
|
|
* visible to this process immediately, reaches disk on its first append or
|
|
* flush, and never existed if the process crashes before that.
|
|
*/
|
|
class JsonlSessionPersistence extends SessionPersistence {
|
|
/** Backend label for diagnostics and effects; shadows `Service.name` without changing the service key. */
|
|
override readonly name = 'session-persistence-jsonl'
|
|
|
|
private root: string
|
|
private compression: JsonlCompression
|
|
/**
|
|
* Bounds directory, header, and stat I/O across every concurrent listing of
|
|
* this backend (`list()`, the root encoding check, historical fingerprints).
|
|
* Queued tasks of a cancelled listing run and reject at their signal check.
|
|
*/
|
|
private readonly listLimit: LimitFunction
|
|
private rootEncodingCheck: Promise<void> | undefined
|
|
private readonly generationFormat: Omit<JsonlGenerationFormatAdapter, 'createRestore'>
|
|
/**
|
|
* Bounded LRU of parsed, validated stored logs keyed by session id and
|
|
* guarded by the stat-derived revision, so an immediate cold-read handoff
|
|
* (observation then resume) parses the artifact once. Every local mutation
|
|
* for an id invalidates its entry; a foreign write misses through the
|
|
* revision guard.
|
|
*/
|
|
private readonly coldLogMemo = new Map<SessionId, StoredLog>()
|
|
/** One joinable decode/migration operation per selected historical Session file revision. */
|
|
private readonly migrationPreparations = new Map<SessionId, MigrationPreparation>()
|
|
|
|
constructor(ctx: Context, public config: Config, private readonly tracker: JsonlBackendTracker) {
|
|
super(ctx)
|
|
/* v8 ignore next 5 -- generated catalog and Session source share one build-time version owner. */
|
|
if (sessionFormatCatalog.currentVersion !== SESSION_FORMAT_VERSION) {
|
|
throw new Error(
|
|
`session-persistence-jsonl: format catalog v${sessionFormatCatalog.currentVersion} `
|
|
+ `does not match Session v${SESSION_FORMAT_VERSION}`,
|
|
)
|
|
}
|
|
// Resolve once so later process.cwd() changes cannot split one backend across roots.
|
|
this.root = resolve(config.root)
|
|
this.compression = config.compression ?? DEFAULT_COMPRESSION
|
|
this.listLimit = pLimit(config.listConcurrency ?? DEFAULT_LIST_CONCURRENCY)
|
|
this.generationFormat = {
|
|
currentVersion: sessionFormatCatalog.currentVersion,
|
|
encodeHeader: (header, inheritedEventCount) =>
|
|
sessionFormatCatalog.encodeCurrentHeader(header, inheritedEventCount),
|
|
encodeEvent: event => sessionFormatCatalog.encodeCurrentEvent(event),
|
|
isUnsupportedMigrationError: (error): error is SessionFormatUnsupportedMigrationError =>
|
|
error instanceof SessionFormatUnsupportedMigrationError,
|
|
}
|
|
this.assertUsableRoot()
|
|
}
|
|
|
|
/**
|
|
* Refusal-diagnostics hook: the absolute target path, without touching the filesystem.
|
|
* @param meta - the stored header naming the session and its cwd.
|
|
* @returns the artifact kind and absolute path.
|
|
*/
|
|
private locate(meta: SessionHeader): SessionLocation {
|
|
return { kind: 'jsonl', path: logPath(this.root, meta.cwd, meta.id, this.compression) }
|
|
}
|
|
|
|
// --- SessionPersistence service API ---
|
|
|
|
/**
|
|
* Create a new stored session and take its write ownership. The session is
|
|
* visible to this process immediately; the physical artifact appears on the
|
|
* first append or flush.
|
|
* @param header - the immutable header to store; must be losslessly
|
|
* JSON-serializable with a non-negative safe-integer `createdAt`.
|
|
* @param options - optional cancellation.
|
|
* @returns the owned write handle.
|
|
*/
|
|
async create(header: SessionHeader, options?: SessionPersistenceCreateOptions): Promise<SessionHandle> {
|
|
options?.signal?.throwIfAborted()
|
|
const snapshot = materializeCreateHeader(header)
|
|
// Fail fast on a seeded/cut mismatch with the exact refusal the header
|
|
// line encoder enforces at materialization.
|
|
toHeaderLine(snapshot, options?.inheritedEventCount)
|
|
const inheritedEventCount = SessionLogOffset(options?.inheritedEventCount ?? 0)
|
|
await this.ensureRootEncoding()
|
|
options?.signal?.throwIfAborted()
|
|
if (this.tracker.hasPending(snapshot.id) || await this.findLog(snapshot.id, options?.signal) !== undefined) {
|
|
throw new SessionAlreadyExistsError(snapshot.id)
|
|
}
|
|
options?.signal?.throwIfAborted()
|
|
// No lock yet: before materialization there is no durable artifact for
|
|
// another process to contend over, so the handle acquires the lock right
|
|
// before its first log bytes publish (ensureLease); an unmaterialized
|
|
// session leaves no filesystem footprint at all.
|
|
this.tracker.registerCreated(snapshot, inheritedEventCount)
|
|
return this.tracker.adopt(new JsonlSessionHandle(this, snapshot.id, snapshot, 'write', { cursor: 0, materialized: false, inheritedEventCount }))
|
|
}
|
|
|
|
/**
|
|
* Open an existing stored session for `read` or single-writer `write`.
|
|
* @param id - the stored session to open.
|
|
* @param access - `read` (no ownership) or `write` (atomic in-process claim).
|
|
* @param options - optional cancellation.
|
|
* @returns the open handle.
|
|
*/
|
|
async open(id: SessionId, access: SessionAccess, options?: SessionPersistenceOpenOptions): Promise<SessionHandle> {
|
|
options?.signal?.throwIfAborted()
|
|
await this.ensureRootEncoding()
|
|
options?.signal?.throwIfAborted()
|
|
const pending = this.tracker.pendingOf(id)
|
|
if (access === 'read') {
|
|
if (pending !== undefined) {
|
|
return this.tracker.adopt(new JsonlSessionHandle(this, id, pending.header, 'read', { cursor: 0, materialized: false, inheritedEventCount: pending.inheritedEventCount }))
|
|
}
|
|
let stored: StoredLog
|
|
try {
|
|
stored = await this.requireStoredLog(id, options?.signal)
|
|
} catch (error: unknown) {
|
|
if (!(error instanceof JsonlGenerationSourceChangedError)) throw error
|
|
stored = await this.requireStoredLog(id, options?.signal)
|
|
}
|
|
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.
|
|
this.tracker.claimWrite(id)
|
|
let lease: SessionWriteLease | undefined
|
|
try {
|
|
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 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,
|
|
}, lease))
|
|
} catch (error) {
|
|
// Free the in-process claim no matter how the kernel-lock release
|
|
// fares, and keep the original diagnostic: a release failure joins it
|
|
// instead of replacing it.
|
|
/* v8 ignore next -- typed backends and fs reject with Error */
|
|
const failure = error instanceof Error ? error : new Error(String(error))
|
|
let releaseFailure: Error | undefined
|
|
try {
|
|
await lease?.release()
|
|
} catch (raw: unknown) {
|
|
/* v8 ignore next -- lock releases reject with Error */
|
|
releaseFailure = raw instanceof Error ? raw : new Error(String(raw))
|
|
}
|
|
this.tracker.releaseClaim(id)
|
|
if (releaseFailure !== undefined) {
|
|
throw new AggregateError([failure, releaseFailure], `session "${id}": write open failed and its lock release failed`)
|
|
}
|
|
throw failure
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Flush every active write handle in one durability barrier; see the seam
|
|
* contract.
|
|
* @returns resolution once every write handle active at the call has flushed.
|
|
*/
|
|
flush(): Promise<void> {
|
|
return this.tracker.flushAll()
|
|
}
|
|
|
|
/**
|
|
* Observe one stored session without reading its event log.
|
|
* @param id - the stored session to observe.
|
|
* @param options - optional cancellation.
|
|
* @returns the snapshot (`sizeBytes` carries the physical artifact size), or
|
|
* `undefined` when the session does not exist.
|
|
*/
|
|
async stat(
|
|
id: SessionId,
|
|
options?: SessionPersistenceStatOptions,
|
|
): Promise<SessionPersistenceSnapshot | undefined> {
|
|
options?.signal?.throwIfAborted()
|
|
await this.ensureRootEncoding()
|
|
options?.signal?.throwIfAborted()
|
|
const pending = this.tracker.pendingOf(id)
|
|
if (pending !== undefined) {
|
|
return { header: pending.header, revision: pending.revision }
|
|
}
|
|
const selected = await this.findLog(id, options?.signal)
|
|
if (selected === undefined) return undefined
|
|
const metadata = await this.readGenerationHeader(selected, id, options?.signal)
|
|
if (metadata === undefined) return undefined
|
|
try {
|
|
const identity = await stat(selected.sourcePath, { bigint: true })
|
|
options?.signal?.throwIfAborted()
|
|
return {
|
|
...metadata,
|
|
revision: selected.sourceVersion < SESSION_FORMAT_VERSION
|
|
? SessionPersistenceRevision(`${fileRevision(identity)}:${await this.historicalCorpusRevision(options?.signal)}`)
|
|
: fileRevision(identity),
|
|
sizeBytes: Number(identity.size),
|
|
}
|
|
} catch (error: unknown) {
|
|
options?.signal?.throwIfAborted()
|
|
if (isENOENT(error)) return undefined
|
|
throw error
|
|
}
|
|
}
|
|
|
|
/**
|
|
* List every stored session visible to this process: materialized artifacts
|
|
* plus this process's created-but-unmaterialized sessions.
|
|
* @param options - optional cancellation.
|
|
* @returns one snapshot per session, in no promised order.
|
|
*/
|
|
async list(options?: SessionPersistenceListOptions): Promise<readonly SessionPersistenceSnapshot[]> {
|
|
const signal = options?.signal
|
|
const snapshots: SessionPersistenceSnapshot[] = []
|
|
const listed = new Set<SessionId>()
|
|
// Snapshot pending entries BEFORE scanning storage: a session whose first
|
|
// append lands mid-scan is then still in this snapshot (its artifact may
|
|
// predate the scan), so create-to-list visibility never has a hole.
|
|
const pending = [...this.tracker.pendingEntries()]
|
|
const artifacts = await this.listArtifacts(signal)
|
|
const corpusRevision = artifacts.some(artifact => artifact.sourceVersion < SESSION_FORMAT_VERSION)
|
|
? await this.historicalCorpusRevision(signal) : undefined
|
|
const statted = settledValues(await settleLimited(this.listLimit, artifacts, async (artifact) => {
|
|
signal?.throwIfAborted()
|
|
try {
|
|
const identity = await stat(artifact.path, { bigint: true })
|
|
signal?.throwIfAborted()
|
|
return { artifact, identity }
|
|
} catch (error: unknown) {
|
|
signal?.throwIfAborted()
|
|
if (!isENOENT(error)) throw error
|
|
return undefined
|
|
}
|
|
}))
|
|
for (const entry of statted) {
|
|
if (entry === undefined) continue
|
|
const { artifact, identity } = entry
|
|
listed.add(artifact.header.id)
|
|
snapshots.push({
|
|
header: artifact.header,
|
|
formatStatus: artifact.formatStatus,
|
|
revision: artifact.sourceVersion < SESSION_FORMAT_VERSION
|
|
? SessionPersistenceRevision(`${fileRevision(identity)}:${corpusRevision}`)
|
|
: fileRevision(identity),
|
|
sizeBytes: Number(identity.size),
|
|
})
|
|
}
|
|
for (const [id, entry] of pending) {
|
|
if (!listed.has(id)) snapshots.push({ header: entry.header, revision: entry.revision })
|
|
}
|
|
signal?.throwIfAborted()
|
|
return snapshots
|
|
}
|
|
|
|
// --- handle-facing storage internals (package-private via the handle class below) ---
|
|
|
|
/** Resolve and read one stored log, refusing loudly when the artifact is absent. */
|
|
private async requireStoredLog(id: SessionId, signal?: AbortSignal): Promise<StoredLog> {
|
|
const selected = await this.findLog(id, signal)
|
|
if (selected === undefined) throw new SessionPersistenceNotFoundError(id)
|
|
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)
|
|
}
|
|
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(
|
|
selected.sourcePath,
|
|
id,
|
|
current.bytes,
|
|
fileRevision(current.identity),
|
|
signal,
|
|
)
|
|
}
|
|
|
|
/** Probe the memo and otherwise decode one historical generation under backend cancellation. */
|
|
private async loadStoredMigration(
|
|
id: SessionId,
|
|
selected: ResolvedJsonlGeneration,
|
|
sourceRevision: PersistenceRevision,
|
|
signal: AbortSignal,
|
|
): Promise<PreparedStoredLog> {
|
|
signal.throwIfAborted()
|
|
const memoized = this.coldLogMemo.get(id)
|
|
if (memoized?.status === 'prepared' && memoized.revision === sourceRevision) {
|
|
try {
|
|
await memoized.validateRelatedSources()
|
|
} catch (error: unknown) {
|
|
this.coldLogMemo.delete(id)
|
|
throw this.generationFailure(id, selected, error)
|
|
}
|
|
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<PreparedStoredLog> {
|
|
preparation.waiters += 1
|
|
try {
|
|
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<PreparedStoredLog> {
|
|
let prepared: Awaited<ReturnType<typeof prepareJsonlMigration>>
|
|
let validateRelatedSources: () => Promise<void>
|
|
try {
|
|
const children = async () => (await this.listArtifacts(signal))
|
|
.filter(source => source.header.origin === 'subagent' && source.header.parentSession === id)
|
|
const sources = await children()
|
|
const related = await prepareCatalogFacts(id, sources, this.compression, signal)
|
|
for (const failure of related.failures) {
|
|
this.ctx.logger.warn(`${this.name}: session "${id}" catalog retained a child with unknown descriptor (raw log: ${failure.path}): ${String(failure.error)}`)
|
|
}
|
|
const membership = sources.map(source => source.path).sort()
|
|
validateRelatedSources = async () => {
|
|
const current = (await children()).map(source => source.path).sort()
|
|
const before = new Set(membership)
|
|
const after = new Set(current)
|
|
const changed = current.find(path => !before.has(path)) ?? membership.find(path => !after.has(path))
|
|
if (changed !== undefined) throw new JsonlGenerationSourceChangedError(changed)
|
|
await related.validate()
|
|
}
|
|
prepared = await prepareJsonlMigration({
|
|
sourcePath: selected.sourcePath,
|
|
sourceVersion: selected.sourceVersion,
|
|
currentPath: selected.currentPath,
|
|
compression: this.compression,
|
|
format: {
|
|
...this.generationFormat,
|
|
createRestore: header => createSessionFormatCatalogWithChildren(related.facts).createRestore(header, {
|
|
recovery: 'recoverable', validation: 'transformed',
|
|
}),
|
|
},
|
|
validateRelatedSources,
|
|
verifyCurrentFile: verifyCurrentGenerationInWorker,
|
|
validateHistoricalHeader: headerValue => this.validateSourceIdentity(
|
|
selected,
|
|
headerValue,
|
|
id,
|
|
signal,
|
|
),
|
|
signal,
|
|
})
|
|
} catch (error: unknown) {
|
|
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',
|
|
validateRelatedSources,
|
|
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<CurrentStoredLog> {
|
|
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 },
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Read, parse, and validate one stored log as the current logical prefix.
|
|
* @param path - the artifact file to read.
|
|
* @param expectedId - the session identity the artifact must carry.
|
|
* @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<CurrentStoredLog> {
|
|
signal?.throwIfAborted()
|
|
const probe = fileRevision(await stat(path, { bigint: true }))
|
|
const memoized = this.coldLogMemo.get(expectedId)
|
|
if (memoized?.status === 'current' && memoized.revision === probe) {
|
|
this.coldLogMemo.delete(expectedId)
|
|
this.coldLogMemo.set(expectedId, memoized)
|
|
return memoized
|
|
}
|
|
const { bytes, identity } = await readStableJsonlFile(path, signal)
|
|
return this.decodeStoredLog(path, expectedId, bytes, fileRevision(identity), signal)
|
|
}
|
|
|
|
/** Decode and memoize one already-stable current physical snapshot. */
|
|
private async decodeStoredLog(
|
|
path: string,
|
|
expectedId: SessionId,
|
|
buffer: Buffer,
|
|
revision: PersistenceRevision,
|
|
signal?: AbortSignal,
|
|
): Promise<CurrentStoredLog> {
|
|
let parsed: {
|
|
meta: SessionHeader
|
|
inheritedEventCount: SessionLogOffsetType
|
|
events: SessionEvent[]
|
|
tornTruncateTo: number | undefined
|
|
recoveredTail: SessionEvent[]
|
|
}
|
|
try {
|
|
if (this.compression === 'zstd') {
|
|
parsed = await this.readZstdPrefix(buffer, signal)
|
|
} else {
|
|
signal?.throwIfAborted()
|
|
const { meta, inheritedEventCount, events, committedBytes } = scanLog(buffer)
|
|
signal?.throwIfAborted()
|
|
parsed = {
|
|
meta,
|
|
inheritedEventCount,
|
|
events,
|
|
tornTruncateTo: committedBytes < buffer.byteLength ? committedBytes : undefined,
|
|
// A torn raw tail is one incomplete JSONL line; it holds no complete
|
|
// record to recover.
|
|
recoveredTail: [],
|
|
}
|
|
}
|
|
} catch (error: unknown) {
|
|
signal?.throwIfAborted()
|
|
// A parse-time format refusal predates any SessionHeader, so attach the
|
|
// artifact this read actually refused; every other parse failure is
|
|
// committed bytes the decoder cannot interpret — damage, classified for
|
|
// the seam's stable error vocabulary.
|
|
if (error instanceof SessionFormatUnsupportedError) {
|
|
throw new SessionFormatUnsupportedError(`${error.message} (raw log: ${path})`, { kind: 'jsonl', path })
|
|
}
|
|
throw new SessionPersistenceCorruptionError(`session "${expectedId}": stored log is corrupt: ${String(error)} (raw log: ${path})`, { cause: error })
|
|
}
|
|
signal?.throwIfAborted()
|
|
await this.assertStoredIdentity(path, SESSION_FORMAT_VERSION, parsed.meta, expectedId, signal)
|
|
signal?.throwIfAborted()
|
|
assertStoredId(expectedId, parsed.meta)
|
|
const location = this.locate(parsed.meta)
|
|
validateStoredEvents(parsed.meta, parsed.events, location)
|
|
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)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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 current artifact path, or `undefined` while only a historical generation exists.
|
|
*/
|
|
async resolveCurrentLog(id: SessionId, signal?: AbortSignal): Promise<string | undefined> {
|
|
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
|
|
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 },
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Durably append one validated batch; lazily materializes on the first write.
|
|
* @param header - the session's stored header.
|
|
* @param events - the validated contiguous batch, in seq order.
|
|
* @param isMaterialized - whether the session already has a durable artifact.
|
|
* @param inheritedEventCount - the exact fork-inherited prefix length written into a materializing header line.
|
|
*/
|
|
async persistBatch(
|
|
header: SessionHeader,
|
|
events: readonly SessionEvent[],
|
|
isMaterialized: boolean,
|
|
inheritedEventCount: SessionLogOffsetType,
|
|
): Promise<void> {
|
|
this.coldLogMemo.delete(header.id)
|
|
await this.ensureRootEncoding()
|
|
if (isMaterialized) {
|
|
await this.appendLines(header, events)
|
|
} else {
|
|
await this.materialize(header, inheritedEventCount, events)
|
|
this.tracker.materialized(header.id)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Materialize a header-only artifact for an explicitly durable empty session.
|
|
* @param header - the session's stored header.
|
|
* @param inheritedEventCount - the exact fork-inherited prefix length written into the header line.
|
|
*/
|
|
async persistHeader(header: SessionHeader, inheritedEventCount: SessionLogOffsetType): Promise<void> {
|
|
this.coldLogMemo.delete(header.id)
|
|
await this.ensureRootEncoding()
|
|
await this.materialize(header, inheritedEventCount, [])
|
|
this.tracker.materialized(header.id)
|
|
}
|
|
|
|
/**
|
|
* Truncate a torn physical tail durably before this session's first new append.
|
|
* @param header - the session's stored header.
|
|
* @param truncateTo - the byte offset the artifact is truncated to.
|
|
*/
|
|
async truncateTornTail(header: SessionHeader, truncateTo: number): Promise<void> {
|
|
this.coldLogMemo.delete(header.id)
|
|
await this.repair(header, truncateTo)
|
|
this.ctx.logger.warn(`${this.name}: session "${header.id}" recovered from a torn tail; incomplete tail bytes were discarded`)
|
|
}
|
|
|
|
/**
|
|
* Whether this process still tracks a created-but-unmaterialized session.
|
|
* @param id - the session to test.
|
|
* @returns true while the pending entry exists.
|
|
*/
|
|
hasPendingSession(id: SessionId): boolean {
|
|
return this.tracker.hasPending(id)
|
|
}
|
|
|
|
/**
|
|
* Release one handle's backend bookkeeping on close.
|
|
* @param handle - the closing handle.
|
|
* @param materialized - whether the session reached durable storage.
|
|
*/
|
|
releaseHandle(handle: JsonlSessionHandle, materialized: boolean): void {
|
|
this.tracker.release(handle, materialized)
|
|
}
|
|
|
|
/**
|
|
* Acquire the session directory's kernel write lock; the kernel holds it
|
|
* until the handle's close releases the descriptor, including on process death.
|
|
* @param id - the session the lock guards.
|
|
* @param cwd - header cwd used to derive the directory for a fresh session.
|
|
* @param dir - the resolved directory of an existing artifact, when known.
|
|
* @returns the held lock.
|
|
*/
|
|
private acquireLease(id: SessionId, cwd: string | undefined, dir = sessionDir(this.root, cwd, id)): Promise<SessionWriteLease> {
|
|
return SessionWriteLease.acquire(dir, id)
|
|
}
|
|
|
|
/**
|
|
* Acquire the cross-process write lock for a materializing created session,
|
|
* called by its handle immediately before the first log bytes publish.
|
|
* @param header - the session's stored header (its cwd derives the directory).
|
|
* @returns the held lock.
|
|
*/
|
|
async acquireWriteLease(header: SessionHeader): Promise<SessionWriteLease> {
|
|
// Refuse an opposite-encoding artifact before the lock's mkdir publishes
|
|
// the session directory — the last moment the directory can be absent.
|
|
await this.rejectOppositeArtifact(header.cwd, header.id)
|
|
return this.acquireLease(header.id, header.cwd)
|
|
}
|
|
|
|
/** Decode complete frames and retain complete JSONL records from a torn final frame. */
|
|
private async readZstdPrefix(
|
|
buffer: Buffer,
|
|
signal?: AbortSignal,
|
|
): Promise<{
|
|
meta: SessionHeader
|
|
inheritedEventCount: SessionLogOffsetType
|
|
events: SessionEvent[]
|
|
tornTruncateTo: number | undefined
|
|
recoveredTail: SessionEvent[]
|
|
}> {
|
|
signal?.throwIfAborted()
|
|
const { frames, tornStart } = scanZstdFrames(buffer)
|
|
signal?.throwIfAborted()
|
|
if (frames.length === 0) throw new Error('empty or header-less Zstandard session log')
|
|
|
|
const decoder = createZstdFrameDecoder()
|
|
let yieldDeadline = performance.now() + ZSTD_DECODE_YIELD_INTERVAL_MS
|
|
try {
|
|
const decodedFrames = decoder.decode(buffer, frames)
|
|
signal?.throwIfAborted()
|
|
const headerFrame = decodedFrames.next()
|
|
signal?.throwIfAborted()
|
|
/* v8 ignore next -- a non-empty structural frame list makes the decoder yield its first frame or throw. */
|
|
if (headerFrame.done) throw new Error('empty or header-less Zstandard session log')
|
|
assertZstdHeaderFrame(headerFrame.value)
|
|
const scanner = new SessionLogScanner(headerFrame.value)
|
|
|
|
let remainingFrames = frames.length - 1
|
|
for (const plaintext of decodedFrames) {
|
|
signal?.throwIfAborted()
|
|
scanner.write(plaintext)
|
|
remainingFrames -= 1
|
|
if (remainingFrames > 0 && performance.now() >= yieldDeadline) {
|
|
await scheduler.yield()
|
|
signal?.throwIfAborted()
|
|
yieldDeadline = performance.now() + ZSTD_DECODE_YIELD_INTERVAL_MS
|
|
}
|
|
}
|
|
signal?.throwIfAborted()
|
|
const complete = scanner.checkpoint()
|
|
if (complete.committedBytes !== complete.inputBytes) {
|
|
throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record')
|
|
}
|
|
if (tornStart === undefined) {
|
|
const prefix = scanner.finish()
|
|
return {
|
|
meta: prefix.meta,
|
|
inheritedEventCount: prefix.inheritedEventCount,
|
|
events: prefix.events,
|
|
tornTruncateTo: undefined,
|
|
recoveredTail: [],
|
|
}
|
|
}
|
|
// A torn final frame's append never resolved, but complete JSONL records
|
|
// already flushed into it are real emitted events: recover them, and let
|
|
// the write path truncate the torn bytes and rewrite them durably.
|
|
let recoveredPlaintext: Buffer = Buffer.alloc(0)
|
|
try {
|
|
signal?.throwIfAborted()
|
|
recoveredPlaintext = await decompressZstdPrefix(buffer.subarray(tornStart))
|
|
} catch {
|
|
/* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */
|
|
if (signal?.aborted) signal.throwIfAborted()
|
|
// A structurally incomplete final frame may end before Node's decoder
|
|
// can emit any plaintext; the complete prior frames remain recoverable.
|
|
}
|
|
signal?.throwIfAborted()
|
|
scanner.write(recoveredPlaintext)
|
|
const prefix = scanner.finish()
|
|
return {
|
|
meta: prefix.meta,
|
|
inheritedEventCount: prefix.inheritedEventCount,
|
|
events: prefix.events,
|
|
tornTruncateTo: tornStart,
|
|
recoveredTail: prefix.events.slice(complete.eventCount),
|
|
}
|
|
} catch (error) {
|
|
/* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */
|
|
if (signal?.aborted) signal.throwIfAborted()
|
|
throw error
|
|
} finally {
|
|
decoder.close()
|
|
}
|
|
}
|
|
|
|
/** Enumerate selected physical generations without interpreting their headers or bodies. */
|
|
private async listGenerations(signal?: AbortSignal): Promise<ResolvedJsonlGeneration[]> {
|
|
const selected = await this.mapSessionDirectories((dir) => {
|
|
signal?.throwIfAborted()
|
|
return this.resolveGenerationInDirectory(dir, signal)
|
|
}, signal)
|
|
return selected.filter(source => source !== undefined)
|
|
}
|
|
|
|
/**
|
|
* Apply `task` to every Session directory through the shared listing
|
|
* limiter. Results and the first failure follow the sequential
|
|
* project-then-directory order; a project whose listing fails ends the
|
|
* traversal after the directories of the projects before it.
|
|
* @param task - per-directory operation; must not wait on the listing limiter.
|
|
* @param signal - optional cancellation for directory listings.
|
|
* @returns per-directory results in traversal order.
|
|
*/
|
|
private async mapSessionDirectories<R>(
|
|
task: (dir: string) => Promise<R>,
|
|
signal?: AbortSignal,
|
|
): Promise<R[]> {
|
|
const limit = this.listLimit
|
|
// The root read counts against the same bound as every other listing read.
|
|
const projectDirs = await limit(() => this.listProjectDirs(signal))
|
|
const projects = await settleLimited(limit, projectDirs, project => this.listSessionDirs(project, signal))
|
|
const dirs: string[] = []
|
|
let projectFailure: PromiseRejectedResult | undefined
|
|
for (const outcome of projects) {
|
|
if (outcome.status === 'rejected') {
|
|
projectFailure = outcome
|
|
break
|
|
}
|
|
// Appended one by one: spreading a large project as call arguments exceeds the engine's argument limit.
|
|
for (const dir of outcome.value) dirs.push(dir)
|
|
}
|
|
const results = settledValues(await settleLimited(limit, dirs, task))
|
|
if (projectFailure !== undefined) throw projectFailure.reason
|
|
return results
|
|
}
|
|
|
|
/** Historical logical events depend on the corpus, including members with unreadable headers. */
|
|
private async historicalCorpusRevision(signal?: AbortSignal): Promise<string> {
|
|
const paths = (await this.listGenerations(signal)).map(source => source.sourcePath).sort()
|
|
const entries = settledValues(await settleLimited(this.listLimit, paths, async (path) => {
|
|
signal?.throwIfAborted()
|
|
try {
|
|
return { path, revision: fileRevision(await stat(path, { bigint: true })) }
|
|
} catch (error: unknown) {
|
|
if (!isENOENT(error)) throw error
|
|
return { path, revision: 'missing' }
|
|
}
|
|
}))
|
|
const hash = createHash('sha256')
|
|
for (const { path, revision } of entries) hash.update(JSON.stringify([path, revision]))
|
|
signal?.throwIfAborted()
|
|
return hash.digest('hex')
|
|
}
|
|
|
|
private async listArtifacts(
|
|
signal?: AbortSignal,
|
|
): Promise<Array<GenerationHeader & { path: string; sourceVersion: number }>> {
|
|
signal?.throwIfAborted()
|
|
await this.ensureRootEncoding()
|
|
signal?.throwIfAborted()
|
|
const generations = await this.listGenerations(signal)
|
|
const headers = await settleLimited(this.listLimit, generations, async (selected) => {
|
|
signal?.throwIfAborted()
|
|
try {
|
|
return { selected, metadata: await this.readGenerationHeader(selected, undefined, signal) }
|
|
} catch (error: unknown) {
|
|
if (error instanceof SessionFormatUnsupportedError || error instanceof SessionPersistenceCorruptionError) {
|
|
return { selected, metadata: undefined }
|
|
}
|
|
throw error
|
|
}
|
|
})
|
|
const artifacts: Array<GenerationHeader & { path: string; sourceVersion: number }> = []
|
|
const ids = new Set<SessionId>()
|
|
for (const outcome of headers) {
|
|
if (outcome.status === 'rejected') throw outcome.reason
|
|
const { selected, metadata } = outcome.value
|
|
if (metadata === undefined) continue
|
|
if (ids.has(metadata.header.id)) {
|
|
throw new Error(`duplicate JSONL session id "${metadata.header.id}" appears in multiple project directories`)
|
|
}
|
|
ids.add(metadata.header.id)
|
|
artifacts.push({ ...metadata, path: selected.sourcePath, sourceVersion: selected.sourceVersion })
|
|
}
|
|
signal?.throwIfAborted()
|
|
return artifacts
|
|
}
|
|
|
|
/** Read and translate one selected generation header without inspecting its body. */
|
|
private async readGenerationHeader(
|
|
selected: ResolvedJsonlGeneration,
|
|
expectedId?: SessionId,
|
|
signal?: AbortSignal,
|
|
): Promise<GenerationHeader | undefined> {
|
|
let first: string | undefined
|
|
try {
|
|
first = this.compression === 'zstd'
|
|
? await this.readFirstZstdLine(selected.sourcePath, signal)
|
|
: await this.readFirstLine(selected.sourcePath, signal)
|
|
} catch (error: unknown) {
|
|
signal?.throwIfAborted()
|
|
if (isENOENT(error)) return undefined
|
|
throw error
|
|
}
|
|
signal?.throwIfAborted()
|
|
if (first === undefined) return undefined
|
|
let value: unknown
|
|
try {
|
|
value = JSON.parse(first)
|
|
} catch {
|
|
return undefined
|
|
}
|
|
assertNoRetiredHeaderFields(value)
|
|
const result = sessionFormatCatalog.readHeader(value)
|
|
if ('storedVersion' in result && result.storedVersion !== selected.sourceVersion) {
|
|
throw new Error(
|
|
`session generation filename identifies v${selected.sourceVersion}, `
|
|
+ `but its header identifies v${result.storedVersion}`,
|
|
)
|
|
}
|
|
if (result.status === 'unsupported') {
|
|
const physicalId = String((value as { id?: unknown }).id)
|
|
let reason = result.reason
|
|
/* v8 ignore else -- released historical header migrations cannot refuse after physical decoding. */
|
|
if (result.storedVersion > SESSION_FORMAT_VERSION) {
|
|
reason = sessionFormatVersionRefusal(physicalId, result.storedVersion)
|
|
}
|
|
throw new SessionFormatUnsupportedError(
|
|
`${reason} (raw log: ${selected.sourcePath})`,
|
|
{ kind: 'jsonl', path: selected.sourcePath },
|
|
)
|
|
}
|
|
if (result.status === 'malformed') return undefined
|
|
const header = this.currentHeader(result.header)
|
|
await this.assertStoredIdentity(
|
|
selected.sourcePath,
|
|
selected.sourceVersion,
|
|
header,
|
|
expectedId,
|
|
signal,
|
|
)
|
|
return {
|
|
header,
|
|
formatStatus: result.status,
|
|
}
|
|
}
|
|
|
|
/** Convert format-catalog string identities to current branded Session metadata. */
|
|
private currentHeader(header: {
|
|
readonly version: number
|
|
readonly id: string
|
|
readonly createdAt: number
|
|
readonly cwd?: string
|
|
readonly parentSession?: string
|
|
readonly isSeeded: boolean
|
|
readonly origin?: 'subagent'
|
|
readonly delegationDepth: number
|
|
readonly agentPreset?: string
|
|
}): SessionHeader {
|
|
/* v8 ignore next 3 -- readable catalog results are restored to its configured current version. */
|
|
if (header.version !== SESSION_FORMAT_VERSION) {
|
|
throw new Error(`format catalog returned non-current logical header v${header.version}`)
|
|
}
|
|
return {
|
|
version: SESSION_FORMAT_VERSION,
|
|
id: makeSessionId(header.id),
|
|
createdAt: header.createdAt,
|
|
...(header.cwd === undefined ? {} : { cwd: header.cwd }),
|
|
...(header.parentSession === undefined
|
|
? {}
|
|
: { parentSession: makeSessionId(header.parentSession) }),
|
|
isSeeded: header.isSeeded,
|
|
...(header.origin === undefined ? {} : { origin: header.origin }),
|
|
delegationDepth: header.delegationDepth,
|
|
...(header.agentPreset === undefined ? {} : { agentPreset: header.agentPreset }),
|
|
}
|
|
}
|
|
|
|
// --- materialization / append / repair (file mechanics) ---
|
|
|
|
/** Atomically write the header line + first batch (temp-write, fsync, publish). */
|
|
private async materialize(
|
|
meta: SessionHeader,
|
|
inheritedEventCount: SessionLogOffsetType,
|
|
events: readonly SessionEvent[],
|
|
): Promise<void> {
|
|
const project = projectDir(this.root, meta.cwd)
|
|
const dir = sessionDir(this.root, meta.cwd, meta.id)
|
|
const finalPath = logPath(this.root, meta.cwd, meta.id, this.compression)
|
|
await this.rejectOppositeArtifact(meta.cwd, meta.id)
|
|
const content = await this.encodeMaterialization(meta, inheritedEventCount, events)
|
|
/* v8 ignore next -- native Windows coverage exercises this platform dispatch; Linux covers the POSIX peer */
|
|
if (process.platform === 'win32') {
|
|
await this.materializeWin32(project, dir, finalPath, meta.id, content)
|
|
} else {
|
|
await this.materializePosix(project, dir, finalPath, meta.id, content)
|
|
}
|
|
}
|
|
|
|
/* v8 ignore start -- Windows uses the Win32 durable-publish path; POSIX coverage exercises this peer. */
|
|
private async materializePosix(
|
|
project: string,
|
|
dir: string,
|
|
finalPath: string,
|
|
id: SessionId,
|
|
content: Buffer | string,
|
|
): Promise<void> {
|
|
await mkdir(this.root, { recursive: true, mode: 0o700 })
|
|
await this.syncDirPosix(dirname(this.root))
|
|
await mkdir(project, { recursive: true, mode: 0o700 })
|
|
await this.syncDirPosix(this.root)
|
|
await mkdir(dir, { recursive: true, mode: 0o700 })
|
|
await this.syncDirPosix(project)
|
|
await this.rejectExistingLog(finalPath, id)
|
|
const tmp = await this.writeSyncedTempFile(finalPath, content)
|
|
// Publish via link()+unlink(), NOT rename(): link fails with EEXIST if the
|
|
// final path already exists, so two processes materializing the same id
|
|
// concurrently cannot clobber each other. rename() would silently overwrite.
|
|
let linked = false
|
|
try {
|
|
await link(tmp, finalPath)
|
|
linked = true
|
|
} finally {
|
|
// Remove an unpublished temp on failure. After publication, defer cleanup
|
|
// until the directory entry is durable so cleanup cannot reject a live log.
|
|
/* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */
|
|
if (!linked) await rm(tmp, { force: true })
|
|
}
|
|
// link() succeeded — the log is published. fsync the directory so the new
|
|
// entry survives a power loss: the new link is not crash-durable until the
|
|
// parent directory's metadata is synced.
|
|
await this.syncDirPosix(dir)
|
|
// Best-effort temp cleanup: the log is already published and durable, so a
|
|
// failure to remove the redundant temp hard link must NOT reject the
|
|
// append. Swallow only the rm failure; nothing else of consequence runs here.
|
|
try {
|
|
await rm(tmp, { force: true })
|
|
} catch {
|
|
/* v8 ignore next -- redundant temp link; publish already durable, rm failure is an unreachable IO edge */
|
|
}
|
|
}
|
|
/* v8 ignore stop */
|
|
|
|
/* v8 ignore start -- native Windows coverage exercises this integration path */
|
|
private async materializeWin32(
|
|
project: string,
|
|
dir: string,
|
|
finalPath: string,
|
|
id: SessionId,
|
|
content: Buffer | string,
|
|
): Promise<void> {
|
|
await ensureDurableDirectoryWin32(this.root)
|
|
await ensureDurableDirectoryWin32(project)
|
|
await ensureDurableDirectoryWin32(dir)
|
|
await this.rejectExistingLog(finalPath, id)
|
|
const tmp = await this.writeSyncedTempFile(finalPath, content)
|
|
try {
|
|
await publishNewFileWin32(tmp, finalPath)
|
|
} catch (error) {
|
|
await rm(tmp, { force: true })
|
|
throw error
|
|
}
|
|
}
|
|
/* v8 ignore stop */
|
|
|
|
private async rejectExistingLog(finalPath: string, id: SessionId): Promise<void> {
|
|
// Never publish over an existing committed log: materialize is the first
|
|
// write of a session the backend believes is new. A file here means a
|
|
// different session shares this id on disk — reject loudly. (create already
|
|
// guards the create path, so this is unreachable-in-practice TOCTOU
|
|
// defense.)
|
|
/* v8 ignore next 3 -- create guards collisions before materialize; this is a TOCTOU backstop */
|
|
if (await this.resolveGenerationInDirectory(dirname(finalPath)) !== undefined) {
|
|
throw new Error(`refusing to materialize "${id}": a log already exists on disk (open it instead)`)
|
|
}
|
|
}
|
|
|
|
private async writeSyncedTempFile(finalPath: string, content: Buffer | string): Promise<string> {
|
|
const tmp = `${finalPath}.${randomBytes(6).toString('hex')}.tmp`
|
|
const handle = await open(tmp, 'wx', 0o600)
|
|
try {
|
|
await handle.writeFile(content)
|
|
await handle.sync()
|
|
} finally {
|
|
await handle.close()
|
|
}
|
|
return tmp
|
|
}
|
|
|
|
/** Encode the header and first batch without combining their frame boundaries. */
|
|
private async encodeMaterialization(
|
|
meta: SessionHeader,
|
|
inheritedEventCount: SessionLogOffsetType,
|
|
events: readonly SessionEvent[],
|
|
): Promise<Buffer | string> {
|
|
const header = JSON.stringify(toHeaderLine(meta, meta.isSeeded ? inheritedEventCount : undefined)) + '\n'
|
|
if (events.length === 0) {
|
|
return this.compression === 'none' ? header : compressZstdFrame(header)
|
|
}
|
|
const body = eventLines(events) + '\n'
|
|
if (this.compression === 'none') return header + body
|
|
const headerFrame = await compressZstdFrame(header)
|
|
const eventFrame = await compressZstdFrame(body)
|
|
return Buffer.concat([headerFrame, eventFrame])
|
|
}
|
|
|
|
/** Encode one durable append batch in the configured physical representation. */
|
|
private async encodeEventBatch(events: readonly SessionEvent[]): Promise<Buffer | string> {
|
|
const body = eventLines(events) + '\n'
|
|
return this.compression === 'zstd' ? compressZstdFrame(body) : body
|
|
}
|
|
|
|
/** fsync a POSIX directory so a just-created/renamed entry is crash-durable. */
|
|
/* v8 ignore start -- Windows uses write-through namespace operations; POSIX coverage exercises directory fsync. */
|
|
private async syncDirPosix(dir: string): Promise<void> {
|
|
const handle = await open(dir, 'r')
|
|
try {
|
|
await handle.sync()
|
|
} finally {
|
|
await handle.close()
|
|
}
|
|
}
|
|
/* v8 ignore stop */
|
|
|
|
/**
|
|
* Append and fsync event lines. On a partial write or sync failure, restore the
|
|
* previous size before rethrowing because the unchanged cursor will retry the
|
|
* batch; leaving partial bytes would create duplicate sequence numbers.
|
|
*/
|
|
private async appendLines(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
|
|
const content = await this.encodeEventBatch(events)
|
|
const path = logPath(this.root, meta.cwd, meta.id, this.compression)
|
|
const handle = await open(path, 'a')
|
|
let closed = false
|
|
const closeAppendHandle = async (): Promise<void> => {
|
|
if (closed) return
|
|
closed = true
|
|
await handle.close()
|
|
}
|
|
|
|
try {
|
|
const { size: before } = await handle.stat()
|
|
try {
|
|
await handle.writeFile(content)
|
|
await handle.sync()
|
|
} catch (error) {
|
|
try {
|
|
await closeAppendHandle()
|
|
await this.rollbackAppend(path, before)
|
|
} catch (rollbackError) {
|
|
throw new AggregateError([error, rollbackError], `failed to roll back append to "${path}"`)
|
|
}
|
|
throw error
|
|
}
|
|
} finally {
|
|
await closeAppendHandle()
|
|
}
|
|
}
|
|
|
|
private async rollbackAppend(path: string, size: number): Promise<void> {
|
|
const handle = await open(path, 'r+')
|
|
try {
|
|
await handle.truncate(size)
|
|
await handle.sync()
|
|
} finally {
|
|
await handle.close()
|
|
}
|
|
}
|
|
|
|
/** Truncate the log file to `offset` bytes and fsync (discard the crash tail). */
|
|
private async repair(meta: SessionHeader, offset: number): Promise<void> {
|
|
const path = logPath(this.root, meta.cwd, meta.id, this.compression)
|
|
await truncate(path, offset)
|
|
const handle = await open(path, 'r+')
|
|
try {
|
|
await handle.sync()
|
|
} finally {
|
|
await handle.close()
|
|
}
|
|
}
|
|
|
|
// --- discovery helpers ---
|
|
|
|
/**
|
|
* Read the first newline-terminated line of a file without loading the whole
|
|
* file. Returns undefined if the file is empty or has no complete first line.
|
|
* Reads in bounded chunks so a huge log costs only the header read.
|
|
*/
|
|
private async readFirstLine(path: string, signal?: AbortSignal): Promise<string | undefined> {
|
|
signal?.throwIfAborted()
|
|
const handle = await open(path, 'r')
|
|
try {
|
|
signal?.throwIfAborted()
|
|
const chunks: Buffer[] = []
|
|
const buf = Buffer.alloc(8192)
|
|
for (;;) {
|
|
signal?.throwIfAborted()
|
|
const { bytesRead } = await handle.read(buf, 0, buf.length, null)
|
|
signal?.throwIfAborted()
|
|
if (bytesRead === 0) return undefined // EOF with no newline → no complete line
|
|
const slice = buf.subarray(0, bytesRead)
|
|
const nl = slice.indexOf(0x0a)
|
|
if (nl !== -1) {
|
|
chunks.push(slice.subarray(0, nl))
|
|
signal?.throwIfAborted()
|
|
return Buffer.concat(chunks).toString('utf8')
|
|
}
|
|
chunks.push(Buffer.from(slice))
|
|
}
|
|
} finally {
|
|
await handle.close()
|
|
}
|
|
}
|
|
|
|
/** Read only the header frame; compression failures reject as corruption, while I/O and cancellation propagate. */
|
|
private async readFirstZstdLine(path: string, signal?: AbortSignal): Promise<string | undefined> {
|
|
signal?.throwIfAborted()
|
|
const handle = await open(path, 'r')
|
|
try {
|
|
signal?.throwIfAborted()
|
|
let content = Buffer.alloc(0)
|
|
const chunk = Buffer.alloc(8192)
|
|
for (;;) {
|
|
signal?.throwIfAborted()
|
|
const { bytesRead } = await handle.read(chunk, 0, chunk.length, null)
|
|
signal?.throwIfAborted()
|
|
if (bytesRead === 0) return undefined
|
|
signal?.throwIfAborted()
|
|
content = Buffer.concat([content, chunk.subarray(0, bytesRead)])
|
|
signal?.throwIfAborted()
|
|
try {
|
|
const first = scanZstdFrames(content, 1).frames[0]
|
|
if (first === undefined) continue
|
|
const plaintext = await decompressZstdFrame(content.subarray(first.start, first.end))
|
|
signal?.throwIfAborted()
|
|
assertZstdHeaderFrame(plaintext)
|
|
return plaintext.subarray(0, -1).toString('utf8')
|
|
} catch (error) {
|
|
/* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */
|
|
if (signal?.aborted) signal.throwIfAborted()
|
|
throw new SessionPersistenceCorruptionError(
|
|
`corrupt Zstandard session log: header frame failed validation: ${String(error)} (raw log: ${path})`,
|
|
{ cause: error },
|
|
)
|
|
}
|
|
}
|
|
} finally {
|
|
await handle.close()
|
|
}
|
|
}
|
|
|
|
/** Select the numerically highest canonical generation in one Session directory. */
|
|
private async resolveGenerationInDirectory(
|
|
dir: string,
|
|
signal?: AbortSignal,
|
|
): Promise<ResolvedJsonlGeneration | undefined> {
|
|
signal?.throwIfAborted()
|
|
let entries: Dirent[]
|
|
try {
|
|
entries = await readdir(dir, { withFileTypes: true })
|
|
} catch (error: unknown) {
|
|
if (isENOENT(error)) return undefined
|
|
throw error
|
|
}
|
|
signal?.throwIfAborted()
|
|
const generations: Array<{ readonly path: string; readonly version: number }> = []
|
|
const opposite: string[] = []
|
|
for (const entry of entries) {
|
|
const version = parseGenerationLogFilename(entry.name, this.compression)
|
|
if (version !== undefined) {
|
|
generations.push({ path: join(dir, entry.name), version })
|
|
continue
|
|
}
|
|
if (parseGenerationLogFilename(entry.name, this.oppositeCompression()) !== undefined) {
|
|
opposite.push(join(dir, entry.name))
|
|
}
|
|
}
|
|
if (opposite.length > 0) throw this.encodingMismatch(opposite[0] as string)
|
|
const latest = generations.sort((left, right) => right.version - left.version)[0]
|
|
if (latest === undefined) return undefined
|
|
return {
|
|
sourcePath: latest.path,
|
|
sourceVersion: latest.version,
|
|
currentPath: join(
|
|
dir,
|
|
generationLogFilename(sessionFormatCatalog.currentVersion, this.compression),
|
|
),
|
|
}
|
|
}
|
|
|
|
/** Find the unique authoritative generation for an id across project directories. */
|
|
private async findLog(id: SessionId, signal?: AbortSignal): Promise<ResolvedJsonlGeneration | undefined> {
|
|
const matches: ResolvedJsonlGeneration[] = []
|
|
for (const project of await this.listProjectDirs(signal)) {
|
|
signal?.throwIfAborted()
|
|
await this.rejectLegacyFlatArtifact(project, id, signal)
|
|
signal?.throwIfAborted()
|
|
const dir = join(project, encodeSegment(id))
|
|
const selected = await this.resolveGenerationInDirectory(dir, signal)
|
|
if (selected !== undefined) matches.push(selected)
|
|
}
|
|
if (matches.length > 1) {
|
|
throw new Error(`duplicate JSONL session id "${id}" appears in multiple project directories`)
|
|
}
|
|
signal?.throwIfAborted()
|
|
return matches[0]
|
|
}
|
|
|
|
/** Require an existing configured root to be a readable directory. */
|
|
private assertUsableRoot(): void {
|
|
try {
|
|
readdirSync(this.root)
|
|
} catch (error) {
|
|
if (isENOENT(error)) return
|
|
throw error
|
|
}
|
|
}
|
|
|
|
/** Reject metadata that does not identify the selected physical log. */
|
|
private async assertStoredIdentity(
|
|
path: string,
|
|
storedVersion: number,
|
|
meta: SessionHeader,
|
|
expectedId?: SessionId,
|
|
signal?: AbortSignal,
|
|
): Promise<void> {
|
|
signal?.throwIfAborted()
|
|
if (expectedId !== undefined && meta.id !== expectedId) {
|
|
throw new Error(`corrupt session log "${path}": requested id "${expectedId}" does not match header id "${meta.id}"`)
|
|
}
|
|
let expectedPath: string
|
|
try {
|
|
expectedPath = generationLogPath(
|
|
this.root,
|
|
meta.cwd,
|
|
meta.id,
|
|
storedVersion,
|
|
this.compression,
|
|
)
|
|
} catch (error) {
|
|
throw new Error(`corrupt session log "${path}": header id cannot name a storage path`, { cause: error })
|
|
}
|
|
if (path !== expectedPath && !await this.sameFile(path, expectedPath, signal)) {
|
|
throw new Error(`corrupt session log "${path}": header id "${meta.id}" and cwd identify "${expectedPath}"`)
|
|
}
|
|
signal?.throwIfAborted()
|
|
}
|
|
|
|
/** Validate a supported historical header against the selected source path. */
|
|
private validateSourceIdentity(
|
|
selected: ResolvedJsonlGeneration,
|
|
headerValue: Readonly<Record<string, unknown>>,
|
|
expectedId: SessionId,
|
|
signal?: AbortSignal,
|
|
): void | Promise<void> {
|
|
const result = sessionFormatCatalog.readHeader(headerValue)
|
|
if (result.status !== 'current' && result.status !== 'migration-required') return
|
|
return this.assertStoredIdentity(
|
|
selected.sourcePath,
|
|
selected.sourceVersion,
|
|
this.currentHeader(result.header),
|
|
expectedId,
|
|
signal,
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Whether two path spellings resolve to the same physical file. This admits
|
|
* case aliases on case-insensitive filesystems without weakening identity
|
|
* checks on case-sensitive stores.
|
|
*/
|
|
private async sameFile(path: string, expectedPath: string, signal?: AbortSignal): Promise<boolean> {
|
|
signal?.throwIfAborted()
|
|
try {
|
|
const [actual, expected] = await Promise.all([realpath(path), realpath(expectedPath)])
|
|
signal?.throwIfAborted()
|
|
return actual === expected
|
|
} catch (error) {
|
|
signal?.throwIfAborted()
|
|
/* v8 ignore else -- non-ENOENT realpath failures require an external permission or I/O fault */
|
|
if (isENOENT(error)) return false
|
|
/* v8 ignore next -- non-ENOENT realpath failures are external I/O faults, propagated unchanged */
|
|
throw error
|
|
}
|
|
}
|
|
|
|
/** The human-readable project directories under the configured root. */
|
|
private async listProjectDirs(signal?: AbortSignal): Promise<string[]> {
|
|
try {
|
|
signal?.throwIfAborted()
|
|
const entries = await readdir(this.root, { withFileTypes: true })
|
|
signal?.throwIfAborted()
|
|
return entries.filter(e => e.isDirectory()).map(e => join(this.root, e.name))
|
|
} catch (error) {
|
|
// Only an absent root means no sessions; rethrow every other I/O failure.
|
|
if (isENOENT(error)) return []
|
|
throw error
|
|
}
|
|
}
|
|
|
|
/** List session-owned directories and reject the obsolete flat-file layout. */
|
|
private async listSessionDirs(project: string, signal?: AbortSignal): Promise<string[]> {
|
|
signal?.throwIfAborted()
|
|
const entries = await readdir(project, { withFileTypes: true })
|
|
signal?.throwIfAborted()
|
|
const legacy = entries.find(entry =>
|
|
entry.isFile() && (entry.name.endsWith('.jsonl') || entry.name.endsWith('.jsonl.zstd')))
|
|
if (legacy !== undefined) throw this.legacyLayout(join(project, legacy.name))
|
|
return entries.filter(entry => entry.isDirectory()).map(entry => join(project, entry.name))
|
|
}
|
|
|
|
/** Reject a root that already belongs to the other physical encoding. */
|
|
private ensureRootEncoding(): Promise<void> {
|
|
this.rootEncodingCheck ??= this.checkRootEncoding()
|
|
return this.rootEncodingCheck
|
|
}
|
|
|
|
private async checkRootEncoding(): Promise<void> {
|
|
await this.mapSessionDirectories(async (dir) => {
|
|
const incompatible = await this.findOppositeGenerationInDirectory(dir)
|
|
if (incompatible !== undefined) throw this.encodingMismatch(incompatible)
|
|
})
|
|
}
|
|
|
|
private async rejectLegacyFlatArtifact(
|
|
project: string,
|
|
id: SessionId,
|
|
signal?: AbortSignal,
|
|
): Promise<void> {
|
|
signal?.throwIfAborted()
|
|
const encoded = encodeSegment(id)
|
|
for (const compression of ['zstd', 'none'] as const) {
|
|
const path = join(project, encoded + logSuffix(compression))
|
|
const artifactExists = await this.exists(path)
|
|
signal?.throwIfAborted()
|
|
if (artifactExists) throw this.legacyLayout(path)
|
|
}
|
|
}
|
|
|
|
private async rejectOppositeArtifact(cwd: string | undefined, id: SessionId): Promise<void> {
|
|
const path = await this.findOppositeGenerationInDirectory(sessionDir(this.root, cwd, id))
|
|
if (path !== undefined) throw this.encodingMismatch(path)
|
|
}
|
|
|
|
/** Return the highest canonical generation encoded with the other configured suffix. */
|
|
private async findOppositeGenerationInDirectory(dir: string): Promise<string | undefined> {
|
|
let entries: Dirent[]
|
|
try {
|
|
entries = await readdir(dir, { withFileTypes: true })
|
|
} catch (error: unknown) {
|
|
if (isENOENT(error)) return undefined
|
|
throw error
|
|
}
|
|
const generations: Array<{ readonly name: string; readonly version: number }> = []
|
|
for (const entry of entries) {
|
|
const version = parseGenerationLogFilename(entry.name, this.oppositeCompression())
|
|
if (version !== undefined) generations.push({ name: entry.name, version })
|
|
}
|
|
const latest = generations.sort((left, right) => right.version - left.version)[0]
|
|
return latest === undefined ? undefined : join(dir, latest.name)
|
|
}
|
|
|
|
private oppositeCompression(): JsonlCompression {
|
|
return this.compression === 'zstd' ? 'none' : 'zstd'
|
|
}
|
|
|
|
private encodingMismatch(path: string): Error {
|
|
return new Error(
|
|
`session artifact ${JSON.stringify(path)} uses ${logSuffix(this.oppositeCompression())}, `
|
|
+ `but this backend is configured for compression ${JSON.stringify(this.compression)}; `
|
|
+ 'use a separate root or select the matching compression mode',
|
|
)
|
|
}
|
|
|
|
private legacyLayout(path: string): Error {
|
|
return new Error(
|
|
`session artifact ${JSON.stringify(path)} uses the unsupported flat-file layout; `
|
|
+ 'use a separate root or move it into a project/session directory before loading',
|
|
)
|
|
}
|
|
|
|
private async exists(path: string): Promise<boolean> {
|
|
try {
|
|
const handle = await open(path, 'r')
|
|
await handle.close()
|
|
return true
|
|
} catch (error) {
|
|
// Only ENOENT means absent. A permission/I/O error must surface rather
|
|
// than letting load or collision checks proceed under false absence.
|
|
/* v8 ignore else -- Windows reports file-valued parents as ENOENT; POSIX covers direct ENOTDIR. */
|
|
if (isENOENT(error)) {
|
|
// Windows reports ENOENT, not ENOTDIR, for `regular-file/child`, so it
|
|
// alone verifies the immediate parent to keep a blocked session
|
|
// directory a storage fault. POSIX open already reported ENOTDIR before
|
|
// this point, where the extra stat would only cost a syscall per probe.
|
|
/* v8 ignore next -- native Windows coverage exercises this platform dispatch; POSIX reports ENOTDIR from open */
|
|
if (process.platform === 'win32') await this.assertLogParentAllowsAbsence(path)
|
|
return false
|
|
}
|
|
/* v8 ignore next -- Windows repairs ENOTDIR from ENOENT above; POSIX covers direct ENOTDIR. */
|
|
throw error
|
|
}
|
|
}
|
|
|
|
/* v8 ignore start -- native Windows coverage exercises this repair; POSIX open reports ENOTDIR before this point. */
|
|
private async assertLogParentAllowsAbsence(path: string): Promise<void> {
|
|
try {
|
|
const parent = dirname(path)
|
|
const info = await stat(parent)
|
|
if (info.isDirectory()) return
|
|
const error = new Error(`ENOTDIR: parent path exists but is not a directory: ${parent}`) as NodeJS.ErrnoException
|
|
error.code = 'ENOTDIR'
|
|
error.path = parent
|
|
throw error
|
|
} catch (error) {
|
|
if (isENOENT(error)) return
|
|
throw error
|
|
}
|
|
}
|
|
/* v8 ignore stop */
|
|
}
|
|
|
|
/** JSONL persistence plugin label. */
|
|
export const name = 'session-persistence-jsonl'
|
|
|
|
/** Loader schema for the backend's root, physical encoding, and listing concurrency. */
|
|
export const Config: z<Config> = z.object({
|
|
root: z.string().required(),
|
|
compression: JsonlCompressionSchema,
|
|
listConcurrency: z.natural().min(1).default(DEFAULT_LIST_CONCURRENCY),
|
|
})
|
|
|
|
/**
|
|
* Mount persistence with live routing and handles retained through the
|
|
* consumer unloads triggered by removing this provider. Consumers must own
|
|
* cleanup in their injecting scope; concurrent root disposal is not ordered.
|
|
* @param ctx - the owning plugin context.
|
|
* @param config - the storage root, physical encoding, and listing concurrency.
|
|
* @returns resolves when the service is ready.
|
|
*/
|
|
export async function apply(ctx: Context, config: Config): Promise<void> {
|
|
let service!: Fiber
|
|
// Keep this generator synchronous so thenable disposers are registered as functions, not awaited.
|
|
ctx.effect(function* () {
|
|
const tracker = new JsonlBackendTracker(name)
|
|
yield tracker.install(ctx)
|
|
service = ctx.plugin({
|
|
name: 'session-persistence-jsonl-service',
|
|
apply(serviceCtx: Context) {
|
|
new JsonlSessionPersistence(serviceCtx, config, tracker)
|
|
},
|
|
})
|
|
yield service.dispose
|
|
}, 'session-persistence-jsonl resources')
|
|
await service.await()
|
|
}
|