Merge remote-tracking branch 'origin/master' into worktree/fix-sqlite-file-permissions
# Conflicts: # docs/config-catalog.md
This commit is contained in:
Yichen Jiang
@@ -19,6 +19,8 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
|
||||
|---|---|---|
|
||||
| `root` | `string` (required) | Root directory for all session files. **No default** — a `process.cwd()` default would scatter files as the process's cwd changes (bash calls, subprocesses). |
|
||||
|
||||
`locate(meta)` returns `{ kind: 'jsonl', path }` using the resolved absolute root and the same cwd-bucket/id encoding as materialization. It performs no filesystem I/O: the target can be returned before the file exists, and an existing file contains only the last flushed prefix.
|
||||
|
||||
## Durability and crash semantics
|
||||
|
||||
- **Lazy materialization.** `create(meta)` writes nothing; on the first `append`, the backend writes and `fsync`s a temporary file, publishes it without overwrite via a hard link, then `fsync`s the directory. A created-but-never-appended session leaves nothing on disk and is absent from `list`.
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
/**
|
||||
* JSONL durable session-persistence backend. It stores a header and contiguous
|
||||
* events in one append-only file per session, and delegates orchestration to
|
||||
* {@link PersistenceCoordinator}.
|
||||
* {@link PersistenceCoordinator}. Its side-effect-free locator returns the
|
||||
* absolute per-session log target before materialization.
|
||||
* @module @deepseek-ai/dsh-session-persistence-jsonl
|
||||
*/
|
||||
|
||||
@@ -12,7 +13,7 @@ import { dirname, resolve } from 'node:path'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import {
|
||||
SessionPersistence, PersistenceCoordinator,
|
||||
type PersistenceBackend, type StoredPrefix,
|
||||
type PersistenceBackend, type SessionLocation, type StoredPrefix,
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
@@ -68,6 +69,11 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
/* jscpd:ignore-start */
|
||||
// --- SessionPersistence service surface (delegated to the coordinator) ---
|
||||
|
||||
/** Resolve the absolute target path without touching the filesystem. */
|
||||
locate(meta: SessionHeader): SessionLocation {
|
||||
return { kind: 'jsonl', path: logPath(this.root, meta.cwd, meta.id) }
|
||||
}
|
||||
|
||||
create(meta: SessionHeader): Promise<void> {
|
||||
return this.coordinator.create(meta)
|
||||
}
|
||||
|
||||
@@ -2,11 +2,11 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { isAbsolute, join, relative, resolve } from 'node:path'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import { encodeSegment, logPath, scanLog, sessionDir } from '../src/format.ts'
|
||||
import { encodeSegment, logPath, scanLog, sessionDir, toHeaderLine } from '../src/format.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
|
||||
|
||||
@@ -112,6 +112,19 @@ describe('SessionPersistenceJsonl: format helpers', () => {
|
||||
it('encodeSegment rejects an empty id', () => {
|
||||
expect(() => encodeSegment('')).toThrow(/empty/)
|
||||
})
|
||||
|
||||
it('resolves a relative custom root before locating a session', async () => {
|
||||
const absoluteRoot = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: relative(process.cwd(), absoluteRoot) })
|
||||
const m = meta('relative-location', '/work')
|
||||
expect(ctx.sessionPersistence.locate(m)).toEqual({
|
||||
kind: 'jsonl',
|
||||
path: logPath(resolve(absoluteRoot), '/work', m.id),
|
||||
})
|
||||
await fiber.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
@@ -126,8 +139,13 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
|
||||
it('lazy materialization: create() writes no file until the first append', async () => {
|
||||
const m = meta('lazy', '/work')
|
||||
const location = ctx.sessionPersistence.locate(m)
|
||||
expect(location).toEqual({ kind: 'jsonl', path: logPath(root, '/work', m.id) })
|
||||
expect(isAbsolute(location!.path)).toBe(true)
|
||||
|
||||
await ctx.sessionPersistence.create(m)
|
||||
// nothing on disk yet
|
||||
// locate() is a pure target-path calculation: neither it nor create()
|
||||
// materializes a file before the first append.
|
||||
const dir = sessionDir(root, '/work')
|
||||
await expect(stat(logPath(root, '/work', m.id))).rejects.toThrow()
|
||||
expect((await ctx.sessionPersistence.list()).map(h => h.id)).not.toContain(m.id)
|
||||
@@ -139,6 +157,26 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
void dir
|
||||
})
|
||||
|
||||
it('keeps the same location on resume and gives a fork its own location', async () => {
|
||||
const parent = meta('location-parent', '/work')
|
||||
const parentLocation = ctx.sessionPersistence.locate(parent)
|
||||
await ctx.sessionPersistence.create(parent)
|
||||
await ctx.sessionPersistence.append(parent.id, oneTurnLog())
|
||||
|
||||
const loaded = await ctx.sessionPersistence.load(parent.id)
|
||||
expect(ctx.sessionPersistence.locate(loaded.meta)).toEqual(parentLocation)
|
||||
|
||||
const child = {
|
||||
...loaded.meta,
|
||||
id: SessionId('location-child'),
|
||||
parentSession: parent.id,
|
||||
seedLength: loaded.events.length,
|
||||
}
|
||||
const childLocation = ctx.sessionPersistence.locate(child)
|
||||
expect(childLocation?.path).not.toBe(parentLocation?.path)
|
||||
expect(childLocation).toEqual({ kind: 'jsonl', path: logPath(root, '/work', child.id) })
|
||||
})
|
||||
|
||||
it('round-trip is byte-identical (incl. assistant/chunk verbatim)', async () => {
|
||||
const m = meta('chunks')
|
||||
const log: SessionEvent[] = [
|
||||
@@ -146,7 +184,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
{ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/chunk', seq: 2, time: 3, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'he' } } },
|
||||
{ type: 'assistant/chunk', seq: 3, time: 4, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'llo' } } },
|
||||
{ type: 'assistant/message', seq: 4, time: 5, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }] }, surfaceOp: 'append', sourceEventSeqs: [2, 3] },
|
||||
{ type: 'assistant/message', seq: 4, time: 5, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append', sourceEventSeqs: [2, 3] },
|
||||
{ type: 'step/end', seq: 5, time: 6, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 6, time: 7, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
@@ -156,6 +194,40 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
expect(loaded.events).toEqual(log) // chunks preserved, contiguous seqs
|
||||
})
|
||||
|
||||
it('rejects a stored v0 log containing a legacy request/header-delta event', async () => {
|
||||
const m = meta('legacy-header-delta', '/legacy')
|
||||
const path = logPath(root, m.cwd, m.id)
|
||||
await mkdir(sessionDir(root, m.cwd), { recursive: true })
|
||||
await writeFile(path, [
|
||||
JSON.stringify(toHeaderLine(m)),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'request/header-delta', seq: 1, time: 2, data: { config: { model: 'legacy' } } }),
|
||||
JSON.stringify({ type: 'turn/end', seq: 2, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
'',
|
||||
].join('\n'))
|
||||
|
||||
await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/unsupported legacy request\/header-delta event at seq 1/)
|
||||
})
|
||||
|
||||
it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
|
||||
const m = meta('legacy-header-fallback', '/legacy')
|
||||
const path = logPath(root, m.cwd, m.id)
|
||||
await mkdir(sessionDir(root, m.cwd), { recursive: true })
|
||||
await writeFile(path, [
|
||||
JSON.stringify(toHeaderLine(m)),
|
||||
JSON.stringify({
|
||||
type: 'request/header',
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: { header: { config: { model: 'legacy' } }, reason: 'fallback' },
|
||||
}),
|
||||
'',
|
||||
].join('\n'))
|
||||
|
||||
await expect(ctx.sessionPersistence.load(m.id))
|
||||
.rejects.toThrow(/unsupported legacy request\/header reason "fallback" at seq 0/)
|
||||
})
|
||||
|
||||
it('persists a forked child seed through the existing session write path', async () => {
|
||||
const source = ctx.sessions.create(SessionId('persist-parent'), { meta: { cwd: '/workspace' } })
|
||||
appendClosedTurn(source)
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
|
||||
A SQLite durable session-persistence backend — a second `SessionPersistence` implementation ([session persistence](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md)), built to validate that the abstract seam and the shared `runPersistenceContract` suite are genuinely backend-agnostic. It satisfies the SAME contract as `dsh-session-persistence-jsonl` (append-only, contiguous-seq, lazy materialization, interrupted-turn close on load), expressed over `node:sqlite` rows instead of file bytes.
|
||||
|
||||
`locate(meta)` returns `undefined`: all sessions share one database, so there is no honest independent per-session transcript path.
|
||||
|
||||
> **TODO:** this backend talks to `node:sqlite` directly. If a cordis database service (`cordis/db` / a `@cordisjs` SQL driver plugin) is adopted, route through that instead of holding a raw `DatabaseSync` here — the contract surface (`SessionPersistence`) would not change, only the storage driver.
|
||||
|
||||
## Storage model
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
/**
|
||||
* SQLite durable session-persistence backend. It maps each session header and
|
||||
* event to rows, and delegates write-path orchestration to
|
||||
* {@link PersistenceCoordinator}.
|
||||
* {@link PersistenceCoordinator}. It has no independent per-session artifact,
|
||||
* so its locator returns `undefined`.
|
||||
* @module @deepseek-ai/dsh-session-persistence-sqlite
|
||||
*/
|
||||
|
||||
@@ -12,7 +13,7 @@ import { mkdir, open } from 'node:fs/promises'
|
||||
import { dirname, resolve } from 'node:path'
|
||||
import {
|
||||
SessionPersistence, PersistenceCoordinator,
|
||||
type PersistenceBackend, type StoredPrefix,
|
||||
type PersistenceBackend, type SessionLocation, type StoredPrefix,
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
@@ -112,6 +113,11 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
|
||||
// --- SessionPersistence service surface (delegated to the coordinator) ---
|
||||
|
||||
/** SQLite has one database, not an independent local artifact per session. */
|
||||
locate(_meta: SessionHeader): SessionLocation | undefined {
|
||||
return undefined
|
||||
}
|
||||
|
||||
create(meta: SessionHeader): Promise<void> {
|
||||
return this.coordinator.create(meta)
|
||||
}
|
||||
|
||||
@@ -153,6 +153,48 @@ describe('scanRows', () => {
|
||||
})
|
||||
|
||||
describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
it('rejects a stored v0 log containing a legacy request/header-delta event', async () => {
|
||||
const path = await freshDbPath()
|
||||
const m = meta('legacy-header-delta', '/legacy')
|
||||
const db = openDatabase(path, 'wal')
|
||||
db.prepare('INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length) VALUES (?, ?, ?, ?, NULL, NULL)')
|
||||
.run(m.id, m.version, m.createdAt, m.cwd ?? null)
|
||||
const insert = db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
|
||||
insert.run(m.id, 0, 'turn/start', 1, JSON.stringify({ turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }))
|
||||
insert.run(m.id, 1, 'request/header-delta', 2, JSON.stringify({ config: { model: 'legacy' } }))
|
||||
insert.run(m.id, 2, 'turn/end', 3, JSON.stringify({ turn: 1, reason: { kind: 'completed' } }))
|
||||
db.close()
|
||||
|
||||
const mounted = await backend(path)
|
||||
await expect(mounted.ctx.sessionPersistence.load(m.id)).rejects.toThrow(/unsupported legacy request\/header-delta event at seq 1/)
|
||||
await mounted.dispose()
|
||||
})
|
||||
|
||||
it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
|
||||
const path = await freshDbPath()
|
||||
const m = meta('legacy-header-fallback', '/legacy')
|
||||
const db = openDatabase(path, 'wal')
|
||||
db.prepare('INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length) VALUES (?, ?, ?, ?, NULL, NULL)')
|
||||
.run(m.id, m.version, m.createdAt, m.cwd ?? null)
|
||||
db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
|
||||
.run(m.id, 0, 'request/header', 1, JSON.stringify({
|
||||
header: { config: { model: 'legacy' } },
|
||||
reason: 'fallback',
|
||||
}))
|
||||
db.close()
|
||||
|
||||
const mounted = await backend(path)
|
||||
await expect(mounted.ctx.sessionPersistence.load(m.id))
|
||||
.rejects.toThrow(/unsupported legacy request\/header reason "fallback" at seq 0/)
|
||||
await mounted.dispose()
|
||||
})
|
||||
|
||||
it('has no independent per-session log location', async () => {
|
||||
const { ctx, dispose } = await backend()
|
||||
expect(ctx.sessionPersistence.locate(meta('sqlite-location'))).toBeUndefined()
|
||||
await dispose()
|
||||
})
|
||||
|
||||
it('an interrupted turn (rows after the last turn/end) is PRESERVED and closed during load', async () => {
|
||||
const path = await freshDbPath()
|
||||
const m = meta('crash')
|
||||
@@ -510,7 +552,7 @@ describe('surface field round-trip', () => {
|
||||
const session = ctx.sessions.create(SessionId('roundtrip-surface'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: 'append', sourceEventSeqs: [0] })
|
||||
session.append('assistant/message', { provenance: { provider: 'mock', model: 'mock' }, turn: 1, step: 1, content: [] }, { surfaceOp: 'append', sourceEventSeqs: [0] })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.parallel('session/flush', session)
|
||||
const loaded = await ctx.sessionPersistence.load(SessionId('roundtrip-surface'))
|
||||
|
||||
@@ -8,6 +8,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
|
||||
|
||||
| Method | Contract |
|
||||
|---|---|
|
||||
| `locate(meta): SessionLocation \| undefined` | Resolve an absolute per-session artifact target without I/O or materialization. Backends without an independent local artifact return `undefined`. |
|
||||
| `create(meta): Promise<void>` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). |
|
||||
| `append(id, events): Promise<void>` | Durably persist a batch (from the `session/flush` drain). Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
|
||||
| `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic closers — an error `tool/result` per unanswered `tool-call`, then `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. |
|
||||
@@ -24,6 +25,10 @@ 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):
|
||||
|
||||
| Hook | Role |
|
||||
@@ -44,9 +49,9 @@ Import `runPersistenceContract` from `tests/contract.ts` (the public-API contrac
|
||||
|
||||
Three backends run these suites: an in-memory reference (in `tests/`), `dsh-session-persistence-jsonl` (append-only file log) and `dsh-session-persistence-sqlite` (`node:sqlite`, each `SessionEvent` one row `(session_id, seq, type, time, data, source_event_seqs, surface_op)`). All passing the same contract + coordinator suite is the proof that the seam is genuinely backend-agnostic — lazy materialization, crash-tail-on-load, and contiguous-seq hold identically over file bytes and over a transactional store.
|
||||
|
||||
## Metadata types
|
||||
## Metadata and location types
|
||||
|
||||
Re-exported from `dsh-session`: `SessionHeader` (immutable session metadata: `version`, `id`, `createdAt`, `cwd?`, `parentSession?`, `seedLength?`).
|
||||
Re-exported from `dsh-session`: `SessionHeader` (immutable session metadata: `version`, `id`, `createdAt`, `cwd?`, `parentSession?`, `seedLength?`). `SessionLocation` is `{ readonly kind: string; readonly path: string }`; its path is an absolute backend target, not proof that the artifact exists or contains an unflushed turn.
|
||||
|
||||
## Model Experience
|
||||
|
||||
|
||||
@@ -118,6 +118,20 @@ function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly Sessio
|
||||
})
|
||||
}
|
||||
|
||||
/** Reject events from an obsolete v0 vocabulary that this build cannot replay. */
|
||||
function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId): void {
|
||||
const legacyType: string = 'request/header-delta'
|
||||
const legacy = events.find(event => event.type === legacyType)
|
||||
if (legacy !== undefined) {
|
||||
throw new Error(`session "${id}" contains unsupported legacy request/header-delta event at seq ${legacy.seq}`)
|
||||
}
|
||||
const fallback = events.find(event => event.type === 'request/header'
|
||||
&& (event.data as { reason?: string }).reason === 'fallback')
|
||||
if (fallback !== undefined) {
|
||||
throw new Error(`session "${id}" contains unsupported legacy request/header reason "fallback" at seq ${fallback.seq}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Owns the backend-agnostic session write-path orchestration. A backend
|
||||
* constructs one (`new PersistenceCoordinator(ctx, this)`), implements
|
||||
@@ -126,7 +140,8 @@ function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly Sessio
|
||||
*
|
||||
* 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.
|
||||
*/
|
||||
@@ -146,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()
|
||||
@@ -204,6 +221,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
}
|
||||
|
||||
private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
|
||||
// Every append route converges here: the public service, live write-behind
|
||||
// drains, and HMR seed/suffix adoption. Keep vocabulary rejection at that
|
||||
// shared boundary so a stale JavaScript plugin cannot persist an event that
|
||||
// this same backend will refuse to load.
|
||||
assertSupportedEvents(events, id)
|
||||
if (events.length === 0) return
|
||||
let state = this.states.get(id)
|
||||
if (state === undefined) state = await this.adopt(id) // calls loadCore, not load
|
||||
@@ -238,6 +260,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
if (stored === undefined) throw new Error(`session "${id}" not found`)
|
||||
const { meta, events, tornMarker } = stored
|
||||
this.assertVersion(meta)
|
||||
assertSupportedEvents(events, id)
|
||||
|
||||
// Preserve complete interrupted events and synthesize only missing closers.
|
||||
const closers = interruptedTurnClosers(events)
|
||||
@@ -267,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
|
||||
}
|
||||
|
||||
@@ -293,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 = [
|
||||
@@ -341,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)
|
||||
@@ -463,6 +529,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
private async adoptLivePrefix(session: Session, seed: readonly SessionEvent[], stored: StoredPrefix<TornMarker>): Promise<void> {
|
||||
const { meta, events, tornMarker } = stored
|
||||
this.assertVersion(meta)
|
||||
assertSupportedEvents(events, session.header.id)
|
||||
if (!seedCoversPrefix(seed, events)) {
|
||||
throw new Error(`session "${session.header.id}" already has a persisted log on disk that does not match this live session (id collision)`)
|
||||
}
|
||||
|
||||
@@ -21,6 +21,18 @@ declare module 'cordis' {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A backend-resolved, per-session local artifact location. The path is an
|
||||
* absolute target path and can name an artifact that has not materialized yet.
|
||||
* Consumers must treat it as a location hint, never as an authorization token.
|
||||
*/
|
||||
export interface SessionLocation {
|
||||
/** Backend-specific artifact kind, for example `jsonl`. */
|
||||
readonly kind: string
|
||||
/** Absolute path to this session's backend-owned artifact. */
|
||||
readonly path: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Durable append-only session storage. Implementations preserve contiguous,
|
||||
* losslessly JSON-serializable events; {@link append} resolves only after
|
||||
@@ -32,6 +44,15 @@ export abstract class SessionPersistence extends Service {
|
||||
super(ctx, 'sessionPersistence')
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve this backend's independent local artifact for a session without
|
||||
* reading, creating, flushing, or otherwise materializing it. Backends such
|
||||
* as SQLite that do not own one artifact per session return `undefined`.
|
||||
* @param meta - the immutable session header whose artifact is requested.
|
||||
* @returns the backend-specific absolute location, when one exists.
|
||||
*/
|
||||
abstract locate(meta: SessionHeader): SessionLocation | undefined
|
||||
|
||||
/**
|
||||
* Register a new session's metadata. A backend MAY defer the physical write
|
||||
* until the first {@link append} (lazy materialization), in which case a
|
||||
|
||||
@@ -36,7 +36,7 @@ export function oneTurnLog(): SessionEvent[] {
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, surfaceOp: 'append' },
|
||||
{ type: 'step/start', seq: 2, time: 3, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 3, time: 4, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }] }, surfaceOp: 'append' },
|
||||
{ type: 'assistant/message', seq: 3, time: 4, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append' },
|
||||
{ type: 'step/end', seq: 4, time: 5, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 5, time: 6, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
@@ -136,7 +136,7 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
{ type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 8, time: 9, data: { turn: 2, step: 1, content: [
|
||||
{ type: 'tool-call', id: CallId('call-x'), name: 'bash', arguments: '{}' },
|
||||
] } },
|
||||
], provenance: { provider: 'mock', model: 'mock' } } },
|
||||
])
|
||||
|
||||
const loaded = await persistence.load(m.id)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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,
|
||||
@@ -12,9 +12,38 @@ import { runCoordinatorContract, type CoordinatorFixture } from './coordinator-c
|
||||
/** The durable store shape: materialized sessions only (no lazy entries). */
|
||||
type MemoryStore = Map<string, { meta: SessionHeader; events: SessionEvent[] }>
|
||||
|
||||
/** An obsolete event fixture that emulates an untyped pre-change producer. */
|
||||
function legacyHeaderDelta(seq = 0): SessionEvent {
|
||||
return {
|
||||
type: 'request/header-delta',
|
||||
seq,
|
||||
time: 1,
|
||||
data: { config: { model: 'legacy' } },
|
||||
} as unknown as SessionEvent
|
||||
}
|
||||
|
||||
/** An obsolete full-header reason fixture from the removed delta codec. */
|
||||
function legacyFallbackHeader(seq = 0): SessionEvent {
|
||||
return {
|
||||
type: 'request/header',
|
||||
seq,
|
||||
time: 1,
|
||||
data: { header: { config: { model: 'legacy' } }, reason: 'fallback' },
|
||||
} as unknown as 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
|
||||
@@ -41,6 +70,10 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
|
||||
|
||||
// --- service surface (delegated to the coordinator) ---
|
||||
|
||||
locate(_meta: SessionHeader): undefined {
|
||||
return undefined
|
||||
}
|
||||
|
||||
create(m: SessionHeader): Promise<void> {
|
||||
return this.coordinator.create(m)
|
||||
}
|
||||
@@ -97,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()
|
||||
@@ -118,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()
|
||||
@@ -151,4 +451,95 @@ describe('SessionPersistence service registration', () => {
|
||||
.rejects.toThrow('session metadata must be losslessly JSON-serializable')
|
||||
await fiber.dispose()
|
||||
})
|
||||
|
||||
it('rejects a legacy header delta from a pre-change live producer', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(MemoryPersistence)
|
||||
const session = ctx.sessions.create(SessionId('legacy-live'), { meta: { cwd: '/legacy' } })
|
||||
// Model the runtime shape available to JavaScript or a hot-loaded plugin
|
||||
// compiled against the obsolete event vocabulary.
|
||||
const appendLegacy = session.append.bind(session) as (type: string, data: unknown) => SessionEvent
|
||||
expect(() => appendLegacy('request/header-delta', { config: { model: 'legacy' } }))
|
||||
.toThrow(/unsupported legacy request\/header-delta format/)
|
||||
expect(session.events).toHaveLength(0)
|
||||
await fiber.dispose()
|
||||
})
|
||||
|
||||
it('rejects a legacy fallback header buffered by a pre-change live producer', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(MemoryPersistence)
|
||||
const session = ctx.sessions.create(SessionId('legacy-fallback-live'), { meta: { cwd: '/legacy' } })
|
||||
const appendLegacy = session.append.bind(session) as (type: string, data: unknown) => SessionEvent
|
||||
|
||||
expect(() => appendLegacy('request/header', legacyFallbackHeader().data))
|
||||
.toThrow('unsupported legacy request/header reason "fallback"')
|
||||
expect(session.events).toHaveLength(0)
|
||||
await fiber.dispose()
|
||||
})
|
||||
|
||||
it('rejects a legacy stored prefix during live HMR adoption', async () => {
|
||||
const id = SessionId('legacy-hmr')
|
||||
const m = meta(id, '/legacy')
|
||||
const legacy = legacyHeaderDelta()
|
||||
const store: MemoryStore = new Map([[id, { meta: m, events: [legacy] }]])
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
// A current live session cannot carry the obsolete event in its seed, but
|
||||
// HMR still has to identify the persisted prefix as unsupported rather than
|
||||
// treating it as an ordinary live-prefix collision.
|
||||
const session = ctx.sessions.create(id, { meta: { cwd: '/legacy' } })
|
||||
const fiber = await ctx.plugin(MemoryPersistence, { store })
|
||||
|
||||
await expect(ctx.sessions.flush(session))
|
||||
.rejects.toThrow(/unsupported legacy request\/header-delta event at seq 0/)
|
||||
await Promise.allSettled([fiber.dispose()])
|
||||
})
|
||||
|
||||
it('rejects a stored legacy fallback header during load', async () => {
|
||||
const id = SessionId('legacy-fallback-load')
|
||||
const m = meta(id, '/legacy')
|
||||
const store: MemoryStore = new Map([[id, { meta: m, events: [legacyFallbackHeader()] }]])
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(MemoryPersistence, { store })
|
||||
|
||||
await expect(ctx.sessionPersistence.load(id))
|
||||
.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()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user