refactor(session): fold the session family into packages/session/

git mv the 12 packages from session-persistence/, session-projection/,
session-title/, and telemetry/ into one session/ group per the
regrouping RFC; merge the four group READMEs into one bilingual
triplet; rewrite the group segment in tsconfig references (intra-group
references shorten to ../<pkg>), tsconfig.base.json paths/globs,
knip.json keys, vitest include, gate scripts, and authored doc/note
citations; regenerate module graph, doc graphs, catalogs, and the
lockfile importer keys. No npm names change.

Full unit suite: 8779 passed; the 18 reported failures reproduce as
env flakes (ambient-proxy IPv6 tunneling, watched-dir inotify
timeouts under parallel load) — each passes in isolation with
NO_PROXY set, matching their known pre-existing behavior on master.

packages/session/session-persistence/src/index.ts

@@ -0,0 +1,203 @@
+/**
+ * Durable session-persistence seam (`ctx.sessionPersistence`). Backends store
+ * {@link SessionEvent}s as the event-sourced log and carry non-replayable
+ * {@link SessionHeader} metadata separately.
+ * @module @deepseek-ai/dsh-session-persistence
+ */
+
+import { Context, Service } from 'cordis'
+import { SessionPreparation } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
+import type { SessionPersistenceRevision } from './revision.ts'
+
+// Re-export the metadata vocabulary so consumers import it from the seam.
+export type { SessionHeader } from '@deepseek-ai/dsh-session'
+export { SessionPersistenceRevision } from './revision.ts'
+
+/** Lightweight immutable source identity returned without loading a full log. */
+export interface SessionPersistenceSnapshot {
+  /** Detached metadata for one materialized session. */
+  header: SessionHeader
+  /** Opaque source-qualified token that changes whenever this stored log changes. */
+  revision: SessionPersistenceRevision
+}
+
+/** Immutable logical session prepared from persistence or a live owner. */
+export interface SessionInspection {
+  /** Validated immutable session metadata. */
+  readonly meta: SessionHeader
+  /** Validated contiguous logical event log. */
+  readonly events: readonly SessionEvent[]
+}
+
+// The backend-agnostic write-path orchestration first-party backends compose.
+export {
+  DEFAULT_PREPARED_SESSION_CACHE_SIZE,
+  DEFAULT_WRITE_BATCH_MAX_DELAY_MS,
+  MAX_WRITE_BATCH_DELAY_MS,
+  PersistenceCoordinator,
+  SessionPersistenceCorruptionError,
+} from './coordinator.ts'
+export type {
+  PersistenceBackend,
+  PersistenceCoordinatorOptions,
+  StoredPrefix,
+  StoredSuffix,
+} from './coordinator.ts'
+
+declare module 'cordis' {
+  interface Context {
+    sessionPersistence: SessionPersistence
+  }
+}
+
+/**
+ * A backend-resolved, per-session local artifact location. The path is an
+ * absolute target path and can name an artifact that has not materialized yet.
+ * Consumers must treat it as a location hint, never as an authorization token.
+ */
+export interface SessionLocation {
+  /** Backend-specific artifact kind, for example `jsonl`. */
+  readonly kind: string
+  /** Absolute path to this session's backend-owned artifact. */
+  readonly path: string
+}
+
+/**
+ * Durable append-only session storage. Implementations preserve contiguous,
+ * losslessly JSON-serializable events; {@link append} resolves only after
+ * durability, and {@link load} balances a complete interrupted tail without
+ * rewriting committed events.
+ */
+export abstract class SessionPersistence extends Service {
+  constructor(ctx: Context) {
+    super(ctx, 'sessionPersistence')
+  }
+
+  /**
+   * Resolve this backend's independent local artifact for a session without
+   * reading, creating, flushing, or otherwise materializing it. Backends such
+   * as SQLite that do not own one artifact per session return `undefined`.
+   * @param meta - the immutable session header whose artifact is requested.
+   * @returns the backend-specific absolute location, when one exists.
+   */
+  abstract locate(meta: SessionHeader): SessionLocation | undefined
+
+  /**
+   * Register a new session's metadata. A backend MAY defer the physical write
+   * until the first {@link append} (lazy materialization), in which case a
+   * created-but-never-appended session is absent from {@link list}
+   * — abandoned sessions leave nothing behind.
+   * @param meta - the immutable header (id, version, cwd, lineage) to record.
+   */
+  abstract create(meta: SessionHeader): Promise<void>
+
+  /**
+   * Durably persist a batch of events. Honors the append-only and contiguous-
+   * seq contracts: the first event's `seq` MUST equal the stored next-seq
+   * (after `load` has durably closed any interrupted turn). Rejects non-JSON-
+   * serializable `event.data` with an error naming the offending event type.
+   * @param id - the session the batch belongs to.
+   * @param events - the contiguous batch to persist, in seq order.
+   */
+  abstract append(id: SessionId, events: readonly SessionEvent[]): Promise<void>
+
+  /**
+   * Prepare the exact unpublished Session used by resume. Implementations may
+   * reuse object graphs retained by an earlier {@link inspect} after confirming
+   * their durable revision is still current; disposal releases an unpublished
+   * reservation. Revision retries require the durable log to remain unchanged
+   * for one read/check round trip; continuous external writers may delay completion.
+   * @param id - persisted session to prepare.
+   * @param signal - optional cancellation for preparation work.
+   * @returns one owned unpublished Session preparation.
+   */
+  async prepare(id: SessionId, signal?: AbortSignal): Promise<SessionPreparation> {
+    signal?.throwIfAborted()
+    const loaded = await this.load(id)
+    signal?.throwIfAborted()
+    const sessions = this.ctx.get('sessions')
+    if (sessions === undefined) {
+      throw new Error('cannot prepare a session: SessionStore is not configured')
+    }
+    return SessionPreparation.create(sessions.prepare(id, {
+      seed: loaded.events.map(event => structuredClone(event)),
+      meta: structuredClone(loaded.meta),
+      seedSource: 'persistence',
+    }))
+  }
+
+  /**
+   * Load an immutable balanced logical view and commit any required cold
+   * recovery. A complete interrupted final turn is preserved and durably
+   * closed with missing tool errors plus any open step and turn boundaries;
+   * only a torn final record is discarded. Unknown versions and corruption in
+   * the committed prefix reject. Implementations MUST NOT crash-repair an
+   * identity still bound to a live Session: a balanced live log may return as a
+   * durable snapshot, while an open live turn rejects. Returned values may be
+   * shared with immutable live or prepared state and must not be mutated.
+   * Revision-based implementations may wait for one stable read/check round trip.
+   * @param id - the persisted session to reload.
+   * @returns the header and a log ending on a balanced `turn/end`.
+   */
+  abstract load(id: SessionId): Promise<SessionInspection>
+
+  /**
+   * Inspect an immutable logical session without committing recovery or
+   * publishing it. A cold complete interrupted turn receives synthetic closers
+   * in memory and a torn physical tail remains untouched. An already-live
+   * Session instead yields its current immutable snapshot, which may contain an
+   * open turn and its `session/end-seed` boundary. Coordinator-backed
+   * implementations retain the exact cold unpublished Session for bounded
+   * reuse by a later {@link prepare}. A stale ready source is reloaded; a source
+   * already committing or reserved for resume remains exclusive, and inspection
+   * may borrow its immutable view. Callers borrow only the immutable header and
+   * log. Continuous external writers may delay revision convergence.
+   * @param id - the persisted session to inspect.
+   * @param signal - optional cancellation for queued and backend read work.
+   * @returns the validated header and current logical event log.
+   */
+  abstract inspect(id: SessionId, signal?: AbortSignal): Promise<SessionInspection>
+
+  /**
+   * Read the stored events from `fromSeq` onward — the read-from-seq
+   * primitive for read models that resume from a watermark (e.g. a persisted
+   * projection cache folding only the tail past its checkpoint). Unlike
+   * {@link inspect}, it is a detached physical suffix read: no preparation
+   * cache, torn-tail truncation, synthetic closers, or coordinator-state
+   * publication. Only events from the valid contiguous stored prefix are
+   * returned, so a torn fragment never reaches the caller. `fromSeq` at or
+   * beyond the stored prefix returns an empty event list (never an error).
+   * Backends whose medium can seek by seq
+   * (SQLite) read only the suffix; sequential media (JSONL, both encodings)
+   * still parse the whole artifact and skip forward — the primitive bounds
+   * what is RETURNED and refolded, not every backend's physical read.
+   * @param id - the persisted session to read.
+   * @param fromSeq - first event seq to include; a non-negative safe integer.
+   * @param signal - optional cancellation for queued and backend read work.
+   * @returns the header and the stored events with `seq >= fromSeq`.
+   */
+  abstract readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal):
+  Promise<{ meta: SessionHeader; events: SessionEvent[] }>
+
+  /**
+   * Lightweight listing from metadata, without a full-log parse.
+   * @param signal - optional cancellation for backend listing work.
+   * @returns one header per materialized session.
+   */
+  abstract list(signal?: AbortSignal): Promise<SessionHeader[]>
+
+  /**
+   * List materialized sessions with cheap per-log change tokens.
+   *
+   * Repeated observations of an unchanged log return the same revision. A
+   * successful mutating {@link load} repair changes the next listed revision.
+   * Revisions also distinguish independently backed stores so backend-local
+   * counters cannot compare equal across different persistence sources.
+   * @param signal - optional cancellation for backend snapshot-listing work.
+   * @returns one header and opaque revision per materialized session without loading full logs.
+   */
+  abstract listSnapshots(signal?: AbortSignal): Promise<SessionPersistenceSnapshot[]>
+}
+
+export default SessionPersistence

packages/session/session-persistence/src/invariant.ts

@@ -0,0 +1,30 @@
+/**
+ * Package-owned invariant companion for `@deepseek-ai/dsh-session-persistence`.
+ * @module @deepseek-ai/dsh-session-persistence/invariant
+ */
+
+/* jscpd:ignore-start */
+import type { Context } from 'cordis'
+import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
+
+const PACKAGE_NAME = '@deepseek-ai/dsh-session-persistence'
+
+/** Cordis companion plugin name. */
+export const name = 'session-persistence-invariant'
+/** Service required before the companion can reserve package ownership. */
+export const inject = ['invariants']
+
+/**
+ * No runtime invariant: persistence correctness requires backend round-trip and crash-tail tests;
+ * this package exposes no continuously observable in-process relation.
+ */
+const install: InvariantInstaller = () => {}
+
+/**
+ * Register this package's invariant companion.
+ * @param ctx - Cordis context carrying the invariant service.
+ * @returns the installed registration's disposer after setup succeeds.
+ */
+export const apply = (ctx: Context): Promise<() => void> =>
+  Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
+/* jscpd:ignore-end */

packages/session/session-persistence/src/preparations.ts

@@ -0,0 +1,348 @@
+/**
+ * Bounded sharing and exclusive reservation of unpublished Sessions.
+ * @module @deepseek-ai/dsh-session-persistence/preparations
+ */
+
+import type { Session, SessionId } from '@deepseek-ai/dsh-session'
+
+interface PreparedSource {
+  readonly session: Session
+}
+
+type PreparationPhase = 'loading' | 'ready' | 'committing' | 'reserved'
+
+interface PreparationEntry<Source, CommitState> {
+  readonly id: SessionId
+  readonly result: Promise<Source>
+  phase: PreparationPhase
+  source?: Source
+  reservation?: SessionPreparationReservation<Source, CommitState>
+  reservationSettled?: Promise<void>
+  settleReservation?: () => void
+}
+
+/** One exclusively held prepared source and its committed persistence state. */
+export interface SessionPreparationReservation<Source, CommitState> {
+  readonly entry: PreparationEntry<Source, CommitState>
+  readonly source: Source
+  readonly state: CommitState
+}
+
+/** Per-coordinator cold-read sharing, exclusive reservation, and ready-entry LRU. */
+export class SessionPreparations<Source extends PreparedSource, CommitState> {
+  private readonly entries = new Map<SessionId, PreparationEntry<Source, CommitState>>()
+
+  constructor(private readonly capacity: number) {}
+
+  /**
+   * Whether this pool currently knows about an unpublished identity.
+   * @param id - session identity.
+   * @returns whether an entry exists for the identity.
+   */
+  has(id: SessionId): boolean {
+    return this.entries.has(id)
+  }
+
+  /**
+   * Observe one prepared source, sharing an in-flight read for the same id.
+   * @param id - session identity.
+   * @param load - cold loader used when no entry exists.
+   * @param signal - optional cancellation signal while waiting.
+   * @returns the shared prepared source.
+   */
+  async inspect(
+    id: SessionId,
+    load: () => Promise<Source>,
+    signal?: AbortSignal,
+  ): Promise<Source> {
+    const entry = this.entryFor(id, load)
+    const loaded = signal === undefined
+      ? await entry.result
+      : await observeQueuedAbort(entry.result, signal)
+    const source = entry.source ?? loaded
+    if (this.entries.get(id) === entry && entry.phase === 'ready') this.touch(entry)
+    return source
+  }
+
+  /**
+   * Reserve one ready source after committing its pending durable repair.
+   * @param id - session identity.
+   * @param load - cold loader used when no entry exists.
+   * @param commit - durable repair and cursor-state commit.
+   * @param signal - optional cancellation signal while waiting.
+   * @returns the exclusive reservation, or undefined if its entry was invalidated.
+   */
+  async reserve(
+    id: SessionId,
+    load: () => Promise<Source>,
+    commit: (source: Source) => Promise<{ source: Source; state: CommitState } | undefined>,
+    signal?: AbortSignal,
+  ): Promise<SessionPreparationReservation<Source, CommitState> | undefined> {
+    const entry = this.entryFor(id, load)
+    await (signal === undefined ? entry.result : observeQueuedAbort(entry.result, signal))
+    while (this.entries.get(id) === entry && entry.phase !== 'ready') {
+      const settled = entry.reservationSettled
+      /* v8 ignore next -- committing/reserved transitions install this waiter synchronously. */
+      if (settled === undefined) throw new Error(`session "${id}" preparation lost its reservation waiter`)
+      if (signal === undefined) await settled
+      else await observeQueuedAbort(settled, signal)
+    }
+    if (this.entries.get(id) !== entry) return undefined
+    const source = entry.source as Source
+    const reservationSettled = Promise.withResolvers<void>()
+    entry.phase = 'committing'
+    entry.reservationSettled = reservationSettled.promise
+    entry.settleReservation = reservationSettled.resolve
+    let committed: { source: Source; state: CommitState } | undefined
+    try {
+      committed = await commit(source)
+    } catch (error: unknown) {
+      this.remove(entry)
+      throw error
+    }
+    if (committed === undefined) {
+      this.remove(entry)
+      return undefined
+    }
+    entry.source = committed.source
+    try {
+      signal?.throwIfAborted()
+    } catch (error: unknown) {
+      this.makeReady(entry)
+      throw error
+    }
+    if (this.entries.get(id) !== entry) return undefined
+    const reservation: SessionPreparationReservation<Source, CommitState> = {
+      entry,
+      source: committed.source,
+      state: committed.state,
+    }
+    entry.phase = 'reserved'
+    entry.reservation = reservation
+    return reservation
+  }
+
+  /**
+   * Return the exact reservation for Session publication, rejecting aliases.
+   * @param session - exact Session candidate for publication.
+   * @returns its reservation, or undefined when no preparation exists.
+   */
+  reservationFor(session: Session): SessionPreparationReservation<Source, CommitState> | undefined {
+    const entry = this.entries.get(session.id)
+    if (entry === undefined) return undefined
+    if (entry.phase === 'reserved'
+      && entry.source?.session === session
+      && entry.reservation !== undefined) {
+      return entry.reservation
+    }
+    throw new Error(`cannot publish session "${session.id}": persisted state already owns this identity`)
+  }
+
+  /**
+   * Consume a reservation after its exact Session has attached.
+   * @param reservation - reservation to consume.
+   */
+  attach(reservation: SessionPreparationReservation<Source, CommitState>): void {
+    const { entry } = reservation
+    if (this.entries.get(entry.id) !== entry || entry.reservation !== reservation) {
+      throw new Error(`session "${entry.id}" preparation is no longer reserved`)
+    }
+    this.remove(entry)
+  }
+
+  /**
+   * Consume a reservation whose caller only needs the committed inspection.
+   * @param reservation - reservation to consume.
+   */
+  discard(reservation: SessionPreparationReservation<Source, CommitState>): void {
+    const { entry } = reservation
+    if (this.entries.get(entry.id) !== entry || entry.reservation !== reservation) return
+    this.remove(entry)
+  }
+
+  /**
+   * Return a reusable unpublished reservation to the ready LRU.
+   * @param reservation - reservation to release.
+   * @param reusable - whether the source remains valid for reuse.
+   */
+  release(
+    reservation: SessionPreparationReservation<Source, CommitState>,
+    reusable: boolean,
+  ): void {
+    const { entry } = reservation
+    if (this.entries.get(entry.id) !== entry
+      || entry.reservation !== reservation
+      || entry.phase !== 'reserved') return
+    if (!reusable) {
+      this.remove(entry)
+      return
+    }
+    delete entry.reservation
+    this.makeReady(entry)
+  }
+
+  /**
+   * Discard a prepared view after the durable log changes.
+   * @param id - changed session identity.
+   */
+  invalidate(id: SessionId): void {
+    const entry = this.entries.get(id)
+    if (entry !== undefined) this.remove(entry)
+  }
+
+  /**
+   * Discard an exact stale ready source without disturbing an exclusive owner.
+   * @param id - changed session identity.
+   * @param expected - exact source observed before its revision check.
+   * @returns whether the source was discarded, retained by a reservation, or is absent.
+   */
+  discardReady(id: SessionId, expected: Source): 'discarded' | 'retained' | 'missing' {
+    const entry = this.entries.get(id)
+    if (entry === undefined || entry.source !== expected) return 'missing'
+    if (entry.phase !== 'ready') return 'retained'
+    this.remove(entry)
+    return 'discarded'
+  }
+
+  /**
+   * Reject writes while an unpublished Session exclusively reserves the id.
+   * @param id - session identity to check.
+   */
+  assertWritable(id: SessionId): void {
+    const phase = this.entries.get(id)?.phase
+    if (phase === 'committing' || phase === 'reserved') {
+      throw new Error(`cannot append session "${id}" while its persisted preparation is reserved`)
+    }
+  }
+
+  /**
+   * Remove a completed entry for an already-serialized append adoption.
+   * @param id - adopted session identity.
+   * @returns the prepared source, or undefined when no ready entry exists.
+   */
+  takeReady(id: SessionId): Source | undefined {
+    const entry = this.entries.get(id)
+    if (entry === undefined || entry.phase !== 'ready' || entry.source === undefined) return undefined
+    this.remove(entry)
+    return entry.source
+  }
+
+  private entryFor(
+    id: SessionId,
+    load: () => Promise<Source>,
+  ): PreparationEntry<Source, CommitState> {
+    const existing = this.entries.get(id)
+    if (existing !== undefined) return existing
+    const deferred = Promise.withResolvers<Source>()
+    const entry: PreparationEntry<Source, CommitState> = {
+      id,
+      result: deferred.promise,
+      phase: 'loading',
+    }
+    this.entries.set(id, entry)
+    let loading: Promise<Source>
+    try {
+      // Start immediately so a same-tick serialized append queues behind this
+      // read. The deferred result settles only after the entry becomes ready.
+      loading = load()
+    } catch (error: unknown) {
+      this.remove(entry)
+      deferred.reject(error)
+      return entry
+    }
+    void loading.then((source) => {
+      if (this.entries.get(id) === entry) {
+        entry.source = source
+        this.makeReady(entry)
+      }
+      deferred.resolve(source)
+    }, (error: unknown) => {
+      this.remove(entry)
+      deferred.reject(error)
+    })
+    return entry
+  }
+
+  private makeReady(entry: PreparationEntry<Source, CommitState>): void {
+    if (this.entries.get(entry.id) !== entry) return
+    entry.phase = 'ready'
+    const settle = entry.settleReservation
+    delete entry.reservationSettled
+    delete entry.settleReservation
+    settle?.()
+    this.touch(entry)
+  }
+
+  private remove(entry: PreparationEntry<Source, CommitState>): void {
+    if (this.entries.get(entry.id) !== entry) return
+    this.entries.delete(entry.id)
+    const settle = entry.settleReservation
+    delete entry.reservationSettled
+    delete entry.settleReservation
+    settle?.()
+  }
+
+  private touch(entry: PreparationEntry<Source, CommitState>): void {
+    this.entries.delete(entry.id)
+    this.entries.set(entry.id, entry)
+    let readyCount = 0
+    for (const candidate of this.entries.values()) {
+      if (candidate.phase === 'ready') readyCount += 1
+    }
+    if (readyCount <= this.capacity) return
+    for (const [id, candidate] of this.entries) {
+      if (candidate.phase !== 'ready') continue
+      this.entries.delete(id)
+      return
+    }
+  }
+}
+
+/**
+ * Give a queued observer a prompt cancellation view without cancelling shared work.
+ * @param operation - shared operation whose settlement remains authoritative.
+ * @param signal - observer-local cancellation signal.
+ * @param started - whether the operation has crossed its cancellation cutoff.
+ * @returns the operation result or the observer's prompt cancellation.
+ */
+export function observeQueuedAbort<T>(
+  operation: Promise<T>,
+  signal: AbortSignal,
+  started: () => boolean = () => false,
+): Promise<T> {
+  return new Promise<T>((resolve, reject) => {
+    let settled = false
+    const finish = (callback: () => void): void => {
+      if (settled) return
+      settled = true
+      signal.removeEventListener('abort', onAbort)
+      callback()
+    }
+    const onAbort = (): void => {
+      if (started()) return
+      finish(() => {
+        try {
+          signal.throwIfAborted()
+        } catch (reason: unknown) {
+          rejectObservation(reject, reason)
+          return
+        }
+        /* v8 ignore next -- a native AbortSignal emits abort only after becoming aborted. */
+        reject(new Error('queued observation abort event lacked an aborted signal'))
+      })
+    }
+    signal.addEventListener('abort', onAbort, { once: true })
+    operation.then(
+      (value) => { finish(() => { resolve(value) }) },
+      (reason: unknown) => {
+        finish(() => { rejectObservation(reject, reason) })
+      },
+    )
+    if (signal.aborted) onAbort()
+  })
+}
+
+/** Preserve an exact loader or AbortSignal reason, including legacy non-Error values. */
+function rejectObservation(reject: (reason?: unknown) => void, reason: unknown): void {
+  reject(reason)
+}

packages/session/session-persistence/src/revision.ts

@@ -0,0 +1,18 @@
+/** Opaque revision identity for lightweight persistence observations. */
+
+import type { Branded } from '@deepseek-ai/dsh-brand'
+
+/**
+ * Backend-owned token that identifies both one storage source and one revision
+ * of a persisted session log.
+ */
+export type SessionPersistenceRevision = Branded<'SessionPersistenceRevision'>
+
+/**
+ * Brand a backend revision for the provider-neutral persistence contract.
+ * @param value - backend-owned opaque revision representation.
+ * @returns the same runtime string with persistence-revision identity.
+ */
+export function SessionPersistenceRevision(value: string): SessionPersistenceRevision {
+  return value as SessionPersistenceRevision
+}

packages/session/session-persistence/src/write-behind.ts

@@ -0,0 +1,159 @@
+/**
+ * Bounded per-session write batching for the shared persistence coordinator.
+ * @module @deepseek-ai/dsh-session-persistence/write-behind
+ */
+
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
+
+/** Dependencies and scheduling policy for one live session's write controller. */
+export interface SessionWriteBehindOptions {
+  /** Maximum intentional batching wait after an idle queue receives work. */
+  readonly maxDelayMs: number
+  /** Persist one stable ordered prefix; resolves only after backend durability. */
+  readonly write: (events: readonly SessionEvent[]) => Promise<void>
+  /** Observe a detached background write failure without rejecting the producer. */
+  readonly reportBackgroundFailure: (error: unknown) => void
+}
+
+/**
+ * Owns one live session's pending events, fixed batching deadline, active write,
+ * failure retention, and explicit quiescence barrier.
+ */
+export class SessionWriteBehind {
+  private pending: SessionEvent[] = []
+  private timer: ReturnType<typeof setTimeout> | undefined
+  private active: Promise<void> | undefined
+  private barrier: Promise<void> | undefined
+  private deadlineExpired = false
+  private automaticPaused = false
+
+  /**
+   * @param options - fixed scheduling policy and durable batch sink.
+   */
+  constructor(private readonly options: SessionWriteBehindOptions) {}
+
+  /** Whether this controller owns queued events or an active durable write. */
+  get hasWork(): boolean {
+    return this.pending.length > 0 || this.active !== undefined
+  }
+
+  /**
+   * Copy one event into the persistence-owned queue and start a fixed deadline
+   * when the automatic path is idle.
+   * @param event - frozen live event to retain independently of its producer.
+   */
+  enqueue(event: SessionEvent): void {
+    const wasEmpty = this.pending.length === 0
+    this.pending.push(structuredClone(event))
+    if (this.barrier !== undefined) return
+    if (this.automaticPaused) {
+      this.automaticPaused = false
+      this.deadlineExpired = false
+      this.armTimer()
+    } else if (wasEmpty) {
+      this.armTimer()
+    }
+  }
+
+  /**
+   * Cancel the batching wait and durably drain through a quiescent point.
+   * Concurrent callers join the same barrier.
+   * @returns a promise that rejects if the barrier's durable retry fails.
+   */
+  flush(): Promise<void> {
+    if (this.barrier !== undefined) return this.barrier
+    this.cancelTimer()
+    this.deadlineExpired = false
+    this.automaticPaused = false
+    const barrier = Promise.withResolvers<void>()
+    this.barrier = barrier.promise
+    void this.drainBarrier(barrier.resolve, barrier.reject)
+    return barrier.promise
+  }
+
+  /** Cancel the current automatic deadline without draining retained work. */
+  cancelAutomaticWait(): void {
+    this.cancelTimer()
+    this.deadlineExpired = false
+  }
+
+  /** Start the one fixed window for the current pending prefix. */
+  private armTimer(): void {
+    this.timer = setTimeout(() => { this.onDeadline() }, this.options.maxDelayMs)
+  }
+
+  /** Cancel any pending automatic deadline. */
+  private cancelTimer(): void {
+    if (this.timer === undefined) return
+    clearTimeout(this.timer)
+    this.timer = undefined
+  }
+
+  /** Start a background write now, or remember that an active write used the budget. */
+  private onDeadline(): void {
+    this.timer = undefined
+    if (this.active !== undefined) {
+      this.deadlineExpired = true
+      return
+    }
+    this.startBackground()
+  }
+
+  /** Start one detached write whose failure is reported and retained. */
+  private startBackground(): void {
+    const active = this.startWrite(true)
+    void active.then(() => { this.continueAutomatic() }, () => {})
+  }
+
+  /** Continue immediately after an over-budget active write, otherwise keep its timer. */
+  private continueAutomatic(): void {
+    if (this.barrier !== undefined || this.pending.length === 0) return
+    if (this.deadlineExpired) {
+      this.deadlineExpired = false
+      this.startBackground()
+    }
+  }
+
+  /** Await overlapping work, drain to quiescence, and settle the shared barrier. */
+  private async drainBarrier(resolve: () => void, reject: (reason?: unknown) => void): Promise<void> {
+    try {
+      const overlapping = this.active
+      if (overlapping !== undefined) {
+        await Promise.allSettled([overlapping])
+        this.automaticPaused = false
+      }
+      while (this.pending.length > 0) await this.startWrite(false)
+    } catch (error: unknown) {
+      this.barrier = undefined
+      reject(error)
+      return
+    }
+    // Close admission to this barrier in the same job that observes the empty
+    // queue, before resolving callers. A later enqueue therefore starts its own
+    // automatic window instead of being stranded behind a settled barrier.
+    this.barrier = undefined
+    resolve()
+  }
+
+  /** Start one stable pending prefix, retaining it in order if durability fails. */
+  private startWrite(background: boolean): Promise<void> {
+    const batch = this.pending.splice(0)
+    this.cancelTimer()
+    this.deadlineExpired = false
+    const operation = Promise.resolve().then(() => this.options.write(batch))
+    const active = operation
+      .catch((error: unknown) => {
+        this.pending = batch.concat(this.pending)
+        this.cancelTimer()
+        this.deadlineExpired = false
+        this.automaticPaused = true
+        if (background) this.options.reportBackgroundFailure(error)
+        throw error
+      })
+      .finally(() => {
+        this.active = undefined
+      })
+    this.active = active
+    return active
+  }
+}