From d6c77412f9600cd8fbba4c7c9d07c6ec3130e8ff Mon Sep 17 00:00:00 2001 From: limityan Date: Sun, 2 Aug 2026 22:49:45 +0800 Subject: [PATCH] perf(skills): cache standard user source discovery --- .../opencode-config-assets-adapter-design.md | 6 + .../opencode-extension-compatibility.md | 2 +- .../tools/implementations/skills/mod.rs | 2 + .../tools/implementations/skills/registry.rs | 586 +++++++++++--- .../implementations/skills/source_cache.rs | 622 +++++++++++++++ .../src/file_watch/service.rs | 734 +++++++++++++++--- .../src/file_watch/types.rs | 5 + .../tests/file_watch_contracts.rs | 69 ++ 8 files changed, 1784 insertions(+), 242 deletions(-) create mode 100644 src/crates/assembly/core/src/agentic/tools/implementations/skills/source_cache.rs diff --git a/docs/architecture/extensions/opencode-config-assets-adapter-design.md b/docs/architecture/extensions/opencode-config-assets-adapter-design.md index 3c63dd4d3b..896504605e 100644 --- a/docs/architecture/extensions/opencode-config-assets-adapter-design.md +++ b/docs/architecture/extensions/opencode-config-assets-adapter-design.md @@ -307,6 +307,12 @@ V1 `skills.paths` 和当前 `skills: string[]`；字段类型错误只拒绝该 拒绝符号链接/reparse point；加载时重新校验规范化根及其稳定 source slot，防止目录整体替换改变已发现来源身份。同 scope 内配置根位于标准 OpenCode 根之前，较后的不同配置根覆盖同名 Skill，但不重排更早的 BitFun/Claude/Codex/Cursor 来源。 +Skill Registry 只为与 workspace 无关的标准用户根维护进程级版本化候选快照，具体文件观察复用 File Watch Service。 +相关文件变化只使快照失效，下一次 Agent/Skill 查询重建，不改变正在执行的 Turn；瞬时读取失败或观察器无法覆盖任一目标根时 +不发布缓存，后续查询保持原有重扫行为。OpenCode 配置根的作用域取决于当前 workspace，因此继续在每次请求中按完整配置来源 +顺序统一发现和扫描，避免把项目内绝对路径误缓存为用户来源，也保证 64 根上限只计算一次。标准项目根与 Remote 项目来源同样 +按请求读取，后者继续通过 `WorkspaceFileSystem` 访问且不回退到控制端同名目录。 + ### 5.2.1 References Workspace Reference 不是第二套 Workspace，也不是文件权限入口。OpenCode adapter 只解析有界 JSON/JSONC 配置， diff --git a/docs/architecture/extensions/opencode-extension-compatibility.md b/docs/architecture/extensions/opencode-extension-compatibility.md index add24be959..e929a8068f 100644 --- a/docs/architecture/extensions/opencode-extension-compatibility.md +++ b/docs/architecture/extensions/opencode-extension-compatibility.md @@ -108,7 +108,7 @@ OpenCode，和 OpenCode 配置/插件进入 BitFun 是两个独立验收方向 | 独立 `tui.json/jsonc` | 融合现有能力 + 转换参数 | 未实现 | 可完整适配 | OC-R1 | 按 global、`OPENCODE_TUI_CONFIG`、project、`.opencode` 独立顺序加载，不能复用主配置优先级 | [TUI 来源](opencode-config-assets-adapter-design.md#32-tui-独立来源顺序) | | Rules / Instructions | 转换参数 | 部分实现：用户/项目本地文件与 glob | 可完整适配 | OC-R1 | OpenCode adapter 已读取用户全局 `AGENTS.md`/Claude fallback 和最终本地 `instructions`；Product Assembly 在 Codex/Claude 用户来源与项目来源之前合成并去重。远程 URL、managed/organization policy、完整配置来源图和变化监听仍未实现 | [声明式资产](opencode-config-assets-adapter-design.md#5-声明式资产映射) | | Agents / Modes | 融合现有能力 + 转换参数 | 部分实现：Subagent 安全子集、模型/profile 绑定与 Agent-local 权限约束 | 可主要适配 | OC-R1 | 已支持当前生产 V1 `agent/prompt/disable/permission` 与 Core V2 `agents/system/disabled/permissions` 的已验证安全子集、全局/项目 Markdown 和 JSON/JSONC、subagent/all、description、`Default`/不透明模型引用、不透明 `variant` 意图和工具映射，并接入唯一精确匹配、显式 BitFun 模型绑定、Web/TUI 可见性、审批、冲突、更新、撤下和 fresh single-run Task；仅显式声明 `model` 的 Agent 才保留 `variant`，未声明模型时与 OpenCode 一样不生效。保留的 `variant` 不推断为 reasoning effort，也不生成请求级 override，需显式绑定现有配置。不维护厂商别名、质量推断或自动 fallback。V1 `disable` 保持 deep-merge，V2 `disabled` 保持 remove/re-add，不能混用生命周期语义。有序 V2 permission rules 与 V1 扁平精确 action map 会成为只可收紧的独立约束。primary/mode、root ambient permission、V1 action pattern/嵌套 resource map、跨路径与命令资源域的歧义 pattern、options、采样与续接明确阻断或降级 | [Agents 与 Skills](opencode-config-assets-adapter-design.md#52-agentsmodes-与-skills) | -| Skills | 转换参数 | 部分实现：标准根与本地配置根 | 可完整适配 | OC-R2 | 现有 Registry 除标准用户/项目根外，也通过 `bitfun-core/external_sources` 组合边界按 OpenCode 配置来源顺序累加 V1 `skills.paths` 与当前迁移后的本地字符串数组；仅接受项目根/用户目录内的本地目录并做有界递归发现。同 scope 配置根覆盖标准 OpenCode 根，但不重排更早的 BitFun/Claude/Codex/Cursor 来源。URL、下载/缓存、完整 allow/deny/ask 顺序及外部来源策略仍未实现 | [Agents 与 Skills](opencode-config-assets-adapter-design.md#52-agentsmodes-与-skills) | +| Skills | 转换参数 | 部分实现：标准根、本地配置根与标准用户根变化失效 | 可完整适配 | OC-R2 | 现有 Registry 除标准用户/项目根外，也通过 `bitfun-core/external_sources` 组合边界按 OpenCode 配置来源顺序累加 V1 `skills.paths` 与当前迁移后的本地字符串数组；仅接受项目根/用户目录内的本地目录并做有界递归发现。与 workspace 无关的标准用户根复用版本化快照，文件变化使其失效并在下一次发现时重建；OpenCode 配置根因作用域依赖当前 workspace 而保持按请求统一发现，标准项目根与 Remote 项目来源也仍按请求读取。同 scope 配置根覆盖标准 OpenCode 根，但不重排更早的 BitFun/Claude/Codex/Cursor 来源。URL、下载/缓存、完整 allow/deny/ask 顺序及外部来源策略仍未实现 | [Agents 与 Skills](opencode-config-assets-adapter-design.md#52-agentsmodes-与-skills) | | References | 融合现有能力 + 转换参数 | 部分实现：本地目录与既有 Workspace 消费点 | 可主要适配 | OC-R2 | 已按 OpenCode V2 `e4bd9757` 的独立来源顺序解析 `references`/旧 `reference` 的本地 path、description/hidden，相同 alias 后者覆盖；通过独立生命周期协调器与 BitFun 原生关联目录合成 native-first 有效快照，接入关联目录弹窗和既有 `@` 目录选择器。外部声明不自动进入 Prompt、不授予文件权限；Git、Remote、下载/缓存明确不支持且不做临时实现 | [References](opencode-config-assets-adapter-design.md#521-references) | | Commands | 补扩展接口 + 转换参数 | 部分实现：prompt、本地文本文件、经审阅的 shell 上下文与显式 Subagent 委派 | 可完整适配 | OC-R2 | 已支持全局/项目 JSON、JSONC、Markdown 命令、参数展开、动态目录、刷新和显式冲突选择；模板中的静态 workspace 相对 `@file` 可在调用时有界读取，`!shell` 经精确计划审阅后仅把 stdout 加入 Prompt，静态计划可记住、参数相关计划仅可单次运行。仅 `agent` 加缺省/`true` 的 `subtask` 可委派给同 workspace、同 OpenCode 生态、已审批且仍有效的精确 Subagent，并复用现有 fresh Task 生命周期；shell 与委派的组合、`model`、`variant`、`subtask: false`、隐式默认 Agent、Remote 与附件上下文保持受限，不回退到当前 Agent 或本机执行 | [Commands](opencode-config-assets-adapter-design.md#53-commands) | | Models / Providers 配置 | 融合现有能力 | 未实现 | 可主要适配 | OC-R1 | 静态字段进入模型归属模块；动态模型、鉴权和请求头交给插件运行时 | [声明式资产](opencode-config-assets-adapter-design.md#5-声明式资产映射) | diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skills/mod.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skills/mod.rs index 1cf5b42667..964d25e389 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/skills/mod.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skills/mod.rs @@ -8,6 +8,8 @@ pub mod mode_overrides; pub mod policy; pub mod registry; pub mod resolver; +#[cfg(feature = "file-watch")] +mod source_cache; pub mod types; pub use registry::SkillRegistry; diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skills/registry.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skills/registry.rs index ad931c6f0b..7897a562cf 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/skills/registry.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skills/registry.rs @@ -7,6 +7,8 @@ use super::mode_overrides::{ load_disabled_mode_skills_local, load_disabled_mode_skills_remote, load_globally_disabled_user_skills, load_user_mode_skill_overrides, UserModeSkillOverrides, }; +#[cfg(feature = "file-watch")] +use super::source_cache::{LocalSkillWatchMonitor, LocalSkillWatchRoot, VersionedSnapshotCache}; use super::types::{ModeSkillInfo, SkillData, SkillInfo, SkillLocation}; use crate::agentic::workspace::WorkspaceFileSystem; #[cfg(feature = "product-full")] @@ -26,10 +28,11 @@ use bitfun_agent_runtime::skills::{ BITFUN_USER_SKILL_SLOT, PROJECT_SKILL_KEY_PREFIX, PROJECT_SKILL_ROOTS, USER_CONFIG_SKILL_ROOTS, USER_HOME_SKILL_ROOTS, USER_SKILL_KEY_PREFIX, }; +use bitfun_services_core::bounded_fs::is_symlink_or_reparse; #[cfg(feature = "product-full")] use bitfun_services_core::bounded_fs::{collect_bounded_regular_files, BoundedDirectoryWalkLimits}; #[cfg(feature = "product-full")] -use bitfun_services_core::bounded_fs::{is_symlink_or_reparse, read_bounded_text, BoundedTextRead}; +use bitfun_services_core::bounded_fs::{read_bounded_text, BoundedTextRead}; #[cfg(feature = "product-full")] use bitfun_services_core::workspace_text::read_workspace_relative_text_bounded; use log::{debug, error, warn}; @@ -39,7 +42,6 @@ use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::OnceLock; use tokio::fs; -use tokio::sync::RwLock; #[cfg(feature = "product-full")] const MAX_OPENCODE_CONFIGURED_SKILL_ROOTS: usize = 64; @@ -76,6 +78,187 @@ struct RemoteSkillRootEntry { priority: usize, } +#[derive(Debug, Clone)] +struct UserSkillSources { + standard: Vec, + cacheable: bool, + #[cfg(feature = "file-watch")] + watch_roots: Vec, +} + +struct LocalSkillScan { + candidates: Vec, + cacheable: bool, +} + +async fn local_source_path_is_cacheable(path: &Path) -> bool { + match fs::symlink_metadata(path).await { + Ok(metadata) => !is_symlink_or_reparse(&metadata), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => true, + Err(_) => false, + } +} + +#[cfg(test)] +mod local_skill_scan_tests { + use super::{SkillLocation, SkillRegistry, SkillRootEntry}; + use std::fs; + use std::path::Path; + + fn write_skill(path: &Path) { + fs::create_dir_all(path).expect("skill directory"); + fs::write( + path.join("SKILL.md"), + "---\nname: shared-review\ndescription: Shared review workflow\n---\n", + ) + .expect("skill markdown"); + } + + fn test_root(path: impl Into) -> SkillRootEntry { + SkillRootEntry { + path: path.into(), + level: SkillLocation::User, + slot: "test", + source_id: "test", + source_label: "Test", + priority: 0, + is_builtin: false, + } + } + + #[cfg(unix)] + fn create_dir_symlink(target: &Path, link: &Path) -> bool { + std::os::unix::fs::symlink(target, link).is_ok() + } + + #[cfg(windows)] + fn create_dir_symlink(target: &Path, link: &Path) -> bool { + std::os::windows::fs::symlink_dir(target, link).is_ok() + } + + #[cfg(unix)] + fn create_file_symlink(target: &Path, link: &Path) -> bool { + std::os::unix::fs::symlink(target, link).is_ok() + } + + #[cfg(windows)] + fn create_file_symlink(target: &Path, link: &Path) -> bool { + std::os::windows::fs::symlink_file(target, link).is_ok() + } + + #[tokio::test] + async fn standard_scan_follows_linked_skill_directories_without_caching() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = temp.path().join("skills"); + let shared_skill = temp.path().join("shared-review"); + fs::create_dir_all(&root).expect("skill root"); + write_skill(&shared_skill); + if !create_dir_symlink(&shared_skill, &root.join("review")) { + return; + } + let entry = test_root(root); + + let scan = SkillRegistry::scan_skills_in_dir_with_status(&entry).await; + + assert!(!scan.cacheable); + assert_eq!(scan.candidates.len(), 1); + assert_eq!(scan.candidates[0].info.name, "shared-review"); + } + + #[tokio::test] + async fn standard_scan_does_not_cache_a_broken_linked_root() { + let temp = tempfile::tempdir().expect("tempdir"); + let missing_target = temp.path().join("missing-skills"); + let root = temp.path().join("skills"); + if !create_dir_symlink(&missing_target, &root) { + return; + } + + let scan = SkillRegistry::scan_skills_in_dir_with_status(&test_root(root)).await; + + assert!(!scan.cacheable); + assert!(scan.candidates.is_empty()); + } + + #[tokio::test] + async fn standard_scan_does_not_cache_linked_skill_markdown() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = temp.path().join("skills"); + let skill_dir = root.join("review"); + let shared_markdown = temp.path().join("shared-SKILL.md"); + fs::create_dir_all(&skill_dir).expect("skill directory"); + fs::write( + &shared_markdown, + "---\nname: shared-review\ndescription: Shared review workflow\n---\n", + ) + .expect("shared skill markdown"); + if !create_file_symlink(&shared_markdown, &skill_dir.join("SKILL.md")) { + return; + } + + let scan = SkillRegistry::scan_skills_in_dir_with_status(&test_root(root)).await; + + assert!(!scan.cacheable); + assert_eq!(scan.candidates.len(), 1); + } + + #[tokio::test] + async fn standard_scan_does_not_cache_linked_openai_policy() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = temp.path().join("skills"); + let skill_dir = root.join("review"); + let shared_policy = temp.path().join("openai.yaml"); + write_skill(&skill_dir); + fs::create_dir_all(skill_dir.join("agents")).expect("policy directory"); + fs::write( + &shared_policy, + "policy:\n allow_implicit_invocation: false\n", + ) + .expect("shared policy"); + if !create_file_symlink(&shared_policy, &skill_dir.join("agents/openai.yaml")) { + return; + } + + let scan = SkillRegistry::scan_skills_in_dir_with_status(&test_root(root)).await; + + assert!(!scan.cacheable); + assert_eq!(scan.candidates.len(), 1); + assert!(!scan.candidates[0].info.allow_implicit_invocation); + } + + #[tokio::test] + async fn transient_policy_read_failure_is_not_cacheable() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = temp.path().join("skills"); + let skill_dir = root.join("review"); + fs::create_dir_all(skill_dir.join("agents/openai.yaml")).expect("policy-shaped directory"); + fs::write( + skill_dir.join("SKILL.md"), + "---\nname: review\ndescription: Review changes\n---\n", + ) + .expect("skill markdown"); + let entry = test_root(root); + + let failed = SkillRegistry::scan_skills_in_dir_with_status(&entry).await; + + assert!(!failed.cacheable); + assert!(failed.candidates[0].info.allow_implicit_invocation); + + fs::remove_dir(skill_dir.join("agents/openai.yaml")) + .expect("remove policy-shaped directory"); + fs::write( + skill_dir.join("agents/openai.yaml"), + "policy:\n allow_implicit_invocation: false\n", + ) + .expect("policy file"); + + let recovered = SkillRegistry::scan_skills_in_dir_with_status(&entry).await; + + assert!(recovered.cacheable); + assert!(!recovered.candidates[0].info.allow_implicit_invocation); + } +} + fn sort_remote_dir_entries(entries: &mut [crate::agentic::workspace::WorkspaceDirEntry]) { entries.sort_by(|a, b| { a.name @@ -133,15 +316,25 @@ fn validate_configured_opencode_skill_root( /// Skill registry pub struct SkillRegistry { - /// Cached raw user-level skills (no workspace-specific project skills). - cache: RwLock>, + #[cfg(feature = "file-watch")] + user_sources: VersionedSnapshotCache, + #[cfg(feature = "file-watch")] + user_source_monitor: LocalSkillWatchMonitor, } impl SkillRegistry { fn new() -> Self { - Self { - cache: RwLock::new(Vec::new()), + #[cfg(feature = "file-watch")] + { + let user_sources = VersionedSnapshotCache::new(); + let user_source_monitor = LocalSkillWatchMonitor::new(user_sources.invalidator()); + Self { + user_sources, + user_source_monitor, + } } + #[cfg(not(feature = "file-watch"))] + Self {} } fn parse_skill_markdown( @@ -178,18 +371,21 @@ impl SkillRegistry { .collect() } - async fn apply_local_openai_policy(skill_data: &mut SkillData, skill_dir: &Path) { - let policy_path = skill_dir.join("agents").join("openai.yaml"); + async fn apply_local_openai_policy(skill_data: &mut SkillData, skill_dir: &Path) -> bool { + let agents_dir = skill_dir.join("agents"); + let policy_path = agents_dir.join("openai.yaml"); + let cacheable = local_source_path_is_cacheable(&agents_dir).await + && local_source_path_is_cacheable(&policy_path).await; let content = match fs::read_to_string(&policy_path).await { Ok(content) => content, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return cacheable, Err(error) => { warn!( "Failed to read optional skill policy {}: {}", policy_path.display(), error ); - return; + return false; } }; @@ -200,6 +396,7 @@ impl SkillRegistry { error ); } + cacheable } #[cfg(feature = "product-full")] @@ -311,28 +508,30 @@ impl SkillRegistry { } } - fn get_possible_paths_for_workspace(workspace_root: Option<&Path>) -> Vec { + fn get_project_skill_roots(workspace_path: &Path) -> Vec { let mut entries = Vec::new(); let mut priority = 0usize; - let mut deferred_home_entries = Vec::new(); - if let Some(workspace_path) = workspace_root { - for spec in PROJECT_SKILL_ROOTS { - let path = workspace_path.join(spec.parent).join(spec.subdir); - if path.exists() && path.is_dir() { - entries.push(SkillRootEntry { - path, - level: SkillLocation::Project, - slot: spec.slot, - source_id: spec.source_id, - source_label: spec.source_label, - priority, - is_builtin: false, - }); - } - priority += 1; - } + for spec in PROJECT_SKILL_ROOTS { + let path = workspace_path.join(spec.parent).join(spec.subdir); + entries.push(SkillRootEntry { + path, + level: SkillLocation::Project, + slot: spec.slot, + source_id: spec.source_id, + source_label: spec.source_label, + priority, + is_builtin: false, + }); + priority += 1; } + entries + } + + fn get_user_skill_roots() -> Vec { + let mut entries = Vec::new(); + let mut priority = 0usize; + let mut deferred_home_entries = Vec::new(); let home_dir = dirs::home_dir(); @@ -346,7 +545,7 @@ impl SkillRegistry { spec.source_id, spec.source_label, )); - } else if path.exists() && path.is_dir() { + } else { entries.push(SkillRootEntry { path, level: SkillLocation::User, @@ -366,83 +565,180 @@ impl SkillRegistry { // while still keeping config-level overrides after BitFun defaults. let path_manager = get_path_manager_arc(); let bitfun_skills = path_manager.user_skills_dir(); - if bitfun_skills.exists() && bitfun_skills.is_dir() { - entries.push(SkillRootEntry { - path: bitfun_skills, - level: SkillLocation::User, - slot: BITFUN_USER_SKILL_SLOT, - source_id: BITFUN_SKILL_SOURCE_ID, - source_label: BITFUN_SKILL_SOURCE_LABEL, - priority, - is_builtin: false, - }); - } + entries.push(SkillRootEntry { + path: bitfun_skills, + level: SkillLocation::User, + slot: BITFUN_USER_SKILL_SLOT, + source_id: BITFUN_SKILL_SOURCE_ID, + source_label: BITFUN_SKILL_SOURCE_LABEL, + priority, + is_builtin: false, + }); priority += 1; let builtin_skills = path_manager.builtin_skills_dir(); - if builtin_skills.exists() && builtin_skills.is_dir() { - entries.push(SkillRootEntry { - path: builtin_skills, - level: SkillLocation::User, - slot: BITFUN_SYSTEM_SKILL_SLOT, - source_id: BITFUN_SKILL_SOURCE_ID, - source_label: BITFUN_SKILL_SOURCE_LABEL, - priority, - is_builtin: true, - }); - } + entries.push(SkillRootEntry { + path: builtin_skills, + level: SkillLocation::User, + slot: BITFUN_SYSTEM_SKILL_SLOT, + source_id: BITFUN_SKILL_SOURCE_ID, + source_label: BITFUN_SKILL_SOURCE_LABEL, + priority, + is_builtin: true, + }); priority += 1; if let Some(config_dir) = dirs::config_dir() { for spec in USER_CONFIG_SKILL_ROOTS { let path = resolve_user_config_skill_root(spec, &config_dir, home_dir.as_deref()); - if path.exists() && path.is_dir() { - entries.push(SkillRootEntry { - path, - level: SkillLocation::User, - slot: spec.slot, - source_id: spec.source_id, - source_label: spec.source_label, - priority, - is_builtin: false, - }); - } - priority += 1; - } - } - - for (path, slot, source_id, source_label) in deferred_home_entries { - if path.exists() && path.is_dir() { entries.push(SkillRootEntry { path, level: SkillLocation::User, - slot, - source_id, - source_label, + slot: spec.slot, + source_id: spec.source_id, + source_label: spec.source_label, priority, is_builtin: false, }); + priority += 1; } + } + + for (path, slot, source_id, source_label) in deferred_home_entries { + entries.push(SkillRootEntry { + path, + level: SkillLocation::User, + slot, + source_id, + source_label, + priority, + is_builtin: false, + }); priority += 1; } entries } + #[cfg(feature = "file-watch")] + fn standard_user_skill_watch_roots() -> Vec { + let mut roots = Vec::new(); + let home_dir = dirs::home_dir(); + if let Some(home) = home_dir.as_deref() { + roots.extend(USER_HOME_SKILL_ROOTS.iter().map(|spec| { + LocalSkillWatchRoot::recursive(home.join(spec.parent).join(spec.subdir)) + })); + } + + let path_manager = get_path_manager_arc(); + roots.push(LocalSkillWatchRoot::recursive( + path_manager.user_skills_dir(), + )); + roots.push(LocalSkillWatchRoot::recursive( + path_manager.builtin_skills_dir(), + )); + + if let Some(config_dir) = dirs::config_dir() { + roots.extend(USER_CONFIG_SKILL_ROOTS.iter().map(|spec| { + LocalSkillWatchRoot::recursive(resolve_user_config_skill_root( + spec, + &config_dir, + home_dir.as_deref(), + )) + })); + } + roots + } + async fn scan_skills_in_dir(entry: &SkillRootEntry) -> Vec { + Self::scan_skills_in_dir_with_status(entry).await.candidates + } + + async fn scan_skills_in_dir_with_status(entry: &SkillRootEntry) -> LocalSkillScan { let mut skills = Vec::new(); - if !entry.path.exists() { - return skills; + let root_cacheable = local_source_path_is_cacheable(&entry.path).await; + match fs::metadata(&entry.path).await { + Ok(metadata) if metadata.is_dir() => {} + Ok(_) => { + return LocalSkillScan { + candidates: skills, + cacheable: root_cacheable, + }; + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return LocalSkillScan { + candidates: skills, + cacheable: root_cacheable, + }; + } + Err(error) => { + debug!( + "Failed to inspect Skill root {}: {}", + entry.path.display(), + error + ); + return LocalSkillScan { + candidates: skills, + cacheable: false, + }; + } } - let Ok(mut read_dir) = fs::read_dir(&entry.path).await else { - return skills; + let mut read_dir = match fs::read_dir(&entry.path).await { + Ok(read_dir) => read_dir, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return LocalSkillScan { + candidates: skills, + cacheable: root_cacheable, + }; + } + Err(error) => { + debug!( + "Failed to read Skill root {}: {}", + entry.path.display(), + error + ); + return LocalSkillScan { + candidates: skills, + cacheable: false, + }; + } }; - - while let Ok(Some(item)) = read_dir.next_entry().await { + let mut cacheable = root_cacheable; + + loop { + let item = match read_dir.next_entry().await { + Ok(Some(item)) => item, + Ok(None) => break, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + skills.clear(); + break; + } + Err(error) => { + debug!( + "Failed while reading Skill root {}: {}", + entry.path.display(), + error + ); + cacheable = false; + break; + } + }; let path = item.path(); - if !path.is_dir() { - continue; + cacheable &= local_source_path_is_cacheable(&path).await; + match fs::metadata(&path).await { + Ok(metadata) if metadata.is_dir() => {} + Ok(_) => continue, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => { + debug!( + "Failed to inspect Skill entry {}: {}", + path.display(), + error + ); + cacheable = false; + continue; + } } let Some(dir_name) = normalize_local_skill_dir_name(&path) else { @@ -454,8 +750,16 @@ impl SkillRegistry { } let skill_md_path = path.join("SKILL.md"); - if !skill_md_path.exists() { - continue; + cacheable &= local_source_path_is_cacheable(&skill_md_path).await; + match fs::metadata(&skill_md_path).await { + Ok(metadata) if metadata.is_file() => {} + Ok(_) => continue, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => { + debug!("Failed to inspect {}: {}", skill_md_path.display(), error); + cacheable = false; + continue; + } } match fs::read_to_string(&skill_md_path).await { @@ -467,7 +771,7 @@ impl SkillRegistry { entry.slot, ) { Ok(mut skill_data) => { - Self::apply_local_openai_policy(&mut skill_data, &path).await; + cacheable &= Self::apply_local_openai_policy(&mut skill_data, &path).await; skill_data.dir_name = dir_name; let key_prefix = match entry.level { SkillLocation::User => USER_SKILL_KEY_PREFIX, @@ -489,44 +793,109 @@ impl SkillRegistry { }, Err(error) => { debug!("Failed to read {}: {}", skill_md_path.display(), error); + if error.kind() != std::io::ErrorKind::NotFound { + cacheable = false; + } } } } - sort_skill_candidates_by_dir(skills) + LocalSkillScan { + candidates: sort_skill_candidates_by_dir(skills), + cacheable, + } + } + + async fn scan_user_skill_sources() -> UserSkillSources { + let mut cacheable = match ensure_builtin_skills_installed().await { + Ok(()) => true, + Err(error) => { + debug!("Failed to install built-in skills: {}", error); + false + } + }; + + let mut standard = Vec::new(); + for entry in Self::get_user_skill_roots() { + let mut scan = Self::scan_skills_in_dir_with_status(&entry).await; + cacheable &= scan.cacheable; + standard.append(&mut scan.candidates); + } + + UserSkillSources { + standard, + cacheable, + #[cfg(feature = "file-watch")] + watch_roots: Self::standard_user_skill_watch_roots(), + } + } + + #[cfg(feature = "file-watch")] + async fn user_skill_sources(&self) -> UserSkillSources { + self.user_source_monitor.start(); + self.user_sources + .get_or_load(|| async { + let sources = Self::scan_user_skill_sources().await; + let cacheable = self + .user_source_monitor + .sync_roots(sources.watch_roots.clone()) + .await + && sources.cacheable; + (sources, cacheable) + }) + .await + } + + #[cfg(not(feature = "file-watch"))] + async fn user_skill_sources(&self) -> UserSkillSources { + Self::scan_user_skill_sources().await } async fn scan_skill_candidates_for_workspace( &self, workspace_root: Option<&Path>, ) -> Vec { - if let Err(error) = ensure_builtin_skills_installed().await { - debug!("Failed to install built-in skills: {}", error); + let mut user_sources = self.user_skill_sources().await; + let mut standard = Vec::new(); + if let Some(workspace_root) = workspace_root { + for entry in Self::get_project_skill_roots(workspace_root) { + let mut part = Self::scan_skills_in_dir(&entry).await; + standard.append(&mut part); + } + for candidate in &mut user_sources.standard { + candidate.priority = candidate.priority.saturating_add(PROJECT_SKILL_ROOTS.len()); + } + standard.append(&mut user_sources.standard); + } else { + standard.append(&mut user_sources.standard); } - let entries = Self::get_possible_paths_for_workspace(workspace_root); - let mut skills = Vec::new(); - for entry in entries { - let mut part = Self::scan_skills_in_dir(&entry).await; - skills.append(&mut part); - } #[cfg(feature = "product-full")] { - let existing_paths = skills + // OpenCode configured roots are workspace-sensitive: an absolute path + // from user config may become project-scoped for the current workspace. + // Discover and scan them once per request so scope and the 64-root cap + // are applied to one coherent OpenCode configuration snapshot. + let roots = opencode_configured_skill_roots(workspace_root); + let mut configured = Self::scan_configured_opencode_candidates(roots).await; + let existing_paths = standard .iter() .map(canonical_candidate_path) .collect::>(); - let roots = opencode_configured_skill_roots(workspace_root); - let mut configured = Self::scan_configured_opencode_candidates(roots).await; - configured - .retain(|candidate| !existing_paths.contains(&canonical_candidate_path(candidate))); - skills = Self::merge_configured_opencode_candidates( - skills, + let mut configured_paths = HashSet::new(); + configured.retain(|candidate| { + let path = canonical_candidate_path(candidate); + !existing_paths.contains(&path) && configured_paths.insert(path) + }); + return Self::merge_configured_opencode_candidates( + standard, configured, workspace_root.is_some(), ); } - skills + + #[cfg(not(feature = "product-full"))] + standard } #[cfg(feature = "product-full")] @@ -998,20 +1367,9 @@ impl SkillRegistry { ) } - async fn ensure_loaded(&self) { - let cache = self.cache.read().await; - if cache.is_empty() { - drop(cache); - self.refresh().await; - } - } - pub async fn refresh(&self) { - let skills = sort_skills(annotate_shadowed_skills( - self.scan_skill_candidates_for_workspace(None).await, - )); - let mut cache = self.cache.write().await; - *cache = skills; + #[cfg(feature = "file-watch")] + self.user_sources.invalidate(); } pub async fn refresh_for_workspace(&self, _workspace_root: Option<&Path>) { @@ -1019,9 +1377,7 @@ impl SkillRegistry { } pub async fn get_all_skills(&self) -> Vec { - self.ensure_loaded().await; - let cache = self.cache.read().await; - cache.clone() + self.get_all_skills_for_workspace(None).await } pub async fn get_all_skills_for_workspace( diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skills/source_cache.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skills/source_cache.rs new file mode 100644 index 0000000000..e22eadf9aa --- /dev/null +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skills/source_cache.rs @@ -0,0 +1,622 @@ +use bitfun_services_integrations::file_watch::{ + FileWatchEventKind, FileWatchService, FileWatcherConfig, +}; +use std::collections::BTreeMap; +use std::future::Future; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, RwLock as StdRwLock}; +use tokio::sync::{Mutex, RwLock}; + +// The cache itself is lazy, so coalescing invalidations has no scan-cost benefit +// and would only create a stale window before the next registry query. +const SKILL_WATCH_DEBOUNCE_MS: u64 = 0; +const MAX_SNAPSHOT_LOAD_ATTEMPTS: usize = 2; + +#[derive(Clone)] +pub(super) struct SnapshotInvalidator { + generation: Arc, +} + +impl SnapshotInvalidator { + fn new() -> Self { + Self { + generation: Arc::new(AtomicU64::new(0)), + } + } + + fn current(&self) -> u64 { + self.generation.load(Ordering::Acquire) + } + + pub(super) fn invalidate(&self) { + self.generation.fetch_add(1, Ordering::AcqRel); + } +} + +struct CachedSnapshot { + generation: u64, + value: T, +} + +/// A lazy, single-flight snapshot that cannot publish a value observed across +/// an invalidation boundary. +pub(super) struct VersionedSnapshotCache { + invalidator: SnapshotInvalidator, + snapshot: RwLock>>, + load_gate: Mutex<()>, +} + +impl VersionedSnapshotCache { + pub(super) fn new() -> Self { + Self { + invalidator: SnapshotInvalidator::new(), + snapshot: RwLock::new(None), + load_gate: Mutex::new(()), + } + } + + pub(super) fn invalidator(&self) -> SnapshotInvalidator { + self.invalidator.clone() + } + + pub(super) fn invalidate(&self) { + self.invalidator.invalidate(); + } + + pub(super) async fn get_or_load(&self, mut load: F) -> T + where + F: FnMut() -> Fut, + Fut: Future, + { + let mut attempts = 0usize; + loop { + let generation = self.invalidator.current(); + if let Some(value) = self.value_for_generation(generation).await { + return value; + } + + let _load_guard = self.load_gate.lock().await; + let generation = self.invalidator.current(); + if let Some(value) = self.value_for_generation(generation).await { + return value; + } + + let (value, cacheable) = load().await; + if !cacheable { + return value; + } + if self.invalidator.current() != generation { + attempts = attempts.saturating_add(1); + if attempts >= MAX_SNAPSHOT_LOAD_ATTEMPTS { + return value; + } + continue; + } + + *self.snapshot.write().await = Some(CachedSnapshot { + generation, + value: value.clone(), + }); + return value; + } + } + + async fn value_for_generation(&self, generation: u64) -> Option { + self.snapshot + .read() + .await + .as_ref() + .filter(|snapshot| snapshot.generation == generation) + .map(|snapshot| snapshot.value.clone()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct LocalSkillWatchRoot { + pub(super) path: PathBuf, + pub(super) recursive: bool, +} + +impl LocalSkillWatchRoot { + pub(super) fn recursive(path: PathBuf) -> Self { + Self { + path, + recursive: true, + } + } +} + +/// Process-lifetime watcher for user-level Skill sources. Project roots are +/// intentionally excluded because they follow workspace/session lifecycles and +/// continue to be read on demand by the registry. +pub(super) struct LocalSkillWatchMonitor { + watcher: Arc, + invalidator: SnapshotInvalidator, + registrations: Mutex>, + desired_paths: Arc>>, + rebind_required: Arc, + started: AtomicBool, +} + +impl LocalSkillWatchMonitor { + pub(super) fn new(invalidator: SnapshotInvalidator) -> Self { + let mut config = FileWatcherConfig::default(); + config.ignore_hidden_files = false; + config.ignore_common_build_directories = false; + config.debounce_interval_ms = SKILL_WATCH_DEBOUNCE_MS; + Self { + watcher: Arc::new(FileWatchService::new(config)), + invalidator, + registrations: Mutex::new(BTreeMap::new()), + desired_paths: Arc::new(StdRwLock::new(Vec::new())), + rebind_required: Arc::new(AtomicBool::new(false)), + started: AtomicBool::new(false), + } + } + + pub(super) fn start(&self) { + if self + .started + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return; + } + + let mut receiver = self.watcher.subscribe(); + let mut health_failures = self.watcher.subscribe_health_failures(); + let invalidator = self.invalidator.clone(); + let health_invalidator = invalidator.clone(); + let desired_paths = self.desired_paths.clone(); + let rebind_required = self.rebind_required.clone(); + let health_rebind_required = rebind_required.clone(); + tokio::spawn(async move { + loop { + let events = match receiver.recv().await { + Ok(events) => events, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { + rebind_required.store(true, Ordering::Release); + invalidator.invalidate(); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + rebind_required.store(true, Ordering::Release); + invalidator.invalidate(); + break; + } + }; + let desired = read_desired_paths(&desired_paths); + let relevant = events.iter().filter(|event| { + let event_path = Path::new(&event.path); + desired + .iter() + .any(|root| event_path.starts_with(root) || root.starts_with(event_path)) + }); + let mut invalidated = false; + for event in relevant { + invalidated = true; + if matches!( + &event.kind, + FileWatchEventKind::Remove + | FileWatchEventKind::Modify + | FileWatchEventKind::Rename { .. } + | FileWatchEventKind::Other + ) { + rebind_required.store(true, Ordering::Release); + } + } + if invalidated { + invalidator.invalidate(); + } + } + }); + tokio::spawn(async move { + loop { + match health_failures.recv().await { + Ok(()) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { + health_rebind_required.store(true, Ordering::Release); + health_invalidator.invalidate(); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + }); + } + + /// Replaces the desired source set. Returns false when any desired root + /// cannot be covered by an OS watcher; callers then keep discovery uncached. + pub(super) async fn sync_roots(&self, roots: Vec) -> bool { + let backend_unhealthy = !self.watcher.is_healthy(); + let force_rebind = self.rebind_required.swap(false, Ordering::AcqRel) && !backend_unhealthy; + let desired = merge_roots(roots); + let desired_paths = desired.keys().cloned().collect::>(); + let desired_changed = { + let mut current = write_desired_paths(&self.desired_paths); + if *current == desired_paths { + false + } else { + *current = desired_paths; + true + } + }; + + let mut registrations = BTreeMap::new(); + let mut healthy = true; + for (path, recursive) in desired { + let Some((watch_path, watch_recursively)) = registration_for(&path, recursive) else { + healthy = false; + continue; + }; + registrations + .entry(watch_path) + .and_modify(|registered_recursive| *registered_recursive |= watch_recursively) + .or_insert(watch_recursively); + } + let registrations = remove_recursively_covered_roots(registrations); + + let mut states = self.registrations.lock().await; + let obsolete = states + .keys() + .filter(|key| !registrations.contains_key(*key)) + .cloned() + .collect::>(); + let mut registration_changed = !obsolete.is_empty(); + for key in obsolete { + if let Err(error) = self.watcher.unwatch_path(&key.to_string_lossy()).await { + log::warn!( + "Failed to remove obsolete Skill watch root {}: {}", + key.display(), + error + ); + } + states.remove(&key); + } + + for (path, recursive) in registrations { + let previous = states.get(&path).copied(); + if !force_rebind && previous == Some((recursive, true)) && path.exists() { + continue; + } + registration_changed = true; + if !path.exists() { + states.insert(path, (recursive, false)); + healthy = false; + continue; + } + if previous.is_some() { + let _ = self.watcher.unwatch_path(&path.to_string_lossy()).await; + } + let mut config = FileWatcherConfig::default(); + config.watch_recursively = recursive; + config.ignore_hidden_files = false; + config.ignore_common_build_directories = false; + config.debounce_interval_ms = SKILL_WATCH_DEBOUNCE_MS; + match self + .watcher + .watch_path(&path.to_string_lossy(), Some(config)) + .await + { + Ok(()) => { + states.insert(path, (recursive, true)); + } + Err(error) => { + states.insert(path.clone(), (recursive, false)); + healthy = false; + log::warn!( + "Failed to watch Skill source root {}: {}", + path.display(), + error + ); + } + } + } + drop(states); + + if backend_unhealthy { + match self.watcher.rebuild_watcher().await { + Ok(()) => registration_changed = true, + Err(error) => { + healthy = false; + log::warn!("Failed to rebuild the Skill file watcher: {}", error); + } + } + } + + if desired_changed || registration_changed { + self.invalidator.invalidate(); + } + healthy && self.watcher.is_healthy() + } +} + +fn merge_roots(roots: Vec) -> BTreeMap { + let mut merged = BTreeMap::new(); + for root in roots { + merged + .entry(root.path) + .and_modify(|recursive| *recursive |= root.recursive) + .or_insert(root.recursive); + } + merged +} + +fn remove_recursively_covered_roots( + registrations: BTreeMap, +) -> BTreeMap { + let mut ordered = registrations.into_iter().collect::>(); + ordered.sort_by(|(left, _), (right, _)| { + left.components() + .count() + .cmp(&right.components().count()) + .then_with(|| left.cmp(right)) + }); + let mut minimal = BTreeMap::new(); + for (path, recursive) in ordered { + let covered = minimal.iter().any(|(ancestor, ancestor_recursive)| { + *ancestor_recursive && path.starts_with(ancestor) + }); + if !covered { + minimal.insert(path, recursive); + } + } + minimal +} + +fn registration_for(path: &Path, recursive: bool) -> Option<(PathBuf, bool)> { + if path.exists() { + return Some((path.to_path_buf(), recursive)); + } + let mut parent = path.to_path_buf(); + while parent.pop() { + if parent.exists() { + return Some((parent, false)); + } + } + None +} + +fn read_desired_paths( + lock: &StdRwLock>, +) -> std::sync::RwLockReadGuard<'_, Vec> { + lock.read().unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn write_desired_paths( + lock: &StdRwLock>, +) -> std::sync::RwLockWriteGuard<'_, Vec> { + lock.write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +#[cfg(test)] +mod tests { + use super::{ + remove_recursively_covered_roots, LocalSkillWatchMonitor, LocalSkillWatchRoot, + VersionedSnapshotCache, MAX_SNAPSHOT_LOAD_ATTEMPTS, + }; + use std::collections::BTreeMap; + use std::fs; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + use std::time::Duration; + + #[test] + fn recursive_parent_registration_covers_nested_skill_roots() { + let parent = std::path::PathBuf::from("skills"); + let child = parent.join(".system"); + let sibling = std::path::PathBuf::from("other-skills"); + let registrations = BTreeMap::from([ + (child, true), + (sibling.clone(), true), + (parent.clone(), true), + ]); + + let minimal = remove_recursively_covered_roots(registrations); + + assert_eq!(minimal, BTreeMap::from([(parent, true), (sibling, true)])); + } + + #[tokio::test] + async fn versioned_cache_reuses_a_stable_snapshot_and_reloads_after_invalidation() { + let cache = VersionedSnapshotCache::new(); + let loads = AtomicUsize::new(0); + + let first = cache + .get_or_load(|| async { + loads.fetch_add(1, Ordering::SeqCst); + ("first".to_string(), true) + }) + .await; + let reused = cache + .get_or_load(|| async { + loads.fetch_add(1, Ordering::SeqCst); + ("unexpected".to_string(), true) + }) + .await; + cache.invalidate(); + let refreshed = cache + .get_or_load(|| async { + loads.fetch_add(1, Ordering::SeqCst); + ("second".to_string(), true) + }) + .await; + + assert_eq!(first, "first"); + assert_eq!(reused, "first"); + assert_eq!(refreshed, "second"); + assert_eq!(loads.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn invalidation_during_load_cannot_publish_the_stale_snapshot() { + let cache = Arc::new(VersionedSnapshotCache::new()); + let loads = Arc::new(AtomicUsize::new(0)); + let invalidator = cache.clone(); + let load_count = loads.clone(); + + let value = cache + .get_or_load(move || { + let invalidator = invalidator.clone(); + let load_count = load_count.clone(); + async move { + let attempt = load_count.fetch_add(1, Ordering::SeqCst); + if attempt == 0 { + invalidator.invalidate(); + } + (attempt + 1, true) + } + }) + .await; + + assert_eq!(value, 2); + assert_eq!(loads.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn continuous_invalidations_return_uncached_instead_of_starving_discovery() { + let cache = VersionedSnapshotCache::new(); + let loads = AtomicUsize::new(0); + + let value = cache + .get_or_load(|| async { + let value = loads.fetch_add(1, Ordering::SeqCst) + 1; + cache.invalidate(); + (value, true) + }) + .await; + + assert_eq!(value, MAX_SNAPSHOT_LOAD_ATTEMPTS); + assert_eq!(loads.load(Ordering::SeqCst), MAX_SNAPSHOT_LOAD_ATTEMPTS); + let next = cache + .get_or_load(|| async { + let value = loads.fetch_add(1, Ordering::SeqCst) + 1; + (value, true) + }) + .await; + assert_eq!(next, MAX_SNAPSHOT_LOAD_ATTEMPTS + 1); + } + + #[tokio::test] + async fn an_unhealthy_monitor_keeps_discovery_uncached() { + let cache = VersionedSnapshotCache::new(); + let loads = AtomicUsize::new(0); + + for expected in [1, 2] { + let value = cache + .get_or_load(|| async { + let value = loads.fetch_add(1, Ordering::SeqCst) + 1; + (value, false) + }) + .await; + assert_eq!(value, expected); + } + + assert_eq!(loads.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn watched_semantic_source_changes_invalidate_the_cached_snapshot() { + let temp = tempfile::tempdir().expect("tempdir"); + let build_skill = temp.path().join("build"); + fs::create_dir_all(&build_skill).expect("build-named skill directory"); + let cache = VersionedSnapshotCache::new(); + let monitor = LocalSkillWatchMonitor::new(cache.invalidator()); + monitor.start(); + let roots = vec![LocalSkillWatchRoot::recursive(temp.path().to_path_buf())]; + let loads = AtomicUsize::new(0); + + let first = cache + .get_or_load(|| async { + let value = loads.fetch_add(1, Ordering::SeqCst) + 1; + let cacheable = monitor.sync_roots(roots.clone()).await; + (value, cacheable) + }) + .await; + + let skill_file = build_skill.join("SKILL.md"); + fs::write( + &skill_file, + "---\nname: build\ndescription: Build workflow\n---\n", + ) + .expect("write skill file"); + + let refreshed = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let value = cache + .get_or_load(|| async { + let value = loads.fetch_add(1, Ordering::SeqCst) + 1; + let cacheable = monitor.sync_roots(roots.clone()).await; + (value, cacheable) + }) + .await; + if value > first { + break value; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .expect("watched Skill change should invalidate the cache"); + + assert!(refreshed > first); + } + + #[tokio::test] + async fn replacing_a_watched_directory_rebinds_before_caching_again() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = temp.path().join("skills"); + fs::create_dir_all(&root).expect("skill root"); + let skill_file = root.join("SKILL.md"); + fs::write(&skill_file, "first").expect("first snapshot"); + let cache = VersionedSnapshotCache::new(); + let monitor = LocalSkillWatchMonitor::new(cache.invalidator()); + monitor.start(); + let roots = vec![LocalSkillWatchRoot::recursive(root.clone())]; + + let first = cache + .get_or_load(|| async { + let value = fs::read_to_string(&skill_file).unwrap_or_default(); + let cacheable = monitor.sync_roots(roots.clone()).await; + (value, cacheable) + }) + .await; + assert_eq!(first, "first"); + + fs::remove_dir_all(&root).expect("remove watched root"); + fs::create_dir_all(&root).expect("replace watched root"); + fs::write(&skill_file, "second").expect("replacement snapshot"); + let second = wait_for_snapshot(&cache, &monitor, &roots, &skill_file, "second").await; + assert_eq!(second, "second"); + + fs::write(&skill_file, "third").expect("modify replacement"); + let third = wait_for_snapshot(&cache, &monitor, &roots, &skill_file, "third").await; + assert_eq!(third, "third"); + } + + async fn wait_for_snapshot( + cache: &VersionedSnapshotCache, + monitor: &LocalSkillWatchMonitor, + roots: &[LocalSkillWatchRoot], + skill_file: &std::path::Path, + expected: &str, + ) -> String { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let value = cache + .get_or_load(|| async { + let value = fs::read_to_string(skill_file).unwrap_or_default(); + let cacheable = monitor.sync_roots(roots.to_vec()).await; + (value, cacheable) + }) + .await; + if value == expected { + break value; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .expect("watched directory update should invalidate the cache") + } +} diff --git a/src/crates/services/services-integrations/src/file_watch/service.rs b/src/crates/services/services-integrations/src/file_watch/service.rs index 20a5d79ac1..11bd8ee6b0 100644 --- a/src/crates/services/services-integrations/src/file_watch/service.rs +++ b/src/crates/services/services-integrations/src/file_watch/service.rs @@ -3,23 +3,91 @@ use log::{debug, error}; use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; use std::collections::hash_map::Entry; use std::collections::HashMap; +use std::future::Future; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex as StdMutex, RwLock as StdRwLock}; -use tokio::sync::{broadcast, Mutex}; +use tokio::sync::{broadcast, mpsc, Mutex}; use super::types::{FileWatchEvent, FileWatchEventKind, FileWatcherConfig}; -type WatchedPaths = StdRwLock>; +#[derive(Clone)] +struct WatchedPath { + config: FileWatcherConfig, + /// Native backends may report canonical paths even when callers register a + /// logical alias (notably `/private/var` for `/var` on macOS). + backend_root: PathBuf, +} + +impl WatchedPath { + fn new(path: &Path, config: FileWatcherConfig) -> Self { + Self { + config, + backend_root: std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()), + } + } +} + +type WatchedPaths = StdRwLock>; +type EmitterDispatch = StdRwLock>>>; + +const EVENT_CHANNEL_CAPACITY: usize = 64; + +struct FileWatchHealth { + healthy: AtomicBool, + sender: broadcast::Sender<()>, +} + +impl FileWatchHealth { + fn new() -> Self { + let (sender, _) = broadcast::channel(8); + Self { + healthy: AtomicBool::new(true), + sender, + } + } + + fn is_healthy(&self) -> bool { + self.healthy.load(Ordering::Acquire) + } + + fn subscribe(&self) -> broadcast::Receiver<()> { + self.sender.subscribe() + } + + fn mark_healthy(&self) { + self.healthy.store(true, Ordering::Release); + } + + fn mark_unhealthy(&self) { + if self.healthy.swap(false, Ordering::AcqRel) { + let _ = self.sender.send(()); + } + } +} fn read_watched_paths( lock: &WatchedPaths, -) -> std::sync::RwLockReadGuard<'_, HashMap> { +) -> std::sync::RwLockReadGuard<'_, HashMap> { lock.read().unwrap_or_else(|poisoned| poisoned.into_inner()) } fn write_watched_paths( lock: &WatchedPaths, -) -> std::sync::RwLockWriteGuard<'_, HashMap> { +) -> std::sync::RwLockWriteGuard<'_, HashMap> { + lock.write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn read_emitter_dispatch( + lock: &EmitterDispatch, +) -> std::sync::RwLockReadGuard<'_, Option>>> { + lock.read().unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn write_emitter_dispatch( + lock: &EmitterDispatch, +) -> std::sync::RwLockWriteGuard<'_, Option>>> { lock.write() .unwrap_or_else(|poisoned| poisoned.into_inner()) } @@ -37,14 +105,17 @@ impl From<&EventKind> for FileWatchEventKind { } pub struct FileWatchService { - emitter: Arc>>>, + emitter_dispatch: Arc, watcher: Arc>>, + registration_gate: Mutex<()>, /// Path table shared with the watcher event thread. Uses a synchronous /// RwLock so per-event filtering never needs `block_on` into the async /// runtime; guards are only held for short, await-free sections. watched_paths: Arc, event_buffer: Arc>>, event_sender: broadcast::Sender>, + health: Arc, + active_watcher_instance: Arc, config: FileWatcherConfig, } @@ -62,13 +133,16 @@ fn lock_event_buffer( impl FileWatchService { pub fn new(config: FileWatcherConfig) -> Self { - let (event_sender, _) = broadcast::channel(64); + let (event_sender, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); Self { - emitter: Arc::new(Mutex::new(None)), + emitter_dispatch: Arc::new(StdRwLock::new(None)), watcher: Arc::new(Mutex::new(None)), + registration_gate: Mutex::new(()), watched_paths: Arc::new(StdRwLock::new(HashMap::new())), event_buffer: Arc::new(StdMutex::new(Vec::new())), event_sender, + health: Arc::new(FileWatchHealth::new()), + active_watcher_instance: Arc::new(AtomicU64::new(0)), config, } } @@ -78,9 +152,36 @@ impl FileWatchService { self.event_sender.subscribe() } + /// Reports failures from the native watcher thread separately from file + /// events so cache owners can stop publishing snapshots until re-registration + /// succeeds. + pub fn subscribe_health_failures(&self) -> broadcast::Receiver<()> { + self.health.subscribe() + } + + pub fn is_healthy(&self) -> bool { + self.health.is_healthy() + } + pub async fn set_emitter(&self, emitter: Arc) { - let mut e = self.emitter.lock().await; - *e = Some(emitter); + self.set_emitter_dispatch(move |events| { + let emitter = emitter.clone(); + async move { Self::emit_events(&emitter, &events).await } + }); + } + + fn set_emitter_dispatch(&self, mut emit: F) + where + F: FnMut(Vec) -> Fut + Send + 'static, + Fut: Future + Send + 'static, + { + let (sender, mut receiver) = mpsc::channel::>(EVENT_CHANNEL_CAPACITY); + *write_emitter_dispatch(&self.emitter_dispatch) = Some(sender); + tokio::spawn(async move { + while let Some(events) = receiver.recv().await { + emit(events).await; + } + }); } pub async fn watch_path( @@ -88,34 +189,46 @@ impl FileWatchService { path: &str, config: Option, ) -> Result<(), String> { + let _registration_guard = self.registration_gate.lock().await; let path_buf = PathBuf::from(path); if !path_buf.exists() { return Err("Path does not exist".to_string()); } + let registration = + WatchedPath::new(&path_buf, config.unwrap_or_else(|| self.config.clone())); let (recursive, is_new) = { let mut watched_paths = write_watched_paths(&self.watched_paths); - let config = config.unwrap_or_else(|| self.config.clone()); + let config = registration.config; + let backend_root = registration.backend_root; match watched_paths.entry(path_buf.clone()) { Entry::Occupied(mut entry) => { // Multiple product services may share a root. Registering a // narrower observer must not silently downgrade an existing // recursive or hidden-file-aware watch. let existing = entry.get_mut(); - existing.watch_recursively |= config.watch_recursively; - existing.ignore_hidden_files &= config.ignore_hidden_files; - existing.debounce_interval_ms = existing + existing.config.watch_recursively |= config.watch_recursively; + existing.config.ignore_hidden_files &= config.ignore_hidden_files; + existing.config.ignore_common_build_directories &= + config.ignore_common_build_directories; + existing.config.debounce_interval_ms = existing + .config .debounce_interval_ms .min(config.debounce_interval_ms); - existing.max_events_per_interval = existing + existing.config.max_events_per_interval = existing + .config .max_events_per_interval .max(config.max_events_per_interval); - (existing.watch_recursively, false) + existing.backend_root = backend_root; + (existing.config.watch_recursively, false) } Entry::Vacant(entry) => { let recursive = config.watch_recursively; - entry.insert(config); + entry.insert(WatchedPath { + config, + backend_root, + }); (recursive, true) } } @@ -133,9 +246,9 @@ impl FileWatchService { }; // A repeat registration must always re-register with the OS // watcher, not just when the recursion mode changed: the path - // may have been absent when the watcher was built (missing - // paths are skipped), or removed and recreated since, leaving a - // stale/absent OS-level watch. `ensure_watch_roots` relies on + // may have failed a prior registration, or been removed and + // recreated since, leaving a stale/absent OS-level watch. + // `ensure_watch_roots` relies on // exactly this to resume watching a root that reappeared, and // the pre-incremental implementation got it for free by // rebuilding the whole watcher on every call. @@ -143,9 +256,14 @@ impl FileWatchService { // May never have been registered; ignore unwatch failures. let _ = watcher.unwatch(&path_buf); } - watcher - .watch(&path_buf, mode) - .map_err(|e| format!("Failed to watch path {}: {}", path_buf.display(), e))?; + if let Err(error) = watcher.watch(&path_buf, mode) { + self.health.mark_unhealthy(); + return Err(format!( + "Failed to watch path {}: {}", + path_buf.display(), + error + )); + } Ok(()) } None => { @@ -156,6 +274,7 @@ impl FileWatchService { } pub async fn unwatch_path(&self, path: &str) -> Result<(), String> { + let _registration_guard = self.registration_gate.lock().await; let path_buf = PathBuf::from(path); let (removed, is_empty) = { @@ -168,7 +287,9 @@ impl FileWatchService { if is_empty { // Dropping the watcher disconnects its channel; the event thread // exits on its own. + self.active_watcher_instance.fetch_add(1, Ordering::AcqRel); *watcher_guard = None; + self.health.mark_healthy(); } else if removed { if let Some(watcher) = watcher_guard.as_mut() { // The path may never have been registered (e.g. it was missing @@ -180,90 +301,127 @@ impl FileWatchService { Ok(()) } + /// Recreates the native watcher and re-registers the complete current path + /// table. A backend failure is only considered recovered after this succeeds. + pub async fn rebuild_watcher(&self) -> Result<(), String> { + let _registration_guard = self.registration_gate.lock().await; + self.create_watcher().await + } + async fn create_watcher(&self) -> Result<(), String> { // Keep the sync read guard scopes free of awaits: check emptiness with // a short-lived guard before touching the async watcher mutex. if read_watched_paths(&self.watched_paths).is_empty() { let mut watcher_guard = self.watcher.lock().await; + self.active_watcher_instance.fetch_add(1, Ordering::AcqRel); *watcher_guard = None; + self.health.mark_healthy(); return Ok(()); } let (tx, rx) = std::sync::mpsc::channel(); - let mut watcher = RecommendedWatcher::new(tx, Config::default()) - .map_err(|e| format!("Failed to create watcher: {}", e))?; - - { - let watched_paths = read_watched_paths(&self.watched_paths); - - for (path, config) in watched_paths.iter() { - // A watched source directory may be removed between events. Its - // stable parent watch remains active and the missing path will be - // re-registered when it reappears. - if !path.exists() { - continue; - } - let mode = if config.watch_recursively { - RecursiveMode::Recursive - } else { - RecursiveMode::NonRecursive - }; + let mut watcher = match RecommendedWatcher::new(tx, Config::default()) { + Ok(watcher) => watcher, + Err(error) => { + self.health.mark_unhealthy(); + return Err(format!("Failed to create watcher: {}", error)); + } + }; - watcher - .watch(path, mode) - .map_err(|e| format!("Failed to watch path {}: {}", path.display(), e))?; + for (path, config) in refresh_watch_registrations(&self.watched_paths) { + let mode = if config.watch_recursively { + RecursiveMode::Recursive + } else { + RecursiveMode::NonRecursive + }; + + if let Err(error) = watcher.watch(&path, mode) { + self.health.mark_unhealthy(); + return Err(format!( + "Failed to watch path {}: {}", + path.display(), + error + )); } } + let watcher_instance; { let mut watcher_guard = self.watcher.lock().await; + watcher_instance = self + .active_watcher_instance + .fetch_add(1, Ordering::AcqRel) + .saturating_add(1); *watcher_guard = Some(watcher); } + self.health.mark_healthy(); let event_buffer = self.event_buffer.clone(); - let emitter_arc = self.emitter.clone(); + let emitter_dispatch = self.emitter_dispatch.clone(); let default_debounce_ms = self.config.debounce_interval_ms; let watched_paths = self.watched_paths.clone(); let event_sender = self.event_sender.clone(); - - tokio::task::spawn_blocking(move || { - let rt = tokio::runtime::Handle::current(); - let poll = std::time::Duration::from_millis(50); - let mut last_event_time: Option = None; - - loop { - match rx.recv_timeout(poll) { - Ok(Ok(event)) => { - // Synchronous path-table snapshot: no block_on needed - // per event. - let file_events = Self::convert_events(&event, &watched_paths); - if !file_events.is_empty() { - lock_event_buffer(&event_buffer).extend(file_events); - last_event_time = Some(std::time::Instant::now()); + let health = self.health.clone(); + let active_watcher_instance = self.active_watcher_instance.clone(); + // The native receiver lives for as long as its OS watcher. Keeping that + // process-lifetime loop in Tokio's blocking pool would make short-lived + // runtimes wait forever during shutdown. Emitter delivery is handed to + // the runtime-owned dispatch queue configured by `set_emitter`. + let worker = std::thread::Builder::new() + .name("bitfun-file-watch".to_string()) + .spawn(move || { + let poll = std::time::Duration::from_millis(50); + let mut last_event_time: Option = None; + + loop { + match rx.recv_timeout(poll) { + Ok(Ok(event)) => { + // Synchronous path-table snapshot: no block_on needed + // per event. + let file_events = Self::convert_events(&event, &watched_paths); + if !file_events.is_empty() { + lock_event_buffer(&event_buffer).extend(file_events); + last_event_time = Some(std::time::Instant::now()); + } + } + Ok(Err(error)) => { + error!("File watch error: {}", error); + if active_watcher_instance.load(Ordering::Acquire) == watcher_instance { + health.mark_unhealthy(); + } + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + if active_watcher_instance.load(Ordering::Acquire) == watcher_instance { + health.mark_unhealthy(); + } + break; } } - Ok(Err(e)) => error!("File watch error: {}", e), - Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} - Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, - } - if let Some(t) = last_event_time { - let debounce_ms = read_watched_paths(&watched_paths) - .values() - .map(|config| config.debounce_interval_ms) - .min() - .unwrap_or(default_debounce_ms); - if t.elapsed() >= std::time::Duration::from_millis(debounce_ms) { - rt.block_on(Self::flush_events_static( - &event_buffer, - &emitter_arc, - &event_sender, - )); - last_event_time = None; + if let Some(t) = last_event_time { + let debounce_ms = read_watched_paths(&watched_paths) + .values() + .map(|registration| registration.config.debounce_interval_ms) + .min() + .unwrap_or(default_debounce_ms); + if t.elapsed() >= std::time::Duration::from_millis(debounce_ms) { + Self::flush_events_static( + &event_buffer, + &emitter_dispatch, + &event_sender, + ); + last_event_time = None; + } } } - } - }); + }); + if let Err(error) = worker { + self.active_watcher_instance.fetch_add(1, Ordering::AcqRel); + *self.watcher.lock().await = None; + self.health.mark_unhealthy(); + return Err(format!("Failed to start file watch worker: {error}")); + } Ok(()) } @@ -274,22 +432,22 @@ impl FileWatchService { .paths .iter() .filter_map(|event_path| { - let config = paths - .iter() - .filter(|(watch_path, _)| event_path.starts_with(watch_path)) - .max_by_key(|(watch_path, _)| watch_path.components().count()) - .map(|(_, config)| config)?; - if Self::is_in_excluded_directory(event_path) - || Self::is_temporary_file(event_path) - || config.ignore_hidden_files - && event_path + let (registered_root, registration) = + preferred_event_registration(event_path, &paths)?; + let projected_path = + project_event_path(registered_root, ®istration.backend_root, event_path); + if registration.config.ignore_common_build_directories + && Self::is_in_excluded_directory(&projected_path) + || Self::is_temporary_file(&projected_path) + || registration.config.ignore_hidden_files + && projected_path .file_name() .and_then(|name| name.to_str()) .is_some_and(|name| name.starts_with('.')) { return None; } - Self::convert_event(&event.kind, event_path) + Self::convert_event(&event.kind, &projected_path) }) .collect() } @@ -380,9 +538,9 @@ impl FileWatchService { }) } - async fn flush_events_static( + fn flush_events_static( event_buffer: &Arc>>, - emitter_arc: &Arc>>>, + emitter_dispatch: &Arc, event_sender: &broadcast::Sender>, ) { let events = { @@ -397,45 +555,55 @@ impl FileWatchService { // may still consume this batch. let _ = event_sender.send(events.clone()); - let emitter_guard = emitter_arc.lock().await; - if let Some(emitter) = emitter_guard.as_ref() { - let mut event_array = Vec::new(); - - for event in &events { - let kind = match event.kind { - FileWatchEventKind::Create => "create", - FileWatchEventKind::Modify => "modify", - FileWatchEventKind::Remove => "remove", - FileWatchEventKind::Rename { ref from, ref to } => { - event_array.push(serde_json::json!({ - "path": to, - "kind": "rename", - "from": from, - "to": to, - "timestamp": event.timestamp - })); - continue; - } - FileWatchEventKind::Other => "other", - }; + match read_emitter_dispatch(emitter_dispatch).as_ref() { + Some(dispatch) => match dispatch.try_send(events) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + debug!("EventEmitter queue full, skipping file watch events"); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + debug!("EventEmitter runtime unavailable, skipping file watch events"); + } + }, + None => debug!("EventEmitter not configured, skipping file watch events"), + } + } - event_array.push(serde_json::json!({ - "path": event.path, - "kind": kind, - "timestamp": event.timestamp - })); - } + async fn emit_events(emitter: &Arc, events: &[FileWatchEvent]) { + let mut event_array = Vec::new(); + + for event in events { + let kind = match event.kind { + FileWatchEventKind::Create => "create", + FileWatchEventKind::Modify => "modify", + FileWatchEventKind::Remove => "remove", + FileWatchEventKind::Rename { ref from, ref to } => { + event_array.push(serde_json::json!({ + "path": to, + "kind": "rename", + "from": from, + "to": to, + "timestamp": event.timestamp + })); + continue; + } + FileWatchEventKind::Other => "other", + }; + + event_array.push(serde_json::json!({ + "path": event.path, + "kind": kind, + "timestamp": event.timestamp + })); + } - if let Err(e) = emitter - .emit("file-system-changed", serde_json::json!(event_array)) - .await - { - error!("Failed to emit file-system-changed events: {}", e); - } else { - debug!("Emitted {} file system change events", event_array.len()); - } + if let Err(error) = emitter + .emit("file-system-changed", serde_json::json!(event_array)) + .await + { + error!("Failed to emit file-system-changed events: {}", error); } else { - debug!("EventEmitter not configured, skipping file watch events"); + debug!("Emitted {} file system change events", event_array.len()); } } @@ -448,6 +616,320 @@ impl FileWatchService { } } +fn refresh_watch_registrations(watched_paths: &WatchedPaths) -> Vec<(PathBuf, FileWatcherConfig)> { + let roots = read_watched_paths(watched_paths) + .keys() + .cloned() + .collect::>(); + let refreshed_roots = roots + .into_iter() + .map(|root| { + let backend_root = std::fs::canonicalize(&root).unwrap_or_else(|_| root.clone()); + (root, backend_root) + }) + .collect::>(); + + let mut paths = write_watched_paths(watched_paths); + for (root, registration) in paths.iter_mut() { + if let Some(backend_root) = refreshed_roots.get(root) { + registration.backend_root.clone_from(backend_root); + } + } + let mut registrations = paths + .iter() + .map(|(root, registration)| (root.clone(), registration.config.clone())) + .collect::>(); + registrations.sort_by(|(left, _), (right, _)| left.cmp(right)); + registrations +} + +fn preferred_event_registration<'a>( + event_path: &Path, + paths: &'a HashMap, +) -> Option<(&'a PathBuf, &'a WatchedPath)> { + paths + .iter() + .filter_map(|(registered_root, registration)| { + let registered_depth = event_path + .starts_with(registered_root) + .then(|| registered_root.components().count()); + let backend_depth = event_path + .starts_with(®istration.backend_root) + .then(|| registration.backend_root.components().count()); + let matched_depth = registered_depth.into_iter().chain(backend_depth).max()?; + let direct_at_matched_depth = registered_depth == Some(matched_depth); + Some(( + registered_root, + registration, + matched_depth, + direct_at_matched_depth, + )) + }) + .max_by(|left, right| { + left.2 + .cmp(&right.2) + .then_with(|| left.3.cmp(&right.3)) + // Keep canonical aliases deterministic when neither registration + // directly matches the backend event namespace. + .then_with(|| right.0.cmp(left.0)) + }) + .map(|(registered_root, registration, _, _)| (registered_root, registration)) +} + +fn project_event_path(registered_root: &Path, backend_root: &Path, event_path: &Path) -> PathBuf { + event_path + .strip_prefix(backend_root) + .map(|relative| registered_root.join(relative)) + .unwrap_or_else(|_| event_path.to_path_buf()) +} + +#[cfg(test)] +mod tests { + use super::{ + preferred_event_registration, read_watched_paths, refresh_watch_registrations, + write_watched_paths, FileWatchService, WatchedPath, + }; + use crate::file_watch::FileWatcherConfig; + use notify::event::CreateKind; + use notify::{Event, EventKind}; + use std::collections::HashMap; + use std::fs; + use std::path::Path; + use std::sync::RwLock as StdRwLock; + use std::time::Duration; + + #[test] + fn backend_canonical_paths_are_projected_to_the_registered_namespace() { + let registered_root = Path::new("/var/folders/bitfun-skills").to_path_buf(); + let backend_root = Path::new("/private/var/folders/bitfun-skills").to_path_buf(); + let backend_event = backend_root.join("example/SKILL.md"); + let watched_paths = StdRwLock::new(HashMap::from([( + registered_root.clone(), + WatchedPath { + config: FileWatcherConfig::default(), + backend_root, + }, + )])); + let event = Event::new(EventKind::Create(CreateKind::File)).add_path(backend_event); + + let converted = FileWatchService::convert_events(&event, &watched_paths); + + assert_eq!(converted.len(), 1); + assert_eq!( + converted[0].path, + registered_root.join("example/SKILL.md").to_string_lossy() + ); + } + + #[test] + fn a_direct_registered_path_wins_a_canonical_alias_at_equal_depth() { + let alias_root = Path::new("/var/folders/bitfun-skills").to_path_buf(); + let direct_root = Path::new("/private/var/folders/bitfun-skills").to_path_buf(); + let watched_paths = HashMap::from([ + ( + alias_root, + WatchedPath { + config: FileWatcherConfig::default(), + backend_root: direct_root.clone(), + }, + ), + ( + direct_root.clone(), + WatchedPath { + config: FileWatcherConfig::default(), + backend_root: direct_root.clone(), + }, + ), + ]); + + let (registered_root, _) = + preferred_event_registration(&direct_root.join("example/SKILL.md"), &watched_paths) + .expect("matching registration"); + + assert_eq!(registered_root, &direct_root); + } + + #[test] + fn canonical_alias_fallback_is_deterministic() { + let first_alias = Path::new("/aliases/first/bitfun-skills").to_path_buf(); + let second_alias = Path::new("/aliases/second/bitfun-skills").to_path_buf(); + let backend_root = Path::new("/private/var/folders/bitfun-skills").to_path_buf(); + let watched_paths = HashMap::from([ + ( + second_alias, + WatchedPath { + config: FileWatcherConfig::default(), + backend_root: backend_root.clone(), + }, + ), + ( + first_alias.clone(), + WatchedPath { + config: FileWatcherConfig::default(), + backend_root: backend_root.clone(), + }, + ), + ]); + + let (registered_root, _) = + preferred_event_registration(&backend_root.join("example/SKILL.md"), &watched_paths) + .expect("matching registration"); + + assert_eq!(registered_root, &first_alias); + } + + #[test] + fn watcher_rebuild_refreshes_the_native_backend_root() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = temp.path().join("skills"); + fs::create_dir_all(&root).expect("skill root"); + let expected = std::fs::canonicalize(&root).expect("canonical skill root"); + let watched_paths = StdRwLock::new(HashMap::from([( + root.clone(), + WatchedPath { + config: FileWatcherConfig::default(), + backend_root: Path::new("/stale/backend/root").to_path_buf(), + }, + )])); + + let registrations = refresh_watch_registrations(&watched_paths); + + assert_eq!(registrations.len(), 1); + assert_eq!( + read_watched_paths(&watched_paths)[&root].backend_root, + expected + ); + } + + #[tokio::test] + async fn backend_health_recovers_only_after_a_full_rebuild() { + let temp = tempfile::tempdir().expect("tempdir"); + let first = temp.path().join("first"); + let second = temp.path().join("second"); + fs::create_dir_all(&first).expect("first root"); + fs::create_dir_all(&second).expect("second root"); + let mut config = FileWatcherConfig::default(); + config.debounce_interval_ms = 0; + let service = FileWatchService::new(config.clone()); + let mut events = service.subscribe(); + let mut failures = service.subscribe_health_failures(); + service + .watch_path(&first.to_string_lossy(), Some(config.clone())) + .await + .expect("initial watch"); + + service.health.mark_unhealthy(); + + failures.recv().await.expect("health failure notification"); + service + .watch_path(&second.to_string_lossy(), Some(config)) + .await + .expect("incremental registration"); + assert!(!service.is_healthy()); + + service.rebuild_watcher().await.expect("full rebuild"); + assert!(service.is_healthy()); + fs::write(second.join("SKILL.md"), "updated").expect("watched update"); + + let batch = tokio::time::timeout(Duration::from_secs(5), events.recv()) + .await + .expect("event after rebuild") + .expect("watch event batch"); + assert!(batch.iter().any(|event| event.path.ends_with("SKILL.md"))); + } + + #[tokio::test] + async fn rebuild_does_not_report_a_missing_registered_path_as_healthy() { + let temp = tempfile::tempdir().expect("tempdir"); + let missing = temp.path().join("missing"); + let service = FileWatchService::new(FileWatcherConfig::default()); + write_watched_paths(&service.watched_paths).insert( + missing.clone(), + WatchedPath::new(&missing, FileWatcherConfig::default()), + ); + + let error = service + .rebuild_watcher() + .await + .expect_err("missing registered path must fail rebuild"); + + assert!(error.contains("Failed to watch path")); + assert!(!service.is_healthy()); + } + + #[test] + fn emitter_dispatch_can_rebind_after_runtime_shutdown() { + let temp = tempfile::tempdir().expect("tempdir"); + let mut config = FileWatcherConfig::default(); + config.debounce_interval_ms = 0; + config.ignore_hidden_files = false; + let service = FileWatchService::new(config.clone()); + + let first_runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("first runtime"); + first_runtime.block_on(async { + service.set_emitter_dispatch(|_| async { + tokio::time::sleep(Duration::from_millis(1)).await; + }); + service + .watch_path(temp.path().to_str().unwrap(), Some(config)) + .await + .expect("watch temp directory"); + }); + drop(first_runtime); + + let second_runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("second runtime"); + second_runtime.block_on(async { + let (observed, mut emitted) = tokio::sync::mpsc::unbounded_channel(); + service.set_emitter_dispatch(move |events| { + let observed = observed.clone(); + async move { + tokio::time::sleep(Duration::from_millis(1)).await; + let _ = observed.send(events); + } + }); + + let first_file = temp.path().join("first.md"); + fs::write(&first_file, "first").expect("first watched update"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + let events = tokio::time::timeout_at(deadline, emitted.recv()) + .await + .expect("dispatch after runtime rebind") + .expect("emitter dispatch remains open"); + if events + .iter() + .any(|event| event.path == first_file.to_string_lossy()) + { + break; + } + } + + tokio::time::sleep(Duration::from_millis(50)).await; + let second_file = temp.path().join("second.md"); + fs::write(&second_file, "second").expect("second watched update"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + let events = tokio::time::timeout_at(deadline, emitted.recv()) + .await + .expect("watch worker survives the first runtime shutdown") + .expect("emitter dispatch remains open"); + if events + .iter() + .any(|event| event.path == second_file.to_string_lossy()) + { + break; + } + } + }); + } +} + static GLOBAL_FILE_WATCH_SERVICE: std::sync::OnceLock> = std::sync::OnceLock::new(); diff --git a/src/crates/services/services-integrations/src/file_watch/types.rs b/src/crates/services/services-integrations/src/file_watch/types.rs index d91a79f3a2..b13e923584 100644 --- a/src/crates/services/services-integrations/src/file_watch/types.rs +++ b/src/crates/services/services-integrations/src/file_watch/types.rs @@ -21,6 +21,10 @@ pub enum FileWatchEventKind { pub struct FileWatcherConfig { pub watch_recursively: bool, pub ignore_hidden_files: bool, + /// Skip common dependency/build output directories for broad workspace watches. + /// Semantic source roots such as Skill catalogs can disable this because names + /// like `build` and `target` are valid source identifiers there. + pub ignore_common_build_directories: bool, pub debounce_interval_ms: u64, pub max_events_per_interval: usize, } @@ -30,6 +34,7 @@ impl Default for FileWatcherConfig { Self { watch_recursively: true, ignore_hidden_files: true, + ignore_common_build_directories: true, debounce_interval_ms: 500, max_events_per_interval: 100, } diff --git a/src/crates/services/services-integrations/tests/file_watch_contracts.rs b/src/crates/services/services-integrations/tests/file_watch_contracts.rs index 0c61ce5c16..e58626b0aa 100644 --- a/src/crates/services/services-integrations/tests/file_watch_contracts.rs +++ b/src/crates/services/services-integrations/tests/file_watch_contracts.rs @@ -4,6 +4,7 @@ use bitfun_services_integrations::file_watch::{ FileWatchEventKind, FileWatchService, FileWatcherConfig, }; use std::fs; +use std::sync::Arc; use std::time::Duration; #[tokio::test] @@ -28,6 +29,36 @@ fn file_watch_event_kind_serializes_snake_case() { assert_eq!(value, "modify"); } +#[test] +fn file_watch_worker_does_not_extend_tokio_runtime_lifetime() { + let temp = tempfile::tempdir().expect("tempdir"); + let service = Arc::new(FileWatchService::new(FileWatcherConfig::default())); + let worker_service = service.clone(); + let watched_path = temp.path().to_string_lossy().into_owned(); + let (finished_tx, finished_rx) = std::sync::mpsc::sync_channel(1); + + let runtime_owner = std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime"); + runtime.block_on(async move { + worker_service + .watch_path(&watched_path, None) + .await + .expect("watch temp directory"); + }); + drop(runtime); + finished_tx.send(()).expect("report runtime shutdown"); + }); + + finished_rx + .recv_timeout(Duration::from_secs(3)) + .expect("file-watch worker must not keep a short-lived runtime alive"); + drop(service); + runtime_owner.join().expect("runtime owner thread"); +} + #[tokio::test] async fn file_watch_publishes_debounced_batches_to_backend_subscribers() { let temp = tempfile::tempdir().expect("tempdir"); @@ -54,6 +85,44 @@ async fn file_watch_publishes_debounced_batches_to_backend_subscribers() { .any(|event| event.path == file.to_string_lossy())); } +#[tokio::test] +async fn file_watch_can_include_build_named_directories_for_semantic_sources() { + let temp = tempfile::tempdir().expect("tempdir"); + let build_skill = temp.path().join("build"); + fs::create_dir_all(&build_skill).expect("build-named skill directory"); + let mut config = FileWatcherConfig::default(); + config.debounce_interval_ms = 40; + config.ignore_hidden_files = false; + config.ignore_common_build_directories = false; + let service = FileWatchService::new(config.clone()); + let mut events = service.subscribe(); + service + .watch_path(temp.path().to_str().unwrap(), Some(config)) + .await + .expect("watch semantic source root"); + + let file = build_skill.join("SKILL.md"); + fs::write( + &file, + "---\nname: build\ndescription: Build workflow\n---\n", + ) + .expect("write skill file"); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + let batch = tokio::time::timeout_at(deadline, events.recv()) + .await + .expect("build-named semantic source should emit events") + .expect("watch broadcast remains open"); + if batch + .iter() + .any(|event| event.path == file.to_string_lossy()) + { + break; + } + } +} + #[tokio::test] async fn a_narrow_duplicate_registration_does_not_downgrade_recursive_watch() { let temp = tempfile::tempdir().expect("tempdir");