refactor(session-persistence): share stored read machinery

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

@@ -106,20 +106,6 @@ interface JsonlStoredHeader {
   readonly revision: PersistenceRevision
 }
 
-function deferred<T>(): {
-  readonly promise: Promise<T>
-  resolve(value: T): void
-  reject(reason: unknown): void
-} {
-  let resolve!: (value: T) => void
-  let reject!: (reason: unknown) => void
-  const promise = new Promise<T>((accept, decline) => {
-    resolve = accept
-    reject = decline
-  })
-  return { promise, resolve, reject }
-}
-
 interface FileRevisionIdentity {
   readonly dev: bigint
   readonly ino: bigint
@@ -254,28 +240,23 @@ export class JsonlSessionPersistence extends SessionPersistence implements Persi
       revision,
       location: { kind: 'jsonl', path },
       readEvents: (options = {}): StoredEventRead<JsonlTornMarker> => {
-        const completed = deferred<{ tornMarker?: JsonlTornMarker }>()
-        const events = (async function* (backend: JsonlSessionPersistence): AsyncIterable<unknown> {
-          try {
-            const prefix = await backend.readPrefix(path, id, signal)
+        const fromSeq = options.fromSeq ?? 0
+        return this.createStoredEventRead(
+          async () => {
+            const prefix = await this.readPrefix(path, id, signal)
             if (prefix.revision !== revision) {
               throw new SessionPersistenceRevisionConflictError(
                 `session "${id}" changed while reading revision ${revision}`,
               )
             }
-            const fromSeq = options.fromSeq ?? 0
-            for (const event of prefix.events) {
-              signal?.throwIfAborted()
-              const seq = (event as { seq?: unknown }).seq
-              if (typeof seq !== 'number' || seq >= fromSeq) yield event
-            }
-            completed.resolve(prefix.tornMarker === undefined ? {} : { tornMarker: prefix.tornMarker })
-          } catch (error: unknown) {
-            completed.reject(error)
-            throw error
-          }
-        })(this)
-        return { events, completed: completed.promise }
+            return prefix
+          },
+          (event) => {
+            const seq = (event as { seq?: unknown }).seq
+            return typeof seq !== 'number' || seq >= fromSeq
+          },
+          signal,
+        )
       },
     }
   }

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

@@ -59,20 +59,6 @@ interface SqliteStoredPrefix {
   readonly tornMarker?: number
 }
 
-function deferred<T>(): {
-  readonly promise: Promise<T>
-  resolve(value: T): void
-  reject(reason: unknown): void
-} {
-  let resolve!: (value: T) => void
-  let reject!: (reason: unknown) => void
-  const promise = new Promise<T>((accept, decline) => {
-    resolve = accept
-    reject = decline
-  })
-  return { promise, resolve, reject }
-}
-
 /**
  * Exclusively create a missing database file with owner-only permissions.
  * Existing files retain their modes, and errors other than `EEXIST` propagate.
@@ -238,29 +224,22 @@ export class SqliteSessionPersistence extends SessionPersistence implements Pers
       meta: rowToStoredMeta(row),
       revision,
       readEvents: (options = {}): StoredEventRead<number> => {
-        const completed = deferred<{ tornMarker?: number }>()
-        const events = (async function* (backend: SqliteSessionPersistence): AsyncIterable<unknown> {
-          try {
+        return this.createStoredEventRead(
+          async () => {
             const fromSeq = options.fromSeq ?? 0
             const stored = fromSeq === 0
-              ? await backend.readPrefix(id, signal)
-              : await backend.readSuffix(id, fromSeq, signal)
+              ? await this.readPrefix(id, signal)
+              : await this.readSuffix(id, fromSeq, signal)
             if (stored === undefined || stored.revision !== revision) {
               throw new SessionPersistenceRevisionConflictError(
                 `session "${id}" changed while reading revision ${revision}`,
               )
             }
-            for (const event of stored.events) {
-              signal?.throwIfAborted()
-              yield event
-            }
-            completed.resolve(stored.tornMarker === undefined ? {} : { tornMarker: stored.tornMarker })
-          } catch (error: unknown) {
-            completed.reject(error)
-            throw error
-          }
-        })(this)
-        return { events, completed: completed.promise }
+            return stored
+          },
+          () => true,
+          signal,
+        )
       },
     }
   }
@@ -280,33 +259,7 @@ export class SqliteSessionPersistence extends SessionPersistence implements Pers
    * region are dropped, never repaired (non-mutating read).
    */
   private async readSuffix(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<SqliteStoredPrefix | undefined> {
-    signal?.throwIfAborted()
-    await this.ready
-    signal?.throwIfAborted()
-    this.db.exec('BEGIN')
-    let snapshot: { row: SessionRow; eventRows: EventRow[] } | undefined
-    try {
-      const row = this.rowFor(id)
-      if (row !== undefined) {
-        const eventRows = this.db
-          .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op, ignorable FROM events WHERE session_id = ? AND seq >= ? ORDER BY seq')
-          .all(id, fromSeq) as unknown as EventRow[]
-        snapshot = { row, eventRows }
-      }
-      this.db.exec('COMMIT')
-    } catch (error: unknown) {
-      this.db.exec('ROLLBACK')
-      throw error
-    }
-    signal?.throwIfAborted()
-    if (snapshot === undefined) return undefined
-    const { row, eventRows } = snapshot
-    const { preserved } = scanRows(eventRows, fromSeq)
-    return {
-      meta: rowToStoredMeta(row),
-      events: preserved,
-      revision: sqliteRevision(this.storeIdentity, row),
-    }
+    return this.readStoredEvents(id, fromSeq, false, signal)
   }
 
   /**
@@ -315,6 +268,15 @@ export class SqliteSessionPersistence extends SessionPersistence implements Pers
    * (`scanRows` already returns it as `number | undefined`).
    */
   private async readPrefix(id: SessionId, signal?: AbortSignal): Promise<SqliteStoredPrefix | undefined> {
+    return this.readStoredEvents(id, 0, true, signal)
+  }
+
+  private async readStoredEvents(
+    id: SessionId,
+    fromSeq: number,
+    includeTornMarker: boolean,
+    signal?: AbortSignal,
+  ): Promise<SqliteStoredPrefix | undefined> {
     signal?.throwIfAborted()
     await this.ready
     signal?.throwIfAborted()
@@ -323,27 +285,28 @@ export class SqliteSessionPersistence extends SessionPersistence implements Pers
     try {
       const row = this.rowFor(id)
       if (row !== undefined) {
-        const eventRows = this.db
-          .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op, ignorable FROM events WHERE session_id = ? ORDER BY seq')
-          .all(id) as unknown as EventRow[]
+        const statement = fromSeq === 0
+          ? this.db.prepare('SELECT seq, type, time, data, source_event_seqs, surface_op, ignorable FROM events WHERE session_id = ? ORDER BY seq')
+          : this.db.prepare('SELECT seq, type, time, data, source_event_seqs, surface_op, ignorable FROM events WHERE session_id = ? AND seq >= ? ORDER BY seq')
+        const eventRows = (fromSeq === 0
+          ? statement.all(id)
+          : statement.all(id, fromSeq)) as unknown as EventRow[]
         snapshot = { row, eventRows }
       }
       this.db.exec('COMMIT')
     } catch (error: unknown) {
-      /* v8 ignore start -- synchronous read failures only need transaction cleanup before propagation. */
       this.db.exec('ROLLBACK')
       throw error
-      /* v8 ignore stop */
     }
     signal?.throwIfAborted()
     if (snapshot === undefined) return undefined
     const { row, eventRows } = snapshot
-    const { preserved, tornFrom } = scanRows(eventRows)
+    const { preserved, tornFrom } = scanRows(eventRows, fromSeq)
     return {
       meta: rowToStoredMeta(row),
       events: preserved,
       revision: sqliteRevision(this.storeIdentity, row),
-      ...tornFrom !== undefined ? { tornMarker: tornFrom } : {},
+      ...includeTornMarker && tornFrom !== undefined ? { tornMarker: tornFrom } : {},
     }
   }
 

packages/session/session-persistence-sqlite/src/schema.ts

@@ -10,7 +10,7 @@
 
 import { randomUUID } from 'node:crypto'
 import { DatabaseSync } from 'node:sqlite'
-import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionHeader, SurfaceOp } from '@deepseek-ai/dsh-session'
 
 /**
  * The on-disk schema version. Bumped only on a breaking change to the table
@@ -180,17 +180,7 @@ export function rowToMeta(row: SessionRow): SessionHeader {
   if (!Number.isSafeInteger(row.created_at) || row.created_at < 0) {
     throw new Error('stored session createdAt must be a non-negative safe integer')
   }
-  return {
-    version: row.version,
-    id: row.id as SessionId,
-    createdAt: row.created_at,
-    ...row.cwd !== null ? { cwd: row.cwd } : {},
-    ...row.parent_session !== null ? { parentSession: row.parent_session as SessionId } : {},
-    ...row.seed_length !== null ? { seedLength: row.seed_length } : {},
-    ...row.origin !== null ? { origin: row.origin } : {},
-    ...row.delegation_depth !== null ? { delegationDepth: row.delegation_depth } : {},
-    ...row.agent_preset !== null ? { agentPreset: row.agent_preset } : {},
-  }
+  return rowToStoredMeta(row) as SessionHeader
 }
 
 /**

packages/session/session-persistence/src/format-decoder.ts

@@ -17,6 +17,7 @@ import {
   unversionedFormatCompatibility,
 } from './format-v0-compat.ts'
 import type { UnversionedFormatCompatibility } from './format-v0-compat.ts'
+import { asStoredRecord, readStoredEventEnvelope } from './format-json.ts'
 import type { SessionLocation } from './index.ts'
 import { SESSION_FORMAT_STEPS } from './format-migrations/index.ts'
 import type { SessionPersistenceRevision } from './revision.ts'
@@ -163,12 +164,6 @@ interface DecodedHeader {
   readonly unversionedCompatibility?: UnversionedFormatCompatibility
 }
 
-function asRecord(value: unknown): Record<string, unknown> | undefined {
-  return typeof value === 'object' && value !== null && !Array.isArray(value)
-    ? value as Record<string, unknown>
-    : undefined
-}
-
 interface StoredHeaderSource {
   readonly meta: unknown
   readonly location?: SessionLocation
@@ -190,7 +185,7 @@ function readSourceHeader(
   expectedId: SessionId,
 ): { meta: Record<string, unknown>; version: number; id: SessionId } {
   const snapshot = snapshotJsonValue(source.meta)
-  const meta = asRecord(snapshot)
+  const meta = asStoredRecord(snapshot)
   if (meta === undefined) throw new Error('stored session header is not a lossless JSON record')
   if (!Number.isSafeInteger(meta['version'])) {
     throw new Error(`stored session header has invalid format version ${String(meta['version'])}`)
@@ -243,7 +238,7 @@ function decodeHeader(
         { cause: error },
       )
     }
-    const record = asRecord(meta)
+    const record = asStoredRecord(meta)
     const actual = record?.['version']
     if (actual !== step.to) {
       throw new Error(`Session format step v${step.from} -> v${step.to} returned header version ${String(actual)}`)
@@ -282,19 +277,7 @@ export function decodeStoredSessionHeader(
 
 function assertCurrentEnvelope(value: unknown, id: SessionId): SessionEvent {
   const snapshot = snapshotJsonValue(value)
-  const event = asRecord(snapshot)
-  if (event === undefined) throw new Error(`session "${id}" contains a non-record event`)
-  if (typeof event['type'] !== 'string') throw new Error(`session "${id}" contains an event without a string type`)
-  if (!Number.isSafeInteger(event['seq']) || (event['seq'] as number) < 0) {
-    throw new Error(`session "${id}" contains event type "${event['type']}" with invalid seq ${String(event['seq'])}`)
-  }
-  if (typeof event['time'] !== 'number' || !Number.isFinite(event['time'])) {
-    throw new Error(`session "${id}" contains event type "${event['type']}" at seq ${String(event['seq'])} with invalid time`)
-  }
-  if (!Object.hasOwn(event, 'data')) {
-    throw new Error(`session "${id}" contains event type "${event['type']}" at seq ${String(event['seq'])} without data`)
-  }
-  return event as unknown as SessionEvent
+  return readStoredEventEnvelope(snapshot, id)
 }
 
 function assertCurrentEventSupported<TornMarker>(
@@ -376,20 +359,6 @@ async function* snapshotStoredEvents(
   }
 }
 
-function deferred<T>(): {
-  readonly promise: Promise<T>
-  resolve(value: T): void
-  reject(reason: unknown): void
-} {
-  let resolve!: (value: T) => void
-  let reject!: (reason: unknown) => void
-  const promise = new Promise<T>((accept, decline) => {
-    resolve = accept
-    reject = decline
-  })
-  return { promise, resolve, reject }
-}
-
 function decodedRead<TornMarker>(
   source: StoredSessionSource<TornMarker>,
   header: DecodedHeader,
@@ -398,7 +367,7 @@ function decodedRead<TornMarker>(
   readonly events: AsyncIterable<SessionEvent>
   readonly completed: Promise<StoredEventReadCompletion<TornMarker>>
 } {
-  const completion = deferred<StoredEventReadCompletion<TornMarker>>()
+  const completion = Promise.withResolvers<StoredEventReadCompletion<TornMarker>>()
   const migrating = header.steps.length > 0
   const compatibility = header.unversionedCompatibility
   let physical: StoredEventRead<TornMarker> | undefined

packages/session/session-persistence/src/format-json.ts

@@ -0,0 +1,36 @@
+/** Shared JSON validation for stored Session format records. */
+
+import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
+
+/**
+ * Narrow an unknown JSON value to a non-array object.
+ * @param value - parsed JSON value.
+ * @returns the object, or `undefined` for every other JSON value.
+ */
+export function asStoredRecord(value: unknown): Record<string, unknown> | undefined {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+    ? value as Record<string, unknown>
+    : undefined
+}
+
+/**
+ * Validate fields common to every stored Session event envelope.
+ * @param value - detached parsed event JSON.
+ * @param id - Session identity used in diagnostics.
+ * @returns the structurally valid event envelope.
+ */
+export function readStoredEventEnvelope(value: unknown, id: SessionId): SessionEvent {
+  const event = asStoredRecord(value)
+  if (event === undefined) throw new Error(`session "${id}" contains a non-record event`)
+  if (typeof event['type'] !== 'string') throw new Error(`session "${id}" contains an event without a string type`)
+  if (!Number.isSafeInteger(event['seq']) || (event['seq'] as number) < 0) {
+    throw new Error(`session "${id}" contains event type "${event['type']}" with invalid seq ${String(event['seq'])}`)
+  }
+  if (typeof event['time'] !== 'number' || !Number.isFinite(event['time'])) {
+    throw new Error(`session "${id}" contains event type "${event['type']}" at seq ${String(event['seq'])} with invalid time`)
+  }
+  if (!Object.hasOwn(event, 'data')) {
+    throw new Error(`session "${id}" contains event type "${event['type']}" at seq ${String(event['seq'])} without data`)
+  }
+  return event as unknown as SessionEvent
+}

packages/session/session-persistence/src/format-v0-compat.ts

@@ -4,6 +4,7 @@
  */
 
 import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
+import { asStoredRecord, readStoredEventEnvelope } from './format-json.ts'
 
 /** One format-specific normalizer selected before adjacent-version migrations. */
 export interface UnversionedFormatCompatibility {
@@ -25,12 +26,6 @@ export interface UnversionedFormatCompatibility {
   canonicalizeEvents(events: AsyncIterable<unknown>, sessionId: SessionId): AsyncIterable<unknown>
 }
 
-function asRecord(value: unknown): Record<string, unknown> | undefined {
-  return typeof value === 'object' && value !== null && !Array.isArray(value)
-    ? value as Record<string, unknown>
-    : undefined
-}
-
 function hasOnlyKeys(
   record: Record<string, unknown>,
   required: readonly string[],
@@ -48,16 +43,16 @@ function legacyMessageId(id: SessionId, seq: number): PersistedMessageId {
 }
 
 function replacementStart(event: SessionEvent): number | undefined {
-  const op = asRecord((event as SessionEvent & { surfaceOp?: unknown }).surfaceOp)
+  const op = asStoredRecord((event as SessionEvent & { surfaceOp?: unknown }).surfaceOp)
   return op?.['op'] === 'replace' && typeof op['start'] === 'number'
     ? op['start']
     : undefined
 }
 
 function requiresV0Prefix(value: unknown): boolean {
-  const event = asRecord(value)
+  const event = asStoredRecord(value)
   if (event === undefined) return false
-  const data = asRecord(event['data'])
+  const data = asStoredRecord(event['data'])
   if (event['type'] === 'steering/message') return true
   if (data === undefined) return false
   switch (event['type']) {
@@ -73,29 +68,17 @@ function requiresV0Prefix(value: unknown): boolean {
 }
 
 function readV0Event(value: unknown, id: SessionId): SessionEvent {
-  const event = asRecord(value)
-  if (event === undefined) throw new Error(`session "${id}" contains a non-record event`)
-  if (typeof event['type'] !== 'string') throw new Error(`session "${id}" contains an event without a string type`)
-  if (!Number.isSafeInteger(event['seq']) || (event['seq'] as number) < 0) {
-    throw new Error(`session "${id}" contains event type "${event['type']}" with invalid seq ${String(event['seq'])}`)
-  }
-  if (typeof event['time'] !== 'number' || !Number.isFinite(event['time'])) {
-    throw new Error(`session "${id}" contains event type "${event['type']}" at seq ${String(event['seq'])} with invalid time`)
-  }
-  if (!Object.hasOwn(event, 'data')) {
-    throw new Error(`session "${id}" contains event type "${event['type']}" at seq ${String(event['seq'])} without data`)
-  }
-  return event as unknown as SessionEvent
+  return readStoredEventEnvelope(value, id)
 }
 
 function canonicalizeLegacySteeringEvent(event: SessionEvent, id: SessionId): SessionEvent {
   const legacyType: string = 'steering/message'
   if (event.type !== legacyType) return event
-  const data = asRecord(event.data)
+  const data = asStoredRecord(event.data)
   if (data === undefined) {
     throw new Error(`session "${id}" contains malformed pre-react-loop steering/message at seq ${event.seq}`)
   }
-  const wrapped = asRecord(data['message'])
+  const wrapped = asStoredRecord(data['message'])
   if (wrapped !== undefined && Number.isSafeInteger(data['turn'])
     && hasOnlyKeys(data, ['turn', 'message'])) {
     return { ...event, type: 'user/message', data: wrapped } as SessionEvent
@@ -113,9 +96,9 @@ function canonicalizeLegacySteeringEvent(event: SessionEvent, id: SessionId): Se
 
 function canonicalizeLegacyTurnStartEvent(event: SessionEvent, id: SessionId): SessionEvent {
   if (event.type !== 'turn/start') return event
-  const data = asRecord(event.data)
+  const data = asStoredRecord(event.data)
   if (data === undefined || !Object.hasOwn(data, 'trigger')) return event
-  const trigger = asRecord(data['trigger'])
+  const trigger = asStoredRecord(data['trigger'])
   if (!Number.isSafeInteger(data['turn']) || (data['turn'] as number) < 1
     || !hasOnlyKeys(data, ['turn', 'trigger'])
     || trigger === undefined || typeof trigger['kind'] !== 'string' || trigger['kind'].length === 0) {
@@ -126,12 +109,12 @@ function canonicalizeLegacyTurnStartEvent(event: SessionEvent, id: SessionId): S
 
 function canonicalizeLegacyTurnEndEvent(event: SessionEvent, id: SessionId): SessionEvent {
   if (event.type !== 'turn/end') return event
-  const data = asRecord(event.data)
+  const data = asStoredRecord(event.data)
   if (data === undefined) return event
   const malformed = (): never => {
     throw new Error(`session "${id}" contains malformed pre-react-loop turn/end at seq ${event.seq}`)
   }
-  const reason = asRecord(data['reason'])
+  const reason = asStoredRecord(data['reason'])
   if (!Number.isSafeInteger(data['turn']) || (data['turn'] as number) < 1
     || !hasOnlyKeys(data, ['turn', 'reason'])
     || reason === undefined || typeof reason['kind'] !== 'string') return malformed()
@@ -156,7 +139,7 @@ function canonicalizeLegacyTurnEndEvent(event: SessionEvent, id: SessionId): Ses
     case 'error': {
       if (Object.hasOwn(reason, 'error')) return event
       if (!Number.isSafeInteger(reason['step']) || (reason['step'] as number) < 0) return malformed()
-      const failure = asRecord(reason['failure'])
+      const failure = asStoredRecord(reason['failure'])
       if (failure !== undefined && hasOnlyKeys(reason, ['kind', 'step', 'failure'])
         && hasOnlyKeys(failure, ['message', 'code'], ['status', 'providerRetryAfterMs', 'requestId'])
         && typeof failure['message'] === 'string' && typeof failure['code'] === 'string'
@@ -192,7 +175,7 @@ function canonicalizeLegacyMessageEvent(
   id: SessionId,
   messageIds: ReadonlyMap<number, PersistedMessageId>,
 ): SessionEvent {
-  const data = asRecord(event.data)
+  const data = asStoredRecord(event.data)
   if (data === undefined) return event
   switch (event.type) {
     case 'user/message':
@@ -212,7 +195,7 @@ function canonicalizeLegacyMessageEvent(
             id: legacyMessageId(id, event.seq),
             role: 'assistant',
             content,
-            source: { ...asRecord(provenance), kind: 'model' },
+            source: { ...asStoredRecord(provenance), kind: 'model' },
           },
         },
       } as SessionEvent
@@ -242,8 +225,8 @@ function canonicalizeLegacyMessageEvent(
 }
 
 function eventMessageId(event: SessionEvent): PersistedMessageId | undefined {
-  const data = asRecord(event.data)
-  const message = event.type === 'user/message' ? data : asRecord(data?.['message'])
+  const data = asStoredRecord(event.data)
+  const message = event.type === 'user/message' ? data : asStoredRecord(data?.['message'])
   return typeof message?.['id'] === 'string' ? message['id'] as PersistedMessageId : undefined
 }
 

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

@@ -6,9 +6,9 @@
  */
 
 import { Context, Service } from '@deepseek-ai/cordis'
-import { SessionPreparation } from '@deepseek-ai/dsh-session'
-import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
+import { SessionPreparation, type SessionEvent, type SessionId, type SessionHeader } from '@deepseek-ai/dsh-session'
 import type { SessionPersistenceRevision } from './revision.ts'
+import type { StoredEventRead, StoredEventReadCompletion } from './format-decoder.ts'
 
 // Re-export the metadata vocabulary so Consumers import it from the Service Definition.
 export type { SessionHeader } from '@deepseek-ai/dsh-session'
@@ -86,6 +86,35 @@ export abstract class SessionPersistence extends Service {
     super(ctx, 'sessionPersistence')
   }
 
+  /**
+   * Build the standard lazy event stream and EOF metadata around one backend read.
+   * @param load - revision-checked batch loader owned by the backend.
+   * @param include - whether one loaded event belongs in this physical read.
+   * @param signal - optional cancellation checked between yielded events.
+   * @returns an independently consumable event read.
+   */
+  protected createStoredEventRead<TornMarker>(
+    load: () => Promise<{ readonly events: readonly unknown[]; readonly tornMarker?: TornMarker }>,
+    include: (event: unknown) => boolean,
+    signal?: AbortSignal,
+  ): StoredEventRead<TornMarker> {
+    const completed = Promise.withResolvers<StoredEventReadCompletion<TornMarker>>()
+    const events = (async function* (): AsyncIterable<unknown> {
+      try {
+        const batch = await load()
+        for (const event of batch.events) {
+          signal?.throwIfAborted()
+          if (include(event)) yield event
+        }
+        completed.resolve(batch.tornMarker === undefined ? {} : { tornMarker: batch.tornMarker })
+      } catch (error: unknown) {
+        completed.reject(error)
+        throw error
+      }
+    })()
+    return { events, completed: completed.promise }
+  }
+
   /**
    * Resolve this backend's independent local artifact for a session without
    * reading, creating, flushing, or otherwise materializing it. Backends such