feat(storage): json backend — one human-readable file per unit

Atomic whole-file replacement (same-dir temp + fsync + rename + parent fsync); the in-memory unit state is authoritative and the file is always the current net state, pretty-printed. Missing files open as empty units and materialize on first write; foreign or unparsable files reject with malformed-medium, stored-version drift with version-mismatch.
2026-07-24 19:07:07 +08:00
parent e90b0d51df
commit 1529be6fd4
9 changed files with 543 additions and 0 deletions
@@ -0,0 +1,24 @@
+# @deepseek-ai/dsh-storage-json
+
+JSON backend for the [storage hub](../storage/README.md): one human-readable `<unit>.json` file per unit under a configured root, registered as backend `json`. Design: [domain KV storage Agent Note](../../../.agents/notes/proposed/architecture/2026-07-24-domain-kv-storage-and-workspace.zh.md).
+
+## Model
+
+- The in-memory unit state is authoritative; every write primitive republishes the whole file via temp-write + fsync + atomic `rename()` replace. A unit file is always the complete current net state — legibility is this backend's reason to exist; scale is the SQLite backend's job.
+- A missing file opens as an empty unit and materializes on the first write. A foreign or unparsable file rejects with `malformed-medium`; a stored version differing from the descriptor rejects with `version-mismatch` (no migration, pre-release stance).
+- Write ordering across calls belongs to the caller (the domain layer's write chain); each single call is atomic and durable once resolved.
+
+## Config
+
+| Key | Type | Default | Meaning |
+| --- | --- | --- | --- |
+| `root` | string | required — no default (a cwd fallback would scatter files) | Directory holding unit files; created `0o700` on demand |
+
+## Model Experience
+
+No model-visible surface: this package serves host-side persistence only; nothing it does reaches prompts, tool schemas, or token budgets.
+
+## Known Limitations and Deferred Work
+
+- Windows durability relies on libuv's `rename()` (`MoveFileExW` with replacement) without an explicit write-through flag; the session-log backend's stricter Win32 write-through publish helper is planned to move down here when the append-log facet lands (see the Agent Note's migration section).
+- No cross-process write locking: two processes writing the same root can interleave whole-file replacements (last write wins). Single-host-process deployments are the current consumer; the multi-process story is deferred per the Agent Note's out-of-scope table.
@@ -0,0 +1,42 @@
+{
+  "name": "@deepseek-ai/dsh-storage-json",
+  "description": "JSON file KV storage backend for the DeepSeek Harness storage hub",
+  "version": "0.0.1",
+  "private": true,
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/types/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/types/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./invariant": {
+      "types": "./lib/types/invariant.d.ts",
+      "default": "./lib/invariant.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "lib/index.js",
+    "lib/invariant.js",
+    "lib/types/**/*.d.ts",
+    "lib/types/**/*.d.ts.map",
+    "src"
+  ],
+  "license": "BSD-3-Clause",
+  "peerDependencies": {
+    "@deepseek-ai/dsh-invariants": "^0.0.1",
+    "@deepseek-ai/dsh-storage": "^0.0.1",
+    "cordis": "^4.0.0-rc.7"
+  },
+  "dependencies": {
+    "schemastery": "^3.18.0"
+  },
+  "devDependencies": {
+    "@deepseek-ai/dsh-invariants": "workspace:^",
+    "@deepseek-ai/dsh-storage": "workspace:^",
+    "cordis": "^4.0.0-rc.7"
+  }
+}
@@ -0,0 +1,53 @@
+/**
+ * Atomic whole-file replacement for the JSON backend.
+ *
+ * Publish protocol: write a same-directory temp file, fsync it, then
+ * `rename()` over the target. Rename is an atomic replace on POSIX and on
+ * Windows (libuv maps it to `MoveFileExW(..., MOVEFILE_REPLACE_EXISTING)`),
+ * and replacement is the intended semantic here — unlike the session-log
+ * backend's link()+unlink() no-clobber protocol, a unit file has exactly one
+ * writer per process and last-write-wins is correct. After the rename the
+ * parent directory is fsynced on POSIX so the new entry is crash-durable.
+ * @module @deepseek-ai/dsh-storage-json/src/atomic
+ */
+
+import { open, rename, rm } from 'node:fs/promises'
+import { dirname, join } from 'node:path'
+import { randomUUID } from 'node:crypto'
+
+/**
+ * Durably replace `path` with `data`.
+ * @param path - Absolute target file path.
+ * @param data - Full new file content.
+ * @returns resolution after the replacement is crash-durable.
+ */
+export async function writeAtomic(path: string, data: string): Promise<void> {
+  const tmp = join(dirname(path), `.${randomUUID()}.tmp`)
+  try {
+    const handle = await open(tmp, 'wx', 0o600)
+    try {
+      await handle.writeFile(data, 'utf8')
+      await handle.sync()
+    } finally {
+      await handle.close()
+    }
+    await rename(tmp, path)
+    await fsyncDirectory(dirname(path))
+  } catch (error) {
+    await rm(tmp, { force: true })
+    throw error
+  }
+}
+
+/** fsync a POSIX directory so a just-renamed entry is crash-durable. */
+/* v8 ignore start -- Windows rejects O_RDONLY directory opens; POSIX coverage exercises this. */
+async function fsyncDirectory(path: string): Promise<void> {
+  if (process.platform === 'win32') return
+  const handle = await open(path, 'r')
+  try {
+    await handle.sync()
+  } finally {
+    await handle.close()
+  }
+}
+/* v8 ignore stop */
@@ -0,0 +1,84 @@
+/**
+ * On-disk JSON unit format: the file is always the current net state, kept
+ * human-readable (pretty-printed, stable key order from insertion) — that
+ * legibility is this backend's reason to exist.
+ * @module @deepseek-ai/dsh-storage-json/src/format
+ */
+
+import { StorageError } from '@deepseek-ai/dsh-storage'
+import type { KvUnitDescriptor } from '@deepseek-ai/dsh-storage'
+
+/** In-memory authoritative state of one unit; the file is its projection. */
+export interface UnitState {
+  version: number
+  global: unknown | null
+  tables: Map<string, Map<string, unknown>>
+}
+
+/**
+ * Serialize a unit state to file content.
+ * @param name - Unit name, stamped into the header.
+ * @param state - Authoritative in-memory state.
+ * @returns pretty-printed JSON document with a trailing newline.
+ */
+export function serialize(name: string, state: UnitState): string {
+  const tables: Record<string, Record<string, unknown>> = {}
+  for (const [table, records] of state.tables) {
+    tables[table] = Object.fromEntries(records)
+  }
+  const document = {
+    unit: { name, version: state.version },
+    global: state.global,
+    tables,
+  }
+  return `${JSON.stringify(document, null, 2)}\n`
+}
+
+/**
+ * Parse file content into unit state, validating shape and version.
+ * @param text - Raw file content.
+ * @param descriptor - Expected identity; version mismatch rejects.
+ * @returns the parsed state.
+ */
+export function parse(text: string, descriptor: KvUnitDescriptor): UnitState {
+  let document: unknown
+  try {
+    document = JSON.parse(text)
+  } catch (error) {
+    throw new StorageError('malformed-medium', `unit '${descriptor.name}': file is not valid JSON`, { cause: error })
+  }
+  if (typeof document !== 'object' || document === null) {
+    throw new StorageError('malformed-medium', `unit '${descriptor.name}': file is not a JSON object`)
+  }
+  const { unit, global: globalValue, tables } = document as Record<string, unknown>
+  if (
+    typeof unit !== 'object' || unit === null ||
+    (unit as Record<string, unknown>)['name'] !== descriptor.name ||
+    typeof (unit as Record<string, unknown>)['version'] !== 'number'
+  ) {
+    throw new StorageError('malformed-medium', `unit '${descriptor.name}': missing or foreign unit header`)
+  }
+  const version = (unit as Record<string, unknown>)['version'] as number
+  if (version !== descriptor.version) {
+    throw new StorageError(
+      'version-mismatch',
+      `unit '${descriptor.name}': stored version ${version} != expected ${descriptor.version}`,
+    )
+  }
+  if (typeof tables !== 'object' || tables === null) {
+    throw new StorageError('malformed-medium', `unit '${descriptor.name}': tables is not an object`)
+  }
+  const state: UnitState = { version, global: globalValue ?? null, tables: new Map() }
+  for (const table of descriptor.tables) {
+    const records = (tables as Record<string, unknown>)[table]
+    if (records === undefined) {
+      state.tables.set(table, new Map())
+      continue
+    }
+    if (typeof records !== 'object' || records === null || Array.isArray(records)) {
+      throw new StorageError('malformed-medium', `unit '${descriptor.name}': table '${table}' is not an object`)
+    }
+    state.tables.set(table, new Map(Object.entries(records as Record<string, unknown>)))
+  }
+  return state
+}
@@ -0,0 +1,96 @@
+/**
+ * JSON storage backend: one human-readable file per unit under a configured
+ * root, published by atomic whole-file rewrite. Registers as backend `json`
+ * on the storage hub.
+ * @module @deepseek-ai/dsh-storage-json
+ */
+
+import { mkdir } from 'node:fs/promises'
+import { join } from 'node:path'
+import type { Context } from 'cordis'
+import z from 'schemastery'
+import { StorageError, UNIT_NAME_RE } from '@deepseek-ai/dsh-storage'
+import type { KvFacet, KvUnit, KvUnitDescriptor, StorageBackend } from '@deepseek-ai/dsh-storage'
+import { openJsonUnit } from './unit.ts'
+
+/** Cordis plugin name. */
+export const name = 'storage-json'
+/** The hub must exist before the backend can register. */
+export const inject = ['storage']
+
+/**
+ * Plugin configuration.
+ * `root` has NO default on purpose: a `process.cwd()` fallback would scatter
+ * unit files wherever the process happens to start; assemblies state the
+ * location explicitly.
+ */
+export interface Config {
+  /** Directory holding one `<unit>.json` file per unit. */
+  root: string
+}
+
+/** Config schema. */
+export const Config: z<Config> = z.object({
+  root: z.string().required(),
+})
+
+/** JSON backend: owns the file-tree root and serves the `kv` facet. */
+export class JsonStorageBackend implements StorageBackend {
+  private readonly open = new Map<string, KvUnit>()
+  private closed = false
+
+  constructor(private readonly root: string) {}
+
+  readonly kv: KvFacet = {
+    open: async (descriptor: KvUnitDescriptor): Promise<KvUnit> => {
+      if (this.closed) throw new StorageError('closed', 'json backend is closed')
+      validateDescriptor(descriptor)
+      if (this.open.has(descriptor.name)) {
+        throw new StorageError(
+          'malformed-medium',
+          `unit '${descriptor.name}' is already open; a unit has exactly one live handle`,
+        )
+      }
+      await mkdir(this.root, { recursive: true, mode: 0o700 })
+      const path = join(this.root, `${descriptor.name}.json`)
+      const unit = await openJsonUnit(descriptor, path, () => this.open.delete(descriptor.name))
+      this.open.set(descriptor.name, unit)
+      return unit
+    },
+  }
+
+  async close(): Promise<void> {
+    if (this.closed) return
+    this.closed = true
+    for (const unit of [...this.open.values()]) {
+      await unit.close()
+    }
+  }
+}
+
+function validateDescriptor(descriptor: KvUnitDescriptor): void {
+  if (!UNIT_NAME_RE.test(descriptor.name)) {
+    throw new StorageError('malformed-medium', `invalid unit name '${descriptor.name}'`)
+  }
+  for (const table of descriptor.tables) {
+    if (!UNIT_NAME_RE.test(table)) {
+      throw new StorageError('malformed-medium', `invalid table name '${table}' in unit '${descriptor.name}'`)
+    }
+  }
+}
+
+/**
+ * Register the `json` backend on the storage hub.
+ * @param ctx - Plugin context.
+ * @param config - Validated configuration.
+ */
+export function apply(ctx: Context, config: Config) {
+  const backend = new JsonStorageBackend(config.root)
+  ctx.effect(() => {
+    const unregister = ctx.storage.backend.register('json', backend)
+    return async () => {
+      unregister()
+      await backend.close()
+    }
+  })
+}
@@ -0,0 +1,32 @@
+/**
+ * Package-owned invariant companion for `@deepseek-ai/dsh-storage-json`.
+ * @module @deepseek-ai/dsh-storage-json/invariant
+ */
+
+/* jscpd:ignore-start */
+import type { Context } from 'cordis'
+import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
+
+const PACKAGE_NAME = '@deepseek-ai/dsh-storage-json'
+
+/** Cordis companion plugin name. */
+export const name = 'storage-json-invariant'
+/** Service required before the companion can reserve package ownership. */
+export const inject = ['invariants']
+
+/**
+ * No runtime invariant: correctness here is write-durability and
+ * publish-then-reparse equivalence, which require medium round-trip tests
+ * (the shared backend conformance suite); the backend exposes no continuously
+ * observable in-process relation.
+ */
+const install: InvariantInstaller = () => {}
+
+/**
+ * Register this package's invariant companion.
+ * @param ctx - Cordis context carrying the invariant service.
+ * @returns the installed registration's disposer after setup succeeds.
+ */
+export const apply = (ctx: Context): Promise<() => void> =>
+  Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
+/* jscpd:ignore-end */
@@ -0,0 +1,105 @@
+/**
+ * One opened JSON unit. The in-memory state is authoritative; every write
+ * primitive mutates it and republishes the whole file atomically. Writes are
+ * NOT queued here — per the backend contract, write ordering belongs to the
+ * caller (the domain layer's write chain); this unit only guarantees that
+ * each single call publishes a complete, durable file.
+ * @module @deepseek-ai/dsh-storage-json/src/unit
+ */
+
+import { readFile } from 'node:fs/promises'
+import { StorageError } from '@deepseek-ai/dsh-storage'
+import type { KvUnit, KvUnitDescriptor } from '@deepseek-ai/dsh-storage'
+import { writeAtomic } from './atomic.ts'
+import { parse, serialize } from './format.ts'
+import type { UnitState } from './format.ts'
+
+/** Open (load or lazily create) one unit backed by `path`. */
+export async function openJsonUnit(
+  descriptor: KvUnitDescriptor,
+  path: string,
+  onClose: () => void,
+): Promise<KvUnit> {
+  let text: string | undefined
+  try {
+    text = await readFile(path, 'utf8')
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error
+    // Missing file = empty unit; materialization defers to the first write.
+  }
+  const state: UnitState =
+    text === undefined
+      ? {
+          version: descriptor.version,
+          global: null,
+          tables: new Map(descriptor.tables.map((table) => [table, new Map()])),
+        }
+      : parse(text, descriptor)
+  return new JsonKvUnit(descriptor, path, state, onClose)
+}
+
+class JsonKvUnit implements KvUnit {
+  private closed = false
+
+  constructor(
+    private readonly descriptor: KvUnitDescriptor,
+    private readonly path: string,
+    private readonly state: UnitState,
+    private readonly onClose: () => void,
+  ) {}
+
+  async loadAll(): Promise<{ tables: Record<string, Record<string, unknown>>; global: unknown | null }> {
+    this.assertOpen()
+    const tables: Record<string, Record<string, unknown>> = {}
+    for (const [table, records] of this.state.tables) {
+      tables[table] = Object.fromEntries(records)
+    }
+    return { tables, global: this.state.global }
+  }
+
+  async putRecord(table: string, key: string, value: unknown): Promise<void> {
+    this.assertOpen()
+    this.records(table).set(key, value)
+    await this.publish()
+  }
+
+  async deleteRecord(table: string, key: string): Promise<void> {
+    this.assertOpen()
+    if (this.records(table).delete(key)) {
+      await this.publish()
+    }
+  }
+
+  async setGlobal(value: unknown): Promise<void> {
+    this.assertOpen()
+    if (!this.descriptor.hasGlobal) {
+      throw new Error(`unit '${this.descriptor.name}' does not declare a global slot`)
+    }
+    this.state.global = value
+    await this.publish()
+  }
+
+  async close(): Promise<void> {
+    if (this.closed) return
+    this.closed = true
+    this.onClose()
+  }
+
+  private assertOpen(): void {
+    if (this.closed) {
+      throw new StorageError('closed', `unit '${this.descriptor.name}' is closed`)
+    }
+  }
+
+  private records(table: string): Map<string, unknown> {
+    const records = this.state.tables.get(table)
+    if (!records) {
+      throw new Error(`unit '${this.descriptor.name}' does not declare table '${table}'`)
+    }
+    return records
+  }
+
+  private publish(): Promise<void> {
+    return writeAtomic(this.path, serialize(this.descriptor.name, this.state))
+  }
+}
@@ -0,0 +1,80 @@
+import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterAll, describe, expect, it } from 'vitest'
+import { runKvBackendContract } from '../../storage/tests/contract.ts'
+import { JsonStorageBackend } from '../src/index.ts'
+
+const roots: string[] = []
+
+async function freshRoot(): Promise<string> {
+  const root = await mkdtemp(join(tmpdir(), 'dsh-storage-json-'))
+  roots.push(root)
+  return root
+}
+
+afterAll(async () => {
+  for (const root of roots) await rm(root, { recursive: true, force: true })
+})
+
+runKvBackendContract('json', async () => {
+  const root = await freshRoot()
+  return {
+    backend: new JsonStorageBackend(root),
+    reopen: async () => new JsonStorageBackend(root),
+  }
+})
+
+describe('json backend specifics', () => {
+  const descriptor = { name: 'shape', version: 1, tables: ['t'], hasGlobal: true }
+
+  it('publishes a human-readable pretty-printed file', async () => {
+    const root = await freshRoot()
+    const backend = new JsonStorageBackend(root)
+    const unit = await backend.kv.open(descriptor)
+    await unit.putRecord('t', 'k', { hello: 'world' })
+    const text = await readFile(join(root, 'shape.json'), 'utf8')
+    expect(text).toBe(`${JSON.stringify(
+      { unit: { name: 'shape', version: 1 }, global: null, tables: { t: { k: { hello: 'world' } } } },
+      null,
+      2,
+    )}\n`)
+    await backend.close()
+  })
+
+  it('defers materialization until the first write', async () => {
+    const root = await freshRoot()
+    const backend = new JsonStorageBackend(root)
+    await backend.kv.open(descriptor)
+    await expect(readFile(join(root, 'shape.json'), 'utf8')).rejects.toMatchObject({ code: 'ENOENT' })
+    await backend.close()
+  })
+
+  it('rejects a malformed medium', async () => {
+    const root = await freshRoot()
+    await writeFile(join(root, 'shape.json'), 'not json at all', 'utf8')
+    const backend = new JsonStorageBackend(root)
+    await expect(backend.kv.open(descriptor)).rejects.toMatchObject({ code: 'malformed-medium' })
+    await backend.close()
+  })
+
+  it('rejects a foreign unit header', async () => {
+    const root = await freshRoot()
+    await writeFile(
+      join(root, 'shape.json'),
+      JSON.stringify({ unit: { name: 'other', version: 1 }, global: null, tables: {} }),
+      'utf8',
+    )
+    const backend = new JsonStorageBackend(root)
+    await expect(backend.kv.open(descriptor)).rejects.toMatchObject({ code: 'malformed-medium' })
+    await backend.close()
+  })
+
+  it('rejects double-open of one unit', async () => {
+    const root = await freshRoot()
+    const backend = new JsonStorageBackend(root)
+    await backend.kv.open(descriptor)
+    await expect(backend.kv.open(descriptor)).rejects.toMatchObject({ code: 'malformed-medium' })
+    await backend.close()
+  })
+})
@@ -0,0 +1,27 @@
+{
+  "extends": "../../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib/types"
+  },
+  "include": [
+    "src"
+  ],
+  "references": [
+    {
+      "path": "../../../vendor/cosmokit"
+    },
+    {
+      "path": "../../../vendor/cordis"
+    },
+    {
+      "path": "../../../vendor/schemastery"
+    },
+    {
+      "path": "../storage"
+    },
+    {
+      "path": "../../support/invariants"
+    }
+  ]
+}