refactor(persistence): share pure backend guards
This commit is contained in:
@@ -25,8 +25,10 @@ import z from 'schemastery'
|
|||||||
import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node:fs/promises'
|
import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node:fs/promises'
|
||||||
import { dirname, resolve } from 'node:path'
|
import { dirname, resolve } from 'node:path'
|
||||||
import { randomBytes } from 'node:crypto'
|
import { randomBytes } from 'node:crypto'
|
||||||
import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
import {
|
||||||
import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
SessionPersistence, assertSerializable, seedCoversPrefix,
|
||||||
|
} from '@deepseek-ai/dsh-session-persistence'
|
||||||
|
import { interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
||||||
import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
||||||
import {
|
import {
|
||||||
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, sidecarPath, toHeaderLine,
|
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, sidecarPath, toHeaderLine,
|
||||||
@@ -59,44 +61,6 @@ interface SessionState {
|
|||||||
owner?: Session
|
owner?: Session
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Whether a live session's `seed` reproduces a persisted `prefix` exactly — the
|
|
||||||
* prefix is no longer than the seed, and each prefix event DEEP-equals the seed
|
|
||||||
* event at the same index. Used to tell a session legitimately continuing a
|
|
||||||
* persisted log (HMR re-seeing its own session, or a resume) from a different
|
|
||||||
* session that merely reuses the id: the latter would have its already-counted
|
|
||||||
* seq 0..prefix-1 events filtered out on flush and its conversation silently
|
|
||||||
* grafted onto the old log.
|
|
||||||
*
|
|
||||||
* The comparison is a full structural equality (via canonical JSON) of each
|
|
||||||
* event INCLUDING its `data` payload, not just `seq`/`type`/`time` — a session
|
|
||||||
* built from loaded events but with mutated message/tool payloads (same seq/
|
|
||||||
* type/time) must NOT be accepted, or the live history and durable log diverge.
|
|
||||||
* Both sides are JSON-serializable by contract (Session.append enforces it), so
|
|
||||||
* JSON.stringify is a sound canonical form here.
|
|
||||||
*/
|
|
||||||
function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
|
|
||||||
return prefix.length <= seed.length
|
|
||||||
&& prefix.every((e, i) => {
|
|
||||||
const s = seed[i]
|
|
||||||
return s !== undefined && JSON.stringify(s) === JSON.stringify(e)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Reject non-JSON-serializable `event.data`, naming the offending type. Used on
|
|
||||||
* the backend's `append(events)` entry point (replay/fork paths that bypass a
|
|
||||||
* live `Session`); events that flow through `Session.append` are already
|
|
||||||
* validated at the source, so the live write path never needs this.
|
|
||||||
*/
|
|
||||||
function assertSerializable(events: readonly SessionEvent[]): void {
|
|
||||||
for (const event of events) {
|
|
||||||
if (!isJsonValue(event.data)) {
|
|
||||||
throw new Error(`event "${event.type}" carries non-JSON-serializable data (seq ${event.seq})`)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Whether `error` is a "no such file/directory" (`ENOENT`) failure — the ONLY
|
* Whether `error` is a "no such file/directory" (`ENOENT`) failure — the ONLY
|
||||||
* filesystem error that legitimately means "this session/root is absent" for a
|
* filesystem error that legitimately means "this session/root is absent" for a
|
||||||
|
|||||||
@@ -24,8 +24,10 @@ import z from 'schemastery'
|
|||||||
import { DatabaseSync } from 'node:sqlite'
|
import { DatabaseSync } from 'node:sqlite'
|
||||||
import { mkdir } from 'node:fs/promises'
|
import { mkdir } from 'node:fs/promises'
|
||||||
import { dirname, resolve } from 'node:path'
|
import { dirname, resolve } from 'node:path'
|
||||||
import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
import {
|
||||||
import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
SessionPersistence, assertSerializable, seedCoversPrefix,
|
||||||
|
} from '@deepseek-ai/dsh-session-persistence'
|
||||||
|
import { interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
||||||
import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
||||||
import {
|
import {
|
||||||
openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow,
|
openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow,
|
||||||
@@ -54,32 +56,6 @@ interface SessionState {
|
|||||||
owner?: Session
|
owner?: Session
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Whether a live session's `seed` reproduces a persisted `prefix` exactly (the
|
|
||||||
* prefix is no longer than the seed and each event DEEP-equals the seed event
|
|
||||||
* at the same index). Distinguishes a session legitimately continuing a
|
|
||||||
* persisted log (HMR re-seeing its own session, or a resume) from a different
|
|
||||||
* session that merely reuses the id. Mirrors the JSONL backend's check; both
|
|
||||||
* sides are JSON-serializable by contract, so `JSON.stringify` is a sound
|
|
||||||
* canonical form.
|
|
||||||
*/
|
|
||||||
function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
|
|
||||||
return prefix.length <= seed.length
|
|
||||||
&& prefix.every((e, i) => {
|
|
||||||
const s = seed[i]
|
|
||||||
return s !== undefined && JSON.stringify(s) === JSON.stringify(e)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
/** Reject non-JSON-serializable `event.data`, naming the offending type. */
|
|
||||||
function assertSerializable(events: readonly SessionEvent[]): void {
|
|
||||||
for (const event of events) {
|
|
||||||
if (!isJsonValue(event.data)) {
|
|
||||||
throw new Error(`event "${event.type}" carries non-JSON-serializable data (seq ${event.seq})`)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unknown[]> {
|
async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unknown[]> {
|
||||||
const settled = await Promise.allSettled([...promises])
|
const settled = await Promise.allSettled([...promises])
|
||||||
const errors: unknown[] = []
|
const errors: unknown[] = []
|
||||||
|
|||||||
@@ -22,6 +22,7 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
import { Context, Service } from 'cordis'
|
import { Context, Service } from 'cordis'
|
||||||
|
import { isJsonValue } from '@deepseek-ai/dsh-session'
|
||||||
import type { SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
import type { SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
||||||
|
|
||||||
// Re-export the metadata vocabulary so consumers import it from the seam.
|
// Re-export the metadata vocabulary so consumers import it from the seam.
|
||||||
@@ -33,6 +34,35 @@ 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.
|
||||||
|
*/
|
||||||
|
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 non-JSON-serializable event data before a backend serializes a batch.
|
||||||
|
* Live session appends already enforce this; persistence append paths also
|
||||||
|
* accept replay/fork batches that may bypass a live session instance.
|
||||||
|
*/
|
||||||
|
export function assertSerializable(events: readonly SessionEvent[]): void {
|
||||||
|
for (const event of events) {
|
||||||
|
if (!isJsonValue(event.data)) {
|
||||||
|
throw new Error(`event "${event.type}" carries non-JSON-serializable data (seq ${event.seq})`)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Abstract durable session-persistence service. Subclass, implement the
|
* Abstract durable session-persistence service. Subclass, implement the
|
||||||
* abstract methods, and load the subclass as a plugin — it registers as
|
* abstract methods, and load the subclass as a plugin — it registers as
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
|
|||||||
import { Context } from 'cordis'
|
import { Context } from 'cordis'
|
||||||
import { SessionId, isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
import { SessionId, isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
||||||
import type { SessionEvent, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
import type { SessionEvent, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
||||||
import { SessionPersistence } from '../src/index.ts'
|
import { SessionPersistence, assertSerializable, seedCoversPrefix } from '../src/index.ts'
|
||||||
import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
|
import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -104,3 +104,38 @@ describe('SessionPersistence service registration', () => {
|
|||||||
await fiber.dispose()
|
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 non-JSON-serializable event data with type and seq context', () => {
|
||||||
|
const bad = [
|
||||||
|
{ type: 'user/message', seq: 0, time: 1, data: { content: 1n } },
|
||||||
|
] as unknown as SessionEvent[]
|
||||||
|
expect(() => { assertSerializable(bad) }).toThrow(/"user\/message".*seq 0/)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user