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

@@ -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';

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<HostFsChange>;
+
+  private readonly emitter: Emitter<HostFsChange>;
+  private readonly watcher: FSWatcher;
+  private disposed = false;
+
+  constructor(path: string, options: HostFsWatchOptions | undefined) {
+    this.emitter = new Emitter<HostFsChange>();
+    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',
+);

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<HostFsChange>;
+}
+
+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<IHostFsWatchService> =
+  createDecorator<IHostFsWatchService>('hostFsWatchService');

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<FsChangeEvent>;
+}
+
+export const ISessionFsWatchService: ServiceIdentifier<ISessionFsWatchService> =
+  createDecorator<ISessionFsWatchService>('sessionFsWatchService');

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<FsChangeEvent>());
+  readonly onDidChangeFiles: Event<FsChangeEvent> = this.emitter.event;
+
+  private watched = new Set<string>();
+  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<string>();
+    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<string>): 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',
+);

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<void> => 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<HostFsChange[]> {
+    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);
+  });
+});

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);
+  });
+});

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<Runnin
     core,
     logger,
   });
+  const fsWatchBridge = new FsWatchBridge({ core, logger });
 
   const snapshotReader = new SnapshotReader({
     homeDir,
@@ -370,6 +372,7 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
     validateCredential,
     registry: connectionRegistry,
     broadcaster,
+    fsWatchBridge,
     logger,
   });
 

packages/kap-server/src/transport/ws/v1/fsWatchBridge.ts

@@ -0,0 +1,284 @@
+/**
+ * `FsWatchBridge` — volatile `/api/v1/ws` delivery for filesystem changes.
+ *
+ * Turns the core `ISessionFsWatchService.onDidChangeFiles` feed into
+ * `event.fs.changed` frames on the v1 WebSocket, byte-compatible with the v1
+ * server (`packages/server/.../fsWatcherService.ts`):
+ *
+ *   client → `{type:'watch_fs_add',    id, payload:{session_id, paths}}`
+ *   client → `{type:'watch_fs_remove', id, payload:{session_id, paths}}`
+ *   server → `{type:'ack', id, code, payload:{watched_paths, current_count}}`
+ *   server → `{type:'event.fs.changed', seq, session_id, timestamp, payload}`
+ *
+ * The bridge is transport state (like {@link ConnectionRegistry} /
+ * {@link SessionEventBroadcaster}); it is **not** DI-registered and carries no
+ * `_serviceBrand`. It owns the per-`(connection, session)` subscription sets,
+ * fans the core feed out to each connection filtered by that connection's
+ * paths, and assigns a per-session monotonic `seq`. Frames are sent straight
+ * to the socket — they never enter the broadcaster / journal (fs changes are
+ * volatile: on overflow the client sees `truncated` and re-syncs).
+ *
+ * The core `ISessionFsWatchService` keeps a single subscription set per
+ * session; the bridge drives it with the **union** of every connection's
+ * paths for that session, then re-filters per connection on the way out.
+ */
+
+import { isAbsolute, relative, sep } from 'node:path';
+
+import {
+  type IDisposable,
+  type ISessionScopeHandle,
+  ISessionFsWatchService,
+  ISessionLifecycleService,
+  ISessionWorkspaceContext,
+  type Scope,
+} from '@moonshot-ai/agent-core-v2';
+import type { FsChangeEntry, FsChangeEvent } from '@moonshot-ai/protocol';
+
+import type { EventEnvelope, JournalLogger } from './sessionEventJournal';
+
+const MAX_PATHS_PER_CONNECTION = 100;
+
+export const FS_WATCH_CODE = {
+  OK: 0,
+  PATH_ESCAPES: 41304,
+  LIMIT_EXCEEDED: 42902,
+  SESSION_NOT_FOUND: 40409,
+} as const;
+
+export interface FsChangedFrame {
+  readonly type: 'event.fs.changed';
+  readonly seq: number;
+  readonly session_id: string;
+  readonly timestamp: string;
+  readonly payload: FsChangeEvent;
+}
+
+/** Minimal connection surface the bridge needs (satisfied by `WsConnectionV1`). */
+export interface FsWatchConnection {
+  readonly id: string;
+  send(envelope: EventEnvelope): void;
+}
+
+export interface FsWatchAck {
+  readonly code: number;
+  readonly msg: string;
+  readonly watched_paths?: readonly string[];
+  readonly current_count?: number;
+}
+
+interface ConnEntry {
+  readonly conn: FsWatchConnection;
+  readonly paths: Set<string>;
+}
+
+interface SessionWatch {
+  readonly id: string;
+  readonly session: ISessionScopeHandle;
+  readonly fsWatch: ISessionFsWatchService;
+  readonly workspace: ISessionWorkspaceContext;
+  readonly conns: Map<string, ConnEntry>;
+  union: Set<string>;
+  seq: number;
+  sub: IDisposable | undefined;
+}
+
+export class FsWatchBridge {
+  private readonly core: Scope;
+  private readonly logger: JournalLogger | undefined;
+  private readonly bySession = new Map<string, SessionWatch>();
+  private readonly connPathCount = new Map<string, number>();
+
+  constructor(opts: { core: Scope; logger?: JournalLogger }) {
+    this.core = opts.core;
+    this.logger = opts.logger;
+  }
+
+  async addWatch(
+    conn: FsWatchConnection,
+    sessionId: string,
+    rawPaths: readonly string[],
+  ): Promise<FsWatchAck> {
+    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<FsWatchAck> {
+    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<string>();
+    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<string>): boolean {
+  for (const parent of parents) {
+    if (parent === '.' || parent === '') return true;
+    if (rel === parent) return true;
+    if (rel.startsWith(`${parent}/`)) return true;
+  }
+  return false;
+}

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,

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<void> {
+    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.
   }
 }

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<RunningServer> {
+  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<string> {
+  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<string, unknown>;
+  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<Conn> {
+  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<WsFrame> {
+  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<WsFrame> {
+  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<void> {
+  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<void> => 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();
+  });
+});