feat(minidb): switch agent-core-v2 query-store to ClusterDb with 16 shards

packages/agent-core-v2/docs/errors.md

@@ -70,7 +70,7 @@ The os / persistence / wire domains show the standard shapes:
 
 - **`os.fs` (`HostFsError`, `os/interface/hostFsErrors.ts`)** — every `IHostFileSystem` backend translates raw errnos at its boundary via the pure `toHostFsError(err, { path, op })`: `ENOENT→os.fs.not_found`, `EISDIR→os.fs.is_directory`, `ENOTDIR→os.fs.not_directory`, `EEXIST→os.fs.already_exists`, `EACCES/EPERM→os.fs.permission_denied`, `ENOTEMPTY→os.fs.not_empty`, everything else `os.fs.unknown`. `details` carries `{ path, op, errno?, syscall? }`. Documented boolean semantics (e.g. `createExclusive` returning `false` on `EEXIST`) stay booleans, not errors.
 - **`os.process` (`HostProcessError`, `os/interface/hostProcess.ts`)** — `os.process.spawn_failed` (details `{ command, args?, cwd?, errno? }`) and `os.process.kill_failed`; both carry the raw error as `cause`. Kill keeps its deliberate tolerances: `ESRCH` is a silent no-op, `EPERM` degrades to `child.kill()`.
-- **`storage` (`StorageError`, `persistence/interface/storage.ts`)** — `storage.not_found` / `decode_failed` / `corrupted` / `io_failed` / `locked`. ENOENT keeps its established absence semantics (`read → undefined`, `list → []`) and is *not* an error; other I/O failures become `storage.io_failed` (`retryable`). Codec parse failures become `storage.decode_failed` with `{ scope, key, format }`; append-log corruption is `AppendLogCorruptedError` (`storage.corrupted`). A query-store open failure (writer lock held by another process) throws `storage.locked` — consumers (e.g. `FileSessionIndex`) catch it explicitly and fall back to their non-read-model path with a one-time warning; there is no silent no-op degradation.
+- **`storage` (`StorageError`, `persistence/interface/storage.ts`)** — `storage.not_found` / `decode_failed` / `corrupted` / `io_failed` / `locked`. ENOENT keeps its established absence semantics (`read → undefined`, `list → []`) and is *not* an error; other I/O failures become `storage.io_failed` (`retryable`). Codec parse failures become `storage.decode_failed` with `{ scope, key, format }`; append-log corruption is `AppendLogCorruptedError` (`storage.corrupted`). `storage.locked` is reserved for a store exclusively held by another process — consumers (e.g. `FileSessionIndex`) catch it explicitly and fall back to their non-read-model path with a one-time warning; there is no silent no-op degradation. (The minidb query-store backend is a multi-process `ClusterDb` and no longer throws it: peers share the store, and per-shard lock contention surfaces as a transient `LockError` instead.)
 - **`wire` (`WireError`, `wire/errors.ts`)** — `DuplicateOpError` (`wire.duplicate_op`, a build-time bug), `CycleError` (`wire.cycle`, details carry the drain depth and a capped op-type sample), and `wire.unknown_record`: replay skips records whose Op type is absent from `OP_REGISTRY` (compatibility), reports each skip through `onUnexpectedError`, and returns `{ unknownRecords }` so the caller knows the restore was lossy.
 
 ## Serialization & boundary translation

packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts

@@ -30,10 +30,11 @@
  * 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 — and as the runtime fallback when
- * the query store reports `storage.locked` (another process holds the writer
- * lock): the first lock warns once and disables the read model for the rest of
- * the process lifetime.
+ * N+1 path remains as the flag-off fallback — and as the runtime fallback if
+ * the query store ever reports `storage.locked`: the first lock warns once and
+ * disables the read model for the rest of the process lifetime. (The minidb
+ * backend is a multi-process `ClusterDb` and no longer produces that error;
+ * the wiring stays as defense in depth.)
  *
  * This is the local-deployment backend of `ISessionIndex`; a server deployment
  * would substitute a database-backed `DbSessionIndex`. Bound at App scope.

packages/agent-core-v2/src/persistence/backends/minidb/miniDbQueryStore.ts

@@ -1,40 +1,53 @@
 /**
- * `minidb` backend — `IQueryStore` implementation over `MiniDb`.
+ * `minidb` backend — `IQueryStore` implementation over `ClusterDb`.
+ *
+ * A rebuildable, in-process derived read-model. The store is a `ClusterDb`
+ * of 16 shards rooted at `<cacheDir>/query-store`: keys are hash-routed over
+ * ordinary `MiniDb` directories, so multiple kimi processes can read and
+ * write the same read model concurrently (a single writer per shard, readers
+ * that never take write locks) instead of failing against a database-wide
+ * single-writer lock. Authoritative data lives in `IAppendLogStore` /
+ * `IAtomicDocumentStore`, never here, so losing the read model is always
+ * safe.
  *
- * 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
- * `<cacheDir>/query-store`.
+ * `query`) and held in memory (`valueMode: 'memory'`); durability is
+ * `everysec`, which is acceptable for a cache. Writes are atomic per shard;
+ * a `batch` spanning shards is best-effort across them — a projector can
+ * always replay from its checkpoint. `lockAcquireTimeoutMs` is lowered from
+ * the 30s default: a cache read must not hang behind a contended shard, and
+ * with `lockHoldMs` yields one second is ample for a live writer.
  *
- * 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.
+ * The database is opened **lazily** on the first actual IO, not at
+ * construction. Construction therefore does no filesystem work — 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.
  *
- * An open failure — typically another kimi process holding the single-writer
- * lock on `<cacheDir>/query-store` — throws `StorageError(storage.locked)`
- * instead of silently degrading to a no-op. The failure is memoized (the
- * rejected open promise is cached), so the error is stable for the process
- * lifetime and consumers can catch it once and fall back to their
- * non-read-model paths.
+ * Corruption handling lifts `MiniDb.openOrRebuild`'s predicate
+ * (`SyntaxError` / `CorruptFrameError`) to the cluster: the first
+ * rebuildable failure triggers one process-lifetime rebuild — close, delete
+ * the directory, reopen empty, retry the operation once — and consumers'
+ * checkpoint-based reprojection repopulates the model. Every other error
+ * propagates as-is; in particular a per-shard `LockError` (a live process
+ * holding a shard beyond the acquire timeout) is transient and must NOT
+ * become `storage.locked`, which consumers would treat as a permanent
+ * read-model outage.
  *
- * A `collection` is encoded as a key prefix (`<collection>\u0000<key>`); 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.
+ * A `collection` is encoded as a key prefix (`<collection>` + NUL + `<key>`); index
+ * names are prefixed with the collection to keep them isolated in the
+ * cluster-wide registry, 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 { promises as fsp } from 'node:fs';
+
 import { join } from 'pathe';
 
-import { MiniDb, type QueryOptions } from '@moonshot-ai/minidb';
+import { type QueryOptions } from '@moonshot-ai/minidb';
+import { ClusterDb } from '@moonshot-ai/minidb/cluster';
 
 import { InstantiationType } from '#/_base/di/extensions';
 import { Disposable, toDisposable } from '#/_base/di/lifecycle';
@@ -51,11 +64,12 @@ import {
   type SortDir,
   type WriteOp,
 } from '#/persistence/interface/queryStore';
-import { StorageError, StorageErrors } from '#/persistence/interface/storage';
 
 const SEP = String.fromCodePoint(0);
 const CHECKPOINT_COLLECTION = '__checkpoint__';
 const STORE_SUBDIR = 'query-store';
+const SHARD_COUNT = 16;
+const LOCK_ACQUIRE_TIMEOUT_MS = 1000;
 
 function physicalKey(collection: string, key: string): string {
   return `${collection}${SEP}${key}`;
@@ -65,11 +79,18 @@ function indexName(collection: string, name: string): string {
   return `${collection}:${name}`;
 }
 
+/** The `MiniDb.openOrRebuild` rebuildable predicate: only unrecoverable
+ *  on-disk corruption justifies wiping the read model. */
+function isRebuildable(error: unknown): boolean {
+  return error instanceof SyntaxError || (error as { name?: string }).name === 'CorruptFrameError';
+}
+
 export class MiniDbQueryStore extends Disposable implements IQueryStore {
   declare readonly _serviceBrand: undefined;
 
   private readonly dir: string;
-  private dbPromise: Promise<MiniDb> | undefined;
+  private dbPromise: Promise<ClusterDb> | undefined;
+  private rebuildPromise: Promise<void> | undefined;
   private readonly ensuredIndexes = new Set<string>();
 
   constructor(
@@ -83,84 +104,112 @@ export class MiniDbQueryStore extends Disposable implements IQueryStore {
     }));
   }
 
-  private openDb(): Promise<MiniDb> {
-    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),
-          });
-        },
-      },
-    ).catch((error) => {
-      throw new StorageError(
-        StorageErrors.codes.STORAGE_LOCKED,
-        'minidb query-store is locked by another process',
-        { details: { dir: this.dir }, cause: error },
-      );
-    });
+  private openDb(): Promise<ClusterDb> {
+    // A rebuild wipes and recreates the directory; opens started while one is
+    // in flight must wait for it instead of racing the rm.
+    if (this.rebuildPromise !== undefined) return this.openDbAfterRebuild();
+    this.dbPromise ??= this.openFresh();
     return this.dbPromise;
   }
 
+  private async openDbAfterRebuild(): Promise<ClusterDb> {
+    await this.rebuildPromise;
+    this.dbPromise ??= this.openFresh();
+    return this.dbPromise;
+  }
+
+  private openFresh(): Promise<ClusterDb> {
+    return ClusterDb.open({
+      dir: this.dir,
+      shardCount: SHARD_COUNT,
+      valueCodec: 'json',
+      valueMode: 'memory',
+      fsyncPolicy: 'everysec',
+      lockAcquireTimeoutMs: LOCK_ACQUIRE_TIMEOUT_MS,
+    });
+  }
+
+  /** One process-lifetime rebuild: wipe the corrupt store and let the next
+   *  open start empty. Concurrent callers share the same in-flight rebuild. */
+  private rebuild(cause: unknown): Promise<void> {
+    this.rebuildPromise ??= (async () => {
+      this.log.warn('minidb query-store rebuilt after corruption', {
+        dir: this.dir,
+        error: String(cause),
+      });
+      const previous = this.dbPromise;
+      // Reset before the first await so a concurrent openDb() waits on
+      // rebuildPromise instead of reusing the corrupt instance.
+      this.dbPromise = undefined;
+      this.ensuredIndexes.clear();
+      if (previous !== undefined) {
+        const db = await previous.catch(() => undefined);
+        await db?.close().catch(() => {});
+      }
+      await fsp.rm(this.dir, { recursive: true, force: true });
+    })();
+    return this.rebuildPromise;
+  }
+
+  private async withDb<T>(op: (db: ClusterDb) => Promise<T>): Promise<T> {
+    try {
+      return await op(await this.openDb());
+    } catch (error) {
+      if (!isRebuildable(error)) throw error;
+      await this.rebuild(error);
+      // One retry on the fresh store; a second failure propagates as-is.
+      return op(await this.openDb());
+    }
+  }
+
   async put<T>(collection: string, key: string, value: T): Promise<void> {
-    const db = await this.openDb();
-    await db.set(physicalKey(collection, key), value);
+    await this.withDb((db) => db.set(physicalKey(collection, key), value));
   }
 
   async batch(ops: readonly WriteOp[]): Promise<void> {
     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) },
+    await this.withDb((db) =>
+      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<void> {
-    const db = await this.openDb();
-    await db.del(physicalKey(collection, key));
+    await this.withDb((db) => db.del(physicalKey(collection, key)));
   }
 
   async get<T>(collection: string, key: string): Promise<T | undefined> {
-    const db = await this.openDb();
-    return db.get(physicalKey(collection, key)) as T | undefined;
+    return this.withDb((db) => db.get(physicalKey(collection, key)) as Promise<T | undefined>);
   }
 
   query<T>(collection: string): IQuery<T> {
-    return new MiniDbQuery<T>(() => this.openDb(), collection);
+    return new MiniDbQuery<T>((op) => this.withDb(op), collection);
   }
 
   async ensureIndex(collection: string, def: IndexDef): Promise<void> {
     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 {
+    await this.withDb(async (db) => {
       try {
-        await db.createTextIndex(name, { fields: def.fields });
+        if (def.kind === 'value') {
+          await db.createIndex(name, { field: def.field, sparse: true, unique: def.unique });
+        } else if (def.kind === 'compound') {
+          await db.createCompoundIndex(name, { groupBy: def.groupBy, orderBy: def.orderBy });
+        } else {
+          await db.createTextIndex(name, { fields: def.fields });
+        }
       } catch (error) {
+        // A raced ensure (a peer process created it first, or a rebuild
+        // replayed this call) is a no-op: the definition already exists.
         if (!(error instanceof Error) || !error.message.includes('already exists')) throw error;
       }
-    }
+    });
     this.ensuredIndexes.add(guard);
   }
 
@@ -173,8 +222,7 @@ export class MiniDbQueryStore extends Disposable implements IQueryStore {
   }
 
   async close(): Promise<void> {
-    if (this.dbPromise === undefined) return;
-    const db = await this.dbPromise.catch(() => undefined);
+    const db = await this.dbPromise?.catch(() => undefined);
     await db?.close();
   }
 }
@@ -187,7 +235,7 @@ class MiniDbQuery<T> implements IQuery<T> {
   private skip = 0;
 
   constructor(
-    private readonly openDb: () => Promise<MiniDb>,
+    private readonly withDb: <R>(op: (db: ClusterDb) => Promise<R>) => Promise<R>,
     private readonly collection: string,
   ) {}
 
@@ -213,7 +261,6 @@ class MiniDbQuery<T> implements IQuery<T> {
   }
 
   async execute(): Promise<Page<T>> {
-    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<string, unknown>;
@@ -222,7 +269,7 @@ class MiniDbQuery<T> implements IQuery<T> {
     }
     q.skip = this.skip;
     if (this.lim !== undefined) q.limit = this.lim + 1;
-    const rows = db.query(q) as ReadonlyArray<{ key: string; value: T }>;
+    const rows = (await this.withDb((db) => 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) {