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

Merge branch 'master' into codex/fix-acp-snapshot-fs-registration

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-14 23:32:41 +08:00
committed by GitHub
parent 466bc01ed3 513a8d2177
commit 7b50c573ca
13 changed files with 42 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/core/session/README.md
View File

@@ -35,7 +35,7 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
- `session.append(type, data, opts?)` snapshots and freezes durable data and surface metadata, commits synchronously, then notifies observers with independent failure containment. Reentrant attached-session appends reject, and runtime checks cover widened unions and loaded logs.
- `session.deriveMessages()` incrementally projects each new surface node once and returns a fresh array over shared frozen messages. A surface rewrite rebuilds the projection; there is no raw-log fallback.
- `session.deriveEventMessage(event)` is the canonical per-event projection used by reconstruction and invariants.
- `session.surface` lazily folds only new `surfaceOp` markers; `replaceGeneration` changes on every rewrite or invalidation.
- `session.surface` lazily folds only new `surfaceOp` markers; `replaceGeneration` changes on every rewrite.
- `session.events` is a cached frozen snapshot invalidated by append; accepted events remain deeply frozen.
- `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`.

18
packages/core/session/src/surface.ts
View File

@@ -193,26 +193,14 @@ export function foldSurface(events: readonly SessionEvent[]): SurfaceFoldResult
export class SurfaceManager {
/** Incremental state shared with the complete surface fold. */
private _state = createFoldState()
/** 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
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
// A wholesale rebuild is a rewrite: bump the generation so incremental
// consumers (the session's derived-message cache) discard their view.
this._state = createFoldState(this._state.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

12
packages/core/session/tests/derived-cache.spec.ts
View File

@@ -1,6 +1,6 @@
/**
* Derived-message cache contract against a scratch oracle: project new nodes
* once, rebuild on surface generation changes, return fresh arrays over shared
* once, rebuild on surface replacements, return fresh arrays over shared
* frozen messages, and remain value-equal to replay at every step.
*/
@@ -61,16 +61,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)
expect(after[0]).not.toBe(before[0])
})
})
describe('Session.deriveEventMessage — the per-event projection', () => {

15
packages/core/session/tests/surface.spec.ts
View File

@@ -81,14 +81,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.
@@ -386,7 +378,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' })
@@ -400,10 +392,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)
})
})

10
packages/session-persistence/session-persistence-jsonl/src/index.ts
View File

@@ -14,7 +14,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'
@@ -83,15 +83,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
// One method serves both public `list` and the backend hook; delegating it to
// the coordinator would call this hook recursively.
/**
* 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
}
/* jscpd:ignore-end */
// --- PersistenceBackend hooks (the file-bytes storage primitives) ---
/** Read a stored prefix by id across all cwd buckets when cwd is unknown. */

15
packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts
View File

@@ -488,12 +488,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
// A new Session object reuses the id. Object-keyed initialization must run independently,
// detect the disk collision, and reject instead of appending through session A's 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 () => {
@@ -511,12 +510,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.
@@ -533,7 +531,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.
@@ -544,7 +541,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 () => {
@@ -557,12 +554,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/)
})
@@ -594,12 +590,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()
})

10
packages/session-persistence/session-persistence-sqlite/src/index.ts
View File

@@ -14,7 +14,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'
@@ -110,14 +110,6 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
// One method serves both public `list` and the backend hook; delegating it to
// the coordinator would call this hook recursively.
/**
* 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). */

16
packages/session-persistence/session-persistence/src/coordinator.ts
View File

@@ -8,7 +8,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 header, valid contiguous event prefix, and optional opaque
@@ -110,6 +109,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
@@ -134,10 +142,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
private chains = new Map<SessionId, Promise<unknown>>()
/**
* Init promises keyed by live session object, preventing an id-reusing
* replacement from inheriting stale initialization. Readonly access supports
* backend white-box tests.
* replacement from inheriting stale initialization. Flush is the public
* observation boundary; 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()

30
packages/session-persistence/session-persistence/src/index.ts
View File

@@ -6,7 +6,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.
@@ -22,35 +21,6 @@ declare module 'cordis' {
}
}
/**
* Check whether a live seed exactly reproduces a durable prefix, including full
* payloads. This distinguishes resume/HMR rebinding from an id collision.
* @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')
}
}
/**
* Durable append-only session storage. Implementations preserve contiguous,
* losslessly JSON-serializable events; {@link append} resolves only after

32
packages/session-persistence/session-persistence/tests/coordinator-contract.ts
View File

@@ -13,7 +13,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'
/**
@@ -39,11 +38,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)
@@ -168,7 +162,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.
@@ -197,7 +191,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)
@@ -370,7 +364,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()
@@ -388,14 +382,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)
@@ -415,7 +409,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' } })
@@ -425,7 +419,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()
@@ -462,7 +456,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 {
@@ -487,7 +481,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()
@@ -511,7 +505,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 {
@@ -531,7 +525,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()
@@ -549,7 +543,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()
@@ -565,7 +559,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()

44
packages/session-persistence/session-persistence/tests/persistence.spec.ts
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'
@@ -53,11 +53,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
@@ -157,38 +152,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/)
})
})