From 84b2989e8f93195995b93ecc74b01f22ee943de8 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 8 Jul 2026 21:47:01 +0800 Subject: [PATCH] feat(fs-watch): add workspace fs watch with v1-compatible WS delivery - os layer: add IHostFsWatchService over chokidar (raw create/modify/delete, .git ignored) - session layer: add ISessionFsWatchService, a workspace-confined, debounced, .gitignore-aware FsChangeEvent feed - kap-server: add FsWatchBridge pushing event.fs.changed over /api/v1/ws (watch_fs_add/remove, volatile, per-connection filter), byte-compatible with v1 - tests: os/session unit tests and kap-server fs-watch e2e --- packages/agent-core-v2/src/index.ts | 4 + .../backends/node-local/hostFsWatchService.ts | 102 +++++ .../src/os/interface/hostFsWatch.ts | 54 +++ .../src/session/sessionFs/fsWatch.ts | 39 ++ .../src/session/sessionFs/fsWatchService.ts | 216 ++++++++++ .../test/os/hostFsWatchService.test.ts | 90 ++++ .../test/sessionFs/fsWatchService.test.ts | 213 ++++++++++ packages/kap-server/src/start.ts | 3 + .../src/transport/ws/v1/fsWatchBridge.ts | 284 +++++++++++++ .../src/transport/ws/v1/registerWsV1.ts | 3 + .../src/transport/ws/v1/wsConnectionV1.ts | 45 +- packages/kap-server/test/fs-watch.e2e.test.ts | 398 ++++++++++++++++++ 12 files changed, 1449 insertions(+), 2 deletions(-) create mode 100644 packages/agent-core-v2/src/os/backends/node-local/hostFsWatchService.ts create mode 100644 packages/agent-core-v2/src/os/interface/hostFsWatch.ts create mode 100644 packages/agent-core-v2/src/session/sessionFs/fsWatch.ts create mode 100644 packages/agent-core-v2/src/session/sessionFs/fsWatchService.ts create mode 100644 packages/agent-core-v2/test/os/hostFsWatchService.test.ts create mode 100644 packages/agent-core-v2/test/sessionFs/fsWatchService.test.ts create mode 100644 packages/kap-server/src/transport/ws/v1/fsWatchBridge.ts create mode 100644 packages/kap-server/test/fs-watch.e2e.test.ts diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index aaa5fccef..18bd3c12f 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -33,11 +33,13 @@ export * from '#/app/bootstrap/bootstrap'; export * from '#/app/bootstrap/bootstrapService'; export * from '#/os/interface/hostEnvironment'; export * from '#/os/interface/hostFileSystem'; +export * from '#/os/interface/hostFsWatch'; export * from '#/os/interface/hostProcess'; export * from '#/os/interface/terminal'; export * from '#/os/interface/terminalErrors'; export * from '#/os/backends/node-local/hostEnvironmentService'; export * from '#/os/backends/node-local/hostFsService'; +export * from '#/os/backends/node-local/hostFsWatchService'; export * from '#/os/backends/node-local/hostProcessService'; export * from '#/os/backends/node-local/hostTerminalService'; export * from '#/os/backends/node-local/tools/bash'; @@ -254,6 +256,8 @@ export * from '#/session/process/processRunnerService'; export * from '#/session/sessionFs/errors'; export * from '#/session/sessionFs/fs'; export * from '#/session/sessionFs/fsService'; +export * from '#/session/sessionFs/fsWatch'; +export * from '#/session/sessionFs/fsWatchService'; export * from '#/session/sessionFs/gitContext'; export * from '#/session/sessionFs/rgLocator'; export * from '#/session/sessionFs/runRg'; diff --git a/packages/agent-core-v2/src/os/backends/node-local/hostFsWatchService.ts b/packages/agent-core-v2/src/os/backends/node-local/hostFsWatchService.ts new file mode 100644 index 000000000..560c07c37 --- /dev/null +++ b/packages/agent-core-v2/src/os/backends/node-local/hostFsWatchService.ts @@ -0,0 +1,102 @@ +/** + * `hostFsWatch` domain (L1) — `IHostFsWatchService` implementation. + * + * Wraps `chokidar` to report raw create/modify/delete events under an absolute + * path. Each `watch()` call owns an independent `FSWatcher`; disposing the + * handle closes it. Bound at App scope. + */ + +import { FSWatcher } from 'chokidar'; + +import { Emitter, type Event } from '#/_base/event'; +import { InstantiationType } from '#/_base/di/extensions'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; + +import { + type HostFsChange, + type HostFsChangeAction, + type HostFsChangeKind, + type HostFsWatchOptions, + type IHostFsWatchHandle, + IHostFsWatchService, +} from '#/os/interface/hostFsWatch'; + +/** Suppress `.git` directories by default — they are high-volume noise. */ +const DEFAULT_IGNORED = (p: string): boolean => /(?:^|[/\\])\.git(?:$|[/\\])/.test(p); + +class HostFsWatchHandle implements IHostFsWatchHandle { + readonly onDidChange: Event; + + private readonly emitter: Emitter; + private readonly watcher: FSWatcher; + private disposed = false; + + constructor(path: string, options: HostFsWatchOptions | undefined) { + this.emitter = new Emitter(); + this.onDidChange = this.emitter.event; + this.watcher = new FSWatcher({ + ignoreInitial: true, + persistent: false, + followSymlinks: false, + depth: options?.recursive === false ? 0 : undefined, + ignored: options?.ignored ?? DEFAULT_IGNORED, + }); + this.watcher.on('all', (eventName: string, absPath: string) => { + const mapped = mapChokidarEvent(eventName, absPath); + if (mapped !== undefined) this.emitter.fire(mapped); + }); + this.watcher.on('error', () => { + // Best-effort: a watcher error must not crash the host. Higher layers + // can always re-subscribe if events stop arriving. + }); + this.watcher.add(path); + } + + dispose(): void { + if (this.disposed) return; + this.disposed = true; + void this.watcher.close().catch(() => undefined); + this.emitter.dispose(); + } +} + +export class HostFsWatchService implements IHostFsWatchService { + declare readonly _serviceBrand: undefined; + + watch(path: string, options?: HostFsWatchOptions): IHostFsWatchHandle { + return new HostFsWatchHandle(path, options); + } +} + +function mapChokidarEvent(eventName: string, absPath: string): HostFsChange | undefined { + const mapped = mapActionAndKind(eventName); + if (mapped === undefined) return undefined; + return { path: absPath, action: mapped.action, kind: mapped.kind }; +} + +function mapActionAndKind( + eventName: string, +): { action: HostFsChangeAction; kind: HostFsChangeKind } | undefined { + switch (eventName) { + case 'add': + return { action: 'created', kind: 'file' }; + case 'addDir': + return { action: 'created', kind: 'directory' }; + case 'change': + return { action: 'modified', kind: 'file' }; + case 'unlink': + return { action: 'deleted', kind: 'file' }; + case 'unlinkDir': + return { action: 'deleted', kind: 'directory' }; + default: + return undefined; + } +} + +registerScopedService( + LifecycleScope.App, + IHostFsWatchService, + HostFsWatchService, + InstantiationType.Delayed, + 'hostFsWatch', +); diff --git a/packages/agent-core-v2/src/os/interface/hostFsWatch.ts b/packages/agent-core-v2/src/os/interface/hostFsWatch.ts new file mode 100644 index 000000000..6577d6870 --- /dev/null +++ b/packages/agent-core-v2/src/os/interface/hostFsWatch.ts @@ -0,0 +1,54 @@ +/** + * `hostFsWatch` domain (L1) — local real-filesystem change notifications. + * + * Defines the `IHostFsWatchService`, a thin primitive over the host OS file + * watcher. It reports raw create/modify/delete events under an absolute path + * and knows nothing about sessions, connections, workspaces or wire frames. + * App-scoped — one shared instance. Higher layers (e.g. `sessionFsWatch`) + * subscribe, confine events to a workspace, debounce/coalesce and re-expose + * them as domain events. + */ + +import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { Event } from '#/_base/event'; +import type { IDisposable } from '#/_base/di/lifecycle'; + +export type HostFsChangeKind = 'file' | 'directory'; +export type HostFsChangeAction = 'created' | 'modified' | 'deleted'; + +export interface HostFsChange { + /** Absolute path that changed. */ + readonly path: string; + readonly action: HostFsChangeAction; + readonly kind: HostFsChangeKind; +} + +export interface HostFsWatchOptions { + /** Watch recursively into subdirectories. Defaults to `true`. */ + readonly recursive?: boolean; + /** + * Predicate returning `true` for paths the watcher should ignore. Defaults + * to a filter that suppresses `.git` directories. Replaces the default when + * provided. + */ + readonly ignored?: (path: string) => boolean; +} + +/** A live watch subscription. Dispose to stop receiving events. */ +export interface IHostFsWatchHandle extends IDisposable { + readonly onDidChange: Event; +} + +export interface IHostFsWatchService { + readonly _serviceBrand: undefined; + + /** + * Watch `path` (absolute, file or directory) and return a handle that fires + * for changes beneath it. Synchronous — the underlying watcher is armed + * immediately; dispose the handle to stop. + */ + watch(path: string, options?: HostFsWatchOptions): IHostFsWatchHandle; +} + +export const IHostFsWatchService: ServiceIdentifier = + createDecorator('hostFsWatchService'); diff --git a/packages/agent-core-v2/src/session/sessionFs/fsWatch.ts b/packages/agent-core-v2/src/session/sessionFs/fsWatch.ts new file mode 100644 index 000000000..30d600454 --- /dev/null +++ b/packages/agent-core-v2/src/session/sessionFs/fsWatch.ts @@ -0,0 +1,39 @@ +/** + * `sessionFsWatch` domain (L2) — workspace-confined filesystem change feed. + * + * Defines the `ISessionFsWatchService` that turns the os `IHostFsWatchService` + * raw events into a workspace-relative, debounced, `.gitignore`-aware change + * feed (`FsChangeEvent`) for the session. Callers declare the set of + * workspace-relative paths they care about; events outside that subtree are + * dropped. Session-scoped — the scope itself is the session, so no + * `sessionId` is threaded through. + */ + +import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { Event } from '#/_base/event'; +import type { FsChangeEvent } from '@moonshot-ai/protocol'; + +export interface ISessionFsWatchService { + readonly _serviceBrand: undefined; + + /** + * Replace the set of workspace-relative paths to observe. `'.'` watches the + * whole workspace. Passing an empty array stops the underlying watcher. + * Paths are confined to the workspace; absolute / `..` / escaping inputs + * throw `FS_PATH_ESCAPES`. + */ + setWatchedPaths(paths: readonly string[]): void; + + /** Currently observed workspace-relative paths (posix). */ + readonly watchedPaths: readonly string[]; + + /** + * Coalesced change feed. Each event carries the changes for one debounce + * window; when the window overflows, `changes` is emptied and `truncated` + * (with `count`) is set so consumers can fall back to a full refresh. + */ + readonly onDidChangeFiles: Event; +} + +export const ISessionFsWatchService: ServiceIdentifier = + createDecorator('sessionFsWatchService'); diff --git a/packages/agent-core-v2/src/session/sessionFs/fsWatchService.ts b/packages/agent-core-v2/src/session/sessionFs/fsWatchService.ts new file mode 100644 index 000000000..52b31bcf5 --- /dev/null +++ b/packages/agent-core-v2/src/session/sessionFs/fsWatchService.ts @@ -0,0 +1,216 @@ +/** + * `sessionFsWatch` domain (L2) — `ISessionFsWatchService` implementation. + * + * Subscribes to the os `IHostFsWatchService` on the workspace root, confines + * events to the caller-declared subtree and to non-`.gitignore`d paths, + * debounces them into fixed windows and re-exposes them as workspace-relative + * `FsChangeEvent`s. The os watcher is started lazily on the first non-empty + * subscription and stopped when the subscription set becomes empty. Path + * confinement is lexical (`ISessionWorkspaceContext.isWithin`), matching + * `sessionFs`. + */ + +import { isAbsolute, join, relative, sep } from 'node:path'; + +import ignore, { type Ignore } from 'ignore'; + +import { Disposable, type IDisposable } from '#/_base/di/lifecycle'; +import { Emitter, type Event } from '#/_base/event'; +import { InstantiationType } from '#/_base/di/extensions'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { ErrorCodes, KimiError } from '#/errors'; +import { IHostFileSystem } from '#/os/interface/hostFileSystem'; +import { + type HostFsChange, + type IHostFsWatchHandle, + IHostFsWatchService, +} from '#/os/interface/hostFsWatch'; +import { ISessionWorkspaceContext } from '#/session/workspaceContext/workspaceContext'; +import type { FsChangeEntry, FsChangeEvent } from '@moonshot-ai/protocol'; + +import { ISessionFsWatchService } from './fsWatch'; + +const DEBOUNCE_MS = 200; +const MAX_CHANGES_PER_WINDOW = 500; + +export class SessionFsWatchService extends Disposable implements ISessionFsWatchService { + declare readonly _serviceBrand: undefined; + + private readonly emitter = this._register(new Emitter()); + readonly onDidChangeFiles: Event = this.emitter.event; + + private watched = new Set(); + private handle: IHostFsWatchHandle | undefined; + private handleSub: IDisposable | undefined; + + private debounceTimer: NodeJS.Timeout | undefined; + private pending: FsChangeEntry[] = []; + private rawCount = 0; + private truncated = false; + + /** Always present; starts with `.git/` and is augmented with `.gitignore` once loaded. */ + private readonly matcher: Ignore = ignore().add('.git/'); + private gitignoreLoaded = false; + + constructor( + @ISessionWorkspaceContext private readonly workspace: ISessionWorkspaceContext, + @IHostFsWatchService private readonly hostFsWatch: IHostFsWatchService, + @IHostFileSystem private readonly hostFs: IHostFileSystem, + ) { + super(); + } + + get watchedPaths(): readonly string[] { + return Array.from(this.watched); + } + + setWatchedPaths(paths: readonly string[]): void { + const next = new Set(); + for (const p of paths) { + const abs = this.resolveWithin(p); + next.add(this.toRel(abs)); + } + this.watched = next; + if (next.size === 0) { + this.teardownHandle(); + this.clearWindow(); + return; + } + this.ensureHandle(); + } + + private ensureHandle(): void { + if (this.handle !== undefined) return; + this.loadGitignore(); + const handle = this.hostFsWatch.watch(this.workspace.workDir, { recursive: true }); + this.handle = handle; + this.handleSub = handle.onDidChange((e) => this.onRaw(e)); + } + + private teardownHandle(): void { + this.handleSub?.dispose(); + this.handleSub = undefined; + this.handle?.dispose(); + this.handle = undefined; + } + + private loadGitignore(): void { + if (this.gitignoreLoaded) return; + this.gitignoreLoaded = true; + void this.hostFs + .readText(join(this.workspace.workDir, '.gitignore')) + .then( + (content) => { + this.matcher.add(content); + }, + () => undefined, + ); + } + + private onRaw(e: HostFsChange): void { + const rel = this.toRel(e.path); + if (rel === '.') return; + const probe = e.kind === 'directory' ? `${rel}/` : rel; + if (this.matcher.ignores(probe)) return; + if (!isUnderAny(rel, this.watched)) return; + + this.pending.push({ path: rel, change: e.action, kind: e.kind }); + this.rawCount += 1; + if (this.pending.length > MAX_CHANGES_PER_WINDOW) { + this.truncated = true; + this.pending = []; + } + if (this.debounceTimer === undefined) { + const timer = setTimeout(() => this.flush(), DEBOUNCE_MS); + timer.unref?.(); + this.debounceTimer = timer; + } + } + + private flush(): void { + this.debounceTimer = undefined; + if (this.rawCount === 0) return; + const truncated = this.truncated; + const count = this.rawCount; + const changes = truncated ? [] : this.pending; + this.pending = []; + this.rawCount = 0; + this.truncated = false; + + const event: FsChangeEvent = { + changes, + coalesced_window_ms: DEBOUNCE_MS, + ...(truncated ? { truncated: true, count } : {}), + }; + this.emitter.fire(event); + } + + private clearWindow(): void { + if (this.debounceTimer !== undefined) { + clearTimeout(this.debounceTimer); + this.debounceTimer = undefined; + } + this.pending = []; + this.rawCount = 0; + this.truncated = false; + } + + override dispose(): void { + this.clearWindow(); + this.teardownHandle(); + super.dispose(); + } + + private resolveWithin(inputPath: string): string { + if (inputPath === '' || inputPath === '/') { + throw new KimiError(ErrorCodes.FS_PATH_ESCAPES, `path "${inputPath}" rejected (empty)`, { + details: { path: inputPath, reason: 'empty' }, + }); + } + if (isAbsolute(inputPath)) { + throw new KimiError(ErrorCodes.FS_PATH_ESCAPES, `path "${inputPath}" rejected (absolute)`, { + details: { path: inputPath, reason: 'absolute' }, + }); + } + const segments = inputPath.split(/[/\\]+/); + if (segments.some((s) => s === '..')) { + throw new KimiError( + ErrorCodes.FS_PATH_ESCAPES, + `path "${inputPath}" rejected (dotdot segment)`, + { details: { path: inputPath, reason: 'dotdot_segment' } }, + ); + } + const abs = this.workspace.resolve(inputPath); + if (!this.workspace.isWithin(abs)) { + throw new KimiError(ErrorCodes.FS_PATH_ESCAPES, `path "${inputPath}" escapes workspace`, { + details: { path: inputPath, reason: 'resolved_outside' }, + }); + } + return abs; + } + + private toRel(abs: string): string { + const cwd = this.workspace.workDir; + if (abs === cwd) return '.'; + const rel = relative(cwd, abs); + if (rel === '') return '.'; + return rel.split(sep).join('/'); + } +} + +function isUnderAny(rel: string, parents: ReadonlySet): boolean { + for (const parent of parents) { + if (parent === '.' || parent === '') return true; + if (rel === parent) return true; + if (rel.startsWith(`${parent}/`)) return true; + } + return false; +} + +registerScopedService( + LifecycleScope.Session, + ISessionFsWatchService, + SessionFsWatchService, + InstantiationType.Delayed, + 'sessionFsWatch', +); diff --git a/packages/agent-core-v2/test/os/hostFsWatchService.test.ts b/packages/agent-core-v2/test/os/hostFsWatchService.test.ts new file mode 100644 index 000000000..e3305dc76 --- /dev/null +++ b/packages/agent-core-v2/test/os/hostFsWatchService.test.ts @@ -0,0 +1,90 @@ +/** + * `hostFsWatch` domain (L1) — integration test against the real `chokidar` + * watcher on a temporary directory. + */ + +import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { afterEach, describe, expect, it } from 'vitest'; + +import { HostFsWatchService } from '#/os/backends/node-local/hostFsWatchService'; +import type { HostFsChange, IHostFsWatchHandle } from '#/os/interface/hostFsWatch'; + +const wait = (ms: number): Promise => new Promise((r) => setTimeout(r, ms)); + +describe('HostFsWatchService', () => { + let root: string; + let handle: IHostFsWatchHandle | undefined; + + afterEach(async () => { + handle?.dispose(); + handle = undefined; + if (root) await rm(root, { recursive: true, force: true }); + }); + + async function start(recursive = true): Promise { + const events: HostFsChange[] = []; + const svc = new HostFsWatchService(); + handle = svc.watch(root, { recursive }); + handle.onDidChange((e) => events.push(e)); + // Let chokidar arm before the test mutates the tree. + await wait(200); + return events; + } + + it('reports create / modify / delete for a file', async () => { + root = await mkdtemp(join(tmpdir(), 'hostfswatch-')); + const events = await start(); + + const file = join(root, 'a.txt'); + await writeFile(file, 'v1'); + await wait(300); + await writeFile(file, 'v2'); + await wait(300); + await rm(file); + await wait(300); + + const actions = events.filter((e) => e.path === file).map((e) => e.action); + expect(actions).toContain('created'); + expect(actions).toContain('modified'); + expect(actions).toContain('deleted'); + expect(events.find((e) => e.path === file)?.kind).toBe('file'); + }); + + it('does not fire for paths ignored by default (.git)', async () => { + root = await mkdtemp(join(tmpdir(), 'hostfswatch-')); + const events = await start(); + + await mkdir(join(root, '.git')); + await writeFile(join(root, '.git', 'config'), 'x'); + await wait(300); + + expect(events.some((e) => e.path.includes('/.git/') || e.path.endsWith('/.git'))).toBe(false); + }); + + it('does not fire for pre-existing files (ignoreInitial)', async () => { + root = await mkdtemp(join(tmpdir(), 'hostfswatch-')); + const preexisting = join(root, 'pre.txt'); + await writeFile(preexisting, 'v0'); + + const events = await start(); + await wait(300); + + expect(events.some((e) => e.path === preexisting)).toBe(false); + }); + + it('stops firing after the handle is disposed', async () => { + root = await mkdtemp(join(tmpdir(), 'hostfswatch-')); + const events = await start(); + + handle?.dispose(); + handle = undefined; + + await writeFile(join(root, 'after-dispose.txt'), 'x'); + await wait(300); + + expect(events).toHaveLength(0); + }); +}); diff --git a/packages/agent-core-v2/test/sessionFs/fsWatchService.test.ts b/packages/agent-core-v2/test/sessionFs/fsWatchService.test.ts new file mode 100644 index 000000000..97eb3a8f2 --- /dev/null +++ b/packages/agent-core-v2/test/sessionFs/fsWatchService.test.ts @@ -0,0 +1,213 @@ +/** + * `sessionFsWatch` domain (L2) — verifies confinement to the declared subtree, + * workspace-relative path mapping, debounce coalescing, window truncation, + * `.gitignore` filtering and handle lifecycle, using a fake os watcher. + */ + +import { isAbsolute, join, relative, resolve } from 'node:path'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { LifecycleScope } from '#/_base/di/scope'; +import { createScopedTestHost, stubPair } from '#/_base/di/test'; +import { IHostFileSystem } from '#/os/interface/hostFileSystem'; +import { + type HostFsChange, + type IHostFsWatchHandle, + IHostFsWatchService, +} from '#/os/interface/hostFsWatch'; +import { ISessionWorkspaceContext } from '#/session/workspaceContext/workspaceContext'; +import type { FsChangeEvent } from '@moonshot-ai/protocol'; + +import { ISessionFsWatchService } from '#/session/sessionFs/fsWatch'; +// Imported for its scoped-registration side effect. +import { SessionFsWatchService } from '#/session/sessionFs/fsWatchService'; + +const WORK_DIR = '/repo'; + +void SessionFsWatchService; + +function stubWorkspace(): ISessionWorkspaceContext { + return { + _serviceBrand: undefined, + workDir: WORK_DIR, + additionalDirs: [], + setWorkDir: () => {}, + setAdditionalDirs: () => {}, + resolve: (rel) => (isAbsolute(rel) ? rel : resolve(WORK_DIR, rel)), + isWithin: (abs) => { + const r = relative(WORK_DIR, abs); + return r === '' || (!r.startsWith('..') && !isAbsolute(r)); + }, + assertAllowed: (abs) => abs, + addAdditionalDir: () => {}, + removeAdditionalDir: () => {}, + }; +} + +interface FakeWatch { + readonly service: IHostFsWatchService; + readonly watchCalls: string[]; + fire: (rel: string, action: HostFsChange['action'], kind?: HostFsChange['kind']) => void; + readonly disposed: () => boolean; +} + +function fakeHostFsWatch(): FakeWatch { + const watchCalls: string[] = []; + let listener: ((e: HostFsChange) => void) | undefined; + let disposed = false; + const handle: IHostFsWatchHandle = { + onDidChange: (l) => { + listener = l; + return { dispose: () => (listener = undefined) }; + }, + dispose: () => { + disposed = true; + listener = undefined; + }, + }; + const service: IHostFsWatchService = { + _serviceBrand: undefined, + watch: (path) => { + watchCalls.push(path); + disposed = false; + return handle; + }, + }; + return { + service, + watchCalls, + fire: (rel, action, kind = 'file') => + listener?.({ path: join(WORK_DIR, rel), action, kind }), + disposed: () => disposed, + }; +} + +function fakeHostFs(gitignore?: string): IHostFileSystem { + return { + _serviceBrand: undefined, + readText: async (p: string) => { + if (gitignore !== undefined && p === join(WORK_DIR, '.gitignore')) return gitignore; + const err = new Error(`ENOENT: ${p}`) as NodeJS.ErrnoException; + err.code = 'ENOENT'; + throw err; + }, + } as unknown as IHostFileSystem; +} + +interface Harness { + readonly svc: ISessionFsWatchService; + readonly watch: FakeWatch; + readonly events: FsChangeEvent[]; +} + +function makeSession(gitignore?: string): Harness { + const watch = fakeHostFsWatch(); + const host = createScopedTestHost(); + const session = host.child(LifecycleScope.Session, 's1', [ + stubPair(ISessionWorkspaceContext, stubWorkspace()), + stubPair(IHostFsWatchService, watch.service), + stubPair(IHostFileSystem, fakeHostFs(gitignore)), + ]); + const svc = session.accessor.get(ISessionFsWatchService); + const events: FsChangeEvent[] = []; + svc.onDidChangeFiles((e) => events.push(e)); + disposers.push(() => host.dispose()); + return { svc, watch, events }; +} + +const disposers: Array<() => void> = []; + +describe('SessionFsWatchService', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + for (const d of disposers.splice(0)) d(); + vi.useRealTimers(); + }); + + it('starts the os watcher on the workspace root for a non-empty subscription', () => { + const { svc, watch } = makeSession(); + svc.setWatchedPaths(['src']); + expect(watch.watchCalls).toEqual([WORK_DIR]); + expect(svc.watchedPaths).toEqual(['src']); + }); + + it('drops events outside the subscribed subtree', () => { + const { svc, watch, events } = makeSession(); + svc.setWatchedPaths(['src']); + + watch.fire('src/a.ts', 'created'); + watch.fire('lib/b.ts', 'created'); + vi.advanceTimersByTime(200); + + expect(events).toHaveLength(1); + expect(events[0]?.changes).toEqual([{ path: 'src/a.ts', change: 'created', kind: 'file' }]); + }); + + it('coalesces changes within a window into one event', () => { + const { svc, watch, events } = makeSession(); + svc.setWatchedPaths(['.']); + + watch.fire('a.ts', 'created'); + watch.fire('b.ts', 'modified'); + watch.fire('c.ts', 'deleted'); + vi.advanceTimersByTime(200); + + expect(events).toHaveLength(1); + expect(events[0]?.coalesced_window_ms).toBe(200); + expect(events[0]?.changes).toHaveLength(3); + }); + + it('marks the event truncated when the window overflows', () => { + const { svc, watch, events } = makeSession(); + svc.setWatchedPaths(['.']); + + for (let i = 0; i < 501; i++) watch.fire(`f${i}.ts`, 'created'); + vi.advanceTimersByTime(200); + + expect(events).toHaveLength(1); + expect(events[0]?.truncated).toBe(true); + expect(events[0]?.changes).toEqual([]); + expect(events[0]?.count).toBe(501); + }); + + it('filters out `.gitignore`d paths once loaded', async () => { + const { svc, watch, events } = makeSession('dist/\n'); + svc.setWatchedPaths(['.']); + // Let the async `.gitignore` load (Promise.then) land on the matcher. + await Promise.resolve(); + await Promise.resolve(); + + watch.fire('dist/x.js', 'created'); + watch.fire('src/keep.ts', 'created'); + vi.advanceTimersByTime(200); + + expect(events).toHaveLength(1); + expect(events[0]?.changes.map((c) => c.path)).toEqual(['src/keep.ts']); + }); + + it('rejects paths that escape the workspace', () => { + const { svc } = makeSession(); + expect(() => svc.setWatchedPaths(['../x'])).toThrowError(/escapes workspace|rejected/); + expect(() => svc.setWatchedPaths(['/abs'])).toThrowError(/rejected/); + }); + + it('disposes the os handle when the subscription set becomes empty', () => { + const { svc, watch } = makeSession(); + svc.setWatchedPaths(['src']); + expect(watch.disposed()).toBe(false); + svc.setWatchedPaths([]); + expect(watch.disposed()).toBe(true); + }); + + it('does not fire after the service is disposed', () => { + const { svc, watch, events } = makeSession(); + svc.setWatchedPaths(['.']); + watch.fire('a.ts', 'created'); + (svc as unknown as { dispose: () => void }).dispose(); + vi.advanceTimersByTime(200); + expect(events).toHaveLength(0); + }); +}); diff --git a/packages/kap-server/src/start.ts b/packages/kap-server/src/start.ts index 5932c20e0..da36665b9 100644 --- a/packages/kap-server/src/start.ts +++ b/packages/kap-server/src/start.ts @@ -48,6 +48,7 @@ import { import { registerWs, WS_PATH as WS_PATH_V2 } from './transport/ws/registerWs'; import { extractWsBearerToken } from './transport/ws/bearerProtocol'; import { SessionEventBroadcaster } from './transport/ws/v1/sessionEventBroadcaster'; +import { FsWatchBridge } from './transport/ws/v1/fsWatchBridge'; import { registerWsV1, WS_PATH as WS_PATH_V1 } from './transport/ws/v1/registerWsV1'; import { getServerVersion } from './version'; import { classify } from './security/bindClassify'; @@ -297,6 +298,7 @@ export async function startServer(opts: ServerStartOptions = {}): Promise; +} + +interface SessionWatch { + readonly id: string; + readonly session: ISessionScopeHandle; + readonly fsWatch: ISessionFsWatchService; + readonly workspace: ISessionWorkspaceContext; + readonly conns: Map; + union: Set; + seq: number; + sub: IDisposable | undefined; +} + +export class FsWatchBridge { + private readonly core: Scope; + private readonly logger: JournalLogger | undefined; + private readonly bySession = new Map(); + private readonly connPathCount = new Map(); + + constructor(opts: { core: Scope; logger?: JournalLogger }) { + this.core = opts.core; + this.logger = opts.logger; + } + + async addWatch( + conn: FsWatchConnection, + sessionId: string, + rawPaths: readonly string[], + ): Promise { + const resolved = this.resolveSession(sessionId); + if (resolved === undefined) { + return { code: FS_WATCH_CODE.SESSION_NOT_FOUND, msg: 'session not found' }; + } + const sw = resolved; + + const normalized: string[] = []; + for (const raw of rawPaths) { + const rel = this.normalize(sw, raw); + if (rel === undefined) { + return { code: FS_WATCH_CODE.PATH_ESCAPES, msg: 'fs.path_escapes_session' }; + } + normalized.push(rel); + } + + let entry = sw.conns.get(conn.id); + const toAdd: string[] = []; + for (const rel of normalized) { + if (entry?.paths.has(rel)) continue; + toAdd.push(rel); + } + const current = this.connPathCount.get(conn.id) ?? 0; + if (current + toAdd.length > MAX_PATHS_PER_CONNECTION) { + return { code: FS_WATCH_CODE.LIMIT_EXCEEDED, msg: 'fs.watch_limit_exceeded' }; + } + + if (entry === undefined) { + entry = { conn, paths: new Set() }; + sw.conns.set(conn.id, entry); + } + for (const rel of toAdd) entry.paths.add(rel); + this.connPathCount.set(conn.id, current + toAdd.length); + this.recomputeAndApply(sw); + + return this.ok(sw, conn); + } + + async removeWatch( + conn: FsWatchConnection, + sessionId: string, + rawPaths: readonly string[], + ): Promise { + const sw = this.bySession.get(sessionId); + const entry = sw?.conns.get(conn.id); + if (sw === undefined || entry === undefined) { + return { code: FS_WATCH_CODE.OK, msg: 'success', watched_paths: [], current_count: this.countFor(conn.id) }; + } + + let removed = 0; + for (const raw of rawPaths) { + const rel = this.normalize(sw, raw) ?? raw; + if (entry.paths.delete(rel)) removed += 1; + } + this.connPathCount.set(conn.id, Math.max(0, this.countFor(conn.id) - removed)); + if (entry.paths.size === 0) sw.conns.delete(conn.id); + this.recomputeAndApply(sw); + if (sw.conns.size === 0) this.teardownSession(sw); + + return this.ok(sw, conn); + } + + /** Drop every subscription held by `conn` (called on socket close). */ + detachConnection(conn: FsWatchConnection): void { + for (const sw of Array.from(this.bySession.values())) { + const entry = sw.conns.get(conn.id); + if (entry === undefined) continue; + sw.conns.delete(conn.id); + this.connPathCount.set(conn.id, Math.max(0, this.countFor(conn.id) - entry.paths.size)); + this.recomputeAndApply(sw); + if (sw.conns.size === 0) this.teardownSession(sw); + } + this.connPathCount.delete(conn.id); + } + + private resolveSession(sessionId: string): SessionWatch | undefined { + const existing = this.bySession.get(sessionId); + if (existing !== undefined) return existing; + const session = this.core.accessor.get(ISessionLifecycleService).get(sessionId); + if (session === undefined) return undefined; + const sw: SessionWatch = { + id: sessionId, + session, + fsWatch: session.accessor.get(ISessionFsWatchService), + workspace: session.accessor.get(ISessionWorkspaceContext), + conns: new Map(), + union: new Set(), + seq: 0, + sub: undefined, + }; + this.bySession.set(sessionId, sw); + return sw; + } + + private recomputeAndApply(sw: SessionWatch): void { + const union = new Set(); + for (const { paths } of sw.conns.values()) { + for (const p of paths) union.add(p); + } + sw.union = union; + sw.fsWatch.setWatchedPaths([...union]); + if (union.size > 0 && sw.sub === undefined) { + sw.sub = sw.fsWatch.onDidChangeFiles((ev) => this.onSessionEvent(sw.id, ev)); + } + } + + private teardownSession(sw: SessionWatch): void { + sw.sub?.dispose(); + sw.sub = undefined; + sw.fsWatch.setWatchedPaths([]); + this.bySession.delete(sw.id); + } + + private onSessionEvent(sessionId: string, ev: FsChangeEvent): void { + const sw = this.bySession.get(sessionId); + if (sw === undefined) return; + for (const { conn, paths } of sw.conns.values()) { + let changes: FsChangeEntry[]; + if (ev.truncated === true) { + changes = []; + } else { + changes = ev.changes.filter((c) => isUnderAny(c.path, paths)); + if (changes.length === 0) continue; + } + sw.seq += 1; + const frame: FsChangedFrame = { + type: 'event.fs.changed', + seq: sw.seq, + session_id: sessionId, + timestamp: new Date().toISOString(), + payload: { + changes, + coalesced_window_ms: ev.coalesced_window_ms, + ...(ev.truncated === true ? { truncated: true, count: ev.count } : {}), + }, + }; + try { + conn.send(frame as EventEnvelope); + } catch (error) { + this.logger?.warn({ sessionId, err: String(error) }, 'fs-watch send failed'); + } + } + } + + /** Lexical confinement + workspace-relative normalization (no `stat`). */ + private normalize(sw: SessionWatch, raw: string): string | undefined { + if (raw === '' || raw === '/') return undefined; + if (isAbsolute(raw)) return undefined; + if (raw.split(/[/\\]+/).some((s) => s === '..')) return undefined; + const abs = sw.workspace.resolve(raw); + if (!sw.workspace.isWithin(abs)) return undefined; + return toRel(sw.workspace.workDir, abs); + } + + private ok(sw: SessionWatch, conn: FsWatchConnection): FsWatchAck { + const entry = sw.conns.get(conn.id); + return { + code: FS_WATCH_CODE.OK, + msg: 'success', + watched_paths: entry === undefined ? [] : [...entry.paths].sort(), + current_count: this.countFor(conn.id), + }; + } + + private countFor(connId: string): number { + return this.connPathCount.get(connId) ?? 0; + } +} + +function toRel(cwd: string, abs: string): string { + if (abs === cwd) return '.'; + const rel = relative(cwd, abs); + if (rel === '') return '.'; + return rel.split(sep).join('/'); +} + +function isUnderAny(rel: string, parents: ReadonlySet): boolean { + for (const parent of parents) { + if (parent === '.' || parent === '') return true; + if (rel === parent) return true; + if (rel.startsWith(`${parent}/`)) return true; + } + return false; +} diff --git a/packages/kap-server/src/transport/ws/v1/registerWsV1.ts b/packages/kap-server/src/transport/ws/v1/registerWsV1.ts index e7f03cd6a..9ccd784a8 100644 --- a/packages/kap-server/src/transport/ws/v1/registerWsV1.ts +++ b/packages/kap-server/src/transport/ws/v1/registerWsV1.ts @@ -14,6 +14,7 @@ import { WebSocketServer } from 'ws'; import type { CredentialValidator } from '../../../services/auth/credentials'; import { type IConnectionRegistry } from '../connectionRegistry'; import type { SessionEventBroadcaster } from './sessionEventBroadcaster'; +import type { FsWatchBridge } from './fsWatchBridge'; import type { JournalLogger } from './sessionEventJournal'; import { WsConnectionV1 } from './wsConnectionV1'; import { selectWsBearerProtocol } from '../bearerProtocol'; @@ -25,6 +26,7 @@ export interface RegisterWsV1Options { readonly validateCredential?: CredentialValidator; readonly registry: IConnectionRegistry; readonly broadcaster: SessionEventBroadcaster; + readonly fsWatchBridge: FsWatchBridge; readonly logger?: JournalLogger; readonly pingIntervalMs?: number; readonly pongTimeoutMs?: number; @@ -43,6 +45,7 @@ export function registerWsV1(core: Scope, opts: RegisterWsV1Options): WebSocketS const conn = new WsConnectionV1({ socket, broadcaster, + fsWatchBridge: opts.fsWatchBridge, connectionRegistry: registry, validateCredential: opts.validateCredential, remoteAddress: req.socket.remoteAddress ?? null, diff --git a/packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts b/packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts index ba4b5ebd2..ea4ecaece 100644 --- a/packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts +++ b/packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts @@ -34,6 +34,7 @@ import { type ResyncReason, type SessionEventBroadcaster, } from './sessionEventBroadcaster'; +import { FsWatchBridge } from './fsWatchBridge'; const DEFAULT_PING_INTERVAL_MS = 30_000; const DEFAULT_PONG_TIMEOUT_MS = 10_000; @@ -57,6 +58,7 @@ interface InboundFrame { export interface WsConnectionV1Options { readonly socket: WebSocket; readonly broadcaster: SessionEventBroadcaster; + readonly fsWatchBridge?: FsWatchBridge; readonly connectionRegistry: IConnectionRegistry; /** * Present-only credential check for the post-connect `client_hello` @@ -88,6 +90,7 @@ export class WsConnectionV1 implements BroadcastTarget { private readonly socket: WebSocket; private readonly broadcaster: SessionEventBroadcaster; + private readonly fsWatchBridge?: FsWatchBridge; private readonly validateCredential?: CredentialValidator; private readonly pingIntervalMs: number; private readonly pongTimeoutMs: number; @@ -119,6 +122,7 @@ export class WsConnectionV1 implements BroadcastTarget { this.userAgent = opts.userAgent; this.socket = opts.socket; this.broadcaster = opts.broadcaster; + this.fsWatchBridge = opts.fsWatchBridge; this.validateCredential = opts.validateCredential; this.logger = opts.logger; this.pingIntervalMs = opts.pingIntervalMs ?? DEFAULT_PING_INTERVAL_MS; @@ -178,12 +182,18 @@ export class WsConnectionV1 implements BroadcastTarget { case 'unsubscribe': void this.onUnsubscribe(frame); return; + case 'watch_fs_add': + void this.onWatchFs(frame, true); + return; + case 'watch_fs_remove': + void this.onWatchFs(frame, false); + return; case 'pong': this.onPong(); return; default: - // Unknown / not-yet-implemented control frame (e.g. terminal_*, abort, - // watch_fs_*) — ignore for now; terminal/abort stay on REST. + // Unknown / not-yet-implemented control frame (e.g. terminal_*, abort) + // — ignore for now; terminal/abort stay on REST. return; } } @@ -276,6 +286,36 @@ export class WsConnectionV1 implements BroadcastTarget { ); } + private async onWatchFs(frame: InboundFrame, isAdd: boolean): Promise { + const payload = frame.payload ?? {}; + const sessionId = typeof payload['session_id'] === 'string' ? payload['session_id'] : ''; + const paths = asStringArray(payload['paths']); + const bridge = this.fsWatchBridge; + if (bridge === undefined) { + this.sendFrame(buildAck(frame.id ?? '', 1, 'fs watch unavailable', {})); + return; + } + let result; + try { + result = isAdd + ? await bridge.addWatch(this, sessionId, paths) + : await bridge.removeWatch(this, sessionId, paths); + } catch (error) { + this.sendFrame( + buildAck(frame.id ?? '', 1, 'internal error', { + message: error instanceof Error ? error.message : String(error), + }), + ); + return; + } + this.sendFrame( + buildAck(frame.id ?? '', result.code, result.msg, { + watched_paths: result.watched_paths ?? [], + current_count: result.current_count ?? 0, + }), + ); + } + private async attachSession( sid: string, cursor: SessionCursor | undefined, @@ -461,6 +501,7 @@ export class WsConnectionV1 implements BroadcastTarget { if (this.backpressureRetryTimer !== undefined) clearTimeout(this.backpressureRetryTimer); this.outbound = []; for (const sid of this.subscriptions.keys()) this.broadcaster.unsubscribe(sid, this); + this.fsWatchBridge?.detachConnection(this); // registry removal is handled by registerWsV1 on the socket 'close' event. } } diff --git a/packages/kap-server/test/fs-watch.e2e.test.ts b/packages/kap-server/test/fs-watch.e2e.test.ts new file mode 100644 index 000000000..af56ac736 --- /dev/null +++ b/packages/kap-server/test/fs-watch.e2e.test.ts @@ -0,0 +1,398 @@ +/** + * `event.fs.changed` end-to-end for kap-server (server-v2). + * + * Mirrors `packages/server/test/fs-watch.e2e.test.ts` (v1) so the wire contract + * stays byte-compatible: + * 1. subscribe `src` → create file → receive `event.fs.changed` + * 2. burst > 500 changes / 200ms → `truncated` event + * 3. two clients, disjoint paths → no cross-delivery + * 4. > 100 paths per connection → `42902 fs.watch_limit_exceeded` + * 5. idempotent add of the same path + * 6. `watch_fs_remove` updates `watched_paths`; `..` → `41304` + * + * Boots `startServer` in-process (loopback, auth disabled) against a tmp + * workspace, drives `/api/v1/ws` clients with the raw `ws` library, and mutates + * the filesystem to trigger chokidar events. + */ + +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { pino } from 'pino'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { WebSocket, type RawData } from 'ws'; + +import { startServer, type RunningServer } from '../src/start'; + +let tmpDir: string; +let lockPath: string; +let bridgeHome: string; +let workspace: string; +let server: RunningServer | undefined; + +beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), 'kap-fswatch-')); + lockPath = join(tmpDir, 'lock'); + bridgeHome = mkdtempSync(join(tmpdir(), 'kap-fswatch-home-')); + workspace = join(tmpDir, 'workspace'); + mkdirSync(workspace, { recursive: true }); + mkdirSync(join(workspace, 'src'), { recursive: true }); + mkdirSync(join(workspace, 'docs'), { recursive: true }); +}); + +afterEach(async () => { + try { + await server?.close(); + } catch { + // ignore + } + server = undefined; + rmSync(tmpDir, { recursive: true, force: true }); + rmSync(bridgeHome, { recursive: true, force: true }); +}); + +async function boot(): Promise { + server = await startServer({ + host: '127.0.0.1', + port: 0, + homeDir: bridgeHome, + lockPath, + logger: pino({ level: 'silent' }), + disableAuth: true, + }); + return server; +} + +function addressOf(r: RunningServer): string { + return `http://${r.host}:${r.port}`; +} + +function wsUrl(r: RunningServer): string { + return `${addressOf(r).replace(/^http/, 'ws')}/api/v1/ws`; +} + +async function createSession(r: RunningServer): Promise { + const res = await fetch(`${addressOf(r)}/api/v1/sessions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ metadata: { cwd: workspace } }), + }); + const env = (await res.json()) as { code: number; data: { id: string } | null }; + if (env.code !== 0 || env.data === null) { + throw new Error(`create session failed: ${JSON.stringify(env)}`); + } + return env.data.id; +} + +interface WsFrame { + type: string; + payload?: Record; + id?: string; + code?: number; + msg?: string; + seq?: number; + session_id?: string; +} + +interface Conn { + ws: WebSocket; + queue: WsFrame[]; + waiters: Array<(frame: WsFrame) => void>; +} + +function rawToString(data: RawData): string { + if (typeof data === 'string') return data; + if (Buffer.isBuffer(data)) return data.toString('utf8'); + if (Array.isArray(data)) return Buffer.concat(data).toString('utf8'); + return Buffer.from(data as ArrayBuffer).toString('utf8'); +} + +function openConn(url: string): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(url); + const queue: WsFrame[] = []; + const waiters: Array<(frame: WsFrame) => void> = []; + ws.on('message', (data) => { + let parsed: WsFrame; + try { + parsed = JSON.parse(rawToString(data)) as WsFrame; + } catch { + return; + } + if (waiters.length > 0) waiters.shift()?.(parsed); + else queue.push(parsed); + }); + ws.once('open', () => resolve({ ws, queue, waiters })); + ws.once('error', (err) => reject(err)); + }); +} + +function receive(conn: Conn, timeoutMs: number): Promise { + return new Promise((resolve, reject) => { + if (conn.queue.length > 0) { + resolve(conn.queue.shift()!); + return; + } + const t = setTimeout(() => { + const idx = conn.waiters.indexOf(waiter); + if (idx >= 0) conn.waiters.splice(idx, 1); + reject(new Error(`no message in ${timeoutMs}ms`)); + }, timeoutMs); + const waiter = (frame: WsFrame): void => { + clearTimeout(t); + resolve(frame); + }; + conn.waiters.push(waiter); + }); +} + +async function receiveType(conn: Conn, type: string, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + const remaining = deadline - Date.now(); + if (remaining <= 0) throw new Error(`no message of type ${type} within ${timeoutMs}ms`); + const frame = await receive(conn, remaining); + if (frame.type === type) return frame; + } +} + +async function helloAndSubscribe(conn: Conn, clientId: string, sessionId: string): Promise { + await receiveType(conn, 'server_hello', 1000); + conn.ws.send( + JSON.stringify({ + type: 'client_hello', + id: `cli_${clientId}`, + payload: { client_id: clientId, subscriptions: [sessionId] }, + }), + ); + await receiveType(conn, 'ack', 1000); +} + +const sleep = (ms: number): Promise => new Promise((r) => setTimeout(r, ms)); + +/** Time given to chokidar to register newly-watched paths before mutating. */ +const WATCH_SETTLE_MS = 150; + +describe('WS fs watch (kap-server)', () => { + it('subscribe src → create file → receive event.fs.changed', async () => { + const r = await boot(); + const sid = await createSession(r); + const conn = await openConn(wsUrl(r)); + await helloAndSubscribe(conn, 'A', sid); + + conn.ws.send( + JSON.stringify({ + type: 'watch_fs_add', + id: 'w1', + payload: { session_id: sid, paths: ['src'] }, + }), + ); + const ack = await receiveType(conn, 'ack', 1000); + expect(ack.code).toBe(0); + expect(ack.payload).toMatchObject({ watched_paths: ['src'] }); + + await sleep(WATCH_SETTLE_MS); + writeFileSync(join(workspace, 'src', 'new.ts'), 'export const x = 1;\n'); + + const ev = await receiveType(conn, 'event.fs.changed', 2000); + expect(ev.session_id).toBe(sid); + const payload = ev.payload as { + changes: Array<{ path: string; change: string; kind: string }>; + coalesced_window_ms: number; + truncated?: boolean; + }; + expect(payload.coalesced_window_ms).toBe(200); + expect(payload.truncated).toBeUndefined(); + expect(payload.changes.length).toBeGreaterThanOrEqual(1); + const paths = payload.changes.map((c) => c.path); + expect(paths.some((p) => p === 'src/new.ts' || p === 'src')).toBe(true); + + conn.ws.close(); + }); + + it.skipIf(process.platform === 'win32')( + 'burst > 500 changes inside 200ms window → truncated:true', + { timeout: 5000 }, + async () => { + const r = await boot(); + const sid = await createSession(r); + const conn = await openConn(wsUrl(r)); + await helloAndSubscribe(conn, 'A', sid); + + conn.ws.send( + JSON.stringify({ + type: 'watch_fs_add', + id: 'w2', + payload: { session_id: sid, paths: ['.'] }, + }), + ); + await receiveType(conn, 'ack', 1000); + await sleep(WATCH_SETTLE_MS); + + const burstDir = join(workspace, 'burst'); + mkdirSync(burstDir, { recursive: true }); + for (let i = 0; i < 600; i++) writeFileSync(join(burstDir, `f${i}.txt`), `x${i}`); + + const deadline = Date.now() + 4000; + let sawTruncated = false; + while (Date.now() < deadline) { + let frame: WsFrame; + try { + frame = await receive(conn, deadline - Date.now()); + } catch { + break; + } + if (frame.type !== 'event.fs.changed') continue; + const payload = frame.payload as { truncated?: boolean; count?: number }; + if (payload.truncated === true) { + expect(payload.count).toBeGreaterThan(500); + sawTruncated = true; + break; + } + } + expect(sawTruncated).toBe(true); + conn.ws.close(); + }, + ); + + it('two clients on disjoint paths receive only their own changes', async () => { + const r = await boot(); + const sid = await createSession(r); + const a = await openConn(wsUrl(r)); + const b = await openConn(wsUrl(r)); + await helloAndSubscribe(a, 'A', sid); + await helloAndSubscribe(b, 'B', sid); + + a.ws.send( + JSON.stringify({ type: 'watch_fs_add', id: 'wA', payload: { session_id: sid, paths: ['src'] } }), + ); + await receiveType(a, 'ack', 1000); + b.ws.send( + JSON.stringify({ type: 'watch_fs_add', id: 'wB', payload: { session_id: sid, paths: ['docs'] } }), + ); + await receiveType(b, 'ack', 1000); + + await sleep(WATCH_SETTLE_MS); + writeFileSync(join(workspace, 'src', 'a.ts'), 'a'); + writeFileSync(join(workspace, 'docs', 'b.md'), 'b'); + + const evA = await receiveType(a, 'event.fs.changed', 2000); + const pathsA = (evA.payload as { changes: Array<{ path: string }> }).changes.map((c) => c.path); + expect(pathsA.some((p) => p.startsWith('src/'))).toBe(true); + expect(pathsA.some((p) => p.startsWith('docs/'))).toBe(false); + + const evB = await receiveType(b, 'event.fs.changed', 2000); + const pathsB = (evB.payload as { changes: Array<{ path: string }> }).changes.map((c) => c.path); + expect(pathsB.some((p) => p.startsWith('docs/'))).toBe(true); + expect(pathsB.some((p) => p.startsWith('src/'))).toBe(false); + + a.ws.close(); + b.ws.close(); + }); + + it('> 100 paths on one connection → 42902 fs.watch_limit_exceeded', async () => { + const r = await boot(); + const sid = await createSession(r); + const conn = await openConn(wsUrl(r)); + await helloAndSubscribe(conn, 'A', sid); + + const paths: string[] = []; + for (let i = 0; i < 101; i++) { + const p = `dir${i}`; + mkdirSync(join(workspace, p), { recursive: true }); + paths.push(p); + } + + conn.ws.send( + JSON.stringify({ + type: 'watch_fs_add', + id: 'w100', + payload: { session_id: sid, paths: paths.slice(0, 100) }, + }), + ); + const ack100 = await receiveType(conn, 'ack', 2000); + expect(ack100.code).toBe(0); + expect((ack100.payload as { current_count: number }).current_count).toBe(100); + + conn.ws.send( + JSON.stringify({ + type: 'watch_fs_add', + id: 'w101', + payload: { session_id: sid, paths: [paths[100]!] }, + }), + ); + const ack101 = await receiveType(conn, 'ack', 2000); + expect(ack101.code).toBe(42902); + + conn.ws.close(); + }); + + it('idempotent: adding the same path twice keeps current_count singular', async () => { + const r = await boot(); + const sid = await createSession(r); + const conn = await openConn(wsUrl(r)); + await helloAndSubscribe(conn, 'A', sid); + + conn.ws.send( + JSON.stringify({ type: 'watch_fs_add', id: 'w1', payload: { session_id: sid, paths: ['src'] } }), + ); + await receiveType(conn, 'ack', 1000); + conn.ws.send( + JSON.stringify({ type: 'watch_fs_add', id: 'w2', payload: { session_id: sid, paths: ['src'] } }), + ); + const ack = await receiveType(conn, 'ack', 1000); + expect((ack.payload as { current_count: number }).current_count).toBe(1); + + conn.ws.close(); + }); + + it('watch_fs_remove drops the subscription and acks updated watched_paths', async () => { + const r = await boot(); + const sid = await createSession(r); + const conn = await openConn(wsUrl(r)); + await helloAndSubscribe(conn, 'A', sid); + + conn.ws.send( + JSON.stringify({ + type: 'watch_fs_add', + id: 'wadd', + payload: { session_id: sid, paths: ['src', 'docs'] }, + }), + ); + await receiveType(conn, 'ack', 1000); + + conn.ws.send( + JSON.stringify({ + type: 'watch_fs_remove', + id: 'wrm', + payload: { session_id: sid, paths: ['src'] }, + }), + ); + const ack = await receiveType(conn, 'ack', 1000); + const payload = ack.payload as { watched_paths: string[]; current_count: number }; + expect(payload.watched_paths).toEqual(['docs']); + expect(payload.current_count).toBe(1); + + conn.ws.close(); + }); + + it('watch_fs_add for `..` path → 41304 fs.path_escapes_session', async () => { + const r = await boot(); + const sid = await createSession(r); + const conn = await openConn(wsUrl(r)); + await helloAndSubscribe(conn, 'A', sid); + + conn.ws.send( + JSON.stringify({ + type: 'watch_fs_add', + id: 'wbad', + payload: { session_id: sid, paths: ['../escape'] }, + }), + ); + const ack = await receiveType(conn, 'ack', 1000); + expect(ack.code).toBe(41304); + + conn.ws.close(); + }); +});