Merge pull request #276 from deepseek-harness/codex/simp-session-dead-surface

refactor: prune dead session surfaces

docs/cordis-catalog/services.md

@@ -169,7 +169,7 @@ abstract list(): Promise<SessionHeader[]>
 
 Types: [SessionEvent](../core-data-structures/core.md)
 
-Source: [`packages/session-persistence/session-persistence/src/index.ts:60`](../../packages/session-persistence/session-persistence/src/index.ts)
+Source: [`packages/session-persistence/session-persistence/src/index.ts:30`](../../packages/session-persistence/session-persistence/src/index.ts)
 
 ## `ctx.sessionQuery` — `SessionQueryService`
 

docs/rfc/implemented/architecture/2026-06-18-session-surface.md

@@ -31,7 +31,7 @@ export type SurfaceOp =
 
 ### SurfaceManager: delta-based, not full rebuild
 
-A `SurfaceManager` class (private to `Session`) maintains the cached linked list. It tracks `_lastProcessedSeq` and processes only the **delta** (new events since the last access) rather than rescanning the entire log. Because the log is append-only, prior events never change — full rebuild is only needed after a wholesale log replacement (e.g., seeding).
+A `SurfaceManager` class (private to `Session`) maintains the cached linked list. It tracks `_lastProcessedSeq` and processes only the **delta** (new events since the last access) rather than rescanning the entire log. Because the log is append-only, prior events never change; a seeded log is simply the initial delta folded on first access.
 
 Delta processing is O(1) when no new events and O(new events) when new events arrive.
 

packages/core/session/README.md

@@ -35,7 +35,7 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
 - `session.append(type, data, opts?)` snapshots and freezes durable data and surface metadata, commits synchronously, then notifies observers with independent failure containment. Reentrant attached-session appends reject, and runtime checks cover widened unions and loaded logs.
 - `session.deriveMessages()` incrementally projects each new surface node once and returns a fresh array over shared frozen messages. A surface rewrite rebuilds the projection; there is no raw-log fallback.
 - `session.deriveEventMessage(event)` is the canonical per-event projection used by reconstruction and invariants.
-- `session.surface` lazily folds only new `surfaceOp` markers; `replaceGeneration` changes on every rewrite or invalidation.
+- `session.surface` lazily folds only new `surfaceOp` markers; `replaceGeneration` changes on every rewrite.
 - `session.events` is a cached frozen snapshot invalidated by append; accepted events remain deeply frozen.
 - `session.seq`, `session.id` — current sequence and readonly typed identity.
 - `session.header: SessionHeader` — detached, deep-frozen creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`/`seedLength`). Construction validates the durable record and requires its id to match `session.id`.

packages/core/session/src/surface.ts

@@ -193,26 +193,14 @@ export function foldSurface(events: readonly SessionEvent[]): SurfaceFoldResult
 export class SurfaceManager {
   /** Incremental state shared with the complete surface fold. */
   private _state = createFoldState()
-  /** The last processed seq. -1 forces a full rebuild on first access. */
+  /** The last processed seq. -1 folds the seeded log on first access. */
   private _lastProcessedSeq = -1
 
   constructor(private log: readonly SessionEvent[]) {}
 
   /**
-   * Reset to unprocessed state. Call after the log has been replaced
-   * wholesale (e.g. after Session seed). Not needed for normal appends —
-   * those are picked up incrementally.
-   */
-  invalidate(): void {
-    this._lastProcessedSeq = -1
-    // A wholesale rebuild is a rewrite: bump the generation so incremental
-    // consumers (the session's derived-message cache) discard their view.
-    this._state = createFoldState(this._state.replaceGeneration + 1)
-  }
-
-  /**
-   * The surface's rewrite generation: bumped by every folded `replace` op and
-   * by {@link invalidate}. A replace is the ONE operation that rewrites the
+   * The surface's rewrite generation, bumped by every folded `replace` op.
+   * A replace is the ONE operation that rewrites the
    * surface non-monotonically, so an incremental consumer of {@link nodes}
    * (the session's derived-message cache) compares this between visits — an
    * unchanged generation guarantees every node it has not seen is a pure tail

packages/core/session/tests/derived-cache.spec.ts

@@ -1,6 +1,6 @@
 /**
  * Derived-message cache contract against a scratch oracle: project new nodes
- * once, rebuild on surface generation changes, return fresh arrays over shared
+ * once, rebuild on surface replacements, return fresh arrays over shared
  * frozen messages, and remain value-equal to replay at every step.
  */
 
@@ -61,16 +61,6 @@ describe('derived-message cache', () => {
     expect(Object.isFrozen(first[0])).toBe(true)
   })
 
-  it('rebuilds after surface.invalidate() (the generation covers wholesale rebuilds too)', () => {
-    const session = new Session(SessionId('cache-invalidate'))
-    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
-    userText(session, 'one')
-    const before = session.deriveMessages()
-    session.surface.invalidate()
-    const after = session.deriveMessages()
-    expect(after).toEqual(before)
-    expect(after[0]).not.toBe(before[0])
-  })
 })
 
 describe('Session.deriveEventMessage — the per-event projection', () => {

packages/core/session/tests/surface.spec.ts

@@ -81,14 +81,6 @@ describe('SurfaceManager', () => {
     expect(nodes[1]!.next).toBeNull()
   })
 
-  it('invalidate resets to full rebuild', () => {
-    const s = surfaceSession()
-    expect(s.surface.nodes.length).toBe(2)
-    // After invalidate, the surface should rebuild from scratch on next access.
-    ;(s.surface).invalidate()
-    expect(s.surface.nodes.length).toBe(2) // same result, but rebuilt
-  })
-
   it('empty surface yields empty nodes', () => {
     const s = new Session(SessionId('empty'))
     // Only turn boundaries, no surface nodes.
@@ -386,7 +378,7 @@ describe('surface type guards', () => {
 })
 
 describe('SurfaceManager.replaceGeneration', () => {
-  it('folds the pending log delta on access and counts replaces and invalidations', () => {
+  it('folds the pending log delta on access and counts replaces', () => {
     const s = new Session(SessionId('gen'))
     s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
     s.append('user/message', { content: [{ type: 'text', text: 'one' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
@@ -400,10 +392,5 @@ describe('SurfaceManager.replaceGeneration', () => {
       content: [{ type: 'text', text: 'summary' }], source: { kind: 'plugin', plugin: 'compact' },
     }, { surfaceOp: { op: 'replace', start: nodes[0]!.seq, end: nodes[1]!.seq }, sourceEventSeqs: [nodes[0]!.seq, nodes[1]!.seq] })
     expect(s.surface.replaceGeneration).toBe(1)
-
-    // invalidate() is a rewrite too: the generation moves forward (and the
-    // refold re-counts the replace), never backwards.
-    s.surface.invalidate()
-    expect(s.surface.replaceGeneration).toBeGreaterThan(1)
   })
 })

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

@@ -14,7 +14,7 @@ import {
   SessionPersistence, PersistenceCoordinator,
   type PersistenceBackend, type StoredPrefix,
 } from '@deepseek-ai/dsh-session-persistence'
-import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
   encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
 } from './format.ts'
@@ -83,15 +83,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
   // One method serves both public `list` and the backend hook; delegating it to
   // the coordinator would call this hook recursively.
 
-  /**
-   * The per-session init promises, exposed for white-box tests that await a
-   * specific session's onCreated (there is no public API to await one init).
-   */
-  get inits(): Map<Session, Promise<void>> {
-    return this.coordinator.inits
-  }
   /* jscpd:ignore-end */
-
   // --- PersistenceBackend hooks (the file-bytes storage primitives) ---
 
   /** Read a stored prefix by id across all cwd buckets when cwd is unknown. */

packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts

@@ -488,12 +488,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
 
     // A new Session object reuses the id. Object-keyed initialization must run independently,
     // detect the disk collision, and reject instead of appending through session A's stale cursor.
-    const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
     let b!: Session
     await ctx.plugin(Object.assign((inner: Context) => {
       b = inner.sessions.create(SessionId('reuse'), { meta: { cwd: '/a' } })
     }, { inject: ['sessions'] }))
-    await expect(backend.inits.get(b)).rejects.toThrow(/already bound to a different live session|already has a persisted log on disk/)
+    await expect(ctx.sessions.flush(b)).rejects.toThrow(/already bound to a different live session|already has a persisted log on disk/)
   })
 
   it('a NO-CWD live session does NOT cross-cwd-adopt a same-id log from a real cwd bucket (loadLive is scope-exact)', async () => {
@@ -511,12 +510,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
     await ctx2.plugin(SessionPersistenceJsonl, { root })
-    const backend = ctx2.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
     let b!: Session
     await ctx2.plugin(Object.assign((inner: Context) => {
       b = inner.sessions.create(SessionId('x')) // no cwd
     }, { inject: ['sessions'] }))
-    await expect(backend.inits.get(b)).rejects.toThrow(/already has a persisted log on disk/)
+    await expect(ctx2.sessions.flush(b)).rejects.toThrow(/already has a persisted log on disk/)
 
     // The "/w" log is untouched — no no-cwd events were grafted onto it, and no
     // `_no-cwd` log for "x" was created.
@@ -533,7 +531,6 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await ctx.sessionPersistence.append(SessionId('divergent'), oneTurnLog())
     await ctx.sessionPersistence.load(SessionId('divergent'))
 
-    const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
     // A seed that keeps every seq/type/time but mutates a payload must NOT be
     // accepted as "the same session" — otherwise drain filters those seqs as
     // already persisted and the divergent payload is silently lost.
@@ -544,7 +541,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await ctx.plugin(Object.assign((inner: Context) => {
       bad = inner.sessions.create(SessionId('divergent'), { seed: tampered, meta: { cwd: '/a' } })
     }, { inject: ['sessions'] }))
-    await expect(backend.inits.get(bad)).rejects.toThrow(/do not match this live session|already has a persisted log/)
+    await expect(ctx.sessions.flush(bad)).rejects.toThrow(/do not match this live session|already has a persisted log/)
   })
 
   it('a second live session reusing a bound id is rejected', async () => {
@@ -557,12 +554,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     for (const s of ctx.sessions.list()) await ctx.parallel('session/flush', s)
     await firstFiber.dispose()
 
-    const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
     let second!: Session
     await ctx.plugin(Object.assign((inner: Context) => {
       second = inner.sessions.create(SessionId('bound'), { meta: { cwd: '/a' } })
     }, { inject: ['sessions'] }))
-    await expect(backend.inits.get(second))
+    await expect(ctx.sessions.flush(second))
       .rejects.toThrow(/already bound to a different live session|already has a persisted log|do not match/)
   })
 
@@ -594,12 +590,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await ctx2.plugin(SessionStore)
     await ctx2.plugin(SessionPersistenceJsonl, { root })
     await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE
-    const backend = ctx2.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
     let s!: Session
     await ctx2.plugin(Object.assign((inner: Context) => {
       s = inner.sessions.create(SessionId('exists-fault'), { meta: { cwd } })
     }, { inject: ['sessions'] }))
-    await expect(backend.inits.get(s)).rejects.toThrow(/ENOTDIR/)
+    await expect(ctx2.sessions.flush(s)).rejects.toThrow(/ENOTDIR/)
     await ctx2.fiber.dispose()
   })
 

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

@@ -14,7 +14,7 @@ import {
   SessionPersistence, PersistenceCoordinator,
   type PersistenceBackend, type StoredPrefix,
 } from '@deepseek-ai/dsh-session-persistence'
-import type { Session, SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
   type JournalMode, openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow,
 } from './schema.ts'
@@ -110,14 +110,6 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
   // One method serves both public `list` and the backend hook; delegating it to
   // the coordinator would call this hook recursively.
 
-  /**
-   * The per-session init promises, exposed for white-box tests that await a
-   * specific session's onCreated (there is no public API to await one init).
-   */
-  get inits(): Map<Session, Promise<void>> {
-    return this.coordinator.inits
-  }
-
   // --- PersistenceBackend hooks (the SQLite storage primitives) ---
 
   /** Read a stored prefix by id (ids are globally unique — no scope to scan). */

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

@@ -8,7 +8,6 @@
 import { Context } from 'cordis'
 import { interruptedTurnClosers, SESSION_FORMAT_VERSION, snapshotJsonValue } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
-import { seedCoversPrefix } from './index.ts'
 
 /**
  * A stored session's header, valid contiguous event prefix, and optional opaque
@@ -110,6 +109,15 @@ async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unkn
   return errors
 }
 
+/** Whether a live session seed reproduces a persisted prefix exactly. */
+function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
+  return prefix.length <= seed.length
+    && prefix.every((event, index) => {
+      const seedEvent = seed[index]
+      return seedEvent !== undefined && JSON.stringify(seedEvent) === JSON.stringify(event)
+    })
+}
+
 /**
  * Owns the backend-agnostic session write-path orchestration. A backend
  * constructs one (`new PersistenceCoordinator(ctx, this)`), implements
@@ -134,10 +142,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   private chains = new Map<SessionId, Promise<unknown>>()
   /**
    * Init promises keyed by live session object, preventing an id-reusing
-   * replacement from inheriting stale initialization. Readonly access supports
-   * backend white-box tests.
+   * replacement from inheriting stale initialization. Flush is the public
+   * observation boundary; callers do not inspect this bookkeeping directly.
    */
-  readonly inits = new Map<Session, Promise<void>>()
+  private inits = new Map<Session, Promise<void>>()
 
   constructor(private ctx: Context, private backend: PersistenceBackend<TornMarker>) {
     this.installWritePath()

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

@@ -6,7 +6,6 @@
  */
 
 import { Context, Service } from 'cordis'
-import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 
 // Re-export the metadata vocabulary so consumers import it from the seam.
@@ -22,35 +21,6 @@ declare module 'cordis' {
   }
 }
 
-/**
- * Check whether a live seed exactly reproduces a durable prefix, including full
- * payloads. This distinguishes resume/HMR rebinding from an id collision.
- * @param seed - the live session's creation-time event snapshot.
- * @param prefix - the persisted prefix the seed must reproduce.
- * @returns `true` when the prefix fits within the seed and every event matches by JSON text.
- */
-export function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
-  return prefix.length <= seed.length
-    && prefix.every((event, index) => {
-      const seedEvent = seed[index]
-      return seedEvent !== undefined && JSON.stringify(seedEvent) === JSON.stringify(event)
-    })
-}
-
-/**
- * Reject a batch that is not wholly losslessly JSON-serializable. Live session
- * appends already enforce this; persistence append paths also accept replay or
- * direct batches that may bypass a live session instance. Validation uses the
- * same one-pass materializer as the coordinator, so getters are read once.
- * @param events - the complete event batch to validate.
- */
-export function assertSerializable(events: readonly SessionEvent[]): void {
-  const snapshot = snapshotJsonValue(events)
-  if (snapshot === undefined) {
-    throw new Error('session event batch is not losslessly JSON-serializable because it contains non-JSON-serializable data')
-  }
-}
-
 /**
  * Durable append-only session storage. Implementations preserve contiguous,
  * losslessly JSON-serializable events; {@link append} resolves only after

packages/session-persistence/session-persistence/tests/coordinator-contract.ts

@@ -13,7 +13,6 @@ import { describe, expect, it } from 'vitest'
 import { Context, type Fiber } from 'cordis'
 import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
-import type { SessionPersistence } from '../src/index.ts'
 import { meta, oneTurnLog, appendLog } from './contract.ts'
 
 /**
@@ -39,11 +38,6 @@ export interface CoordinatorFixture {
 const WORK = '/w'
 const OTHER = '/other'
 
-/** The per-session init map a backend exposes for white-box init awaits. */
-function inits(persistence: SessionPersistence): Map<Session, Promise<void>> {
-  return (persistence as unknown as { inits: Map<Session, Promise<void>> }).inits
-}
-
 /** Append a whole event log to a live session, event by event (drives session/event). */
 function send(session: Session, events: readonly SessionEvent[]): void {
   appendLog(session, events)
@@ -168,7 +162,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         const seed = oneTurnLog()
         // A fork: a brand-new id whose seed came from elsewhere.
         const forked = ctx.sessions.create(SessionId('forked'), { seed, meta: { cwd: WORK } })
-        await inits(ctx.sessionPersistence).get(forked) // onCreated persisted the seed
+        await ctx.sessions.flush(forked) // onCreated persisted the seed
         const loaded = await ctx.sessionPersistence.load(SessionId('forked'))
         expect(loaded.events).toEqual(seed)
         // A flush with no NEW events must not double-write.
@@ -197,7 +191,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       try {
         const loaded = await second.ctx.sessionPersistence.load(SessionId('resumed'))
         const s2 = second.ctx.sessions.create(SessionId('resumed'), { seed: loaded.events, meta: { cwd: WORK } })
-        await inits(second.ctx.sessionPersistence).get(s2) // let onCreated adopt
+        await second.ctx.sessions.flush(s2) // let onCreated adopt
         s2.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
         s2.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
         await second.ctx.parallel('session/flush', s2)
@@ -370,7 +364,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       try {
         const s2 = second.ctx.sessions.create(SessionId('collide'), { meta: { cwd: WORK } })
         s2.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
-        await expect(inits(second.ctx.sessionPersistence).get(s2))
+        await expect(second.ctx.sessions.flush(s2))
           .rejects.toThrow(/already has a persisted log|id collision/)
       } finally {
         await second.fiber.dispose()
@@ -388,14 +382,14 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
           firstSession = inner.sessions.create(SessionId('abandoned'), { meta: { cwd: WORK } })
         }, { inject: ['sessions'] }))
-        await inits(ctx.sessionPersistence).get(firstSession) // register the lazy state
+        await ctx.sessions.flush(firstSession) // register the lazy state
         await firstFiber.dispose() // disposed before any append → never materialized
 
         let reuse!: Session
         await ctx.plugin(Object.assign((inner: Context) => {
           reuse = inner.sessions.create(SessionId('abandoned'), { meta: { cwd: WORK } })
         }, { inject: ['sessions'] }))
-        await expect(inits(ctx.sessionPersistence).get(reuse)).resolves.toBeUndefined()
+        await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined()
         reuse.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
         reuse.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
         await ctx.parallel('session/flush', reuse)
@@ -415,7 +409,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
           first = inner.sessions.create(SessionId('buffered'), { meta: { cwd: WORK } })
         }, { inject: ['sessions'] }))
-        await inits(ctx.sessionPersistence).get(first)
+        await ctx.sessions.flush(first)
         // Append a turn but do NOT flush — events sit in the write-behind buffer.
         first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
         first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
@@ -425,7 +419,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         await ctx.plugin(Object.assign((inner: Context) => {
           reuse = inner.sessions.create(SessionId('buffered'), { meta: { cwd: WORK } })
         }, { inject: ['sessions'] }))
-        await expect(inits(ctx.sessionPersistence).get(reuse)).rejects.toThrow(/already bound to a different live session/)
+        await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/already bound to a different live session/)
       } finally {
         await fiber.dispose()
         await fix.cleanup()
@@ -462,7 +456,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         // A live session with that id arrives and claims it (cursor 0 matches
         // trivially), persisting its seed.
         const live = ctx.sessions.create(SessionId('lazy-claim'), { seed: oneTurnLog(), meta: { cwd: WORK } })
-        await expect(inits(ctx.sessionPersistence).get(live)).resolves.toBeUndefined()
+        await expect(ctx.sessions.flush(live)).resolves.toBeUndefined()
         const loaded = await ctx.sessionPersistence.load(SessionId('lazy-claim'))
         expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
       } finally {
@@ -487,7 +481,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         await ctx.plugin(Object.assign((inner: Context) => {
           fresh = inner.sessions.create(SessionId('preview'), { meta: { cwd: WORK } })
         }, { inject: ['sessions'] }))
-        await expect(inits(ctx.sessionPersistence).get(fresh))
+        await expect(ctx.sessions.flush(fresh))
           .rejects.toThrow(/do not match this live session|already has a persisted log|id collision/)
       } finally {
         await fiber.dispose()
@@ -511,7 +505,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
           { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
           { type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
         ], meta: { cwd: WORK } })
-        await inits(ctx.sessionPersistence).get(cont)
+        await ctx.sessions.flush(cont)
         const loaded = await ctx.sessionPersistence.load(SessionId('claim'))
         expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
       } finally {
@@ -531,7 +525,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         // cwd scope is the fence (without it, WORK events would append under the
         // OTHER header). Rejected as a collision.
         const live = ctx.sessions.create(SessionId('wrong-cwd-claim'), { seed: oneTurnLog(), meta: { cwd: WORK } })
-        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+        await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
       } finally {
         await fiber.dispose()
         await fix.cleanup()
@@ -549,7 +543,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         // A live session whose SEED matches the loaded prefix but whose cwd is
         // WORK must still be rejected — the cwd guard runs before the seed check.
         const live = ctx.sessions.create(SessionId('wrong-cwd-load'), { seed: events, meta: { cwd: WORK } })
-        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+        await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
       } finally {
         await fiber.dispose()
         await fix.cleanup()
@@ -565,7 +559,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         // A live session reusing the id but WITH cwd WORK is a cwd mismatch
         // (undefined vs WORK) and must be rejected.
         const live = ctx.sessions.create(SessionId('no-cwd-state'), { seed: oneTurnLog(), meta: { cwd: WORK } })
-        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+        await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
       } finally {
         await fiber.dispose()
         await fix.cleanup()

packages/session-persistence/session-persistence/tests/persistence.spec.ts

@@ -1,9 +1,9 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
 import SessionStore, { SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
-import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
-  SessionPersistence, PersistenceCoordinator, assertSerializable, seedCoversPrefix,
+  SessionPersistence, PersistenceCoordinator,
   type PersistenceBackend, type StoredPrefix,
 } from '../src/index.ts'
 import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
@@ -53,11 +53,6 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
     return this.coordinator.load(id)
   }
 
-  /** White-box accessor: await a specific session's onCreated init. */
-  get inits(): Map<Session, Promise<void>> {
-    return this.coordinator.inits
-  }
-
   // --- PersistenceBackend hooks (the Map storage primitives) ---
 
   // A Map-backed store has no torn tails, so `tornMarker` is never set. Ids are
@@ -157,38 +152,3 @@ describe('SessionPersistence service registration', () => {
     await fiber.dispose()
   })
 })
-
-describe('shared persistence helpers', () => {
-  it('accepts a seed that reproduces the persisted prefix exactly', () => {
-    const log = oneTurnLog()
-    expect(seedCoversPrefix(log, log.slice(0, 3))).toBe(true)
-    expect(seedCoversPrefix(log, [])).toBe(true)
-  })
-
-  it('rejects a prefix longer than the seed', () => {
-    const log = oneTurnLog()
-    expect(seedCoversPrefix(log.slice(0, 2), log)).toBe(false)
-  })
-
-  it('rejects a same-envelope event with mutated data', () => {
-    const log = oneTurnLog()
-    const tampered = structuredClone(log)
-    const event = tampered[1]!
-    tampered[1] = {
-      ...event,
-      data: { ...event.data, content: [{ type: 'text', text: 'tampered' }] },
-    } as SessionEvent
-    expect(seedCoversPrefix(tampered, log.slice(0, 2))).toBe(false)
-  })
-
-  it('accepts JSON-serializable event data', () => {
-    expect(() => { assertSerializable(oneTurnLog()) }).not.toThrow()
-  })
-
-  it('rejects a batch containing non-JSON-serializable event data', () => {
-    const bad = [
-      { type: 'user/message', seq: 0, time: 1, data: { content: 1n } },
-    ] as unknown as SessionEvent[]
-    expect(() => { assertSerializable(bad) }).toThrow(/batch is not losslessly JSON-serializable/)
-  })
-})