deepseek-harness/packages/session/session-persistence-jsonl/src/index.ts

/**
 * 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 } from '@deepseek-ai/cordis'
import z from '@deepseek-ai/schemastery'
import {
  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 { randomBytes } from 'node:crypto'
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 {
  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)

/** Plugin config for the JSONL backend's root and physical encoding. */
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
}

/** 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 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<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'
}

/** 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 persistence backend. Load as a plugin; it registers as
 * `ctx.sessionPersistence`. 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 {
  static Config: z<Config> = z.object({
    root: z.string().required(),
    compression: JsonlCompressionSchema,
  })

  /** 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
  private rootEncodingCheck: Promise<void> | undefined
  private readonly tracker = new JsonlBackendTracker(this.name)
  private readonly generationFormat: JsonlGenerationFormatAdapter
  /**
   * 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) {
    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.generationFormat = {
      currentVersion: sessionFormatCatalog.currentVersion,
      createRestore: header => sessionFormatCatalog.createRestore(header, {
        recovery: 'recoverable',
        validation: 'transformed',
      }),
      encodeHeader: (header, inheritedEventCount) =>
        sessionFormatCatalog.encodeCurrentHeader(header, inheritedEventCount),
      encodeEvent: event => sessionFormatCatalog.encodeCurrentEvent(event),
      isUnsupportedMigrationError: (error): error is SessionFormatUnsupportedMigrationError =>
        error instanceof SessionFormatUnsupportedMigrationError,
    }
    this.assertUsableRoot()
    this.tracker.install(ctx)
  }

  /**
   * 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 }))
      }
      const 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 header = await this.readGenerationHeader(selected, id, options?.signal)
    if (header === undefined) return undefined
    try {
      const identity = await stat(selected.sourcePath, { bigint: true })
      options?.signal?.throwIfAborted()
      return {
        header,
        revision: 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()]
    for (const artifact of await this.listArtifacts(signal)) {
      signal?.throwIfAborted()
      try {
        const identity = await stat(artifact.path, { bigint: true })
        signal?.throwIfAborted()
        listed.add(artifact.header.id)
        snapshots.push({
          header: artifact.header,
          revision: fileRevision(identity),
          sizeBytes: Number(identity.size),
        })
      } catch (error: unknown) {
        signal?.throwIfAborted()
        if (!isENOENT(error)) throw error
      }
    }
    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) {
      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>>
    try {
      prepared = await prepareJsonlMigration({
        sourcePath: selected.sourcePath,
        sourceVersion: selected.sourceVersion,
        currentPath: selected.currentPath,
        compression: this.compression,
        format: this.generationFormat,
        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',
      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()
    }
  }

  private async listArtifacts(signal?: AbortSignal): Promise<Array<{ header: SessionHeader; path: string }>> {
    signal?.throwIfAborted()
    await this.ensureRootEncoding()
    signal?.throwIfAborted()
    const artifacts: Array<{ header: SessionHeader; path: string }> = []
    const ids = new Set<SessionId>()
    for (const project of await this.listProjectDirs(signal)) {
      signal?.throwIfAborted()
      for (const dir of await this.listSessionDirs(project, signal)) {
        signal?.throwIfAborted()
        const selected = await this.resolveGenerationInDirectory(dir, signal)
        if (selected === undefined) continue
        let header: SessionHeader | undefined
        try {
          header = await this.readGenerationHeader(selected, undefined, signal)
        } catch (error: unknown) {
          // Listing skips a foreign format while opening its id still refuses
          // with the selected physical location.
          if (error instanceof SessionFormatUnsupportedError) continue
          throw error
        }
        if (header === undefined) continue
        if (ids.has(header.id)) {
          throw new Error(`duplicate JSONL session id "${header.id}" appears in multiple project directories`)
        }
        ids.add(header.id)
        artifacts.push({ header, path: selected.sourcePath })
      }
    }
    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<SessionHeader | 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
  }

  /** 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 and validate only the independently compressed header frame. */
  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()
        const first = scanZstdFrames(content, 1).frames[0]
        signal?.throwIfAborted()
        if (first === undefined) continue
        let plaintext: Buffer
        try {
          signal?.throwIfAborted()
          plaintext = await decompressZstdFrame(content.subarray(first.start, first.end))
        } catch (error) {
          /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */
          if (signal?.aborted) signal.throwIfAborted()
          throw new Error('corrupt Zstandard session log: header frame failed validation', { cause: error })
        }
        signal?.throwIfAborted()
        assertZstdHeaderFrame(plaintext)
        return plaintext.subarray(0, -1).toString('utf8')
      }
    } 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> {
    for (const project of await this.listProjectDirs()) {
      for (const dir of await this.listSessionDirs(project)) {
        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 */
}

/**
 * One open channel onto a JSONL-stored session: the shared storage-handle
 * scaffolding over this backend's file primitives. Reads re-scan the artifact
 * under the stable-read loop.
 */

export default JsonlSessionPersistence