feat(agent-core-v2): watch user-level skill roots so the catalog stays fresh

Author: elkaix

.changeset/watch-user-skill-roots.md

@@ -0,0 +1,5 @@
+---
+"@pymodel/pythinker-code": minor
+---
+
+Refresh the skill catalog automatically when user-level skills are created, changed, or deleted while Pythinker Code is running.

packages/agent-core-v2/src/features/skill/catalog/userFileSkillSource.ts

@@ -1,10 +1,16 @@
+import { existsSync } from 'node:fs';
+
+import { join } from 'pathe';
 import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation';
-import { Disposable } from '#/_base/di/lifecycle';
+import { Disposable, DisposableStore } from '#/_base/di/lifecycle';
 import { Emitter, type Event } from '#/_base/event';
+import { TimeoutTimer } from '#/_base/utils/timer';
+import { subtreeWatchFilter } from '#/_base/utils/paths';
 import { LifecycleScope } from '#/app/scopes';
 import { ScopeActivation, registerScopedService } from '#/_base/di/scope';
 import { IBootstrapService } from '#/app/bootstrap/bootstrap';
 import { IConfigService } from '#/app/config/config';
+import { IHostFsWatchService } from '#/os/interface/hostFsWatch';
 
 import {
   MERGE_ALL_AVAILABLE_SKILLS_SECTION,
@@ -21,28 +27,38 @@ export interface IUserFileSkillSource extends ISkillSource {
 export const IUserFileSkillSource: ServiceIdentifier<IUserFileSkillSource> =
   createDecorator<IUserFileSkillSource>('userFileSkillSource');
 
+const WATCH_DEBOUNCE_MS = 200;
+
 export class UserFileSkillSource extends Disposable implements IUserFileSkillSource {
   declare readonly _serviceBrand: undefined;
 
   readonly id = 'user';
   readonly priority = SKILL_SOURCE_PRIORITY.user;
   private readonly onDidChangeEmitter = this._register(new Emitter<void>());
   readonly onDidChange: Event<void> = this.onDidChangeEmitter.event;
+  private readonly watchDebounce = this._register(new TimeoutTimer());
+  private readonly watchResources = this._register(new DisposableStore());
+  private watchReady: Promise<void> = Promise.resolve();
 
   constructor(
     @ISkillDiscovery private readonly discovery: ISkillDiscovery,
     @IBootstrapService private readonly bootstrap: IBootstrapService,
     @IConfigService private readonly config: IConfigService,
+    @IHostFsWatchService private readonly fsWatch: IHostFsWatchService,
   ) {
     super();
     this._register(
       this.config.onDidSectionChange((event) => {
         if (event.domain === MERGE_ALL_AVAILABLE_SKILLS_SECTION) this.onDidChangeEmitter.fire();
       }),
     );
+    if ((this.bootstrap.args.skillDirs?.length ?? 0) === 0) {
+      this.watchUserSkillRoots();
+    }
   }
 
   async load(): Promise<SkillContribution> {
+    await this.watchReady;
     if ((this.bootstrap.args.skillDirs?.length ?? 0) > 0) {
       return { skills: [] };
     }
@@ -53,6 +69,36 @@ export class UserFileSkillSource extends Disposable implements IUserFileSkillSou
       await userRoots(this.bootstrap.homeDir, this.bootstrap.osHomeDir, { mergeAllAvailableSkills }),
     );
   }
+
+  private watchUserSkillRoots(): void {
+    const candidatesByBase = new Map<string, string[]>();
+    for (const [base, root] of [
+      [this.bootstrap.homeDir, join(this.bootstrap.homeDir, 'skills')],
+      [this.bootstrap.osHomeDir, join(this.bootstrap.osHomeDir, '.agents', 'skills')],
+    ] as const) {
+      const existing = candidatesByBase.get(base);
+      if (existing === undefined) candidatesByBase.set(base, [root]);
+      else existing.push(root);
+    }
+    const ready: Promise<void>[] = [];
+    for (const [base, candidates] of candidatesByBase) {
+      if (!existsSync(base)) continue;
+      const handle = this.fsWatch.watch(base, {
+        ignored: subtreeWatchFilter(base, candidates),
+        signal: true,
+      });
+      this.watchResources.add(handle);
+      this.watchResources.add(
+        handle.onDidChange(() => {
+          this.watchDebounce.cancelAndSet(() => {
+            this.onDidChangeEmitter.fire();
+          }, WATCH_DEBOUNCE_MS);
+        }),
+      );
+      ready.push(handle.ready);
+    }
+    this.watchReady = Promise.allSettled(ready).then(() => undefined);
+  }
 }
 
 registerScopedService(

packages/agent-core-v2/src/os/backends/node-local/hostFsWatchService.ts

@@ -184,9 +184,13 @@ class SignalWatchHandle implements IHostFsWatchHandle {
       this.fireInvalidation();
       return;
     }
-    onUnexpectedError(error);
+    if (error.code === 'ENOENT') {
+      this.readiness.resolve();
+    } else {
+      onUnexpectedError(error);
+      this.fireInvalidation();
+    }
     this.recovering = true;
-    this.fireInvalidation();
     const delay = Math.min(NATIVE_RETRY_BASE_MS * 2 ** this.retryAttempts, NATIVE_RETRY_MAX_MS);
     this.retryAttempts += 1;
     this.retry?.dispose();

packages/agent-core-v2/src/program/program.ts

@@ -298,7 +298,7 @@ export class Program {
       const extraAgentProfiles = own(new ExtraAgentProfileLoaderService(this.dependencies.config, this.context, this.dependencies.bootstrap, runtime.fs!, this.dependencies.log, userAgentProfiles, this.dependencies.agentProfiles));
       const agentProfiles = own(new WorkspaceAgentProfileLoaderService(this.context, runtime.fs!, this.dependencies.log, userAgentProfiles, runtime.watch!, this.dependencies.agentProfiles));
       const skillDiscovery = new RuntimeSkillDiscovery(this.dependencies.log, runtime.fs!);
-      const userSkills = own(new UserFileSkillSource(skillDiscovery, this.dependencies.bootstrap, this.dependencies.config));
+      const userSkills = own(new UserFileSkillSource(skillDiscovery, this.dependencies.bootstrap, this.dependencies.config, runtime.watch!));
       const explicitSkills = new ExplicitFileSkillSource(skillDiscovery, this.context, this.dependencies.bootstrap);
       const extraSkills = own(new ExtraFileSkillSource(skillDiscovery, this.dependencies.config, this.context, this.dependencies.bootstrap));
       const workspaceSkills = own(new WorkspaceRootSkillSource(skillDiscovery, this.context, this.dependencies.config, this.dependencies.bootstrap, runtime.watch!));
@@ -329,7 +329,7 @@ export class Program {
         retired: false,
       };
     } catch (error) {
-      for (const disposable of disposables.reverse()) void disposable.dispose();
+      for (const disposable of disposables.toReversed()) void disposable.dispose();
       lease.dispose();
       throw error;
     }
@@ -368,7 +368,7 @@ export class Program {
   private releaseGeneration(generation: ProgramGeneration): void {
     generation.references -= 1;
     if (generation.references !== 0 || !generation.retired) return;
-    for (const disposable of [...generation.disposables].reverse()) void disposable.dispose();
+    for (const disposable of [...generation.disposables].toReversed()) void disposable.dispose();
     generation.lease.dispose();
   }
 

packages/agent-core-v2/test/app/bootstrap/stubs.ts

@@ -16,6 +16,7 @@ export function stubBootstrap(
   homeDir = '/tmp/pythinker-home',
   env: NodeJS.ProcessEnv = {},
   args: HostArgsInput = {},
+  osHomeDir = '/home/test',
 ): IBootstrapService {
   const scopes: Record<PersistenceScopeName, string> = {
     config: '',
@@ -31,7 +32,7 @@ export function stubBootstrap(
     platform: 'linux',
     arch: 'x64',
     cwd: '/tmp',
-    osHomeDir: '/home/test',
+    osHomeDir,
     homeDir,
     configPath: `${homeDir}/config.toml`,
     configKey: 'config.toml',

packages/agent-core-v2/test/features/skill/workspace/skillCatalog.test.ts

@@ -164,6 +164,32 @@ function fsWatchStub(
   };
 }
 
+function recordingWatchService(): {
+  service: IHostFsWatchService;
+  calls: { path: string; ignored: ((path: string) => boolean) | undefined }[];
+  handles: { disposed: boolean }[];
+} {
+  const calls: { path: string; ignored: ((path: string) => boolean) | undefined }[] = [];
+  const handles: { disposed: boolean }[] = [];
+  const service: IHostFsWatchService = {
+    _serviceBrand: undefined,
+    watch: (path, options) => {
+      calls.push({ path, ignored: options?.ignored });
+      const handle: IHostFsWatchHandle & { disposed: boolean } = {
+        ready: Promise.resolve(),
+        onDidChange: Event.None as Event<HostFsChange>,
+        disposed: false,
+        dispose: () => {
+          handle.disposed = true;
+        },
+      };
+      handles.push(handle);
+      return handle;
+    },
+  };
+  return { service, calls, handles };
+}
+
 function makeHost(
   store: ISkillDiscovery,
   ws: IWorkspaceContext,
@@ -1096,4 +1122,188 @@ describe('WorkspaceSkillCatalogService', () => {
       await rm(workDir, { recursive: true, force: true });
     }
   }, 15000);
+  it('watches both user-level skill roots and prunes unrelated paths', async () => {
+    const homeDir = await mkdtemp(join(tmpdir(), 'skill-user-home-'));
+    const osHomeDir = await mkdtemp(join(tmpdir(), 'skill-user-os-'));
+    await mkdir(join(homeDir, 'skills'), { recursive: true });
+    await mkdir(join(osHomeDir, '.agents', 'skills'), { recursive: true });
+    const { service, calls } = recordingWatchService();
+    const host = createScopedTestHost([
+      stubPair(IFlagService, stubFlag(true)),
+      stubPair(IBootstrapService, stubBootstrap(homeDir, {}, {}, osHomeDir)),
+      stubPair(IConfigService, configStub()),
+      stubPair(IPluginService, pluginStub()),
+      stubPair(ILogService, stubLog()),
+      stubPair(IHostFsWatchService, service),
+      stubPair(ISkillDiscovery, new FileSkillDiscovery(stubLog())),
+    ]);
+
+    try {
+      const source = host.app.accessor.get(IUserFileSkillSource);
+      await source.load();
+
+      const home = calls.find((call) => call.path === homeDir);
+      const osHome = calls.find((call) => call.path === osHomeDir);
+      expect(home).toBeDefined();
+      expect(osHome).toBeDefined();
+      expect(home?.ignored?.(join(homeDir, 'skills/demo/SKILL.md'))).toBe(false);
+      expect(home?.ignored?.(join(homeDir, 'sessions/s1/state.json'))).toBe(true);
+      expect(osHome?.ignored?.(join(osHomeDir, '.agents/skills/demo/SKILL.md'))).toBe(false);
+      expect(osHome?.ignored?.(join(osHomeDir, 'Downloads/x.zip'))).toBe(true);
+    } finally {
+      host.dispose();
+      await rm(homeDir, { recursive: true, force: true });
+      await rm(osHomeDir, { recursive: true, force: true });
+    }
+  });
+
+  it('merges both skill-root candidates into one watch when homeDir equals osHomeDir', async () => {
+    const homeDir = await mkdtemp(join(tmpdir(), 'skill-user-same-'));
+    await mkdir(join(homeDir, 'skills'), { recursive: true });
+    await mkdir(join(homeDir, '.agents', 'skills'), { recursive: true });
+    const { service, calls } = recordingWatchService();
+    const host = createScopedTestHost([
+      stubPair(IFlagService, stubFlag(true)),
+      stubPair(IBootstrapService, stubBootstrap(homeDir, {}, {}, homeDir)),
+      stubPair(IConfigService, configStub()),
+      stubPair(IPluginService, pluginStub()),
+      stubPair(ILogService, stubLog()),
+      stubPair(IHostFsWatchService, service),
+      stubPair(ISkillDiscovery, new FileSkillDiscovery(stubLog())),
+    ]);
+
+    try {
+      const source = host.app.accessor.get(IUserFileSkillSource);
+      await source.load();
+
+      const homeCalls = calls.filter((call) => call.path === homeDir);
+      expect(homeCalls).toHaveLength(1);
+      const ignored = homeCalls[0]?.ignored;
+      expect(ignored?.(join(homeDir, 'skills/demo/SKILL.md'))).toBe(false);
+      expect(ignored?.(join(homeDir, '.agents/skills/demo/SKILL.md'))).toBe(false);
+      expect(ignored?.(join(homeDir, 'sessions/s1/state.json'))).toBe(true);
+    } finally {
+      host.dispose();
+      await rm(homeDir, { recursive: true, force: true });
+    }
+  });
+
+  it('does not watch the user skill roots when explicit skillDirs are set', async () => {
+    const homeDir = await mkdtemp(join(tmpdir(), 'skill-user-explicit-'));
+    await mkdir(join(homeDir, 'skills'), { recursive: true });
+    const { service, calls } = recordingWatchService();
+    const host = createScopedTestHost([
+      stubPair(IFlagService, stubFlag(true)),
+      stubPair(IBootstrapService, stubBootstrap(homeDir, {}, { skillDirs: ['/explicit'] })),
+      stubPair(IConfigService, configStub()),
+      stubPair(IPluginService, pluginStub()),
+      stubPair(ILogService, stubLog()),
+      stubPair(IHostFsWatchService, service),
+      stubPair(ISkillDiscovery, new FileSkillDiscovery(stubLog())),
+    ]);
+
+    try {
+      const source = host.app.accessor.get(IUserFileSkillSource);
+      await source.load();
+      expect(calls.map((call) => call.path)).toEqual([]);
+    } finally {
+      host.dispose();
+      await rm(homeDir, { recursive: true, force: true });
+    }
+  });
+
+  it('disposes the user root watches when the app scope is disposed', async () => {
+    const homeDir = await mkdtemp(join(tmpdir(), 'skill-user-dispose-'));
+    await mkdir(join(homeDir, 'skills'), { recursive: true });
+    const { service, handles } = recordingWatchService();
+    const host = createScopedTestHost([
+      stubPair(IFlagService, stubFlag(true)),
+      stubPair(IBootstrapService, stubBootstrap(homeDir)),
+      stubPair(IConfigService, configStub()),
+      stubPair(IPluginService, pluginStub()),
+      stubPair(ILogService, stubLog()),
+      stubPair(IHostFsWatchService, service),
+      stubPair(ISkillDiscovery, new FileSkillDiscovery(stubLog())),
+    ]);
+
+    const source = host.app.accessor.get(IUserFileSkillSource);
+    await source.load();
+    expect(handles.length).toBeGreaterThan(0);
+
+    host.dispose();
+    expect(handles.every((handle) => handle.disposed)).toBe(true);
+    await rm(homeDir, { recursive: true, force: true });
+  });
+
+  it('rescans the user source when skills appear, change and disappear under the user roots', async () => {
+    const homeDir = await mkdtemp(join(tmpdir(), 'skill-user-watch-'));
+    const osHomeDir = await mkdtemp(join(tmpdir(), 'skill-os-watch-'));
+    const host = createScopedTestHost([
+      stubPair(IFlagService, stubFlag(true)),
+      stubPair(IBootstrapService, stubBootstrap(homeDir, {}, {}, osHomeDir)),
+      stubPair(IConfigService, configStub()),
+      stubPair(IPluginService, pluginStub()),
+      stubPair(ILogService, stubLog()),
+      stubPair(ISkillDiscovery, new FileSkillDiscovery(stubLog())),
+      stubPair(IHostFsWatchService, new HostFsWatchService()),
+    ]);
+    const workspace = host.child('program', 'w1', [
+      stubPair(IWorkspaceContext, workspaceContextStub('/work')),
+    ]);
+    const writeSkill = (dir: string, description: string) =>
+      writeFile(
+        join(dir, 'SKILL.md'),
+        `---\nname: watched-user-skill\ndescription: ${description}\n---\nbody`,
+        'utf8',
+      );
+
+    try {
+      const catalog = workspace.accessor.get(IWorkspaceSkillCatalog);
+      await catalog.load();
+      expect(catalog.catalog.getSkill('watched-user-skill')).toBeUndefined();
+
+      const waitForUserChange = (): Promise<string> => {
+        const refreshed = new Promise<string>((resolvePromise) => {
+          const d = catalog.onDidChange((sourceId) => {
+            if (sourceId !== 'user') return;
+            d.dispose();
+            resolvePromise(sourceId);
+          });
+        });
+        const timedOut = new Promise<never>((_resolve, reject) => {
+          setTimeout(() => reject(new Error('user watch refresh timed out')), 10000);
+        });
+        return Promise.race([refreshed, timedOut]);
+      };
+
+      const created = waitForUserChange();
+      const skillDir = join(homeDir, 'skills', 'watched-user-skill');
+      await mkdir(skillDir, { recursive: true });
+      await writeSkill(skillDir, 'v1');
+      await created;
+      expect(catalog.catalog.getSkill('watched-user-skill')?.description).toBe('v1');
+
+      const modified = waitForUserChange();
+      await writeSkill(skillDir, 'v2');
+      await modified;
+      expect(catalog.catalog.getSkill('watched-user-skill')?.description).toBe('v2');
+
+      const deleted = waitForUserChange();
+      await rm(skillDir, { recursive: true, force: true });
+      await deleted;
+      expect(catalog.catalog.getSkill('watched-user-skill')).toBeUndefined();
+
+      const osCreated = waitForUserChange();
+      const osSkillDir = join(osHomeDir, '.agents', 'skills', 'watched-user-skill');
+      await mkdir(osSkillDir, { recursive: true });
+      await writeSkill(osSkillDir, 'os');
+      await osCreated;
+      expect(catalog.catalog.getSkill('watched-user-skill')?.description).toBe('os');
+    } finally {
+      host.dispose();
+      await rm(homeDir, { recursive: true, force: true });
+      await rm(osHomeDir, { recursive: true, force: true });
+    }
+  }, 20000);
+
 });