diff --git a/tests/tools/test_skills_hub.py b/tests/tools/test_skills_hub.py index c608a7be10..c2eb9866a4 100644 --- a/tests/tools/test_skills_hub.py +++ b/tests/tools/test_skills_hub.py @@ -1765,11 +1765,13 @@ class _FakeSource(SkillSource): self._sid = sid self._sleep = sleep self._results = results or [] + self.calls = 0 def source_id(self) -> str: return self._sid def search(self, query: str, limit: int = 10) -> List[SkillMeta]: + self.calls += 1 if self._sleep: time.sleep(self._sleep) return list(self._results) @@ -1829,6 +1831,50 @@ class TestParallelSearchSourcesTimeout: assert len(all_results) == 2 +class TestIndexMissFallback: + """An available hermes-index stands in for the external registries; when it + has no match for a query the registries it displaced must still be asked + (#112503: a skill live on skills.sh but not yet in the index returned zero + results on every surface).""" + + def _meta(self, sid: str) -> SkillMeta: + return SkillMeta(name="humanizar", description="x", source=sid, + identifier=f"{sid}/humanizar", trust_level="community") + + def _sources(self, index_results): + index = _FakeSource("hermes-index", results=index_results) + index.is_available = True + skills_sh = _FakeSource("skills-sh", results=[self._meta("skills-sh")]) + github = _FakeSource("github", results=[self._meta("github")]) + return index, skills_sh, github + + def test_index_miss_consults_displaced_registries_but_not_github(self): + index, skills_sh, github = self._sources([]) + + results, source_counts, timed_out = parallel_search_sources( + [index, skills_sh, github], query="humanizar", overall_timeout=5.0) + + assert [r.identifier for r in results] == ["skills-sh/humanizar"] + assert source_counts == {"hermes-index": 0, "skills-sh": 1} + assert timed_out == [] + assert github.calls == 0 # one miss must not spend the unauthenticated GitHub budget + + # A browse (empty query) with an empty index is not a miss: no fan-out. + index, skills_sh, github = self._sources([]) + results, _, _ = parallel_search_sources([index, skills_sh, github], query="", overall_timeout=5.0) + assert results == [] and skills_sh.calls == 0 + + def test_index_hit_leaves_registries_untouched(self): + index, skills_sh, github = self._sources([self._meta("hermes-index")]) + + results, source_counts, _ = parallel_search_sources( + [index, skills_sh, github], query="humanizar", overall_timeout=5.0) + + assert [r.identifier for r in results] == ["hermes-index/humanizar"] + assert source_counts == {"hermes-index": 1} + assert skills_sh.calls == 0 and github.calls == 0 + + # --------------------------------------------------------------------------- # _load_hermes_index — centralized index fetch (Browse-hub landing / search) # --------------------------------------------------------------------------- diff --git a/tools/skills_hub_search.py b/tools/skills_hub_search.py index 2ecfca9914..9f04ae4737 100644 --- a/tools/skills_hub_search.py +++ b/tools/skills_hub_search.py @@ -11,6 +11,7 @@ from __future__ import annotations import logging import httpx import json +import time from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from tools.skills_hub_clawhub import ClawHubSource @@ -82,6 +83,11 @@ def _load_stale_index_cache() -> Optional[dict]: # index is available and no source filter is active (~70 GitHub calls/search # for unauthenticated users otherwise). _API_SOURCE_IDS = frozenset({"github", "skills-sh", "clawhub", "lobehub", "well-known"}) +# Consulted only when the index answered a non-empty query with nothing: the +# index is rebuilt asynchronously and lags the registries, so a skill that is +# live on skills.sh may not be in it yet. GitHub stays out — one miss (a typo) +# would burn an unauthenticated user's whole hourly GitHub budget. +_INDEX_MISS_FALLBACK_IDS = _API_SOURCE_IDS - {"github"} def create_source_router(auth: Optional[GitHubAuth] = None) -> List[SkillSource]: @@ -141,32 +147,37 @@ def _select_active_sources(sources: List[SkillSource], source_filter: str) -> Li return active -def parallel_search_sources( - sources: List[SkillSource], query: str = "", per_source_limits: Optional[Dict[str, int]] = None, - source_filter: str = "all", overall_timeout: float = 30, on_source_done: Optional[Any] = None, -) -> Tuple[List[SkillMeta], Dict[str, int], List[str]]: - """Search all sources in parallel with an overall timeout. +def _index_miss_fallback_sources( + sources: List[SkillSource], active: List[SkillSource], query: str, source_counts: Dict[str, int], +) -> List[SkillSource]: + """Registries to consult after the index stood in for them and found nothing. - Returns ``(all_results, source_counts, timed_out_ids)``. *on_source_done* - is an optional ``(source_id, count) -> None`` progress callback. Under a - provider filter every source's results are narrowed before they are - counted and merged, so callers need no provider logic of their own. + Empty for a browse (no query), when the index was not consulted (no skip + happened), or when it returned matches. """ + if not query.strip() or not any(src.source_id() == "hermes-index" for src in active): + return [] + if source_counts.get("hermes-index"): + return [] + return [src for src in sources if src.source_id() in _INDEX_MISS_FALLBACK_IDS and src not in active] + + +def _fan_out( + active: List[SkillSource], query: str, per_source_limits: Dict[str, int], provider_filter: str, + deadline: float, on_source_done: Optional[Any], all_results: List[SkillMeta], + source_counts: Dict[str, int], timed_out_ids: List[str], +) -> None: + """Query ``active`` in parallel until ``deadline`` (monotonic), merging into the accumulators.""" from concurrent.futures import as_completed + from tools.daemon_pool import DaemonThreadPoolExecutor - per_source_limits = per_source_limits or {} - active = _select_active_sources(sources, source_filter) - provider_filter = _provider_filter_of(source_filter) - all_results: List[SkillMeta] = [] - source_counts: Dict[str, int] = {} - timed_out_ids: List[str] = [] - if not active: - return all_results, source_counts, timed_out_ids - + remaining = deadline - time.monotonic() + if remaining <= 0: + timed_out_ids.extend(src.source_id() for src in active) + return # Not a ``with`` block: its shutdown(wait=True) would block on a slow source # (ClawHub) for minutes and defeat ``overall_timeout``. Daemon workers so an # abandoned source cannot block interpreter exit either. - from tools.daemon_pool import DaemonThreadPoolExecutor pool = DaemonThreadPoolExecutor(max_workers=min(len(active), 8)) futures = { pool.submit( @@ -175,7 +186,7 @@ def parallel_search_sources( for src in active } try: - for fut in as_completed(futures, timeout=overall_timeout): + for fut in as_completed(futures, timeout=remaining): try: sid, results = fut.result(timeout=0) if provider_filter: @@ -190,11 +201,46 @@ def parallel_search_sources( except Exception: pass except TimeoutError: - timed_out_ids = [futures[f] for f in futures if not f.done()] - if timed_out_ids: - logger.debug("Skills browse timed out waiting for: %s", ", ".join(timed_out_ids)) + late = [futures[f] for f in futures if not f.done()] + timed_out_ids.extend(late) + if late: + logger.debug("Skills browse timed out waiting for: %s", ", ".join(late)) finally: pool.shutdown(wait=False, cancel_futures=True) + + +def parallel_search_sources( + sources: List[SkillSource], query: str = "", per_source_limits: Optional[Dict[str, int]] = None, + source_filter: str = "all", overall_timeout: float = 30, on_source_done: Optional[Any] = None, +) -> Tuple[List[SkillMeta], Dict[str, int], List[str]]: + """Search all sources in parallel with an overall timeout. + + Returns ``(all_results, source_counts, timed_out_ids)``. *on_source_done* + is an optional ``(source_id, count) -> None`` progress callback. Under a + provider filter every source's results are narrowed before they are + counted and merged, so callers need no provider logic of their own. + + When the centralized index stood in for the external registries and found + nothing for a non-empty query, those registries are queried within the + same ``overall_timeout`` so every caller (CLI, TUI gateway, dashboard) + still finds skills the index has not picked up yet. + """ + per_source_limits = per_source_limits or {} + active = _select_active_sources(sources, source_filter) + provider_filter = _provider_filter_of(source_filter) + all_results: List[SkillMeta] = [] + source_counts: Dict[str, int] = {} + timed_out_ids: List[str] = [] + if not active: + return all_results, source_counts, timed_out_ids + + deadline = time.monotonic() + overall_timeout + _fan_out(active, query, per_source_limits, provider_filter, deadline, on_source_done, + all_results, source_counts, timed_out_ids) + fallback = _index_miss_fallback_sources(sources, active, query, source_counts) + if fallback: + _fan_out(fallback, query, per_source_limits, provider_filter, deadline, on_source_done, + all_results, source_counts, timed_out_ids) return all_results, source_counts, timed_out_ids diff --git a/website/docs/user-guide/features/skills.md b/website/docs/user-guide/features/skills.md index ec07b8ec24..e084880892 100644 --- a/website/docs/user-guide/features/skills.md +++ b/website/docs/user-guide/features/skills.md @@ -702,6 +702,8 @@ in the pending JSON file). Memory writes have the same gate under Browse, search, install, and manage skills from online registries, `skills.sh`, direct well-known skill endpoints, and official optional skills. +Unfiltered searches (CLI, TUI, and the dashboard) are answered from a cached centralized index that covers the external registries. That index is rebuilt periodically, so when it has no match for your query Hermes also asks `skills.sh`, ClawHub, LobeHub and well-known endpoints directly within the same search budget — a skill published minutes ago still shows up. Custom GitHub taps are not part of that fallback (search them with `--source github`, or via the index once it catches up). + ### Common commands ```bash