diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index 225431c8b6..ff66e3a4a3 100644 --- a/docs/event-producer-consumer.md +++ b/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) | diff --git a/docs/rfc/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md b/docs/rfc/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md index b1773b480d..fef6ad034c 100644 --- a/docs/rfc/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md +++ b/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`) 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. diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index a294575fd9..3390b2cc30 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/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` hooks (the only seam between the coordinator and storage): diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index a3c24b32bf..beb5fca483 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/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 { * observation boundary; callers do not inspect this bookkeeping directly. */ private inits = new Map>() + /** Final drains started by fire-and-forget session disposal notifications. */ + private retirements = new Set>() constructor(private ctx: Context, private backend: PersistenceBackend) { this.installWritePath() @@ -287,7 +290,13 @@ export class PersistenceCoordinator { 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 { 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 { } }, `${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 { + 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 { + 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 { const existing = this.inits.get(session) diff --git a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts index f6692b9175..defcaefb6d 100644 --- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts +++ b/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() diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 0b044b8e90..e083ac3543 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/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 + buffers: Map + chains: Map + inits: Map + retirements: Set> +} + /** * 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 { + readonly name = 'session-persistence-controlled' + readonly store: MemoryStore = new Map() + readonly lifecycle: string[] = [] + appendAttempts = 0 + loadAttempts = 0 + beforeAppend?: (attempt: number) => Promise + beforeLoadStored?: (attempt: number) => Promise + + async loadStored(id: SessionId): Promise | 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 | undefined> { + return this.loadStored(id) + } + + async appendBatch(m: SessionHeader, events: readonly SessionEvent[], _isMaterialized: boolean): Promise { + 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 {} + + async list(): Promise { + return [...this.store.values()].map(entry => structuredClone(entry.meta)) + } + + async close(): Promise { + 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 => { } }) +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 + const backendFiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + const loadGate = Promise.withResolvers() + + 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 + const backendFiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + const loadGate = Promise.withResolvers() + + 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 + 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() + const second = Promise.withResolvers() + 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 + 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 + 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() + 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() + } + }) })