imccyu
92f9e79fac
A Session-list row no longer carries every registered client view: the three units whose values a list row never renders (`turnOutline`, `sessionStats`, and `todos`) are excluded, which removes the whole-log turn outline and the figure and todo payloads from every row. An opened Session is unchanged — `session/follow`, the `session/control` baseline, and `session.projections` still serve every projected key. The exclusion is one type-level mark plus one registration interlock: `SessionProjectionSummaryExcluded<T>` marks a map entry, and `ProjectionDefinition.wire` requires `summary: false` exactly on a marked key and rejects it everywhere else, so the mark and the declaration cannot drift. A nullable entry keeps `null` outside the wrapper, so `todos` stays `TodoItem[] | null`. `summaryKeys()` returns the derived set, stable until a registration or its disposal changes the key table, and the existing key-taking read faces (`cachedSnapshot`, `viewCheckpoint`) accept that set directly, so summaries need no separate read face and no per-read key copy. `packages/api/session-controller/src/list.ts` passes the set to the live and cold cached faces; `cachedPredecessorTitle` is untouched. Client types only: `SessionSummary.projectionValues` narrows to `SessionProjectionSummaryValues`, `SessionProjectionSnapshot.values` keeps every view value with the mark stripped, and no wire field or client name changed. The cache, subsystem page, and package README document the summary rule and the two read faces.
855 lines
37 KiB
TypeScript
855 lines
37 KiB
TypeScript
/**
|
|
* SessionProjectionCache behavior: mandatory-point writes (turn/end, detach),
|
|
* count/interval throttling between them, fail-soft durability (a failed
|
|
* write logs and stays stale, never throws into the event path), and the
|
|
* synchronous cached listing read. The durable medium is the
|
|
* `session_projcache` storage domain in per-record layout: one
|
|
* version-stamped document per session under the json backend root at
|
|
* `<root>/session_projcache/sessions/<id>.json`. Reads never touch the
|
|
* medium — they come from the domain's in-memory tables, which writes mutate
|
|
* only after durability.
|
|
*/
|
|
|
|
import { afterEach, describe, expect, it, vi } from 'vitest'
|
|
import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
|
|
import { tmpdir } from 'node:os'
|
|
import { dirname, join } from 'node:path'
|
|
import { Context } from '@deepseek-ai/cordis'
|
|
import { z } from 'zod'
|
|
import SessionStore, {
|
|
SESSION_FORMAT_VERSION,
|
|
Session,
|
|
SessionId,
|
|
SessionLogOffset,
|
|
SessionSeq,
|
|
} from '@deepseek-ai/dsh-session'
|
|
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
|
|
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
|
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
|
|
import type { SessionProjectionSummaryExcluded } from '@deepseek-ai/dsh-session-projection/types'
|
|
import Storage from '@deepseek-ai/dsh-storage'
|
|
import {
|
|
apply as storageJsonApply, Config as storageJsonConfig, inject as storageJsonInject, name as storageJsonName,
|
|
} from '@deepseek-ai/dsh-storage-json'
|
|
import {
|
|
apply as storageDomainApply, Config as storageDomainConfig, inject as storageDomainInject, name as storageDomainName,
|
|
} from '@deepseek-ai/dsh-storage-domain'
|
|
import SessionProjectionCache from '../src/index.ts'
|
|
import { checkpointRecord, projectionCacheDomainSpec } from '../src/spec.ts'
|
|
import type { CheckpointRecord } from '../src/spec.ts'
|
|
|
|
declare module '@deepseek-ai/dsh-session-projection/types' {
|
|
interface SessionProjectionStateMap {
|
|
'cache-test/marks': MarksState
|
|
'cache-test/secondary-marks': MarksState
|
|
'cache-test/marks2': Map<string, string>
|
|
'cache-test/count': number
|
|
'cache-test/secret': string
|
|
'cache-test/marks3': MarksState
|
|
'cache-test/summary-excluded': { n: number }
|
|
title: string | null
|
|
}
|
|
interface SessionProjectionMap {
|
|
'cache-test/marks': { marks: string[] }
|
|
'cache-test/secondary-marks': { marks: string[] }
|
|
'cache-test/marks3': { marks: string[] }
|
|
'cache-test/summary-excluded': SessionProjectionSummaryExcluded<{ n: number }>
|
|
title: string | null
|
|
}
|
|
}
|
|
|
|
declare module '@deepseek-ai/dsh-session/types' {
|
|
interface SessionEventMap {
|
|
'cache-test/mark': { marks: string[] }
|
|
}
|
|
|
|
interface OutOfBandSessionEventMap {
|
|
'cache-test/mark': true
|
|
}
|
|
}
|
|
|
|
type MarksState = { marks: string[] } | null
|
|
const marksUnit = (stateVersion = 1) => ({
|
|
key: 'cache-test/marks',
|
|
stateSchema: z.object({ marks: z.array(z.string()) }).nullable(),
|
|
init: () => null,
|
|
apply: (state, event) => (event.type === 'cache-test/mark' ? (event).data : state),
|
|
wire: {
|
|
viewSchema: z.object({ marks: z.array(z.string()) }),
|
|
view: state => state ?? { marks: [] },
|
|
},
|
|
stateVersion,
|
|
}) satisfies ProjectionDefinition<'cache-test/marks', MarksState>
|
|
|
|
const marks3Unit = {
|
|
key: 'cache-test/marks3',
|
|
stateSchema: z.object({ marks: z.array(z.string()) }).nullable(),
|
|
init: () => null,
|
|
apply: state => state,
|
|
wire: {
|
|
viewSchema: z.object({ marks: z.array(z.string()) }),
|
|
view: state => state ?? { marks: [] },
|
|
},
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/marks3', MarksState>
|
|
|
|
const secretUnit = {
|
|
key: 'cache-test/secret',
|
|
stateSchema: z.string(),
|
|
init: () => '',
|
|
apply: state => state,
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/secret', string>
|
|
|
|
const secondaryMarksUnit = {
|
|
key: 'cache-test/secondary-marks',
|
|
stateSchema: z.object({ marks: z.array(z.string()) }).nullable(),
|
|
init: () => null,
|
|
apply: (state, event) => event.type === 'cache-test/mark' ? event.data : state,
|
|
wire: {
|
|
viewSchema: z.object({ marks: z.array(z.string()) }),
|
|
view: state => state ?? { marks: [] },
|
|
},
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/secondary-marks', MarksState>
|
|
|
|
/** A summary-excluded unit: its rows must never reach the summary block. */
|
|
const summaryExcludedUnit = {
|
|
key: 'cache-test/summary-excluded',
|
|
stateSchema: z.object({ n: z.number() }),
|
|
init: () => ({ n: 0 }),
|
|
apply: state => state,
|
|
wire: { viewSchema: z.object({ n: z.number() }), view: state => state, summary: false },
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/summary-excluded', { n: number }>
|
|
|
|
/** Mirrors the shipped title unit's storage face (stateVersion 1, bare-string state). */
|
|
const titleUnit = {
|
|
key: 'title',
|
|
stateSchema: z.string().nullable(),
|
|
init: () => null,
|
|
apply: state => state,
|
|
wire: { viewSchema: z.string().nullable(), view: state => state },
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'title', string | null>
|
|
|
|
/** One session's record document on the per-record medium. */
|
|
const recordPath = (root: string, id: Session['id']): string =>
|
|
join(root, projectionCacheDomainSpec.name, 'sessions', `${String(id)}.json`)
|
|
|
|
/** Header shape for cachedSnapshot calls. */
|
|
const headerOf = (id: SessionId, createdAt = 0, cwd?: string): SessionHeader =>
|
|
({ version: SESSION_FORMAT_VERSION, id, createdAt, isSeeded: false, ...cwd === undefined ? {} : { cwd } })
|
|
|
|
interface HarnessOptions {
|
|
root?: string
|
|
config?: { writeEveryEvents: number; writeIntervalMs: number }
|
|
stateVersion?: number
|
|
}
|
|
|
|
const contexts: Context[] = []
|
|
const roots: string[] = []
|
|
|
|
async function harness(options: HarnessOptions = {}) {
|
|
const root = options.root ?? await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const ctx = new Context()
|
|
contexts.push(ctx)
|
|
// The cache opens its domain through the storage stack; the json backend
|
|
// lands the per-record tree under this tmp root.
|
|
await ctx.plugin(Storage)
|
|
await ctx.plugin({ name: storageJsonName, inject: storageJsonInject, apply: storageJsonApply, Config: storageJsonConfig }, { root })
|
|
await ctx.plugin({ name: storageDomainName, inject: storageDomainInject, apply: storageDomainApply, Config: storageDomainConfig }, { backend: 'json' })
|
|
await ctx.plugin(SessionStore)
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
ctx.sessionProjections.register(marksUnit(options.stateVersion))
|
|
const fiber = await ctx.plugin(SessionProjectionCache, options.config ?? { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
return { ctx, root, fiber, cache: ctx.sessionProjectionCache }
|
|
}
|
|
|
|
const mark = (session: Session, marks: string[]): SessionEvent =>
|
|
session.append('cache-test/mark', { marks })
|
|
|
|
const endTurn = (session: Session): SessionEvent =>
|
|
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
|
|
|
/** Resolve after this Session's next durable cache replacement. */
|
|
function whenWritten(ctx: Context, id: SessionId): Promise<void> {
|
|
return new Promise((resolve) => {
|
|
const dispose = ctx.on('domain/changed', (change) => {
|
|
if (change.domain !== projectionCacheDomainSpec.name
|
|
|| change.table !== 'sessions' || change.key !== id || change.operation !== 'put') return
|
|
dispose()
|
|
resolve()
|
|
})
|
|
})
|
|
}
|
|
|
|
/** The stored record for one session id (undefined = absent or unreadable). */
|
|
async function storedRecord(root: string, id: Session['id']): Promise<CheckpointRecord | undefined> {
|
|
try {
|
|
const document = JSON.parse(await readFile(recordPath(root, id), 'utf8')) as { record: unknown }
|
|
return checkpointRecord.parse(document.record)
|
|
} catch {
|
|
return undefined
|
|
}
|
|
}
|
|
|
|
/** The stored rows for one session id (undefined = absent or unreadable). */
|
|
async function storedRows(root: string, id: Session['id']): Promise<CheckpointRecord['rows'] | undefined> {
|
|
return (await storedRecord(root, id))?.rows
|
|
}
|
|
|
|
/** Pre-seed one session's record document with a stored checkpoint record. */
|
|
async function seedRecord(
|
|
root: string,
|
|
id: string,
|
|
rows: CheckpointRecord['rows'],
|
|
identity: CheckpointRecord['identity'] = {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 0,
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
},
|
|
): Promise<void> {
|
|
const path = recordPath(root, SessionId(id))
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, JSON.stringify({ version: projectionCacheDomainSpec.version, record: { identity, rows } }))
|
|
}
|
|
|
|
afterEach(async () => {
|
|
vi.useRealTimers()
|
|
await Promise.all(contexts.splice(0).map(ctx => ctx.fiber.dispose()))
|
|
await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 })))
|
|
})
|
|
|
|
describe('SessionProjectionCache write policy', () => {
|
|
it('writes a durable checkpoint at turn/end (mandatory point)', async () => {
|
|
const { ctx, root } = await harness()
|
|
// The interval cannot substitute for the mandatory turn/end trigger.
|
|
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
|
const id = SessionId('turn-end')
|
|
const created = whenWritten(ctx, id)
|
|
const session = ctx.sessions.create(id)
|
|
mark(session, ['a'])
|
|
// The mark is throttled, so the creation cut has no marks folded.
|
|
await created
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks']?.seq).toBe(-1)
|
|
const written = whenWritten(ctx, id)
|
|
const end = endTurn(session)
|
|
await written
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks'])
|
|
.toEqual({ ver: 1, seq: end.seq, val: { marks: ['a'] } })
|
|
})
|
|
|
|
it('writes a checkpoint at session creation, capturing the seed-derived cut', async () => {
|
|
const { ctx, root } = await harness()
|
|
// A forked child seeded with its ancestor's title-like event: no
|
|
// conversation follows, yet the creation write must capture the fold so
|
|
// a crash or a live-held fork still lists the derived value.
|
|
const id = SessionId('seeded')
|
|
const created = whenWritten(ctx, id)
|
|
const session = ctx.sessions.create(id, {
|
|
seed: [{ type: 'cache-test/mark', seq: 0, time: 1, data: { marks: ['seed'] } }] as SessionEvent[],
|
|
})
|
|
await created
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks']?.val)
|
|
.toEqual({ marks: ['seed'] })
|
|
})
|
|
|
|
it('writes at session disposal (detach, the live-to-cold moment)', async () => {
|
|
const { ctx, root } = await harness()
|
|
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
|
const id = SessionId('detach')
|
|
const created = whenWritten(ctx, id)
|
|
// Sessions dispose with their owning fiber: create in a child plugin.
|
|
let session: Session | undefined
|
|
const owner = await ctx.plugin(Object.assign((inner: Context) => {
|
|
session = inner.sessions.create(id)
|
|
}, { inject: ['sessions'] }))
|
|
if (session === undefined) throw new Error('session was not created')
|
|
await created
|
|
mark(session, ['live'])
|
|
const written = whenWritten(ctx, id)
|
|
await owner.dispose()
|
|
await written
|
|
expect((await storedRows(root, id))?.['cache-test/marks']?.val).toEqual({ marks: ['live'] })
|
|
})
|
|
|
|
it('flushes when the in-turn event count reaches the configured threshold', async () => {
|
|
const { ctx, root } = await harness({ config: { writeEveryEvents: 3, writeIntervalMs: 60_000 } })
|
|
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
|
const id = SessionId('count')
|
|
const created = whenWritten(ctx, id)
|
|
const session = ctx.sessions.create(id)
|
|
mark(session, ['1'])
|
|
mark(session, ['2'])
|
|
await created
|
|
expect((await storedRows(root, id))?.['cache-test/marks'])
|
|
.toEqual({ ver: 1, seq: -1, val: null }) // still the creation cut
|
|
const written = whenWritten(ctx, id)
|
|
mark(session, ['3'])
|
|
await written
|
|
expect((await storedRows(root, id))?.['cache-test/marks']?.val).toEqual({ marks: ['3'] })
|
|
})
|
|
|
|
it('flushes on the configured interval when the count threshold is not reached', async () => {
|
|
const { ctx, cache } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 20 } })
|
|
const write = vi.spyOn(cache, 'write').mockResolvedValue()
|
|
vi.useFakeTimers()
|
|
const session = ctx.sessions.create(SessionId('interval'))
|
|
write.mockClear()
|
|
mark(session, ['slow'])
|
|
await vi.advanceTimersByTimeAsync(19)
|
|
expect(write).not.toHaveBeenCalled()
|
|
await vi.advanceTimersByTimeAsync(1)
|
|
expect(write).toHaveBeenCalledExactlyOnceWith(session)
|
|
})
|
|
|
|
it('write() on a never-dirty session checkpoints directly and rejects a non-JSON unit state', async () => {
|
|
const { ctx, root } = await harness()
|
|
// Never dirtied: no events — write() still lands the init-derived cut.
|
|
const clean = ctx.sessions.create(SessionId('clean-write'))
|
|
await ctx.sessionProjectionCache.write(clean)
|
|
expect((await storedRows(root, clean.id))?.['cache-test/marks']).toEqual({ ver: 1, seq: -1, val: null })
|
|
// A unit whose state violates the plain-JSON contract fails the write loud.
|
|
ctx.sessionProjections.register({
|
|
key: 'cache-test/marks2',
|
|
stateSchema: z.custom<Map<string, string>>(() => true),
|
|
init: () => new Map<string, string>(),
|
|
apply: state => state,
|
|
stateVersion: 1,
|
|
})
|
|
await expect(ctx.sessionProjectionCache.write(clean)).rejects.toThrow('not losslessly JSON-serializable')
|
|
})
|
|
|
|
it('plugin disposal clears armed interval timers and leaves cleaned sessions alone', async () => {
|
|
vi.useFakeTimers()
|
|
const { ctx, root, fiber } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 5000 } })
|
|
const armed = ctx.sessions.create(SessionId('armed'))
|
|
const cleaned = ctx.sessions.create(SessionId('cleaned'))
|
|
mark(armed, ['pending']) // timer armed, no write yet
|
|
mark(cleaned, ['done'])
|
|
endTurn(cleaned) // mandatory write; markClean leaves {pending: 0, timer: undefined} in the map
|
|
await vi.advanceTimersByTimeAsync(0)
|
|
await fiber.dispose()
|
|
// The armed timer died with the plugin: advancing time writes nothing.
|
|
await vi.advanceTimersByTimeAsync(10_000)
|
|
// Only the creation cut exists: the armed mark never wrote.
|
|
expect((await storedRows(root, armed.id))?.['cache-test/marks']?.seq).toBe(-1)
|
|
})
|
|
|
|
it('contains a durable write failure: logs a warning, event path unharmed, next write self-heals', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const ctx = new Context()
|
|
contexts.push(ctx)
|
|
await ctx.plugin(Storage)
|
|
await ctx.plugin({ name: storageJsonName, inject: storageJsonInject, apply: storageJsonApply, Config: storageJsonConfig }, { root })
|
|
await ctx.plugin({ name: storageDomainName, inject: storageDomainInject, apply: storageDomainApply, Config: storageDomainConfig }, { backend: 'json' })
|
|
await ctx.plugin(SessionStore)
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
ctx.sessionProjections.register(marksUnit())
|
|
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
|
// A directory where the record document must land makes the atomic
|
|
// rename fail — including the creation write, so no row ever lands.
|
|
const blocker = recordPath(root, SessionId('fail-soft'))
|
|
await mkdir(blocker, { recursive: true })
|
|
const session = ctx.sessions.create(SessionId('fail-soft'))
|
|
mark(session, ['x'])
|
|
endTurn(session)
|
|
// The failed creation/turn-end writes are fire-and-forget: wait for the
|
|
// warn (the write actually failed), then assert no row landed — the
|
|
// property under test is that a failed write leaves no partial row.
|
|
await vi.waitFor(() => {
|
|
expect(warn).toHaveBeenCalledWith(expect.stringContaining('turn/end write for "fail-soft" failed'))
|
|
}, { timeout: 5_000 })
|
|
expect(await storedRows(root, session.id)).toBeUndefined()
|
|
// Self-heal: once the blocker clears, the next mandatory point writes.
|
|
await rm(recordPath(root, session.id), { recursive: true })
|
|
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
|
mark(session, ['y'])
|
|
const written = whenWritten(ctx, session.id)
|
|
endTurn(session)
|
|
await written
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['y'] })
|
|
})
|
|
})
|
|
|
|
describe('SessionProjectionCache listing read', () => {
|
|
it('rejects a nonzero inherited cut for an unseeded header on the fold face', async () => {
|
|
const { cache } = await harness()
|
|
|
|
expect(() => cache.coldSnapshot(
|
|
headerOf(SessionId('invalid-unseeded-cut')),
|
|
SessionLogOffset(1),
|
|
[],
|
|
)).toThrow('unseeded projection-cache identity inherited event count must be 0')
|
|
})
|
|
|
|
it('serves rows with differing watermarks as one block at the lowest served watermark', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'watermark-lower', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['primary'] } },
|
|
'cache-test/secondary-marks': { ver: 1, seq: SessionSeq(2), val: { marks: ['secondary'] } },
|
|
})
|
|
await seedRecord(root, 'watermark-higher', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['primary'] } },
|
|
'cache-test/secondary-marks': { ver: 1, seq: SessionSeq(6), val: { marks: ['secondary'] } },
|
|
})
|
|
const { ctx, cache } = await harness({ root })
|
|
ctx.sessionProjections.register(secondaryMarksUnit)
|
|
|
|
// The block carries one watermark: the seq every served value has folded
|
|
// through at least, so the lower of the two rows names the block.
|
|
for (const [id, asOfSeq] of [['watermark-lower', 2], ['watermark-higher', 4]] as const) {
|
|
expect(cache.cachedSnapshot(headerOf(SessionId(id)))).toEqual({
|
|
asOfSeq,
|
|
values: {
|
|
'cache-test/marks': { marks: ['primary'] },
|
|
'cache-test/secondary-marks': { marks: ['secondary'] },
|
|
},
|
|
})
|
|
}
|
|
})
|
|
|
|
it('serves a seeded lifecycle by header alone while the fold face still refuses another cut', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('cut-identity')
|
|
await seedRecord(
|
|
root,
|
|
id,
|
|
{ 'cache-test/marks': { ver: 1, seq: SessionSeq(1), val: { marks: ['seed'] } } },
|
|
{
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 0,
|
|
isSeeded: true,
|
|
inheritedEventCount: SessionLogOffset(2),
|
|
},
|
|
)
|
|
const { cache } = await harness({ root })
|
|
const seededHeader = { ...headerOf(id), isSeeded: true }
|
|
const log: SessionEvent[] = [
|
|
{ type: 'turn/start', seq: SessionSeq(0), time: 0, data: { turn: 1 } },
|
|
{ type: 'cache-test/mark', seq: SessionSeq(1), time: 1, data: { marks: ['logged'] } },
|
|
]
|
|
|
|
// The listing knows only the header: the record's lifecycle matches, so
|
|
// its rows are viewed without any cut.
|
|
expect(cache.cachedSnapshot(seededHeader))
|
|
.toEqual({ asOfSeq: 1, values: { 'cache-test/marks': { marks: ['seed'] } } })
|
|
// An unseeded header names another lifecycle even though the id matches.
|
|
expect(cache.cachedSnapshot(headerOf(id))).toBeUndefined()
|
|
// The fold face holds the exact cut: the matching cut seeds from the row,
|
|
// another cut refolds the whole log instead of continuing a foreign state.
|
|
expect(cache.coldSnapshot(seededHeader, SessionLogOffset(2), log).values['cache-test/marks'])
|
|
.toEqual({ marks: ['seed'] })
|
|
expect(cache.coldSnapshot(seededHeader, SessionLogOffset(1), log).values['cache-test/marks'])
|
|
.toEqual({ marks: ['logged'] })
|
|
})
|
|
|
|
it('serves a seeded lifecycle\'s predecessor title by header alone', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('seeded-predecessor')
|
|
await seedRecord(
|
|
root,
|
|
id,
|
|
{
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(3), val: { marks: ['fork'] } },
|
|
title: { ver: 1, seq: SessionSeq(3), val: 'forked title' },
|
|
},
|
|
{
|
|
formatVersion: SESSION_FORMAT_VERSION - 1,
|
|
createdAt: 0,
|
|
isSeeded: true,
|
|
inheritedEventCount: SessionLogOffset(5),
|
|
},
|
|
)
|
|
const { ctx, cache } = await harness({ root })
|
|
ctx.sessionProjections.register(titleUnit)
|
|
const seededHeader = { ...headerOf(id), isSeeded: true }
|
|
|
|
// An older format generation never serves the full block, but its title
|
|
// survives the format edge, for a seeded lifecycle as for an unseeded one.
|
|
expect(cache.cachedSnapshot(seededHeader)).toBeUndefined()
|
|
expect(cache.cachedPredecessorTitle(seededHeader))
|
|
.toEqual({ asOfSeq: 3, values: { title: 'forked title' } })
|
|
// An unseeded header names another lifecycle: no title either.
|
|
expect(cache.cachedPredecessorTitle(headerOf(id))).toBeUndefined()
|
|
})
|
|
|
|
it('serves a creation-time checkpoint at the before-first-event cursor', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'before-first-event', {
|
|
'cache-test/marks': { ver: 1, seq: -1, val: null },
|
|
})
|
|
const { cache } = await harness({ root })
|
|
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('before-first-event'))))
|
|
.toEqual({ asOfSeq: -1, values: { 'cache-test/marks': { marks: [] } } })
|
|
})
|
|
|
|
it('refuses a checkpoint folded from another Session format generation', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('format-identity')
|
|
await seedRecord(
|
|
root,
|
|
id,
|
|
{ 'cache-test/marks': { ver: 1, seq: SessionSeq(0), val: { marks: ['stale'] } } },
|
|
{
|
|
formatVersion: SESSION_FORMAT_VERSION + 1,
|
|
createdAt: 0,
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
},
|
|
)
|
|
const { cache } = await harness({ root })
|
|
|
|
expect(cache.cachedSnapshot(headerOf(id))).toBeUndefined()
|
|
})
|
|
|
|
it('keeps host-only checkpoint state out of cached wire snapshots', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'host-state', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['wire'] } },
|
|
'cache-test/secret': { ver: 1, seq: SessionSeq(4), val: 'private prompt text' },
|
|
})
|
|
const { ctx, cache } = await harness({ root })
|
|
ctx.sessionProjections.register(secretUnit)
|
|
const header = headerOf(SessionId('host-state'))
|
|
|
|
expect(cache.cachedSnapshot(header)).toEqual({
|
|
asOfSeq: 4,
|
|
values: { 'cache-test/marks': { marks: ['wire'] } },
|
|
})
|
|
expect(JSON.stringify(cache.cachedSnapshot(header)))
|
|
.not.toContain('private prompt text')
|
|
})
|
|
|
|
it('serves lifecycle-matching rows as a cached block and refuses unrelated ones', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'listed', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['t'] } },
|
|
})
|
|
const { cache } = await harness({ root })
|
|
const id = SessionId('listed')
|
|
// Matching header: the values at the stored row's own watermark.
|
|
expect(cache.cachedSnapshot(headerOf(id)))
|
|
.toEqual({ asOfSeq: 4, values: { 'cache-test/marks': { marks: ['t'] } } })
|
|
// A recreated id (different createdAt): the record is unrelated — no block.
|
|
expect(cache.cachedSnapshot(headerOf(id, 777))).toBeUndefined()
|
|
// Unknown id: no block.
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('never-cached'))))
|
|
.toBeUndefined()
|
|
})
|
|
|
|
it('views every served wire row of one record in one cached block', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'multi-row', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['a'] } },
|
|
'cache-test/marks3': { ver: 1, seq: SessionSeq(4), val: { marks: ['b'] } },
|
|
})
|
|
const { ctx, cache } = await harness({ root })
|
|
ctx.sessionProjections.register(marks3Unit)
|
|
const block = cache.cachedSnapshot(headerOf(SessionId('multi-row')))
|
|
expect(block?.values).toEqual({
|
|
'cache-test/marks': { marks: ['a'] },
|
|
'cache-test/marks3': { marks: ['b'] },
|
|
})
|
|
expect(block?.asOfSeq).toBe(4)
|
|
})
|
|
|
|
it('returns undefined when the stored record version is not accepted', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// Version 2 is neither current nor declared compatible, so the document
|
|
// is discarded at open and the record reads as absent.
|
|
const path = recordPath(root, SessionId('all-stale'))
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, JSON.stringify({
|
|
version: 2,
|
|
record: {
|
|
identity: {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 0,
|
|
isSeeded: false,
|
|
inheritedEventCount: 0,
|
|
},
|
|
rows: { 'cache-test/marks': { ver: 1, seq: 4, val: { marks: ['old'] } } },
|
|
},
|
|
}))
|
|
const { cache } = await harness({ root })
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('all-stale'))))
|
|
.toBeUndefined()
|
|
})
|
|
|
|
it('serves a pre-lineage record (accepted old version) to an unseeded caller only', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// A document stamped with an accepted older version whose identity
|
|
// predates the lineage fields: absent lineage reads as unseeded.
|
|
const path = recordPath(root, SessionId('pre-lineage'))
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, JSON.stringify({
|
|
version: 4,
|
|
record: {
|
|
identity: { formatVersion: SESSION_FORMAT_VERSION, createdAt: 0 },
|
|
rows: { 'cache-test/marks': { ver: 1, seq: 4, val: { marks: ['kept'] } } },
|
|
},
|
|
}))
|
|
const { cache } = await harness({ root })
|
|
const id = SessionId('pre-lineage')
|
|
// Unseeded caller: the absent lineage is exactly its identity — served.
|
|
expect(cache.cachedSnapshot(headerOf(id)))
|
|
.toEqual({ asOfSeq: 4, values: { 'cache-test/marks': { marks: ['kept'] } } })
|
|
// Seeded caller: the lineage-less record cannot vouch for a seeded lifecycle — refused.
|
|
expect(cache.cachedSnapshot({ ...headerOf(id), isSeeded: true }))
|
|
.toBeUndefined()
|
|
// The fold face reads the absent lineage as the unseeded cut 0: the
|
|
// unseeded caller continues from the row, the seeded caller refolds.
|
|
const log: SessionEvent[] = [0, 1, 2, 3, 4].map(seq => ({
|
|
type: 'cache-test/mark', seq: SessionSeq(seq), time: seq, data: { marks: [`m${seq}`] },
|
|
}))
|
|
expect(cache.coldSnapshot(headerOf(id), SessionLogOffset(0), log).values['cache-test/marks'])
|
|
.toEqual({ marks: ['kept'] })
|
|
expect(cache.coldSnapshot({ ...headerOf(id), isSeeded: true }, SessionLogOffset(1), log).values['cache-test/marks'])
|
|
.toEqual({ marks: ['m4'] })
|
|
})
|
|
|
|
it('refuses an accepted predecessor record without a Session format generation', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('pre-format-identity')
|
|
const path = recordPath(root, id)
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, JSON.stringify({
|
|
version: 6,
|
|
record: {
|
|
identity: { createdAt: 0, isSeeded: false, inheritedEventCount: 0 },
|
|
rows: { 'cache-test/marks': { ver: 1, seq: 4, val: { marks: ['unbound'] } } },
|
|
},
|
|
}))
|
|
const { cache } = await harness({ root })
|
|
|
|
expect(cache.cachedSnapshot(headerOf(id))).toBeUndefined()
|
|
})
|
|
|
|
it('returns undefined when every stored row is version-mismatched', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// A current document whose rows all fail the live unit's stateVersion:
|
|
// the listing view is empty, so no block is served.
|
|
await seedRecord(root, 'row-stale', {
|
|
'cache-test/marks': { ver: 99, seq: SessionSeq(4), val: { marks: ['old'] } },
|
|
})
|
|
const { cache } = await harness({ root })
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('row-stale'))))
|
|
.toBeUndefined()
|
|
})
|
|
|
|
it('binds identity on cwd too: a matching cwd serves, a moved session does not', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'homed', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(2), val: { marks: ['w'] } },
|
|
}, {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 0,
|
|
cwd: '/work',
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
})
|
|
const { cache } = await harness({ root })
|
|
const id = SessionId('homed')
|
|
expect(cache.cachedSnapshot(headerOf(id, 0, '/work'))?.values['cache-test/marks'])
|
|
.toEqual({ marks: ['w'] })
|
|
expect(cache.cachedSnapshot(headerOf(id, 0, '/elsewhere'))).toBeUndefined()
|
|
expect(cache.cachedSnapshot(headerOf(id, 0))).toBeUndefined()
|
|
})
|
|
|
|
it('returns undefined for a malformed record document (refold from the log on the caller side)', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const path = recordPath(root, SessionId('malformed'))
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, 'not json at all')
|
|
const { cache } = await harness({ root })
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('malformed'))))
|
|
.toBeUndefined()
|
|
})
|
|
})
|
|
|
|
describe('SessionProjectionCache cold-read seeding', () => {
|
|
/** One session's event log: turn/start, one mark per group, turn/end. */
|
|
const storedLog = (marks: string[][]): SessionEvent[] => {
|
|
const events: SessionEvent[] = [
|
|
{ type: 'turn/start', seq: SessionSeq(0), time: 0, data: { turn: 1 } },
|
|
]
|
|
for (const m of marks) {
|
|
events.push({
|
|
type: 'cache-test/mark',
|
|
seq: SessionSeq(events.length),
|
|
time: events.length,
|
|
data: { marks: m },
|
|
})
|
|
}
|
|
events.push({
|
|
type: 'turn/end',
|
|
seq: SessionSeq(events.length),
|
|
time: events.length,
|
|
data: { turn: 1, reason: { kind: 'completed' } },
|
|
})
|
|
return events
|
|
}
|
|
|
|
it('hydratePrepared seeds from a matching row and retries from the exact log on a malformed one', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// Records land on disk before the domain opens, so the in-memory table
|
|
// picks them up at init.
|
|
await seedRecord(root, 'prepared-seeded', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(1), val: { marks: ['cached'] } },
|
|
})
|
|
await seedRecord(root, 'prepared-fallback', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(1), val: { marks: 'malformed' } },
|
|
})
|
|
const { cache } = await harness({ root })
|
|
const events = storedLog([['fresh']])
|
|
|
|
// A matching row hydrates the prepared Session without a persistence read.
|
|
const seeded = headerOf(SessionId('prepared-seeded'))
|
|
const seededSession = Session.create(seeded.id, events, seeded)
|
|
expect(cache.hydratePrepared(seededSession, events)).toEqual({
|
|
asOfSeq: 2,
|
|
values: { 'cache-test/marks': { marks: ['cached'] } },
|
|
})
|
|
|
|
// A malformed row cannot seed the fold; hydration falls back to the
|
|
// exact log so a valid Session stays readable.
|
|
const fallback = headerOf(SessionId('prepared-fallback'))
|
|
const fallbackSession = Session.create(fallback.id, events, fallback)
|
|
expect(cache.hydratePrepared(fallbackSession, events)).toEqual({
|
|
asOfSeq: 2,
|
|
values: { 'cache-test/marks': { marks: ['fresh'] } },
|
|
})
|
|
|
|
// No row at all: hydrate from init over the exact log.
|
|
const bare = headerOf(SessionId('prepared-bare'))
|
|
const bareSession = Session.create(bare.id, events, bare)
|
|
expect(cache.hydratePrepared(bareSession, events)).toEqual({
|
|
asOfSeq: 2,
|
|
values: { 'cache-test/marks': { marks: ['fresh'] } },
|
|
})
|
|
})
|
|
|
|
it('coldSnapshot traverses the full log but applies only the events after each cached watermark', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// A cached row covering the prefix through seq 2 (three applies folded).
|
|
await seedRecord(root, 'cold-snap', {
|
|
'cache-test/count': { ver: 1, seq: SessionSeq(2), val: 3 },
|
|
}, {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 9,
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
})
|
|
const { cache, ctx } = await harness({ root })
|
|
const apply = vi.fn((_state: number, _event: SessionEvent) => 1)
|
|
ctx.sessionProjections.register({
|
|
key: 'cache-test/count',
|
|
stateSchema: z.number().int().nonnegative(),
|
|
init: () => 0,
|
|
apply,
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/count', number>)
|
|
const meta = headerOf(SessionId('cold-snap'), 9)
|
|
const events = Array.from({ length: 5 }, (_, seq) => ({
|
|
type: 'cache-test/mark', seq: SessionSeq(seq), time: seq, data: { marks: [`m${seq}`] },
|
|
})) as SessionEvent[]
|
|
const refreshed = whenWritten(ctx, meta.id)
|
|
const snapshot = cache.coldSnapshot(meta, SessionLogOffset(0), events)
|
|
// The full log was traversed, but the fold applied only seqs 3 and 4.
|
|
expect(apply).toHaveBeenCalledTimes(2)
|
|
expect(apply.mock.calls.map(call => call[1].seq)).toEqual([3, 4])
|
|
expect(snapshot.asOfSeq).toBe(4)
|
|
// Host-only unit: folded but not served; the refreshed row is written
|
|
// back (fail-soft, fire-and-forget) once the write lands.
|
|
expect(Object.keys(snapshot.values)).not.toContain('cache-test/count')
|
|
await refreshed
|
|
expect((await storedRows(root, meta.id))?.['cache-test/count']?.seq).toBe(4)
|
|
// No cached row yet: the first cold read folds from init over the full
|
|
// log and creates the cache row (the `?? {}` seed path).
|
|
const fresh = headerOf(SessionId('cold-fresh'), 10)
|
|
const created = whenWritten(ctx, fresh.id)
|
|
cache.coldSnapshot(fresh, SessionLogOffset(0), events)
|
|
expect(apply).toHaveBeenCalledTimes(7) // 2 tail + 5 full
|
|
await created
|
|
expect((await storedRows(root, fresh.id))?.['cache-test/count']?.seq).toBe(4)
|
|
})
|
|
|
|
it('coldSnapshot write-back is fail-soft: a failed durable write logs and never throws', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const ctx = new Context()
|
|
contexts.push(ctx)
|
|
await ctx.plugin(Storage)
|
|
await ctx.plugin({ name: storageJsonName, inject: storageJsonInject, apply: storageJsonApply, Config: storageJsonConfig }, { root })
|
|
await ctx.plugin({ name: storageDomainName, inject: storageDomainInject, apply: storageDomainApply, Config: storageDomainConfig }, { backend: 'json' })
|
|
await ctx.plugin(SessionStore)
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
ctx.sessionProjections.register(marksUnit())
|
|
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
|
// A directory where the record document must land makes the write-back
|
|
// fail; the cold read itself still succeeds and never throws.
|
|
const meta = headerOf(SessionId('cold-fail'))
|
|
await mkdir(recordPath(root, meta.id), { recursive: true })
|
|
expect(ctx.sessionProjectionCache.coldSnapshot(meta, SessionLogOffset(0), [])).toBeDefined()
|
|
// The failed write-back is fire-and-forget: poll for the warn instead of
|
|
// assuming a fixed settle window (slow runners exceed it).
|
|
await vi.waitFor(() => {
|
|
expect(warn).toHaveBeenCalledWith(expect.stringContaining('cold-read write-back for "cold-fail" failed'))
|
|
}, { timeout: 5_000 })
|
|
})
|
|
})
|
|
|
|
describe('SessionProjectionCache summary reads', () => {
|
|
it('serves the summary block from stored rows and excludes the marked unit', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'summary-watermark', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(5), val: { marks: ['primary'] } },
|
|
'cache-test/summary-excluded': { ver: 1, seq: SessionSeq(1), val: { n: 3 } },
|
|
})
|
|
const { ctx, cache } = await harness({ root })
|
|
ctx.sessionProjections.register(summaryExcludedUnit)
|
|
const summaryKeys = ctx.sessionProjections.summaryKeys()
|
|
|
|
// The summary watermark follows the served rows only: the marked row sits
|
|
// at seq 1, and its absence must not drag the summary block down.
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('summary-watermark')), summaryKeys)).toEqual({
|
|
asOfSeq: 5,
|
|
values: { 'cache-test/marks': { marks: ['primary'] } },
|
|
})
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('summary-watermark')))).toEqual({
|
|
asOfSeq: 1,
|
|
values: {
|
|
'cache-test/marks': { marks: ['primary'] },
|
|
'cache-test/summary-excluded': { n: 3 },
|
|
},
|
|
})
|
|
// No record for this id: a key-subset read reports absence like the complete face.
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('summary-absent')), summaryKeys)).toBeUndefined()
|
|
})
|
|
})
|