kevin/deepseek-harness
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: prune dead session surfaces

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-13 23:25:20 +08:00
parent 62eba83f4f
commit d060c0dd4f
13 changed files with 43 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/cordis-catalog/services.md
+1 -1
View File
@@ -199,7 +199,7 @@ abstract list(): Promise<SessionHeader[]>
Types: [SessionEvent](../core-data-structures/core.md)
Source: [`packages/session-persistence/session-persistence/src/index.ts:102`](../../packages/session-persistence/session-persistence/src/index.ts)
Source: [`packages/session-persistence/session-persistence/src/index.ts:68`](../../packages/session-persistence/session-persistence/src/index.ts)
## `ctx.sessions` — `SessionStore`
docs/rfc/implemented/architecture/2026-06-18-session-surface.md
+1 -1
View File
@@ -31,7 +31,7 @@ export type SurfaceOp =
### SurfaceManager: delta-based, not full rebuild
A `SurfaceManager` class (private to `Session`) maintains the cached linked list. It tracks `_lastProcessedSeq` and processes only the **delta** (new events since the last access) rather than rescanning the entire log. Because the log is append-only, prior events never change — full rebuild is only needed after a wholesale log replacement (e.g., seeding).
A `SurfaceManager` class (private to `Session`) maintains the cached linked list. It tracks `_lastProcessedSeq` and processes only the **delta** (new events since the last access) rather than rescanning the entire log. Because the log is append-only, prior events never change; a seeded log is simply the initial delta folded on first access.
Delta processing is O(1) when no new events and O(new events) when new events arrive.
packages/core/session/README.md
+1 -1
View File
@@ -35,7 +35,7 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
- `session.append(type, data, opts?): SessionEvent` — synchronous, never blocks on I/O. At this durable boundary, data and surface metadata are lossless-JSON snapshotted and deep-frozen. For an attached session, a reentrant append during dispatch/observer publication rejects, and detach waits for that publication to unwind. Callbacks resolve before the log push; the push is the commit point, after which each observer failure is contained independently. Runtime surface validation covers widened unions and raw seed/load logs.
- `session.deriveMessages(): Message[]` — the LLM message history, CACHED: each surface node is projected exactly once, when first seen (O(new nodes) per call; a surface rewrite rebuilds via `surface.replaceGeneration`). Returns a fresh array per call over shared, deep-frozen `Message` objects. Each projection reuses the already deep-frozen content in its durable log event, so no second deep clone is needed and a consumer still cannot mutate logged data. The surface is the single source of derived history — there is no raw-log fallback.
- `session.deriveEventMessage(event): Message | null` — the per-event projection `deriveMessages()` folds: a fresh message wrapper that reuses the event's already frozen content, or `null` when the event produces none (a non-surface event, or an empty-content `assistant/message` hosting only usage). External reconstructors and the dev invariant fold the same function over a log prefix's surface, so no two paths can disagree about what a request's messages were (the reconstructability RFC).
- `session.surface: SurfaceManager` — the derived surface, lazily rebuilt from `surfaceOp` markers in the log. Processes only new events (delta) on each access — the log is append-only, so prior events never change. `surface.replaceGeneration` is the rewrite signal: bumped by every folded `replace` and by `invalidate()`, never reset, so an incremental consumer comparing generations cannot be fooled.
- `session.surface: SurfaceManager` — the derived surface, lazily folded from `surfaceOp` markers in the log. Processes only new events (delta) on each access — the log is append-only, so prior events never change. `surface.replaceGeneration` is the rewrite signal: bumped by every folded `replace` and never reset, so an incremental consumer comparing generations cannot be fooled.
- `session.events` — a cached, frozen array snapshot over deep-frozen events. Repeated reads without an append return the same array; an append invalidates the cache and the next read returns a new snapshot, while earlier snapshots stay unchanged. Neither a cast nor a retained reference can push into the live log or rewrite an accepted event.
- `session.seq`, `session.id` — current sequence and readonly typed identity.
- `session.header: SessionHeader` — detached, deep-frozen creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`/`seedLength`). Construction validates the durable record and requires its id to match `session.id`.
packages/core/session/src/surface.ts
+3 -17
View File
@@ -73,7 +73,7 @@ export class SurfaceManager {
private _nodes: SurfaceNode[] = []
/** Map from event seq → node. */
private _nodeBySeq = new Map<number, SurfaceNode>()
/** The last processed seq. -1 forces a full rebuild on first access. */
/** The last processed seq. -1 folds the seeded log on first access. */
private _lastProcessedSeq = -1
/** Rewrite generation — see {@link replaceGeneration}. */
@@ -82,22 +82,8 @@ export class SurfaceManager {
constructor(private log: readonly SessionEvent[]) {}
/**
* Reset to unprocessed state. Call after the log has been replaced
* wholesale (e.g. after Session seed). Not needed for normal appends —
* those are picked up incrementally.
*/
invalidate(): void {
this._lastProcessedSeq = -1
this._nodes = []
this._nodeBySeq.clear()
// A wholesale rebuild is a rewrite: bump the generation so incremental
// consumers (the session's derived-message cache) discard their view.
this._replaceGeneration += 1
}
/**
* The surface's rewrite generation: bumped by every folded `replace` op and
* by {@link invalidate}. A replace is the ONE operation that rewrites the
* The surface's rewrite generation, bumped by every folded `replace` op.
* A replace is the ONE operation that rewrites the
* surface non-monotonically, so an incremental consumer of {@link nodes}
* (the session's derived-message cache) compares this between visits — an
* unchanged generation guarantees every node it has not seen is a pure tail
packages/core/session/tests/derived-cache.spec.ts
+2 -13
View File
@@ -1,7 +1,7 @@
/**
* Derived-message cache tests: the session projects each surface node exactly
* once (O(new nodes) per call), rebuilds on a surface rewrite (replace /
* invalidate — the replaceGeneration signal), returns a fresh array snapshot
* once (O(new nodes) per call), rebuilds on a surface replace (the
* replaceGeneration signal), returns a fresh array snapshot
* per call over shared frozen messages, and stays deep-equal to a from-scratch
* replay derivation at every step — the incremental==scratch property the
* reconstructability RFC's invariant enforces in dev at request time.
@@ -66,17 +66,6 @@ describe('derived-message cache', () => {
expect(Object.isFrozen(first[0])).toBe(true)
})
it('rebuilds after surface.invalidate() (the generation covers wholesale rebuilds too)', () => {
const session = new Session(SessionId('cache-invalidate'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
userText(session, 'one')
const before = session.deriveMessages()
session.surface.invalidate()
const after = session.deriveMessages()
expect(after).toEqual(before)
// A rebuild re-projects: fresh objects, same values.
expect(after[0]).not.toBe(before[0])
})
})
describe('Session.deriveEventMessage — the per-event projection', () => {
packages/core/session/tests/surface.spec.ts
+1 -14
View File
@@ -28,14 +28,6 @@ describe('SurfaceManager', () => {
expect(nodes[1]!.next).toBeNull()
})
it('invalidate resets to full rebuild', () => {
const s = surfaceSession()
expect(s.surface.nodes.length).toBe(2)
// After invalidate, the surface should rebuild from scratch on next access.
;(s.surface).invalidate()
expect(s.surface.nodes.length).toBe(2) // same result, but rebuilt
})
it('empty surface yields empty nodes', () => {
const s = new Session(SessionId('empty'))
// Only turn boundaries, no surface nodes.
@@ -336,7 +328,7 @@ describe('surface type guards', () => {
})
describe('SurfaceManager.replaceGeneration', () => {
it('folds the pending log delta on access and counts replaces and invalidations', () => {
it('folds the pending log delta on access and counts replaces', () => {
const s = new Session(SessionId('gen'))
s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
s.append('user/message', { content: [{ type: 'text', text: 'one' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
@@ -350,10 +342,5 @@ describe('SurfaceManager.replaceGeneration', () => {
content: [{ type: 'text', text: 'summary' }], source: { kind: 'plugin', plugin: 'compact' },
}, { surfaceOp: { op: 'replace', start: nodes[0]!.seq, end: nodes[1]!.seq }, sourceEventSeqs: [nodes[0]!.seq, nodes[1]!.seq] })
expect(s.surface.replaceGeneration).toBe(1)
// invalidate() is a rewrite too: the generation moves forward (and the
// refold re-counts the replace), never backwards.
s.surface.invalidate()
expect(s.surface.replaceGeneration).toBeGreaterThan(1)
})
})
packages/session-persistence/session-persistence-jsonl/src/index.ts
+1 -9
View File
@@ -26,7 +26,7 @@ import {
SessionPersistence, PersistenceCoordinator,
type PersistenceBackend, type StoredPrefix,
} from '@deepseek-ai/dsh-session-persistence'
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import {
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
} from './format.ts'
@@ -108,14 +108,6 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
// this same method; routing it through the coordinator would recurse. Defined
// once, in the "PersistenceBackend hooks" section.
/**
* The per-session init promises, exposed for white-box tests that await a
* specific session's onCreated (there is no public API to await one init).
*/
get inits(): Map<Session, Promise<void>> {
return this.coordinator.inits
}
// --- PersistenceBackend hooks (the file-bytes storage primitives) ---
/** Read a stored prefix by id across ALL cwd buckets (cwd unknown). */
packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts
+5 -10
View File
@@ -496,12 +496,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
// the Session OBJECT, so this gets its OWN onCreated (not A's stale promise)
// — which detects the on-disk collision and rejects, rather than silently
// appending the new session's events onto A's log under a stale cursor.
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
let b!: Session
await ctx.plugin(Object.assign((inner: Context) => {
b = inner.sessions.create(SessionId('reuse'), { meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await expect(backend.inits.get(b)).rejects.toThrow(/already bound to a different live session|already has a persisted log on disk/)
await expect(ctx.sessions.flush(b)).rejects.toThrow(/already bound to a different live session|already has a persisted log on disk/)
})
it('a NO-CWD live session does NOT cross-cwd-adopt a same-id log from a real cwd bucket (loadLive is scope-exact)', async () => {
@@ -523,12 +522,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
const backend = ctx2.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
let b!: Session
await ctx2.plugin(Object.assign((inner: Context) => {
b = inner.sessions.create(SessionId('x')) // no cwd
}, { inject: ['sessions'] }))
await expect(backend.inits.get(b)).rejects.toThrow(/already has a persisted log on disk/)
await expect(ctx2.sessions.flush(b)).rejects.toThrow(/already has a persisted log on disk/)
// The "/w" log is untouched — no no-cwd events were grafted onto it, and no
// `_no-cwd` log for "x" was created.
@@ -545,7 +543,6 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await ctx.sessionPersistence.append(SessionId('divergent'), oneTurnLog())
await ctx.sessionPersistence.load(SessionId('divergent'))
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
// A seed that keeps every seq/type/time but mutates a payload must NOT be
// accepted as "the same session" — otherwise drain filters those seqs as
// already persisted and the divergent payload is silently lost.
@@ -556,7 +553,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await ctx.plugin(Object.assign((inner: Context) => {
bad = inner.sessions.create(SessionId('divergent'), { seed: tampered, meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await expect(backend.inits.get(bad)).rejects.toThrow(/do not match this live session|already has a persisted log/)
await expect(ctx.sessions.flush(bad)).rejects.toThrow(/do not match this live session|already has a persisted log/)
})
it('a second live session reusing a bound id is rejected', async () => {
@@ -569,12 +566,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
for (const s of ctx.sessions.list()) await ctx.parallel('session/flush', s)
await firstFiber.dispose()
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
let second!: Session
await ctx.plugin(Object.assign((inner: Context) => {
second = inner.sessions.create(SessionId('bound'), { meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await expect(backend.inits.get(second))
await expect(ctx.sessions.flush(second))
.rejects.toThrow(/already bound to a different live session|already has a persisted log|do not match/)
})
@@ -610,12 +606,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE
const backend = ctx2.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
let s!: Session
await ctx2.plugin(Object.assign((inner: Context) => {
s = inner.sessions.create(SessionId('exists-fault'), { meta: { cwd } })
}, { inject: ['sessions'] }))
await expect(backend.inits.get(s)).rejects.toThrow(/ENOTDIR/)
await expect(ctx2.sessions.flush(s)).rejects.toThrow(/ENOTDIR/)
await ctx2.fiber.dispose()
})
packages/session-persistence/session-persistence-sqlite/src/index.ts
+1 -9
View File
@@ -26,7 +26,7 @@ import {
SessionPersistence, PersistenceCoordinator,
type PersistenceBackend, type StoredPrefix,
} from '@deepseek-ai/dsh-session-persistence'
import type { Session, SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import {
type JournalMode, openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow,
} from './schema.ts'
@@ -126,14 +126,6 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
// listing, so routing it through the coordinator would just recurse. Defined
// once, in the "PersistenceBackend hooks" section.
/**
* The per-session init promises, exposed for white-box tests that await a
* specific session's onCreated (there is no public API to await one init).
*/
get inits(): Map<Session, Promise<void>> {
return this.coordinator.inits
}
// --- PersistenceBackend hooks (the SQLite storage primitives) ---
/** Read a stored prefix by id (ids are globally unique — no scope to scan). */
packages/session-persistence/session-persistence/src/coordinator.ts
+12 -5
View File
@@ -27,7 +27,6 @@
import { Context } from 'cordis'
import { interruptedTurnClosers, SESSION_FORMAT_VERSION, snapshotJsonValue } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import { seedCoversPrefix } from './index.ts'
/**
* A stored session's durable prefix as read back from a backend: its
@@ -143,6 +142,15 @@ async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unkn
return errors
}
/** Whether a live session seed reproduces a persisted prefix exactly. */
function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
return prefix.length <= seed.length
&& prefix.every((event, index) => {
const seedEvent = seed[index]
return seedEvent !== undefined && JSON.stringify(seedEvent) === JSON.stringify(event)
})
}
/**
* Owns the backend-agnostic session write-path orchestration. A backend
* constructs one (`new PersistenceCoordinator(ctx, this)`), implements
@@ -171,11 +179,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
* Session reusing the same id (HMR, an ACP reconnect), and an id-keyed cache
* would hand the new object the old object's init promise.
*
* Public (readonly) so a backend can expose it for white-box tests that await
* a specific session's init (there is no public API to await one init); the
* coordinator itself only ever mutates it internally.
* Flush is the public observation boundary for initialization; callers do
* not inspect this bookkeeping directly.
*/
readonly inits = new Map<Session, Promise<void>>()
private inits = new Map<Session, Promise<void>>()
constructor(private ctx: Context, private backend: PersistenceBackend<TornMarker>) {
this.installWritePath()
packages/session-persistence/session-persistence/src/index.ts
-34
View File
@@ -22,7 +22,6 @@
*/
import { Context, Service } from 'cordis'
import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
// Re-export the metadata vocabulary so consumers import it from the seam.
@@ -38,39 +37,6 @@ declare module 'cordis' {
}
}
/**
* Whether a live session's seed reproduces a persisted prefix exactly. Backends
* use this collision check to distinguish a legitimate resume/HMR rebind from a
* different live session reusing an existing session id.
*
* The comparison includes the full event payload, not just seq/type/time, so a
* mutated seed cannot be grafted onto a durable log with the same envelope.
* @param seed - the live session's creation-time event snapshot.
* @param prefix - the persisted prefix the seed must reproduce.
* @returns `true` when the prefix fits within the seed and every event matches by JSON text.
*/
export function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
return prefix.length <= seed.length
&& prefix.every((event, index) => {
const seedEvent = seed[index]
return seedEvent !== undefined && JSON.stringify(seedEvent) === JSON.stringify(event)
})
}
/**
* Reject a batch that is not wholly losslessly JSON-serializable. Live session
* appends already enforce this; persistence append paths also accept replay or
* direct batches that may bypass a live session instance. Validation uses the
* same one-pass materializer as the coordinator, so getters are read once.
* @param events - the complete event batch to validate.
*/
export function assertSerializable(events: readonly SessionEvent[]): void {
const snapshot = snapshotJsonValue(events)
if (snapshot === undefined) {
throw new Error('session event batch is not losslessly JSON-serializable because it contains non-JSON-serializable data')
}
}
/**
* Abstract durable session-persistence service. Subclass, implement the
* abstract methods, and load the subclass as a plugin — it registers as
packages/session-persistence/session-persistence/tests/coordinator-contract.ts
+13 -19
View File
@@ -30,7 +30,6 @@ import { describe, expect, it } 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'
import type { SessionPersistence } from '../src/index.ts'
import { meta, oneTurnLog, appendLog } from './contract.ts'
/**
@@ -69,11 +68,6 @@ export interface CoordinatorFixture {
const WORK = '/w'
const OTHER = '/other'
/** The per-session init map a backend exposes for white-box init awaits. */
function inits(persistence: SessionPersistence): Map<Session, Promise<void>> {
return (persistence as unknown as { inits: Map<Session, Promise<void>> }).inits
}
/** Append a whole event log to a live session, event by event (drives session/event). */
function send(session: Session, events: readonly SessionEvent[]): void {
appendLog(session, events)
@@ -199,7 +193,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const seed = oneTurnLog()
// A fork: a brand-new id whose seed came from elsewhere.
const forked = ctx.sessions.create(SessionId('forked'), { seed, meta: { cwd: WORK } })
await inits(ctx.sessionPersistence).get(forked) // onCreated persisted the seed
await ctx.sessions.flush(forked) // onCreated persisted the seed
const loaded = await ctx.sessionPersistence.load(SessionId('forked'))
expect(loaded.events).toEqual(seed)
// A flush with no NEW events must not double-write.
@@ -231,7 +225,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
try {
const loaded = await second.ctx.sessionPersistence.load(SessionId('resumed'))
const s2 = second.ctx.sessions.create(SessionId('resumed'), { seed: loaded.events, meta: { cwd: WORK } })
await inits(second.ctx.sessionPersistence).get(s2) // let onCreated adopt
await second.ctx.sessions.flush(s2) // let onCreated adopt
s2.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
s2.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
await second.ctx.parallel('session/flush', s2)
@@ -406,7 +400,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
try {
const s2 = second.ctx.sessions.create(SessionId('collide'), { meta: { cwd: WORK } })
s2.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
await expect(inits(second.ctx.sessionPersistence).get(s2))
await expect(second.ctx.sessions.flush(s2))
.rejects.toThrow(/already has a persisted log|id collision/)
} finally {
await second.fiber.dispose()
@@ -424,14 +418,14 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
firstSession = inner.sessions.create(SessionId('abandoned'), { meta: { cwd: WORK } })
}, { inject: ['sessions'] }))
await inits(ctx.sessionPersistence).get(firstSession) // register the lazy state
await ctx.sessions.flush(firstSession) // register the lazy state
await firstFiber.dispose() // disposed before any append → never materialized
let reuse!: Session
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create(SessionId('abandoned'), { meta: { cwd: WORK } })
}, { inject: ['sessions'] }))
await expect(inits(ctx.sessionPersistence).get(reuse)).resolves.toBeUndefined()
await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined()
reuse.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
reuse.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', reuse)
@@ -451,7 +445,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
first = inner.sessions.create(SessionId('buffered'), { meta: { cwd: WORK } })
}, { inject: ['sessions'] }))
await inits(ctx.sessionPersistence).get(first)
await ctx.sessions.flush(first)
// 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' } })
@@ -461,7 +455,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create(SessionId('buffered'), { meta: { cwd: WORK } })
}, { inject: ['sessions'] }))
await expect(inits(ctx.sessionPersistence).get(reuse)).rejects.toThrow(/already bound to a different live session/)
await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/already bound to a different live session/)
} finally {
await fiber.dispose()
await fix.cleanup()
@@ -498,7 +492,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
// A live session with that id arrives and claims it (cursor 0 matches
// trivially), persisting its seed.
const live = ctx.sessions.create(SessionId('lazy-claim'), { seed: oneTurnLog(), meta: { cwd: WORK } })
await expect(inits(ctx.sessionPersistence).get(live)).resolves.toBeUndefined()
await expect(ctx.sessions.flush(live)).resolves.toBeUndefined()
const loaded = await ctx.sessionPersistence.load(SessionId('lazy-claim'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
} finally {
@@ -523,7 +517,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
await ctx.plugin(Object.assign((inner: Context) => {
fresh = inner.sessions.create(SessionId('preview'), { meta: { cwd: WORK } })
}, { inject: ['sessions'] }))
await expect(inits(ctx.sessionPersistence).get(fresh))
await expect(ctx.sessions.flush(fresh))
.rejects.toThrow(/do not match this live session|already has a persisted log|id collision/)
} finally {
await fiber.dispose()
@@ -547,7 +541,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
], meta: { cwd: WORK } })
await inits(ctx.sessionPersistence).get(cont)
await ctx.sessions.flush(cont)
const loaded = await ctx.sessionPersistence.load(SessionId('claim'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
} finally {
@@ -567,7 +561,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
// cwd scope is the fence (without it, WORK events would append under the
// OTHER header). Rejected as a collision.
const live = ctx.sessions.create(SessionId('wrong-cwd-claim'), { seed: oneTurnLog(), meta: { cwd: WORK } })
await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
} finally {
await fiber.dispose()
await fix.cleanup()
@@ -585,7 +579,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
// A live session whose SEED matches the loaded prefix but whose cwd is
// WORK must still be rejected — the cwd guard runs before the seed check.
const live = ctx.sessions.create(SessionId('wrong-cwd-load'), { seed: events, meta: { cwd: WORK } })
await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
} finally {
await fiber.dispose()
await fix.cleanup()
@@ -601,7 +595,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
// A live session reusing the id but WITH cwd WORK is a cwd mismatch
// (undefined vs WORK) and must be rejected.
const live = ctx.sessions.create(SessionId('no-cwd-state'), { seed: oneTurnLog(), meta: { cwd: WORK } })
await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
} finally {
await fiber.dispose()
await fix.cleanup()
packages/session-persistence/session-persistence/tests/persistence.spec.ts
+2 -42
View File
@@ -1,9 +1,9 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import SessionStore, { SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import {
SessionPersistence, PersistenceCoordinator, assertSerializable, seedCoversPrefix,
SessionPersistence, PersistenceCoordinator,
type PersistenceBackend, type StoredPrefix,
} from '../src/index.ts'
import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
@@ -61,11 +61,6 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
return this.coordinator.load(id)
}
/** White-box accessor: await a specific session's onCreated init. */
get inits(): Map<Session, Promise<void>> {
return this.coordinator.inits
}
// --- PersistenceBackend hooks (the Map storage primitives) ---
// A Map-backed store has no torn tails, so `tornMarker` is never set. Ids are
@@ -170,38 +165,3 @@ describe('SessionPersistence service registration', () => {
await fiber.dispose()
})
})
describe('shared persistence helpers', () => {
it('accepts a seed that reproduces the persisted prefix exactly', () => {
const log = oneTurnLog()
expect(seedCoversPrefix(log, log.slice(0, 3))).toBe(true)
expect(seedCoversPrefix(log, [])).toBe(true)
})
it('rejects a prefix longer than the seed', () => {
const log = oneTurnLog()
expect(seedCoversPrefix(log.slice(0, 2), log)).toBe(false)
})
it('rejects a same-envelope event with mutated data', () => {
const log = oneTurnLog()
const tampered = structuredClone(log)
const event = tampered[1]!
tampered[1] = {
...event,
data: { ...event.data, content: [{ type: 'text', text: 'tampered' }] },
} as SessionEvent
expect(seedCoversPrefix(tampered, log.slice(0, 2))).toBe(false)
})
it('accepts JSON-serializable event data', () => {
expect(() => { assertSerializable(oneTurnLog()) }).not.toThrow()
})
it('rejects a batch containing non-JSON-serializable event data', () => {
const bad = [
{ type: 'user/message', seq: 0, time: 1, data: { content: 1n } },
] as unknown as SessionEvent[]
expect(() => { assertSerializable(bad) }).toThrow(/batch is not losslessly JSON-serializable/)
})
})