Merge remote-tracking branch 'origin/master' into session-query-search
# Conflicts: # .agents/notes/implemented/feature/2026-07-10-session-query-service.md # .agents/notes/implemented/feature/2026-07-10-sqlite-session-query-provider.md # .agents/notes/proposed/feature/2026-07-10-sqlite-session-query-provider.md # docs/architecture.md # docs/capability-seams.md # docs/config-catalog.md # docs/cordis-catalog/services.md # docs/core-data-structures/core.md # docs/core-data-structures/persistence.md # docs/core-data-structures/session-query.md # docs/module-graph.md # docs/rfc/INDEX.md # packages/README.md # packages/cordis/tool-cordis/src/api-catalog.ts # packages/hooks/hooks-claude/tests/coverage.spec.ts # packages/session-persistence/session-persistence-jsonl/src/index.ts # packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts # packages/session-persistence/session-persistence-sqlite/README.md # packages/session-persistence/session-persistence-sqlite/src/index.ts # packages/session-persistence/session-persistence-sqlite/src/schema.ts # packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts # packages/session-persistence/session-persistence/README.md # packages/session-persistence/session-persistence/package.json # packages/session-query/README.md # packages/session-query/session-query/README.md # packages/session-query/session-query/package.json # packages/session-query/session-query/src/config.ts # packages/session-query/session-query/src/index.ts # packages/session-query/session-query/src/types.ts # pnpm-lock.yaml # scripts/gen-doc-graphs.ts # scripts/type-equiv.manifest.json # tsconfig.host.json # tsconfig.json
This commit is contained in:
Hypatia May
@@ -2,11 +2,11 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { isAbsolute, join, relative, resolve } from 'node:path'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import { encodeSegment, logPath, scanLog, sessionDir } from '../src/format.ts'
|
||||
import { encodeSegment, eventLines, logPath, scanLog, sessionDir, toHeaderLine } from '../src/format.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
|
||||
|
||||
@@ -20,17 +20,15 @@ function mutableHeader(header: SessionHeader): MutableSessionHeader {
|
||||
return header
|
||||
}
|
||||
|
||||
async function expectParallelFlushError(promise: Promise<unknown>, message: RegExp): Promise<void> {
|
||||
async function expectFlushError(promise: Promise<unknown>, message: RegExp): Promise<void> {
|
||||
try {
|
||||
await promise
|
||||
} catch (error) {
|
||||
expect(error).toBeInstanceOf(AggregateError)
|
||||
const [cause] = (error as AggregateError).errors as unknown[]
|
||||
expect(cause).toBeInstanceOf(Error)
|
||||
expect((cause as Error).message).toMatch(message)
|
||||
expect(error).toBeInstanceOf(Error)
|
||||
expect((error as Error).message).toMatch(message)
|
||||
return
|
||||
}
|
||||
throw new Error('expected parallel flush to reject')
|
||||
throw new Error('expected flush to reject')
|
||||
}
|
||||
|
||||
async function freshRoot(): Promise<string> {
|
||||
@@ -39,7 +37,12 @@ async function freshRoot(): Promise<string> {
|
||||
return dir
|
||||
}
|
||||
|
||||
function rawLogPath(root: string, cwd: string | undefined, id: SessionId): string {
|
||||
return logPath(root, cwd, id, 'none')
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks()
|
||||
for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
@@ -53,11 +56,11 @@ function appendClosedTurn(session: Session): void {
|
||||
}
|
||||
|
||||
// Run the shared backend contract against the real JSONL backend.
|
||||
runPersistenceContract('jsonl', async () => {
|
||||
runPersistenceContract('jsonl-none', async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-'))
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir })
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
|
||||
return {
|
||||
persistence: ctx.sessionPersistence,
|
||||
dispose: async () => {
|
||||
@@ -69,18 +72,18 @@ runPersistenceContract('jsonl', async () => {
|
||||
|
||||
// Two mounts share this temp root to exercise reload. `corruptTail` appends a partial,
|
||||
// newline-less fragment past the committed region so coordinator repair runs on real file bytes.
|
||||
runCoordinatorContract('jsonl', async (): Promise<CoordinatorFixture> => {
|
||||
runCoordinatorContract('jsonl-none', async (): Promise<CoordinatorFixture> => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-coord-'))
|
||||
return {
|
||||
mount: async (ctx) => {
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir })
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
|
||||
return fiber
|
||||
},
|
||||
corruptTail: async (id, cwd) => {
|
||||
// A half-written record with no trailing newline: scanLog treats it as an
|
||||
// uncommitted crash fragment and reports committedBytes < byteLength, so
|
||||
// the coordinator sees a tornMarker to truncate.
|
||||
await appendFile(logPath(dir, cwd, id), '{"type":"assistant/chunk","seq":8,"ti')
|
||||
await appendFile(rawLogPath(dir, cwd, id), '{"type":"assistant/chunk","seq":8,"ti')
|
||||
},
|
||||
cleanup: async () => { await rm(dir, { recursive: true, force: true }) },
|
||||
}
|
||||
@@ -112,6 +115,22 @@ describe('SessionPersistenceJsonl: format helpers', () => {
|
||||
it('encodeSegment rejects an empty id', () => {
|
||||
expect(() => encodeSegment('')).toThrow(/empty/)
|
||||
})
|
||||
|
||||
it('resolves a relative custom root before locating a session', async () => {
|
||||
const absoluteRoot = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, {
|
||||
root: relative(process.cwd(), absoluteRoot),
|
||||
compression: 'none',
|
||||
})
|
||||
const m = meta('relative-location', '/work')
|
||||
expect(ctx.sessionPersistence.locate(m)).toEqual({
|
||||
kind: 'jsonl',
|
||||
path: rawLogPath(resolve(absoluteRoot), '/work', m.id),
|
||||
})
|
||||
await fiber.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
@@ -120,25 +139,50 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
root = await freshRoot()
|
||||
ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
})
|
||||
afterEach(async () => { await ctx.fiber.dispose() })
|
||||
|
||||
it('lazy materialization: create() writes no file until the first append', async () => {
|
||||
const m = meta('lazy', '/work')
|
||||
const location = ctx.sessionPersistence.locate(m)
|
||||
expect(location).toEqual({ kind: 'jsonl', path: rawLogPath(root, '/work', m.id) })
|
||||
expect(isAbsolute(location!.path)).toBe(true)
|
||||
|
||||
await ctx.sessionPersistence.create(m)
|
||||
// nothing on disk yet
|
||||
// locate() is a pure target-path calculation: neither it nor create()
|
||||
// materializes a file before the first append.
|
||||
const dir = sessionDir(root, '/work')
|
||||
await expect(stat(logPath(root, '/work', m.id))).rejects.toThrow()
|
||||
await expect(stat(rawLogPath(root, '/work', m.id))).rejects.toThrow()
|
||||
expect((await ctx.sessionPersistence.list()).map(h => h.id)).not.toContain(m.id)
|
||||
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
// now materialized
|
||||
expect((await stat(logPath(root, '/work', m.id))).isFile()).toBe(true)
|
||||
expect((await stat(rawLogPath(root, '/work', m.id))).isFile()).toBe(true)
|
||||
expect((await ctx.sessionPersistence.list()).map(h => h.id)).toContain(m.id)
|
||||
void dir
|
||||
})
|
||||
|
||||
it('keeps the same location on resume and gives a fork its own location', async () => {
|
||||
const parent = meta('location-parent', '/work')
|
||||
const parentLocation = ctx.sessionPersistence.locate(parent)
|
||||
await ctx.sessionPersistence.create(parent)
|
||||
await ctx.sessionPersistence.append(parent.id, oneTurnLog())
|
||||
|
||||
const loaded = await ctx.sessionPersistence.load(parent.id)
|
||||
expect(ctx.sessionPersistence.locate(loaded.meta)).toEqual(parentLocation)
|
||||
|
||||
const child = {
|
||||
...loaded.meta,
|
||||
id: SessionId('location-child'),
|
||||
parentSession: parent.id,
|
||||
seedLength: loaded.events.length,
|
||||
}
|
||||
const childLocation = ctx.sessionPersistence.locate(child)
|
||||
expect(childLocation?.path).not.toBe(parentLocation?.path)
|
||||
expect(childLocation).toEqual({ kind: 'jsonl', path: rawLogPath(root, '/work', child.id) })
|
||||
})
|
||||
|
||||
it('round-trip is byte-identical (incl. assistant/chunk verbatim)', async () => {
|
||||
const m = meta('chunks')
|
||||
const log: SessionEvent[] = [
|
||||
@@ -146,7 +190,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
{ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/chunk', seq: 2, time: 3, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'he' } } },
|
||||
{ type: 'assistant/chunk', seq: 3, time: 4, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'llo' } } },
|
||||
{ type: 'assistant/message', seq: 4, time: 5, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }] }, surfaceOp: 'append', sourceEventSeqs: [2, 3] },
|
||||
{ type: 'assistant/message', seq: 4, time: 5, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append', sourceEventSeqs: [2, 3] },
|
||||
{ type: 'step/end', seq: 5, time: 6, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 6, time: 7, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
@@ -164,13 +208,13 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
|
||||
const reopenedCtx = new Context()
|
||||
await reopenedCtx.plugin(SessionStore)
|
||||
await reopenedCtx.plugin(SessionPersistenceJsonl, { root })
|
||||
await reopenedCtx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
expect((await reopenedCtx.sessionPersistence.listSnapshots())[0]?.revision).toBe(revision)
|
||||
|
||||
const otherRoot = await freshRoot()
|
||||
const otherCtx = new Context()
|
||||
await otherCtx.plugin(SessionStore)
|
||||
await otherCtx.plugin(SessionPersistenceJsonl, { root: otherRoot })
|
||||
await otherCtx.plugin(SessionPersistenceJsonl, { root: otherRoot, compression: 'none' })
|
||||
await otherCtx.sessionPersistence.create(m)
|
||||
await otherCtx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
expect((await otherCtx.sessionPersistence.listSnapshots())[0]?.revision).not.toBe(revision)
|
||||
@@ -212,12 +256,46 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
discovery.mockRestore()
|
||||
})
|
||||
|
||||
it('rejects a stored v0 log containing a legacy request/header-delta event', async () => {
|
||||
const m = meta('legacy-header-delta', '/legacy')
|
||||
const path = rawLogPath(root, m.cwd, m.id)
|
||||
await mkdir(sessionDir(root, m.cwd), { recursive: true })
|
||||
await writeFile(path, [
|
||||
JSON.stringify(toHeaderLine(m)),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'request/header-delta', seq: 1, time: 2, data: { config: { model: 'legacy' } } }),
|
||||
JSON.stringify({ type: 'turn/end', seq: 2, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
'',
|
||||
].join('\n'))
|
||||
|
||||
await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/unsupported legacy request\/header-delta event at seq 1/)
|
||||
})
|
||||
|
||||
it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
|
||||
const m = meta('legacy-header-fallback', '/legacy')
|
||||
const path = rawLogPath(root, m.cwd, m.id)
|
||||
await mkdir(sessionDir(root, m.cwd), { recursive: true })
|
||||
await writeFile(path, [
|
||||
JSON.stringify(toHeaderLine(m)),
|
||||
JSON.stringify({
|
||||
type: 'request/header',
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: { header: { config: { model: 'legacy' } }, reason: 'fallback' },
|
||||
}),
|
||||
'',
|
||||
].join('\n'))
|
||||
|
||||
await expect(ctx.sessionPersistence.load(m.id))
|
||||
.rejects.toThrow(/unsupported legacy request\/header reason "fallback" at seq 0/)
|
||||
})
|
||||
|
||||
it('persists a forked child seed through the existing session write path', async () => {
|
||||
const source = ctx.sessions.create(SessionId('persist-parent'), { meta: { cwd: '/workspace' } })
|
||||
appendClosedTurn(source)
|
||||
|
||||
const child = ctx.sessions.fork(source, undefined, SessionId('persist-child'))
|
||||
await ctx.parallel('session/flush', child)
|
||||
await ctx.sessions.flush(child)
|
||||
const loaded = await ctx.sessionPersistence.load(child.id)
|
||||
|
||||
expect(loaded.events).toEqual(source.events)
|
||||
@@ -237,7 +315,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
// Simulate a crash mid-second-turn: append raw lines that are NOT closed by
|
||||
// a turn/end (turn/start + step/start are fully written), plus a final
|
||||
// partial line with no newline (a torn fragment never fully flushed).
|
||||
const path = logPath(root, '/proj', m.id)
|
||||
const path = rawLogPath(root, '/proj', m.id)
|
||||
await writeFile(path, [
|
||||
JSON.stringify({ type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 7, time: 9, data: { turn: 2, step: 1 } }),
|
||||
@@ -270,17 +348,17 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
const m = meta('append-only')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
const before = await readFile(logPath(root, undefined, m.id), 'utf8')
|
||||
const before = await readFile(rawLogPath(root, undefined, m.id), 'utf8')
|
||||
const committedPrefix = before // the whole committed log
|
||||
|
||||
// A crash tail then a repair-append.
|
||||
await writeFile(logPath(root, undefined, m.id), '\n{"partial', { flag: 'a' })
|
||||
await writeFile(rawLogPath(root, undefined, m.id), '\n{"partial', { flag: 'a' })
|
||||
await ctx.sessionPersistence.load(m.id)
|
||||
await ctx.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
] as SessionEvent[])
|
||||
const after = await readFile(logPath(root, undefined, m.id), 'utf8')
|
||||
const after = await readFile(rawLogPath(root, undefined, m.id), 'utf8')
|
||||
// the committed prefix is byte-for-byte intact at the head of the file
|
||||
expect(after.startsWith(committedPrefix)).toBe(true)
|
||||
})
|
||||
@@ -289,12 +367,12 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
const m = meta('truncate-retry')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog()) // materialized, seqs 0..5
|
||||
const sizeBefore = (await stat(logPath(root, undefined, m.id))).size
|
||||
const sizeBefore = (await stat(rawLogPath(root, undefined, m.id))).size
|
||||
|
||||
// Force the NEXT fsync (inside appendLines) to fail once, AFTER writeFile
|
||||
// has already put bytes on disk — simulating an ENOSPC/fsync error
|
||||
// mid-append. The recovery truncate() also fsyncs, so allow that one.
|
||||
const handle = await (await import('node:fs/promises')).open(logPath(root, undefined, m.id), 'r')
|
||||
const handle = await (await import('node:fs/promises')).open(rawLogPath(root, undefined, m.id), 'r')
|
||||
const proto = Object.getPrototypeOf(handle) as { sync: () => Promise<void> }
|
||||
await handle.close()
|
||||
const realSync = proto.sync
|
||||
@@ -311,7 +389,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
// The append rejects, but the partial bytes are truncated back: the file is
|
||||
// its pre-append size and the cursor is unchanged.
|
||||
await expect(ctx.sessionPersistence.append(m.id, turn2)).rejects.toThrow(/ENOSPC/)
|
||||
expect((await stat(logPath(root, undefined, m.id))).size).toBe(sizeBefore)
|
||||
expect((await stat(rawLogPath(root, undefined, m.id))).size).toBe(sizeBefore)
|
||||
spy.mockRestore()
|
||||
|
||||
// The retry now succeeds with NO seq gap — the log is contiguous 0..7.
|
||||
@@ -320,6 +398,45 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
})
|
||||
|
||||
it('reports both the append failure and a failed rollback', async () => {
|
||||
const m = meta('rollback-failure')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
|
||||
const path = rawLogPath(root, undefined, m.id)
|
||||
const handle = await (await import('node:fs/promises')).open(path, 'r')
|
||||
const proto = Object.getPrototypeOf(handle) as { sync: () => Promise<void> }
|
||||
await handle.close()
|
||||
const realSync = proto.sync
|
||||
let failed = false
|
||||
const syncSpy = vi.spyOn(proto, 'sync').mockImplementation(async function (this: unknown) {
|
||||
if (!failed) { failed = true; throw new Error('simulated append fsync failure') }
|
||||
return realSync.call(this)
|
||||
})
|
||||
const backend = ctx.sessionPersistence as unknown as {
|
||||
rollbackAppend: (path: string, size: number) => Promise<void>
|
||||
}
|
||||
const realRollback = backend.rollbackAppend.bind(backend)
|
||||
backend.rollbackAppend = () => Promise.reject(new Error('simulated rollback failure'))
|
||||
|
||||
try {
|
||||
await ctx.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
] as SessionEvent[])
|
||||
throw new Error('expected append to reject')
|
||||
} catch (error) {
|
||||
expect(error).toBeInstanceOf(AggregateError)
|
||||
const aggregate = error as AggregateError
|
||||
expect(aggregate.message).toContain(`failed to roll back append to "${path}"`)
|
||||
expect(aggregate.errors).toHaveLength(2)
|
||||
expect(aggregate.errors[0]).toMatchObject({ message: 'simulated append fsync failure' })
|
||||
expect(aggregate.errors[1]).toMatchObject({ message: 'simulated rollback failure' })
|
||||
} finally {
|
||||
backend.rollbackAppend = realRollback
|
||||
syncSpy.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('load returns a meta copy: mutating it does not corrupt backend pathing', async () => {
|
||||
const m = meta('meta-copy', '/proj')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
@@ -370,16 +487,18 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () =>
|
||||
root = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
|
||||
const a = ctx.sessions.create(SessionId('sa'))
|
||||
const b = ctx.sessions.create(SessionId('sb'))
|
||||
a.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
b.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
a.append('user/message', { content: [{ type: 'text', text: 'A' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
b.append('user/message', { content: [{ type: 'text', text: 'B' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
a.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
b.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.parallel('session/flush', a)
|
||||
await ctx.parallel('session/flush', b)
|
||||
await ctx.sessions.flush(a)
|
||||
await ctx.sessions.flush(b)
|
||||
|
||||
const la = await ctx.sessionPersistence.load(SessionId('sa'))
|
||||
const lb = await ctx.sessionPersistence.load(SessionId('sb'))
|
||||
@@ -406,9 +525,30 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
expect(() => scanLog(Buffer.from('{"type":"event"}\n'))).toThrow(/session header/)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['missing', undefined],
|
||||
['a string', '1'],
|
||||
['fractional', 1.5],
|
||||
['negative', -1],
|
||||
])('rejects a session header with %s delegationDepth', (_label, delegationDepth) => {
|
||||
const log = JSON.stringify({
|
||||
type: 'session',
|
||||
version: 0,
|
||||
id: 'invalid-depth',
|
||||
createdAt: 1,
|
||||
...delegationDepth === undefined ? {} : { delegationDepth },
|
||||
}) + '\n'
|
||||
expect(() => scanLog(Buffer.from(log))).toThrow(/session header/)
|
||||
})
|
||||
|
||||
it('rejects a session header with negative-zero delegationDepth', () => {
|
||||
const log = '{"type":"session","version":0,"id":"invalid-depth","createdAt":1,"delegationDepth":-0}\n'
|
||||
expect(() => scanLog(Buffer.from(log))).toThrow(/session header/)
|
||||
})
|
||||
|
||||
it('a seq gap after the last turn/end bounds the preserved tail (torn fragment tolerated)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'g', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'g', createdAt: 1, delegationDepth: 0 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
|
||||
].join('\n') + '\n'
|
||||
@@ -420,7 +560,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
|
||||
it('rejects a seq gap BEFORE a later committed turn/end (committed data damaged)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'g2', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'g2', createdAt: 1, delegationDepth: 0 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
|
||||
JSON.stringify({ type: 'turn/end', seq: 3, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
@@ -432,7 +572,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
|
||||
it('rejects a corrupt line BEFORE a later committed turn/end (committed data damaged)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'c', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'c', createdAt: 1, delegationDepth: 0 }),
|
||||
'{not json', // corrupt, sits in the committed region (a turn/end follows)
|
||||
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
].join('\n') + '\n'
|
||||
@@ -440,7 +580,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
})
|
||||
|
||||
it('a header-only log (no event lines at all) preserves nothing — committedBytes is the header', () => {
|
||||
const log = JSON.stringify({ type: 'session', version: 0, id: 'h0', createdAt: 1 }) + '\n'
|
||||
const log = JSON.stringify({ type: 'session', version: 0, id: 'h0', createdAt: 1, delegationDepth: 0 }) + '\n'
|
||||
const scanned = scanLog(Buffer.from(log))
|
||||
expect(scanned.events).toEqual([])
|
||||
// committedBytes falls back to the header line's end (no preserved events).
|
||||
@@ -449,7 +589,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
|
||||
it('a corrupt line after the last turn/end bounds the preserved tail', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'c2', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'c2', createdAt: 1, delegationDepth: 0 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
'{not json', // corrupt crash fragment, no turn/end committed
|
||||
].join('\n') + '\n'
|
||||
@@ -460,7 +600,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
|
||||
it('tolerates a seq gap AFTER a turn/end (uncommitted tail)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 't', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'session', version: 0, id: 't', createdAt: 1, delegationDepth: 0 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 9, time: 3, data: { turn: 2, step: 1 } }), // gap in uncommitted tail
|
||||
@@ -470,13 +610,128 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: packed chunk rows (packChunks: true)', () => {
|
||||
let ctx: Context
|
||||
beforeEach(async () => {
|
||||
root = await freshRoot()
|
||||
ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
// compression: 'none' — these tests assert the textual storage-record layout
|
||||
// (row tags per line); packing is orthogonal to the physical encoding.
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root, packChunks: true, compression: 'none' })
|
||||
})
|
||||
afterEach(async () => { await ctx.fiber.dispose() })
|
||||
|
||||
/** A one-turn log whose step streams a five-member text-delta run. */
|
||||
function chunkRunLog(): SessionEvent[] {
|
||||
const deltas: SessionEvent[] = Array.from({ length: 5 }, (_, k) => ({
|
||||
type: 'assistant/chunk',
|
||||
seq: 2 + k,
|
||||
time: 3 + k,
|
||||
data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: `t${k}` } },
|
||||
}))
|
||||
return [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
|
||||
...deltas,
|
||||
{ type: 'assistant/message', seq: 7, time: 8, data: { turn: 1, step: 1, content: [{ type: 'text', text: 't0t1t2t3t4' }], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append', sourceEventSeqs: [2, 3, 4, 5, 6] },
|
||||
{ type: 'step/end', seq: 8, time: 9, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 9, time: 10, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
}
|
||||
|
||||
it('writes a delta run as one text-chunks row and loads back identical events', async () => {
|
||||
const m = meta('packed', '/work')
|
||||
const log = chunkRunLog()
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, log)
|
||||
|
||||
const raw = (await readFile(rawLogPath(root, '/work', m.id), 'utf8')).split('\n').filter(Boolean)
|
||||
const tags = raw.slice(1).map(line => (JSON.parse(line) as { type: string }).type)
|
||||
expect(tags).toEqual(['turn/start', 'step/start', 'text-chunks', 'assistant/message', 'step/end', 'turn/end'])
|
||||
|
||||
const loaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events).toEqual(log)
|
||||
})
|
||||
|
||||
it('loads a mixed file: verbatim lines from an unpacked writer, then packed appends', async () => {
|
||||
const m = meta('mixed', '/work')
|
||||
const log = chunkRunLog()
|
||||
// First turn written line-per-event by an unpacked-config writer (an old
|
||||
// file, hand-planted so this packed-config backend adopts it on load).
|
||||
await mkdir(sessionDir(root, '/work'), { recursive: true })
|
||||
await writeFile(rawLogPath(root, '/work', m.id), [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'mixed', createdAt: 1000, cwd: '/work', delegationDepth: 0 }),
|
||||
...log.map(e => JSON.stringify(e)),
|
||||
].join('\n') + '\n')
|
||||
// Adopt the stored log (cursor = stored length), then append a second turn
|
||||
// through THIS packed-config backend.
|
||||
expect((await ctx.sessionPersistence.load(m.id)).events).toEqual(log)
|
||||
const secondTurn: SessionEvent[] = JSON.parse(JSON.stringify(log)) as SessionEvent[]
|
||||
for (const [k, e] of secondTurn.entries()) {
|
||||
;(e as { seq: number }).seq = 10 + k
|
||||
;(e.data as { turn: number }).turn = 2
|
||||
}
|
||||
await ctx.sessionPersistence.append(m.id, secondTurn)
|
||||
|
||||
const loaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events).toEqual([...log, ...secondTurn])
|
||||
// The packed append really packed: the file's tail carries a text-chunks row.
|
||||
const tags = (await readFile(rawLogPath(root, '/work', m.id), 'utf8')).split('\n').filter(Boolean)
|
||||
.map(line => (JSON.parse(line) as { type: string }).type)
|
||||
expect(tags.filter(t => t === 'text-chunks')).toHaveLength(1)
|
||||
expect(tags.filter(t => t === 'assistant/chunk')).toHaveLength(5)
|
||||
})
|
||||
|
||||
it('scanLog: a packed row advances the seq cursor by its whole run', () => {
|
||||
const logText = [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'rows', createdAt: 1, delegationDepth: 0 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'text-chunks', seq0: 1, time0: 2, data: { turn: 1, step: 1, index: 0, dt: [1, 1], texts: ['a', 'b', 'c'] } }),
|
||||
JSON.stringify({ type: 'turn/end', seq: 4, time: 5, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
].join('\n') + '\n'
|
||||
const { events } = scanLog(Buffer.from(logText))
|
||||
expect(events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4])
|
||||
expect(events[2]).toEqual({ type: 'assistant/chunk', seq: 2, time: 3, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'b' } } })
|
||||
})
|
||||
|
||||
it('scanLog: a malformed packed row in the committed region rejects like corrupt JSON', () => {
|
||||
const logText = [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'bad-row', createdAt: 1, delegationDepth: 0 }),
|
||||
// dt arity mismatch — row validation throws, so the line is a committed hole.
|
||||
JSON.stringify({ type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], texts: ['a', 'b'] } }),
|
||||
JSON.stringify({ type: 'turn/end', seq: 2, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
].join('\n') + '\n'
|
||||
expect(() => scanLog(Buffer.from(logText))).toThrow(/unparsable committed event/)
|
||||
})
|
||||
|
||||
it('scanLog: a packed row with a mid-run seq gap after the last turn/end drops the whole row', () => {
|
||||
const logText = [
|
||||
JSON.stringify({ type: 'session', version: 0, id: 'row-gap', createdAt: 1, delegationDepth: 0 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
// seq0 skips 1 — the run's first member is already a gap; no turn/end follows.
|
||||
JSON.stringify({ type: 'text-chunks', seq0: 2, time0: 2, data: { turn: 1, step: 1, index: 0, dt: [1, 1], texts: ['a', 'b', 'c'] } }),
|
||||
].join('\n') + '\n'
|
||||
const scanned = scanLog(Buffer.from(logText))
|
||||
expect(scanned.events.map(e => e.seq)).toEqual([0])
|
||||
// committedBytes stays on the line boundary BEFORE the dropped row.
|
||||
const headerAndTurn = logText.split('\n').slice(0, 2).join('\n') + '\n'
|
||||
expect(scanned.committedBytes).toBe(Buffer.byteLength(headerAndTurn, 'utf8'))
|
||||
})
|
||||
|
||||
it('eventLines(packChunks: false) is byte-identical to the pre-packing layout', () => {
|
||||
const log = chunkRunLog()
|
||||
expect(eventLines(log, false)).toBe(log.map(e => JSON.stringify(e)).join('\n'))
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
let ctx: Context
|
||||
beforeEach(async () => {
|
||||
root = await freshRoot()
|
||||
ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
})
|
||||
afterEach(async () => { await ctx.fiber.dispose() })
|
||||
|
||||
@@ -496,8 +751,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
await p
|
||||
await ctx.sessionPersistence.append(SessionId('create-snap'), oneTurnLog())
|
||||
// The log materialized under the ORIGINAL cwd, not the mutated one.
|
||||
expect((await stat(logPath(root, '/orig', SessionId('create-snap')))).isFile()).toBe(true)
|
||||
await expect(stat(logPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
|
||||
expect((await stat(rawLogPath(root, '/orig', SessionId('create-snap')))).isFile()).toBe(true)
|
||||
await expect(stat(rawLogPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
|
||||
})
|
||||
|
||||
it('list discovers sessions across multiple cwd buckets', async () => {
|
||||
@@ -538,7 +793,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
// `readFirstLine` accumulates chunks before `list()` parses it.
|
||||
const bucket = join(root, '_no-cwd')
|
||||
await mkdir(bucket, { recursive: true })
|
||||
const bigHeader = JSON.stringify({ type: 'session', version: 0, id: 'big', createdAt: 1, pad: 'x'.repeat(9000) })
|
||||
const bigHeader = JSON.stringify({ type: 'session', version: 0, id: 'big', createdAt: 1, delegationDepth: 0, pad: 'x'.repeat(9000) })
|
||||
await writeFile(join(bucket, 'big.jsonl'), bigHeader + '\n')
|
||||
const ids = (await ctx.sessionPersistence.list()).map(x => x.id)
|
||||
expect(ids).toContain('big')
|
||||
@@ -552,7 +807,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
}, { inject: ['sessions'] }))
|
||||
// Drain A, then dispose ITS fiber (the live session A is gone) while the
|
||||
// backend stays loaded.
|
||||
for (const s of ctx.sessions.list()) await ctx.parallel('session/flush', s)
|
||||
for (const s of ctx.sessions.list()) await ctx.sessions.flush(s)
|
||||
await sessFiberA.dispose()
|
||||
|
||||
// A new Session object reuses the id. Object-keyed initialization must run independently,
|
||||
@@ -578,7 +833,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
// of grafting no-cwd events onto a log with mismatched cwd.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
let b!: Session
|
||||
await ctx2.plugin(Object.assign((inner: Context) => {
|
||||
b = inner.sessions.create(SessionId('x')) // no cwd
|
||||
@@ -587,10 +842,10 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
|
||||
// The "/w" log is untouched — no no-cwd events were grafted onto it, and no
|
||||
// `_no-cwd` log for "x" was created.
|
||||
const inW = scanLog(await readFile(logPath(root, '/w', SessionId('x'))))
|
||||
const inW = scanLog(await readFile(rawLogPath(root, '/w', SessionId('x'))))
|
||||
expect(inW.meta.cwd).toBe('/w')
|
||||
expect(inW.events).toHaveLength(6)
|
||||
await expect(stat(logPath(root, undefined, SessionId('x')))).rejects.toThrow()
|
||||
await expect(stat(rawLogPath(root, undefined, SessionId('x')))).rejects.toThrow()
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
@@ -620,7 +875,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
a.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
a.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
}, { inject: ['sessions'] }))
|
||||
for (const s of ctx.sessions.list()) await ctx.parallel('session/flush', s)
|
||||
for (const s of ctx.sessions.list()) await ctx.sessions.flush(s)
|
||||
await firstFiber.dispose()
|
||||
|
||||
let second!: Session
|
||||
@@ -634,7 +889,10 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
it('list returns nothing when the root directory does not exist', async () => {
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root: join(root, 'does-not-exist-yet') })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, {
|
||||
root: join(root, 'does-not-exist-yet'),
|
||||
compression: 'none',
|
||||
})
|
||||
expect(await ctx2.sessionPersistence.list()).toEqual([])
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
@@ -646,7 +904,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
await writeFile(filePath, 'x')
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root: filePath })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root: filePath, compression: 'none' })
|
||||
await expect(ctx2.sessionPersistence.list()).rejects.toThrow(/ENOTDIR/)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
@@ -657,7 +915,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
const cwd = '/x'
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE
|
||||
let s!: Session
|
||||
await ctx2.plugin(Object.assign((inner: Context) => {
|
||||
@@ -672,14 +930,14 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
const m = meta('disk-append', '/d')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
await writeFile(logPath(root, '/d', m.id), '\n{"partial crash', { flag: 'a' })
|
||||
await writeFile(rawLogPath(root, '/d', m.id), '\n{"partial crash', { flag: 'a' })
|
||||
|
||||
// A FRESH backend with no in-memory state: append directly (no prior load)
|
||||
// → append must adopt from disk, and the adopt's load schedules a repair
|
||||
// that the same append then performs before writing.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
await ctx2.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
@@ -715,7 +973,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
// nondeterministic. create scans every bucket, not just meta.cwd's.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
await expect(ctx2.sessionPersistence.create(meta('dup-id', '/projB')))
|
||||
.rejects.toThrow(/already has a persisted log on disk/)
|
||||
await ctx2.fiber.dispose()
|
||||
@@ -725,21 +983,22 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
root = await freshRoot()
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
const session = ctx2.sessions.create(SessionId('flush-fail'))
|
||||
// A full turn lands in the write-behind buffer.
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
// Make the durable materialize fail on the next flush.
|
||||
const backend = ctx2.sessionPersistence as unknown as { materialize: (...args: unknown[]) => Promise<void> }
|
||||
const origMat = backend.materialize.bind(backend)
|
||||
backend.materialize = () => Promise.reject(new Error('disk full'))
|
||||
await expectParallelFlushError(ctx2.parallel('session/flush', session), /disk full/)
|
||||
await expectFlushError(ctx2.sessions.flush(session), /disk full/)
|
||||
// The events are STILL buffered (not silently dropped): a retry persists them.
|
||||
backend.materialize = origMat
|
||||
await ctx2.parallel('session/flush', session)
|
||||
await ctx2.sessions.flush(session)
|
||||
const loaded = await ctx2.sessionPersistence.load(SessionId('flush-fail'))
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1])
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2])
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
/**
|
||||
* Unit tests for the Windows durable namespace helper with a mocked kernel32
|
||||
* binding. The real JSONL suite exercises the helper on native Windows; these
|
||||
* tests keep the Win32 error mapping and race handling covered on every host.
|
||||
*/
|
||||
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs'
|
||||
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
|
||||
const MOVEFILE_WRITE_THROUGH = 0x00000008
|
||||
const ERROR_FILE_NOT_FOUND = 2
|
||||
const ERROR_PATH_NOT_FOUND = 3
|
||||
const ERROR_ACCESS_DENIED = 5
|
||||
const ERROR_NOT_SAME_DEVICE = 17
|
||||
const ERROR_FILE_EXISTS = 80
|
||||
const ERROR_INVALID_NAME = 123
|
||||
const ERROR_ALREADY_EXISTS = 183
|
||||
|
||||
type MoveFileExW = (existing: string, replacement: string, flags: number, setLastError: (code: number) => void) => number
|
||||
|
||||
const roots: string[] = []
|
||||
|
||||
function stripNamespace(path: string): string {
|
||||
if (path.startsWith('\\\\?\\UNC\\')) return `\\\\${path.slice('\\\\?\\UNC\\'.length)}`
|
||||
if (path.startsWith('\\\\?\\')) return path.slice('\\\\?\\'.length)
|
||||
return path
|
||||
}
|
||||
|
||||
async function tempRoot(): Promise<string> {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-win32-'))
|
||||
roots.push(dir)
|
||||
return dir
|
||||
}
|
||||
|
||||
async function importWithMove(moveFileExW: MoveFileExW): Promise<typeof import('../src/win32.ts')> {
|
||||
vi.resetModules()
|
||||
vi.doMock('koffi', () => {
|
||||
let lastError = 0
|
||||
const setLastError = (code: number): void => { lastError = code }
|
||||
const move: MoveFileExW = (existing, replacement, flags, setError) => {
|
||||
const ok = moveFileExW(existing, replacement, flags, setError)
|
||||
lastError = ok === 0 ? lastError : 0
|
||||
return ok
|
||||
}
|
||||
return {
|
||||
default: {
|
||||
load: () => ({
|
||||
func: (_convention: string, name: string, result: string) => {
|
||||
if (name === 'MoveFileExW') return (existing: string, replacement: string, flags: number) => {
|
||||
expect(result).toBe('int')
|
||||
const ok = move(existing, replacement, flags, setLastError)
|
||||
return ok
|
||||
}
|
||||
return () => lastError
|
||||
},
|
||||
}),
|
||||
},
|
||||
}
|
||||
})
|
||||
return import('../src/win32.ts')
|
||||
}
|
||||
|
||||
async function importWithError(code: number): Promise<typeof import('../src/win32.ts')> {
|
||||
vi.resetModules()
|
||||
vi.doMock('koffi', () => ({
|
||||
default: {
|
||||
load: () => ({
|
||||
func: (_convention: string, name: string) => {
|
||||
if (name === 'MoveFileExW') return () => 0
|
||||
return () => code
|
||||
},
|
||||
}),
|
||||
},
|
||||
}))
|
||||
return import('../src/win32.ts')
|
||||
}
|
||||
|
||||
async function importWithFilesystemMove(): Promise<typeof import('../src/win32.ts')> {
|
||||
return importWithMove((existing, replacement, flags, setLastError) => {
|
||||
expect(flags).toBe(MOVEFILE_WRITE_THROUGH)
|
||||
const from = stripNamespace(existing)
|
||||
const to = stripNamespace(replacement)
|
||||
if (!existsSync(from)) { setLastError(ERROR_FILE_NOT_FOUND); return 0 }
|
||||
if (existsSync(to)) { setLastError(ERROR_ALREADY_EXISTS); return 0 }
|
||||
renameSync(from, to)
|
||||
return 1
|
||||
})
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
vi.doUnmock('koffi')
|
||||
vi.resetModules()
|
||||
for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('Windows durable namespace helpers', () => {
|
||||
it('publishes a new file with write-through MoveFileExW semantics', async () => {
|
||||
const { publishNewFileWin32 } = await importWithFilesystemMove()
|
||||
const root = await tempRoot()
|
||||
const tmp = join(root, 'log.tmp')
|
||||
const final = join(root, 'log.jsonl')
|
||||
await writeFile(tmp, 'content')
|
||||
|
||||
await publishNewFileWin32(tmp, final)
|
||||
expect(existsSync(tmp)).toBe(false)
|
||||
expect(readFileSync(final, 'utf8')).toBe('content')
|
||||
})
|
||||
|
||||
it('maps Win32 publish failures to Node-style errno codes', async () => {
|
||||
const cases = [
|
||||
[ERROR_FILE_NOT_FOUND, 'ENOENT'],
|
||||
[ERROR_PATH_NOT_FOUND, 'ENOENT'],
|
||||
[ERROR_ACCESS_DENIED, 'EACCES'],
|
||||
[ERROR_NOT_SAME_DEVICE, 'EXDEV'],
|
||||
[ERROR_FILE_EXISTS, 'EEXIST'],
|
||||
[ERROR_ALREADY_EXISTS, 'EEXIST'],
|
||||
[ERROR_INVALID_NAME, 'EINVAL'],
|
||||
[9999, 'EIO'],
|
||||
] as const
|
||||
for (const [win32Code, code] of cases) {
|
||||
const { publishNewFileWin32 } = await importWithError(win32Code)
|
||||
await expect(publishNewFileWin32('from', 'to')).rejects.toMatchObject({ code, win32Code, path: 'from', dest: 'to' })
|
||||
}
|
||||
})
|
||||
|
||||
it('creates missing directories through staging siblings and tolerates an already-created race', async () => {
|
||||
const root = await tempRoot()
|
||||
const raced = join(root, 'raced')
|
||||
const { ensureDurableDirectoryWin32 } = await importWithMove((existing, replacement, flags, setLastError) => {
|
||||
expect(flags).toBe(MOVEFILE_WRITE_THROUGH)
|
||||
const from = stripNamespace(existing)
|
||||
const to = stripNamespace(replacement)
|
||||
if (to === raced) {
|
||||
mkdirSync(to)
|
||||
setLastError(ERROR_ALREADY_EXISTS)
|
||||
return 0
|
||||
}
|
||||
if (!existsSync(from)) { setLastError(ERROR_FILE_NOT_FOUND); return 0 }
|
||||
if (existsSync(to)) { setLastError(ERROR_ALREADY_EXISTS); return 0 }
|
||||
renameSync(from, to)
|
||||
return 1
|
||||
})
|
||||
|
||||
await ensureDurableDirectoryWin32(join(root, 'a', 'b'))
|
||||
expect(existsSync(join(root, 'a', 'b'))).toBe(true)
|
||||
await ensureDurableDirectoryWin32(join(root, 'a', 'b'))
|
||||
await ensureDurableDirectoryWin32(raced)
|
||||
expect(existsSync(raced)).toBe(true)
|
||||
})
|
||||
|
||||
it('surfaces directory publication failures other than an existing-target race', async () => {
|
||||
const { ensureDurableDirectoryWin32 } = await importWithError(ERROR_ACCESS_DENIED)
|
||||
const root = await tempRoot()
|
||||
|
||||
await expect(ensureDurableDirectoryWin32(join(root, 'denied'))).rejects.toMatchObject({ code: 'EACCES' })
|
||||
})
|
||||
|
||||
it('rejects a non-directory component instead of treating it as missing', async () => {
|
||||
const { ensureDurableDirectoryWin32 } = await importWithFilesystemMove()
|
||||
const root = await tempRoot()
|
||||
const blocked = join(root, 'blocked')
|
||||
writeFileSync(blocked, 'x')
|
||||
|
||||
await expect(ensureDurableDirectoryWin32(join(blocked, 'child'))).rejects.toMatchObject({ code: 'ENOTDIR' })
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,24 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts'
|
||||
|
||||
describe('JSONL Zstandard compatibility', () => {
|
||||
it('round-trips concatenated checksummed frames through the built-in Node API', async () => {
|
||||
const encoded = Buffer.concat([
|
||||
await compressZstdFrame('{"type":"session","version":0,"id":"compat","createdAt":1}\n'),
|
||||
await compressZstdFrame('{"type":"turn/start","seq":0,"turn":1}\n'),
|
||||
])
|
||||
const { frames, tornStart } = scanZstdFrames(encoded)
|
||||
|
||||
expect(tornStart).toBeUndefined()
|
||||
expect(frames).toHaveLength(2)
|
||||
expect(frames.map(frame => encoded.subarray(frame.start, frame.start + 4).toString('hex')))
|
||||
.toEqual(['28b52ffd', '28b52ffd'])
|
||||
const decoded = await Promise.all(frames.map(frame => decompressZstdFrame(encoded.subarray(frame.start, frame.end))))
|
||||
expect(Buffer.concat(decoded).toString()).toContain('"type":"turn/start"')
|
||||
|
||||
const eventFrame = encoded.subarray(frames[1]!.start, frames[1]!.end)
|
||||
const missingChecksumByte = eventFrame.subarray(0, -1)
|
||||
expect(scanZstdFrames(missingChecksumByte)).toEqual({ frames: [], tornStart: 0 })
|
||||
expect((await decompressZstdFrame(missingChecksumByte)).toString()).toContain('"type":"turn/start"')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,483 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { appendFile, mkdir, mkdtemp, open, readFile, readdir, rm, stat, writeFile } from 'node:fs/promises'
|
||||
import type { FileHandle } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import { logPath, scanLog, sessionDir, toHeaderLine, type JsonlCompression } from '../src/format.ts'
|
||||
import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
|
||||
|
||||
const MAGIC = Buffer.from([0x28, 0xB5, 0x2F, 0xFD])
|
||||
const roots: string[] = []
|
||||
const contexts: Context[] = []
|
||||
|
||||
async function freshRoot(prefix = 'dsh-jsonl-zstd-'): Promise<string> {
|
||||
const root = await mkdtemp(join(tmpdir(), prefix))
|
||||
roots.push(root)
|
||||
return root
|
||||
}
|
||||
|
||||
async function mount(root: string, compression?: JsonlCompression): Promise<Context> {
|
||||
const ctx = new Context()
|
||||
contexts.push(ctx)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, {
|
||||
root,
|
||||
...(compression === undefined ? {} : { compression }),
|
||||
})
|
||||
return ctx
|
||||
}
|
||||
|
||||
async function decodeCompleteFrames(buffer: Buffer): Promise<Buffer> {
|
||||
const { frames, tornStart } = scanZstdFrames(buffer)
|
||||
expect(tornStart).toBeUndefined()
|
||||
const plaintext: Buffer[] = []
|
||||
for (const frame of frames) {
|
||||
plaintext.push(await decompressZstdFrame(buffer.subarray(frame.start, frame.end)))
|
||||
}
|
||||
return Buffer.concat(plaintext)
|
||||
}
|
||||
|
||||
async function tornFrame(
|
||||
plaintext: string,
|
||||
accepts: (decoded: string) => boolean,
|
||||
): Promise<Buffer> {
|
||||
const frame = await compressZstdFrame(plaintext)
|
||||
const candidateEnds = [
|
||||
frame.length - 1,
|
||||
frame.length - 4,
|
||||
...[0.9, 0.75, 0.6, 0.5, 0.4, 0.25].map(ratio => Math.floor(frame.length * ratio)),
|
||||
]
|
||||
for (const end of candidateEnds) {
|
||||
const candidate = frame.subarray(0, end)
|
||||
if (scanZstdFrames(candidate).tornStart !== 0) continue
|
||||
try {
|
||||
const decoded = (await decompressZstdFrame(candidate)).toString('utf8')
|
||||
if (accepts(decoded)) return candidate
|
||||
} catch {
|
||||
// Some early cuts precede the first decodable block; keep searching for
|
||||
// a cut that exercises partial-plaintext recovery.
|
||||
}
|
||||
}
|
||||
throw new Error('test fixture could not produce the requested torn Zstandard frame')
|
||||
}
|
||||
|
||||
function deterministicNoise(length: number): string {
|
||||
let state = 0x12345678
|
||||
let output = ''
|
||||
for (let index = 0; index < length; index++) {
|
||||
state = (Math.imul(state, 1_664_525) + 1_013_904_223) >>> 0
|
||||
output += String.fromCharCode(33 + (state % 90))
|
||||
}
|
||||
return output
|
||||
}
|
||||
|
||||
function emptyStructuralFrame(descriptor: number): Buffer {
|
||||
const contentSizeFlag = descriptor >>> 6
|
||||
const singleSegment = (descriptor & 0x20) !== 0
|
||||
const dictionaryBytes = [0, 1, 2, 4][descriptor & 0x03]!
|
||||
const contentSizeBytes = contentSizeFlag === 0 ? (singleSegment ? 1 : 0) : 1 << contentSizeFlag
|
||||
const variableHeader = Buffer.alloc((singleSegment ? 0 : 1) + dictionaryBytes + contentSizeBytes)
|
||||
const lastEmptyRawBlock = Buffer.from([1, 0, 0])
|
||||
const checksum = (descriptor & 0x04) === 0 ? Buffer.alloc(0) : Buffer.alloc(4)
|
||||
return Buffer.concat([MAGIC, Buffer.from([descriptor]), variableHeader, lastEmptyRawBlock, checksum])
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks()
|
||||
for (const ctx of contexts.splice(0).reverse()) await ctx.fiber.dispose()
|
||||
for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
runPersistenceContract('jsonl-zstd', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-jsonl-zstd-contract-'))
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
return {
|
||||
persistence: ctx.sessionPersistence,
|
||||
dispose: async () => {
|
||||
await fiber.dispose()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
},
|
||||
}
|
||||
})
|
||||
|
||||
runCoordinatorContract('jsonl-zstd', async (): Promise<CoordinatorFixture> => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-jsonl-zstd-coordinator-'))
|
||||
return {
|
||||
mount: async ctx => ctx.plugin(SessionPersistenceJsonl, { root }),
|
||||
corruptTail: async (id, cwd) => {
|
||||
const line = JSON.stringify({
|
||||
type: 'assistant/chunk',
|
||||
seq: 8,
|
||||
time: 9,
|
||||
data: { turn: 2, step: 1, chunk: { type: 'text-delta', index: 0, text: deterministicNoise(300_000) } },
|
||||
}) + '\n'
|
||||
const partial = await tornFrame(line, decoded => decoded.length > 0 && !decoded.endsWith('\n'))
|
||||
await appendFile(logPath(root, cwd, id, 'zstd'), partial)
|
||||
},
|
||||
cleanup: async () => { await rm(root, { recursive: true, force: true }) },
|
||||
}
|
||||
})
|
||||
|
||||
describe('Zstandard frame structure', () => {
|
||||
it('scans concatenated checksummed frames and honors a frame limit', async () => {
|
||||
const first = await compressZstdFrame('header\n')
|
||||
const second = await compressZstdFrame('event\n')
|
||||
const stream = Buffer.concat([first, second])
|
||||
expect(scanZstdFrames(Buffer.alloc(0))).toEqual({ frames: [] })
|
||||
expect(scanZstdFrames(stream)).toEqual({
|
||||
frames: [{ start: 0, end: first.length }, { start: first.length, end: stream.length }],
|
||||
})
|
||||
expect(scanZstdFrames(stream, 1)).toEqual({ frames: [{ start: 0, end: first.length }] })
|
||||
expect(first[4]! & 0x04).toBe(0x04)
|
||||
expect(second[4]! & 0x04).toBe(0x04)
|
||||
expect((await decompressZstdFrame(first)).toString()).toBe('header\n')
|
||||
})
|
||||
|
||||
it('distinguishes incomplete frame regions from invalid complete structure', () => {
|
||||
expect(scanZstdFrames(MAGIC.subarray(0, 2))).toEqual({ frames: [], tornStart: 0 })
|
||||
expect(scanZstdFrames(MAGIC)).toEqual({ frames: [], tornStart: 0 })
|
||||
expect(() => scanZstdFrames(Buffer.alloc(4))).toThrow(/invalid frame magic/)
|
||||
expect(() => scanZstdFrames(Buffer.concat([MAGIC, Buffer.from([0x08])]))).toThrow(/reserved frame-header bit/)
|
||||
|
||||
// Non-single-segment descriptor with no window descriptor.
|
||||
expect(scanZstdFrames(Buffer.concat([MAGIC, Buffer.from([0x00])]))).toEqual({ frames: [], tornStart: 0 })
|
||||
// Single-segment header followed by only two bytes of the three-byte block header.
|
||||
expect(scanZstdFrames(Buffer.concat([MAGIC, Buffer.from([0x20, 0x00, 0x01, 0x00])]))).toEqual({
|
||||
frames: [],
|
||||
tornStart: 0,
|
||||
})
|
||||
|
||||
const rawFiveBytes = Buffer.from([(5 << 3) | 1, 0, 0])
|
||||
expect(scanZstdFrames(Buffer.concat([
|
||||
MAGIC,
|
||||
Buffer.from([0x20, 0x00]),
|
||||
rawFiveBytes,
|
||||
Buffer.from([0x01, 0x02]),
|
||||
]))).toEqual({ frames: [], tornStart: 0 })
|
||||
|
||||
const reservedBlock = Buffer.concat([
|
||||
MAGIC,
|
||||
Buffer.from([0x20, 0x00, 0x07, 0x00, 0x00]),
|
||||
])
|
||||
expect(() => scanZstdFrames(reservedBlock)).toThrow(/reserved block type/)
|
||||
})
|
||||
|
||||
it('covers standard header variants, RLE blocks, multiple blocks, and checksums', () => {
|
||||
for (const descriptor of [0x00, 0x21, 0x42, 0x83, 0xE3]) {
|
||||
const frame = emptyStructuralFrame(descriptor)
|
||||
expect(scanZstdFrames(frame)).toEqual({ frames: [{ start: 0, end: frame.length }] })
|
||||
}
|
||||
|
||||
const rle = Buffer.concat([
|
||||
MAGIC,
|
||||
Buffer.from([0x20, 0x01]),
|
||||
Buffer.from([(1 << 3) | (1 << 1) | 1, 0, 0]),
|
||||
Buffer.from([0x41]),
|
||||
])
|
||||
expect(scanZstdFrames(rle)).toEqual({ frames: [{ start: 0, end: rle.length }] })
|
||||
|
||||
const twoBlocks = Buffer.concat([
|
||||
MAGIC,
|
||||
Buffer.from([0x20, 0x00]),
|
||||
Buffer.from([0, 0, 0]),
|
||||
Buffer.from([1, 0, 0]),
|
||||
])
|
||||
expect(scanZstdFrames(twoBlocks)).toEqual({ frames: [{ start: 0, end: twoBlocks.length }] })
|
||||
|
||||
const checksummed = emptyStructuralFrame(0x24)
|
||||
expect(scanZstdFrames(checksummed.subarray(0, -1))).toEqual({ frames: [], tornStart: 0 })
|
||||
expect(scanZstdFrames(checksummed)).toEqual({ frames: [{ start: 0, end: checksummed.length }] })
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
|
||||
it('writes .jsonl.zstd by default with one header frame and one first-batch frame', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('default-zstd', '/work')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
|
||||
const path = logPath(root, header.cwd, header.id, 'zstd')
|
||||
const buffer = await readFile(path)
|
||||
expect(buffer.subarray(0, 4)).toEqual(MAGIC)
|
||||
await expect(stat(logPath(root, header.cwd, header.id, 'none'))).rejects.toThrow()
|
||||
expect(ctx.sessionPersistence.locate(header)).toEqual({ kind: 'jsonl', path })
|
||||
|
||||
const scan = scanZstdFrames(buffer)
|
||||
expect(scan.frames).toHaveLength(2)
|
||||
const plaintext = await decodeCompleteFrames(buffer)
|
||||
expect(plaintext.toString()).toBe([
|
||||
JSON.stringify(toHeaderLine(header)),
|
||||
...oneTurnLog().map(e => JSON.stringify(e)),
|
||||
'',
|
||||
].join('\n'))
|
||||
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual(oneTurnLog())
|
||||
})
|
||||
|
||||
it('resolves the default when a programmatic wrapper bypasses Loader schema normalization', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = new Context()
|
||||
contexts.push(ctx)
|
||||
await ctx.plugin(SessionStore)
|
||||
let backend!: SessionPersistenceJsonl
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
backend = new SessionPersistenceJsonl(inner, { root })
|
||||
}, { inject: ['sessions'] }))
|
||||
const header = meta('direct-default')
|
||||
expect(backend.locate(header)).toEqual({
|
||||
kind: 'jsonl',
|
||||
path: logPath(root, header.cwd, header.id, 'zstd'),
|
||||
})
|
||||
})
|
||||
|
||||
it('appends one frame per durable batch without rewriting prior bytes', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('append-frame')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
const path = logPath(root, header.cwd, header.id, 'zstd')
|
||||
const before = await readFile(path)
|
||||
const secondTurn = [
|
||||
{ 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' } } },
|
||||
] as SessionEvent[]
|
||||
await ctx.sessionPersistence.append(header.id, secondTurn)
|
||||
|
||||
const after = await readFile(path)
|
||||
expect(after.subarray(0, before.length)).toEqual(before)
|
||||
expect(scanZstdFrames(after).frames).toHaveLength(3)
|
||||
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual([...oneTurnLog(), ...secondTurn])
|
||||
})
|
||||
|
||||
it('lists from a multi-chunk header frame without decoding a corrupt event frame', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('large-header', `/work/${'x'.repeat(24_000)}`)
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
const path = logPath(root, header.cwd, header.id, 'zstd')
|
||||
const buffer = Buffer.from(await readFile(path))
|
||||
const eventFrame = scanZstdFrames(buffer).frames[1]!
|
||||
buffer[eventFrame.end - 1] = buffer[eventFrame.end - 1]! ^ 0xFF
|
||||
await writeFile(path, buffer)
|
||||
|
||||
expect((await ctx.sessionPersistence.list()).map(item => item.id)).toEqual([header.id])
|
||||
await expect(ctx.sessionPersistence.load(header.id)).rejects.toThrow(/frame at byte .* failed validation/)
|
||||
})
|
||||
|
||||
it('preserves complete records from a torn frame and re-encodes them with crash closers', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('recover-torn', '/proj')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
const path = logPath(root, header.cwd, header.id, 'zstd')
|
||||
const committed = await readFile(path)
|
||||
const openTurn = [
|
||||
{ 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/chunk', seq: 8, time: 9, data: { turn: 2, step: 1, chunk: { type: 'text-delta', index: 0, text: deterministicNoise(300_000) } } },
|
||||
] as SessionEvent[]
|
||||
const plaintext = openTurn.map(e => JSON.stringify(e)).join('\n') + '\n'
|
||||
const partial = await tornFrame(plaintext, (decoded) => {
|
||||
const newlines = decoded.match(/\n/g)?.length ?? 0
|
||||
return newlines >= 2 && !decoded.endsWith('\n')
|
||||
})
|
||||
await appendFile(path, partial)
|
||||
|
||||
const loaded = await ctx.sessionPersistence.load(header.id)
|
||||
expect(loaded.events.map(event => event.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
|
||||
expect(loaded.events[6]).toEqual(openTurn[0])
|
||||
expect(loaded.events[7]).toEqual(openTurn[1])
|
||||
expect(loaded.events.some(event => event.type === 'assistant/chunk' && event.seq === 8)).toBe(false)
|
||||
expect(loaded.events[8]?.type).toBe('step/end')
|
||||
expect(loaded.events[9]?.type).toBe('turn/end')
|
||||
|
||||
const repaired = await readFile(path)
|
||||
expect(repaired.subarray(0, committed.length)).toEqual(committed)
|
||||
expect(scanZstdFrames(repaired).tornStart).toBeUndefined()
|
||||
expect(scanLog(await decodeCompleteFrames(repaired)).events).toEqual(loaded.events)
|
||||
})
|
||||
|
||||
it('drops a frame torn in its header before it has produced plaintext', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('partial-magic')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
const path = logPath(root, header.cwd, header.id, 'zstd')
|
||||
const committed = await readFile(path)
|
||||
await appendFile(path, MAGIC.subarray(0, 2))
|
||||
|
||||
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual(oneTurnLog())
|
||||
expect(await readFile(path)).toEqual(committed)
|
||||
})
|
||||
|
||||
it('recovers complete events when EOF tears only the final frame checksum', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('partial-checksum')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
const path = logPath(root, header.cwd, header.id, 'zstd')
|
||||
const secondTurn = [
|
||||
{ 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' } } },
|
||||
] as SessionEvent[]
|
||||
const frame = await compressZstdFrame(secondTurn.map(e => JSON.stringify(e)).join('\n') + '\n')
|
||||
await appendFile(path, frame.subarray(0, -1))
|
||||
|
||||
const loaded = await ctx.sessionPersistence.load(header.id)
|
||||
expect(loaded.events).toEqual([...oneTurnLog(), ...secondTurn])
|
||||
const repaired = await readFile(path)
|
||||
expect(scanZstdFrames(repaired).tornStart).toBeUndefined()
|
||||
expect(scanLog(await decodeCompleteFrames(repaired)).events).toEqual(loaded.events)
|
||||
})
|
||||
|
||||
it('rejects a complete frame containing a torn JSONL record', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('complete-bad-jsonl')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
await appendFile(
|
||||
logPath(root, header.cwd, header.id, 'zstd'),
|
||||
await compressZstdFrame('{"type":"turn/start"'),
|
||||
)
|
||||
await expect(ctx.sessionPersistence.load(header.id)).rejects.toThrow(/complete frame contains a torn JSONL record/)
|
||||
})
|
||||
|
||||
it('rolls back a checksummed append frame when fsync fails', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('zstd-fsync-rollback')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
const path = logPath(root, header.cwd, header.id, 'zstd')
|
||||
const before = await readFile(path)
|
||||
|
||||
const handle = await open(path, 'r')
|
||||
const prototype = Object.getPrototypeOf(handle) as { sync: () => Promise<void> }
|
||||
await handle.close()
|
||||
const realSync = prototype.sync
|
||||
let failed = false
|
||||
const spy = vi.spyOn(prototype, 'sync').mockImplementation(async function (this: FileHandle) {
|
||||
if (!failed) {
|
||||
failed = true
|
||||
throw new Error('simulated Zstandard fsync failure')
|
||||
}
|
||||
return realSync.call(this)
|
||||
})
|
||||
const secondTurn = [
|
||||
{ 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' } } },
|
||||
] as SessionEvent[]
|
||||
await expect(ctx.sessionPersistence.append(header.id, secondTurn)).rejects.toThrow(/simulated Zstandard fsync failure/)
|
||||
expect(await readFile(path)).toEqual(before)
|
||||
spy.mockRestore()
|
||||
await ctx.sessionPersistence.append(header.id, secondTurn)
|
||||
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual([...oneTurnLog(), ...secondTurn])
|
||||
})
|
||||
|
||||
it('skips empty, incomplete, and non-header compressed artifacts while rejecting malformed header frames', async () => {
|
||||
const root = await freshRoot()
|
||||
const bucket = sessionDir(root, undefined)
|
||||
await mkdir(bucket, { recursive: true })
|
||||
await writeFile(join(bucket, 'empty.jsonl.zstd'), '')
|
||||
await writeFile(join(bucket, 'partial.jsonl.zstd'), MAGIC)
|
||||
await writeFile(join(bucket, 'not-header.jsonl.zstd'), await compressZstdFrame('{"type":"turn/start"}\n'))
|
||||
const ctx = await mount(root)
|
||||
expect(await ctx.sessionPersistence.list()).toEqual([])
|
||||
|
||||
await writeFile(join(bucket, 'two-lines.jsonl.zstd'), await compressZstdFrame([
|
||||
JSON.stringify(toHeaderLine(meta('two-lines'))),
|
||||
JSON.stringify({ type: 'turn/start' }),
|
||||
'',
|
||||
].join('\n')))
|
||||
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/first frame is not exactly one header line/)
|
||||
await expect(ctx.sessionPersistence.load(SessionId('two-lines')))
|
||||
.rejects.toThrow(/first frame is not exactly one header line/)
|
||||
})
|
||||
|
||||
it('rejects missing, empty, and checksum-corrupt header frames on targeted reads', async () => {
|
||||
const root = await freshRoot()
|
||||
const bucket = sessionDir(root, undefined)
|
||||
await mkdir(bucket, { recursive: true })
|
||||
await writeFile(logPath(root, undefined, SessionId('partial-only'), 'zstd'), MAGIC)
|
||||
await writeFile(logPath(root, undefined, SessionId('empty-header'), 'zstd'), await compressZstdFrame(''))
|
||||
const corruptHeader = Buffer.from(await compressZstdFrame(`${JSON.stringify(toHeaderLine(meta('bad-checksum')))}\n`))
|
||||
corruptHeader[corruptHeader.length - 1] = corruptHeader[corruptHeader.length - 1]! ^ 0xFF
|
||||
await writeFile(logPath(root, undefined, SessionId('bad-checksum'), 'zstd'), corruptHeader)
|
||||
const ctx = await mount(root)
|
||||
|
||||
await expect(ctx.sessionPersistence.load(SessionId('partial-only')))
|
||||
.rejects.toThrow(/empty or header-less Zstandard session log/)
|
||||
await expect(ctx.sessionPersistence.load(SessionId('empty-header')))
|
||||
.rejects.toThrow(/first frame is not exactly one header line/)
|
||||
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/header frame failed validation/)
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: encoding selection', () => {
|
||||
it('rejects roots owned by the opposite encoding in both directions', async () => {
|
||||
const rawRoot = await freshRoot('dsh-jsonl-raw-mismatch-')
|
||||
const raw = await mount(rawRoot, 'none')
|
||||
const rawHeader = meta('raw-log')
|
||||
await raw.sessionPersistence.create(rawHeader)
|
||||
await raw.sessionPersistence.append(rawHeader.id, oneTurnLog())
|
||||
const defaultBackend = await mount(rawRoot)
|
||||
await expect(defaultBackend.sessionPersistence.list()).rejects.toThrow(/configured for compression "zstd"/)
|
||||
|
||||
const zstdRoot = await freshRoot('dsh-jsonl-zstd-mismatch-')
|
||||
const zstd = await mount(zstdRoot)
|
||||
const zstdHeader = meta('zstd-log')
|
||||
await zstd.sessionPersistence.create(zstdHeader)
|
||||
await zstd.sessionPersistence.append(zstdHeader.id, oneTurnLog())
|
||||
const rawBackend = await mount(zstdRoot, 'none')
|
||||
await expect(rawBackend.sessionPersistence.list()).rejects.toThrow(/configured for compression "none"/)
|
||||
})
|
||||
|
||||
it('rechecks targeted artifacts and listing after an initially empty root', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
expect(await ctx.sessionPersistence.list()).toEqual([])
|
||||
|
||||
const loadHeader = meta('late-raw-load', '/late')
|
||||
await mkdir(sessionDir(root, loadHeader.cwd), { recursive: true })
|
||||
await writeFile(logPath(root, loadHeader.cwd, loadHeader.id, 'none'), [
|
||||
JSON.stringify(toHeaderLine(loadHeader)),
|
||||
...oneTurnLog().map(e => JSON.stringify(e)),
|
||||
'',
|
||||
].join('\n'))
|
||||
await expect(ctx.sessionPersistence.load(loadHeader.id)).rejects.toThrow(/uses \.jsonl/)
|
||||
await expect((ctx.sessionPersistence as SessionPersistenceJsonl).loadLive(loadHeader.id, loadHeader.cwd))
|
||||
.rejects.toThrow(/uses \.jsonl/)
|
||||
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/uses \.jsonl/)
|
||||
})
|
||||
|
||||
it('refuses materialization when an opposite artifact appears after create', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
await ctx.sessionPersistence.list()
|
||||
const header = meta('late-raw-materialize', '/late')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await mkdir(sessionDir(root, header.cwd), { recursive: true })
|
||||
await writeFile(logPath(root, header.cwd, header.id, 'none'), [
|
||||
JSON.stringify(toHeaderLine(header)),
|
||||
...oneTurnLog().map(e => JSON.stringify(e)),
|
||||
'',
|
||||
].join('\n'))
|
||||
await expect(ctx.sessionPersistence.append(header.id, oneTurnLog())).rejects.toThrow(/uses \.jsonl/)
|
||||
expect((await readdir(sessionDir(root, header.cwd))).some(name => name.endsWith('.jsonl.zstd'))).toBe(false)
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user