From 187e78e1fb0139f6e0ede2a9347cecca300b4b89 Mon Sep 17 00:00:00 2001 From: morluto Date: Wed, 1 Jul 2026 13:23:18 +0000 Subject: [PATCH] fix(agent-core): release filesystem watchers on session close --- .changeset/session-fs-watchers.md | 5 ++ .../src/services/fs/fsWatcherService.ts | 20 ++++++- packages/server/test/services.test.ts | 52 +++++++++++++++++-- 3 files changed, 73 insertions(+), 4 deletions(-) create mode 100644 .changeset/session-fs-watchers.md diff --git a/.changeset/session-fs-watchers.md b/.changeset/session-fs-watchers.md new file mode 100644 index 000000000..6263bb8a8 --- /dev/null +++ b/.changeset/session-fs-watchers.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Release session filesystem watchers when sessions close. diff --git a/packages/agent-core/src/services/fs/fsWatcherService.ts b/packages/agent-core/src/services/fs/fsWatcherService.ts index 8811c3a47..1e17a3179 100644 --- a/packages/agent-core/src/services/fs/fsWatcherService.ts +++ b/packages/agent-core/src/services/fs/fsWatcherService.ts @@ -109,7 +109,7 @@ export class FsWatcherService extends Disposable implements IFsWatcher { private readonly lookup: FsWatcherConnectionLookup, options: FsWatcherServiceOptions, @ILogService private readonly logger: ILogService, - @ISessionService _sessionService: ISessionService, + @ISessionService sessionService: ISessionService, ) { super(); this.sessions = this._register(new DisposableMap()); @@ -126,6 +126,11 @@ export class FsWatcherService extends Disposable implements IFsWatcher { persistent: false, ignored: (p: string) => /(?:^|[/\\])\.git(?:$|[/\\])/.test(p), })); + this._register( + sessionService.onDidClose(({ sessionId }) => { + this.disposeSessionEntry(sessionId); + }), + ); } addPaths( @@ -257,6 +262,19 @@ export class FsWatcherService extends Disposable implements IFsWatcher { } } + private disposeSessionEntry(sessionId: string): void { + const entry = this.sessions.get(sessionId); + if (!entry) return; + for (const connectionId of entry.connectionPathRefs.keys()) { + const connSessions = this.connections.get(connectionId); + connSessions?.delete(sessionId); + if (connSessions !== undefined && connSessions.size === 0) { + this.connections.delete(connectionId); + } + } + this.sessions.deleteAndDispose(sessionId); + } + private getOrCreateConnection( connectionId: string, ): Map>> { diff --git a/packages/server/test/services.test.ts b/packages/server/test/services.test.ts index 186338697..41350b3d8 100644 --- a/packages/server/test/services.test.ts +++ b/packages/server/test/services.test.ts @@ -7,7 +7,7 @@ import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { InstantiationService, ServiceCollection, EventService, FsWatcherService, IApprovalService, IEventService, ILogService, IQuestionService, type ApprovalResponse, type QuestionResult, type FsWatcherServiceOptions, type IEnvironmentService, type ILogService as ILoggerT, type ISessionService } from '@moonshot-ai/agent-core'; +import { Emitter, InstantiationService, ServiceCollection, EventService, FsWatcherService, IApprovalService, IEventService, ILogService, IQuestionService, type ApprovalResponse, type QuestionResult, type FsWatcherServiceOptions, type IEnvironmentService, type ILogService as ILoggerT, type ISessionService } from '@moonshot-ai/agent-core'; import type { Event } from '@moonshot-ai/protocol'; import { ApprovalService } from '#/services/approval/approvalService'; @@ -135,6 +135,29 @@ class FakeWatcher { type TestFsWatcher = ReturnType>; +function makeSessionService( + closeEmitter = new Emitter<{ sessionId: string }>(), +): ISessionService { + const createEmitter = new Emitter(); + return { + _serviceBrand: undefined, + create: async () => { throw new Error('not implemented'); }, + list: async () => ({ items: [], has_more: false }), + get: async () => { throw new Error('not implemented'); }, + update: async () => { throw new Error('not implemented'); }, + fork: async () => { throw new Error('not implemented'); }, + listChildren: async () => ({ items: [], has_more: false }), + createChild: async () => { throw new Error('not implemented'); }, + getStatus: async () => { throw new Error('not implemented'); }, + getSessionWarnings: async () => [], + compact: async () => { throw new Error('not implemented'); }, + undo: async () => { throw new Error('not implemented'); }, + archive: async () => { throw new Error('not implemented'); }, + onDidCreate: createEmitter.event, + onDidClose: closeEmitter.event, + }; +} + let ix: InstantiationService; let testLogger: TestLogger; @@ -463,7 +486,7 @@ describe('FsWatcherService', () => { { resolve: () => undefined }, { watcherFactory: () => watcher as unknown as TestFsWatcher }, testLogger, - {} as ISessionService, + makeSessionService(), ); const path = '/workspace/src'; @@ -496,7 +519,7 @@ describe('FsWatcherService', () => { { resolve: () => undefined }, { watcherFactory: () => watcher as unknown as TestFsWatcher }, testLogger, - {} as ISessionService, + makeSessionService(), ); const paths = ['/workspace/src', '/workspace/docs', '/workspace/notes']; watcher.unwatchErrors.set(paths[0]!, new Error('unwatch-src')); @@ -517,6 +540,29 @@ describe('FsWatcherService', () => { expect(watcher.closeCalls).toBe(0); service.dispose(); }); + + it('releases a session watcher when the session closes', () => { + const watcher = new FakeWatcher(); + const closeEmitter = new Emitter<{ sessionId: string }>(); + const service = new FsWatcherService( + { resolve: () => undefined }, + { watcherFactory: () => watcher as unknown as TestFsWatcher }, + testLogger, + makeSessionService(closeEmitter), + ); + const path = '/workspace/src'; + + service.addPaths('sid', 'conn-a', [path]); + service.addPaths('sid', 'conn-b', [path]); + closeEmitter.fire({ sessionId: 'sid' }); + + expect(watcher.closeCalls).toBe(1); + expect(service.watchedPaths('conn-a', 'sid')).toEqual([]); + expect(service.watchedPaths('conn-b', 'sid')).toEqual([]); + expect(service.countForConnection('conn-a')).toBe(0); + expect(service.countForConnection('conn-b')).toBe(0); + service.dispose(); + }); }); describe('ApprovalService (broadcasts + resolve-by-approval_id)', () => {