From 9226d9bbf6e5dfc216ede9cf69a9b33558afd98c Mon Sep 17 00:00:00 2001 From: _Kerman Date: Thu, 20 Aug 2026 14:38:33 +0800 Subject: [PATCH] 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: /session_projcache/sessions/.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. --- .../apiproxy/tests/api-proxy-cold.spec.ts | 2 +- .../tests/api-proxy-projections.spec.ts | 2 +- .../session-projection-cache/package.json | 4 +- .../tests/cache.spec.ts | 103 ++++++++++------ .../session-projection-cache/tsconfig.json | 13 +- .../storage-domain/tests/domain.spec.ts | 13 +- .../storage-json/tests/json-backend.spec.ts | 116 +++++++++++++++++- packages/subagent/subagent/package.json | 3 + .../subagent/tests/list-children.spec.ts | 28 +++-- pnpm-lock.yaml | 15 +++ 10 files changed, 246 insertions(+), 53 deletions(-) diff --git a/packages/host/apiproxy/tests/api-proxy-cold.spec.ts b/packages/host/apiproxy/tests/api-proxy-cold.spec.ts index f18d0ac0e2..ef642082ac 100644 --- a/packages/host/apiproxy/tests/api-proxy-cold.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-cold.spec.ts @@ -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 } } } } diff --git a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts index f57e1338ac..6d9c53a6f8 100644 --- a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts @@ -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), diff --git a/packages/session/session-projection-cache/package.json b/packages/session/session-projection-cache/package.json index 78e3adc706..08d3e3cac1 100644 --- a/packages/session/session-projection-cache/package.json +++ b/packages/session/session-projection-cache/package.json @@ -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:^" } } diff --git a/packages/session/session-projection-cache/tests/cache.spec.ts b/packages/session/session-projection-cache/tests/cache.spec.ts index d083425c4e..c91ad1d438 100644 --- a/packages/session/session-projection-cache/tests/cache.spec.ts +++ b/packages/session/session-projection-cache/tests/cache.spec.ts @@ -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 - * (`//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 + * `/session_projcache/sessions/.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 { 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 { - 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() }) }) diff --git a/packages/session/session-projection-cache/tsconfig.json b/packages/session/session-projection-cache/tsconfig.json index b4d2f25a6b..fd5c7ab456 100644 --- a/packages/session/session-projection-cache/tsconfig.json +++ b/packages/session/session-projection-cache/tsconfig.json @@ -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" } ] } diff --git a/packages/storage/storage-domain/tests/domain.spec.ts b/packages/storage/storage-domain/tests/domain.spec.ts index 5e1afbc863..3b3c8fa679 100644 --- a/packages/storage/storage-domain/tests/domain.spec.ts +++ b/packages/storage/storage-domain/tests/domain.spec.ts @@ -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', () => { diff --git a/packages/storage/storage-json/tests/json-backend.spec.ts b/packages/storage/storage-json/tests/json-backend.spec.ts index 60bb3a61db..4365a5e3e0 100644 --- a/packages/storage/storage-json/tests/json-backend.spec.ts +++ b/packages/storage/storage-json/tests/json-backend.spec.ts @@ -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) + }) +}) diff --git a/packages/subagent/subagent/package.json b/packages/subagent/subagent/package.json index 189984cfbf..34e8d05463 100644 --- a/packages/subagent/subagent/package.json +++ b/packages/subagent/subagent/package.json @@ -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:^", diff --git a/packages/subagent/subagent/tests/list-children.spec.ts b/packages/subagent/subagent/tests/list-children.spec.ts index e90ef052d9..3b9514cd23 100644 --- a/packages/subagent/subagent/tests/list-children.spec.ts +++ b/packages/subagent/subagent/tests/list-children.spec.ts @@ -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') diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 083f8d1faf..85aedad3bc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -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