Fix profile skills-stats thundering herd at cold startup (#5364)

The two-tier mtime cache from #4783 fixed the per-request SKILL.md rescan
but left two concurrency holes that only bite at container cold start,
when the frontend fires several profile-data requests at once and the
caches are empty:

1. `_get_profile_skills_stats()` had no lock, so concurrent misses on the
   same profile each ran `os.walk(followlinks=True)` + parsed every
   SKILL.md simultaneously.
2. `_build_profile_rows_fast()` ran outside `_LIST_PROFILES_CACHE_LOCK`
   in `list_profiles_api()`, so every concurrent request rebuilt all rows
   (each walking every profile's skill tree) at once.

With ThreadingHTTPServer (one OS thread per request) and Docker overlay2,
this stacked thousands of concurrent stat() calls and stalled workers
57-70s (per the report's thread dumps).

Fix:
- Add a per-profile compute lock (registry guarded by a meta-lock) and
  use double-checked locking in `_get_profile_skills_stats()`: concurrent
  misses on one profile collapse to a single compute, while independent
  profiles still compute in parallel.
- Single-flight the row build in `list_profiles_api()` by holding
  `_LIST_PROFILES_CACHE_LOCK` across the build + cache write. Lock order
  is strictly list-lock -> per-profile skills-lock, so no deadlock.

The report's third suggestion (debounce the mtime probe) is deliberately
NOT taken: the every-call cheap probe is the #4783 out-of-band
change-detection contract (test_issue4783 asserts it MUST run on every
call). Serializing the misses removes the herd without weakening that
contract, since only the expensive compute is guarded, not the probe.

Adds tests/test_issue5364_skills_stats_thundering_herd.py proving the
herd collapses (single compute / single build under a concurrent burst),
independent profiles still parallelize, and the every-call probe contract
is preserved. All existing #4783 contract tests still pass.

Co-authored-by: claw-io <claw-io@users.noreply.github.com>

tests/test_issue5364_skills_stats_thundering_herd.py

@@ -0,0 +1,298 @@
+"""Tests for the cold-startup thundering-herd fix in api.profiles (#5364).
+
+Background
+----------
+The two-tier mtime cache from #4783 fixed the per-request SKILL.md rescan, but
+left two concurrency holes that only bite at container cold start, when the
+frontend fires several profile-data requests at once and the caches are empty:
+
+  1. ``_get_profile_skills_stats`` had NO lock, so concurrent misses on the same
+     profile each ran ``os.walk`` + parsed every SKILL.md simultaneously.
+  2. ``_build_profile_rows_fast`` ran OUTSIDE ``_LIST_PROFILES_CACHE_LOCK`` in
+     ``list_profiles_api``, so every concurrent request rebuilt all rows (each
+     walking every profile's skill tree) instead of one building while the rest
+     waited.
+
+Under Docker overlay2 with 9 profiles this stacked ~45k concurrent ``stat``
+calls and stalled worker threads for 57–70 s (per the report).
+
+These tests prove:
+  * concurrent misses on one profile collapse to a SINGLE compute;
+  * the per-profile lock registry returns a stable lock per profile;
+  * a concurrent ``list_profiles_api`` burst builds the rows exactly ONCE;
+  * the #4783 contract is preserved — the cheap mtime probe still runs on every
+    call so out-of-band changes stay promptly visible.
+"""
+import sys
+import threading
+import time
+import types
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Import harness (mirrors tests/test_issue4783_profile_skills_mtime_cache.py)
+# ---------------------------------------------------------------------------
+
+def _make_profiles_module():
+    """Import api.profiles with minimal stubs for heavy dependencies."""
+    stubs = {
+        "flask": types.ModuleType("flask"),
+        "yaml": types.ModuleType("yaml"),
+        "agent": types.ModuleType("agent"),
+        "agent.skill_utils": types.ModuleType("agent.skill_utils"),
+    }
+
+    flask_mod = stubs["flask"]
+    flask_mod.request = MagicMock()
+    flask_mod.g = MagicMock()
+    flask_mod.Blueprint = MagicMock(return_value=MagicMock())
+    flask_mod.jsonify = MagicMock(side_effect=lambda x: x)
+    flask_mod.abort = MagicMock()
+    flask_mod.current_app = MagicMock()
+
+    stubs["yaml"].safe_load = MagicMock(return_value=None)
+
+    su = stubs["agent.skill_utils"]
+    su.iter_skill_index_files = lambda skills_dir, filename: skills_dir.rglob(filename)
+    su.parse_frontmatter = MagicMock(return_value=({}, ""))
+    su.skill_matches_platform = MagicMock(return_value=True)
+
+    for name, mod in stubs.items():
+        sys.modules.setdefault(name, mod)
+
+    mod_name = "api.profiles"
+    if mod_name in sys.modules:
+        del sys.modules[mod_name]
+
+    api_pkg = types.ModuleType("api")
+    sys.modules["api"] = api_pkg
+
+    import importlib.util
+    spec_path = Path(__file__).parent.parent / "api" / "profiles.py"
+    spec = importlib.util.spec_from_file_location(mod_name, spec_path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[mod_name] = mod
+    try:
+        spec.loader.exec_module(mod)
+    except Exception:
+        pass
+    return mod
+
+
+@pytest.fixture()
+def mod(tmp_path):
+    try:
+        m = _make_profiles_module()
+        assert hasattr(m, "_get_profile_skills_stats")
+        assert hasattr(m, "_skills_stats_lock_for")
+    except Exception:
+        pytest.skip("api.profiles not importable in this environment")
+    # Reset both caches + the per-profile lock registry for isolation.
+    if hasattr(m, "_SKILLS_STATS_CACHE"):
+        m._SKILLS_STATS_CACHE.clear()
+    if hasattr(m, "_SKILLS_STATS_LOCKS"):
+        m._SKILLS_STATS_LOCKS.clear()
+    if hasattr(m, "_LIST_PROFILES_CACHE"):
+        m._LIST_PROFILES_CACHE = None
+    yield m
+    saved = {k: v for k, v in list(sys.modules.items())
+             if k == "api" or k.startswith("api.")}
+    for k in list(saved):
+        sys.modules.pop(k, None)
+
+
+# ---------------------------------------------------------------------------
+# 1. Per-profile compute lock collapses a concurrent-miss herd to one compute
+# ---------------------------------------------------------------------------
+
+class TestConcurrentMissComputesOnce:
+    def test_concurrent_cache_miss_computes_exactly_once(self, mod, tmp_path):
+        profile_dir = tmp_path / "p"
+        (profile_dir / "skills").mkdir(parents=True)
+
+        compute_calls = []
+        compute_lock = threading.Lock()
+
+        def _slow_compute(_pdir):
+            with compute_lock:
+                compute_calls.append(1)
+            # Widen the race window so every thread would pile in without a lock.
+            time.sleep(0.15)
+            return (3, 5)
+
+        fixed_mtime = 1_700_000_000_000_000_000
+        results = []
+        results_lock = threading.Lock()
+
+        n_threads = 24
+        barrier = threading.Barrier(n_threads)
+
+        def _worker():
+            barrier.wait()  # release all threads simultaneously (cold-boot burst)
+            r = mod._get_profile_skills_stats(profile_dir)
+            with results_lock:
+                results.append(r)
+
+        with (
+            patch.object(mod, "_compute_profile_skills_stats", side_effect=_slow_compute),
+            patch.object(mod, "_skill_tree_max_mtime_ns", return_value=fixed_mtime),
+        ):
+            threads = [threading.Thread(target=_worker) for _ in range(n_threads)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+
+        assert sum(compute_calls) == 1, (
+            "concurrent cache misses on ONE profile must collapse to a single "
+            f"_compute_profile_skills_stats call, got {sum(compute_calls)}"
+        )
+        assert len(results) == n_threads
+        assert all(r == (3, 5) for r in results), \
+            "every waiting thread must see the single shared computed result"
+
+    def test_distinct_profiles_compute_in_parallel(self, mod, tmp_path):
+        """Different profiles must NOT serialize on each other (per-profile lock,
+        not a single global lock) — independent trees compute concurrently."""
+        n = 4
+        dirs = []
+        for i in range(n):
+            d = tmp_path / f"p{i}"
+            (d / "skills").mkdir(parents=True)
+            dirs.append(d)
+
+        in_compute = []
+        max_concurrent = [0]
+        gate = threading.Lock()
+
+        def _slow_compute(_pdir):
+            with gate:
+                in_compute.append(1)
+                max_concurrent[0] = max(max_concurrent[0], sum(in_compute))
+            time.sleep(0.15)
+            with gate:
+                in_compute.pop()
+            return (1, 1)
+
+        barrier = threading.Barrier(n)
+
+        def _worker(pdir):
+            barrier.wait()
+            mod._get_profile_skills_stats(pdir)
+
+        # A single patched probe returns per-path deterministic mtimes so each
+        # profile is a distinct cache key (distinct lock).
+        path_mtimes = {Path(d).resolve(): 1_700_000_000_000_000_000 + i
+                       for i, d in enumerate(dirs)}
+
+        def _probe(skills_dir, config_path):
+            return path_mtimes.get(Path(skills_dir).parent.resolve(), 0)
+
+        with (
+            patch.object(mod, "_compute_profile_skills_stats", side_effect=_slow_compute),
+            patch.object(mod, "_skill_tree_max_mtime_ns", side_effect=_probe),
+        ):
+            threads = [threading.Thread(target=_worker, args=(d,)) for d in dirs]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+
+        assert max_concurrent[0] >= 2, (
+            "distinct profiles must be able to compute concurrently — a single "
+            "global lock would force max_concurrent==1"
+        )
+
+
+# ---------------------------------------------------------------------------
+# 2. Lock registry returns a stable per-profile lock
+# ---------------------------------------------------------------------------
+
+class TestLockRegistry:
+    def test_same_profile_returns_same_lock(self, mod, tmp_path):
+        d = (tmp_path / "p").resolve()
+        l1 = mod._skills_stats_lock_for(d)
+        l2 = mod._skills_stats_lock_for(d)
+        assert l1 is l2
+
+    def test_distinct_profiles_get_distinct_locks(self, mod, tmp_path):
+        a = (tmp_path / "a").resolve()
+        b = (tmp_path / "b").resolve()
+        assert mod._skills_stats_lock_for(a) is not mod._skills_stats_lock_for(b)
+
+
+# ---------------------------------------------------------------------------
+# 3. list_profiles_api single-flights the row build under a concurrent burst
+# ---------------------------------------------------------------------------
+
+class TestListProfilesSingleFlight:
+    def test_concurrent_list_profiles_builds_rows_once(self, mod):
+        build_calls = []
+        build_lock = threading.Lock()
+
+        def _slow_build():
+            with build_lock:
+                build_calls.append(1)
+            time.sleep(0.15)
+            return [{"name": "default", "path": "/x"}]
+
+        n_threads = 16
+        barrier = threading.Barrier(n_threads)
+        results = []
+        results_lock = threading.Lock()
+
+        def _worker():
+            barrier.wait()
+            r = mod.list_profiles_api()
+            with results_lock:
+                results.append(r)
+
+        with (
+            patch.object(mod, "_is_isolated_profile_mode", return_value=False),
+            patch.object(mod, "get_active_profile_name", return_value="default"),
+            patch.object(mod, "_build_profile_rows_fast", side_effect=_slow_build),
+        ):
+            mod._LIST_PROFILES_CACHE = None
+            threads = [threading.Thread(target=_worker) for _ in range(n_threads)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+
+        assert sum(build_calls) == 1, (
+            "a concurrent cold-start burst must build the profile rows exactly "
+            f"once (single-flight), got {sum(build_calls)}"
+        )
+        assert len(results) == n_threads
+        assert all(r and r[0]["name"] == "default" for r in results)
+
+
+# ---------------------------------------------------------------------------
+# 4. #4783 contract preserved: the cheap mtime probe still runs on every call
+# ---------------------------------------------------------------------------
+
+class TestProbeStillRunsEveryCall:
+    def test_probe_runs_on_cache_hit(self, mod, tmp_path):
+        profile_dir = tmp_path / "p"
+        (profile_dir / "skills").mkdir(parents=True)
+
+        with (
+            patch.object(mod, "_compute_profile_skills_stats",
+                         wraps=mod._compute_profile_skills_stats) as mock_compute,
+            patch.object(mod, "_skill_tree_max_mtime_ns",
+                         wraps=mod._skill_tree_max_mtime_ns) as mock_probe,
+        ):
+            mod._get_profile_skills_stats(profile_dir)
+            compute_after_first = mock_compute.call_count
+            probe_after_first = mock_probe.call_count
+
+            mod._get_profile_skills_stats(profile_dir)  # within TTL, unchanged
+
+            assert mock_compute.call_count == compute_after_first, \
+                "expensive compute must be skipped within TTL when unchanged"
+            assert mock_probe.call_count > probe_after_first, \
+                "cheap mtime probe MUST still run on every call (#4783 contract)"