From d060c0dd4f5947e02b52114ea968307bb1b272e5 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 13 Jul 2026 23:25:20 +0800 Subject: [PATCH] refactor: prune dead session surfaces --- docs/cordis-catalog/services.md | 2 +- .../2026-06-18-session-surface.md | 2 +- packages/core/session/README.md | 2 +- packages/core/session/src/surface.ts | 20 ++------- .../core/session/tests/derived-cache.spec.ts | 15 +------ packages/core/session/tests/surface.spec.ts | 15 +------ .../session-persistence-jsonl/src/index.ts | 10 +---- .../tests/jsonl.spec.ts | 15 +++---- .../session-persistence-sqlite/src/index.ts | 10 +---- .../session-persistence/src/coordinator.ts | 17 ++++--- .../session-persistence/src/index.ts | 34 -------------- .../tests/coordinator-contract.ts | 32 ++++++-------- .../tests/persistence.spec.ts | 44 +------------------ 13 files changed, 43 insertions(+), 175 deletions(-) diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 0aea3f1589..13de873d0c 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -199,7 +199,7 @@ abstract list(): Promise 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` diff --git a/docs/rfc/implemented/architecture/2026-06-18-session-surface.md b/docs/rfc/implemented/architecture/2026-06-18-session-surface.md index 4c8b5a81e3..d7a15a22af 100644 --- a/docs/rfc/implemented/architecture/2026-06-18-session-surface.md +++ b/docs/rfc/implemented/architecture/2026-06-18-session-surface.md @@ -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. diff --git a/packages/core/session/README.md b/packages/core/session/README.md index 97b18b0504..25d6b40be6 100644 --- a/packages/core/session/README.md +++ b/packages/core/session/README.md @@ -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`. diff --git a/packages/core/session/src/surface.ts b/packages/core/session/src/surface.ts index 7219856bdb..d6322c72e7 100644 --- a/packages/core/session/src/surface.ts +++ b/packages/core/session/src/surface.ts @@ -73,7 +73,7 @@ export class SurfaceManager { private _nodes: SurfaceNode[] = [] /** Map from event seq → node. */ private _nodeBySeq = new Map() - /** 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 diff --git a/packages/core/session/tests/derived-cache.spec.ts b/packages/core/session/tests/derived-cache.spec.ts index 46015106c8..493a5a96f0 100644 --- a/packages/core/session/tests/derived-cache.spec.ts +++ b/packages/core/session/tests/derived-cache.spec.ts @@ -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', () => { diff --git a/packages/core/session/tests/surface.spec.ts b/packages/core/session/tests/surface.spec.ts index 257828e466..c03a77d2c3 100644 --- a/packages/core/session/tests/surface.spec.ts +++ b/packages/core/session/tests/surface.spec.ts @@ -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) }) }) diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index a69c979756..5928332b81 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -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> { - return this.coordinator.inits - } - // --- PersistenceBackend hooks (the file-bytes storage primitives) --- /** Read a stored prefix by id across ALL cwd buckets (cwd unknown). */ diff --git a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts index 757ba03aac..e5c2955ef2 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts @@ -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> } 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> } 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> } // 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> } 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> } 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() }) diff --git a/packages/session-persistence/session-persistence-sqlite/src/index.ts b/packages/session-persistence/session-persistence-sqlite/src/index.ts index 30387b4837..72419ef51e 100644 --- a/packages/session-persistence/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence/session-persistence-sqlite/src/index.ts @@ -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> { - return this.coordinator.inits - } - // --- PersistenceBackend hooks (the SQLite storage primitives) --- /** Read a stored prefix by id (ids are globally unique — no scope to scan). */ diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index b1fc118a21..a3bae9cf87 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -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 { + 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 { * 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>() + private inits = new Map>() constructor(private ctx: Context, private backend: PersistenceBackend) { this.installWritePath() diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index b03691d17b..ec252fc027 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -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 diff --git a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts index 61d37964ce..c4cc714646 100644 --- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts +++ b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts @@ -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> { - return (persistence as unknown as { inits: Map> }).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() diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 9c766d4b66..e28bd32851 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -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> { - 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/) - }) -})