Hypatia May
828c3f85c9
P1: both merge parents shipped SCHEMA_VERSION=3 for different layouts (surface columns vs seed_length), so an on-disk 3 was ambiguous and wrongly accepted. Bump to 4 (merged layout) so the version check rejects both sibling v3s. P2: a surface-eligible event with no surfaceOp lands in the log but vanishes from deriveMessages() (surface is the sole derivation path). The typed append overload enforces the marker only when the type arg is a literal; it collapses to optional when widened to the union (a caller iterating raw events). Guard at runtime in both append() and the seed constructor — no backward-compat for surface-less logs. Shared seed fixtures carry surfaceOp explicitly and the appendLog helper forwards it verbatim (no synthesized default). Exports isSurfaceEligibleType. Regression tests for all three, each verified to fail on the unfixed code. Gates: typecheck, test (1115), snapshot (14), doc-sync, lint, build, hygiene green.
256 lines
12 KiB
TypeScript
256 lines
12 KiB
TypeScript
/**
|
|
* Reusable contract test for any {@link SessionPersistence} backend. A backend
|
|
* package imports {@link runPersistenceContract} and calls it with a factory
|
|
* that yields a fresh, empty backend (and a teardown), so every backend is held
|
|
* to the same append-only / contiguous-seq / lazy-materialization / crash
|
|
* semantics. The JSONL backend's own spec adds file-specific tests on top.
|
|
*
|
|
* @module @deepseek-ai/dsh-session-persistence/tests/contract
|
|
*/
|
|
|
|
import { describe, expect, it } from 'vitest'
|
|
import { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
|
|
import type { Session, SessionEvent, SessionHeader, SurfaceEventType, SurfaceIntent } from '@deepseek-ai/dsh-session'
|
|
import { CallId } from '@deepseek-ai/dsh-llm'
|
|
import type { SessionPersistence } from '../src/index.ts'
|
|
|
|
/** A backend under test plus its teardown. */
|
|
export interface ContractBackend {
|
|
persistence: SessionPersistence
|
|
dispose: () => Promise<void>
|
|
}
|
|
|
|
/** Build a minimal {@link SessionHeader} for a session id. */
|
|
export function meta(id: string, cwd?: string): SessionHeader {
|
|
return {
|
|
version: SESSION_FORMAT_VERSION,
|
|
id: SessionId(id),
|
|
createdAt: 1000,
|
|
...cwd !== undefined ? { cwd } : {},
|
|
}
|
|
}
|
|
|
|
/** A well-formed one-turn event log (contiguous seqs from 0). */
|
|
export function oneTurnLog(): SessionEvent[] {
|
|
return [
|
|
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
|
{ type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, surfaceOp: 'append' },
|
|
{ type: 'step/start', seq: 2, time: 3, data: { turn: 1, step: 1 } },
|
|
{ type: 'assistant/message', seq: 3, time: 4, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }] }, surfaceOp: 'append' },
|
|
{ type: 'step/end', seq: 4, time: 5, data: { turn: 1, step: 1 } },
|
|
{ type: 'turn/end', seq: 5, time: 6, data: { turn: 1, reason: { kind: 'completed' } } },
|
|
]
|
|
}
|
|
|
|
/**
|
|
* Append a whole event log to a LIVE session, event by event, forwarding the
|
|
* surface metadata each event already carries. A bare `append(e.type, e.data)`
|
|
* over a `SessionEvent[]` widens the type argument to the union, where the
|
|
* typed overload's mandatory-marker rule collapses to optional — and `append`'s
|
|
* runtime guard then rejects a surface-eligible event with no marker. This
|
|
* helper forwards the `surfaceOp`/`sourceEventSeqs` VERBATIM from the source
|
|
* event (it does not synthesize a default), so a well-formed recorded log
|
|
* round-trips through a live session intact and a fixture that forgot a marker
|
|
* still trips the guard.
|
|
*/
|
|
export function appendLog(session: Session, events: readonly SessionEvent[]): void {
|
|
for (const e of events) {
|
|
const se = e as SessionEvent<SurfaceEventType>
|
|
if (se.surfaceOp !== undefined) {
|
|
const intent: SurfaceIntent = {
|
|
surfaceOp: se.surfaceOp,
|
|
...se.sourceEventSeqs !== undefined ? { sourceEventSeqs: se.sourceEventSeqs } : {},
|
|
}
|
|
session.append(e.type, e.data, intent)
|
|
} else {
|
|
session.append(e.type, e.data)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Run the backend-agnostic contract suite. `make()` MUST return a fresh, empty
|
|
* backend each call.
|
|
*/
|
|
export function runPersistenceContract(name: string, make: () => Promise<ContractBackend>): void {
|
|
describe(`SessionPersistence contract: ${name}`, () => {
|
|
it('round-trips a session: create + append → load returns identical meta and byte-identical events', async () => {
|
|
const { persistence, dispose } = await make()
|
|
try {
|
|
const m = meta('s1', '/work')
|
|
const log = oneTurnLog()
|
|
await persistence.create(m)
|
|
await persistence.append(m.id, log)
|
|
|
|
const loaded = await persistence.load(m.id)
|
|
expect(loaded.meta).toMatchObject({ version: SESSION_FORMAT_VERSION, id: m.id, cwd: '/work' })
|
|
expect(loaded.events).toEqual(log)
|
|
} finally {
|
|
await dispose()
|
|
}
|
|
})
|
|
|
|
it('crash recovery: load preserves an interrupted (unclosed) turn and closes it with turn/end {interrupted}', async () => {
|
|
const { persistence, dispose } = await make()
|
|
try {
|
|
const m = meta('interrupted')
|
|
await persistence.create(m)
|
|
await persistence.append(m.id, oneTurnLog()) // turn 1, committed (seqs 0..5)
|
|
// A second turn that crashed mid-flight: turn/start + step/start were
|
|
// durably written, but no step/end / turn/end ever arrived.
|
|
await persistence.append(m.id, [
|
|
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
|
{ type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
|
])
|
|
|
|
// load PRESERVES the interrupted turn's events (a turn can be huge — they
|
|
// must not be truncated) and closes the orphaned turn with synthetic
|
|
// boundary events: step/end (the step was open) then turn/end {interrupted}.
|
|
const loaded = await persistence.load(m.id)
|
|
expect(loaded.events.map(e => e.type)).toEqual([
|
|
'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', // turn 1
|
|
'turn/start', 'step/start', 'step/end', 'turn/end', // turn 2: real events + synthetic closers
|
|
])
|
|
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
|
|
const last = loaded.events.at(-1)!
|
|
expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' })
|
|
|
|
// The closed log is durable and continuable: a fresh append continues at
|
|
// the balanced length (seq 10), and a reload round-trips identically.
|
|
await persistence.append(m.id, [
|
|
{ type: 'turn/start', seq: 10, time: 9, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
|
{ type: 'turn/end', seq: 11, time: 10, data: { turn: 3, reason: { kind: 'completed' } } },
|
|
])
|
|
const reloaded = await persistence.load(m.id)
|
|
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
|
|
} finally {
|
|
await dispose()
|
|
}
|
|
})
|
|
|
|
it('crash recovery: an interrupted tool call gets a synthetic error result so resume is a valid transcript', async () => {
|
|
const { persistence, dispose } = await make()
|
|
try {
|
|
const m = meta('interrupted-toolcall')
|
|
await persistence.create(m)
|
|
await persistence.append(m.id, oneTurnLog()) // turn 1, committed (seqs 0..5)
|
|
// Turn 2 crashed AFTER the assistant message asked for a tool call but
|
|
// BEFORE the tool/result was written (the loop runs tools after logging
|
|
// the assistant message — a process killed mid-tool lands exactly here).
|
|
await persistence.append(m.id, [
|
|
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
|
{ type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
|
{ type: 'assistant/message', seq: 8, time: 9, data: { turn: 2, step: 1, content: [
|
|
{ type: 'tool-call', id: CallId('call-x'), name: 'bash', arguments: '{}' },
|
|
] } },
|
|
])
|
|
|
|
const loaded = await persistence.load(m.id)
|
|
// The orphaned call is answered by a synthetic error tool/result BEFORE
|
|
// step/end + turn/end {interrupted}, so the step (and turn) are balanced
|
|
// and a resumed session derives a valid transcript (no dangling call).
|
|
expect(loaded.events.map(e => e.type)).toEqual([
|
|
'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', // turn 1
|
|
'turn/start', 'step/start', 'assistant/message', 'tool/result', 'step/end', 'turn/end', // turn 2
|
|
])
|
|
const synthetic = loaded.events.find(e => e.type === 'tool/result')
|
|
expect(synthetic?.type === 'tool/result' && synthetic.data).toMatchObject({
|
|
callId: CallId('call-x'), isError: true, error: { code: 'interrupted' },
|
|
})
|
|
// The synthetic result carries the SAME callId as the orphaned tool-call,
|
|
// so deriveMessages() pairs them — no provider-invalid dangling call.
|
|
const call = loaded.events.findLast(e => e.type === 'assistant/message')
|
|
const callId = call?.type === 'assistant/message'
|
|
&& call.data.content.find(b => b.type === 'tool-call')
|
|
expect(callId && callId.type === 'tool-call' && callId.id).toBe(CallId('call-x'))
|
|
} finally {
|
|
await dispose()
|
|
}
|
|
})
|
|
|
|
it('list() excludes a created-but-never-appended (zero-event) session', async () => {
|
|
const { persistence, dispose } = await make()
|
|
try {
|
|
await persistence.create(meta('empty'))
|
|
expect((await persistence.list()).map(m => m.id)).not.toContain(SessionId('empty'))
|
|
} finally {
|
|
await dispose()
|
|
}
|
|
})
|
|
|
|
it('list() includes a session once it has events', async () => {
|
|
const { persistence, dispose } = await make()
|
|
try {
|
|
const m = meta('s2')
|
|
await persistence.create(m)
|
|
await persistence.append(m.id, oneTurnLog())
|
|
expect((await persistence.list()).map(x => x.id)).toContain(m.id)
|
|
} finally {
|
|
await dispose()
|
|
}
|
|
})
|
|
|
|
it('append rejects a batch whose first seq does not match the stored next-seq', async () => {
|
|
const { persistence, dispose } = await make()
|
|
try {
|
|
const m = meta('s3')
|
|
await persistence.create(m)
|
|
await persistence.append(m.id, oneTurnLog()) // seqs 0..5, next-seq = 6
|
|
// A re-append of an already-stored seq must be rejected, not duplicated.
|
|
const restated = oneTurnLog()
|
|
await expect(persistence.append(m.id, restated)).rejects.toThrow()
|
|
} finally {
|
|
await dispose()
|
|
}
|
|
})
|
|
|
|
it('append rejects a mid-batch seq gap', async () => {
|
|
const { persistence, dispose } = await make()
|
|
try {
|
|
const m = meta('s4')
|
|
await persistence.create(m)
|
|
const gapped: SessionEvent[] = [
|
|
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
|
{ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // gap: missing seq 1
|
|
]
|
|
await expect(persistence.append(m.id, gapped)).rejects.toThrow()
|
|
} finally {
|
|
await dispose()
|
|
}
|
|
})
|
|
|
|
it('append rejects non-JSON-serializable event data, naming the event type', async () => {
|
|
const { persistence, dispose } = await make()
|
|
try {
|
|
// Every value `isJsonValue` rejects must be rejected by the backend, not
|
|
// just BigInt — otherwise a backend could pass this contract while still
|
|
// accepting values that corrupt the durable round-trip. Each is a
|
|
// plugin-added `extra` field on a single user/message (seq 0).
|
|
const cyclic: Record<string, unknown> = { type: 'text', text: 'x' }
|
|
cyclic['self'] = cyclic
|
|
const badValues: unknown[] = [
|
|
1n, // BigInt
|
|
undefined, // dropped by JSON.stringify
|
|
Infinity, // → null
|
|
() => 0, // function
|
|
Symbol('s'), // symbol
|
|
new Map(), // exotic object
|
|
cyclic, // circular ref
|
|
]
|
|
for (const [i, bad] of badValues.entries()) {
|
|
// A fresh session per value isolates each rejection (a rejected append
|
|
// must leave no state behind, but isolating keeps the assertion clean).
|
|
const mi = meta(`s5-${i}`)
|
|
await persistence.create(mi)
|
|
const events = [
|
|
{ type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: bad } },
|
|
] as unknown as SessionEvent[]
|
|
await expect(persistence.append(mi.id, events)).rejects.toThrow(/user\/message/)
|
|
}
|
|
} finally {
|
|
await dispose()
|
|
}
|
|
})
|
|
})
|
|
}
|