Skip to content

Commit e76df69

Browse files
fede-kamelfede-kamel
authored
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 <[email protected]>
1 parent f983111 commit e76df69

3 files changed

Lines changed: 145 additions & 8 deletions

File tree

‎src/skills/runtime/refresh-state.ts‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ export function bumpSkillsSnapshotVersion(params?: {
4343
const reason = params?.reason ?? "manual";
4444
const changedPath = params?.changedPath;
4545
if (params?.workspaceDir) {
46-
const current = workspaceVersions.get(params.workspaceDir) ?? 0;
46+
const current = Math.max(globalVersion, workspaceVersions.get(params.workspaceDir) ?? 0);
4747
const next = bumpVersion(current);
4848
workspaceVersions.set(params.workspaceDir, next);
4949
emit({ workspaceDir: params.workspaceDir, reason, changedPath });
@@ -62,6 +62,16 @@ export function getSkillsSnapshotVersion(workspaceDir?: string): number {
6262
return Math.max(globalVersion, local);
6363
}
6464

65+
export function clearSkillsSnapshotVersionForWorkspace(workspaceDir: string): void {
66+
const local = workspaceVersions.get(workspaceDir);
67+
if (typeof local === "number" && local > globalVersion) {
68+
// Keep pending workspace invalidation visible after dropping the workspace
69+
// key; otherwise teardown can hide a skill change from cached snapshots.
70+
globalVersion = local;
71+
}
72+
workspaceVersions.delete(workspaceDir);
73+
}
74+
6575
export function shouldRefreshSnapshotForVersion(
6676
cachedVersion?: number,
6777
nextVersion?: number,

‎src/skills/runtime/refresh.test.ts‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -657,6 +657,7 @@ describe("ensureSkillsWatcher", () => {
657657
workspaceDir: "/tmp/ws-a",
658658
config: { skills: { load: { extraDirs: ["/tmp/shared"], watch: false } } },
659659
});
660+
seen.length = 0;
660661

661662
const callPaths = (watchMock.mock.calls as unknown as Array<[string]>).map((call) => call[0]);
662663
const sharedIndex = callPaths.findIndex((target) => target.includes("/tmp/shared"));
@@ -673,6 +674,99 @@ describe("ensureSkillsWatcher", () => {
673674
expect(seen.some((change) => change.workspaceDir === "/tmp/ws-a")).toBe(false);
674675
});
675676

677+
it("clears workspace version state on watch disable without losing pending invalidation", () => {
678+
vi.useFakeTimers();
679+
vi.setSystemTime(new Date("2026-01-01T00:00:00Z"));
680+
const workspaceDir = "/tmp/workspace-version-cleanup";
681+
refreshModule.ensureSkillsWatcher({
682+
workspaceDir,
683+
config: { skills: { load: { watchDebounceMs: 10 } } },
684+
});
685+
686+
const firstVersion = refreshModule.bumpSkillsSnapshotVersion({
687+
workspaceDir,
688+
reason: "watch",
689+
changedPath: `${workspaceDir}/skills/demo/SKILL.md`,
690+
});
691+
refreshModule.ensureSkillsWatcher({
692+
workspaceDir,
693+
config: { skills: { load: { watch: false } } },
694+
});
695+
696+
const nextVersion = refreshModule.getSkillsSnapshotVersion(workspaceDir);
697+
expect(nextVersion).toBeGreaterThan(firstVersion);
698+
expect(refreshModule.shouldRefreshSnapshotForVersion(firstVersion, nextVersion)).toBe(true);
699+
vi.setSystemTime(new Date(nextVersion));
700+
const followupVersion = refreshModule.bumpSkillsSnapshotVersion({
701+
workspaceDir,
702+
reason: "watch",
703+
});
704+
expect(followupVersion).toBeGreaterThan(nextVersion);
705+
});
706+
707+
it("evicts idle workspace subscriptions on a later ensure call", () => {
708+
vi.useFakeTimers();
709+
vi.setSystemTime(new Date("2026-01-01T00:00:00Z"));
710+
const idleWorkspaceDir = "/tmp/workspace-idle";
711+
refreshModule.ensureSkillsWatcher({
712+
workspaceDir: idleWorkspaceDir,
713+
config: { skills: { load: { watchDebounceMs: 10 } } },
714+
});
715+
const callPaths = (watchMock.mock.calls as unknown as Array<[string]>).map((call) => call[0]);
716+
const idleSkillsIndex = callPaths.findIndex(
717+
(target) => target === `${idleWorkspaceDir}/skills`,
718+
);
719+
expect(idleSkillsIndex).toBeGreaterThanOrEqual(0);
720+
const firstVersion = refreshModule.bumpSkillsSnapshotVersion({
721+
workspaceDir: idleWorkspaceDir,
722+
reason: "watch",
723+
});
724+
725+
vi.advanceTimersByTime(60 * 60_000 + 1_000);
726+
refreshModule.ensureSkillsWatcher({
727+
workspaceDir: "/tmp/workspace-active",
728+
config: { skills: { load: { watchDebounceMs: 10 } } },
729+
});
730+
731+
expect(createdWatchers[idleSkillsIndex]?.close).toHaveBeenCalledTimes(1);
732+
const evictedVersion = refreshModule.getSkillsSnapshotVersion(idleWorkspaceDir);
733+
expect(evictedVersion).toBeGreaterThan(firstVersion);
734+
expect(refreshModule.shouldRefreshSnapshotForVersion(firstVersion, evictedVersion)).toBe(true);
735+
vi.setSystemTime(new Date(evictedVersion));
736+
const followupVersion = refreshModule.bumpSkillsSnapshotVersion({
737+
workspaceDir: idleWorkspaceDir,
738+
});
739+
expect(followupVersion).toBeGreaterThan(evictedVersion);
740+
});
741+
742+
it("keeps refreshed workspace subscriptions within the idle TTL", () => {
743+
vi.useFakeTimers();
744+
vi.setSystemTime(new Date("2026-01-01T00:00:00Z"));
745+
const activeWorkspaceDir = "/tmp/workspace-active-refresh";
746+
refreshModule.ensureSkillsWatcher({
747+
workspaceDir: activeWorkspaceDir,
748+
config: { skills: { load: { watchDebounceMs: 10 } } },
749+
});
750+
const callPaths = (watchMock.mock.calls as unknown as Array<[string]>).map((call) => call[0]);
751+
const activeSkillsIndex = callPaths.findIndex(
752+
(target) => target === `${activeWorkspaceDir}/skills`,
753+
);
754+
expect(activeSkillsIndex).toBeGreaterThanOrEqual(0);
755+
756+
vi.advanceTimersByTime(30 * 60_000);
757+
refreshModule.ensureSkillsWatcher({
758+
workspaceDir: activeWorkspaceDir,
759+
config: { skills: { load: { watchDebounceMs: 10 } } },
760+
});
761+
vi.advanceTimersByTime(31 * 60_000);
762+
refreshModule.ensureSkillsWatcher({
763+
workspaceDir: "/tmp/workspace-other",
764+
config: { skills: { load: { watchDebounceMs: 10 } } },
765+
});
766+
767+
expect(createdWatchers[activeSkillsIndex]?.close).not.toHaveBeenCalled();
768+
});
769+
676770
it("rebuilds a shared watcher with last-writer debounce while preserving subscribers", async () => {
677771
vi.useFakeTimers();
678772
const seen: SkillsChangeEvent[] = [];

‎src/skills/runtime/refresh.ts‎

Lines changed: 40 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import { CONFIG_DIR, resolveUserPath } from "../../utils.js";
99
import { resolvePluginSkillDirs } from "../loading/plugin-skills.js";
1010
import {
1111
bumpSkillsSnapshotVersion,
12+
clearSkillsSnapshotVersionForWorkspace,
1213
resetSkillsRefreshStateForTest,
1314
setSkillsChangeListenerErrorHandler,
1415
} from "./refresh-state.js";
@@ -59,6 +60,10 @@ const workspaceWatchTargets = new Map<string, WatchTarget[]>();
5960
// per-turn watcher reconciliation path stays cheap until config or watched
6061
// filesystem changes require a fresh root scan.
6162
const workspaceWatchTargetCache = new Map<string, WatchTargetCacheEntry>();
63+
const workspaceWatchLastEnsuredAt = new Map<string, number>();
64+
// Session turns re-ensure their workspace; entries older than this are treated
65+
// as abandoned subscriptions and evicted by the next ensure call.
66+
const SKILLS_WORKSPACE_WATCH_IDLE_TTL_MS = 60 * 60_000;
6267

6368
setSkillsChangeListenerErrorHandler((err) => {
6469
log.warn(`skills change listener failed: ${String(err)}`);
@@ -523,26 +528,51 @@ function unsubscribeWorkspaceFromPath(workspaceDir: string, watchTarget: WatchTa
523528
}
524529
}
525530

531+
function disposeWorkspaceWatchState(
532+
workspaceDir: string,
533+
watchTargets: readonly WatchTarget[] = workspaceWatchTargets.get(workspaceDir) ?? [],
534+
): void {
535+
const hadWatchTargets = watchTargets.length > 0;
536+
for (const watchTarget of watchTargets) {
537+
unsubscribeWorkspaceFromPath(workspaceDir, watchTarget);
538+
}
539+
workspaceWatchTargets.delete(workspaceDir);
540+
workspaceWatchTargetCache.delete(workspaceDir);
541+
workspaceWatchLastEnsuredAt.delete(workspaceDir);
542+
if (hadWatchTargets) {
543+
// Watcher disposal creates an unwatched interval; mark the workspace dirty
544+
// so the next turn rebuilds skills even if file events were missed.
545+
bumpSkillsSnapshotVersion({ workspaceDir, reason: "watch-targets" });
546+
}
547+
clearSkillsSnapshotVersionForWorkspace(workspaceDir);
548+
}
549+
550+
function evictIdleWorkspaceWatchStates(now: number): void {
551+
const cutoff = now - SKILLS_WORKSPACE_WATCH_IDLE_TTL_MS;
552+
for (const [workspaceDir, lastEnsuredAt] of workspaceWatchLastEnsuredAt) {
553+
if (lastEnsuredAt < cutoff) {
554+
disposeWorkspaceWatchState(workspaceDir);
555+
}
556+
}
557+
}
558+
526559
export function ensureSkillsWatcher(params: { workspaceDir: string; config?: OpenClawConfig }) {
527560
const workspaceDir = params.workspaceDir.trim();
528561
if (!workspaceDir) {
529562
return;
530563
}
564+
const now = Date.now();
531565
const watchEnabled = params.config?.skills?.load?.watch !== false;
532566
const debounceMs = resolveWatchDebounceMs(params.config);
533567
const previousTargets = workspaceWatchTargets.get(workspaceDir) ?? [];
534568

535569
if (!watchEnabled) {
536-
if (previousTargets.length > 0) {
537-
for (const watchTarget of previousTargets) {
538-
unsubscribeWorkspaceFromPath(workspaceDir, watchTarget);
539-
}
540-
workspaceWatchTargets.delete(workspaceDir);
541-
workspaceWatchTargetCache.delete(workspaceDir);
542-
}
570+
disposeWorkspaceWatchState(workspaceDir, previousTargets);
571+
evictIdleWorkspaceWatchStates(now);
543572
return;
544573
}
545574

575+
workspaceWatchLastEnsuredAt.set(workspaceDir, now);
546576
const watchTargets = resolveWatchTargets(workspaceDir, params.config);
547577
const targetsUnchanged = sameWatchTargets(previousTargets, watchTargets);
548578
const debounceUnchanged = watchTargets.every(
@@ -553,6 +583,7 @@ export function ensureSkillsWatcher(params: { workspaceDir: string; config?: Ope
553583
},
554584
);
555585
if (targetsUnchanged && debounceUnchanged) {
586+
evictIdleWorkspaceWatchStates(now);
556587
return;
557588
}
558589
const watchTargetsChanged = previousTargets.length > 0 && !targetsUnchanged;
@@ -575,6 +606,7 @@ export function ensureSkillsWatcher(params: { workspaceDir: string; config?: Ope
575606
changedPath: watchTargets.map((target) => target.path).join("|"),
576607
});
577608
}
609+
evictIdleWorkspaceWatchStates(now);
578610
}
579611

580612
export async function resetSkillsRefreshForTest(): Promise<void> {
@@ -584,6 +616,7 @@ export async function resetSkillsRefreshForTest(): Promise<void> {
584616
pathWatchers.clear();
585617
workspaceWatchTargets.clear();
586618
workspaceWatchTargetCache.clear();
619+
workspaceWatchLastEnsuredAt.clear();
587620
await Promise.all(
588621
active.map(async (state) => {
589622
if (state.timer) {

0 commit comments

Comments
 (0)