diff --git a/packages/agent-core-v2/package.json b/packages/agent-core-v2/package.json index a02a0698e..9725df441 100644 --- a/packages/agent-core-v2/package.json +++ b/packages/agent-core-v2/package.json @@ -64,6 +64,7 @@ "@modelcontextprotocol/sdk": "^1.29.0", "@moonshot-ai/kimi-code-oauth": "workspace:^", "@moonshot-ai/kimi-telemetry": "workspace:^", + "@moonshot-ai/minidb": "workspace:^", "@moonshot-ai/protocol": "workspace:^", "@mozilla/readability": "^0.6.0", "chokidar": "^4.0.3", diff --git a/packages/agent-core-v2/scripts/check-domain-layers.mjs b/packages/agent-core-v2/scripts/check-domain-layers.mjs index 8d82a72b0..ae827d7ef 100644 --- a/packages/agent-core-v2/scripts/check-domain-layers.mjs +++ b/packages/agent-core-v2/scripts/check-domain-layers.mjs @@ -324,6 +324,11 @@ const ALLOWED_EXCEPTIONS = new Set([ 'terminal>os/backends', 'sessionFs>os/backends', 'blobStore>persistence/backends', + // `sessionIndex` (L2) reads the `persistence_minidb_readmodel` experimental + // flag (L3) to switch session listings between the legacy N+1 disk read and + // the minidb-backed derived read model. A genuine, planned upward dependency + // on a cross-cutting capability switch — surfaced here for review. + 'sessionIndex>flag', ]); // Matches: import ... from 'x' | export ... from 'x' | import('x') | require('x') diff --git a/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts b/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts index 8ae63834b..05f01db70 100644 --- a/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts +++ b/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts @@ -14,6 +14,16 @@ * sessions written before the layouts were unified. Both timestamp * representations are normalized to epoch ms. * + * Read model (flag `persistence_minidb_readmodel`): when enabled, summaries are + * served from the `IQueryStore` derived read model instead of re-reading and + * re-parsing `state.json` on every call. Listing still enumerates the directory + * (a cheap `readdir`) to discover `(workspaceId, sessionId)` pairs, but each + * summary is resolved through the read model — falling back to a disk read + + * backfill on a cold miss. Writes (create / archive / metadata update) keep the + * read model warm via `SessionMetadata`; new sessions that have not been + * mirrored yet are simply a cold miss and backfilled on first read. The legacy + * N+1 path remains as the flag-off fallback. + * * This is the local-deployment backend of `ISessionIndex`; a server deployment * would substitute a database-backed `DbSessionIndex`. Bound at App scope. */ @@ -21,14 +31,17 @@ import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { IBootstrapService } from '#/app/bootstrap'; +import { IFlagService } from '#/app/flag'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; +import { IQueryStore, type Page } from '#/persistence/interface/queryStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; -import type { Page } from '#/persistence/interface/queryStore'; import { ISessionIndex, type SessionListQuery, type SessionSummary } from './sessionIndex'; const META_SCOPE = 'session-meta'; const META_KEY = 'state.json'; +const SESSION_COLLECTION = 'session'; +const READ_MODEL_FLAG = 'persistence_minidb_readmodel'; /** Accept both v2 (epoch ms number) and v1 (ISO string) timestamps. */ function parseTime(value: unknown): number { @@ -43,13 +56,20 @@ function parseTime(value: unknown): number { export class FileSessionIndex implements ISessionIndex { declare readonly _serviceBrand: undefined; + private indexesEnsured = false; + constructor( @IBootstrapService private readonly bootstrap: IBootstrapService, @IFileSystemStorageService private readonly storage: IFileSystemStorageService, @IAtomicDocumentStore private readonly docs: IAtomicDocumentStore, + @IQueryStore private readonly queryStore: IQueryStore, + @IFlagService private readonly flags: IFlagService, ) {} async list(query: SessionListQuery): Promise> { + if (!this.readModelEnabled()) return this.listLegacy(query); + + await this.ensureIndexes(); if (query.sessionId !== undefined) { const summary = await this.get(query.sessionId); const items = @@ -59,6 +79,92 @@ export class FileSessionIndex implements ISessionIndex { return { items: query.limit !== undefined ? items.slice(0, query.limit) : items }; } + const workspaceIds = + query.workspaceId !== undefined ? [query.workspaceId] : await this.listWorkspaceIds(); + const items: SessionSummary[] = []; + for (const workspaceId of workspaceIds) { + for (const sessionId of await this.listSessionIds(workspaceId)) { + const summary = await this.getCachedSummary(workspaceId, sessionId); + if (summary === undefined) continue; + if (summary.archived && query.includeArchived !== true) continue; + items.push(summary); + } + } + items.sort((a, b) => b.updatedAt - a.updatedAt); + return { items: query.limit !== undefined ? items.slice(0, query.limit) : items }; + } + + async get(id: string): Promise { + if (!this.readModelEnabled()) return this.getLegacy(id); + + const cached = await this.queryStore.get(SESSION_COLLECTION, id); + if (cached !== undefined) return cached; + // Cold miss: locate the session on disk, then read + backfill. + for (const workspaceId of await this.listWorkspaceIds()) { + if (!(await this.hasSession(workspaceId, id))) continue; + return this.getCachedSummary(workspaceId, id); + } + return undefined; + } + + async countActive(workspaceId: string): Promise { + if (!this.readModelEnabled()) return this.countActiveLegacy(workspaceId); + + let count = 0; + for (const sessionId of await this.listSessionIds(workspaceId)) { + const summary = await this.getCachedSummary(workspaceId, sessionId); + if (summary !== undefined && !summary.archived) count += 1; + } + return count; + } + + private readModelEnabled(): boolean { + return this.flags.enabled(READ_MODEL_FLAG); + } + + private async ensureIndexes(): Promise { + if (this.indexesEnsured) return; + await this.queryStore.ensureIndex(SESSION_COLLECTION, { + kind: 'value', + name: 'byWorkspace', + field: 'workspaceId', + }); + await this.queryStore.ensureIndex(SESSION_COLLECTION, { + kind: 'compound', + name: 'byWsUpdated', + groupBy: 'workspaceId', + orderBy: 'updatedAt', + }); + this.indexesEnsured = true; + } + + /** + * Resolve a summary through the read model, backfilling from disk on a cold + * miss. The read model is keyed by session id (globally unique). + */ + private async getCachedSummary( + workspaceId: string, + sessionId: string, + ): Promise { + const cached = await this.queryStore.get(SESSION_COLLECTION, sessionId); + if (cached !== undefined) return cached; + const summary = await this.readSummary(workspaceId, sessionId); + if (summary !== undefined) { + await this.queryStore.put(SESSION_COLLECTION, sessionId, summary); + } + return summary; + } + + private async listLegacy(query: SessionListQuery): Promise> { + if (query.sessionId !== undefined) { + const summary = await this.getLegacy(query.sessionId); + const items = + summary !== undefined && (!summary.archived || query.includeArchived === true) + ? [summary] + : []; + return { items: query.limit !== undefined ? items.slice(0, query.limit) : items }; + } + const workspaceIds = query.workspaceId !== undefined ? [query.workspaceId] : await this.listWorkspaceIds(); const items: SessionSummary[] = []; @@ -74,7 +180,7 @@ export class FileSessionIndex implements ISessionIndex { return { items: query.limit !== undefined ? items.slice(0, query.limit) : items }; } - async get(id: string): Promise { + private async getLegacy(id: string): Promise { for (const workspaceId of await this.listWorkspaceIds()) { if (!(await this.hasSession(workspaceId, id))) continue; const summary = await this.readSummary(workspaceId, id); @@ -83,7 +189,7 @@ export class FileSessionIndex implements ISessionIndex { return undefined; } - async countActive(workspaceId: string): Promise { + private async countActiveLegacy(workspaceId: string): Promise { let count = 0; for (const sessionId of await this.listSessionIds(workspaceId)) { const summary = await this.readSummary(workspaceId, sessionId); diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index 55f9ab4d4..ab5a24eb5 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -11,9 +11,6 @@ export { IAgentWireService, ISessionWireService, type IWireService, - type Signal, - type SignalMap, - type WireEmission, } from '#/wire'; export * from '#/session/sessionLog'; export * from '#/app/telemetry'; @@ -22,7 +19,7 @@ export * from '#/os/interface'; export * from '#/os/backends/node-local'; export * from '#/session/terminal'; export * from '#/app/task'; -export { IEventService, type DomainEvent } from '#/app/event'; +export { IEventService, IEventBus, type DomainEvent, type GlobalEvent } from '#/app/event'; export * from '#/app/llmProtocol'; export * from '#/app/sessionIndex'; @@ -75,6 +72,7 @@ export * from '#/session/sessionFs'; export * from '#/app/hostFolderBrowser'; export * from '#/persistence/interface'; export * from '#/persistence/backends/node-fs'; +export * from '#/persistence/backends/minidb'; export * from '#/persistence/backends/memory'; export * from '#/app/auth'; export * from '#/app/authLegacy'; diff --git a/packages/agent-core-v2/src/persistence/backends/minidb/flag.ts b/packages/agent-core-v2/src/persistence/backends/minidb/flag.ts new file mode 100644 index 000000000..7703bd8c9 --- /dev/null +++ b/packages/agent-core-v2/src/persistence/backends/minidb/flag.ts @@ -0,0 +1,23 @@ +/** + * `minidb` persistence backend — flag contribution. + * + * Gates the minidb-backed derived read-model (`IQueryStore`) and the consumers + * that read through it. Off by default; enable via + * `KIMI_CODE_EXPERIMENTAL_PERSISTENCE_MINIDB_READMODEL` or the `[experimental]` + * config section. Imported for its side effect (registers the definition) from + * the backend barrel. + */ + +import { type FlagDefinitionInput, registerFlagDefinition } from '#/app/flag'; + +export const persistenceMiniDbReadModelFlag: FlagDefinitionInput = { + id: 'persistence_minidb_readmodel', + title: 'minidb read model', + description: + 'Use the minidb-backed IQueryStore as a derived read model for session indexing and wire replay.', + env: 'KIMI_CODE_EXPERIMENTAL_PERSISTENCE_MINIDB_READMODEL', + default: false, + surface: 'core', +}; + +registerFlagDefinition(persistenceMiniDbReadModelFlag); diff --git a/packages/agent-core-v2/src/persistence/backends/minidb/index.ts b/packages/agent-core-v2/src/persistence/backends/minidb/index.ts new file mode 100644 index 000000000..bcc2b83c5 --- /dev/null +++ b/packages/agent-core-v2/src/persistence/backends/minidb/index.ts @@ -0,0 +1,3 @@ +import './flag'; + +export * from './miniDbQueryStore'; diff --git a/packages/agent-core-v2/src/persistence/backends/minidb/miniDbQueryStore.ts b/packages/agent-core-v2/src/persistence/backends/minidb/miniDbQueryStore.ts new file mode 100644 index 000000000..e993676a8 --- /dev/null +++ b/packages/agent-core-v2/src/persistence/backends/minidb/miniDbQueryStore.ts @@ -0,0 +1,231 @@ +/** + * `minidb` backend — `IQueryStore` implementation over `MiniDb`. + * + * A rebuildable, in-process derived read-model. `MiniDb` is opened with + * `openOrRebuild`, so on-disk corruption becomes a clean rebuild rather than a + * hard failure: authoritative data lives in `IAppendLogStore` / + * `IAtomicDocumentStore`, never here, so losing the read model is always safe. + * Values are JSON (`valueCodec: 'json'`, required by secondary indexes and + * `query`) and held in memory (`valueMode: 'memory'`); durability is `everysec`, + * which is acceptable for a cache. The store is rooted at + * `/query-store`. + * + * The database is opened **lazily** on the first actual IO, not at construction. + * Construction therefore does no filesystem work and never touches the single + * writer lock — important because `MiniDbQueryStore` is resolved transitively + * whenever a consumer (e.g. `SessionMetadata`) is constructed, including in + * tests that share a home dir and never read or write the read model. Only a + * real `put`/`get`/`query`/... opens the database. + * + * A `collection` is encoded as a key prefix (`\u0000`); indexes + * are global to the `MiniDb` instance, so index names are prefixed with the + * collection to keep them isolated, and value indexes are created `sparse` so + * documents from other collections (which lack the indexed field) are skipped. + * + * Bound at App scope as a peer of the other access-pattern stores. + */ + +import { join } from 'pathe'; + +import { MiniDb, type QueryOptions } from '@moonshot-ai/minidb'; + +import { InstantiationType } from '#/_base/di/extensions'; +import { Disposable, toDisposable } from '#/_base/di/lifecycle'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { ILogService } from '#/_base/log'; +import { IBootstrapService } from '#/app/bootstrap'; +import { + IQueryStore, + type Checkpoint, + type IndexDef, + type IQuery, + type Page, + type QueryFilter, + type SortDir, + type WriteOp, +} from '#/persistence/interface/queryStore'; + +const SEP = String.fromCodePoint(0); +const CHECKPOINT_COLLECTION = '__checkpoint__'; +const STORE_SUBDIR = 'query-store'; + +function physicalKey(collection: string, key: string): string { + return `${collection}${SEP}${key}`; +} + +function indexName(collection: string, name: string): string { + return `${collection}:${name}`; +} + +export class MiniDbQueryStore extends Disposable implements IQueryStore { + declare readonly _serviceBrand: undefined; + + private readonly dir: string; + private dbPromise: Promise | undefined; + private readonly ensuredIndexes = new Set(); + + constructor( + @IBootstrapService private readonly bootstrap: IBootstrapService, + @ILogService private readonly log: ILogService, + ) { + super(); + this.dir = join(this.bootstrap.cacheDir, STORE_SUBDIR); + this._register(toDisposable(() => { + void this.close(); + })); + } + + private openDb(): Promise { + if (this.dbPromise !== undefined) return this.dbPromise; + this.dbPromise = MiniDb.openOrRebuild( + { + dir: this.dir, + valueCodec: 'json', + valueMode: 'memory', + fsyncPolicy: 'everysec', + }, + { + onRebuild: (err) => { + this.log.warn('minidb query-store rebuilt after corruption', { + dir: this.dir, + error: String(err), + }); + }, + }, + ); + return this.dbPromise; + } + + async put(collection: string, key: string, value: T): Promise { + const db = await this.openDb(); + await db.set(physicalKey(collection, key), value); + } + + async batch(ops: readonly WriteOp[]): Promise { + if (ops.length === 0) return; + const db = await this.openDb(); + await db.batch( + ops.map((op) => + op.kind === 'put' + ? { op: 'set' as const, key: physicalKey(op.collection, op.key), value: op.value } + : { op: 'del' as const, key: physicalKey(op.collection, op.key) }, + ), + ); + } + + async delete(collection: string, key: string): Promise { + const db = await this.openDb(); + await db.del(physicalKey(collection, key)); + } + + async get(collection: string, key: string): Promise { + const db = await this.openDb(); + return db.get(physicalKey(collection, key)) as T | undefined; + } + + query(collection: string): IQuery { + return new MiniDbQuery(() => this.openDb(), collection); + } + + async ensureIndex(collection: string, def: IndexDef): Promise { + const guard = `${collection}:${def.kind}:${def.name}`; + if (this.ensuredIndexes.has(guard)) return; + const db = await this.openDb(); + const name = indexName(collection, def.name); + if (def.kind === 'value') { + if (!db.listIndexes().some((i) => i.name === name)) { + await db.createIndex(name, { field: def.field, sparse: true, unique: def.unique }); + } + } else if (def.kind === 'compound') { + if (!db.listCompoundIndexes().some((i) => i.name === name)) { + await db.createCompoundIndex(name, { groupBy: def.groupBy, orderBy: def.orderBy }); + } + } else { + // A text index that already exists (rebuilt from persisted definitions on + // reopen) makes `createTextIndex` throw; treat that as already-ensured. + try { + await db.createTextIndex(name, { fields: def.fields }); + } catch (error) { + if (!(error instanceof Error) || !error.message.includes('already exists')) throw error; + } + } + this.ensuredIndexes.add(guard); + } + + async getCheckpoint(source: string): Promise { + return this.get(CHECKPOINT_COLLECTION, source); + } + + async setCheckpoint(source: string, checkpoint: Checkpoint): Promise { + await this.put(CHECKPOINT_COLLECTION, source, checkpoint); + } + + async close(): Promise { + if (this.dbPromise === undefined) return; + const db = await this.dbPromise; + await db.close(); + } +} + +class MiniDbQuery implements IQuery { + private filter: QueryFilter = {}; + private sortField?: string; + private sortDir: SortDir = 'asc'; + private lim?: number; + private skip = 0; + + constructor( + private readonly openDb: () => Promise, + private readonly collection: string, + ) {} + + where(filter: QueryFilter): IQuery { + this.filter = { ...this.filter, ...filter }; + return this; + } + + orderBy(field: string, dir: SortDir = 'asc'): IQuery { + this.sortField = field; + this.sortDir = dir; + return this; + } + + limit(n: number): IQuery { + this.lim = n; + return this; + } + + cursor(cursor: string | undefined): IQuery { + this.skip = cursor !== undefined && cursor.length > 0 ? Number(cursor) : 0; + return this; + } + + async execute(): Promise> { + const db = await this.openDb(); + const prefix = `${this.collection}${SEP}`; + const q: QueryOptions = { key: { prefix } }; + if (Object.keys(this.filter).length > 0) q.filter = this.filter as Record; + if (this.sortField !== undefined) { + q.sort = { [this.sortField]: this.sortDir === 'desc' ? -1 : 1 }; + } + q.skip = this.skip; + // Fetch one extra row to know whether a next page exists. + if (this.lim !== undefined) q.limit = this.lim + 1; + const rows = db.query(q) as ReadonlyArray<{ key: string; value: T }>; + let items = rows.map((r) => r.value); + let nextCursor: string | undefined; + if (this.lim !== undefined && items.length > this.lim) { + items = items.slice(0, this.lim); + nextCursor = String(this.skip + this.lim); + } + return { items, nextCursor }; + } +} + +registerScopedService( + LifecycleScope.App, + IQueryStore, + MiniDbQueryStore, + InstantiationType.Delayed, + 'storage', +); diff --git a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts index f5c74dcba..efb79dc99 100644 --- a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts +++ b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts @@ -6,6 +6,14 @@ * namespace from `sessionContext`. Loads the existing document on * construction (creating it on first run), and logs through `log`. Bound at * Session scope. + * + * Read-model mirroring (flag `persistence_minidb_readmodel`): after a metadata + * update is persisted, the fresh summary is mirrored into the `IQueryStore` + * derived read model so `FileSessionIndex` can serve listings without + * re-reading `state.json`. Mirroring is best-effort (a failure is logged, not + * thrown) and is a no-op when the flag is off. Initial creation in `load()` is + * intentionally not mirrored — a not-yet-mirrored session is simply a cold + * read-model miss that `FileSessionIndex` backfills on first read. */ import { InstantiationType } from '#/_base/di/extensions'; @@ -13,8 +21,10 @@ import { Disposable } from '#/_base/di/lifecycle'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { Emitter, type Event } from '#/_base/event'; import { ILogService } from '#/_base/log'; -import { ISessionContext } from '#/session/sessionContext'; +import { IFlagService } from '#/app/flag'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; +import { IQueryStore } from '#/persistence/interface/queryStore'; +import { ISessionContext } from '#/session/sessionContext'; import { ISessionMetadata, @@ -26,6 +36,8 @@ import { } from './sessionMetadata'; const META_KEY = 'state.json'; +const SESSION_COLLECTION = 'session'; +const READ_MODEL_FLAG = 'persistence_minidb_readmodel'; export class SessionMetadata extends Disposable implements ISessionMetadata { declare readonly _serviceBrand: undefined; @@ -42,6 +54,8 @@ export class SessionMetadata extends Disposable implements ISessionMetadata { @ISessionContext private readonly ctx: ISessionContext, @IAtomicDocumentStore private readonly store: IAtomicDocumentStore, @ILogService private readonly log: ILogService, + @IQueryStore private readonly queryStore: IQueryStore, + @IFlagService private readonly flags: IFlagService, ) { super(); this.scope = ctx.metaScope; @@ -58,6 +72,7 @@ export class SessionMetadata extends Disposable implements ISessionMetadata { await this.ready; this.data = { ...this.data, ...patch, updatedAt: Date.now() }; await this.store.set(this.scope, META_KEY, this.data); + await this.mirrorToReadModel(); this._onDidChangeMetadata.fire({ changed: Object.keys(patch) as (keyof SessionMeta)[], }); @@ -73,10 +88,31 @@ export class SessionMetadata extends Disposable implements ISessionMetadata { async registerAgent(agentId: string, meta: AgentMeta): Promise { await this.ready; - const agents = { ...(this.data.agents ?? {}), [agentId]: meta }; + const agents = { ...this.data.agents, [agentId]: meta }; await this.update({ agents }); } + private async mirrorToReadModel(): Promise { + if (!this.flags.enabled(READ_MODEL_FLAG)) return; + try { + await this.queryStore.put(SESSION_COLLECTION, this.ctx.sessionId, { + id: this.data.id, + workspaceId: this.ctx.workspaceId, + title: this.data.title, + lastPrompt: this.data.lastPrompt, + createdAt: this.data.createdAt, + updatedAt: this.data.updatedAt, + archived: this.data.archived, + custom: this.data.custom, + }); + } catch (error) { + this.log.warn('failed to mirror session metadata to read model', { + sessionId: this.ctx.sessionId, + error: String(error), + }); + } + } + private async load(): Promise { const existing = await this.store.get(this.scope, META_KEY); if (existing !== undefined) { diff --git a/packages/agent-core-v2/test/flag/stubs.ts b/packages/agent-core-v2/test/flag/stubs.ts new file mode 100644 index 000000000..b37954f3e --- /dev/null +++ b/packages/agent-core-v2/test/flag/stubs.ts @@ -0,0 +1,37 @@ +/** + * `flag` test stubs — minimal `IFlagService` for unit tests. + * + * Lives under `test/` (not `src/`). Import from a relative path. + */ + +import { IFlagService } from '#/app/flag/flag'; +import type { + ExperimentalFeatureState, + ExperimentalFlagConfig, + ExperimentalFlagMap, +} from '#/app/flag/flag'; +import type { IFlagRegistry } from '#/app/flag/flagRegistry'; + +/** + * A minimal `IFlagService`. `enabled` is either a fixed boolean or a per-id + * predicate; everything else is a no-op / empty. + */ +export function stubFlag(enabled: boolean | ((id: string) => boolean) = false): IFlagService { + const isEnabled = typeof enabled === 'function' ? enabled : (): boolean => enabled; + const registry: IFlagRegistry = { + _serviceBrand: undefined, + register: () => ({ dispose: () => {} }), + get: () => undefined, + list: () => [], + }; + return { + _serviceBrand: undefined, + registry, + enabled: isEnabled, + snapshot: (): ExperimentalFlagMap => ({}), + enabledIds: () => [], + explain: (): ExperimentalFeatureState | undefined => undefined, + explainAll: () => [], + setConfigOverrides: (_overrides: ExperimentalFlagConfig | undefined) => {}, + }; +} diff --git a/packages/agent-core-v2/test/persistence/miniDbQueryStore.test.ts b/packages/agent-core-v2/test/persistence/miniDbQueryStore.test.ts new file mode 100644 index 000000000..464039aff --- /dev/null +++ b/packages/agent-core-v2/test/persistence/miniDbQueryStore.test.ts @@ -0,0 +1,150 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import { promises as fsp } from 'node:fs'; +import os from 'node:os'; +import { join } from 'node:path'; + +import { InstantiationType } from '#/_base/di/extensions'; +import { LifecycleScope, _clearScopedRegistryForTests, registerScopedService } from '#/_base/di/scope'; +import { createScopedTestHost, stubPair } from '#/_base/di/test'; +import { ILogService } from '#/_base/log'; +import { IBootstrapService } from '#/app/bootstrap'; +import { MiniDbQueryStore } from '#/persistence/backends/minidb/miniDbQueryStore'; +import { IQueryStore } from '#/persistence/interface/queryStore'; +import { stubBootstrap } from '../bootstrap/stubs'; +import { stubLog } from '../log/stubs'; + +const COLLECTION = 'session'; + +describe('MiniDbQueryStore', () => { + let homeDir: string; + let disposeHost: (() => void) | undefined; + + beforeEach(async () => { + _clearScopedRegistryForTests(); + registerScopedService( + LifecycleScope.App, + IQueryStore, + MiniDbQueryStore, + InstantiationType.Delayed, + 'storage', + ); + homeDir = await fsp.mkdtemp(join(os.tmpdir(), 'minidb-qs-')); + }); + + afterEach(async () => { + disposeHost?.(); + disposeHost = undefined; + await fsp.rm(homeDir, { recursive: true, force: true }); + }); + + function build(): IQueryStore { + const host = createScopedTestHost([ + stubPair(IBootstrapService, stubBootstrap(homeDir)), + stubPair(ILogService, stubLog()), + ]); + disposeHost = () => { host.dispose(); }; + return host.app.accessor.get(IQueryStore); + } + + it('put/get/delete round-trip', async () => { + const store = build(); + await store.put(COLLECTION, 'a', { id: 'a', v: 1 }); + expect(await store.get(COLLECTION, 'a')).toEqual({ id: 'a', v: 1 }); + expect(await store.get(COLLECTION, 'missing')).toBeUndefined(); + await store.delete(COLLECTION, 'a'); + expect(await store.get(COLLECTION, 'a')).toBeUndefined(); + }); + + it('batch applies put and delete atomically', async () => { + const store = build(); + await store.batch([ + { kind: 'put', collection: COLLECTION, key: 'a', value: { v: 1 } }, + { kind: 'put', collection: COLLECTION, key: 'b', value: { v: 2 } }, + ]); + expect(await store.get(COLLECTION, 'a')).toEqual({ v: 1 }); + await store.batch([{ kind: 'delete', collection: COLLECTION, key: 'a' }]); + expect(await store.get(COLLECTION, 'a')).toBeUndefined(); + expect(await store.get(COLLECTION, 'b')).toEqual({ v: 2 }); + }); + + it('isolates collections by prefix', async () => { + const store = build(); + await store.batch([ + { kind: 'put', collection: 'c1', key: 'k', value: { v: 1 } }, + { kind: 'put', collection: 'c2', key: 'k', value: { v: 2 } }, + ]); + expect(await store.get('c1', 'k')).toEqual({ v: 1 }); + expect(await store.get('c2', 'k')).toEqual({ v: 2 }); + }); + + it('query filters, orders, limits and paginates with cursor', async () => { + const store = build(); + await store.ensureIndex(COLLECTION, { kind: 'value', name: 'byWs', field: 'ws' }); + await store.batch( + ( + [ + ['a', 'x', 1], + ['b', 'x', 3], + ['c', 'y', 5], + ['d', 'x', 2], + ] as const + ).map(([id, ws, n]) => ({ kind: 'put' as const, collection: COLLECTION, key: id, value: { id, ws, n } })), + ); + + const page1 = await store + .query<{ id: string; ws: string; n: number }>(COLLECTION) + .where({ ws: 'x' }) + .orderBy('n', 'desc') + .limit(2) + .execute(); + expect(page1.items.map((i) => i.id)).toEqual(['b', 'd']); + expect(page1.nextCursor).toBe('2'); + + const page2 = await store + .query<{ id: string; ws: string; n: number }>(COLLECTION) + .where({ ws: 'x' }) + .orderBy('n', 'desc') + .limit(2) + .cursor(page1.nextCursor) + .execute(); + expect(page2.items.map((i) => i.id)).toEqual(['a']); + expect(page2.nextCursor).toBeUndefined(); + }); + + it('ensureIndex is idempotent across value, compound and text kinds', async () => { + const store = build(); + await store.put(COLLECTION, 'a', { id: 'a', ws: 'x', n: 1, body: 'hello world' }); + await store.ensureIndex(COLLECTION, { kind: 'value', name: 'byWs', field: 'ws' }); + await store.ensureIndex(COLLECTION, { kind: 'value', name: 'byWs', field: 'ws' }); + await store.ensureIndex(COLLECTION, { kind: 'compound', name: 'byWsN', groupBy: 'ws', orderBy: 'n' }); + await store.ensureIndex(COLLECTION, { kind: 'text', name: 'body', fields: ['body'] }); + await store.ensureIndex(COLLECTION, { kind: 'text', name: 'body', fields: ['body'] }); + const page = await store.query(COLLECTION).where({ ws: 'x' }).execute(); + expect(page.items).toHaveLength(1); + }); + + it('stores checkpoints', async () => { + const store = build(); + expect(await store.getCheckpoint('wire:abc')).toBeUndefined(); + await store.setCheckpoint('wire:abc', { seq: 42 }); + expect(await store.getCheckpoint('wire:abc')).toEqual({ seq: 42 }); + }); + + it('rebuilds a clean store after on-disk corruption', async () => { + const first = build(); + await first.put(COLLECTION, 'a', { v: 1 }); + await first.ensureIndex(COLLECTION, { kind: 'value', name: 'byV', field: 'v' }); + await first.close(); + disposeHost?.(); + disposeHost = undefined; + + // Corrupt the persisted index-definition JSON so `open` throws SyntaxError, + // which `openOrRebuild` turns into a wipe + fresh open. + const indexFile = join(homeDir, 'cache', 'query-store', 'db.indexes.json'); + await fsp.writeFile(indexFile, '{ definitely not valid json'); + + const second = build(); + expect(await second.get(COLLECTION, 'a')).toBeUndefined(); + }); +}); diff --git a/packages/agent-core-v2/test/persistence/stubs.ts b/packages/agent-core-v2/test/persistence/stubs.ts new file mode 100644 index 000000000..06d72f15a --- /dev/null +++ b/packages/agent-core-v2/test/persistence/stubs.ts @@ -0,0 +1,40 @@ +/** + * `persistence` test stubs — minimal no-op `IQueryStore` for unit tests. + * + * Lives under `test/` (not `src/`). Import from a relative path. + */ + +import { + IQueryStore, + type Checkpoint, + type IQuery, + type Page, +} from '#/persistence/interface/queryStore'; + +/** A no-op `IQueryStore`: every read is empty / undefined, every write is dropped. */ +export function stubQueryStore(): IQueryStore { + return { + _serviceBrand: undefined, + put: async (_c: string, _k: string, _v: unknown) => {}, + batch: async (_ops) => {}, + delete: async (_c: string, _k: string) => {}, + get: async (_c: string, _k: string) => undefined as T | undefined, + query: (_c: string) => emptyQuery(), + ensureIndex: async (_c, _d) => {}, + getCheckpoint: async (_s: string) => undefined as Checkpoint | undefined, + setCheckpoint: async (_s: string, _c: Checkpoint) => {}, + close: async () => {}, + }; +} + +function emptyQuery(): IQuery { + const page: Page = { items: [] }; + const q: IQuery = { + where: () => q, + orderBy: () => q, + limit: () => q, + cursor: () => q, + execute: () => Promise.resolve(page), + }; + return q; +} diff --git a/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts b/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts index 174731bcc..1fc277141 100644 --- a/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts +++ b/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts @@ -5,21 +5,29 @@ import os from 'node:os'; import { join } from 'node:path'; import { InstantiationType } from '#/_base/di/extensions'; +import { ILogService } from '#/_base/log'; import { LifecycleScope, _clearScopedRegistryForTests, registerScopedService } from '#/_base/di/scope'; import { createScopedTestHost, stubPair } from '#/_base/di/test'; import { encodeWorkDirKey } from '#/_base/utils/workdir-slug'; import { IBootstrapService } from '#/app/bootstrap'; -import { ISessionIndex } from '#/app/sessionIndex/sessionIndex'; +import { IFlagService } from '#/app/flag'; +import { ISessionIndex, type SessionSummary } from '#/app/sessionIndex/sessionIndex'; import { FileSessionIndex } from '#/app/sessionIndex/sessionIndexService'; -import { stubBootstrap } from '../bootstrap/stubs'; +import { MiniDbQueryStore } from '#/persistence/backends/minidb/miniDbQueryStore'; import { JsonAtomicDocumentStore } from '#/persistence/backends/node-fs/atomicDocumentStore'; import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; +import { IQueryStore } from '#/persistence/interface/queryStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; +import { stubBootstrap } from '../bootstrap/stubs'; +import { stubFlag } from '../flag/stubs'; +import { stubLog } from '../log/stubs'; +import { stubQueryStore } from '../persistence/stubs'; const WORK_DIR = '/home/user/repo'; +const SESSION_COLLECTION = 'session'; -describe('FileSessionIndex', () => { +describe('FileSessionIndex (legacy)', () => { let homeDir: string; let sessionsDir: string; let workspaceId: string; @@ -45,8 +53,10 @@ describe('FileSessionIndex', () => { stubPair(IFileSystemStorageService, fileStorage), stubPair(IAtomicDocumentStore, new JsonAtomicDocumentStore(fileStorage)), stubPair(IBootstrapService, stubBootstrap(homeDir)), + stubPair(IQueryStore, stubQueryStore()), + stubPair(IFlagService, stubFlag(false)), ]); - disposeHost = () => host.dispose(); + disposeHost = () => { host.dispose(); }; return host.app.accessor.get(ISessionIndex); } @@ -121,3 +131,97 @@ describe('FileSessionIndex', () => { expect(await store.countActive('wd_unknown')).toBe(0); }); }); + +describe('FileSessionIndex (read model)', () => { + let homeDir: string; + let sessionsDir: string; + let workspaceId: string; + let disposeHost: (() => void) | undefined; + let queryStore: IQueryStore; + + beforeEach(async () => { + _clearScopedRegistryForTests(); + registerScopedService(LifecycleScope.App, ISessionIndex, FileSessionIndex, InstantiationType.Delayed, 'sessionIndex'); + registerScopedService(LifecycleScope.App, IQueryStore, MiniDbQueryStore, InstantiationType.Delayed, 'storage'); + homeDir = await fsp.mkdtemp(join(os.tmpdir(), 'ws-sessions-rm-')); + sessionsDir = join(homeDir, 'sessions'); + workspaceId = encodeWorkDirKey(WORK_DIR); + }); + + afterEach(async () => { + disposeHost?.(); + disposeHost = undefined; + await fsp.rm(homeDir, { recursive: true, force: true }); + }); + + function build(): ISessionIndex { + const fileStorage = new FileStorageService(homeDir); + const host = createScopedTestHost([ + stubPair(IFileSystemStorageService, fileStorage), + stubPair(IAtomicDocumentStore, new JsonAtomicDocumentStore(fileStorage)), + stubPair(IBootstrapService, stubBootstrap(homeDir)), + stubPair(ILogService, stubLog()), + stubPair(IFlagService, stubFlag(true)), + ]); + disposeHost = () => { host.dispose(); }; + queryStore = host.app.accessor.get(IQueryStore); + return host.app.accessor.get(ISessionIndex); + } + + async function seedSession( + sessionId: string, + meta: Record, + wsId: string = workspaceId, + ): Promise { + const dir = join(sessionsDir, wsId, sessionId, 'session-meta'); + await fsp.mkdir(dir, { recursive: true }); + await fsp.writeFile(join(dir, 'state.json'), JSON.stringify(meta)); + } + + function summary(id: string, overrides: Partial = {}): SessionSummary { + return { + id, + workspaceId, + createdAt: 1, + updatedAt: 2, + archived: false, + ...overrides, + }; + } + + it('list backfills from disk on a cold read model, then serves from it', async () => { + await seedSession('active', { title: 'hello', createdAt: 1, updatedAt: 2 }); + await seedSession('archived', { archived: true }); + + const store = build(); + const first = await store.list({ workspaceId }); + expect(first.items.map((s) => s.id)).toEqual(['active']); + expect(first.items[0]?.title).toBe('hello'); + + // A second list is served from the read model: mutate the read model to + // prove the disk is not re-read. + await queryStore.put(SESSION_COLLECTION, 'active', summary('active', { title: 'renamed', updatedAt: 3 })); + const second = await store.list({ workspaceId }); + expect(second.items[0]?.title).toBe('renamed'); + }); + + it('get prefers the read model over disk', async () => { + const store = build(); + // Not seeded on disk — only present in the read model. + await queryStore.put(SESSION_COLLECTION, 'warm', summary('warm', { title: 'cached' })); + const got = await store.get('warm'); + expect(got?.title).toBe('cached'); + }); + + it('countActive reflects read-model updates', async () => { + await seedSession('a', {}); + await seedSession('b', { archived: true }); + + const store = build(); + expect(await store.countActive(workspaceId)).toBe(1); + + // Archive `a` through the read model (as SessionMetadata would). + await queryStore.put(SESSION_COLLECTION, 'a', summary('a', { archived: true })); + expect(await store.countActive(workspaceId)).toBe(0); + }); +}); diff --git a/packages/agent-core-v2/test/sessionMetadata/sessionMetadata.test.ts b/packages/agent-core-v2/test/sessionMetadata/sessionMetadata.test.ts index 718fb8fa8..35362876a 100644 --- a/packages/agent-core-v2/test/sessionMetadata/sessionMetadata.test.ts +++ b/packages/agent-core-v2/test/sessionMetadata/sessionMetadata.test.ts @@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; +import { IFlagService } from '#/app/flag'; import { ILogService } from '#/_base/log'; import { ISessionContext, makeSessionContext } from '#/session/sessionContext'; import { ISessionMetadata } from '#/session/sessionMetadata'; @@ -11,8 +12,11 @@ import { JsonAtomicDocumentStore } from '#/persistence/backends/node-fs/atomicDo import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; +import { IQueryStore } from '#/persistence/interface/queryStore'; +import { stubFlag } from '../flag/stubs'; import { stubLog } from '../log/stubs'; +import { stubQueryStore } from '../persistence/stubs'; const META_SCOPE = 'sessions/wd_test/s1/session-meta'; @@ -36,12 +40,14 @@ describe('SessionMetadata', () => { ix = disposables.add(new TestInstantiationService()); ix.stub(ILogService, stubLog()); ix.stub(ISessionContext, makeContext()); + ix.stub(IQueryStore, stubQueryStore()); + ix.stub(IFlagService, stubFlag(false)); ix.set(IFileSystemStorageService, new SyncDescriptor(InMemoryStorageService)); ix.set(IAtomicDocumentStore, new SyncDescriptor(JsonAtomicDocumentStore)); ix.set(ISessionMetadata, new SyncDescriptor(SessionMetadata)); }); - afterEach(() => disposables.dispose()); + afterEach(() => { disposables.dispose(); }); it('creates an initial document on first read', async () => { const meta = ix.get(ISessionMetadata); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 55add119d..9fad009ad 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -496,6 +496,9 @@ importers: '@moonshot-ai/kimi-telemetry': specifier: workspace:^ version: link:../telemetry + '@moonshot-ai/minidb': + specifier: workspace:^ + version: link:../minidb '@moonshot-ai/protocol': specifier: workspace:^ version: link:../protocol