diff --git a/packages/agent-core-v2/package.json b/packages/agent-core-v2/package.json index be2f3a9ed..987b417ee 100644 --- a/packages/agent-core-v2/package.json +++ b/packages/agent-core-v2/package.json @@ -62,6 +62,7 @@ "@moonshot-ai/kimi-telemetry": "workspace:^", "@moonshot-ai/kosong": "workspace:^", "@moonshot-ai/protocol": "workspace:^", + "chokidar": "^4.0.3", "ignore": "^5.3.2", "js-yaml": "^4.1.1", "nunjucks": "^3.2.4", diff --git a/packages/agent-core-v2/src/storage/fileStorageService.ts b/packages/agent-core-v2/src/storage/fileStorageService.ts index aa5009f24..24bdf8690 100644 --- a/packages/agent-core-v2/src/storage/fileStorageService.ts +++ b/packages/agent-core-v2/src/storage/fileStorageService.ts @@ -10,9 +10,9 @@ * fsync, so the replacement is both atomic and durable. * - `append` → `open('a')` + write + `fh.sync()` (when `durable`), plus a * one-time directory fsync per scope. - * - `watch` → `fs.watch` the parent directory (filtered by filename and - * debounced), so it survives atomic-replace renames and observes - * a file that does not exist yet. + * - `watch` → chokidar on the parent directory, filtered to the exact key and + * debounced, so it survives atomic-replace renames and observes a + * file that does not exist yet at subscription time. * * It uses raw `node:fs` rather than `kaos`: the storage kernel needs direct * control over append offsets, fsync, atomic rename and streaming, which the @@ -21,9 +21,10 @@ * this backend, never `node:fs` directly. */ -import { createReadStream, mkdirSync, watch as fsWatch, type FSWatcher } from 'node:fs'; +import { createReadStream, mkdirSync } from 'node:fs'; import { mkdir, open, readFile, readdir, unlink } from 'node:fs/promises'; -import { basename, dirname, join } from 'pathe'; +import { FSWatcher } from 'chokidar'; +import { dirname, join, normalize } from 'pathe'; import { DisposableStore, @@ -137,7 +138,7 @@ export class FileStorageService implements IStorageService { watch(scope: string, key: string): Event { const target = this.path(scope, key); const dir = dirname(target); - const name = basename(target); + const normalizedTarget = normalize(target); const emitter = new Emitter(); let watcher: FSWatcher | undefined; @@ -149,16 +150,24 @@ export class FileStorageService implements IStorageService { timer = setTimeout(() => emitter.fire(), WATCH_DEBOUNCE_MS); }; - // Watch the parent directory and filter by filename: the directory survives - // atomic-replace renames (which would detach an inode watcher), and it lets - // us observe a file that does not exist yet at subscription time. + // Watch the parent directory and filter by exact path: the directory survives + // atomic-replace renames (which would detach a single-file watcher) and it + // lets us observe a file that does not exist yet at subscription time. Events + // are debounced to collapse the burst a single save (plus its atomic-replace + // temp file) emits. const arm = (): void => { try { mkdirSync(dir, { recursive: true, mode: this.dirMode }); - watcher = fsWatch(dir, (_event, filename) => { - if (filename === null || filename === name) schedule(); + watcher = new FSWatcher({ + ignoreInitial: true, + awaitWriteFinish: false, + depth: 0, + }); + watcher.on('all', (_event, changedPath) => { + if (normalize(changedPath) === normalizedTarget) schedule(); }); watcher.on('error', () => undefined); + watcher.add(dir); } catch { // Best effort: callers can still reload explicitly when watching fails. } @@ -169,7 +178,8 @@ export class FileStorageService implements IStorageService { clearTimeout(timer); timer = undefined; } - watcher?.close(); + const closeResult = watcher?.close(); + if (closeResult !== undefined) void closeResult.catch(() => undefined); watcher = undefined; }; diff --git a/packages/agent-core-v2/test/storage/storageService.test.ts b/packages/agent-core-v2/test/storage/storageService.test.ts index 2bbb56027..b8d9501b6 100644 --- a/packages/agent-core-v2/test/storage/storageService.test.ts +++ b/packages/agent-core-v2/test/storage/storageService.test.ts @@ -33,6 +33,11 @@ function storageServiceSuite( await cleanup?.(); }); + // chokidar-backed storage attaches the OS watcher asynchronously; give it a + // moment to register before mutating. Harmless for the in-memory backend. + const settle = (): Promise => + new Promise((resolve) => setTimeout(resolve, 100)); + it('read returns undefined for a missing key', async () => { expect(await service.read('s', 'missing')).toBeUndefined(); }); @@ -94,6 +99,7 @@ function storageServiceSuite( resolve(); }); }); + await settle(); await service.write('s', 'k', enc.encode('v')); await fired; }); @@ -119,6 +125,7 @@ function storageServiceSuite( resolve(); }); }); + await settle(); await service.delete('s', 'k'); await fired; }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8f4d81976..1c33e4668 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -453,6 +453,9 @@ importers: '@moonshot-ai/protocol': specifier: workspace:^ version: link:../protocol + chokidar: + specifier: ^4.0.3 + version: 4.0.3 ignore: specifier: ^5.3.2 version: 5.3.2