Merge latest master into codex/simp-session-log-representation

docs/event-producer-consumer.md

@@ -26,7 +26,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac
 | `fs/write-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:53`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) |
 | `llm/stream` | `waterfall` | [`packages/llm/llm/src/index.ts:39`](../packages/llm/llm/src/index.ts) | [`llm`](../packages/llm/llm) (`waterfall`) | [`invariants`](../packages/support/invariants), [`llm-replay`](../packages/support/llm-replay) |
 | `session/created` | `emit` | [`packages/core/session/src/index.ts:47`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence) |
-| `session/disposed` | `emit` | [`packages/core/session/src/index.ts:57`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | - |
+| `session/disposed` | `emit` | [`packages/core/session/src/index.ts:57`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`session-persistence`](../packages/session-persistence/session-persistence) |
 | `session/event` | `emit` | [`packages/core/session/src/index.ts:69`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`acp`](../packages/ui/acp), [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence), [`stdio`](../packages/ui/stdio), [`workspace-context`](../packages/context/workspace-context) |
 | `session/flush` | `parallel` | [`packages/core/session/src/index.ts:79`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`session-persistence`](../packages/session-persistence/session-persistence) |
 | `subagent/end` | `emit` | [`packages/subagent/subagent/src/index.ts:108`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`jsonrpc`](../packages/ui/jsonrpc) |

docs/rfc/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md

@@ -12,6 +12,8 @@ Extract a backend-agnostic `PersistenceCoordinator` into `dsh-session-persistenc
 
 Composition, not inheritance. The coordinator is a concrete class the backend holds, not a base class the backend extends. The RFC's risk — "a coordinator must not make unusual backends fight an inheritance hierarchy" — is avoided: a backend exposes only the hooks; it cannot reach the coordinator's private orchestration state, and the public `SessionPersistence` service shape is unchanged, so a third-party backend MAY still implement the abstract service directly without the coordinator at all.
 
+The coordinator retires each live session from its `session/disposed` notification: it waits for that exact Session object's initialization, serializes a final drain, and then removes the owned state, buffer, and init entries. Failed drains retain their buffers for backend teardown to retry. Settled per-id chain tails remove themselves only when they are still the current tail, so a completion cannot erase a newer operation for the same id. Backend teardown unregisters the write-path listeners before awaiting all admitted retirements, remaining buffers, and chains, then closes the backend.
+
 ### The hook interface (`PersistenceBackend<TornMarker>`)
 
 Six methods (five required + an optional lifecycle hook) — the only seam between the coordinator and storage:
@@ -30,7 +32,7 @@ The single design choice that keeps the seam clean: the crash-repair "where is t
 
 ## Testing
 
-The shared `runPersistenceContract` (public-API contract) keeps running for every backend. A new `runCoordinatorContract` (`tests/coordinator-contract.ts`) holds the write-path orchestration — adoption, HMR, collision, dispose-drain, crash-tail repair — and runs once per backend through a `CoordinatorFixture` (an in-memory reference + jsonl + sqlite). The per-backend specs shrank to storage mechanics only (JSONL: path safety, fsync rollback, bucket listing; SQLite: schema version, `scanRows`, transaction rollback). A through-coordinator torn-tail→load→`commitRepair` test per real backend (via a `corruptTail` fixture hook) keeps the coordinator's torn-marker repair branch covered under the 100% per-file gate — the contract crash test only produces synthetic closers, never a torn marker, so it could not reach that branch.
+The shared `runPersistenceContract` (public-API contract) keeps running for every backend. `runCoordinatorContract` (`tests/coordinator-contract.ts`) holds the write-path orchestration — adoption, HMR, collision, session and backend disposal drains, and crash-tail repair — and runs once per backend through a `CoordinatorFixture` (an in-memory reference + jsonl + sqlite). Coordinator-specific tests pin retirement map cleanup, same-id chain-tail races, failed-drain retry, and close ordering. The per-backend specs retain storage mechanics only (JSONL: path safety, fsync rollback, bucket listing; SQLite: schema version, `scanRows`, transaction rollback). A through-coordinator torn-tail→load→`commitRepair` test per real backend (via a `corruptTail` fixture hook) keeps the coordinator's torn-marker repair branch covered under the 100% per-file gate — the contract crash test only produces synthetic closers, never a torn marker, so it could not reach that branch.
 
 ## Alternatives considered
 
@@ -39,4 +41,4 @@ The shared `runPersistenceContract` (public-API contract) keeps running for ever
 
 ## Consequences
 
-The coordinator adds one indirection and an opaque torn marker, but centralizes correctness-heavy orchestration previously duplicated by every backend. Its hook surface stays narrow: collision checks reuse `loadStored`, materialization stays atomic inside `appendBatch`, and listing bypasses the coordinator. New backends implement storage primitives rather than copy the event-buffer-flush lifecycle.
+The coordinator adds one indirection, an opaque torn marker, and detached session-retirement tasks, but centralizes correctness-heavy orchestration previously duplicated by every backend. Session disposal remains an observe-only event, so the session owner does not await persistence retirement; the coordinator contains failures, preserves uncommitted buffers, and makes backend teardown the quiescence boundary. Its hook surface stays narrow: collision checks reuse `loadStored`, materialization stays atomic inside `appendBatch`, and listing bypasses the coordinator. New backends implement storage primitives rather than copy the event-buffer-flush lifecycle.

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

@@ -25,6 +25,8 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 
 `PersistenceCoordinator` owns per-id state, write-behind buffers and serialization, the `session/event` → `session/flush` drain, 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 four public service methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives; see the [coordinator RFC](../../../docs/rfc/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
 
+When a live session emits `session/disposed`, the coordinator waits for its initialization, serializes a final buffer drain, then releases every map entry owned by that exact `Session` object. A failed final drain keeps the pending buffer for backend teardown to retry. Backend teardown stops event admission first, awaits all in-flight session retirements and remaining per-id operations, drains any retained buffers, and only then closes the storage handle.
+
 The side-effect-free `locate` query remains backend-owned because it describes storage topology rather than write orchestration.
 
 The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinator and storage):

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

@@ -140,7 +140,8 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId):
  *
  * All per-id operations are serialized (a per-id promise chain) so concurrent
  * flushes / a flush racing a load never interleave storage writes. The
- * constructor installs the write-path listeners and the dispose effect.
+ * constructor installs the write-path listeners, per-session retirement, and
+ * the backend dispose effect.
  *
  * @typeParam TornMarker - the backend's opaque torn-tail repair token.
  */
@@ -160,6 +161,8 @@ export class PersistenceCoordinator<TornMarker = unknown> {
    * observation boundary; callers do not inspect this bookkeeping directly.
    */
   private inits = new Map<Session, Promise<void>>()
+  /** Final drains started by fire-and-forget session disposal notifications. */
+  private retirements = new Set<Promise<void>>()
 
   constructor(private ctx: Context, private backend: PersistenceBackend<TornMarker>) {
     this.installWritePath()
@@ -287,7 +290,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     const next = prior.then(op, op)
     // Keep the chain alive but swallow this op's rejection for the NEXT waiter
     // (the caller still sees the real rejection via `next`).
-    this.chains.set(id, next.then(() => undefined, () => undefined))
+    const tail = next.then(() => undefined, () => undefined)
+    this.chains.set(id, tail)
+    // Settled tails carry no serialization value. Delete only the exact tail
+    // installed above: a later operation may already have replaced it.
+    void tail.then(() => {
+      if (this.chains.get(id) === tail) this.chains.delete(id)
+    })
     return next
   }
 
@@ -313,27 +322,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   private installWritePath(): void {
     const ctx = this.ctx
 
-    // Capture the header on creation; persist a fork's seed once. Record the init
-    // promise so flush/dispose can await it (onCreated is async).
-    ctx.on('session/created', (session) => { void this.initFor(session) })
-
-    // Session emits an owned frozen event. Keep a persistence-owned copy anyway
-    // so the write-behind queue owns exactly the record it will flush rather than
-    // retaining a product-layer record by identity. Serializability is guaranteed
-    // at the source, so structuredClone is safe.
-    ctx.on('session/event', (session, event) => {
-      let buffer = this.buffers.get(session)
-      if (!buffer) this.buffers.set(session, buffer = [])
-      buffer.push(structuredClone(event))
-    })
-
-    // Drain to the backend at the durability checkpoint.
-    ctx.on('session/flush', session => this.flush(session))
-
-    // Dispose must reach quiescence: await every init + final drain BEFORE
-    // returning, then close the backend's own resources (AFTER the drain), so no
-    // write lands after teardown and a close failure never MASKS a drain error.
+    // Register the disposer BEFORE the listeners. Cordis tears effects down in
+    // reverse registration order, so event admission closes before this final
+    // drain reaches quiescence and closes the backend.
     ctx.effect(() => async () => {
+      await this.awaitRetirements()
+
       let disposeError: unknown
       try {
         const errors = [
@@ -361,11 +355,63 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       }
     }, `${this.backend.name} write path`)
 
+    // Capture the header on creation; persist a fork's seed once. Record the init
+    // promise so flush/dispose can await it (onCreated is async).
+    ctx.on('session/created', (session) => { void this.initFor(session) })
+
+    // Session emits an owned frozen event. Keep a persistence-owned copy anyway
+    // so the write-behind queue owns exactly the record it will flush rather than
+    // retaining a product-layer record by identity. Serializability is guaranteed
+    // at the source, so structuredClone is safe.
+    ctx.on('session/event', (session, event) => {
+      let buffer = this.buffers.get(session)
+      if (!buffer) this.buffers.set(session, buffer = [])
+      buffer.push(structuredClone(event))
+    })
+
+    // Drain to the backend at the durability checkpoint.
+    ctx.on('session/flush', session => this.flush(session))
+
+    // Session disposal is observe-only, so the coordinator observes the
+    // detached task itself and backend teardown awaits quiescence.
+    ctx.on('session/disposed', (session) => { this.retire(session) })
+
     // HMR: a hot reload does not replay session/created, so seed existing live
     // sessions (mirrors dsh-invariants).
     for (const session of ctx.sessions.list()) void this.initFor(session)
   }
 
+  /** Start, observe, and track one disposed session's final drain. */
+  private retire(session: Session): void {
+    const task = this.retireCore(session)
+    this.retirements.add(task)
+    const settled = (): void => { this.retirements.delete(task) }
+    void task.then(settled, (error: unknown) => {
+      settled()
+      this.ctx.logger.warn(`${this.backend.name}: session "${session.id}" retirement failed: ${String(error)}`)
+    })
+  }
+
+  /** Drain and release state owned by one exact disposed Session lifecycle. */
+  private async retireCore(session: Session): Promise<void> {
+    await this.inits.get(session)
+
+    const id = session.header.id
+    await this.serialize(id, async () => {
+      await this.drain(session)
+      this.buffers.delete(session)
+      this.inits.delete(session)
+      if (this.states.get(id)?.owner === session) this.states.delete(id)
+    })
+  }
+
+  /** Await every retirement admitted before listener teardown. */
+  private async awaitRetirements(): Promise<void> {
+    while (this.retirements.size > 0) {
+      await Promise.allSettled([...this.retirements])
+    }
+  }
+
   /** Start (once) the async init for a session and remember its promise. */
   private initFor(session: Session): Promise<void> {
     const existing = this.inits.get(session)

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

@@ -9,7 +9,7 @@
  * @module @deepseek-ai/dsh-session-persistence/tests/coordinator-contract
  */
 
-import { describe, expect, it } from 'vitest'
+import { describe, expect, it, vi } 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'
@@ -401,7 +401,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       }
     })
 
-    it('does NOT reclaim an id whose abandoned owner still has buffered (unflushed) events', async () => {
+    it('session disposal drains buffered events before retiring ownership', async () => {
       const fix = await makeFixture()
       const { ctx, fiber } = await freshCtx(fix)
       try {
@@ -413,13 +413,20 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         // 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' } })
-        await firstFiber.dispose() // disposed before flush; not materialized, buffer pending
+        await firstFiber.dispose()
+
+        // Disposal is an observe-only notification. Poll storage rather than
+        // assuming the owning fiber awaits the coordinator's detached drain.
+        await vi.waitFor(async () => {
+          expect((await ctx.sessionPersistence.list()).map(meta => meta.id)).toContain(SessionId('buffered'))
+        })
+        expect((await ctx.sessionPersistence.load(SessionId('buffered'))).events.map(event => event.seq)).toEqual([0, 1])
 
         let reuse!: Session
         await ctx.plugin(Object.assign((inner: Context) => {
           reuse = inner.sessions.create(SessionId('buffered'), { meta: { cwd: WORK } })
         }, { inject: ['sessions'] }))
-        await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/already bound to a different live session/)
+        await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/persisted log|id collision/)
       } finally {
         await fiber.dispose()
         await fix.cleanup()

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

@@ -1,7 +1,7 @@
-import { describe, expect, it } from 'vitest'
+import { describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import SessionStore, { SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
-import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
+import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
   SessionPersistence, PersistenceCoordinator,
   type PersistenceBackend, type StoredPrefix,
@@ -35,6 +35,15 @@ function legacyFallbackHeader(seq = 0): SessionEvent {
 /** Optional plugin config: an EXTERNAL store shared across backend instances. */
 interface MemoryConfig { store?: MemoryStore }
 
+/** Test-only view of the coordinator containers whose retirement is the contract under test. */
+interface CoordinatorInternals {
+  states: Map<unknown, unknown>
+  buffers: Map<unknown, unknown>
+  chains: Map<unknown, unknown>
+  inits: Map<unknown, unknown>
+  retirements: Set<Promise<void>>
+}
+
 /**
  * Reference {@link PersistenceCoordinator} vehicle and abstract-service coverage, backed by a
  * dependency-free map with atomic writes and no torn-tail marker. Supplying the map lets multiple
@@ -121,6 +130,49 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
   }
 }
 
+/** Controllable storage primitive for serialization and retirement failure tests. */
+class ControlledBackend implements PersistenceBackend<never> {
+  readonly name = 'session-persistence-controlled'
+  readonly store: MemoryStore = new Map()
+  readonly lifecycle: string[] = []
+  appendAttempts = 0
+  loadAttempts = 0
+  beforeAppend?: (attempt: number) => Promise<void>
+  beforeLoadStored?: (attempt: number) => Promise<void>
+
+  async loadStored(id: SessionId): Promise<StoredPrefix<never> | undefined> {
+    await this.beforeLoadStored?.(++this.loadAttempts)
+    const entry = this.store.get(id)
+    if (entry === undefined) return undefined
+    return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) }
+  }
+
+  loadLive(id: SessionId, _cwd: string | undefined): Promise<StoredPrefix<never> | undefined> {
+    return this.loadStored(id)
+  }
+
+  async appendBatch(m: SessionHeader, events: readonly SessionEvent[], _isMaterialized: boolean): Promise<void> {
+    const attempt = ++this.appendAttempts
+    await this.beforeAppend?.(attempt)
+    const entry = this.store.get(m.id)
+    if (entry === undefined) {
+      this.store.set(m.id, { meta: structuredClone(m), events: structuredClone(events) as SessionEvent[] })
+    } else {
+      entry.events.push(...structuredClone(events) as SessionEvent[])
+    }
+  }
+
+  async commitRepair(_m: SessionHeader, _tornMarker: undefined, _closers: readonly SessionEvent[]): Promise<void> {}
+
+  async list(): Promise<SessionHeader[]> {
+    return [...this.store.values()].map(entry => structuredClone(entry.meta))
+  }
+
+  async close(): Promise<void> {
+    this.lifecycle.push('close')
+  }
+}
+
 // Run the shared contract against the in-memory backend.
 runPersistenceContract('memory', async () => {
   const ctx = new Context()
@@ -142,6 +194,230 @@ runCoordinatorContract('memory', async (): Promise<CoordinatorFixture> => {
   }
 })
 
+describe('PersistenceCoordinator retirement', () => {
+  it('a retiring unmaterialized owner without buffered events releases its id', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    let coordinator!: PersistenceCoordinator<never>
+    const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const loadGate = Promise.withResolvers<boolean>()
+
+    try {
+      const id = SessionId('retiring-lazy-owner')
+      let first!: Session
+      const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        first = inner.sessions.create(id)
+      }, { inject: ['sessions'] }))
+      await ctx.sessions.flush(first)
+
+      const baselineLoads = backend.loadAttempts
+      backend.beforeLoadStored = async () => { await loadGate.promise }
+      const blockingLoad = coordinator.load(id)
+      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) })
+      await firstFiber.dispose()
+
+      let reuse!: Session
+      await ctx.plugin(Object.assign((inner: Context) => {
+        reuse = inner.sessions.create(id)
+      }, { inject: ['sessions'] }))
+      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 2) })
+
+      loadGate.resolve(true)
+      await expect(blockingLoad).rejects.toThrow(/not found/)
+      await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined()
+    } finally {
+      loadGate.resolve(true)
+      await backendFiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('a retiring owner with buffered events still rejects same-id reuse', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    let coordinator!: PersistenceCoordinator<never>
+    const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const loadGate = Promise.withResolvers<boolean>()
+
+    try {
+      const id = SessionId('retiring-buffered-owner')
+      let first!: Session
+      const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        first = inner.sessions.create(id)
+      }, { inject: ['sessions'] }))
+      await ctx.sessions.flush(first)
+      first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+      first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+
+      const baselineLoads = backend.loadAttempts
+      backend.beforeLoadStored = async () => { await loadGate.promise }
+      const blockingLoad = coordinator.load(id)
+      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) })
+      await firstFiber.dispose()
+
+      let reuse!: Session
+      await ctx.plugin(Object.assign((inner: Context) => {
+        reuse = inner.sessions.create(id)
+      }, { inject: ['sessions'] }))
+      await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/bound to a different live session/)
+
+      loadGate.resolve(true)
+      await expect(blockingLoad).rejects.toThrow(/not found/)
+      await vi.waitFor(() => {
+        expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1])
+      })
+    } finally {
+      loadGate.resolve(true)
+      await backendFiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('a settled chain tail cannot delete a newer operation for the same id', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const internals = coordinator as unknown as CoordinatorInternals
+    const first = Promise.withResolvers<boolean>()
+    const second = Promise.withResolvers<boolean>()
+    backend.beforeAppend = async (attempt) => {
+      if (attempt === 1) await first.promise
+      if (attempt === 2) await second.promise
+    }
+
+    try {
+      const id = SessionId('chain-tail')
+      await coordinator.create(meta(id))
+      const firstAppend = coordinator.append(id, [{
+        type: 'turn/start',
+        seq: 0,
+        time: 1,
+        data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
+      }])
+      const secondAppend = coordinator.append(id, [{
+        type: 'turn/end',
+        seq: 1,
+        time: 2,
+        data: { turn: 1, reason: { kind: 'completed' } },
+      }])
+
+      await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
+      first.resolve(true)
+      await vi.waitFor(() => { expect(backend.appendAttempts).toBe(2) })
+      expect(internals.chains.size).toBe(1)
+      second.resolve(true)
+      await Promise.all([firstAppend, secondAppend])
+      await vi.waitFor(() => { expect(internals.chains.size).toBe(0) })
+      expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1])
+    } finally {
+      first.resolve(true)
+      second.resolve(true)
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('backend teardown retries a failed session retirement before close', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    let coordinator!: PersistenceCoordinator<never>
+    const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const internals = coordinator as unknown as CoordinatorInternals
+    backend.beforeAppend = async (attempt) => {
+      if (attempt === 1) {
+        backend.lifecycle.push('append-failed')
+        throw new Error('transient append failure')
+      }
+      backend.lifecycle.push('append-committed')
+    }
+
+    try {
+      let session!: Session
+      const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        session = inner.sessions.create(SessionId('retry-retirement'))
+      }, { inject: ['sessions'] }))
+      session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+      session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+      await sessionFiber.dispose()
+
+      await vi.waitFor(() => {
+        expect(backend.appendAttempts).toBe(1)
+        expect(internals.retirements.size).toBe(0)
+      })
+      expect([...internals.buffers.values()]).toEqual([expect.arrayContaining([
+        expect.objectContaining({ seq: 0 }),
+        expect.objectContaining({ seq: 1 }),
+      ])])
+
+      await backendFiber.dispose()
+      expect(backend.store.get(SessionId('retry-retirement'))?.events.map(event => event.seq)).toEqual([0, 1])
+      expect(backend.lifecycle).toEqual(['append-failed', 'append-committed', 'close'])
+    } finally {
+      await backendFiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('backend teardown waits for an in-flight session retirement before close', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    let coordinator!: PersistenceCoordinator<never>
+    const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const internals = coordinator as unknown as CoordinatorInternals
+    const appendGate = Promise.withResolvers<boolean>()
+    backend.beforeAppend = async () => {
+      backend.lifecycle.push('append-started')
+      await appendGate.promise
+      backend.lifecycle.push('append-committed')
+    }
+
+    try {
+      let session!: Session
+      const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        session = inner.sessions.create(SessionId('inflight-retirement'))
+      }, { inject: ['sessions'] }))
+      session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+      session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+      await sessionFiber.dispose()
+      await vi.waitFor(() => {
+        expect(backend.appendAttempts).toBe(1)
+        expect(internals.retirements.size).toBe(1)
+      })
+
+      let disposed = false
+      const teardown = backendFiber.dispose().then(() => { disposed = true })
+      await Promise.resolve()
+      expect(disposed).toBe(false)
+      expect(backend.lifecycle).toEqual(['append-started'])
+
+      appendGate.resolve(true)
+      await teardown
+      expect(backend.store.get(SessionId('inflight-retirement'))?.events.map(event => event.seq)).toEqual([0, 1])
+      expect(backend.lifecycle).toEqual(['append-started', 'append-committed', 'close'])
+    } finally {
+      appendGate.resolve(true)
+      await backendFiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+})
+
 describe('SessionPersistence service registration', () => {
   it('registers as ctx.sessionPersistence and is removed on fiber dispose (HMR safety)', async () => {
     const ctx = new Context()
@@ -233,4 +509,37 @@ describe('SessionPersistence service registration', () => {
       .rejects.toThrow('unsupported legacy request/header reason "fallback" at seq 0')
     await fiber.dispose()
   })
+
+  it('retires all coordinator bookkeeping for disposed sessions', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(MemoryPersistence)
+    const { coordinator } = ctx.sessionPersistence as unknown as { coordinator: CoordinatorInternals }
+
+    try {
+      for (let index = 0; index < 3; index += 1) {
+        let session!: Session
+        const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => {
+          session = inner.sessions.create(SessionId(`disposed-${index}`))
+        }, { inject: ['sessions'] }))
+        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)
+        await sessionFiber.dispose()
+      }
+
+      await vi.waitFor(() => {
+        expect(ctx.sessions.list()).toHaveLength(0)
+        expect({
+          states: coordinator.states.size,
+          buffers: coordinator.buffers.size,
+          chains: coordinator.chains.size,
+          inits: coordinator.inits.size,
+          retirements: coordinator.retirements.size,
+        }).toEqual({ states: 0, buffers: 0, chains: 0, inits: 0, retirements: 0 })
+      })
+    } finally {
+      await fiber.dispose()
+    }
+  })
 })