fix(skills): honour overall_timeout and bound ClawHub catalog walk

tests/tools/test_skills_hub.py

@@ -1,6 +1,8 @@
 """Tests for tools/skills_hub.py — source adapters, lock file, taps, dedup logic."""
 
 import json
+import time
+from typing import List, Optional
 from unittest.mock import patch, MagicMock
 
 import httpx
@@ -14,13 +16,15 @@
     UrlSource,
     WellKnownSkillSource,
     OptionalSkillSource,
-    SkillMeta,
+    SkillSource,
     SkillBundle,
+    SkillMeta,
     HubLockFile,
     TapsManager,
     bundle_content_hash,
     check_for_skill_updates,
     create_source_router,
+    parallel_search_sources,
     unified_search,
     append_audit_log,
     _skill_meta_to_dict,
@@ -2201,3 +2205,80 @@ def test_install_from_quarantine_rejects_symlinks(self, tmp_path):
 
         assert not (skills_dir / "bad-skill" / "leak.txt").exists()
         assert secret.read_text() == "data exfiltration payload\n"
+
+
+# ---------------------------------------------------------------------------
+# parallel_search_sources — overall_timeout must be honoured even when a
+# source blocks for far longer than the budget (regression: the executor used
+# `with ... as pool`, whose __exit__ calls shutdown(wait=True) and blocked the
+# caller on the slow worker, making overall_timeout a no-op).
+# ---------------------------------------------------------------------------
+
+
+class _FakeSource(SkillSource):
+    def __init__(self, sid: str, sleep: float = 0.0, results=None):
+        self._sid = sid
+        self._sleep = sleep
+        self._results = results or []
+
+    def source_id(self) -> str:
+        return self._sid
+
+    def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
+        if self._sleep:
+            time.sleep(self._sleep)
+        return list(self._results)
+
+    def fetch(self, identifier: str) -> Optional[SkillBundle]:
+        return None
+
+    def inspect(self, identifier: str) -> Optional[SkillMeta]:
+        return None
+
+
+class TestParallelSearchSourcesTimeout:
+    def _meta(self, sid: str) -> SkillMeta:
+        return SkillMeta(
+            name=f"{sid}-skill",
+            description="x",
+            source=sid,
+            identifier=f"{sid}/x",
+            trust_level="community",
+        )
+
+    def test_slow_source_does_not_block_caller(self):
+        """A source sleeping well past overall_timeout must not stall the
+        return. Before the fix the executor's `with` block waited on the slow
+        worker (~5s); now the call returns promptly and reports the source as
+        timed out."""
+        fast = _FakeSource("fast", sleep=0.0, results=[self._meta("fast")])
+        slow = _FakeSource("slow", sleep=5.0, results=[self._meta("slow")])
+
+        start = time.monotonic()
+        all_results, source_counts, timed_out_ids = parallel_search_sources(
+            [fast, slow], query="q", overall_timeout=0.3,
+        )
+        elapsed = time.monotonic() - start
+
+        # Must return long before the slow source's 5s sleep finishes.
+        assert elapsed < 2.0, f"call blocked for {elapsed:.2f}s (timeout not honoured)"
+        assert "slow" in timed_out_ids
+        # Fast source still delivered its result and is not flagged timed out.
+        assert source_counts.get("fast") == 1
+        assert "fast" not in timed_out_ids
+        assert any(r.source == "fast" for r in all_results)
+
+    def test_all_fast_sources_complete_without_timeout(self):
+        """Happy path: when every source finishes within budget, none are
+        flagged and all results are collected."""
+        a = _FakeSource("a", results=[self._meta("a")])
+        b = _FakeSource("b", results=[self._meta("b")])
+
+        all_results, source_counts, timed_out_ids = parallel_search_sources(
+            [a, b], query="q", overall_timeout=5.0,
+        )
+
+        assert timed_out_ids == []
+        assert source_counts.get("a") == 1
+        assert source_counts.get("b") == 1
+        assert len(all_results) == 2

tests/tools/test_skills_hub_clawhub.py

@@ -350,6 +350,78 @@ def side_effect(url, *args, **kwargs):
         self.assertIn("b-skill-199", identifiers)
         self.assertIn("c-skill-49", identifiers)
 
+    @patch("tools.skills_hub._write_index_cache")
+    @patch("tools.skills_hub._read_index_cache", return_value=None)
+    @patch("tools.skills_hub.httpx.get")
+    def test_catalog_walk_aborts_on_budget_and_does_not_poison_cache(
+        self, mock_get, _mock_read_cache, mock_write_cache
+    ):
+        """A walk truncated by the wall-clock budget must stop early and must
+        NOT write the (partial) result to the cache. Before the budget guard
+        the walk ran up to 750 pages and cached unconditionally — a truncated
+        walk poisoned the cache with incomplete catalog data."""
+        page_calls = {"n": 0}
+
+        def side_effect(url, *args, **kwargs):
+            if url.endswith("/skills"):
+                idx = page_calls["n"]
+                page_calls["n"] += 1
+                # Always advertise another page so the walk would never stop
+                # on its own — only the budget can break it.
+                return _MockResponse(
+                    status_code=200,
+                    json_data={
+                        "items": [
+                            {"slug": f"skill-{idx}", "displayName": f"Skill {idx}"}
+                        ],
+                        "nextCursor": f"cursor-{idx + 1}",
+                    },
+                )
+            return _MockResponse(status_code=404, json_data={})
+
+        mock_get.side_effect = side_effect
+
+        # Force the deadline to be in the past immediately.
+        with patch.object(ClawHubSource, "CATALOG_WALK_BUDGET_SECONDS", -1):
+            results = self.src._load_catalog_index()
+
+        # Walk broke well before the 750-page cap.
+        self.assertLess(page_calls["n"], 750)
+        # Truncated walk must not poison the cache.
+        mock_write_cache.assert_not_called()
+        # Whatever was gathered is still returned to the caller.
+        self.assertIsInstance(results, list)
+
+    @patch("tools.skills_hub._write_index_cache")
+    @patch("tools.skills_hub._read_index_cache", return_value=None)
+    @patch("tools.skills_hub.httpx.get")
+    def test_catalog_walk_caches_when_terminating_naturally_within_budget(
+        self, mock_get, _mock_read_cache, mock_write_cache
+    ):
+        """Happy path: a walk that exhausts the cursor within the budget DOES
+        write the cache."""
+
+        def side_effect(url, *args, **kwargs):
+            if url.endswith("/skills"):
+                return _MockResponse(
+                    status_code=200,
+                    json_data={
+                        "items": [
+                            {"slug": "only-skill", "displayName": "Only Skill"}
+                        ],
+                        # No nextCursor -> natural termination.
+                    },
+                )
+            return _MockResponse(status_code=404, json_data={})
+
+        mock_get.side_effect = side_effect
+
+        results = self.src._load_catalog_index()
+
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0].identifier, "only-skill")
+        mock_write_cache.assert_called_once()
+
 
 if __name__ == "__main__":
     unittest.main()

tools/skills_hub.py

@@ -1946,6 +1946,12 @@ class ClawHubSource(SkillSource):
 
     BASE_URL = "https://clawhub.ai/api/v1"
 
+    # Wall-clock budget for a full catalog walk. ClawHub has 50k+ skills and
+    # the walk is sequential (~250 requests, each under per-request
+    # timeout=30 so nothing errors), so an unbounded walk can block for
+    # minutes. Bound it so a slow/large catalog cannot hang the caller.
+    CATALOG_WALK_BUDGET_SECONDS = 12
+
     def source_id(self) -> str:
         return "clawhub"
 
@@ -2258,8 +2264,13 @@ def _load_catalog_index(self) -> List[SkillMeta]:
         # terminates well before this on `nextCursor` going None — the cap is
         # a safety rail against an infinite-cursor loop.
         max_pages = 750
+        deadline = time.monotonic() + self.CATALOG_WALK_BUDGET_SECONDS
+        hit_deadline = False
 
         for _ in range(max_pages):
+            if time.monotonic() > deadline:
+                hit_deadline = True
+                break
             params: Dict[str, Any] = {"limit": 200}
             if cursor:
                 params["cursor"] = cursor
@@ -2297,7 +2308,11 @@ def _load_catalog_index(self) -> List[SkillMeta]:
             if not isinstance(cursor, str) or not cursor:
                 break
 
-        _write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in results])
+        # Only cache a walk that reached a natural stop (cursor exhausted or
+        # page cap). A walk truncated by the wall-clock budget is partial, so
+        # writing it would poison the cache with incomplete catalog data.
+        if not hit_deadline:
+            _write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in results])
         return results
 
     def _get_json(self, url: str, timeout: int = 20) -> Optional[Any]:
@@ -3774,13 +3789,20 @@ def parallel_search_sources(
     if not active:
         return all_results, source_counts, timed_out_ids
 
-    with ThreadPoolExecutor(max_workers=min(len(active), 8)) as pool:
-        futures = {}
-        for src in active:
-            lim = per_source_limits.get(src.source_id(), 50)
-            fut = pool.submit(_search_one_source, src, query, lim)
-            futures[fut] = src.source_id()
+    # NOTE: a `with ThreadPoolExecutor(...) as pool` block calls
+    # ``shutdown(wait=True)`` on exit, which blocks until every submitted
+    # worker finishes — so a single slow source (e.g. ClawHub) keeps the
+    # caller blocked for minutes and renders ``overall_timeout`` a no-op.
+    # Manage the executor manually and shut it down with ``wait=False`` so
+    # the timeout is actually honoured.
+    pool = ThreadPoolExecutor(max_workers=min(len(active), 8))
+    futures = {}
+    for src in active:
+        lim = per_source_limits.get(src.source_id(), 50)
+        fut = pool.submit(_search_one_source, src, query, lim)
+        futures[fut] = src.source_id()
 
+    try:
         try:
             for fut in as_completed(futures, timeout=overall_timeout):
                 try:
@@ -3800,6 +3822,10 @@ def parallel_search_sources(
                     "Skills browse timed out waiting for: %s",
                     ", ".join(timed_out_ids),
                 )
+    finally:
+        # wait=False so a slow source cannot block the caller's return;
+        # cancel_futures drops not-yet-started work.
+        pool.shutdown(wait=False, cancel_futures=True)
 
     return all_results, source_counts, timed_out_ids