From e76df691fe436493c28af81e5607a04e4736bfe4 Mon Sep 17 00:00:00 2001 From: Federico Kamelhar Date: Sun, 31 May 2026 14:35:42 -0400 Subject: [PATCH] fix(skills): bound watcher workspace state Bounds skills watcher subscriptions and workspace snapshot-version state to active workspaces on the current `src/skills/runtime` implementation. The fix keeps shared path watchers as the owner boundary, evicts idle workspace subscriptions after 1 hour without closing watchers still used by other workspaces, and clears per-workspace version keys only after preserving/advancing invalidation so cached skill snapshots cannot miss changes across teardown or re-enable. Thanks @fede-kamel. Fixes #77997. Co-authored-by: Federico Kamelhar --- src/skills/runtime/refresh-state.ts | 12 +++- src/skills/runtime/refresh.test.ts | 94 +++++++++++++++++++++++++++++ src/skills/runtime/refresh.ts | 47 ++++++++++++--- 3 files changed, 145 insertions(+), 8 deletions(-) diff --git a/src/skills/runtime/refresh-state.ts b/src/skills/runtime/refresh-state.ts index 90fe04886c21..3d1708e638f1 100644 --- a/src/skills/runtime/refresh-state.ts +++ b/src/skills/runtime/refresh-state.ts @@ -43,7 +43,7 @@ export function bumpSkillsSnapshotVersion(params?: { const reason = params?.reason ?? "manual"; const changedPath = params?.changedPath; if (params?.workspaceDir) { - const current = workspaceVersions.get(params.workspaceDir) ?? 0; + const current = Math.max(globalVersion, workspaceVersions.get(params.workspaceDir) ?? 0); const next = bumpVersion(current); workspaceVersions.set(params.workspaceDir, next); emit({ workspaceDir: params.workspaceDir, reason, changedPath }); @@ -62,6 +62,16 @@ export function getSkillsSnapshotVersion(workspaceDir?: string): number { return Math.max(globalVersion, local); } +export function clearSkillsSnapshotVersionForWorkspace(workspaceDir: string): void { + const local = workspaceVersions.get(workspaceDir); + if (typeof local === "number" && local > globalVersion) { + // Keep pending workspace invalidation visible after dropping the workspace + // key; otherwise teardown can hide a skill change from cached snapshots. + globalVersion = local; + } + workspaceVersions.delete(workspaceDir); +} + export function shouldRefreshSnapshotForVersion( cachedVersion?: number, nextVersion?: number, diff --git a/src/skills/runtime/refresh.test.ts b/src/skills/runtime/refresh.test.ts index a1f615719376..3bf96e24e91d 100644 --- a/src/skills/runtime/refresh.test.ts +++ b/src/skills/runtime/refresh.test.ts @@ -657,6 +657,7 @@ describe("ensureSkillsWatcher", () => { workspaceDir: "/tmp/ws-a", config: { skills: { load: { extraDirs: ["/tmp/shared"], watch: false } } }, }); + seen.length = 0; const callPaths = (watchMock.mock.calls as unknown as Array<[string]>).map((call) => call[0]); const sharedIndex = callPaths.findIndex((target) => target.includes("/tmp/shared")); @@ -673,6 +674,99 @@ describe("ensureSkillsWatcher", () => { expect(seen.some((change) => change.workspaceDir === "/tmp/ws-a")).toBe(false); }); + it("clears workspace version state on watch disable without losing pending invalidation", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00Z")); + const workspaceDir = "/tmp/workspace-version-cleanup"; + refreshModule.ensureSkillsWatcher({ + workspaceDir, + config: { skills: { load: { watchDebounceMs: 10 } } }, + }); + + const firstVersion = refreshModule.bumpSkillsSnapshotVersion({ + workspaceDir, + reason: "watch", + changedPath: `${workspaceDir}/skills/demo/SKILL.md`, + }); + refreshModule.ensureSkillsWatcher({ + workspaceDir, + config: { skills: { load: { watch: false } } }, + }); + + const nextVersion = refreshModule.getSkillsSnapshotVersion(workspaceDir); + expect(nextVersion).toBeGreaterThan(firstVersion); + expect(refreshModule.shouldRefreshSnapshotForVersion(firstVersion, nextVersion)).toBe(true); + vi.setSystemTime(new Date(nextVersion)); + const followupVersion = refreshModule.bumpSkillsSnapshotVersion({ + workspaceDir, + reason: "watch", + }); + expect(followupVersion).toBeGreaterThan(nextVersion); + }); + + it("evicts idle workspace subscriptions on a later ensure call", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00Z")); + const idleWorkspaceDir = "/tmp/workspace-idle"; + refreshModule.ensureSkillsWatcher({ + workspaceDir: idleWorkspaceDir, + config: { skills: { load: { watchDebounceMs: 10 } } }, + }); + const callPaths = (watchMock.mock.calls as unknown as Array<[string]>).map((call) => call[0]); + const idleSkillsIndex = callPaths.findIndex( + (target) => target === `${idleWorkspaceDir}/skills`, + ); + expect(idleSkillsIndex).toBeGreaterThanOrEqual(0); + const firstVersion = refreshModule.bumpSkillsSnapshotVersion({ + workspaceDir: idleWorkspaceDir, + reason: "watch", + }); + + vi.advanceTimersByTime(60 * 60_000 + 1_000); + refreshModule.ensureSkillsWatcher({ + workspaceDir: "/tmp/workspace-active", + config: { skills: { load: { watchDebounceMs: 10 } } }, + }); + + expect(createdWatchers[idleSkillsIndex]?.close).toHaveBeenCalledTimes(1); + const evictedVersion = refreshModule.getSkillsSnapshotVersion(idleWorkspaceDir); + expect(evictedVersion).toBeGreaterThan(firstVersion); + expect(refreshModule.shouldRefreshSnapshotForVersion(firstVersion, evictedVersion)).toBe(true); + vi.setSystemTime(new Date(evictedVersion)); + const followupVersion = refreshModule.bumpSkillsSnapshotVersion({ + workspaceDir: idleWorkspaceDir, + }); + expect(followupVersion).toBeGreaterThan(evictedVersion); + }); + + it("keeps refreshed workspace subscriptions within the idle TTL", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00Z")); + const activeWorkspaceDir = "/tmp/workspace-active-refresh"; + refreshModule.ensureSkillsWatcher({ + workspaceDir: activeWorkspaceDir, + config: { skills: { load: { watchDebounceMs: 10 } } }, + }); + const callPaths = (watchMock.mock.calls as unknown as Array<[string]>).map((call) => call[0]); + const activeSkillsIndex = callPaths.findIndex( + (target) => target === `${activeWorkspaceDir}/skills`, + ); + expect(activeSkillsIndex).toBeGreaterThanOrEqual(0); + + vi.advanceTimersByTime(30 * 60_000); + refreshModule.ensureSkillsWatcher({ + workspaceDir: activeWorkspaceDir, + config: { skills: { load: { watchDebounceMs: 10 } } }, + }); + vi.advanceTimersByTime(31 * 60_000); + refreshModule.ensureSkillsWatcher({ + workspaceDir: "/tmp/workspace-other", + config: { skills: { load: { watchDebounceMs: 10 } } }, + }); + + expect(createdWatchers[activeSkillsIndex]?.close).not.toHaveBeenCalled(); + }); + it("rebuilds a shared watcher with last-writer debounce while preserving subscribers", async () => { vi.useFakeTimers(); const seen: SkillsChangeEvent[] = []; diff --git a/src/skills/runtime/refresh.ts b/src/skills/runtime/refresh.ts index e6f7b9cbcd0d..dcd4c451f5f9 100644 --- a/src/skills/runtime/refresh.ts +++ b/src/skills/runtime/refresh.ts @@ -9,6 +9,7 @@ import { CONFIG_DIR, resolveUserPath } from "../../utils.js"; import { resolvePluginSkillDirs } from "../loading/plugin-skills.js"; import { bumpSkillsSnapshotVersion, + clearSkillsSnapshotVersionForWorkspace, resetSkillsRefreshStateForTest, setSkillsChangeListenerErrorHandler, } from "./refresh-state.js"; @@ -59,6 +60,10 @@ const workspaceWatchTargets = new Map(); // per-turn watcher reconciliation path stays cheap until config or watched // filesystem changes require a fresh root scan. const workspaceWatchTargetCache = new Map(); +const workspaceWatchLastEnsuredAt = new Map(); +// Session turns re-ensure their workspace; entries older than this are treated +// as abandoned subscriptions and evicted by the next ensure call. +const SKILLS_WORKSPACE_WATCH_IDLE_TTL_MS = 60 * 60_000; setSkillsChangeListenerErrorHandler((err) => { log.warn(`skills change listener failed: ${String(err)}`); @@ -523,26 +528,51 @@ function unsubscribeWorkspaceFromPath(workspaceDir: string, watchTarget: WatchTa } } +function disposeWorkspaceWatchState( + workspaceDir: string, + watchTargets: readonly WatchTarget[] = workspaceWatchTargets.get(workspaceDir) ?? [], +): void { + const hadWatchTargets = watchTargets.length > 0; + for (const watchTarget of watchTargets) { + unsubscribeWorkspaceFromPath(workspaceDir, watchTarget); + } + workspaceWatchTargets.delete(workspaceDir); + workspaceWatchTargetCache.delete(workspaceDir); + workspaceWatchLastEnsuredAt.delete(workspaceDir); + if (hadWatchTargets) { + // Watcher disposal creates an unwatched interval; mark the workspace dirty + // so the next turn rebuilds skills even if file events were missed. + bumpSkillsSnapshotVersion({ workspaceDir, reason: "watch-targets" }); + } + clearSkillsSnapshotVersionForWorkspace(workspaceDir); +} + +function evictIdleWorkspaceWatchStates(now: number): void { + const cutoff = now - SKILLS_WORKSPACE_WATCH_IDLE_TTL_MS; + for (const [workspaceDir, lastEnsuredAt] of workspaceWatchLastEnsuredAt) { + if (lastEnsuredAt < cutoff) { + disposeWorkspaceWatchState(workspaceDir); + } + } +} + export function ensureSkillsWatcher(params: { workspaceDir: string; config?: OpenClawConfig }) { const workspaceDir = params.workspaceDir.trim(); if (!workspaceDir) { return; } + const now = Date.now(); const watchEnabled = params.config?.skills?.load?.watch !== false; const debounceMs = resolveWatchDebounceMs(params.config); const previousTargets = workspaceWatchTargets.get(workspaceDir) ?? []; if (!watchEnabled) { - if (previousTargets.length > 0) { - for (const watchTarget of previousTargets) { - unsubscribeWorkspaceFromPath(workspaceDir, watchTarget); - } - workspaceWatchTargets.delete(workspaceDir); - workspaceWatchTargetCache.delete(workspaceDir); - } + disposeWorkspaceWatchState(workspaceDir, previousTargets); + evictIdleWorkspaceWatchStates(now); return; } + workspaceWatchLastEnsuredAt.set(workspaceDir, now); const watchTargets = resolveWatchTargets(workspaceDir, params.config); const targetsUnchanged = sameWatchTargets(previousTargets, watchTargets); const debounceUnchanged = watchTargets.every( @@ -553,6 +583,7 @@ export function ensureSkillsWatcher(params: { workspaceDir: string; config?: Ope }, ); if (targetsUnchanged && debounceUnchanged) { + evictIdleWorkspaceWatchStates(now); return; } const watchTargetsChanged = previousTargets.length > 0 && !targetsUnchanged; @@ -575,6 +606,7 @@ export function ensureSkillsWatcher(params: { workspaceDir: string; config?: Ope changedPath: watchTargets.map((target) => target.path).join("|"), }); } + evictIdleWorkspaceWatchStates(now); } export async function resetSkillsRefreshForTest(): Promise { @@ -584,6 +616,7 @@ export async function resetSkillsRefreshForTest(): Promise { pathWatchers.clear(); workspaceWatchTargets.clear(); workspaceWatchTargetCache.clear(); + workspaceWatchLastEnsuredAt.clear(); await Promise.all( active.map(async (state) => { if (state.timer) {