feat(schedule): add durable after reminders

packages/session/session-persistence/README.md

@@ -31,7 +31,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 
 `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. 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.
+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, then returns the Session Store's literal `true` durability acknowledgement. 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/session-persistence/README.zh.md

@@ -31,7 +31,7 @@
 
 `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。第一个待处理事件会开启固定批处理窗口；后续事件会加入该批次，但不会重置截止时间。配置的 `writeBatchMaxDelayMs` 只限制这段有意等待，而不限制事件循环、初始化、串行化操作或后端延迟。写入期间接纳的事件会形成一个新的有界批次。`session/flush` 会取消等待，并作为共享的完全停稳屏障，排空屏障运行期间接纳的事件。后台写入失败只记录一次日志，保留顺序不变的批次，并暂停自动重试；新事件会开启新的固定窗口，而显式 flush 或后端拆卸会立即重试，并在失败再次发生时向调用方暴露失败。
+每个 `session/event` 将事件复制到会话 controller。第一个待处理事件会开启固定批处理窗口；后续事件会加入该批次，但不会重置截止时间。配置的 `writeBatchMaxDelayMs` 只限制这段有意等待，而不限制事件循环、初始化、串行化操作或后端延迟。写入期间接纳的事件会形成一个新的有界批次。`session/flush` 会取消等待，并作为共享的完全停稳屏障，排空屏障运行期间接纳的事件，然后返回 Session Store 所需的字面量 `true` 持久化确认。后台写入失败只记录一次日志，保留顺序不变的批次，并暂停自动重试；新事件会开启新的固定窗口，而显式 flush 或后端拆卸会立即重试，并在失败再次发生时向调用方暴露失败。
 
 崩溃修复只适用于冷状态。对于实时 id，`load(id)` 为权威内存日志制作快照，等待该快照持久，并只在平衡时返回；开放实时轮次会被拒绝，而不会收到合成中断 closer。对于冷 id，检查只读取、验证、冻结并构造一次未发布 Session；只有来源修订值仍然是当前值时，重复检查才会复用该对象图。`prepare(id)` 在修复前执行相同校验，预留该 Session 本身，提交任何待处理的撕裂尾部或中断轮次修复，并将其返回用于发布。HMR 接管通过 `loadStored` 读取，应用协调器 cwd 检查，并绝不关闭活动轮次。
 

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

@@ -1038,8 +1038,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       live.writes.enqueue(event)
     })
 
-    // Callers use flush as the immediate durability barrier for buffered writes.
-    ctx.on('session/flush', session => this.flush(session))
+    // A completed bounded drain acknowledges the caller's durability barrier.
+    ctx.on('session/flush', async (session) => {
+      await this.flush(session)
+      return true as const
+    })
 
     // Session disposal is observe-only, so retirement contains its own failure.
     ctx.on('session/disposed', (session) => { this.retire(session) })
@@ -1089,8 +1092,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       writes: this.createWriteBehind(session, () => live.init),
     }
     this.live.set(session, live)
-    live.init = this.serialize(session.header.id, () => this.onCreated(session, seed))
-    live.init.catch(() => { /* observed by flush/dispose through the controller */ })
+    void this.ensureInitialized(session, live).catch(() => {
+      /* observed by flush/dispose through the controller or retried by a later barrier */
+    })
     return live
   }
 
@@ -1151,8 +1155,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     const tracked = this.states.get(id)
     if (tracked !== undefined) {
       // case 1: already tracked.
-      /* v8 ignore next -- initFor dedupes per session object; same-object re-entry can't occur */
-      if (tracked.owner === session) return
+      if (tracked.owner === session) {
+        await this.reconcileOwnedSeed(session, seed, tracked)
+        return
+      }
       if (tracked.owner === undefined) {
         // Ownerless state from the public create()/load() API. The FIRST live
         // session claims it — but ONLY if BOTH the cwd scope and the seed match.
@@ -1205,6 +1211,43 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     if (seed.length > 0) await this.appendCore(id, seed)
   }
 
+  /**
+   * Reconcile a retrying live owner with the backend's actual durable cursor.
+   * An initialization write may have committed before its promise rejected, so
+   * retry from storage rather than from the coordinator's last acknowledged
+   * cursor. This also completes a suffix whose first attempt never committed.
+   */
+  private async reconcileOwnedSeed(
+    session: Session,
+    seed: readonly SessionEvent[],
+    tracked: SessionState,
+  ): Promise<void> {
+    const stored = await this.backend.loadStored(session.header.id)
+    if (stored === undefined) {
+      if (tracked.materialized || tracked.cursor !== 0) {
+        throw new Error(`session "${session.header.id}" lost its persisted artifact during live initialization`)
+      }
+      if (seed.length > 0) await this.appendCore(session.header.id, seed)
+      return
+    }
+    const { meta, events, tornMarker } = stored
+    this.assertStoredId(session.header.id, meta)
+    if (meta.cwd !== session.header.cwd) {
+      throw new Error(`session "${session.header.id}" is already persisted at a different cwd (persisted: ${String(meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`)
+    }
+    this.assertVersion(meta)
+    const storedEvents = snapshotStoredEvents(events, session.header.id)
+    if (!seedCoversPrefix(seed, storedEvents)) {
+      throw new Error(`session "${session.header.id}" already has a persisted log on disk that does not match this live session (id collision)`)
+    }
+    if (tornMarker !== undefined) await this.backend.commitRepair(meta, tornMarker, [])
+    tracked.meta = { ...meta }
+    tracked.cursor = storedEvents.length
+    tracked.materialized = true
+    const suffix = seed.slice(storedEvents.length)
+    if (suffix.length > 0) await this.appendCore(session.header.id, suffix)
+  }
+
   /**
    * Adopt a stored prefix as a live session's history (HMR/reload): verify the
    * seed covers the stored prefix, truncate any torn tail (NOT the open turn —

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

@@ -182,6 +182,7 @@ class ControlledBackend implements PersistenceBackend<never> {
   loadAttempts = 0
   repairAttempts = 0
   beforeAppend?: (attempt: number) => Promise<void>
+  afterAppend?: (attempt: number) => Promise<void>
   beforeLoadStored?: (attempt: number, signal?: AbortSignal) => Promise<void>
   /** When set, the declared seek hook delegates here so readFrom exercises it; unset throws (tests set it first). */
   seekHook?: (id: SessionId, fromSeq: number, signal?: AbortSignal) => Promise<StoredSuffix | undefined>
@@ -218,6 +219,7 @@ class ControlledBackend implements PersistenceBackend<never> {
     } else {
       entry.events.push(...structuredClone(events) as SessionEvent[])
     }
+    await this.afterAppend?.(attempt)
   }
 
   async commitRepair(m: SessionHeader, _tornMarker: undefined, closers: readonly SessionEvent[]): Promise<void> {
@@ -373,6 +375,130 @@ describe('PersistenceCoordinator bounded writes', () => {
   })
 })
 
+describe('PersistenceCoordinator retryable live initialization', () => {
+  it('retries a rejected first storage read for a new empty session', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const loadGate = Promise.withResolvers<undefined>()
+    const retryGate = Promise.withResolvers<undefined>()
+    backend.beforeLoadStored = async (attempt) => {
+      if (attempt === 1) {
+        await loadGate.promise
+        throw new Error('transient init read failure')
+      }
+      if (attempt === 2) await retryGate.promise
+    }
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+
+    try {
+      const session = ctx.sessions.create(SessionId('retry-new-empty'))
+      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) })
+      const first = ctx.sessions.flush(session)
+      loadGate.resolve(undefined)
+      await expect(first).rejects.toThrow('transient init read failure')
+      const retries = [ctx.sessions.flush(session), ctx.sessions.flush(session)]
+      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(2) })
+      retryGate.resolve(undefined)
+      await expect(Promise.all(retries)).resolves.toEqual([true, true])
+      // The one shared retry performs the normal new-session probe and
+      // createCore's collision recheck; a second initialization would add two
+      // more reads.
+      expect(backend.loadAttempts).toBe(3)
+
+      session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+      session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+      await ctx.sessions.flush(session)
+      expect(backend.store.get(session.id)?.events.map(event => event.seq)).toEqual([0, 1])
+
+      const live = [...(coordinator as unknown as CoordinatorInternals).live.values()][0]
+      expect(live).toMatchObject({ seedEnd: 0, initialized: true })
+      expect(live).not.toHaveProperty('seed')
+    } finally {
+      loadGate.resolve(undefined)
+      retryGate.resolve(undefined)
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('uses the backend cursor when a fork seed committed before initialization rejected', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const appendGate = Promise.withResolvers<undefined>()
+    backend.afterAppend = async (attempt) => {
+      if (attempt === 1) {
+        await appendGate.promise
+        throw new Error('uncertain init write')
+      }
+    }
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+
+    try {
+      const seed = oneTurnLog()
+      const session = ctx.sessions.create(SessionId('retry-fork-seed'), {
+        seed,
+        meta: { cwd: '/w', seedLength: seed.length },
+      })
+      await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
+      const first = ctx.sessions.flush(session)
+      appendGate.resolve(undefined)
+      await expect(first).rejects.toThrow('uncertain init write')
+      await expect(ctx.sessions.flush(session)).resolves.toBe(true)
+
+      expect(backend.appendAttempts).toBe(1)
+      expect(backend.store.get(session.id)?.events.map(event => event.seq))
+        .toEqual([0, 1, 2, 3, 4, 5, 6])
+    } finally {
+      appendGate.resolve(undefined)
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('retries only a missing suffix after stored-session adoption rejects', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const id = SessionId('retry-resume-adoption')
+    const stored = oneTurnLog()
+    backend.store.set(id, { meta: meta(id, '/w'), events: structuredClone(stored) })
+    const appendGate = Promise.withResolvers<undefined>()
+    backend.beforeAppend = async (attempt) => {
+      if (attempt === 1) {
+        await appendGate.promise
+        throw new Error('transient adoption write failure')
+      }
+    }
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+
+    try {
+      const session = ctx.sessions.create(id, { seed: stored, meta: { cwd: '/w' } })
+      await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
+      const first = ctx.sessions.flush(session)
+      appendGate.resolve(undefined)
+      await expect(first).rejects.toThrow('transient adoption write failure')
+      await expect(ctx.sessions.flush(session)).resolves.toBe(true)
+
+      expect(backend.appendAttempts).toBe(2)
+      expect(backend.store.get(id)?.events.map(event => event.seq))
+        .toEqual([0, 1, 2, 3, 4, 5, 6])
+    } finally {
+      appendGate.resolve(undefined)
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+})
+
 describe('PersistenceCoordinator stored identity', () => {
   it('rejects a mismatched backend header before repair or state publication', async () => {
     const ctx = new Context()