fix(daemon): automatically discover config Skills changes (#16262)

This commit is contained in:
callmeYe authored and GitHub committed 2026-11-04 09:12:02 +00:00
1 parent 815c1d3768
commit 641e258546
15 files changed
+1494 -26

No files matched your search

@@ -0,0 +1,118 @@
# Automatic discovery for the Skills configuration catalog
[English](skills-config-automatic-discovery.md) | [简体中文](skills-config-automatic-discovery.zh-CN.md)
## Problem and scope
Issue #15645: after a daemon warms its independent configuration catalog, an
external tool can create a valid project `SKILL.md` that runtime discovery sees
but management cannot. The provider retains its `SkillManager` indefinitely and
only invalidates it after configuration mutations. Settings reload on every
read, but directory discovery does not.
The fix covers both `/workspace/config/skills` (legacy-primary) and
`/workspaces/:workspace/config/skills` (persisted-workspace). Management remains
independent of child and session availability. Applying changes to existing
sessions, runtime fallback policy, extension activation, and core discovery
semantics are outside this change.
## Design
Only the server's two configuration providers enable `autoDiscover`. Other
callers keep their existing behavior. A cold read performs one shared discovery.
A warm read projects the retained manager through freshly loaded settings and
does not await background filesystem work.
Trusted, non-safe catalogs subscribe to project/user directories using chokidar
and the runtime's `watcherIgnored` and `WATCHER_MAX_DEPTH` policy. Subscriptions
are shared by directory within the provider, including user directories shared
by multiple workspaces. Only existing directories are subscribed; a later read or
publication arms newly created roots. Failed watchers retry after 30 seconds,
while read reconciliation remains available. Events debounce for 150 ms and start
background discovery without a GET. An event during a scan queues one further
observation.
Reads also request background reconciliation at most once per second, measured
from task start. This discovers directories created after startup, covers missed
events and watcher failures, and refreshes inert untrusted catalogs without
holding persistent watchers on untrusted workspaces. The one-second interval is
a start throttle, not a guaranteed visibility deadline. An idle catalog whose
change was not watched converges after another read and a successful scan.
Each background observation builds into a fresh temporary manager map. A complete
successful result atomically replaces the retained manager. Failed reads or
incomplete source enumeration preserve the previous inventory. Invalid
project/user manifests are omitted without vetoing unrelated valid changes. Extension discovery errors
remain conservative because the extension loader does not classify their cause.
Core exposes a separate read-failure signal for this distinction; its existing completeness signal and
the snapshot export's strict rejection of incomplete catalogs stay unchanged.
Every warm GET still rereads settings; enable/disable changes do not wait for a
discovery window. The initial discovery retains its existing response semantics.
Background work releases its shared slot after five seconds so a slow filesystem
does not permanently prevent retries. At most two background scans per workspace
generation may remain unsettled: the original scan and one timeout retry. Once
both are occupied, further retries wait for a scan to settle. This does not cancel
underlying I/O. A late
successful observation may publish if its workspace generation is still current
and no newer successful observation has published. Start sequence, not completion
time or wall clock, orders publication. Completion only clears its own slot.
Cold reads wait for actual discovery and do not use this timeout.
## Trust and lifecycle
Safe catalogs remain bundled-only. Effective safe-mode changes discard the
previous manager and subscriptions before answering another read. Inert untrusted
catalogs may read manifests under the existing policy, but never load workspace
settings/extensions or create persistent watchers. Trust replacement invalidates
both providers before runtime draining, even without another configuration read.
Dispatch under a different trust policy invalidates that workspace in the
opposite provider before reading.
No watcher creates directories.
Explicit invalidation advances the existing epoch, drops the manager and active
read/refresh slots, cancels debounce timers, and removes that workspace's watcher
subscriptions. Workspace removal uses this same path. Detached scans cannot
install managers or watchers afterward. Other workspaces keep their subscriptions.
Provider disposal seals all new work before invalidating every workspace,
including cold in-flight reads, and closes the owned watchers. The server
adds this cleanup independently of the owner's replaceable app drain with both
supplied and locally owned lifecycles, preserving ownership of unrelated daemon
resources. The throwaway `snapshot()`
path performs fresh discovery and never starts watchers or background tasks.
## Tradeoffs and attribution
This replaces #15703 and builds on 2SinceJuly's watcher-based direction, failure
cases, and regression scenarios. Core discovery and its existing completeness
consumers remain unchanged; only an additive read-failure signal is introduced.
Its only consumer is the configuration provider's background publication gate.
Fresh background builds scan all enabled levels, including bundled and extension
metadata, rather than introducing partial-cache publication into core. This is
more work per refresh, bounded by the unsettled-scan cap and start throttle; warm reads
remain cache-backed. A hung OS operation is not forcibly cancellable and can
outlive its timeout, but cannot overwrite newer successful observations.
The existing omission of custom `skills.directories` from this provider is not
expanded here. This change does not claim to reconcile every runtime source or
to make an existing session adopt newly discovered skills automatically.
## Validation and acceptance
- Independently reproduce the warmed-catalog defect on the baseline. Attempt the
global CLI first and record version/API limitations separately.
- On the built daemon, verify external creation, editing, deletion and recreation
through both configuration routes with no child/session; confirm workspace
isolation and concurrent readers.
- Verify watcher-driven discovery without reads, missing-directory reconciliation,
consecutive refresh windows, failed/partial scans, and slow refresh recovery.
- Verify invalidation and disposal during cold discovery and background work,
late-result ordering, shared subscriptions, and complete cleanup.
- Preserve per-read settings, explicit invalidation, snapshot freshness,
bundled-only safe mode, and inert untrusted behavior across refresh cycles.
- Run build, bundle, typecheck and focused tests; inspect the complete diff in
two clean self-audit passes and obtain independent review.
Acceptance requires observed behavior and recorded checks. PR creation or green
CI does not establish deployment or merge.
@@ -0,0 +1,91 @@
# Skills 配置目录的自动发现
[English](skills-config-automatic-discovery.md) | [简体中文](skills-config-automatic-discovery.zh-CN.md)
## 问题与范围
Issue #15645:daemon 预热独立配置目录后,外部工具创建的合法项目 `SKILL.md`
能被 runtime 发现,管理端却看不到。provider 长期保留 `SkillManager`,只有配置
变更才使其失效。每次读取都会重新加载设置,但不会重新发现目录。
修复覆盖 `/workspace/config/skills`(legacy-primary)和
`/workspaces/:workspace/config/skills`(persisted-workspace)。管理目录仍独立于
child 和 session。向已有 session 应用变更、runtime 回退策略、扩展激活以及 core
发现语义不在本次范围内。
## 设计
只有 server 的两个配置 provider 启用 `autoDiscover`,其他调用方保留原有行为。
冷读取共享一次发现;热读取将保留的 manager 与重新加载的设置组合,不等待后台
文件系统工作。
受信任且非 safe 的目录使用 chokidar 订阅 project/user 目录,复用 runtime 的
`watcherIgnored` 和 `WATCHER_MAX_DEPTH` 策略。provider 内按目录共享订阅,多个
workspace 共用的 user 目录也只监听一次。只订阅已存在的目录,后续读取或发布会
为新建根目录建立监听。失败 watcher 在 30 秒后重试,其间读取核对仍可用。
事件经 150 ms 防抖后启动后台发现,
无需 GET;扫描期间到达的事件会合并为后续一次观察。
读取还会触发后台核对,从任务开始计时,最多每秒启动一次。该路径发现启动后
才创建的目录,覆盖丢失事件、watcher 失败,并在不持久监听不受信任 workspace
的前提下刷新 inert untrusted 目录。1 秒是启动节流,不是保证可见的期限。
没有被监听到变更的空闲目录,要在后续读取及一次成功扫描后才收敛。
每次后台观察都使用全新的临时 manager map。完整成功的结果原子替换保留的
manager。读取失败或来源枚举不完整时保留上一份目录。格式错误的 project/user 清单被忽略,
不会否决无关合法变更。扩展加载器不区分错误来源,因此扩展发现错误仍保守阻止发布。
core 增加独立的读取失败标记以区分两类错误;既有完整性
标记及 snapshot 对不完整目录的严格拒绝保持不变。
每次热 GET 仍重新读取设置,启用/禁用变更不必等待发现窗口。首次发现保留既有
响应语义。
后台任务在 5 秒后释放共享槽,避免慢文件系统永久阻止重试;这不会取消底层
I/O。每个 workspace 代际最多保留两个未结束后台扫描:原始扫描及一次超时重试。
两者都未结束时,后续重试等待任一扫描结束。只要 workspace 代际仍有效,且没有更新的成功观察已发布,迟到的成功结果
仍可发布。发布顺序由启动序号决定,与完成时间和系统时钟无关;完成只清理自己
占用的槽。冷读取等待真实发现,不采用这个超时。
## 信任与生命周期
safe 目录保持仅 bundled。有效 safe-mode 变化时,在回答后续读取前丢弃原 manager
及订阅。inert untrusted 目录按已有策略读取清单,但不会加载 workspace 设置或
扩展,也不创建持久 watcher。信任替换会在 runtime 开始 drain 前使两个 provider
失效,即使没有后续配置读取也会清理。按不同信任策略分发时,读取前使另一 provider 中该
workspace 的状态失效。watcher 不创建任何目录。
显式失效推进既有 epoch,删除 manager 和活跃的读取/刷新槽,取消防抖计时器,
移除该 workspace 的 watcher 订阅。workspace 删除复用此路径。已分离的扫描
之后不能再安装 manager 或 watcher;其他 workspace 的订阅继续保留。
provider 关闭先禁止新工作,再使所有 workspace 失效,包括仍在途的冷读取,
并关闭其拥有的 watcher。server 在外部传入和本地创建的 lifecycle 上,都独立于
所有者可替换的 app drain 注册该
清理,不改变其他 daemon 资源的所有权。一次性的 `snapshot()` 路径执行全新
发现,不启动 watcher 或后台任务。
## 取舍与归属
本实现替代 #15703,参考 2SinceJuly 的 watcher 方案、失败案例和回归场景。
core 发现逻辑及既有完整性消费者保持不变,仅增加独立读取失败标记,唯一消费者
是配置 provider 的后台发布门。后台全新构建扫描所有已启用层级,包括 bundled
和扩展元数据,避免在 core 中引入部分缓存发布。这增加了每次刷新的工作量,
由未结束扫描数量上限及启动节流限制;热读取仍使用缓存。挂起的操作系统 I/O 无法强制取消,
可能超出超时时间继续存在,但不能覆盖更新的成功观察。
本次不扩展 provider 既有的 `skills.directories` 缺失。因此不宣称所有 runtime
来源都已对齐,也不承诺已有 session 会自动采用新发现的 Skill。
## 验证与验收
- 独立复现基线的预热目录缺陷;先尝试全局 CLI,单独记录版本/API 限制。
- 使用构建后的 daemon,在无 child/session 时通过两条配置路由验证外部新增、
编辑、删除及重新创建,并确认 workspace 隔离与并发读取。
- 验证无读取时的 watcher 发现、缺失目录的后台核对、连续刷新窗口、失败或
不完整扫描,以及慢刷新恢复。
- 验证冷发现和后台工作期间的失效与关闭、迟到结果顺序、共享订阅和完整清理。
- 跨刷新周期保留每次读取设置、显式失效、snapshot 新鲜度、safe 模式仅 bundled
及 inert untrusted 行为。
- 执行 build、bundle、typecheck 和针对性测试;完成两轮干净的全 diff 自审,
并取得独立审阅。
验收依赖实测行为和检查记录。创建 PR 或 CI 通过不等于已部署或合并。
-1
View File
@@ -186,7 +186,6 @@ export const legacyCoreBarrelImports = [
'packages/cli/src/serve/workspace-remember.ts',
'packages/cli/src/serve/workspace-runtime-storage.ts',
'packages/cli/src/serve/workspace-skill-management.ts',
'packages/cli/src/serve/workspace-skills-status.ts',
'packages/cli/src/services/BuiltinCommandLoader.ts',
'packages/cli/src/services/BundledSkillLoader.ts',
'packages/cli/src/services/CommandService.ts',
+4
View File
@@ -8978,6 +8978,10 @@ async function runQwenServeImpl(
});
},
drainRuntime: async (runtimeToDrain) => {
const invalidateSkillsConfigStatus = app.locals[
'invalidateSkillsConfigStatus'
] as (workspaceCwd: string) => void;
invalidateSkillsConfigStatus(runtimeToDrain.workspaceCwd);
workspaceRuntimeRemoval.beginDrain(runtimeToDrain);
acpHandle()?.beginWorkspaceDrain(runtimeToDrain.workspaceId);
},
@@ -50,6 +50,7 @@ export class ServeAppLifecycleController implements ServeAppLifecycle {
private sealed = false;
private closePending?: Promise<void>;
private appDrain?: () => Promise<void>;
private readonly additionalAppDrains: Array<() => Promise<void>> = [];
private hostDrain?: () => Promise<void>;
private bootStarter?: () => Promise<void> | void;
private bootPending?: Promise<void>;
@@ -114,6 +115,10 @@ export class ServeAppLifecycleController implements ServeAppLifecycle {
this.appDrain = drain;
}
addAppDrain(drain: () => Promise<void>): void {
this.additionalAppDrains.push(drain);
}
setBootStarter(starter: () => Promise<void> | void): void {
this.bootStarter = starter;
this.openAdmissionIfReady();
@@ -194,6 +199,7 @@ export class ServeAppLifecycleController implements ServeAppLifecycle {
this.seal();
const drains = [
this.startDrain(this.appDrain),
...this.additionalAppDrains.map((drain) => this.startDrain(drain)),
this.startDrain(this.hostDrain),
this.bootPending?.catch(() => undefined),
].filter((value): value is Promise<void> => value !== undefined);
@@ -7,6 +7,11 @@
import { describe, expect, it, vi } from 'vitest';
import type { SettingScope } from '../config/settings.js';
import type { ServeOptions } from './types.js';
import type { WorkspaceSkillsStatusProviderOptions } from './workspace-skills-status.js';
import {
getServeAppLifecycle,
ServeAppLifecycleController,
} from './serve-app-lifecycle.js';
const mockCreateDaemonWorkspaceService = vi.hoisted(() =>
vi.fn((_deps: unknown) => ({})),
@@ -16,6 +21,15 @@ vi.mock('./workspace-service/index.js', () => ({
createDaemonWorkspaceService: mockCreateDaemonWorkspaceService,
}));
const mockCreateSkillsProvider = vi.hoisted(() =>
vi.fn((_options: WorkspaceSkillsStatusProviderOptions) =>
Object.assign(vi.fn(), { dispose: vi.fn(async () => {}) }),
),
);
vi.mock('./workspace-skills-status.js', () => ({
createWorkspaceSkillsStatusProvider: mockCreateSkillsProvider,
}));
const { createServeApp } = await import('./server.js');
const baseOpts: ServeOptions = {
@@ -34,10 +48,62 @@ function makeBridge() {
failed: 0,
})),
publishWorkspaceEvent: vi.fn(),
shutdown: vi.fn(async () => {}),
};
}
describe('createServeApp workspace service wiring', () => {
it.each(['before', 'after'] as const)(
'preserves an injected lifecycle drain registered %s app creation',
async (order) => {
mockCreateSkillsProvider.mockClear();
const lifecycle = new ServeAppLifecycleController();
const drain = vi.fn(async () => {});
if (order === 'before') lifecycle.setAppDrain(drain);
const app = createServeApp(baseOpts, undefined, {
boundWorkspace: '/workspace',
bridge: makeBridge(),
serveAppLifecycle: lifecycle,
} as never);
if (order === 'after') lifecycle.setAppDrain(drain);
await getServeAppLifecycle(app).close();
expect(drain).toHaveBeenCalledOnce();
for (const [
index,
[options],
] of mockCreateSkillsProvider.mock.calls.entries()) {
if (options.autoDiscover)
expect(
mockCreateSkillsProvider.mock.results[index]!.value.dispose,
).toHaveBeenCalledOnce();
}
},
);
it.each([false, true])(
'disposes both automatic catalogs with an externally supplied lifecycle: %s',
async (supplied) => {
mockCreateSkillsProvider.mockClear();
const app = createServeApp(baseOpts, undefined, {
boundWorkspace: '/workspace',
bridge: makeBridge(),
...(supplied
? { serveAppLifecycle: new ServeAppLifecycleController() }
: {}),
} as never);
const automatic = mockCreateSkillsProvider.mock.calls.flatMap(
([options], index) =>
options.autoDiscover
? [mockCreateSkillsProvider.mock.results[index]!.value]
: [],
);
expect(automatic).toHaveLength(2);
await getServeAppLifecycle(app).close();
for (const provider of automatic)
expect(provider.dispose).toHaveBeenCalledOnce();
},
);
it('forwards batch settings persistence to the workspace service', () => {
mockCreateDaemonWorkspaceService.mockClear();
const persistSetting = vi.fn(
+14
View File
@@ -1101,12 +1101,18 @@ export function createServeApp(
const primaryEffectiveEnv = getRuntimeEffectiveEnv(primaryRuntimeEnvMetadata);
const trustedSkillsConfigStatus = createWorkspaceSkillsStatusProvider({
workspaceTrusted: true,
autoDiscover: true,
});
const untrustedSkillsConfigStatus = createWorkspaceSkillsStatusProvider({
workspaceTrusted: false,
includeUntrustedSkills: true,
autoDiscover: true,
});
const getSkillsConfigStatus = (workspaceCwd: string, trusted: boolean) => {
const obsolete = trusted
? untrustedSkillsConfigStatus
: trustedSkillsConfigStatus;
obsolete.invalidate?.(workspaceCwd);
const provider = trusted
? trustedSkillsConfigStatus
: untrustedSkillsConfigStatus;
@@ -1116,6 +1122,7 @@ export function createServeApp(
trustedSkillsConfigStatus.invalidate?.(workspaceCwd);
untrustedSkillsConfigStatus.invalidate?.(workspaceCwd);
};
app.locals['invalidateSkillsConfigStatus'] = invalidateSkillsConfigStatus;
const webTerminalRegistry = new WebTerminalRegistry();
const webTerminalLocals = app.locals as {
stopWebTerminalRegistry?: () => void;
@@ -4339,6 +4346,13 @@ export function createServeApp(
}
let appDrainComplete = false;
const disposeSkillsConfigStatus = async () => {
await Promise.all([
trustedSkillsConfigStatus.dispose?.(),
untrustedSkillsConfigStatus.dispose?.(),
]);
};
serveAppLifecycle.addAppDrain(disposeSkillsConfigStatus);
if (!deps.serveAppLifecycle)
serveAppLifecycle.setAppDrain(async () => {
if (appDrainComplete) return;
@@ -47,6 +47,10 @@ import {
} from '@qwen-code/acp-bridge/bridgeErrors';
import { MCP_RESTART_SERVER_DEADLINE_MS } from '@qwen-code/acp-bridge/mcpTimeouts';
import {
WORKSPACE_SKILLS_REVALIDATE_INTERVAL_MS,
WORKSPACE_SKILLS_REVALIDATE_TIMEOUT_MS,
} from '../workspace-skills-refresh-policy.js';
import { loadSettings } from '../../config/settings.js';
import {
@@ -135,11 +139,8 @@ export {
// Helpers
// ---------------------------------------------------------------------------
// Start throttle for background revalidation, measured from task start.
const WORKSPACE_SKILLS_REVALIDATE_INTERVAL_MS = 1_000;
// Wire-visible age at which snapshot.stale becomes true.
const WORKSPACE_SKILLS_STALE_AFTER_MS = 1_000;
const WORKSPACE_SKILLS_REVALIDATE_TIMEOUT_MS = 5_000;
function createSkillToggleMutation(input: {
skills: ReadonlyArray<{ name: string; enabled: boolean }>;
@@ -0,0 +1,9 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
// Start throttle for background revalidation, measured from task start.
export const WORKSPACE_SKILLS_REVALIDATE_INTERVAL_MS = 1_000;
export const WORKSPACE_SKILLS_REVALIDATE_TIMEOUT_MS = 5_000;
@@ -0,0 +1,480 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { EventEmitter } from 'node:events';
import * as fs from 'node:fs/promises';
import * as os from 'node:os';
import * as path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { SkillManager } from '@qwen-code/qwen-code-core/skills/skill-manager.js';
import { createWorkspaceSkillsStatusProvider } from './workspace-skills-status.js';
const { watch, stderr } = vi.hoisted(() => ({
watch: vi.fn(),
stderr: vi.fn(),
}));
vi.mock('chokidar', () => ({ watch }));
vi.mock('node:fs/promises', async (importOriginal) => ({
...(await importOriginal<typeof fs>()),
}));
vi.mock('../utils/stdioHelpers.js', () => ({ writeStderrLine: stderr }));
class Watcher extends EventEmitter {
close = vi.fn(async () => {
this.removeAllListeners();
});
}
function deferred() {
let resolve!: () => void;
const promise = new Promise<void>((done) => {
resolve = done;
});
return { promise, resolve };
}
describe('configuration catalog automatic discovery', () => {
let root: string;
let workspace: string;
let home: string;
let now: number;
let providers: Array<ReturnType<typeof createWorkspaceSkillsStatusProvider>>;
let watchers: Map<string, Watcher>;
beforeEach(async () => {
root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-catalog-discovery-'));
workspace = path.join(root, 'workspace');
home = path.join(root, 'home');
await fs.mkdir(workspace);
await fs.mkdir(home);
await fs.mkdir(path.join(home, 'skills'));
vi.stubEnv('QWEN_HOME', home);
vi.stubEnv('QWEN_CODE_SAFE_MODE', '0');
now = 1_000_000;
vi.spyOn(Date, 'now').mockImplementation(() => now);
providers = [];
watchers = new Map();
watch.mockImplementation((directory: string) => {
const watcher = new Watcher();
watchers.set(directory, watcher);
return watcher;
});
});
afterEach(async () => {
await Promise.all(providers.map((provider) => provider.dispose?.()));
vi.useRealTimers();
vi.restoreAllMocks();
vi.unstubAllEnvs();
watch.mockReset();
stderr.mockReset();
await fs.rm(root, { recursive: true, force: true });
});
function create(
options: Parameters<typeof createWorkspaceSkillsStatusProvider>[0] = {},
) {
const provider = createWorkspaceSkillsStatusProvider({
autoDiscover: true,
...options,
});
providers.push(provider);
return provider;
}
async function writeSkill(name: string, description = name, cwd = workspace) {
const directory = path.join(cwd, '.qwen', 'skills', name);
await fs.mkdir(directory, { recursive: true });
await fs.writeFile(
path.join(directory, 'SKILL.md'),
`---\nname: ${name}\ndescription: ${description}\n---\nInstructions\n`,
);
return directory;
}
it('refreshes from filesystem events without a GET and coalesces events during a scan', async () => {
await writeSkill('existing');
const provider = create();
await provider(workspace);
const gate = deferred();
const scanned = deferred();
const original = SkillManager.prototype.listSkills;
let scans = 0;
vi.spyOn(SkillManager.prototype, 'listSkills').mockImplementation(
async function (this: SkillManager, options) {
const cold = this.getCachedSkills() === null;
const result = await original.call(this, options);
if (cold && ++scans === 1) {
scanned.resolve();
await gate.promise;
}
return result;
},
);
const watcher = watchers.get(path.join(workspace, '.qwen', 'skills'))!;
await writeSkill('first');
watcher.emit('all', 'add', 'first/SKILL.md');
await scanned.promise;
await writeSkill('second');
watcher.emit('all', 'add', 'second/SKILL.md');
gate.resolve();
await vi.waitFor(() => expect(scans).toBe(2));
await vi.waitFor(async () =>
expect((await provider(workspace)).skills.map((s) => s.name)).toContain(
'second',
),
);
await provider(workspace);
expect(scans).toBe(2);
});
it('reconciles missing directories over successive windows and applies settings on warm reads', async () => {
const provider = create();
const original = SkillManager.prototype.listSkills;
let scans = 0;
vi.spyOn(SkillManager.prototype, 'listSkills').mockImplementation(function (
this: SkillManager,
options,
) {
if (this.getCachedSkills() === null) scans++;
return original.call(this, options);
});
await provider(workspace);
const directory = await writeSkill('external', 'first');
now += 999;
expect((await provider(workspace)).skills.map((s) => s.name)).not.toContain(
'external',
);
expect(scans).toBe(1);
now += 1;
await vi.waitFor(async () =>
expect((await provider(workspace)).skills.map((s) => s.name)).toContain(
'external',
),
);
expect(scans).toBe(2);
await fs.writeFile(
path.join(workspace, '.qwen', 'settings.json'),
JSON.stringify({ skills: { disabled: ['external'] } }),
);
expect(
(await provider(workspace)).skills.find((s) => s.name === 'external')
?.status,
).toBe('disabled');
await writeSkill('external', 'edited');
now += 1000;
await vi.waitFor(async () =>
expect(
(await provider(workspace)).skills.find((s) => s.name === 'external')
?.description,
).toBe('edited'),
);
await fs.rm(directory, { recursive: true });
now += 1000;
await vi.waitFor(async () =>
expect(
(await provider(workspace)).skills.map((s) => s.name),
).not.toContain('external'),
);
});
it('shares subscriptions across workspaces and releases only the removed workspace', async () => {
const other = path.join(root, 'other');
await fs.mkdir(other);
await writeSkill('one');
await writeSkill('two', 'two', other);
const provider = create();
await Promise.all([provider(workspace), provider(other)]);
const userDirectory = path.join(home, 'skills');
expect(
watch.mock.calls.filter(([dir]) => dir === userDirectory),
).toHaveLength(1);
expect((await provider(other)).skills.map((s) => s.name)).not.toContain(
'one',
);
const shared = watchers.get(userDirectory)!;
const own = watchers.get(path.join(workspace, '.qwen', 'skills'))!;
provider.invalidate?.(workspace);
expect(own.close).toHaveBeenCalledOnce();
expect(shared.close).not.toHaveBeenCalled();
await provider.dispose?.();
expect(shared.close).toHaveBeenCalledOnce();
expect((await provider(workspace)).initialized).toBe(false);
});
it('keeps safe and inert untrusted catalogs inside their trust policy across cycles', async () => {
await writeSkill('project');
await fs.writeFile(
path.join(workspace, '.qwen', 'settings.json'),
JSON.stringify({ skills: { disabled: ['project'] } }),
);
const provider = create({
workspaceTrusted: false,
includeUntrustedSkills: true,
});
expect(
(await provider(workspace)).skills.find((s) => s.name === 'project')
?.status,
).toBe('ok');
await writeSkill('later');
now += 1000;
await vi.waitFor(async () =>
expect((await provider(workspace)).skills.map((s) => s.name)).toContain(
'later',
),
);
expect(watch).not.toHaveBeenCalled();
await expect(
fs.stat(path.join(workspace, '.agents')),
).rejects.toMatchObject({
code: 'ENOENT',
});
vi.stubEnv('QWEN_CODE_SAFE_MODE', '1');
expect(
(await provider(workspace)).skills.every((s) => s.level === 'bundled'),
).toBe(true);
now += 1000;
expect(
(await provider(workspace)).skills.every((s) => s.level === 'bundled'),
).toBe(true);
expect(watch).not.toHaveBeenCalled();
});
it('drops trusted subscriptions immediately on entering safe mode', async () => {
await writeSkill('project');
const provider = create();
await provider(workspace);
const priorWatchers = [...watchers.values()];
vi.stubEnv('QWEN_CODE_SAFE_MODE', '1');
expect(
(await provider(workspace)).skills.every((s) => s.level === 'bundled'),
).toBe(true);
for (const watcher of priorWatchers)
expect(watcher.close).toHaveBeenCalledOnce();
});
it('does not start watchers for snapshots or non-config fallback callers', async () => {
await writeSkill('project');
const provider = create();
expect(
(await provider.snapshot(workspace)).skills.map((s) => s.name),
).toContain('project');
const fallback = create({ autoDiscover: false });
await fallback(workspace);
now += 10_000;
await fallback(workspace);
expect(watch).not.toHaveBeenCalled();
});
it.each(['invalidate', 'dispose'] as const)(
'fences a cold scan during %s',
async (operation) => {
await writeSkill('old');
const gate = deferred();
const scanned = deferred();
const original = SkillManager.prototype.listSkills;
vi.spyOn(SkillManager.prototype, 'listSkills').mockImplementationOnce(
async function (this: SkillManager, options) {
const result = await original.call(this, options);
scanned.resolve();
await gate.promise;
return result;
},
);
const provider = create();
const oldRead = provider(workspace);
await scanned.promise;
if (operation === 'dispose') await provider.dispose?.();
else provider.invalidate?.(workspace);
gate.resolve();
await oldRead;
expect(watch).not.toHaveBeenCalled();
if (operation === 'invalidate') {
await writeSkill('new');
expect((await provider(workspace)).skills.map((s) => s.name)).toContain(
'new',
);
} else {
expect((await provider(workspace)).initialized).toBe(false);
}
},
);
it('shares a slow background scan without delaying concurrent readers', async () => {
await writeSkill('retained');
const provider = create();
await provider(workspace);
const gate = deferred();
const scanned = deferred();
const original = SkillManager.prototype.listSkills;
let scans = 0;
vi.spyOn(SkillManager.prototype, 'listSkills').mockImplementation(
async function (this: SkillManager, options) {
if (this.getCachedSkills() === null) {
scans++;
scanned.resolve();
await gate.promise;
}
return original.call(this, options);
},
);
now += 1000;
await provider(workspace);
await scanned.promise;
now += 1000;
const results = await Promise.all(
Array.from({ length: 20 }, () => provider(workspace)),
);
expect(scans).toBe(1);
expect(
results.every((result) =>
result.skills.some((s) => s.name === 'retained'),
),
).toBe(true);
gate.resolve();
});
it.each(['late-success', 'newer-success', 'invalidate'] as const)(
'releases a timed-out slot and fences %s',
async (scenario) => {
await writeSkill('initial');
const provider = create();
await provider(workspace);
const gate = deferred();
const scanned = deferred();
const original = SkillManager.prototype.listSkills;
let scans = 0;
vi.spyOn(SkillManager.prototype, 'listSkills').mockImplementation(
async function (this: SkillManager, options) {
const cold = this.getCachedSkills() === null;
const result = await original.call(this, options);
if (cold && ++scans === 1) {
scanned.resolve();
await gate.promise;
}
return result;
},
);
await writeSkill('slow');
now += 1000;
await provider(workspace);
await scanned.promise;
// The timer is already running on real time; switching timers would lose it.
await new Promise((resolve) => setTimeout(resolve, 5100));
if (scenario !== 'late-success') {
await fs.rm(path.join(workspace, '.qwen', 'skills', 'slow'), {
recursive: true,
});
await writeSkill('newer');
if (scenario === 'invalidate') provider.invalidate?.(workspace);
now += 6000;
await vi.waitFor(async () =>
expect(
(await provider(workspace)).skills.map((s) => s.name),
).toContain('newer'),
);
}
gate.resolve();
if (scenario === 'late-success') {
await vi.waitFor(async () =>
expect(
(await provider(workspace)).skills.map((s) => s.name),
).toContain('slow'),
);
} else {
await new Promise((resolve) => setImmediate(resolve));
expect(
(await provider(workspace)).skills.map((s) => s.name),
).not.toContain('slow');
}
},
15000,
);
it('retains a complete catalog after an unreadable directory and retries after recovery', async () => {
await writeSkill('retained');
const provider = create();
await provider(workspace);
const original = fs.readdir;
const failing = vi.spyOn(fs, 'readdir').mockImplementation((...args) => {
if (String(args[0]) === path.join(workspace, '.qwen', 'skills')) {
return Promise.reject(
Object.assign(new Error('unreadable'), { code: 'EACCES' }),
);
}
return original(...args);
});
now += 1000;
await provider(workspace);
await vi.waitFor(() =>
expect(stderr).toHaveBeenCalledWith(
expect.stringContaining('unreadable'),
),
);
expect((await provider(workspace)).skills.map((s) => s.name)).toContain(
'retained',
);
failing.mockRestore();
await writeSkill('recovered');
now += 1000;
await vi.waitFor(async () =>
expect((await provider(workspace)).skills.map((s) => s.name)).toContain(
'recovered',
),
);
});
it('rejects a partially discovered catalog and publishes a later complete scan', async () => {
const directory = await writeSkill('retained');
const provider = create();
await provider(workspace);
await fs.rm(directory, { recursive: true });
await writeSkill('replacement');
const failedDiscovery = vi
.spyOn(SkillManager.prototype, 'hasDiscoveryReadErrors')
.mockReturnValue(true);
now += 1000;
await provider(workspace);
await vi.waitFor(() => expect(failedDiscovery).toHaveBeenCalled());
const retained = (await provider(workspace)).skills.map((s) => s.name);
expect(retained).toContain('retained');
expect(retained).not.toContain('replacement');
failedDiscovery.mockRestore();
now += 1000;
await vi.waitFor(async () => {
const refreshed = (await provider(workspace)).skills.map((s) => s.name);
expect(refreshed).toContain('replacement');
expect(refreshed).not.toContain('retained');
});
});
it('recovers from asynchronous watcher failure using read reconciliation', async () => {
await writeSkill('initial');
const provider = create();
await provider(workspace);
const watcher = watchers.get(path.join(workspace, '.qwen', 'skills'))!;
watcher.emit('error', new Error('watch unavailable'));
expect(watcher.close).toHaveBeenCalledOnce();
for (let read = 0; read < 3; read++) await provider(workspace);
const projectWatches = () =>
watch.mock.calls.filter(
([dir]) => dir === path.join(workspace, '.qwen', 'skills'),
).length;
expect(projectWatches()).toBe(1);
expect(stderr).toHaveBeenCalledTimes(1);
await writeSkill('recovered');
now += 1000;
await vi.waitFor(async () =>
expect((await provider(workspace)).skills.map((s) => s.name)).toContain(
'recovered',
),
);
expect(projectWatches()).toBe(1);
now += 30_000;
await provider(workspace);
expect(projectWatches()).toBe(2);
});
});
@@ -0,0 +1,174 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import express from 'express';
import request from 'supertest';
import * as fs from 'node:fs/promises';
import * as os from 'node:os';
import * as path from 'node:path';
import { afterEach, expect, it, vi } from 'vitest';
import { watch } from 'chokidar';
import { createWorkspaceSkillsStatusProvider } from './workspace-skills-status.js';
import {
createWorkspaceRegistry,
type WorkspaceRuntime,
} from './workspace-registry.js';
import {
registerWorkspaceQualifiedSkillsRoutes,
registerWorkspaceSkillsRoutes,
} from './routes/workspace-skills.js';
import { sendBridgeError } from './server/error-response.js';
vi.mock('chokidar', async (importOriginal) => {
const actual = await importOriginal<typeof import('chokidar')>();
return { ...actual, watch: vi.fn(actual.watch) };
});
afterEach(() => vi.unstubAllEnvs());
it('arms a real watcher after initially missing parent directories are created', async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-missing-roots-'));
const workspace = path.join(root, 'workspace');
const home = path.join(root, 'home');
await fs.mkdir(workspace);
await fs.mkdir(home);
vi.stubEnv('QWEN_HOME', home);
const provider = createWorkspaceSkillsStatusProvider({ autoDiscover: true });
const directory = path.join(workspace, '.qwen', 'skills');
const subscriptions = () =>
vi
.mocked(watch)
.mock.results.flatMap((result, index) =>
vi.mocked(watch).mock.calls[index]?.[0] === directory &&
result.type === 'return'
? [result.value]
: [],
);
try {
await provider(workspace);
expect(subscriptions()).toHaveLength(0);
await fs.mkdir(directory, { recursive: true });
await new Promise((resolve) => setTimeout(resolve, 1050));
await provider(workspace);
await vi.waitFor(() => expect(subscriptions()).toHaveLength(1));
const watcher = subscriptions()[0]!;
await vi.waitFor(() => expect(watcher.getWatched()[directory]).toEqual([]));
const event = new Promise<void>((resolve) =>
watcher.once('all', () => resolve()),
);
const skill = path.join(directory, 'external');
await fs.mkdir(skill);
await fs.writeFile(
path.join(skill, 'SKILL.md'),
'---\nname: external\ndescription: external\n---\nInstructions\n',
);
await event;
} finally {
await provider.dispose?.();
await fs.rm(root, { recursive: true, force: true });
}
}, 10000);
it('discovers externally created Skills through both config routes without a child', async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-15645-routes-'));
const workspace = path.join(root, 'workspace');
const qwenHome = path.join(root, 'home');
await fs.mkdir(workspace, { recursive: true });
await fs.mkdir(qwenHome, { recursive: true });
await fs.writeFile(path.join(qwenHome, 'settings.json'), '{}');
vi.stubEnv('QWEN_HOME', qwenHome);
const provider = createWorkspaceSkillsStatusProvider({ autoDiscover: true });
const runtime = {
workspaceId: 'selected',
workspaceCwd: workspace,
primary: true,
trusted: true,
} as WorkspaceRuntime;
const registry = createWorkspaceRegistry([runtime]);
const app = express();
const deps = {
workspaceRegistry: registry,
getSkillsConfigStatus: (cwd: string) => provider(cwd),
invalidateSkillsConfigStatus: (cwd: string) => provider.invalidate?.(cwd),
sendBridgeError,
mutate:
() =>
(
_req: express.Request,
_res: express.Response,
next: express.NextFunction,
) =>
next(),
safeBody: (req: express.Request) => req.body as Record<string, unknown>,
};
registerWorkspaceSkillsRoutes(app, {
...deps,
workspaceRuntime: runtime,
parseAndValidateClientId: () => null,
installSkillConfig: async () => {
throw new Error('unused mutation');
},
deleteSkillConfig: async () => {
throw new Error('unused mutation');
},
});
registerWorkspaceQualifiedSkillsRoutes(app, deps);
const routes = [
'/workspace/config/skills',
'/workspaces/selected/config/skills',
];
try {
for (const route of routes) {
const response = await request(app).get(route);
expect(response.status).toBe(200);
expect(response.body.initialized).toBe(true);
expect(
response.body.skills.some(
(skill: { name: string }) => skill.name === 'external-project',
),
).toBe(false);
}
const skillDir = path.join(
workspace,
'.qwen',
'skills',
'external-project',
);
await fs.mkdir(skillDir, { recursive: true });
await fs.writeFile(
path.join(skillDir, 'SKILL.md'),
'---\nname: external-project\ndescription: Created externally\n---\nInstructions\n',
);
expect(
(await provider.snapshot(workspace)).skills.some(
(skill) => skill.name === 'external-project',
),
).toBe(true);
const started = Date.now();
let seen = [false, false];
do {
seen = await Promise.all(
routes.map(async (route) => {
const response = await request(app).get(route);
expect(response.status).toBe(200);
expect(response.body.initialized).toBe(true);
return response.body.skills.some(
(skill: { name: string }) => skill.name === 'external-project',
) as boolean;
}),
);
if (seen.every(Boolean)) break;
await new Promise((resolve) => setTimeout(resolve, 1000));
} while (Date.now() - started < 35000);
expect(
seen,
`Routes did not observe external creation after ${Date.now() - started} ms`,
).toEqual([true, true]);
} finally {
await provider.dispose?.();
await fs.rm(root, { recursive: true, force: true });
}
}, 45000);
@@ -0,0 +1,214 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { EventEmitter } from 'node:events';
import * as fs from 'node:fs/promises';
import * as os from 'node:os';
import * as path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { SkillManager } from '@qwen-code/qwen-code-core/skills/skill-manager.js';
import { createWorkspaceSkillsStatusProvider } from './workspace-skills-status.js';
import { getServeAppLifecycle } from './serve-app-lifecycle.js';
const { watch, routeDeps } = vi.hoisted(() => ({
watch: vi.fn(),
routeDeps: { current: undefined as unknown },
}));
vi.mock('chokidar', () => ({ watch }));
vi.mock('./workspace-service/index.js', () => ({
createDaemonWorkspaceService: vi.fn(() => ({})),
}));
vi.mock('./routes/workspace-skills.js', async (importOriginal) => {
const actual =
await importOriginal<typeof import('./routes/workspace-skills.js')>();
return {
...actual,
registerWorkspaceSkillsRoutes: (
...args: Parameters<typeof actual.registerWorkspaceSkillsRoutes>
) => {
routeDeps.current = args[1];
actual.registerWorkspaceSkillsRoutes(...args);
},
};
});
class Watcher extends EventEmitter {
close = vi.fn(async () => {
this.removeAllListeners();
});
}
describe('PR 16262 Critical reproduction', () => {
let root: string;
let workspace: string;
let home: string;
let now: number;
let providers: Array<ReturnType<typeof createWorkspaceSkillsStatusProvider>>;
let watchers: Map<string, Watcher>;
let closeApp: (() => Promise<unknown>) | undefined;
beforeEach(async () => {
root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-review-repro-'));
workspace = path.join(root, 'workspace');
home = path.join(root, 'home');
await fs.mkdir(workspace);
await fs.mkdir(home);
vi.stubEnv('QWEN_HOME', home);
vi.stubEnv('QWEN_CODE_SAFE_MODE', '0');
now = 1_000_000;
vi.spyOn(Date, 'now').mockImplementation(() => now);
providers = [];
watchers = new Map();
watch.mockImplementation((directory: string) => {
const watcher = new Watcher();
watchers.set(directory, watcher);
return watcher;
});
});
afterEach(async () => {
await closeApp?.();
closeApp = undefined;
await Promise.all(providers.map((provider) => provider.dispose?.()));
vi.restoreAllMocks();
vi.unstubAllEnvs();
watch.mockReset();
await fs.rm(root, { recursive: true, force: true });
});
async function writeSkill(name: string, cwd = workspace) {
const directory = path.join(cwd, '.qwen', 'skills', name);
await fs.mkdir(directory, { recursive: true });
await fs.writeFile(
path.join(directory, 'SKILL.md'),
`---\nname: ${name}\ndescription: ${name}\n---\nInstructions\n`,
);
return directory;
}
function create() {
const provider = createWorkspaceSkillsStatusProvider({
autoDiscover: true,
});
providers.push(provider);
return provider;
}
it('R1-1 discovers an unrelated valid addition despite a persistent malformed manifest', async () => {
await writeSkill('retained');
const provider = create();
await provider(workspace);
const malformed = await writeSkill('broken');
await fs.writeFile(
path.join(malformed, 'SKILL.md'),
'---\nname: broken\n---\nMissing description\n',
);
await writeSkill('valid-addition');
for (let window = 0; window < 4; window++) {
now += 1000;
await provider(workspace);
await new Promise((resolve) => setTimeout(resolve, 100));
}
await expect(provider.snapshot(workspace)).rejects.toThrow(
'could not be completely discovered',
);
expect((await provider(workspace)).skills.map((s) => s.name)).toContain(
'valid-addition',
);
});
it('R1-2 bounds outstanding discovery after repeated scheduling timeouts', async () => {
await writeSkill('retained');
const provider = create();
await provider(workspace);
const original = SkillManager.prototype.listSkills;
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
let scans = 0;
vi.spyOn(SkillManager.prototype, 'listSkills').mockImplementation(
async function (this: SkillManager, options) {
const cold = this.getCachedSkills() === null;
const result = await original.call(this, options);
if (cold) {
scans++;
await gate;
}
return result;
},
);
try {
for (let window = 0; window < 4; window++) {
now += 6000;
await provider(workspace);
await new Promise((resolve) => setTimeout(resolve, 5100));
}
expect(scans).toBe(2);
} finally {
release();
await new Promise((resolve) => setTimeout(resolve, 50));
}
}, 30000);
it.each(['untrusted-read', 'trust-drain'] as const)(
'R1-3 releases trusted subscriptions on %s while preserving shared watchers',
async (transition) => {
await writeSkill('project');
await fs.mkdir(path.join(home, 'skills'));
const other = path.join(root, 'other');
await fs.mkdir(other);
const { createServeApp } = await import('./server.js');
const app = createServeApp(
{ hostname: '127.0.0.1', port: 4170, mode: 'http-bridge' },
undefined,
{
boundWorkspace: workspace,
bridge: {
isChannelLive: vi.fn(() => false),
queryWorkspaceStatus: vi.fn(),
invokeWorkspaceCommand: vi.fn(),
refreshExtensionsForAllSessions: vi.fn(async () => ({
refreshed: 0,
failed: 0,
})),
publishWorkspaceEvent: vi.fn(),
shutdown: vi.fn(async () => {}),
},
} as never,
);
closeApp = () => getServeAppLifecycle(app).close();
const deps = routeDeps.current as {
getSkillsConfigStatus: (
cwd: string,
trusted: boolean,
) => Promise<unknown>;
};
await deps.getSkillsConfigStatus(workspace, true);
await deps.getSkillsConfigStatus(other, true);
const projectWatcher = watchers.get(
path.join(workspace, '.qwen', 'skills'),
)!;
const sharedWatcher = watchers.get(path.join(home, 'skills'))!;
expect(projectWatcher).toBeDefined();
if (transition === 'untrusted-read') {
await deps.getSkillsConfigStatus(workspace, false);
} else {
const invalidate = app.locals['invalidateSkillsConfigStatus'] as (
cwd: string,
) => void;
invalidate(workspace);
}
expect(projectWatcher.close).toHaveBeenCalledOnce();
expect(sharedWatcher.close).not.toHaveBeenCalled();
const scanned = vi.spyOn(SkillManager.prototype, 'listSkills');
projectWatcher.emit('all', 'add', 'project/SKILL.md');
await new Promise((resolve) => setTimeout(resolve, 250));
expect(scanned).not.toHaveBeenCalled();
},
30000,
);
});
+263 -18
View File
@@ -30,29 +30,40 @@
* runtime Config to the ExtensionManager.
*/
import { ExtensionManager } from '@qwen-code/qwen-code-core/extension/extensionManager.js';
import { ExtensionStore } from '@qwen-code/qwen-code-core/extension/extension-store.js';
import {
ExtensionManager,
ExtensionStore,
SkillManager,
Storage,
isSafeModeEnv,
getExtensionDisplayName,
watcherIgnored,
WATCHER_MAX_DEPTH,
} from '@qwen-code/qwen-code-core/skills/skill-manager.js';
import { Storage } from '@qwen-code/qwen-code-core/config/storage.js';
import { isSafeModeEnv } from '@qwen-code/qwen-code-core/utils/safe-mode.js';
import { closeFileWatcher } from '@qwen-code/qwen-code-core/utils/file-watcher-cleanup.js';
import { getExtensionDisplayName } from '@qwen-code/qwen-code-core/extension/i18n.js';
import {
authoredSkillName,
qualifySkillName,
} from '@qwen-code/qwen-code-core';
} from '@qwen-code/qwen-code-core/skills/types.js';
import type { Config, Extension, SkillLevel } from '@qwen-code/qwen-code-core';
import type { ServeWorkspaceSkillsStatus } from '@qwen-code/acp-bridge/status';
import { STATUS_SCHEMA_VERSION } from '@qwen-code/acp-bridge/status';
import * as fs from 'node:fs/promises';
import { watch, type FSWatcher } from 'chokidar';
import { loadSettings } from '../config/settings.js';
import { resolveLanguage, resolveLanguageSetting } from '../i18n/index.js';
import { writeStderrLine } from '../utils/stdioHelpers.js';
import { mapSkillConfigToStatus } from '../runtime/workspace-skills-mapping.js';
import { resolveSkillSettings } from '../config/skill-settings.js';
import {
WORKSPACE_SKILLS_REVALIDATE_INTERVAL_MS,
WORKSPACE_SKILLS_REVALIDATE_TIMEOUT_MS,
} from './workspace-skills-refresh-policy.js';
export interface WorkspaceSkillsStatusProvider {
(workspaceCwd: string): Promise<ServeWorkspaceSkillsStatus>;
invalidate?(workspaceCwd: string): void;
dispose?(): Promise<void>;
}
export interface WorkspaceSkillsSnapshot {
@@ -70,6 +81,8 @@ export interface WorkspaceSkillsStatusProviderOptions {
workspaceTrusted?: boolean;
/** Read inert on-disk Skill manifests without loading workspace settings. */
includeUntrustedSkills?: boolean;
/** Enable independent automatic discovery for the configuration catalog. */
autoDiscover?: boolean;
}
const VALID_SKILL_LEVELS: ReadonlySet<string> = new Set<SkillLevel>([
@@ -101,6 +114,19 @@ interface WorkspaceSkillManagers {
skillManager: SkillManager;
extensionManager?: ExtensionManager;
extensionSkillStates: Map<Extension, Map<string, boolean>>;
watchDirectories: string[];
}
interface SkillsDiscoveryState {
safeMode: boolean;
startedAt: number;
sequence: number;
publishedSequence: number;
activeScans: number;
dirty: boolean;
directories: Set<string>;
debounce?: NodeJS.Timeout;
pending?: { timeout: NodeJS.Timeout };
}
export function createWorkspaceSkillsStatusProvider(
@@ -108,31 +134,224 @@ export function createWorkspaceSkillsStatusProvider(
): WorkspaceSkillsStatusProvider & {
snapshot(workspaceCwd: string): Promise<WorkspaceSkillsSnapshot>;
} {
// Reuse one SkillManager per workspace so repeat queries hit its in-memory
// skills cache instead of re-scanning (and re-parsing frontmatter / compiling
// globs for) every level on each call. This is a best-effort pre-child
// fallback, so slight staleness between explicit invalidation points is
// acceptable: the live child re-lists authoritatively once a session exists.
// Warm reads reuse a committed manager while discovery builds its replacement
// separately. Runtime fallback callers keep explicit invalidation semantics.
const managers = new Map<string, WorkspaceSkillManagers>();
// An enumeration that overlaps an invalidation captured pre-invalidation
// settings, so it must not install its managers; the next read rebuilds.
const epochs = new Map<string, number>();
const inFlight = new Map<string, Promise<ServeWorkspaceSkillsStatus>>();
const discovery = new Map<string, SkillsDiscoveryState>();
const watchers = new Map<
string,
{ watcher: FSWatcher; workspaces: Set<string> }
>();
const closing = new Set<Promise<void>>();
const watcherRetryAfter = new Map<string, number>();
let disposed = false;
function closeWatcher(directory: string): void {
const subscription = watchers.get(directory);
if (!subscription) return;
watchers.delete(directory);
const pending = closeFileWatcher(subscription.watcher)
.catch((error: unknown) => {
writeStderrLine(
`qwen serve: Skills watcher close failed: ${String(error)}`,
);
})
.finally(() => closing.delete(pending));
closing.add(pending);
}
function releaseDirectories(
workspaceCwd: string,
state: SkillsDiscoveryState,
) {
for (const directory of state.directories) {
const subscription = watchers.get(directory);
subscription?.workspaces.delete(workspaceCwd);
if (subscription?.workspaces.size === 0) closeWatcher(directory);
}
state.directories.clear();
}
function scheduleChange(workspaceCwd: string, state: SkillsDiscoveryState) {
if (disposed || discovery.get(workspaceCwd) !== state) return;
state.dirty = true;
if (state.debounce) clearTimeout(state.debounce);
state.debounce = setTimeout(() => {
state.debounce = undefined;
refresh(workspaceCwd, state);
}, 150);
state.debounce.unref();
}
function subscribe(workspaceCwd: string, state: SkillsDiscoveryState) {
const cached = managers.get(workspaceCwd);
if (!cached || disposed || discovery.get(workspaceCwd) !== state) return;
const directories = new Set(cached.watchDirectories);
for (const directory of state.directories) {
if (directories.has(directory)) continue;
const subscription = watchers.get(directory);
subscription?.workspaces.delete(workspaceCwd);
if (subscription?.workspaces.size === 0) closeWatcher(directory);
}
state.directories = directories;
for (const directory of directories) {
const existing = watchers.get(directory);
if (existing) {
existing.workspaces.add(workspaceCwd);
continue;
}
if (Date.now() < (watcherRetryAfter.get(directory) ?? 0)) continue;
try {
const workspaces = new Set([workspaceCwd]);
const watcher = watch(directory, {
ignoreInitial: true,
ignored: watcherIgnored,
depth: WATCHER_MAX_DEPTH,
});
watchers.set(directory, { watcher, workspaces });
watcherRetryAfter.delete(directory);
watcher.on('all', () => {
for (const cwd of workspaces) {
const current = discovery.get(cwd);
if (current) scheduleChange(cwd, current);
}
});
watcher.on('error', (error: unknown) => {
if (watchers.get(directory)?.watcher !== watcher) return;
watcherRetryAfter.set(directory, Date.now() + 30_000);
writeStderrLine(
`qwen serve: Skills watcher failed: ${String(error)}`,
);
if (watchers.get(directory)?.watcher === watcher)
closeWatcher(directory);
});
} catch (error) {
watcherRetryAfter.set(directory, Date.now() + 30_000);
writeStderrLine(`qwen serve: Skills watcher failed: ${String(error)}`);
}
}
}
function refresh(workspaceCwd: string, state: SkillsDiscoveryState): void {
if (
disposed ||
discovery.get(workspaceCwd) !== state ||
state.pending ||
state.activeScans >= 2
)
return;
state.activeScans++;
state.dirty = false;
state.startedAt = Date.now();
const sequence = ++state.sequence;
const freshManagers = new Map<string, WorkspaceSkillManagers>();
const release = () => {
clearTimeout(pending.timeout);
if (state.pending === pending) state.pending = undefined;
if (!state.pending && state.dirty && state.activeScans < 2)
scheduleChange(workspaceCwd, state);
};
const pending = {
timeout: setTimeout(release, WORKSPACE_SKILLS_REVALIDATE_TIMEOUT_MS),
};
pending.timeout.unref();
state.pending = pending;
void buildWorkspaceSkillsStatus(
workspaceCwd,
freshManagers,
options.workspaceTrusted ?? true,
options.includeUntrustedSkills ?? false,
)
.then((status) => {
const candidate = freshManagers.get(workspaceCwd);
if (
!status.initialized ||
status.errors?.length ||
!candidate ||
candidate.skillManager.hasDiscoveryReadErrors()
)
return;
// A timeout releases the slot, not publication rights. A valid slow
// scan can still publish, but cannot overwrite a newer successful one.
if (
disposed ||
discovery.get(workspaceCwd) !== state ||
state.safeMode !== isSafeModeEnv() ||
sequence <= state.publishedSequence
)
return;
state.publishedSequence = sequence;
managers.set(workspaceCwd, candidate);
subscribe(workspaceCwd, state);
})
.catch((error: unknown) => {
writeStderrLine(
`qwen serve: Skills revalidation failed: ${String(error)}`,
);
})
.finally(() => {
state.activeScans--;
release();
});
}
const provider = ((workspaceCwd: string) => {
if (disposed) {
return Promise.resolve({
v: STATUS_SCHEMA_VERSION,
workspaceCwd,
initialized: false,
skills: [],
} satisfies ServeWorkspaceSkillsStatus);
}
let state = discovery.get(workspaceCwd);
if (state && state.safeMode !== isSafeModeEnv()) {
provider.invalidate?.(workspaceCwd);
state = undefined;
}
if (options.autoDiscover && !state) {
state = {
safeMode: isSafeModeEnv(),
startedAt: Date.now(),
sequence: 0,
publishedSequence: 0,
activeScans: 0,
dirty: false,
directories: new Set(),
};
discovery.set(workspaceCwd, state);
}
const pending = inFlight.get(workspaceCwd);
if (pending) return pending;
const epoch = epochs.get(workspaceCwd) ?? 0;
const current = () =>
!disposed && (epochs.get(workspaceCwd) ?? 0) === epoch;
if (state && managers.has(workspaceCwd)) {
const elapsed = Date.now() - state.startedAt;
if (elapsed < 0 || elapsed >= WORKSPACE_SKILLS_REVALIDATE_INTERVAL_MS) {
refresh(workspaceCwd, state);
}
}
const promise = buildWorkspaceSkillsStatus(
workspaceCwd,
managers,
options.workspaceTrusted ?? true,
options.includeUntrustedSkills ?? false,
() => (epochs.get(workspaceCwd) ?? 0) === epoch,
).finally(() => {
if (inFlight.get(workspaceCwd) === promise) {
inFlight.delete(workspaceCwd);
}
});
current,
)
.then((status) => {
if (state && current()) subscribe(workspaceCwd, state);
return status;
})
.finally(() => {
if (inFlight.get(workspaceCwd) === promise) {
inFlight.delete(workspaceCwd);
}
});
inFlight.set(workspaceCwd, promise);
return promise;
}) as WorkspaceSkillsStatusProvider & {
@@ -144,6 +363,25 @@ export function createWorkspaceSkillsStatusProvider(
// An enumeration that started before this invalidation must not answer a
// read issued after it; the caller needs its own observation.
inFlight.delete(workspaceCwd);
const state = discovery.get(workspaceCwd);
discovery.delete(workspaceCwd);
if (state) {
clearTimeout(state.debounce);
clearTimeout(state.pending?.timeout);
releaseDirectories(workspaceCwd, state);
}
};
provider.dispose = async () => {
disposed = true;
for (const cwd of new Set([
...managers.keys(),
...inFlight.keys(),
...discovery.keys(),
])) {
provider.invalidate?.(cwd);
}
await Promise.all(closing);
watcherRetryAfter.clear();
};
provider.snapshot = async (workspaceCwd) => {
if (options.workspaceTrusted === false)
@@ -291,12 +529,14 @@ async function buildWorkspaceSkillsStatus(
getDisabledSkillLevels: () => disabledLevels,
};
const skillManager = new SkillManager(shim as Config);
const watchDirectories: string[] = [];
if (!safeMode) {
for (const level of ['project', 'user'] as const) {
if (disabledLevels.has(level)) continue;
for (const directory of skillManager.getSkillsBaseDirs(level)) {
try {
await fs.readdir(directory);
if (workspaceTrusted) watchDirectories.push(directory);
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') {
throw error;
@@ -305,7 +545,12 @@ async function buildWorkspaceSkillsStatus(
}
}
}
cached = { skillManager, extensionManager, extensionSkillStates };
cached = {
skillManager,
extensionManager,
extensionSkillStates,
watchDirectories,
};
if (!canInstall || canInstall()) managers.set(workspaceCwd, cached);
}
const { disablements, enabledNames } = resolveSkillSettings(settings);
@@ -1204,6 +1204,37 @@ describe('SkillManager', () => {
});
describe('discovery completeness', () => {
it('distinguishes invalid manifests from unreadable sources without weakening completeness', async () => {
mockReaddir(bareDirent('broken'));
vi.mocked(fs.access).mockResolvedValue(undefined);
vi.mocked(fs.readFile).mockResolvedValue('---\nname: broken\n---\n');
await manager.refreshCache();
expect(manager.hasDiscoveryErrors()).toBe(true);
expect(manager.hasDiscoveryReadErrors()).toBe(false);
vi.mocked(fs.readFile).mockRejectedValue(fsError('unreadable', 'EACCES'));
await manager.refreshCache();
expect(manager.hasDiscoveryReadErrors()).toBe(true);
vi.mocked(fs.readFile).mockResolvedValue(validMarkdown);
await manager.refreshCache();
expect(manager.hasDiscoveryErrors()).toBe(false);
expect(manager.hasDiscoveryReadErrors()).toBe(false);
});
it('keeps a source failure visible alongside a malformed manifest', async () => {
mockReaddir(bareDirent('broken'), bareDirent('unreadable'));
vi.mocked(fs.access).mockImplementation(async (file) => {
if (path.basename(path.dirname(String(file))) === 'unreadable')
throw fsError('unreadable', 'EACCES');
});
vi.mocked(fs.readFile).mockResolvedValue('---\nname: broken\n---\n');
await manager.refreshCache();
expect(manager.getParseErrors().size).toBeGreaterThan(0);
expect(manager.hasDiscoveryErrors()).toBe(true);
expect(manager.hasDiscoveryReadErrors()).toBe(true);
});
function createReadGate() {
let resolve!: (value: []) => void;
const promise = new Promise<[]>((release) => {
+20 -4
View File
@@ -95,6 +95,7 @@ export class SkillManager {
private slashReloadSuppressed = false;
private parseErrors: Map<string, SkillError> = new Map();
private discoveryHasErrors = false;
private discoveryHasReadErrors = false;
private readonly watchers: Map<string, FSWatcher> = new Map();
private watchStarted = false;
private refreshTimer: NodeJS.Timeout | null = null;
@@ -307,6 +308,11 @@ export class SkillManager {
return this.discoveryHasErrors;
}
/** Whether discovery could not read every enabled source. */
hasDiscoveryReadErrors(): boolean {
return this.discoveryHasReadErrors;
}
private collectCachedSkills(level?: SkillLevel): SkillConfig[] {
const skills: SkillConfig[] = [];
const seenNames = new Set<string>();
@@ -455,8 +461,17 @@ export class SkillManager {
const skillsCache = new Map<SkillLevel, SkillConfig[]>();
this.parseErrors.clear();
let discoveryHasErrors = false;
const onDiscoveryError = () => {
let discoveryHasReadErrors = false;
const onDiscoveryError = (error?: unknown) => {
discoveryHasErrors = true;
if (
!(
error instanceof SkillError &&
error.code === SkillErrorCode.PARSE_ERROR
)
) {
discoveryHasReadErrors = true;
}
};
// Safe mode: only load bundled (system) skills
@@ -500,6 +515,7 @@ export class SkillManager {
// Publish completeness with the cache produced by this scan.
this.discoveryHasErrors = discoveryHasErrors;
this.discoveryHasReadErrors = discoveryHasReadErrors;
this.skillsCache = skillsCache;
// Rebuild the activation registry so that newly added/removed `paths:`
@@ -1010,7 +1026,7 @@ export class SkillManager {
*/
private async listSkillsAtLevel(
level: SkillLevel,
onDiscoveryError?: () => void,
onDiscoveryError?: (error?: unknown) => void,
): Promise<SkillConfig[]> {
if (this.config.getBareMode()) {
debugLogger.debug(`Skipping ${level} level skills in bare mode`);
@@ -1135,7 +1151,7 @@ export class SkillManager {
async loadSkillsFromDir(
baseDir: string,
level: SkillLevel,
onDiscoveryError?: () => void,
onDiscoveryError?: (error?: unknown) => void,
): Promise<SkillConfig[]> {
debugLogger.debug(`Loading skills from directory: ${baseDir}`);
try {
@@ -1210,7 +1226,7 @@ export class SkillManager {
return await this.parseSkillFileInternal(skillManifest, level);
} catch (error) {
if (!(isNodeError(error) && error.code === 'ENOENT')) {
onDiscoveryError?.();
onDiscoveryError?.(error);
}
if (error instanceof SkillError) {
debugLogger.error(