feat(session): bound persistence write batching

packages/session-persistence/session-persistence/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/session-persistence/session-persistence/README.md
-README.md: b29ff5ba17f384e8d3b1700ed3ad6c80aeeb184c
-README.zh.md: 8ca4b6a7128383fdca706f35914a06eb1f26de02
+README.md: 89c7cd5ebaff6f9ce9df9b60a50121dff4ddeeb5
+README.zh.md: 1ef7eb6c6c507167f61bb5df7c4ce8183aac5d0d

packages/session-persistence/session-persistence/README.md

@@ -29,9 +29,9 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 
 ## The write coordinator
 
-`PersistenceCoordinator` owns per-id state and serialization, one eager write controller per live session, lazy materialization, crash-tail repair, session adoption, and quiescent disposal. A first-party backend composes one, implements the small `PersistenceBackend` storage hook interface, and delegates its stateful methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives; see the [coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md) and [flush-controller simplification](../../../.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md).
+`PersistenceCoordinator` owns per-id state and serialization, one bounded write controller per live session, lazy materialization, crash-tail repair, session adoption, and quiescent disposal. A first-party backend composes one, implements the small `PersistenceBackend` storage hook interface, and delegates its stateful methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives; see the [coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md), [flush-controller simplification](../../../.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md), and [bounded batching decision](../../../.agents/notes/implemented/architecture/2026-08-08-bounded-session-persistence-write-batching.md).
 
-Each `session/event` copies its event into the session controller and starts an eager drain without blocking the producer. Concurrent notifications share the current drain; events admitted during a write remain pending and trigger the next batch. `session/flush` is an observation barrier that waits until the controller has no current or pending batch. An eager failure is logged and retains the batch; the next explicit flush or backend teardown retries it and surfaces failure to its caller.
+Each `session/event` copies its event into the session controller. The first pending event starts a fixed batching window; later events join without resetting its deadline. The configured `writeBatchMaxDelayMs` bounds this intentional wait, not event-loop, initialization, serialized-operation, or backend latency. Events admitted during a write form a new bounded batch. `session/flush` cancels the wait and is a shared quiescence barrier that drains events admitted while it runs. A background failure is logged once, retains the ordered batch, and pauses automatic retry; a new event starts a fresh window, while explicit flush or backend teardown retries immediately and surfaces a repeated failure.
 
 Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. For a cold id, inspection reads, validates, freezes, and constructs one unpublished Session; repeated inspection reuses that object graph only while its source revision remains current. `prepare(id)` performs the same check before repair, reserves the exact Session, commits any pending torn-tail/interrupted-turn repair, and returns it for publication. HMR adoption reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn.
 

packages/session-persistence/session-persistence/README.zh.md

@@ -29,9 +29,9 @@
 
 ## 写入协调器
 
-`PersistenceCoordinator` 负责每 id 状态和串行化、每个活动会话各自的主动写入 controller、延迟实体化、崩溃尾部修复、会话接管和完全停稳的 dispose（资源释放）。第一方后端组合一个协调器，实现小型 `PersistenceBackend` 存储钩子接口，并委托其有状态方法。因此 JSONL 和 SQLite 共享生命周期正确性，同时保留不同存储原语；见[协调器 Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md) 和 [flush controller 简化](../../../.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md)。
+`PersistenceCoordinator` 负责每 id 状态和串行化、每个活动会话各自的有界写入 controller、延迟实体化、崩溃尾部修复、会话接管和完全停稳的 dispose（资源释放）。第一方后端组合一个协调器，实现小型 `PersistenceBackend` 存储钩子接口，并委托其有状态方法。因此 JSONL 和 SQLite 共享生命周期正确性，同时保留不同存储原语；见[协调器 Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md)、[flush controller 简化](../../../.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md)和[有界批处理决策](../../../.agents/notes/implemented/architecture/2026-08-08-bounded-session-persistence-write-batching.md)。
 
-每个 `session/event` 将事件复制到会话 controller，并在不阻塞生产者的情况下立即启动 drain。并发通知共享当前 drain；写入期间接纳的事件保持 pending，并触发下一批。`session/flush` 是观察屏障，会等待 controller 无当前或 pending 批次。即时写入失败会记录日志并保留批次；下一次显式 flush 或后端拆卸会重试该批次，并将失败返回给调用方。
+每个 `session/event` 将事件复制到会话 controller。第一个待处理事件会开启固定批处理窗口；后续事件会加入该批次，但不会重置截止时间。配置的 `writeBatchMaxDelayMs` 只限制这段有意等待，而不限制事件循环、初始化、串行化操作或后端延迟。写入期间接纳的事件会形成一个新的有界批次。`session/flush` 会取消等待，并作为共享的完全停稳屏障，排空屏障运行期间接纳的事件。后台写入失败只记录一次日志，保留顺序不变的批次，并暂停自动重试；新事件会开启新的固定窗口，而显式 flush 或后端拆卸会立即重试，并在失败再次发生时向调用方暴露失败。
 
 崩溃修复只适用于冷状态。对于实时 id，`load(id)` 为权威内存日志制作快照，等待该快照持久，并只在平衡时返回；开放实时轮次会被拒绝，而不会收到合成中断 closer。对于冷 id，检查只读取、验证、冻结并构造一次未发布 Session；只有来源 revision 仍然是当前值时，重复检查才会复用该对象图。`prepare(id)` 在修复前执行相同校验，预留精确 Session，提交任何待处理的撕裂尾部或中断轮次修复，并将其返回用于发布。HMR 接管通过 `loadStored` 读取，应用协调器 cwd 检查，并绝不关闭活动轮次。
 

packages/session-persistence/session-persistence/package.json

@@ -28,6 +28,7 @@
     "@deepseek-ai/dsh-brand": "^0.0.1",
     "@deepseek-ai/dsh-invariants": "^0.0.1",
     "@deepseek-ai/dsh-session": "^0.0.1",
+    "@deepseek-ai/dsh-timeout": "^0.0.1",
     "cordis": "^4.0.0-rc.7"
   },
   "devDependencies": {
@@ -35,6 +36,7 @@
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-scope": "workspace:^",
     "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-timeout": "workspace:^",
     "cordis": "^4.0.0-rc.7"
   }
 }

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

@@ -15,14 +15,22 @@ import {
   snapshotSessionEvent,
 } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
+import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
 import type { SessionInspection } from './index.ts'
 import type { SessionPersistenceRevision } from './revision.ts'
 import { observeQueuedAbort, SessionPreparations } from './preparations.ts'
 import type { SessionPreparationReservation } from './preparations.ts'
+import { SessionWriteBehind } from './write-behind.ts'
 
 /** Default number of detached session preparations retained by a coordinator. */
 export const DEFAULT_PREPARED_SESSION_CACHE_SIZE = 5
 
+/** Default maximum intentional wait before a live session batch starts writing. */
+export const DEFAULT_WRITE_BATCH_MAX_DELAY_MS = 200
+
+/** Largest write batching delay accepted by Node's timer implementation. */
+export const MAX_WRITE_BATCH_DELAY_MS = MAX_TIMER_DELAY_MS
+
 /** Durable session contents failed validation after a successful backend read. */
 export class SessionPersistenceCorruptionError extends Error {
   /**
@@ -39,6 +47,8 @@ export class SessionPersistenceCorruptionError extends Error {
 export interface PersistenceCoordinatorOptions {
   /** Maximum completed unpublished preparations retained for reuse. */
   readonly preparedSessionCacheSize: number
+  /** Maximum intentional batching wait after an idle live queue receives work. */
+  readonly writeBatchMaxDelayMs: number
 }
 
 /**
@@ -174,11 +184,10 @@ interface SessionState {
   owner?: Session
 }
 
-/** One live session's initialization and eager write-behind controller. */
+/** One live session's initialization and bounded write-behind controller. */
 interface LiveSessionState {
-  pending: SessionEvent[]
   init: Promise<void>
-  flush: Promise<void> | undefined
+  writes: SessionWriteBehind
 }
 
 /** One validated cold source and the exact unpublished Session built from it. */
@@ -531,7 +540,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   private states = new Map<SessionId, SessionState>()
   /** Lifecycle and write-behind state keyed by the exact live Session. */
   private live = new Map<Session, LiveSessionState>()
-  /** Exact disposed lifecycles whose eager tail is still draining. */
+  /** Exact disposed lifecycles whose buffered tail is still draining. */
   private retirements = new Map<SessionId, Promise<void>>()
   /** Shared cold reads, unpublished reservations, and completed LRU entries. */
   private readonly preparations: SessionPreparations<PreparedSessionSource<TornMarker>, SessionState>
@@ -540,18 +549,27 @@ export class PersistenceCoordinator<TornMarker = unknown> {
    * same id, so writes for one session never interleave. Keyed by session id.
    */
   private chains = new Map<SessionId, Promise<unknown>>()
+  /** Resolved fixed write-batching window shared by per-session controllers. */
+  private readonly writeBatchMaxDelayMs: number
 
   constructor(
     private ctx: Context,
     private backend: PersistenceBackend<TornMarker>,
     options: PersistenceCoordinatorOptions = {
       preparedSessionCacheSize: DEFAULT_PREPARED_SESSION_CACHE_SIZE,
+      writeBatchMaxDelayMs: DEFAULT_WRITE_BATCH_MAX_DELAY_MS,
     },
   ) {
     if (!Number.isSafeInteger(options.preparedSessionCacheSize)
       || options.preparedSessionCacheSize < 1) {
       throw new TypeError('preparedSessionCacheSize must be a positive safe integer')
     }
+    if (!Number.isSafeInteger(options.writeBatchMaxDelayMs)
+      || options.writeBatchMaxDelayMs < 1
+      || options.writeBatchMaxDelayMs > MAX_WRITE_BATCH_DELAY_MS) {
+      throw new TypeError(`writeBatchMaxDelayMs must be an integer between 1 and ${MAX_WRITE_BATCH_DELAY_MS}`)
+    }
+    this.writeBatchMaxDelayMs = options.writeBatchMaxDelayMs
     this.preparations = new SessionPreparations(options.preparedSessionCacheSize)
     this.installWritePath()
   }
@@ -1014,14 +1032,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       void this.initFor(session)
     })
 
-    // Keep a persistence-owned copy of each frozen event and start an eager drain.
+    // Keep a persistence-owned copy of each frozen event and start its bounded window.
     ctx.on('session/event', (session, event) => {
       const live = this.initFor(session)
-      live.pending.push(structuredClone(event))
-      if (live.flush === undefined) this.scheduleDrain(session, live)
+      live.writes.enqueue(event)
     })
 
-    // Callers use flush as the observation barrier for the eager write path.
+    // Callers use flush as the immediate durability barrier for buffered writes.
     ctx.on('session/flush', session => this.flush(session))
 
     // Session disposal is observe-only, so retirement contains its own failure.
@@ -1067,9 +1084,14 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       return restored
     }
     const seed = session.events.map(e => structuredClone(e))
-    const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined }
+    let init = Promise.resolve()
+    const live: LiveSessionState = {
+      init,
+      writes: this.createWriteBehind(session, () => init),
+    }
     this.live.set(session, live)
-    live.init = this.serialize(session.header.id, () => this.onCreated(session, seed))
+    init = this.serialize(session.header.id, () => this.onCreated(session, seed))
+    live.init = init
     live.init.catch(() => { /* observed by flush/dispose through the controller */ })
     return live
   }
@@ -1088,9 +1110,14 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     const suffix = session.events.slice(state.cursor).map(event => structuredClone(event))
     this.preparations.attach(reservation)
     state.owner = session
-    const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined }
+    let init = Promise.resolve()
+    const live: LiveSessionState = {
+      init,
+      writes: this.createWriteBehind(session, () => init),
+    }
     if (suffix.length > 0) {
-      live.init = this.serialize(session.id, () => this.appendCore(session.id, suffix))
+      init = this.serialize(session.id, () => this.appendCore(session.id, suffix))
+      live.init = init
       live.init.catch(() => { /* observed by flush/dispose through the controller */ })
     }
     return live
@@ -1152,7 +1179,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
         return
       }
       const owner = this.live.get(tracked.owner)
-      if (!tracked.materialized && !owner?.pending.length) {
+      if (!tracked.materialized && !owner?.writes.hasWork) {
         this.states.delete(id)
       } else {
         throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`)
@@ -1214,41 +1241,29 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   private async flush(session: Session): Promise<void> {
     const live = this.initFor(session)
     await live.init
-    const overlapping = live.flush
-    if (overlapping !== undefined) await Promise.allSettled([overlapping])
-    while (live.flush !== undefined || live.pending.length > 0) {
-      if (live.flush !== undefined) await live.flush
-      else await this.ensureFlush(session, live)
-    }
+    await live.writes.flush()
   }
 
-  /** Start an eager drain without exposing its failure to the synchronous append. */
-  private scheduleDrain(session: Session, live: LiveSessionState): void {
-    void this.ensureFlush(session, live).catch((error: unknown) => {
-      this.ctx.logger.warn(`${this.backend.name}: eager drain for session "${session.id}" failed (buffered events retained): ${String(error)}`)
+  /** Build one package-private write controller around initialization and id serialization. */
+  private createWriteBehind(session: Session, ready: () => Promise<void>): SessionWriteBehind {
+    return new SessionWriteBehind({
+      maxDelayMs: this.writeBatchMaxDelayMs,
+      write: async (batch) => {
+        await ready()
+        await this.serialize(session.header.id, () => this.appendLiveBatch(session.header.id, batch))
+      },
+      reportBackgroundFailure: (error) => {
+        this.ctx.logger.warn(`${this.backend.name}: background write for session "${session.id}" failed (buffered events retained): ${String(error)}`)
+      },
     })
   }
 
-  /** Start one drain for the complete pending batch. */
-  private ensureFlush(session: Session, live: LiveSessionState): Promise<void> {
-    const flush = live.init
-      .then(() => this.serialize(session.header.id, () => this.drain(session.header.id, live)))
-      .finally(() => { live.flush = undefined })
-    live.flush = flush
-    void flush.then(() => {
-      if (live.pending.length > 0) this.scheduleDrain(session, live)
-    }, () => {})
-    return flush
-  }
-
-  /** Drain one stable prefix; events admitted during the write remain pending. */
-  private async drain(id: SessionId, live: LiveSessionState): Promise<void> {
-    const batch = live.pending.slice()
+  /** Append one controller-owned prefix after filtering events initialization already stored. */
+  private async appendLiveBatch(id: SessionId, batch: readonly SessionEvent[]): Promise<void> {
     const state = this.states.get(id)
-    /* v8 ignore next -- state is always set by the awaited init before flush */
+    /* v8 ignore next -- state is always set by the awaited initialization */
     const cursor = state?.cursor ?? 0
     const fresh = batch.filter(e => e.seq >= cursor)
     await this.appendCore(id, fresh)
-    live.pending.splice(0, batch.length)
   }
 }

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

@@ -33,6 +33,8 @@ export interface SessionInspection {
 // 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'

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

@@ -0,0 +1,153 @@
+/**
+ * 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
+  }
+
+  /** 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.unshift(...batch)
+        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
+  }
+}

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

@@ -3,6 +3,7 @@ import { Context } from 'cordis'
 import SessionStore, { Session, SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
+  DEFAULT_PREPARED_SESSION_CACHE_SIZE, DEFAULT_WRITE_BATCH_MAX_DELAY_MS, MAX_WRITE_BATCH_DELAY_MS,
   SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
   type PersistenceBackend, type SessionPersistenceSnapshot, type StoredPrefix, type StoredSuffix,
 } from '../src/index.ts'
@@ -53,7 +54,9 @@ interface MemoryConfig { store?: MemoryStore }
 /** Test-only view of the coordinator containers whose retirement is the contract under test. */
 interface CoordinatorInternals {
   states: Map<unknown, unknown>
-  live: Map<unknown, { pending: unknown[]; flush: Promise<void> | undefined }>
+  live: Map<unknown, {
+    writes: { pending: unknown[]; active: Promise<void> | undefined; hasWork: boolean }
+  }>
   chains: Map<unknown, unknown>
   retirements: Map<unknown, Promise<void>>
 }
@@ -253,7 +256,7 @@ runCoordinatorContract('memory', async (): Promise<CoordinatorFixture> => {
   }
 })
 
-describe('PersistenceCoordinator eager writes', () => {
+describe('PersistenceCoordinator bounded writes', () => {
   it('starts a follow-up batch for events admitted during an in-flight write', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
@@ -263,11 +266,14 @@ describe('PersistenceCoordinator eager writes', () => {
       if (attempt === 1) await appendGate.promise
     }
     const fiber = await ctx.plugin(Object.assign((inner: Context) => {
-      new PersistenceCoordinator(inner, backend)
+      new PersistenceCoordinator(inner, backend, {
+        preparedSessionCacheSize: DEFAULT_PREPARED_SESSION_CACHE_SIZE,
+        writeBatchMaxDelayMs: 1,
+      })
     }, { inject: ['sessions'] }))
 
     try {
-      const session = ctx.sessions.create(SessionId('eager-follow-up'))
+      const session = ctx.sessions.create(SessionId('bounded-follow-up'))
       await ctx.sessions.flush(session)
       session.append('turn/start', { turn: 1 })
       await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
@@ -286,7 +292,7 @@ describe('PersistenceCoordinator eager writes', () => {
     }
   })
 
-  it('retries a failed overlapping eager write at the explicit flush barrier', async () => {
+  it('retries a failed overlapping background write at the explicit flush barrier', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
     const backend = new ControlledBackend()
@@ -294,15 +300,18 @@ describe('PersistenceCoordinator eager writes', () => {
     backend.beforeAppend = async (attempt) => {
       if (attempt === 1) {
         await appendGate.promise
-        throw new Error('transient eager failure')
+        throw new Error('transient background failure')
       }
     }
     const fiber = await ctx.plugin(Object.assign((inner: Context) => {
-      new PersistenceCoordinator(inner, backend)
+      new PersistenceCoordinator(inner, backend, {
+        preparedSessionCacheSize: DEFAULT_PREPARED_SESSION_CACHE_SIZE,
+        writeBatchMaxDelayMs: 1,
+      })
     }, { inject: ['sessions'] }))
 
     try {
-      const session = ctx.sessions.create(SessionId('eager-flush-retry'))
+      const session = ctx.sessions.create(SessionId('bounded-flush-retry'))
       await ctx.sessions.flush(session)
       session.append('turn/start', { turn: 1 })
       session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
@@ -401,9 +410,20 @@ describe('PersistenceCoordinator session preparations', () => {
 
     expect(() => new PersistenceCoordinator(ctx, backend, {
       preparedSessionCacheSize: capacity,
+      writeBatchMaxDelayMs: DEFAULT_WRITE_BATCH_MAX_DELAY_MS,
     })).toThrow(/positive safe integer/)
   })
 
+  it.each([0, 1.5, MAX_WRITE_BATCH_DELAY_MS + 1])('rejects invalid write batch delay %s', (delay) => {
+    const ctx = new Context()
+    const backend = new ControlledBackend()
+
+    expect(() => new PersistenceCoordinator(ctx, backend, {
+      preparedSessionCacheSize: DEFAULT_PREPARED_SESSION_CACHE_SIZE,
+      writeBatchMaxDelayMs: delay,
+    })).toThrow(/writeBatchMaxDelayMs must be an integer between/)
+  })
+
   it('retries invalidated prepare and load reservations', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
@@ -592,6 +612,39 @@ describe('PersistenceCoordinator session preparations', () => {
     }
   })
 
+  it('writes new events after publishing a preparation with no unpublished suffix', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const id = SessionId('prepared-live-write')
+    const stored = [
+      ...oneTurnLog(),
+      { type: 'session/end-seed', seq: 6, time: 7, data: {} } as SessionEvent,
+    ]
+    backend.store.set(id, { meta: meta(id), events: stored })
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const preparation = await coordinator.prepare(id)
+    const detach = ctx.sessions.enter(preparation.session)
+
+    try {
+      ctx.sessions.announce(preparation.session)
+      preparation.session.append('turn/start', { turn: 2 })
+      preparation.session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
+
+      await expect(ctx.sessions.flush(preparation.session)).resolves.toBe(true)
+      expect(backend.store.get(id)?.events.map(event => event.seq))
+        .toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8])
+    } finally {
+      detach()
+      preparation[Symbol.dispose]()
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
   it('reuses the exact Session from inspect through repeated unpublished prepare calls', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
@@ -1004,7 +1057,10 @@ describe('PersistenceCoordinator session preparations', () => {
     backend.store.set(secondId, { meta: meta(secondId), events: oneTurnLog() })
     let coordinator!: PersistenceCoordinator<never>
     const fiber = await ctx.plugin(Object.assign((inner: Context) => {
-      coordinator = new PersistenceCoordinator(inner, backend, { preparedSessionCacheSize: 1 })
+      coordinator = new PersistenceCoordinator(inner, backend, {
+        preparedSessionCacheSize: 1,
+        writeBatchMaxDelayMs: DEFAULT_WRITE_BATCH_MAX_DELAY_MS,
+      })
     }, { inject: ['sessions'] }))
 
     try {
@@ -1529,7 +1585,7 @@ describe('PersistenceCoordinator retirement', () => {
 
       await vi.waitFor(() => {
         expect(backend.appendAttempts).toBeGreaterThanOrEqual(1)
-        expect([...internals.live.values()][0]?.pending).toEqual(expect.arrayContaining([
+        expect([...internals.live.values()][0]?.writes.pending).toEqual(expect.arrayContaining([
           expect.objectContaining({ seq: 0 }),
           expect.objectContaining({ seq: 1 }),
         ]))
@@ -1573,7 +1629,7 @@ describe('PersistenceCoordinator retirement', () => {
       await vi.waitFor(() => {
         expect(backend.appendAttempts).toBe(1)
         expect(internals.live.size).toBe(1)
-        expect([...internals.live.values()][0]?.flush).toBeInstanceOf(Promise)
+        expect([...internals.live.values()][0]?.writes.active).toBeInstanceOf(Promise)
       })
 
       let disposed = false

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

@@ -0,0 +1,252 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
+import { SessionWriteBehind } from '../src/write-behind.ts'
+
+/** Minimal ordered event fixture; batching does not interpret event vocabulary. */
+function event(seq: number): SessionEvent<'turn/start'> {
+  return {
+    type: 'turn/start',
+    seq,
+    time: seq,
+    data: { turn: seq + 1 },
+  }
+}
+
+afterEach(() => {
+  vi.useRealTimers()
+})
+
+describe('SessionWriteBehind', () => {
+  it('uses one fixed window from the first queued event and owns its copy', async () => {
+    vi.useFakeTimers()
+    const batches: SessionEvent[][] = []
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => { batches.push(structuredClone(events) as SessionEvent[]) },
+      reportBackgroundFailure: vi.fn(),
+    })
+    const first = event(0)
+
+    controller.enqueue(first)
+    first.data.turn = 99
+    await vi.advanceTimersByTimeAsync(150)
+    controller.enqueue(event(1))
+    await vi.advanceTimersByTimeAsync(49)
+    expect(batches).toEqual([])
+
+    await vi.advanceTimersByTimeAsync(1)
+    expect(batches).toEqual([[
+      expect.objectContaining({ seq: 0, data: { turn: 1 } }),
+      expect.objectContaining({ seq: 1 }),
+    ]])
+    expect(controller.hasWork).toBe(false)
+  })
+
+  it('coalesces twenty events admitted ten milliseconds apart into one 200 ms batch', async () => {
+    vi.useFakeTimers()
+    const batches: number[][] = []
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => { batches.push(events.map(item => item.seq)) },
+      reportBackgroundFailure: vi.fn(),
+    })
+
+    controller.enqueue(event(0))
+    for (let seq = 1; seq < 20; seq += 1) {
+      await vi.advanceTimersByTimeAsync(10)
+      controller.enqueue(event(seq))
+    }
+    expect(batches).toEqual([])
+
+    await vi.advanceTimersByTimeAsync(10)
+    expect(batches).toEqual([Array.from({ length: 20 }, (_, seq) => seq)])
+    await controller.flush()
+  })
+
+  it('makes concurrent flushes one immediate barrier that drains admitted tails', async () => {
+    vi.useFakeTimers()
+    const gate = Promise.withResolvers<boolean>()
+    const batches: number[][] = []
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => {
+        batches.push(events.map(item => item.seq))
+        if (batches.length === 1) await gate.promise
+      },
+      reportBackgroundFailure: vi.fn(),
+    })
+
+    controller.enqueue(event(0))
+    const first = controller.flush()
+    const second = controller.flush()
+    expect(second).toBe(first)
+    await Promise.resolve()
+    expect(batches).toEqual([[0]])
+
+    controller.enqueue(event(1))
+    gate.resolve(true)
+    await first
+    expect(batches).toEqual([[0], [1]])
+    expect(controller.hasWork).toBe(false)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('starts a new window for work admitted after an already-quiescent barrier', async () => {
+    vi.useFakeTimers()
+    const batches: number[][] = []
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => { batches.push(events.map(item => item.seq)) },
+      reportBackgroundFailure: vi.fn(),
+    })
+
+    const barrier = controller.flush()
+    controller.enqueue(event(0))
+    await barrier
+    expect(batches).toEqual([])
+    expect(vi.getTimerCount()).toBe(1)
+
+    await vi.advanceTimersByTimeAsync(200)
+    expect(batches).toEqual([[0]])
+    expect(controller.hasWork).toBe(false)
+  })
+
+  it('starts an over-budget tail immediately after the active write', async () => {
+    vi.useFakeTimers()
+    const gate = Promise.withResolvers<boolean>()
+    const batches: number[][] = []
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => {
+        batches.push(events.map(item => item.seq))
+        if (batches.length === 1) await gate.promise
+      },
+      reportBackgroundFailure: vi.fn(),
+    })
+
+    controller.enqueue(event(0))
+    await vi.advanceTimersByTimeAsync(200)
+    expect(batches).toEqual([[0]])
+    controller.enqueue(event(1))
+    await vi.advanceTimersByTimeAsync(200)
+    expect(batches).toEqual([[0]])
+
+    gate.resolve(true)
+    await vi.advanceTimersByTimeAsync(0)
+    expect(batches).toEqual([[0], [1]])
+    await controller.flush()
+  })
+
+  it('keeps a tail deadline that has not expired when the active write finishes', async () => {
+    vi.useFakeTimers()
+    const gate = Promise.withResolvers<boolean>()
+    const batches: number[][] = []
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => {
+        batches.push(events.map(item => item.seq))
+        if (batches.length === 1) await gate.promise
+      },
+      reportBackgroundFailure: vi.fn(),
+    })
+
+    controller.enqueue(event(0))
+    await vi.advanceTimersByTimeAsync(200)
+    controller.enqueue(event(1))
+    await vi.advanceTimersByTimeAsync(50)
+    gate.resolve(true)
+    await vi.advanceTimersByTimeAsync(0)
+    expect(batches).toEqual([[0]])
+
+    await vi.advanceTimersByTimeAsync(149)
+    expect(batches).toEqual([[0]])
+    await vi.advanceTimersByTimeAsync(1)
+    expect(batches).toEqual([[0], [1]])
+    await controller.flush()
+  })
+
+  it('pauses automatic retries after failure and preserves order for new work', async () => {
+    vi.useFakeTimers()
+    const failure = new Error('storage unavailable')
+    const report = vi.fn()
+    const batches: number[][] = []
+    let attempt = 0
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => {
+        batches.push(events.map(item => item.seq))
+        if (++attempt === 1) throw failure
+      },
+      reportBackgroundFailure: report,
+    })
+
+    controller.enqueue(event(0))
+    await vi.advanceTimersByTimeAsync(200)
+    expect(report).toHaveBeenCalledWith(failure)
+    expect(controller.hasWork).toBe(true)
+    await vi.advanceTimersByTimeAsync(1_000)
+    expect(batches).toEqual([[0]])
+
+    controller.enqueue(event(1))
+    await vi.advanceTimersByTimeAsync(199)
+    expect(batches).toEqual([[0]])
+    await vi.advanceTimersByTimeAsync(1)
+    expect(batches).toEqual([[0], [0, 1]])
+    await controller.flush()
+  })
+
+  it('observes an overlapping background failure and retries it inside flush', async () => {
+    vi.useFakeTimers()
+    const gate = Promise.withResolvers<boolean>()
+    const report = vi.fn()
+    const batches: number[][] = []
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => {
+        batches.push(events.map(item => item.seq))
+        if (batches.length === 1) {
+          await gate.promise
+          throw new Error('transient')
+        }
+      },
+      reportBackgroundFailure: report,
+    })
+
+    controller.enqueue(event(0))
+    await vi.advanceTimersByTimeAsync(200)
+    const first = controller.flush()
+    const second = controller.flush()
+    gate.resolve(true)
+
+    await expect(Promise.all([first, second])).resolves.toEqual([undefined, undefined])
+    expect(batches).toEqual([[0], [0]])
+    expect(report).toHaveBeenCalledOnce()
+    expect(controller.hasWork).toBe(false)
+  })
+
+  it('surfaces a barrier failure without detached logging and retains its batch', async () => {
+    vi.useFakeTimers()
+    const failure = new Error('durability failed')
+    const report = vi.fn()
+    const batches: number[][] = []
+    let attempt = 0
+    const controller = new SessionWriteBehind({
+      maxDelayMs: 200,
+      write: async (events) => {
+        batches.push(events.map(item => item.seq))
+        if (++attempt === 1) throw failure
+      },
+      reportBackgroundFailure: report,
+    })
+
+    controller.enqueue(event(0))
+    await expect(controller.flush()).rejects.toBe(failure)
+    expect(report).not.toHaveBeenCalled()
+    expect(controller.hasWork).toBe(true)
+
+    controller.enqueue(event(1))
+    await vi.advanceTimersByTimeAsync(200)
+    expect(batches).toEqual([[0], [0, 1]])
+    await controller.flush()
+  })
+})