From c330c1cd3ec627a2020ff6f80fad8765e358fec5 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Tue, 28 Jul 2026 01:26:19 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20dsh-session-projection-cache=20?= =?UTF-8?q?=E2=80=94=20durable=20projection=20checkpoints=20and=20the=20co?= =?UTF-8?q?ld-read=20ladder?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New package on the domain data form: one session_projcache record per session (key → {stateVersion, observedSeq, state}), landing beside workspace.json under the shipped json backend. Write policy: two mandatory points (turn/end + session disposal) with count/interval throttling between them (both Config fields required — flush cadence is a deployment choice); every background write is fail-soft (log + stay stale, self-heal on the next write or cold read). coldSnapshot(id) runs the read ladder — cached rows + persistence readFrom from the registry's anchored restore floor + registry restore + fail-soft write-back — detecting crash-repair-shrunk logs via the one-below anchor and degrading to a single full re-read. Mounted in apps/cli/cordis.yml (writeEveryEvents 200 / writeIntervalMs 5000). --- apps/cli/cordis.yml | 10 + apps/cli/package.json | 1 + packages/session-projection/README.md | 1 + .../session-projection-cache/README.md | 60 +++++ .../session-projection-cache/package.json | 52 ++++ .../session-projection-cache/src/index.ts | 237 ++++++++++++++++ .../session-projection-cache/src/invariant.ts | 35 +++ .../session-projection-cache/src/spec.ts | 51 ++++ .../tests/cache.spec.ts | 255 ++++++++++++++++++ .../session-projection-cache/tsconfig.json | 39 +++ pnpm-lock.yaml | 37 +++ tsconfig.host.json | 1 + 12 files changed, 779 insertions(+) create mode 100644 packages/session-projection/session-projection-cache/README.md create mode 100644 packages/session-projection/session-projection-cache/package.json create mode 100644 packages/session-projection/session-projection-cache/src/index.ts create mode 100644 packages/session-projection/session-projection-cache/src/invariant.ts create mode 100644 packages/session-projection/session-projection-cache/src/spec.ts create mode 100644 packages/session-projection/session-projection-cache/tests/cache.spec.ts create mode 100644 packages/session-projection/session-projection-cache/tsconfig.json diff --git a/apps/cli/cordis.yml b/apps/cli/cordis.yml index 0dbd6c839e..bf0ab30598 100644 --- a/apps/cli/cordis.yml +++ b/apps/cli/cordis.yml @@ -116,6 +116,16 @@ - id: workspace name: '@deepseek-ai/dsh-workspace' +# Persisted projection cache: durable per-session checkpoints of every +# registered projection unit (json backend → ./.storages/session_projcache.json, +# beside workspace.json), throttled between the two mandatory points +# (turn/end + detach), serving cold listings without full-log loads. +- id: session-projection-cache + name: '@deepseek-ai/dsh-session-projection-cache' + config: + writeEveryEvents: 200 + writeIntervalMs: 5000 + # Managed child-process groups for the bash executor (spawn/kill/output plumbing). - id: subprocess name: '@deepseek-ai/dsh-subprocess-local' diff --git a/apps/cli/package.json b/apps/cli/package.json index 7d984ee1b1..6aa491221e 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -57,6 +57,7 @@ "@deepseek-ai/dsh-session": "workspace:^", "@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^", "@deepseek-ai/dsh-session-projection": "workspace:^", + "@deepseek-ai/dsh-session-projection-cache": "workspace:^", "@deepseek-ai/dsh-session-title": "workspace:^", "@deepseek-ai/dsh-session-title-first-message-llm": "workspace:^", "@deepseek-ai/dsh-skill": "workspace:^", diff --git a/packages/session-projection/README.md b/packages/session-projection/README.md index 81c67d56e1..ae80a90570 100644 --- a/packages/session-projection/README.md +++ b/packages/session-projection/README.md @@ -7,3 +7,4 @@ Session-projection capability family: the seam through which domain host plugins | Package | ctx key | Role | |---|---|---| | [`session-projection`](session-projection/README.md) | `sessionProjections` | The interface package: the merge-extensible `SessionProjectionMap` type table, the `ProjectionDefinition` unit contract, and the eagerly driven registry carriers read synchronously | +| [`session-projection-cache`](session-projection-cache/README.md) | `sessionProjectionCache` | Persisted projection cache: durable per-session unit checkpoints over the domain data form, throttled write-behind with mandatory turn/end + detach points, and the cold-read ladder (cache row + persistence tail replay) | diff --git a/packages/session-projection/session-projection-cache/README.md b/packages/session-projection/session-projection-cache/README.md new file mode 100644 index 0000000000..8bb8efe4cd --- /dev/null +++ b/packages/session-projection/session-projection-cache/README.md @@ -0,0 +1,60 @@ +# @deepseek-ai/dsh-session-projection-cache + +The persisted projection cache (`ctx.sessionProjectionCache`): durable checkpoints of every registered projection unit's state, one record per session on the domain data form (`session_projcache` domain — the shipped json backend lands it beside `workspace.json` under the configured storage root). Design authority: the [session-projection RFC](../../../.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md) (persisted projection cache section). + +A stored row `(key → {stateVersion, observedSeq, state})` is a fold shortcut, never an authority: possibly stale (`observedSeq` says exactly how stale) but never wrong. Consequences the implementation commits to: + +- **Every background write is fail-soft.** A failed durable write logs a warning and keeps the cache stale; the next write or cold read self-heals. A crash between writes costs a longer tail replay, never a wrong value. +- **`stateVersion` mismatch discards, never migrates.** A unit bump invalidates its rows at read time; the key refolds from the log. +- **Whole-record writes.** Each write replaces the session's full checkpoint (the registry cut is always complete), snapshotted through the lossless-JSON boundary — a unit state violating the plain-JSON contract fails loud. + +## Write policy + +Two mandatory points, throttled in between: + +| Trigger | Nature | +|---|---| +| `turn/end` | Mandatory — the turn-final value is what cold reads want. | +| Session disposal (detach) | Mandatory — the live-to-cold moment; after it the cold ladder serves this session. | +| `writeEveryEvents` committed events | Config throttle (count). | +| `writeIntervalMs` since the first dirty event | Config throttle (interval). | + +Both `Config` fields are required (no defaults): flush cadence is a deployment choice with no universally correct value, stated in cordis.yml. + +## Cold read (`coldSnapshot(id, signal?)`) + +The read ladder, zero full-log load on the happy path: cached rows → `sessionProjections.restoreFloor` (anchored one event below the lowest usable watermark) → persistence `readFrom(id, floor)` → `sessionProjections.restore` → fail-soft write-back of the refreshed rows. The anchor makes a shrunk log (crash-repair truncation) provable: an overreaching row triggers exactly one full re-read from seq 0 instead of serving a ghost value. No registered units serve `{asOfSeq: -1, values: {}}` without touching persistence; a session with no persisted log rejects with the seam's `not found`. + +`write(session)` is the synchronous-cut checkpoint both mandatory points use; carriers may call it directly (not fail-soft — the fail-soft wrappers own containment). + +## Composition + +```yaml +- id: session-projection-cache + name: '@deepseek-ai/dsh-session-projection-cache' + config: + writeEveryEvents: 200 + writeIntervalMs: 5000 +``` + +Injects `storageDomain`, `sessionProjections`, `sessionPersistence`, `sessions`. Without this row the projection system runs live-only (watermark cache; cold reads fall back to full log loads wherever a carrier implements them). + +## Model Experience + +### What the model sees + +Nothing. The cache is a host read-model accelerator; no prompt, schema, or tool surface. + +### Token effect + +Zero. + +### KV Cache effect + +None — no request content changes. + +## Known Limitations and Deferred Work + +- **No eviction or retention surface** — records accumulate per session; pruning stored checkpoints is out-of-band maintenance, same stance as session persistence itself. +- **Interval throttle is per-session coarse** — the timer arms at the first dirty event after a clean write; a steady sub-threshold trickle writes once per interval, not a sliding window. +- **`coldSnapshot` reads are not deduplicated** — two concurrent cold reads of one session each run the ladder; last write-back wins (rows are equivalent), acceptable for listing-scale call rates. diff --git a/packages/session-projection/session-projection-cache/package.json b/packages/session-projection/session-projection-cache/package.json new file mode 100644 index 0000000000..5080713412 --- /dev/null +++ b/packages/session-projection/session-projection-cache/package.json @@ -0,0 +1,52 @@ +{ + "name": "@deepseek-ai/dsh-session-projection-cache", + "description": "Persisted projection cache (ctx.sessionProjectionCache): durable per-session projection checkpoints over the domain data form, throttled write-behind, and the cold-read ladder (cache row + persistence tail replay)", + "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/**/*.js", + "lib/types/**/*.d.ts", + "lib/types/**/*.d.ts.map", + "src" + ], + "license": "BSD-3-Clause", + "dependencies": { + "schemastery": "^3.18.0", + "zod": "^4.4.3" + }, + "peerDependencies": { + "@deepseek-ai/dsh-invariants": "^0.0.1", + "@deepseek-ai/dsh-session": "^0.0.1", + "@deepseek-ai/dsh-session-persistence": "^0.0.1", + "@deepseek-ai/dsh-session-projection": "^0.0.1", + "@deepseek-ai/dsh-storage-domain": "^0.0.1", + "cordis": "^4.0.0-rc.7" + }, + "devDependencies": { + "@deepseek-ai/dsh-invariants": "workspace:^", + "@deepseek-ai/dsh-session": "workspace:^", + "@deepseek-ai/dsh-session-persistence": "workspace:^", + "@deepseek-ai/dsh-session-projection": "workspace:^", + "@deepseek-ai/dsh-storage": "workspace:^", + "@deepseek-ai/dsh-storage-domain": "workspace:^", + "@deepseek-ai/dsh-storage-json": "workspace:^", + "cordis": "^4.0.0-rc.7" + } +} diff --git a/packages/session-projection/session-projection-cache/src/index.ts b/packages/session-projection/session-projection-cache/src/index.ts new file mode 100644 index 0000000000..c9336dd848 --- /dev/null +++ b/packages/session-projection/session-projection-cache/src/index.ts @@ -0,0 +1,237 @@ +/** + * Persisted projection cache (`ctx.sessionProjectionCache`): durable + * checkpoints of every registered projection unit's state, one record per + * session on the domain data form (`session_projcache` domain — the shipped + * json backend lands it beside `workspace.json`). The cache is a fold + * shortcut, never an authority: a row is possibly stale (its `observedSeq` + * says how stale) but never wrong, so every write path is fail-soft (a lost + * write costs a longer tail replay on the next cold read) and a + * `stateVersion` mismatch discards the row instead of migrating it. Design + * authority: the session-projection RFC + * (.agents/notes/proposed/architecture/2026-07-27-session-projection-and-command-log.md). + * @module @deepseek-ai/dsh-session-projection-cache + */ + +import { Context, Service } from 'cordis' +import z from 'schemastery' +import { snapshotJsonValue } from '@deepseek-ai/dsh-session' +import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session' +// Empty type import: applies the package's cordis Context merge +// (`ctx.sessionPersistence`), which this service reads on the cold path. +import type {} from '@deepseek-ai/dsh-session-persistence' +import type { ProjectionCheckpoint, ProjectionSnapshot } from '@deepseek-ai/dsh-session-projection' +import type { KvTable } from '@deepseek-ai/dsh-storage-domain' +import { projectionCacheDomainSpec } from './spec.ts' +import type { CheckpointRecord } from './spec.ts' + +export { checkpointRecord, checkpointRow, projectionCacheDomainSpec } from './spec.ts' +export type { CheckpointRecord } from './spec.ts' + +declare module 'cordis' { + interface Context { + sessionProjectionCache: SessionProjectionCache + } +} + +/** + * Plugin config. Both throttle triggers are deployment choices with no + * universally correct value, so the composition states them explicitly + * (cordis.yml); the two mandatory write points (`turn/end` and session + * disposal) are policy, not tunables, and always fire. + */ +export interface Config { + /** Committed events per session that force a durable checkpoint write between mandatory points. */ + writeEveryEvents: number + /** Longest time (milliseconds) a dirty checkpoint may stay unwritten between mandatory points. */ + writeIntervalMs: number +} + +export const Config: z = z.object({ + writeEveryEvents: z.natural().min(1).required(), + writeIntervalMs: z.natural().min(1).required(), +}) + +/** Per-session write-behind bookkeeping (live sessions only; dropped at retire). */ +interface DirtyState { + /** Committed events since the last durable write. */ + pending: number + /** Interval trigger armed at the first dirty event after a clean write. */ + timer: ReturnType | undefined +} + +/** + * The persisted projection cache service. Opens the `session_projcache` + * domain at init, checkpoints live sessions on a throttled write-behind + * (count/interval triggers from {@link Config}) plus two mandatory points — + * `turn/end` and session disposal (the live-to-cold moment) — and serves the + * cold-read ladder: cached row, persistence `readFrom` tail, registry + * `restore`, durable write-back. Every durable write is fail-soft: failures + * log a warning and the cache self-heals on the next write or cold read. + */ +export class SessionProjectionCache extends Service { + static inject = ['storageDomain', 'sessionProjections', 'sessionPersistence', 'sessions'] + + static Config: z = Config + + private table?: KvTable + private readonly dirty = new Map() + + constructor(ctx: Context, public config: Config) { + super(ctx, 'sessionProjectionCache') + } + + /** Open the domain and install the write-behind listeners. */ + protected async [Service.init](): Promise { + const domain = await this.ctx.storageDomain.open(projectionCacheDomainSpec) + this.ctx.effect(() => () => domain.close(), 'sessionProjectionCache.domainClose') + this.table = domain.table('sessions') + this.installWritePath() + } + + /** + * The stored checkpoint rows for one session, or an empty checkpoint when + * none is stored. Synchronous from the domain's in-memory state. + * @param id - the session whose cached rows are read. + * @returns the persisted `key → row` checkpoint (possibly empty). + */ + checkpointOf(id: SessionId): ProjectionCheckpoint { + return this.requireTable().get(id)?.rows ?? {} + } + + /** + * Durably checkpoint one live session NOW (both mandatory points call + * this; tests and carriers may too). The registry cut is snapshotted at + * this boundary (states are live references), then the whole record is + * replaced. NOT fail-soft — callers on the fail-soft paths contain it. + * @param session - the live session to checkpoint. + * @returns resolution after durability and event emission. + */ + async write(session: Session): Promise { + const rows = this.ctx.sessionProjections.checkpoint(session) + this.markClean(session) + await this.put(session.id, rows) + } + + /** + * Cold-read one persisted session's projections with zero full-log load: + * cached rows + a persistence `readFrom` tail from the registry's restore + * floor, refolded by the registry and written back (fail-soft) so the next + * cold read starts closer. A cache row invalidated by a shrunk log + * (crash-repair truncation) triggers one full re-read from seq 0 — the + * ladder's slow rung, still no crash. Rejects when the session has no + * persisted log (`not found` from the persistence seam). + * @param id - the persisted session to read. + * @param signal - optional cancellation for the persistence reads. + * @returns the snapshot cut at the stored log end. + */ + async coldSnapshot(id: SessionId, signal?: AbortSignal): Promise { + const cached = this.checkpointOf(id) + const floor = this.ctx.sessionProjections.restoreFloor(cached) + if (floor === undefined) return { asOfSeq: -1, values: {} } + const persistence = this.ctx.sessionPersistence + let restored: { snapshot: ProjectionSnapshot; checkpoint: ProjectionCheckpoint } + const tail = await persistence.readFrom(id, floor, signal) + try { + restored = this.ctx.sessionProjections.restore(cached, tail.events, floor) + } catch { + // The one recoverable restore failure: a row overreaching the stored + // log end (or predating the floor), detected by the registry. Both + // resolve identically — discard the cache and refold the full log. + const whole = await persistence.readFrom(id, 0, signal) + restored = this.ctx.sessionProjections.restore({}, whole.events, 0) + } + await this.putSoft(id, restored.checkpoint, 'cold-read write-back') + return restored.snapshot + } + + // --- write-behind (throttle + mandatory points) --- + + private installWritePath(): void { + // Every committed event advances the dirty counter; turn/end is a + // mandatory point (the durable value most reads want is the turn-final + // one), count/interval throttle the in-turn stream. + this.ctx.on('session/event', (session: Session, event: SessionEvent) => { + if (event.type === 'turn/end') { + void this.flushSoft(session, 'turn/end') + return + } + const state = this.dirty.get(session) ?? { pending: 0, timer: undefined } + this.dirty.set(session, state) + state.pending += 1 + if (state.pending >= this.config.writeEveryEvents) { + void this.flushSoft(session, 'count threshold') + return + } + state.timer ??= setTimeout(() => { + void this.flushSoft(session, 'interval') + }, this.config.writeIntervalMs) + }) + + // Detach (the live-to-cold moment): the second mandatory point. After + // this write the cold-read ladder serves the session from the cache. + // flushSoft's synchronous prefix reads and resets the dirty state, so + // dropping it (timer already cleared by markClean) right after is safe. + this.ctx.on('session/disposed', (session: Session) => { + void this.flushSoft(session, 'detach') + this.markClean(session) + this.dirty.delete(session) + }) + + // Clear pending timers with the plugin (their sessions outlive the cache). + this.ctx.effect(() => () => { + for (const state of this.dirty.values()) { + if (state.timer !== undefined) clearTimeout(state.timer) + } + this.dirty.clear() + }, 'sessionProjectionCache.timers') + } + + /** One fail-soft durable checkpoint: skip when clean, log on failure. */ + private async flushSoft(session: Session, trigger: string): Promise { + const state = this.dirty.get(session) + const mandatory = trigger === 'turn/end' || trigger === 'detach' + if (!mandatory && (state === undefined || state.pending === 0)) return + try { + await this.write(session) + } catch (error) { + this.ctx.logger.warn(`session projection cache: ${trigger} write for "${session.id}" failed (cache stays stale): ${String(error)}`) + } + } + + /** Reset one session's dirty bookkeeping (its checkpoint is being written). */ + private markClean(session: Session): void { + const state = this.dirty.get(session) + if (state === undefined) return + state.pending = 0 + if (state.timer !== undefined) { + clearTimeout(state.timer) + state.timer = undefined + } + } + + /** Replace one session's stored record with a detached snapshot of `rows`. */ + private async put(id: SessionId, rows: ProjectionCheckpoint): Promise { + const detached = snapshotJsonValue(rows) + if (detached === undefined) { + throw new TypeError('projection checkpoint is not losslessly JSON-serializable (a unit state violates the plain-JSON contract)') + } + await this.requireTable().put(id, { rows: detached as CheckpointRecord['rows'] }) + } + + /** Fail-soft {@link put}: cache writes must never fail their caller's read or event path. */ + private async putSoft(id: SessionId, rows: ProjectionCheckpoint, what: string): Promise { + try { + await this.put(id, rows) + } catch (error) { + this.ctx.logger.warn(`session projection cache: ${what} for "${id}" failed (cache stays stale): ${String(error)}`) + } + } + + private requireTable(): KvTable { + /* v8 ignore next -- Service.init assigns the table before the service becomes injectable */ + if (this.table === undefined) throw new Error('session projection cache is not initialized') + return this.table + } +} + +export default SessionProjectionCache diff --git a/packages/session-projection/session-projection-cache/src/invariant.ts b/packages/session-projection/session-projection-cache/src/invariant.ts new file mode 100644 index 0000000000..a5a5515f22 --- /dev/null +++ b/packages/session-projection/session-projection-cache/src/invariant.ts @@ -0,0 +1,35 @@ +/** + * Package-owned invariant companion for `@deepseek-ai/dsh-session-projection-cache`. + * @module @deepseek-ai/dsh-session-projection-cache/invariant + */ + +/* jscpd:ignore-start */ +import type { Context } from 'cordis' +import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants' + +const PACKAGE_NAME = '@deepseek-ai/dsh-session-projection-cache' + +/** Cordis companion plugin name. */ +export const name = 'session-projection-cache-invariant' +/** Service required before the companion can reserve package ownership. */ +export const inject = ['invariants'] + +/** + * No runtime invariant: the cache's correctness relation (a stored row equals + * the registry fold at its `observedSeq`) is only checkable by re-running the + * fold over the persisted log — duplicating the implementation rather than + * detecting drift — and its staleness is by design (fail-soft writes). The + * durable boundary is already schema-validated by the storage-domain layer + * on every reopen, and the read ladder's version/watermark guards are proven + * by the package spec. + */ +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 */ diff --git a/packages/session-projection/session-projection-cache/src/spec.ts b/packages/session-projection/session-projection-cache/src/spec.ts new file mode 100644 index 0000000000..1cc4931032 --- /dev/null +++ b/packages/session-projection/session-projection-cache/src/spec.ts @@ -0,0 +1,51 @@ +/** + * The session-projcache domain declaration: one `sessions` table keyed by + * {@link SessionId}, each record the full projection checkpoint for one + * session (`key → {stateVersion, observedSeq, state}` rows). The spec object + * is the single source of the domain's identity, version, and record schema; + * the storage-domain routing decides the medium (the shipped composition's + * json backend lands it at `/session_projcache.json`, beside + * `workspace.json`). + * @module @deepseek-ai/dsh-session-projection-cache/src/spec + */ + +import { z } from 'zod' +import { SessionId } from '@deepseek-ai/dsh-session' +import { defineDomain, domainTable } from '@deepseek-ai/dsh-storage-domain' + +/** + * One persisted checkpoint row (the RFC's `(sessionId, key, stateVersion, + * observedSeq, state)` minus the two record keys). `state` is the unit's + * internal state — plain JSON by the unit contract; `z.json()` enforces that + * at the durable boundary. A row is never wrong, only possibly stale: + * `observedSeq` says exactly how stale, and a `stateVersion` mismatch + * discards it at read time (never a migration). + */ +export const checkpointRow = z.object({ + stateVersion: z.number().int().nonnegative(), + observedSeq: z.number().int().gte(-1), + state: z.json(), +}) + +/** + * One session's stored record: its checkpoint rows keyed by projection key. + * The whole record is replaced on every write (whole-value discipline — the + * registry checkpoint is always the complete per-session cut). + */ +export const checkpointRecord = z.object({ + rows: z.record(z.string(), checkpointRow), +}) + +/** One stored per-session checkpoint record, inferred from {@link checkpointRecord}. */ +export type CheckpointRecord = z.infer + +/** + * The session-projcache domain spec. Version bumps discard the whole medium + * (cache semantics: a stale or unreadable cache costs a longer tail replay, + * never a wrong value). + */ +export const projectionCacheDomainSpec = defineDomain({ + name: 'session_projcache', + version: 1, + tables: { sessions: domainTable(checkpointRecord) }, +}) diff --git a/packages/session-projection/session-projection-cache/tests/cache.spec.ts b/packages/session-projection/session-projection-cache/tests/cache.spec.ts new file mode 100644 index 0000000000..defeddfd32 --- /dev/null +++ b/packages/session-projection/session-projection-cache/tests/cache.spec.ts @@ -0,0 +1,255 @@ +/** + * 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 + * cold-read ladder (cached row + readFrom tail + registry restore + + * write-back; version bump and shrunk-log rows degrade to a full re-read). + */ + +import { afterEach, describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import { z } from 'zod' +import Storage from '@deepseek-ai/dsh-storage' +import { DomainFacility } from '@deepseek-ai/dsh-storage-domain' +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 { MemoryMediaPool, MemoryStorageBackend } from '../../../storage/storage-domain/tests/helpers/memory-backend.ts' +import SessionProjectionCache from '../src/index.ts' + +declare module '@deepseek-ai/dsh-session-projection/types' { + interface SessionProjectionMap { + 'cache-test/marks': { marks: string[] } + } +} + +declare module '@deepseek-ai/dsh-session' { + interface SessionEventMap { + 'cache-test/mark': { marks: string[] } + } + + interface OutOfBandSessionEventMap { + 'cache-test/mark': true + } +} + +type MarksState = { marks: string[] } | null +const marksUnit = (stateVersion = 1): ProjectionDefinition<'cache-test/marks', MarksState> => ({ + key: 'cache-test/marks', + schema: z.object({ marks: z.array(z.string()) }), + init: () => null, + apply: (state, event) => (event.type === 'cache-test/mark' ? (event as SessionEvent<'cache-test/mark'>).data : state), + view: state => state ?? { marks: [] }, + stateVersion, +}) + +/** A persistence double serving readFrom over a fixed per-id stored log. */ +function fakePersistence(logs: Map) { + const readFrom = vi.fn(async (id: SessionId, fromSeq: number) => { + const events = logs.get(String(id)) + if (events === undefined) throw new Error(`session "${id}" not found`) + return { + meta: { version: 0, id, createdAt: 0 }, + events: events.filter(event => event.seq >= fromSeq), + } + }) + return { readFrom } +} + +interface HarnessOptions { + pool?: MemoryMediaPool + config?: { writeEveryEvents: number; writeIntervalMs: number } + stateVersion?: number + logs?: Map +} + +const contexts: Context[] = [] + +async function harness(options: HarnessOptions = {}) { + const pool = options.pool ?? new MemoryMediaPool() + const logs = options.logs ?? new Map() + const ctx = new Context() + contexts.push(ctx) + await ctx.plugin(Storage) + ctx.storage.backend.register('memory', new MemoryStorageBackend(pool)) + const facility = new DomainFacility(ctx, { backend: 'memory', routes: {} }) + ctx.storage.mount('domain', facility) + ctx.provide('storageDomain', facility) + await ctx.plugin(SessionStore) + await ctx.plugin(SessionProjectionRegistry) + ctx.sessionProjections.register(marksUnit(options.stateVersion)) + const persistence = fakePersistence(logs) + ctx.provide('sessionPersistence', persistence as never) + const fiber = await ctx.plugin(SessionProjectionCache, options.config ?? { writeEveryEvents: 100, writeIntervalMs: 60_000 }) + return { ctx, pool, logs, fiber, persistence, 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' } }) + +/** The stored medium rows for one session id (undefined = never written). */ +function storedRows(pool: MemoryMediaPool, id: Session['id']) { + const record = pool.media.get('session_projcache')?.tables.get('sessions')?.get(String(id)) as + { rows: Record } | undefined + return record?.rows +} + +/** Wait until queued fail-soft writes (event-listener fire-and-forget) drain. */ +const settle = () => new Promise(resolve => setTimeout(resolve, 0)) + +afterEach(async () => { + vi.useRealTimers() + await Promise.all(contexts.splice(0).map(ctx => ctx.fiber.dispose())) +}) + +describe('SessionProjectionCache write policy', () => { + it('writes a durable checkpoint at turn/end (mandatory point)', async () => { + const { ctx, pool } = await harness() + const session = ctx.sessions.create(SessionId('turn-end')) + mark(session, ['a']) + expect(storedRows(pool, session.id)).toBeUndefined() // throttled: no write yet + const end = endTurn(session) + await settle() + const rows = storedRows(pool, session.id) + expect(rows?.['cache-test/marks']).toEqual({ stateVersion: 1, observedSeq: end.seq, state: { marks: ['a'] } }) + }) + + it('writes at session disposal (detach, the live-to-cold moment)', async () => { + const { ctx, pool } = await harness() + // 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(SessionId('detach')) + }, { inject: ['sessions'] })) + if (session === undefined) throw new Error('session was not created') + mark(session, ['live']) + await owner.dispose() + await settle() + expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['live'] }) + }) + + it('flushes when the in-turn event count reaches the configured threshold', async () => { + const { ctx, pool } = await harness({ config: { writeEveryEvents: 3, writeIntervalMs: 60_000 } }) + const session = ctx.sessions.create(SessionId('count')) + mark(session, ['1']) + mark(session, ['2']) + await settle() + expect(storedRows(pool, session.id)).toBeUndefined() + mark(session, ['3']) + await settle() + expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['3'] }) + }) + + it('flushes on the configured interval when the count threshold is not reached', async () => { + vi.useFakeTimers() + const { ctx, pool } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 250 } }) + const session = ctx.sessions.create(SessionId('interval')) + mark(session, ['slow']) + await vi.advanceTimersByTimeAsync(249) + expect(storedRows(pool, session.id)).toBeUndefined() + await vi.advanceTimersByTimeAsync(1) + await vi.runAllTicks() + expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['slow'] }) + }) + + it('contains a durable write failure: logs a warning, event path unharmed, next write self-heals', async () => { + const { ctx, pool } = await harness() + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {}) + const session = ctx.sessions.create(SessionId('fail-soft')) + mark(session, ['x']) + pool.failNextWrites = 1 + endTurn(session) + await settle() + expect(storedRows(pool, session.id)).toBeUndefined() + expect(warn).toHaveBeenCalledWith(expect.stringContaining('turn/end write for "fail-soft" failed')) + // Self-heal: the next mandatory point writes the current cut. + mark(session, ['y']) + endTurn(session) + await settle() + expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['y'] }) + }) +}) + +describe('SessionProjectionCache cold read', () => { + const storedLog = (marks: string[][]): SessionEvent[] => { + const events: SessionEvent[] = [ + { type: 'turn/start', seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }, + ] + for (const m of marks) { + events.push({ type: 'cache-test/mark', seq: events.length, time: events.length, data: { marks: m } } as SessionEvent) + } + events.push({ type: 'turn/end', seq: events.length, time: events.length, data: { turn: 1, reason: { kind: 'completed' } } }) + return events + } + + /** Pre-seed the medium with one stored checkpoint record (before the domain opens). */ + function seedRow(pool: MemoryMediaPool, id: string, row: { stateVersion: number; observedSeq: number; state: unknown }): void { + pool.versions.set('session_projcache', 1) + pool.media.set('session_projcache', { + tables: new Map([['sessions', new Map([[id, { rows: { 'cache-test/marks': row } }]])]]), + global: null, + }) + } + + it('serves a cold session from the cache row plus a bounded tail read, and writes the refresh back', async () => { + const pool = new MemoryMediaPool() + const logs = new Map([['cold', storedLog([['a'], ['a', 'b']])]]) + // A warm-era checkpoint at watermark 1 (only ['a'] folded). + seedRow(pool, 'cold', { stateVersion: 1, observedSeq: 1, state: { marks: ['a'] } }) + const { cache, persistence, pool: samePool } = await harness({ pool, logs }) + const id = SessionId('cold') + const snapshot = await cache.coldSnapshot(id) + expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a', 'b'] }) + expect(snapshot.asOfSeq).toBe(3) + // The tail read was bounded by the anchored floor (watermark 1 -> floor 1), not 0. + expect(persistence.readFrom).toHaveBeenCalledWith(id, 1, undefined) + // Write-back: the stored row advanced to the served cut. + expect(storedRows(samePool, id)?.['cache-test/marks']) + .toEqual({ stateVersion: 1, observedSeq: 3, state: { marks: ['a', 'b'] } }) + }) + + it('discards a version-mismatched row and refolds the full log', async () => { + const pool = new MemoryMediaPool() + const logs = new Map([['bumped', storedLog([['a']])]]) + seedRow(pool, 'bumped', { stateVersion: 1, observedSeq: 2, state: { marks: ['stale'] } }) + const { cache, persistence } = await harness({ pool, logs, stateVersion: 2 }) + const snapshot = await cache.coldSnapshot(SessionId('bumped')) + expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] }) + // Mismatch pulls the floor to 0: one full read, no second pass needed. + expect(persistence.readFrom).toHaveBeenCalledTimes(1) + expect(persistence.readFrom).toHaveBeenCalledWith(SessionId('bumped'), 0, undefined) + }) + + it('detects a log shrunk below the row watermark and degrades to one full re-read', async () => { + const pool = new MemoryMediaPool() + const logs = new Map([['shrunk', storedLog([['a']])]]) // seqs 0..2 + seedRow(pool, 'shrunk', { stateVersion: 1, observedSeq: 9, state: { marks: ['ghost'] } }) + const { cache, persistence } = await harness({ pool, logs }) + const snapshot = await cache.coldSnapshot(SessionId('shrunk')) + expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] }) + expect(snapshot.asOfSeq).toBe(2) + // Anchored tail read (floor 9) came back empty -> full re-read from 0. + expect(persistence.readFrom).toHaveBeenNthCalledWith(1, SessionId('shrunk'), 9, undefined) + expect(persistence.readFrom).toHaveBeenNthCalledWith(2, SessionId('shrunk'), 0, undefined) + }) + + it('write-back failure is contained: the snapshot is still served', async () => { + const pool = new MemoryMediaPool() + const logs = new Map([['soft', storedLog([['a']])]]) + const { ctx, cache } = await harness({ pool, logs }) + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {}) + pool.failNextWrites = 1 + const snapshot = await cache.coldSnapshot(SessionId('soft')) + expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] }) + expect(warn).toHaveBeenCalledWith(expect.stringContaining('cold-read write-back for "soft" failed')) + }) + + it('rejects for a session with no persisted log', async () => { + const { cache } = await harness() + await expect(cache.coldSnapshot(SessionId('absent'))).rejects.toThrow('not found') + }) +}) diff --git a/packages/session-projection/session-projection-cache/tsconfig.json b/packages/session-projection/session-projection-cache/tsconfig.json new file mode 100644 index 0000000000..baba6c805b --- /dev/null +++ b/packages/session-projection/session-projection-cache/tsconfig.json @@ -0,0 +1,39 @@ +{ + "extends": "../../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "lib/types" + }, + "include": [ + "src" + ], + "references": [ + { + "path": "../../../vendor/cosmokit" + }, + { + "path": "../../../vendor/cordis" + }, + { + "path": "../../../vendor/schemastery" + }, + { + "path": "../../core/session" + }, + { + "path": "../../session-persistence/session-persistence" + }, + { + "path": "../session-projection" + }, + { + "path": "../../storage/storage" + }, + { + "path": "../../storage/storage-domain" + }, + { + "path": "../../support/invariants" + } + ] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f5309a6838..7e189366cb 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -239,6 +239,9 @@ importers: '@deepseek-ai/dsh-session-projection': specifier: workspace:^ version: link:../../packages/session-projection/session-projection + '@deepseek-ai/dsh-session-projection-cache': + specifier: workspace:^ + version: link:../../packages/session-projection/session-projection-cache '@deepseek-ai/dsh-session-title': specifier: workspace:^ version: link:../../packages/session-title/session-title @@ -3460,6 +3463,40 @@ importers: specifier: ^4.0.0-rc.7 version: 4.0.0-rc.7(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.5) + packages/session-projection/session-projection-cache: + dependencies: + schemastery: + specifier: ^3.18.0 + version: 3.18.0 + zod: + specifier: ^4.4.3 + version: 4.4.3 + devDependencies: + '@deepseek-ai/dsh-invariants': + specifier: workspace:^ + version: link:../../support/invariants + '@deepseek-ai/dsh-session': + specifier: workspace:^ + version: link:../../core/session + '@deepseek-ai/dsh-session-persistence': + specifier: workspace:^ + version: link:../../session-persistence/session-persistence + '@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 + cordis: + specifier: ^4.0.0-rc.7 + version: 4.0.0-rc.7(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.5) + packages/session-query/session-query: devDependencies: '@deepseek-ai/dsh-brand': diff --git a/tsconfig.host.json b/tsconfig.host.json index 0b4a30f0b6..d211330396 100644 --- a/tsconfig.host.json +++ b/tsconfig.host.json @@ -58,6 +58,7 @@ { "path": "./packages/session-persistence/session-persistence-jsonl" }, { "path": "./packages/session-persistence/session-persistence-sqlite" }, { "path": "./packages/session-projection/session-projection" }, + { "path": "./packages/session-projection/session-projection-cache" }, { "path": "./packages/session-query/session-query" }, { "path": "./packages/session-query/session-query-sqlite" }, { "path": "./packages/session-query/tool-session-query" },