From a29799af2b3c767bff7028409a3d2b0ccf05c386 Mon Sep 17 00:00:00 2001 From: 370025263 <370025263@qq.com> Date: Tue, 14 Jul 2026 16:00:06 +0800 Subject: [PATCH 1/3] =?UTF-8?q?perf(skillhub):=20=E7=BC=93=E5=AD=98?= =?UTF-8?q?=E5=85=A8=E7=9B=98=E6=89=AB=E6=8F=8F=E5=B9=B6=E5=89=AA=E6=9E=9D?= =?UTF-8?q?=E9=9A=90=E8=97=8F=E7=9B=AE=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将高频 entry/fingerprint/search/index 调用收敛到实例级短 TTL 快照，避免并发 sync 重复遍历和哈希整棵 skillhub。 - 用 single-flight 快照合并并发扫描，并建立 O(1) 条目索引 - 用 os.walk 提前剪枝点目录，按 mtime_ns 与 size 复用解析和内容哈希 - 上传后强制刷新快照，标签云复用 5 秒扫描结果 - 覆盖缓存失效、删除、并发、隐藏目录和上传即时可见 --- src/xskill/dashboard/metrics.py | 30 +++- src/xskill/recommend/skillhub.py | 213 ++++++++++++++++++++------ src/xskill/team/server/api.py | 3 +- tests/test_dashboard_metrics.py | 76 +++++++++ tests/test_skill_hub_search_upload.py | 6 + tests/test_skillhub.py | 133 ++++++++++++++++ 6 files changed, 408 insertions(+), 53 deletions(-) diff --git a/src/xskill/dashboard/metrics.py b/src/xskill/dashboard/metrics.py index d4585c99..5e157247 100644 --- a/src/xskill/dashboard/metrics.py +++ b/src/xskill/dashboard/metrics.py @@ -11,7 +11,7 @@ import threading import time from pathlib import Path -from typing import Optional +from typing import Callable, Optional from xskill.pipeline.registry import pooled_connection @@ -421,7 +421,9 @@ class DashboardMetrics: def __init__(self, db_path: Optional[Path] = None, *, skill_dir: Optional[Path] = None, unknown_harness: str = "unknown", - unknown_model: str = "unknown"): + unknown_model: str = "unknown", + tag_cloud_ttl_seconds: float = 5.0, + clock: Callable[[], float] = time.monotonic): self._db = db_path # 使用/UX 类指标的事实源目录（//.ux_scores.jsonl）。 self._skill_dir = skill_dir @@ -429,6 +431,11 @@ def __init__(self, db_path: Optional[Path] = None, *, # 默认 'unknown'；看板路由按 config.dashboard.default_harness/_model 传入覆盖。 self._unknown_harness = unknown_harness self._unknown_model = unknown_model + self._tag_cloud_ttl_seconds = max(0.0, float(tag_cloud_ttl_seconds)) + self._clock = clock + self._tag_cloud_lock = threading.Lock() + self._tag_cloud_expires_at = 0.0 + self._tag_cloud_rows: list[dict] | None = None def _usage(self) -> list[dict]: return load_usage_records(self._skill_dir) @@ -548,6 +555,20 @@ def tag_cloud(self, top_n: int = 40) -> list[dict]: 本机(非 team)目录的原子计入 count 但不归属任何用户。 返回按出现次数降序的 ``[{tag, count, users}]`` 前 top_n。 """ + now = self._clock() + if self._tag_cloud_rows is None or now >= self._tag_cloud_expires_at: + with self._tag_cloud_lock: + now = self._clock() + if self._tag_cloud_rows is None or now >= self._tag_cloud_expires_at: + self._tag_cloud_rows = self._scan_tag_cloud() + self._tag_cloud_expires_at = ( + self._clock() + self._tag_cloud_ttl_seconds + ) + limit = max(0, int(top_n)) + return copy.deepcopy(self._tag_cloud_rows[:limit]) + + def _scan_tag_cloud(self) -> list[dict]: + """执行一次标签全量扫描；由 :meth:`tag_cloud` 合并并发调用。""" from collections import Counter, defaultdict from xskill.pipeline.atom import AtomTaskStore from xskill.config import get_registry_db_path @@ -572,8 +593,9 @@ def tag_cloud(self, top_n: int = 40) -> list[dict]: tag_users[t].add(client) except OSError: continue # 某个目录不可读/路径异常,跳过不阻断整体聚合 - return [{"tag": t, "count": n, "users": sorted(tag_users.get(t, ()))} - for t, n in counter.most_common(top_n)] + return [{"tag": tag, "count": count, + "users": sorted(tag_users.get(tag, ()))} + for tag, count in counter.most_common()] def canary_sides(self) -> list[dict]: """灰度分桶分布：使用打分记录按 side 聚合（与 check_and_decide 裁决同源）。 diff --git a/src/xskill/recommend/skillhub.py b/src/xskill/recommend/skillhub.py index be44a8bf..e1375778 100644 --- a/src/xskill/recommend/skillhub.py +++ b/src/xskill/recommend/skillhub.py @@ -11,10 +11,15 @@ from __future__ import annotations import hashlib +import os import pickle import re +import threading +import time +from dataclasses import dataclass from datetime import datetime from pathlib import Path +from typing import Callable import numpy as np @@ -42,64 +47,184 @@ def _safe_id_part(value: str) -> str: return safe or "skill" -def _content_sha(md: Path) -> str: - return hashlib.sha256(md.read_bytes()).hexdigest()[:16] - - def _path_hash(source_path: str) -> str: return hashlib.sha256(source_path.encode("utf-8")).hexdigest()[:12] +@dataclass(frozen=True) +class _SkillFileMemo: + """一次稳定文件版本对应的解析结果与内容哈希。""" + + mtime_ns: int + size: int + content_sha: str + entry: dict + + +@dataclass(frozen=True) +class _SkillHubSnapshot: + """一次全树扫描结果；条目和名称索引只在实例内部使用。""" + + entries: tuple[dict, ...] + by_name: dict[str, dict] + + class SkillHub: """三方 skill 扫描器 + ux 查询。``enabled=False``（缺省）时为 no-op。""" - def __init__(self, *, enabled: bool, hub_dir: Path | str, embed_client): + def __init__( + self, *, enabled: bool, hub_dir: Path | str, embed_client, + scan_ttl_seconds: float = 5.0, + clock: Callable[[], float] = time.monotonic, + ): self.enabled = bool(enabled) self.dir = Path(hub_dir) self.embed_client = embed_client + self._scan_ttl_seconds = max(0.0, float(scan_ttl_seconds)) + self._clock = clock + self._scan_lock = threading.Lock() + self._snapshot: _SkillHubSnapshot | None = None + self._snapshot_expires_at = 0.0 + self._snapshot_generation = 0 + self._file_memo: dict[Path, _SkillFileMemo] = {} @classmethod def from_config(cls, config: dict, embed_client) -> "SkillHub": cfg = skillhub_config(config) return cls(enabled=cfg["enabled"], hub_dir=cfg["dir"], embed_client=embed_client) - def _entries(self, *, include_vec: bool, require_description: bool) -> list[dict]: - if not self.enabled: - return [] + @staticmethod + def _signature(stat_result: os.stat_result) -> tuple[int, int]: + return stat_result.st_mtime_ns, stat_result.st_size + + def _read_entry( + self, md: Path, rel: str, stat_result: os.stat_result, + ) -> _SkillFileMemo: + """读取变化过的 SKILL.md；若读取期间被替换则按新版本再读一次。""" + content = md.read_bytes() + stable_stat = md.stat() + if self._signature(stable_stat) != self._signature(stat_result): + content = md.read_bytes() + stable_stat = md.stat() + fm, _body = fm_parse(content.decode("utf-8")) + sub = md.parent + raw_name = fm.get("name") or sub.name + display_name = str(raw_name).strip() or sub.name + desc = (fm.get("description") or "").strip() + content_sha = hashlib.sha256(content).hexdigest()[:16] + path_hash = _path_hash(rel) + skill_id = f"{_safe_id_part(display_name)}@{path_hash}" + entry = { + "source": "skillhub", + "name": skill_id, + "skill_id": skill_id, + "display_name": display_name, + "source_path": rel, + "path_hash": path_hash, + "content_sha": content_sha, + "description": desc, + "path": sub, + } + return _SkillFileMemo( + mtime_ns=stable_stat.st_mtime_ns, + size=stable_stat.st_size, + content_sha=content_sha, + entry=entry, + ) + + @staticmethod + def _build_snapshot(entries: list[dict]) -> _SkillHubSnapshot: + """构造 O(1) 精确身份/唯一定义名查询索引。""" + by_name: dict[str, dict] = {} + display_counts: dict[str, int] = {} + for entry in entries: + for key in (entry["skill_id"], entry["name"], entry["source_path"]): + by_name.setdefault(key, entry) + display = entry["display_name"] + display_counts[display] = display_counts.get(display, 0) + 1 + for entry in entries: + display = entry["display_name"] + if display_counts[display] == 1: + by_name.setdefault(display, entry) + return _SkillHubSnapshot(entries=tuple(entries), by_name=by_name) + + def _scan_entries(self) -> _SkillHubSnapshot: + """遍历 skillhub 一次；隐藏目录不下钻，未变化文件不再读取。""" if not self.dir.is_dir(): raise FileNotFoundError( f"skillhub.dir 不存在: {self.dir}（启用 skillhub 前请放置三方 skill）" ) + next_memo: dict[Path, _SkillFileMemo] = {} entries: list[dict] = [] - for md in sorted(self.dir.rglob("SKILL.md")): - sub = md.parent - try: - rel = sub.relative_to(self.dir).as_posix() - except ValueError: + for root, dir_names, file_names in os.walk(self.dir, topdown=True): + dir_names[:] = sorted( + name for name in dir_names if not name.startswith(".") + ) + if "SKILL.md" not in file_names: continue - if any(part.startswith(".") for part in Path(rel).parts): + md = Path(root) / "SKILL.md" + try: + stat_result = md.stat() + rel = md.parent.relative_to(self.dir).as_posix() + except FileNotFoundError: continue - fm, _body = fm_parse(md.read_text(encoding="utf-8")) - raw_name = fm.get("name") or sub.name - display_name = str(raw_name).strip() or sub.name - desc = (fm.get("description") or "").strip() - if require_description and not desc: + except ValueError: continue - content_sha = _content_sha(md) - path_hash = _path_hash(rel) - skill_id = f"{_safe_id_part(display_name)}@{path_hash}" - entry = { - "source": "skillhub", - "name": skill_id, - "skill_id": skill_id, - "display_name": display_name, - "source_path": rel, - "path_hash": path_hash, - "content_sha": content_sha, - "description": desc, - "path": sub, - } - entries.append(entry) + memo = self._file_memo.get(md) + if ( + memo is None + or (memo.mtime_ns, memo.size) != self._signature(stat_result) + ): + try: + memo = self._read_entry(md, rel, stat_result) + except FileNotFoundError: + continue + next_memo[md] = memo + entries.append(memo.entry) + entries.sort(key=lambda entry: entry["source_path"]) + self._file_memo = next_memo + return self._build_snapshot(entries) + + def _current_snapshot(self, *, force_refresh: bool = False) -> _SkillHubSnapshot: + """返回短 TTL 快照；同一代并发过期请求只允许一个线程真扫。""" + observed_generation = self._snapshot_generation + now = self._clock() + if ( + not force_refresh + and self._snapshot is not None + and now < self._snapshot_expires_at + ): + return self._snapshot + with self._scan_lock: + # 等锁期间已有线程完成刷新时直接共享该结果，包括 TTL=0/force 场景。 + if ( + self._snapshot is not None + and self._snapshot_generation != observed_generation + ): + return self._snapshot + now = self._clock() + if ( + not force_refresh + and self._snapshot is not None + and now < self._snapshot_expires_at + ): + return self._snapshot + snapshot = self._scan_entries() + self._snapshot = snapshot + self._snapshot_expires_at = self._clock() + self._scan_ttl_seconds + self._snapshot_generation += 1 + return snapshot + + def _entries( + self, *, include_vec: bool, require_description: bool, + force_refresh: bool = False, + ) -> list[dict]: + if not self.enabled: + return [] + snapshot = self._current_snapshot(force_refresh=force_refresh) + entries = [dict(entry) for entry in snapshot.entries] + if require_description: + entries = [entry for entry in entries if entry["description"]] if include_vec and entries: # 批量 + 按内容哈希复用：重启/改一个 SKILL.md 不再全量重 embed。 embed_store = EmbedStore( @@ -187,23 +312,15 @@ def search(self, query: str, limit: int = 5) -> list[dict]: scored.sort(key=lambda pair: (-pair[0], pair[1]["skill_id"])) return [entry for _score, entry in scored[:limit]] - def entry(self, name: str) -> dict | None: + def entry(self, name: str, *, force_refresh: bool = False) -> dict | None: """按 skill_id / source_path / 唯一 display_name 找当前磁盘上的 skill。""" if not self.enabled: return None - matches: list[dict] = [] - for entry in self._entries(include_vec=False, require_description=False): - if name in { - entry["skill_id"], entry["name"], entry["source_path"], - entry["display_name"], - }: - matches.append(entry) - if len(matches) == 1: - return matches[0] - for entry in matches: - if name in {entry["skill_id"], entry["name"], entry["source_path"]}: - return entry - return None + snapshot = self._current_snapshot(force_refresh=force_refresh) + entry = snapshot.by_name.get(name) + if entry is None or not (Path(entry["path"]) / "SKILL.md").is_file(): + return None + return dict(entry) # ── §7 三方 skill ux 定位 / 版本 / 查询 ────────────────────── # 三方 skill 无 git → 版本号用 SKILL.md 内容 sha256 前 16 位；side 恒 "main" diff --git a/src/xskill/team/server/api.py b/src/xskill/team/server/api.py index 97cd4fec..bb4fe142 100644 --- a/src/xskill/team/server/api.py +++ b/src/xskill/team/server/api.py @@ -883,7 +883,8 @@ def _store_user_skill(hub, owner_dir: str, payload: bytes) -> dict: shutil.rmtree(dest_dir, ignore_errors=True) tmp_dir.replace(dest_dir) source_path = dest_dir.relative_to(Path(hub.dir)).as_posix() - entry = hub.entry(source_path) + # 上传前可能已经建立过 SkillHub TTL 快照；强制刷新保证本次响应立即可见。 + entry = hub.entry(source_path, force_refresh=True) if entry is None: raise HTTPException(status_code=500, detail="stored skill not visible in skillhub scan") diff --git a/tests/test_dashboard_metrics.py b/tests/test_dashboard_metrics.py index 44385e96..b1e5bc0e 100644 --- a/tests/test_dashboard_metrics.py +++ b/tests/test_dashboard_metrics.py @@ -494,6 +494,82 @@ def test_tag_cloud_aggregates_atom_tags(tmp_path): assert cloud["migrate"] == 1 and cloud["nginx"] == 1 +def test_tag_cloud_ttl_reuses_scan_and_refreshes_after_expiry( + tmp_path, monkeypatch): + from xskill.pipeline.atom import AtomTask, AtomTaskStore + wd = tmp_path / "wd"; wd.mkdir() + store = AtomTaskStore(root=wd) + + def save(index, tags): + store.save(AtomTask( + atom_id=f"atom_t_{index:04d}", traj_id="t", + offset_start=1, offset_end=2, intent="i", summary="s", + tags=tags, used_skills=[], ux_score=7, + pre_atom_id=None, post_atom_id=None, context_prefix="", raw_segment="", + )) + + save(0, ["first"]) + db = tmp_path / "tg-cache.db" + conn = get_connection(db) + conn.execute("INSERT INTO watch_dirs(path,label,ecosystem) VALUES(?,?,?)", + (str(wd), "w", "claude_code")) + conn.commit(); conn.close() + now = [0.0] + metrics = DashboardMetrics( + db_path=db, tag_cloud_ttl_seconds=5.0, clock=lambda: now[0], + ) + original = AtomTaskStore.all_atoms + scans = 0 + + def counted(self): + nonlocal scans + scans += 1 + yield from original(self) + + monkeypatch.setattr(AtomTaskStore, "all_atoms", counted) + assert metrics.tag_cloud() == [{"tag": "first", "count": 1, "users": []}] + save(1, ["second"]) + assert metrics.tag_cloud() == [{"tag": "first", "count": 1, "users": []}] + assert scans == 1 + + now[0] = 6.0 + assert {row["tag"] for row in metrics.tag_cloud()} == {"first", "second"} + assert scans == 2 + + +def test_tag_cloud_concurrent_calls_share_one_scan(tmp_path, monkeypatch): + db = tmp_path / "tg-flight.db" + get_connection(db).close() + metrics = DashboardMetrics(db_path=db) + original = metrics._scan_tag_cloud + entered = threading.Event() + release = threading.Event() + calls = 0 + calls_lock = threading.Lock() + + def counted(): + nonlocal calls + with calls_lock: + calls += 1 + entered.set() + assert release.wait(timeout=5) + return original() + + monkeypatch.setattr(metrics, "_scan_tag_cloud", counted) + barrier = threading.Barrier(16) + + def load(): + barrier.wait() + return metrics.tag_cloud() + + with ThreadPoolExecutor(max_workers=16) as pool: + futures = [pool.submit(load) for _ in range(16)] + assert entered.wait(timeout=5) + release.set() + assert all(future.result(timeout=5) == [] for future in futures) + assert calls == 1 + + def test_by_model(tmp_path): db = tmp_path / "r.db" _seed(db) diff --git a/tests/test_skill_hub_search_upload.py b/tests/test_skill_hub_search_upload.py index a652d1ed..cb33e219 100644 --- a/tests/test_skill_hub_search_upload.py +++ b/tests/test_skill_hub_search_upload.py @@ -130,6 +130,12 @@ def test_search_and_upload_503_when_skillhub_disabled(tmp_path): def test_upload_stores_under_user_skill_hub_and_is_searchable(hub_env, tmp_path): cid, hdr = _register(hub_env.client, user_name="alice") + # 先建立不含上传件的 TTL 快照，验证 upload 会显式刷新而非等 5 秒。 + before = hub_env.client.get( + "/api/v1/team/skill_hub/search", + params={"query": "terraform"}, headers=hdr, + ) + assert before.status_code == 200 and before.json()["results"] == [] src = tmp_path / "my-skill-src" src.mkdir() (src / "SKILL.md").write_text( diff --git a/tests/test_skillhub.py b/tests/test_skillhub.py index a91860a4..6f8eb25f 100644 --- a/tests/test_skillhub.py +++ b/tests/test_skillhub.py @@ -5,9 +5,12 @@ """ from __future__ import annotations +import os import pickle import shutil import subprocess +import threading +from concurrent.futures import ThreadPoolExecutor from pathlib import Path import numpy as np @@ -18,6 +21,7 @@ from xskill.recommend.client_interest import ClientInterest from xskill.recommend.client_user import ClientUser from xskill.recommend.engine import SkillRecommendEngine +from xskill.recommend import skillhub as skillhub_module from xskill.recommend.skillhub import SkillHub from xskill.team.client.daemon import TeamClient from xskill.team.client.state import ClientState @@ -41,6 +45,17 @@ def encode_batch(self, texts): return np.stack([self.encode(t) for t in texts]) +class FakeClock: + def __init__(self): + self.now = 0.0 + + def __call__(self): + return self.now + + def advance(self, seconds: float): + self.now += seconds + + def _git(args, cwd): subprocess.run(["git"] + args, cwd=str(cwd), capture_output=True, text=True, check=True) @@ -200,6 +215,121 @@ def test_same_name_and_content_are_still_distinguished_by_relative_path( assert entries[0]["content_sha"] == entries[1]["content_sha"] assert entries[0]["skill_id"] != entries[1]["skill_id"] + def test_ttl_snapshot_reuses_unchanged_files_and_evicts_deleted_memo( + self, tmp_path, monkeypatch, + ): + hub_dir = tmp_path / "hub" + _write_hub_skill(hub_dir, "foo", "first version") + _write_hub_skill(hub_dir, "bar", "unchanged") + clock = FakeClock() + hub = SkillHub( + enabled=True, hub_dir=hub_dir, embed_client=None, + scan_ttl_seconds=5.0, clock=clock, + ) + original = hub._read_entry + reads: list[str] = [] + + def counted(md, rel, stat_result): + reads.append(rel) + return original(md, rel, stat_result) + + monkeypatch.setattr(hub, "_read_entry", counted) + first_sha = hub.entry("foo")["content_sha"] + assert sorted(reads) == ["bar", "foo"] + + # TTL 内不遍历；TTL 后只 stat，未变化文件不再读取/解析/哈希。 + assert hub.entry("foo")["content_sha"] == first_sha + clock.advance(6) + hub.fingerprint() + assert sorted(reads) == ["bar", "foo"] + + skill_md = hub_dir / "foo" / "SKILL.md" + old_mtime = skill_md.stat().st_mtime_ns + os.utime(skill_md, ns=(old_mtime + 1_000_000, old_mtime + 1_000_000)) + clock.advance(6) + assert hub.entry("foo")["content_sha"] == first_sha + assert reads.count("foo") == 2 + + skill_md.write_text( + "---\nname: foo\ndescription: second version\n---\n# foo\n", + encoding="utf-8", + ) + # 保证内容变更测试不依赖文件系统时间戳粒度。 + os.utime(skill_md, ns=(old_mtime + 2_000_000, old_mtime + 2_000_000)) + shutil.rmtree(hub_dir / "bar") + clock.advance(6) + + refreshed = hub.entry("foo") + assert refreshed["content_sha"] != first_sha + assert reads.count("foo") == 3 + assert hub.entry("bar") is None + assert set(hub._file_memo) == {skill_md} + + def test_concurrent_expired_calls_share_one_scan(self, tmp_path, monkeypatch): + hub_dir = tmp_path / "hub" + _write_hub_skill(hub_dir, "foo", "django helper") + hub = SkillHub(enabled=True, hub_dir=hub_dir, embed_client=None) + original = hub._scan_entries + entered = threading.Event() + release = threading.Event() + calls = 0 + calls_lock = threading.Lock() + + def counted(): + nonlocal calls + with calls_lock: + calls += 1 + entered.set() + assert release.wait(timeout=5) + return original() + + monkeypatch.setattr(hub, "_scan_entries", counted) + barrier = threading.Barrier(32) + + def load(): + barrier.wait() + return hub.entry("foo") + + with ThreadPoolExecutor(max_workers=32) as pool: + futures = [pool.submit(load) for _ in range(32)] + assert entered.wait(timeout=5) + release.set() + results = [future.result(timeout=5) for future in futures] + + assert calls == 1 + assert all(entry and entry["display_name"] == "foo" for entry in results) + + def test_scan_prunes_hidden_directories_before_descent( + self, tmp_path, monkeypatch, + ): + hub_dir = tmp_path / "hub" + _write_hub_skill(hub_dir, "visible", "visible helper") + _write_hub_skill(hub_dir, ".git/objects/deep/hidden", "must stay hidden") + visited: list[Path] = [] + original_walk = skillhub_module.os.walk + + def recording_walk(*args, **kwargs): + for root, dirs, files in original_walk(*args, **kwargs): + visited.append(Path(root)) + yield root, dirs, files + + monkeypatch.setattr(skillhub_module.os, "walk", recording_walk) + hub = SkillHub(enabled=True, hub_dir=hub_dir, embed_client=None) + + assert [entry["source_path"] for entry in hub._entries( + include_vec=False, require_description=False, + )] == ["visible"] + assert all(".git" not in path.relative_to(hub_dir).parts for path in visited) + + def test_force_refresh_makes_new_skill_visible_inside_ttl(self, tmp_path): + hub_dir = tmp_path / "hub" + hub_dir.mkdir() + hub = SkillHub(enabled=True, hub_dir=hub_dir, embed_client=None) + assert hub.entry("new-skill") is None + _write_hub_skill(hub_dir, "new-skill", "new helper") + assert hub.entry("new-skill") is None + assert hub.entry("new-skill", force_refresh=True)["source_path"] == "new-skill" + # ── 引擎检索池合并 ─────────────────────────────────────────────── @@ -346,6 +476,8 @@ def test_adding_skillhub_after_empty_scan_refreshes_recommendations( embed_client=FakeEmbed(dim=4), profile_db=tmp_path / "p.db", ) + clock = FakeClock() + eng.skillhub._clock = clock q = FakeEmbed(dim=4).encode("django migration helper") q = q / np.linalg.norm(q) eng.profile_store.upsert( @@ -369,6 +501,7 @@ def test_adding_skillhub_after_empty_scan_refreshes_recommendations( _write_hub_skill( hub_dir, "hub-a/foo", "django migration helper", name="foo", ) + clock.advance(6) second = build_manifest( client_id="client-one", skill_dir=skill_dir, From a42a563080783b014cf9fc0613dff0194ae1ad90 Mon Sep 17 00:00:00 2001 From: 370025263 <370025263@qq.com> Date: Tue, 14 Jul 2026 16:00:17 +0800 Subject: [PATCH 2/3] =?UTF-8?q?test(stress):=20=E5=A2=9E=E5=8A=A0=20skillh?= =?UTF-8?q?ub=20=E4=BA=8B=E4=BB=B6=E5=BE=AA=E7=8E=AF=E9=97=A8=E7=A6=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在发布压力任务中加入面板轮询与 30 路 sync 并发场景，防止全盘扫描再次造成 health 超时。 - 构造 300 个 skill、300 个 atom 和点目录深树 - 断言 SkillHub 与标签云并发请求各只扫描一次 - 连续采样纯异步 health 并约束 p99 小于 500ms - CI 与 release 压力任务共同执行新门禁 --- .github/workflows/ci.yml | 5 +- .github/workflows/release.yml | 5 +- tests/stress/test_skillhub_event_loop.py | 152 +++++++++++++++++++++++ 3 files changed, 158 insertions(+), 4 deletions(-) create mode 100644 tests/stress/test_skillhub_event_loop.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c796dceb..aaf1b95d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -117,9 +117,10 @@ jobs: cache: pip - name: Install package with dev extras run: pip install -e .[dev] - - name: Run 300x300 control-plane stress test + - name: Run control-plane stress tests run: >- - pytest tests/stress/test_control_plane_300.py -v + pytest tests/stress/test_control_plane_300.py + tests/stress/test_skillhub_event_loop.py -v --override-ini="addopts=" -m stress --basetemp=.stress-artifacts - name: Upload stress-test evidence diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index cf897d3d..ea7b1f72 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -42,9 +42,10 @@ jobs: cache: pip - name: Install package with dev extras run: pip install -e .[dev] - - name: Run 300x300 control-plane stress test + - name: Run control-plane stress tests run: >- - pytest tests/stress/test_control_plane_300.py -v + pytest tests/stress/test_control_plane_300.py + tests/stress/test_skillhub_event_loop.py -v --override-ini="addopts=" -m stress --basetemp=.stress-artifacts - name: Upload stress-test evidence diff --git a/tests/stress/test_skillhub_event_loop.py b/tests/stress/test_skillhub_event_loop.py new file mode 100644 index 00000000..e64a5baa --- /dev/null +++ b/tests/stress/test_skillhub_event_loop.py @@ -0,0 +1,152 @@ +"""SkillHub/dashboard 并发扫描不得饿死 ASGI 事件循环。""" +from __future__ import annotations + +import asyncio +import math +import threading +import time +from pathlib import Path + +import httpx +import pytest +from fastapi import FastAPI + +from xskill.dashboard.metrics import DashboardMetrics +from xskill.pipeline.atom import AtomTask, AtomTaskStore +from xskill.pipeline.registry import get_connection +from xskill.recommend.skillhub import SkillHub + + +def _write_skill(root: Path, index: int) -> str: + name = f"hub-skill-{index:03d}" + skill_dir = root / name + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text( + "---\n" + f"name: {name}\n" + f"description: Load-test helper {index:03d} for skillhub scanning\n" + "---\n\n" + f"# {name}\n\n" + ("fixture body\n" * 32), + encoding="utf-8", + ) + return name + + +def _write_atom(store: AtomTaskStore, index: int) -> None: + store.save(AtomTask( + atom_id=f"atom_load_{index:04d}", + traj_id=f"traj_load_{index:04d}", + offset_start=1, + offset_end=2, + intent="load test", + summary="dashboard tag fixture", + tags=["dashboard", f"bucket-{index % 10}"], + used_skills=[], + )) + + +async def _sample_health(client: httpx.AsyncClient, duration: float) -> list[float]: + latencies: list[float] = [] + deadline = time.monotonic() + duration + while time.monotonic() < deadline: + started = time.monotonic() + response = await client.get("/api/v1/health") + latencies.append(time.monotonic() - started) + assert response.status_code == 200 + await asyncio.sleep(0.01) + return latencies + + +def _percentile(values: list[float], percentile: float) -> float: + ordered = sorted(values) + index = min(len(ordered) - 1, math.ceil(percentile * len(ordered)) - 1) + return ordered[index] + + +@pytest.mark.stress +@pytest.mark.timeout(30) +def test_skillhub_and_dashboard_load_keeps_health_responsive( + tmp_path, monkeypatch) -> None: + hub_dir = tmp_path / "skillhub" + hub_dir.mkdir() + names = [_write_skill(hub_dir, index) for index in range(300)] + + # 旧 rglob 会进入点目录并递归这些无关层级；新扫描必须在 .git 处剪枝。 + hidden_root = hub_dir / ".git" / "objects" / "deep" / "nested" / "tree" + for index in range(200): + hidden = hidden_root / f"object-{index:03d}" + hidden.mkdir(parents=True) + (hidden / "SKILL.md").write_text("ignored", encoding="utf-8") + + atom_root = tmp_path / "atoms" + store = AtomTaskStore(atom_root) + for index in range(300): + _write_atom(store, index) + db = tmp_path / "registry.db" + conn = get_connection(db) + conn.execute( + "INSERT INTO watch_dirs(path,label,ecosystem) VALUES(?,?,?)", + (str(atom_root), "load-user", "team_client"), + ) + conn.commit() + conn.close() + + hub = SkillHub(enabled=True, hub_dir=hub_dir, embed_client=None) + metrics = DashboardMetrics(db_path=db) + scan_counts = {"skillhub": 0, "atoms": 0} + count_lock = threading.Lock() + original_hub_scan = hub._scan_entries + original_all_atoms = AtomTaskStore.all_atoms + + def counted_hub_scan(): + with count_lock: + scan_counts["skillhub"] += 1 + return original_hub_scan() + + def counted_all_atoms(self): + with count_lock: + scan_counts["atoms"] += 1 + yield from original_all_atoms(self) + + monkeypatch.setattr(hub, "_scan_entries", counted_hub_scan) + monkeypatch.setattr(AtomTaskStore, "all_atoms", counted_all_atoms) + + app = FastAPI() + + @app.get("/api/v1/health") + async def health(): + return {"status": "ok"} + + @app.get("/api/v1/team/sync") + def sync(): + return {"hits": sum(hub.entry(name) is not None for name in names[:100])} + + @app.get("/api/v1/dashboard/tags") + def dashboard_tags(): + return {"tags": metrics.tag_cloud()} + + async def scenario(): + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, base_url="http://stress", + ) as client: + health_task = asyncio.create_task(_sample_health(client, 1.0)) + await asyncio.sleep(0) + requests = [ + client.get("/api/v1/team/sync") for _ in range(30) + ] + [ + client.get("/api/v1/dashboard/tags") for _ in range(10) + ] + responses = await asyncio.wait_for( + asyncio.gather(*requests), timeout=15, + ) + latencies = await health_task + return responses, latencies + + responses, health_latencies = asyncio.run(scenario()) + + assert all(response.status_code == 200 for response in responses) + assert all(response.json().get("hits") == 100 for response in responses[:30]) + assert scan_counts == {"skillhub": 1, "atoms": 1} + assert len(health_latencies) >= 20 + assert _percentile(health_latencies, 0.99) < 0.5 From 110ea0b43737515c53c7f79fe7bf8324be97fa04 Mon Sep 17 00:00:00 2001 From: 370025263 <370025263@qq.com> Date: Tue, 14 Jul 2026 17:48:33 +0800 Subject: [PATCH 3/3] fix(ci): handle cross-platform paths and file locks --- src/xskill/pipeline/registry.py | 22 ++++++++++++++++++++++ src/xskill/skill/skill.py | 13 ++++++++++++- tests/test_pooled_connection.py | 1 + tests/test_skill_delete.py | 33 +++++++++++++++++++++++++++++++++ tests/test_skill_tools_atom.py | 9 +++++++-- 5 files changed, 75 insertions(+), 3 deletions(-) create mode 100644 tests/test_skill_delete.py diff --git a/src/xskill/pipeline/registry.py b/src/xskill/pipeline/registry.py index f8642788..6e0f1e70 100644 --- a/src/xskill/pipeline/registry.py +++ b/src/xskill/pipeline/registry.py @@ -363,6 +363,28 @@ def pooled_connection(db_path: Optional[Path] = None) -> Iterator[sqlite3.Connec slot.busy = False +def close_pooled_connection(db_path: Optional[Path] = None) -> bool: + """Close this thread's idle pooled connection for ``db_path``. + + Call this before replacing or deleting a database file on platforms such + as Windows, where an open SQLite handle prevents filesystem removal. + """ + if db_path is None: + db_path = get_registry_db_path() + db_key = db_path.expanduser().resolve() + slots = getattr(_REGISTRY_THREAD_POOL, "slots", None) + if not slots: + return False + slot = slots.get(db_key) + if slot is None: + return False + if slot.busy: + raise RuntimeError("cannot close a pooled connection while it is in use") + slots.pop(db_key) + slot.conn.close() + return True + + def _migrate(conn: sqlite3.Connection) -> None: """Add columns missing from older schema versions.""" # ── trajectories ── diff --git a/src/xskill/skill/skill.py b/src/xskill/skill/skill.py index 600040a3..a9ee20fc 100644 --- a/src/xskill/skill/skill.py +++ b/src/xskill/skill/skill.py @@ -15,7 +15,9 @@ import json import logging +import os import shutil +import stat from datetime import datetime, date from pathlib import Path from typing import TYPE_CHECKING, Literal, Optional @@ -34,6 +36,15 @@ logger = logging.getLogger("xskill.skill_manager") +def _remove_readonly(func, path: str, exc_info) -> None: + """Make a read-only path writable and retry its failed removal.""" + error = exc_info[1] + if not isinstance(error, PermissionError): + raise error + os.chmod(path, stat.S_IWRITE) + func(path) + + # ═════════════════════════════════════════════════════════════════ # CandidateBuffer (internal — 不暴露) # ═════════════════════════════════════════════════════════════════ @@ -552,7 +563,7 @@ def delete_skill(skill_dir: Path, name: str) -> bool: logger.error(f"skill not found: {name}") return False - shutil.rmtree(skill_path) + shutil.rmtree(skill_path, onerror=_remove_readonly) committed = commit_changes(str(skill_dir), f"delete skill: {name}") if committed: logger.info(f"deleted: {name}") diff --git a/tests/test_pooled_connection.py b/tests/test_pooled_connection.py index f7a52971..7c3adadf 100644 --- a/tests/test_pooled_connection.py +++ b/tests/test_pooled_connection.py @@ -82,6 +82,7 @@ def test_deleted_db_file_triggers_reopen(tmp_path): first.execute("INSERT INTO llm_usage(step,model,prompt,completion," "total,cost_usd,price_source) VALUES('s','m',1,1,2,0,'x')") first.commit() + assert registry.close_pooled_connection(db) db.unlink() for suffix in ("-wal", "-shm"): sidecar = db.with_name(db.name + suffix) diff --git a/tests/test_skill_delete.py b/tests/test_skill_delete.py new file mode 100644 index 00000000..e97ec53d --- /dev/null +++ b/tests/test_skill_delete.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import os +import stat + +from xskill.skill import skill + + +def test_delete_skill_retries_readonly_git_object(tmp_path, monkeypatch): + skill_dir = tmp_path / "skills" + git_object = skill_dir / "demo" / ".git" / "objects" / "aa" / "object" + git_object.parent.mkdir(parents=True) + git_object.write_bytes(b"git object") + git_object.chmod(stat.S_IREAD) + monkeypatch.setattr(skill, "commit_changes", lambda *_args: True) + + original_unlink = os.unlink + attempts = 0 + + def windows_unlink(path, *args, **kwargs): + nonlocal attempts + if os.fspath(path).endswith("object"): + attempts += 1 + mode = os.stat(path, dir_fd=kwargs.get("dir_fd")).st_mode + if not mode & stat.S_IWRITE: + raise PermissionError("read-only Git object") + return original_unlink(path, *args, **kwargs) + + monkeypatch.setattr(os, "unlink", windows_unlink) + + assert skill.delete_skill(skill_dir, "demo") + assert attempts == 2 + assert not (skill_dir / "demo").exists() diff --git a/tests/test_skill_tools_atom.py b/tests/test_skill_tools_atom.py index 33dd831f..b860bcc4 100644 --- a/tests/test_skill_tools_atom.py +++ b/tests/test_skill_tools_atom.py @@ -1,12 +1,17 @@ """agent_tools v2 atom-era 工具集单测""" from __future__ import annotations +import tempfile from pathlib import Path from xskill.pipeline.atom import AtomTask, AtomTaskStore from xskill.agents import agent_tools +def _skilleditagent_tmp_dir() -> Path: + return Path(tempfile.gettempdir()) / "xskill" / "skilleditagent" + + def _setup(tmp_path: Path) -> tuple[Path, AtomTaskStore]: skill_dir = tmp_path / "skill" skill_dir.mkdir() @@ -86,7 +91,7 @@ def test_reads_tmp_spill_file_with_path_context(self, tmp_path): agent_tools.init_skill_authoring_tool_context( skill_dir, skill_dir, {"skill_opt": {"enabled": False}}, ) - spill = Path("/tmp/xskill/skilleditagent") / f"{tmp_path.name}-spill.txt" + spill = _skilleditagent_tmp_dir() / f"{tmp_path.name}-spill.txt" spill.parent.mkdir(parents=True, exist_ok=True) spill.write_text("spilled raw tool result\n", encoding="utf-8") @@ -102,7 +107,7 @@ def test_reads_line_window_with_offset_and_limit(self, tmp_path): agent_tools.init_skill_authoring_tool_context( skill_dir, skill_dir, {"skill_opt": {"enabled": False}}, ) - spill = Path("/tmp/xskill/skilleditagent") / f"{tmp_path.name}-window.txt" + spill = _skilleditagent_tmp_dir() / f"{tmp_path.name}-window.txt" spill.parent.mkdir(parents=True, exist_ok=True) spill.write_text("L1\nL2\nL3\nL4\n", encoding="utf-8")