mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-08-29 04:26:38 +00:00
test(session-projection-cache): rewrite for the per-record domain medium
cache.spec now boots the real storage stack (storage, storage-json,
storage-domain) and asserts the per-record medium directly:
<root>/session_projcache/sessions/<id>.json carries a version-stamped
{version, record} document, cachedSnapshot is synchronous (zero-I/O from
the domain's in-memory tables), and the write-policy / fail-soft / listing
coverage is preserved at 100%. json-backend.spec gains a per-record layout
block (per-record documents, overwrite/delete/reopen, unsafe keys and
undeclared tables rejecting, foreign-document discard on open, closed
guard, close drain, unreadable-as-absent); storage-domain domain.spec
covers layout validation and descriptorOf projection. list-children.spec
mounts the storage stack for its projectionCache cases and its
cachedSnapshot mocks and reads go synchronous; the api-proxy specs' cache
mocks go synchronous too. devDeps and tsconfig references updated for the
storage stack.
This commit is contained in:
_Kerman
@@ -91,7 +91,7 @@ describe('sessions.list cold merge', () => {
|
||||
readFrom,
|
||||
} as never)
|
||||
ctx.provide('sessionProjectionCache', {
|
||||
cachedSnapshot: async (meta: SessionHeader) => {
|
||||
cachedSnapshot: (meta: SessionHeader) => {
|
||||
if (meta.id === sid('small-blank')) {
|
||||
return { asOfSeq: 0, values: { sessionListMetadata: { blank: true, lastPromptAt: null } } }
|
||||
}
|
||||
|
||||
@@ -274,7 +274,7 @@ describe('session.list projections column', () => {
|
||||
} as never)
|
||||
ctx.provide('sessionProjectionCache', {
|
||||
// The carrier hands the listed header through as the identity witness.
|
||||
cachedSnapshot: async (meta: { id: unknown; createdAt: number }) =>
|
||||
cachedSnapshot: (meta: { id: unknown; createdAt: number }) =>
|
||||
(meta.id === coldId && meta.createdAt === 5
|
||||
? { asOfSeq: 7, values: { 'test/last-user': { text: 'cached' } } }
|
||||
: undefined),
|
||||
|
||||
@@ -43,10 +43,12 @@
|
||||
"@deepseek-ai/dsh-storage-domain": "workspace:^"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/cordis": "workspace:^",
|
||||
"@deepseek-ai/dsh-invariants": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-projection": "workspace:^",
|
||||
"@deepseek-ai/dsh-storage": "workspace:^",
|
||||
"@deepseek-ai/dsh-storage-domain": "workspace:^",
|
||||
"@deepseek-ai/cordis": "workspace:^"
|
||||
"@deepseek-ai/dsh-storage-json": "workspace:^"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,24 +2,33 @@
|
||||
* 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
|
||||
* cached listing read. The durable medium is one `projection_cache.json`
|
||||
* per session under the cache's own configured root
|
||||
* (`<root>/<session-id>/projection_cache.json`); the cache never consults
|
||||
* the persistence layer.
|
||||
* 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 { mkdtemp, mkdir, readFile, rm, writeFile } from 'node:fs/promises'
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { dirname, join } from 'node:path'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import { z } from 'zod'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
||||
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
|
||||
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 } from '../src/spec.ts'
|
||||
import { checkpointRecord, projectionCacheDomainSpec } from '../src/spec.ts'
|
||||
import type { CheckpointRecord } from '../src/spec.ts'
|
||||
|
||||
declare module '@deepseek-ai/dsh-session-projection/types' {
|
||||
@@ -55,9 +64,9 @@ const marksUnit = (stateVersion = 1) => ({
|
||||
stateVersion,
|
||||
}) satisfies ProjectionDefinition<'cache-test/marks', MarksState>
|
||||
|
||||
/** One session's cache file under the cache's own root. */
|
||||
const cachePath = (root: string, id: Session['id']): string =>
|
||||
join(root, String(id), 'projection_cache.json')
|
||||
/** 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) =>
|
||||
@@ -77,13 +86,15 @@ async function harness(options: HarnessOptions = {}) {
|
||||
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, {
|
||||
root,
|
||||
...options.config ?? { writeEveryEvents: 100, writeIntervalMs: 60_000 },
|
||||
})
|
||||
const fiber = await ctx.plugin(SessionProjectionCache, options.config ?? { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
||||
return { ctx, root, fiber, cache: ctx.sessionProjectionCache }
|
||||
}
|
||||
|
||||
@@ -96,7 +107,8 @@ const endTurn = (session: Session): SessionEvent =>
|
||||
/** The stored record for one session id (undefined = absent or unreadable). */
|
||||
async function storedRecord(root: string, id: Session['id']): Promise<CheckpointRecord | undefined> {
|
||||
try {
|
||||
return checkpointRecord.parse(JSON.parse(await readFile(cachePath(root, id), 'utf8')))
|
||||
const document = JSON.parse(await readFile(recordPath(root, id), 'utf8')) as { record: unknown }
|
||||
return checkpointRecord.parse(document.record)
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
@@ -107,15 +119,16 @@ async function storedRows(root: string, id: Session['id']): Promise<CheckpointRe
|
||||
return (await storedRecord(root, id))?.rows
|
||||
}
|
||||
|
||||
/** Pre-seed one session's cache file with a stored checkpoint record. */
|
||||
/** 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'] = { createdAt: 0 },
|
||||
): Promise<void> {
|
||||
await mkdir(join(root, id), { recursive: true })
|
||||
await writeFile(cachePath(root, SessionId(id)), JSON.stringify({ identity, rows }))
|
||||
const path = recordPath(root, SessionId(id))
|
||||
await mkdir(dirname(path), { recursive: true })
|
||||
await writeFile(path, JSON.stringify({ version: projectionCacheDomainSpec.version, record: { identity, rows } }))
|
||||
}
|
||||
|
||||
/** Wait until queued fail-soft writes (event-listener fire-and-forget over real fs I/O) drain. */
|
||||
@@ -212,22 +225,25 @@ describe('SessionProjectionCache write policy', () => {
|
||||
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, { root, writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
||||
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
const session = ctx.sessions.create(SessionId('fail-soft'))
|
||||
// A directory where the cache file must land makes the atomic rename
|
||||
// fail on the first write...
|
||||
await mkdir(cachePath(root, session.id), { recursive: true })
|
||||
// A directory where the record document must land makes the atomic
|
||||
// rename fail on the first write...
|
||||
await mkdir(recordPath(root, session.id), { recursive: true })
|
||||
mark(session, ['x'])
|
||||
endTurn(session)
|
||||
await settle()
|
||||
expect(await storedRows(root, session.id)).toBeUndefined()
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('turn/end write for "fail-soft" failed'))
|
||||
// Self-heal: once the blocker clears, the next mandatory point writes.
|
||||
await rm(cachePath(root, session.id), { recursive: true })
|
||||
await rm(recordPath(root, session.id), { recursive: true })
|
||||
mark(session, ['y'])
|
||||
endTurn(session)
|
||||
await settle()
|
||||
@@ -243,19 +259,35 @@ describe('SessionProjectionCache listing read', () => {
|
||||
const { cache } = await harness({ root })
|
||||
const id = SessionId('listed')
|
||||
// Matching header: values plus the watermark the client seeds under.
|
||||
expect(await cache.cachedSnapshot(headerOf(id))).toEqual({ asOfSeq: 4, values: { 'cache-test/marks': { marks: ['t'] } } })
|
||||
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(await cache.cachedSnapshot(headerOf(id, 777))).toBeUndefined()
|
||||
expect(cache.cachedSnapshot(headerOf(id, 777))).toBeUndefined()
|
||||
// Unknown id: no block.
|
||||
expect(await cache.cachedSnapshot(headerOf(SessionId('never-cached')))).toBeUndefined()
|
||||
expect(cache.cachedSnapshot(headerOf(SessionId('never-cached')))).toBeUndefined()
|
||||
})
|
||||
|
||||
it('returns undefined when the stored record is version-mismatched', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
||||
roots.push(root)
|
||||
// A stale version-stamped document is discarded at open: absent record.
|
||||
const path = recordPath(root, SessionId('all-stale'))
|
||||
await mkdir(dirname(path), { recursive: true })
|
||||
await writeFile(path, JSON.stringify({
|
||||
version: projectionCacheDomainSpec.version + 1,
|
||||
record: { identity: { createdAt: 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('returns undefined when every stored row is version-mismatched', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
||||
roots.push(root)
|
||||
await seedRecord(root, 'all-stale', { 'cache-test/marks': { ver: 99, seq: 4, val: { marks: ['old'] } } })
|
||||
// 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: 4, val: { marks: ['old'] } } })
|
||||
const { cache } = await harness({ root })
|
||||
expect(await cache.cachedSnapshot(headerOf(SessionId('all-stale')))).toBeUndefined()
|
||||
expect(cache.cachedSnapshot(headerOf(SessionId('row-stale')))).toBeUndefined()
|
||||
})
|
||||
|
||||
it('binds identity on cwd too: a matching cwd serves, a moved session does not', async () => {
|
||||
@@ -264,17 +296,18 @@ describe('SessionProjectionCache listing read', () => {
|
||||
await seedRecord(root, 'homed', { 'cache-test/marks': { ver: 1, seq: 2, val: { marks: ['w'] } } }, { createdAt: 0, cwd: '/work' })
|
||||
const { cache } = await harness({ root })
|
||||
const id = SessionId('homed')
|
||||
expect((await cache.cachedSnapshot(headerOf(id, 0, '/work')))?.values['cache-test/marks']).toEqual({ marks: ['w'] })
|
||||
expect(await cache.cachedSnapshot(headerOf(id, 0, '/elsewhere'))).toBeUndefined()
|
||||
expect(await cache.cachedSnapshot(headerOf(id, 0))).toBeUndefined()
|
||||
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 cache file (refold from the log on the caller side)', async () => {
|
||||
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)
|
||||
await mkdir(join(root, 'malformed'), { recursive: true })
|
||||
await writeFile(cachePath(root, SessionId('malformed')), 'not json at all')
|
||||
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(await cache.cachedSnapshot(headerOf(SessionId('malformed')))).toBeUndefined()
|
||||
expect(cache.cachedSnapshot(headerOf(SessionId('malformed')))).toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -9,10 +9,10 @@
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/schemastery"
|
||||
@@ -20,9 +20,6 @@
|
||||
{
|
||||
"path": "../../core/session"
|
||||
},
|
||||
{
|
||||
"path": "../session-projection"
|
||||
},
|
||||
{
|
||||
"path": "../../runtime-diagnostics/invariants"
|
||||
},
|
||||
@@ -31,6 +28,12 @@
|
||||
},
|
||||
{
|
||||
"path": "../../storage/storage-domain"
|
||||
},
|
||||
{
|
||||
"path": "../../storage/storage-json"
|
||||
},
|
||||
{
|
||||
"path": "../session-projection"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@ import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import { z } from 'zod'
|
||||
import Storage, { storageBackendServiceKey } from '@deepseek-ai/dsh-storage'
|
||||
import { apply, DomainFacility, defineDomain, domainTable } from '../src/index.ts'
|
||||
import { apply, defineDomain, descriptorOf, DomainFacility, domainTable } from '../src/index.ts'
|
||||
import type { Config } from '../src/index.ts'
|
||||
import type { DomainChanged } from '../src/events.ts'
|
||||
import { MemoryMediaPool, MemoryStorageBackend } from './helpers/memory-backend.ts'
|
||||
@@ -57,6 +57,17 @@ describe('defineDomain', () => {
|
||||
tables: {},
|
||||
})).toThrow(/must not accept null/)
|
||||
})
|
||||
|
||||
it('rejects an invalid layout and projects the declared one onto the descriptor', () => {
|
||||
// A spec built from config can carry any value; the union type is
|
||||
// compile-time only, so the runtime boundary check must reject it.
|
||||
expect(() => defineDomain({ name: 'ok', version: 1, layout: 'every-record' as 'single', tables: {} }))
|
||||
.toThrow(/layout/)
|
||||
expect(descriptorOf(defineDomain({ name: 'per', version: 1, layout: 'per-record', tables: {} })))
|
||||
.toMatchObject({ name: 'per', layout: 'per-record' })
|
||||
// The default (single) layout is absent from the descriptor.
|
||||
expect(descriptorOf(spec)).not.toHaveProperty('layout')
|
||||
})
|
||||
})
|
||||
|
||||
describe('DomainFacility.open', () => {
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { mkdir, mkdtemp, readFile, rename, rm, writeFile } from 'node:fs/promises'
|
||||
import { chmod, mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterAll, describe, expect, it } from 'vitest'
|
||||
@@ -226,3 +226,117 @@ describe('json backend specifics', () => {
|
||||
await closing
|
||||
})
|
||||
})
|
||||
|
||||
describe('per-record layout', () => {
|
||||
const descriptor = { name: 'recs', version: 2, layout: 'per-record' as const, tables: ['t'], hasGlobal: true }
|
||||
const recordPath = (root: string, key: string): string => join(root, 'recs', 't', `${key}.json`)
|
||||
|
||||
it('stores one version-stamped document per record and defers materialization', async () => {
|
||||
const root = await freshRoot()
|
||||
const backend = new JsonStorageBackend(root)
|
||||
const unit = await backend.kv.open(descriptor)
|
||||
// Missing directory = empty unit; nothing materialized on the medium yet.
|
||||
expect(await unit.loadAll()).toEqual({ tables: { t: {} }, global: null })
|
||||
await unit.putRecord('t', 'k1', { v: 1 })
|
||||
await unit.putRecord('t', 'k2', { v: 2 })
|
||||
await unit.setGlobal('G')
|
||||
expect(await readFile(recordPath(root, 'k1'), 'utf8'))
|
||||
.toBe(`${JSON.stringify({ version: 2, record: { v: 1 } }, null, 2)}\n`)
|
||||
expect((await readdir(join(root, 'recs', 't'))).sort()).toEqual(['k1.json', 'k2.json'])
|
||||
expect(JSON.parse(await readFile(join(root, 'recs', 'global.json'), 'utf8')))
|
||||
.toEqual({ version: 2, record: 'G' })
|
||||
expect(await unit.loadAll()).toEqual({ tables: { t: { k1: { v: 1 }, k2: { v: 2 } } }, global: 'G' })
|
||||
await backend.close()
|
||||
})
|
||||
|
||||
it('overwrites and deletes one document at a time and persists across reopen', async () => {
|
||||
const root = await freshRoot()
|
||||
const backend = new JsonStorageBackend(root)
|
||||
const unit = await backend.kv.open(descriptor)
|
||||
await unit.putRecord('t', 'k', { v: 1 })
|
||||
await unit.putRecord('t', 'k', { v: 2 }) // overwrite the same document
|
||||
await unit.deleteRecord('t', 'missing') // idempotent no-op
|
||||
await unit.close()
|
||||
const unit2 = await backend.kv.open(descriptor)
|
||||
expect(await unit2.loadAll()).toEqual({ tables: { t: { k: { v: 2 } } }, global: null })
|
||||
await unit2.deleteRecord('t', 'k')
|
||||
expect(await unit2.loadAll()).toEqual({ tables: { t: {} }, global: null })
|
||||
await backend.close()
|
||||
})
|
||||
|
||||
it('rejects unsafe keys and undeclared tables, and enforces the closed guard', async () => {
|
||||
const root = await freshRoot()
|
||||
const backend = new JsonStorageBackend(root)
|
||||
const unit = await backend.kv.open(descriptor)
|
||||
await expect(unit.putRecord('t', 'a/b', {})).rejects.toThrow(/not path-safe/)
|
||||
await expect(unit.deleteRecord('t', '..')).rejects.toThrow(/not path-safe/)
|
||||
await expect(unit.putRecord('bogus', 'k', {})).rejects.toThrow(/does not declare table/)
|
||||
await unit.close()
|
||||
await expect(unit.putRecord('t', 'k', {})).rejects.toMatchObject({ code: 'closed' })
|
||||
await expect(unit.deleteRecord('t', 'k')).rejects.toMatchObject({ code: 'closed' })
|
||||
await expect(unit.setGlobal('x')).rejects.toMatchObject({ code: 'closed' })
|
||||
await expect(unit.loadAll()).rejects.toMatchObject({ code: 'closed' })
|
||||
await backend.close()
|
||||
})
|
||||
|
||||
it('discards foreign documents (stale version, malformed, non-object, unsafe key) on open', async () => {
|
||||
const root = await freshRoot()
|
||||
const backend = new JsonStorageBackend(root)
|
||||
const unit = await backend.kv.open(descriptor)
|
||||
await unit.putRecord('t', 'good', { v: 1 })
|
||||
await unit.close()
|
||||
await writeFile(recordPath(root, 'stale'), JSON.stringify({ version: 1, record: { v: 0 } }), 'utf8')
|
||||
await writeFile(recordPath(root, 'broken'), '{oops', 'utf8')
|
||||
await writeFile(recordPath(root, 'scalar'), JSON.stringify(5), 'utf8')
|
||||
await writeFile(recordPath(root, 'unsafe%2Fkey'), JSON.stringify({ version: 2, record: { v: 0 } }), 'utf8')
|
||||
await writeFile(join(root, 'recs', 't', 'not-json.txt'), 'ignored', 'utf8')
|
||||
await writeFile(join(root, 'recs', 'global.json'), JSON.stringify({ version: 1, record: 'old' }), 'utf8')
|
||||
// Stray unit-root entries: an undeclared directory and a non-document file.
|
||||
await mkdir(join(root, 'recs', 'stray-dir'), { recursive: true })
|
||||
await writeFile(join(root, 'recs', 'stray.txt'), 'ignored', 'utf8')
|
||||
const unit2 = await backend.kv.open(descriptor)
|
||||
expect(await unit2.loadAll()).toEqual({ tables: { t: { good: { v: 1 } } }, global: null })
|
||||
await backend.close()
|
||||
})
|
||||
|
||||
it('propagates non-ENOENT read failures and refuses a global slot that is not declared', async () => {
|
||||
const root = await freshRoot()
|
||||
const backend = new JsonStorageBackend(root)
|
||||
// A file where the unit directory should be: the lazy loadAll readdir
|
||||
// fails with ENOTDIR (opening itself touches nothing on the medium).
|
||||
await writeFile(join(root, 'recs'), 'not a directory', 'utf8')
|
||||
const unit = await backend.kv.open(descriptor)
|
||||
await expect(unit.loadAll()).rejects.toMatchObject({ code: 'ENOTDIR' })
|
||||
await unit.close()
|
||||
const noGlobal = { name: 'plain', version: 1, layout: 'per-record' as const, tables: ['t'], hasGlobal: false }
|
||||
const unit2 = await backend.kv.open(noGlobal)
|
||||
await expect(unit2.setGlobal('x')).rejects.toThrow(/does not declare a global slot/)
|
||||
await backend.close()
|
||||
})
|
||||
|
||||
it('close drains in-flight writes and an unreadable record document reads as absent', async () => {
|
||||
const root = await freshRoot()
|
||||
const backend = new JsonStorageBackend(root)
|
||||
const unit = await backend.kv.open(descriptor)
|
||||
const big = unit.putRecord('t', 'big', { blob: 'x'.repeat(4 * 1024 * 1024) })
|
||||
await unit.close()
|
||||
await unit.close() // idempotent
|
||||
await expect(big).resolves.toBeUndefined()
|
||||
const onDisk = JSON.parse(await readFile(recordPath(root, 'big'), 'utf8')) as { record: { blob: string } }
|
||||
expect(onDisk.record).toEqual({ blob: 'x'.repeat(4 * 1024 * 1024) })
|
||||
await backend.close()
|
||||
})
|
||||
|
||||
it('reads an unreadable record document as absent (per-record contract)', async () => {
|
||||
const root = await freshRoot()
|
||||
const path = recordPath(root, 'locked')
|
||||
await mkdir(join(root, 'recs', 't'), { recursive: true })
|
||||
await writeFile(path, JSON.stringify({ version: 2, record: { v: 1 } }), 'utf8')
|
||||
await chmod(path, 0o000)
|
||||
const backend = new JsonStorageBackend(root)
|
||||
const unit = await backend.kv.open(descriptor)
|
||||
expect(await unit.loadAll()).toEqual({ tables: { t: {} }, global: null })
|
||||
await backend.close()
|
||||
await chmod(path, 0o600)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -96,6 +96,9 @@
|
||||
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-projection": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-projection-cache": "workspace:^",
|
||||
"@deepseek-ai/dsh-storage": "workspace:^",
|
||||
"@deepseek-ai/dsh-storage-domain": "workspace:^",
|
||||
"@deepseek-ai/dsh-storage-json": "workspace:^",
|
||||
"@deepseek-ai/dsh-jobs": "workspace:^",
|
||||
"@deepseek-ai/dsh-tools": "workspace:^",
|
||||
"@deepseek-ai/dsh-user-approval": "workspace:^",
|
||||
|
||||
@@ -13,6 +13,13 @@ import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
||||
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
|
||||
import SessionProjectionCache from '@deepseek-ai/dsh-session-projection-cache'
|
||||
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 SubagentRuntime, {
|
||||
SUBAGENT_DESCRIPTOR_VERSION,
|
||||
SubagentError,
|
||||
@@ -46,7 +53,12 @@ async function setup(
|
||||
if (options.projectionCache === true) {
|
||||
const root = mkdtempSync(join(tmpdir(), 'dsh-subagent-projcache-'))
|
||||
projCacheRoots.push(root)
|
||||
await ctx.plugin(SessionProjectionCache, { root, writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
||||
// The cache opens its domain through the storage stack; the json backend
|
||||
// lands it 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(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
||||
}
|
||||
await ctx.plugin(SubagentRuntime)
|
||||
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
|
||||
@@ -421,7 +433,7 @@ describe('SubagentRuntime.listChildren', () => {
|
||||
// seq 2 >= seedLength 0: the cached identity provably comes from the
|
||||
// child's own suffix, so it is final and the log is never re-read — the
|
||||
// divergent label proves the row, not the log, produced the entry.
|
||||
ctx.sessionProjectionCache.cachedSnapshot = async () => ({
|
||||
ctx.sessionProjectionCache.cachedSnapshot = () => ({
|
||||
asOfSeq: 2,
|
||||
values: { subagent: { mode: 'continuable', label: 'cached own', seq: 2 } },
|
||||
})
|
||||
@@ -451,7 +463,7 @@ describe('SubagentRuntime.listChildren', () => {
|
||||
}, events)
|
||||
// A creation-window checkpoint carried the ANCESTOR identity: its seq 2
|
||||
// fails the own-suffix gate (< seedLength 4), so preparation rules.
|
||||
ctx.sessionProjectionCache.cachedSnapshot = async () => ({
|
||||
ctx.sessionProjectionCache.cachedSnapshot = () => ({
|
||||
asOfSeq: 2,
|
||||
values: { subagent: { mode: 'continuable', label: 'ancestor label', seq: 2 } },
|
||||
})
|
||||
@@ -500,7 +512,7 @@ describe('SubagentRuntime.listChildren', () => {
|
||||
origin: 'subagent',
|
||||
}, childEvents(descriptorPayload('actually valid')))
|
||||
// A stale cached sentinel must not out-rank the authoritative re-fold.
|
||||
ctx.sessionProjectionCache.cachedSnapshot = async () => ({ asOfSeq: 0, values: { subagent: null } })
|
||||
ctx.sessionProjectionCache.cachedSnapshot = () => ({ asOfSeq: 0, values: { subagent: null } })
|
||||
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
|
||||
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([{
|
||||
kind: 'child', id: healthy, label: 'actually valid', mode: 'continuable',
|
||||
@@ -772,8 +784,8 @@ describe('SubagentRuntime.listChildren', () => {
|
||||
// The child's turn/end and disposal are the cache's mandatory checkpoint
|
||||
// points; both writes are fail-soft asynchronous, so wait for the row.
|
||||
const header = (await ctx.sessionPersistence.list()).find(meta => meta.id === childId)
|
||||
await vi.waitFor(async () => {
|
||||
expect((await ctx.sessionProjectionCache.cachedSnapshot(header!))?.values.subagent).toBeDefined()
|
||||
await vi.waitFor(() => {
|
||||
expect(ctx.sessionProjectionCache.cachedSnapshot(header!)?.values.subagent).toBeDefined()
|
||||
}, { timeout: 5_000 })
|
||||
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
|
||||
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([{
|
||||
@@ -799,7 +811,7 @@ describe('SubagentRuntime.listChildren', () => {
|
||||
expect(inspect).toHaveBeenCalledTimes(1)
|
||||
// A stored row whose cut predates the descriptor: the subagent key is
|
||||
// absent from the served values, and preparation still rules.
|
||||
ctx.sessionProjectionCache.cachedSnapshot = async () => ({ asOfSeq: 0, values: {} })
|
||||
ctx.sessionProjectionCache.cachedSnapshot = () => ({ asOfSeq: 0, values: {} })
|
||||
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual(expected)
|
||||
expect(inspect).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
@@ -825,7 +837,7 @@ describe('SubagentRuntime.listChildren', () => {
|
||||
parentSession: parent.id,
|
||||
origin: 'subagent',
|
||||
}, childEvents(descriptorPayload('recovered child')))
|
||||
ctx.sessionProjectionCache.cachedSnapshot = async () => {
|
||||
ctx.sessionProjectionCache.cachedSnapshot = () => {
|
||||
// A poisoned stored row (any unit's) detonates at view time; the cache
|
||||
// is derived data, so its failure must not become a verdict.
|
||||
throw new Error('poisoned cache row')
|
||||
|
||||
Generated
+15
@@ -6236,9 +6236,15 @@ importers:
|
||||
'@deepseek-ai/dsh-session-projection':
|
||||
specifier: workspace:^
|
||||
version: link:../session-projection
|
||||
'@deepseek-ai/dsh-storage':
|
||||
specifier: workspace:^
|
||||
version: link:../../storage/storage
|
||||
'@deepseek-ai/dsh-storage-domain':
|
||||
specifier: workspace:^
|
||||
version: link:../../storage/storage-domain
|
||||
'@deepseek-ai/dsh-storage-json':
|
||||
specifier: workspace:^
|
||||
version: link:../../storage/storage-json
|
||||
|
||||
packages/session/session-stats:
|
||||
dependencies:
|
||||
@@ -7111,6 +7117,15 @@ importers:
|
||||
'@deepseek-ai/dsh-session-projection-cache':
|
||||
specifier: workspace:^
|
||||
version: link:../../session/session-projection-cache
|
||||
'@deepseek-ai/dsh-storage':
|
||||
specifier: workspace:^
|
||||
version: link:../../storage/storage
|
||||
'@deepseek-ai/dsh-storage-domain':
|
||||
specifier: workspace:^
|
||||
version: link:../../storage/storage-domain
|
||||
'@deepseek-ai/dsh-storage-json':
|
||||
specifier: workspace:^
|
||||
version: link:../../storage/storage-json
|
||||
'@deepseek-ai/dsh-tools':
|
||||
specifier: workspace:^
|
||||
version: link:../../core/tools
|
||||
|
||||
Reference in New Issue
Block a user