perf(dashboard): collapse per-request full table scans on my page

src/xskill/dashboard/console.py

@@ -44,6 +44,28 @@
 _reco_trigger_cache = SingleFlightTtlCache(
     ttl_seconds=_RECO_TRIGGER_TTL_SECONDS, max_entries=32)
 
+_ADOPTION_ROWS_TTL_SECONDS = 5.0
+_adoption_rows_cache = SingleFlightTtlCache(
+    ttl_seconds=_ADOPTION_ROWS_TTL_SECONDS, max_entries=8)
+
+
+def _adoption_rows(db_path: Optional[Path]) -> list[dict]:
+    """全量 atom_adoption 行（5s TTL + 单飞）。
+
+    贡献归属判定要在 Python 侧按 atom_id 内嵌 traj_id 做包含匹配，无法下推
+    SQL；每请求整表搬运随 adoption 体量线性放大。短窗缓存把一波请求收敛成
+    一次扫描。返回共享只读 list，调用方逐条 dict() 后再改。
+    """
+    def build() -> list[dict]:
+        with pooled_connection(db_path) as conn:
+            return [
+                dict(r) for r in conn.execute(
+                    "SELECT atom_id, skill, weightscore FROM atom_adoption",
+                ).fetchall()
+            ]
+
+    return _adoption_rows_cache.get_or_build(str(db_path or ""), build)
+
 
 def _team_ctx():
     from xskill.team.server.api import team_context
@@ -472,9 +494,7 @@ def my_contributions(ident=Depends(require_user)):
                     (user,),
                 ).fetchall()
             }
-            adoption = conn.execute(
-                "SELECT atom_id, skill, weightscore FROM atom_adoption"
-            ).fetchall()
+            adoption = _adoption_rows(db_path)
         # atom_id 内嵌 traj_id（atom_<traj_id>_NNNN）——按包含判定归属
         my_adopted = [dict(a) for a in adoption
                       if any(stem in (a["atom_id"] or "") for stem in my_stems)]

src/xskill/dashboard/explore.py

@@ -16,7 +16,11 @@
     dashboard_visible_trajectory_sql,
     pooled_connection,
 )
-from xskill.dashboard.metrics import _resolve_local_root, load_usage_records
+from xskill.dashboard.metrics import (
+    SingleFlightTtlCache,
+    _resolve_local_root,
+    load_usage_records,
+)
 
 
 class TrajExplorer:
@@ -141,6 +145,42 @@ def _atom_destinations(self, atom_id: str) -> list[dict]:
         return out
 
 
+_VISIBLE_TRAJ_TTL_SECONDS = 5.0
+_visible_traj_cache = SingleFlightTtlCache(
+    ttl_seconds=_VISIBLE_TRAJ_TTL_SECONDS, max_entries=8)
+
+
+def _visible_traj_info(db_path: Optional[Path]) -> dict[str, dict]:
+    """可见轨迹 stem → {user, model, root}（5s TTL + 单飞）。
+
+    本映射要全量 JOIN ``trajectories × watch_dirs``；skill_lineage 被
+    「我的」页贡献图按 skill 逐个调用，逐调逐扫会随轨迹量线性放大。
+    短窗缓存把一页 N 个 skill 的请求收敛成一次扫描。返回共享只读 dict。
+    """
+    from xskill.config import get_registry_db_path
+
+    def build() -> dict[str, dict]:
+        with pooled_connection(db_path) as conn:
+            wd_rows = conn.execute(
+                "SELECT t.filename fn, t.source_model model, w.label label,"
+                " w.path wpath FROM trajectories t"
+                " JOIN watch_dirs w ON t.watch_dir_id=w.id"
+                f" WHERE {dashboard_visible_trajectory_sql('t', 'w')}"
+            ).fetchall()
+        db_dir = (
+            Path(db_path).parent if db_path else get_registry_db_path().parent
+        )
+        out: dict[str, dict] = {}
+        for r in wd_rows:
+            stem = r["fn"][:-3] if r["fn"].endswith(".md") else r["fn"]
+            out[stem] = {"user": r["label"] or "(local)",
+                         "model": r["model"] or "",
+                         "root": _resolve_local_root(r["wpath"], db_dir)}
+        return out
+
+    return _visible_traj_cache.get_or_build(str(db_path or ""), build)
+
+
 def skill_lineage(skill_dir: Path, name: str,
                   db_path: Optional[Path] = None) -> dict:
     """skill 血缘（图①下半区）：贡献原子（adoption 事件 + 在途 candidates）
@@ -153,20 +193,7 @@ def skill_lineage(skill_dir: Path, name: str,
         adoption = conn.execute(
             "SELECT atom_id, weightscore, ts FROM atom_adoption WHERE skill=?"
             " ORDER BY ts", (name,)).fetchall()
-        wd_rows = conn.execute(
-            "SELECT t.filename fn, t.source_model model, w.label label,"
-            " w.path wpath FROM trajectories t"
-            " JOIN watch_dirs w ON t.watch_dir_id=w.id"
-            f" WHERE {dashboard_visible_trajectory_sql('t', 'w')}"
-        ).fetchall()
-    from xskill.config import get_registry_db_path
-    db_dir = Path(db_path).parent if db_path else get_registry_db_path().parent
-    traj_info: dict[str, dict] = {}
-    for r in wd_rows:
-        stem = r["fn"][:-3] if r["fn"].endswith(".md") else r["fn"]
-        traj_info[stem] = {"user": r["label"] or "(local)",
-                           "model": r["model"] or "",
-                           "root": _resolve_local_root(r["wpath"], db_dir)}
+    traj_info = _visible_traj_info(db_path)
     from xskill.skill.candidates import load_candidates
     from xskill.dashboard.metrics import _traj_of_atom
     entries: dict[str, dict] = {}

src/xskill/events.py

@@ -26,6 +26,8 @@
 
 import json
 import logging
+import threading
+import time
 from pathlib import Path
 from typing import Optional
 
@@ -59,15 +61,36 @@ def _traj_of_atom(atom_id: str) -> str:
     return stem if stem and idx.isdigit() else ""
 
 
+_TRAJ_USER_TTL_SECONDS = 5.0
+_traj_user_cache: dict[str, tuple[float, dict[str, str]]] = {}
+_traj_user_lock = threading.Lock()
+
+
 def _traj_user_map(db_path: Optional[Path] = None) -> dict[str, str]:
-    """filename stem → user_key。调用方批量算贡献人时只建一次。"""
-    with pooled_connection(db_path) as conn:
-        return {
-            (r["filename"][:-3] if r["filename"].endswith(".md")
-             else r["filename"]): (r["user_key"] or "")
-            for r in conn.execute(
-                "SELECT filename, user_key FROM trajectories").fetchall()
-        }
+    """filename stem → user_key（5s TTL + 单飞）。
+
+    本映射要全表扫 ``trajectories``，而「我的」页一次首屏会经多个端点
+    各自调到这里；短窗缓存把一波请求收敛成一次扫描，锁内构建保证到期
+    瞬间只有一个线程真扫。返回的是缓存内共享 dict，调用方只读不改写。
+    """
+    key = str(db_path or "")
+    with _traj_user_lock:
+        hit = _traj_user_cache.get(key)
+        if hit and hit[0] > time.monotonic():
+            return hit[1]
+        if len(_traj_user_cache) > 8:
+            _traj_user_cache.clear()
+        with pooled_connection(db_path) as conn:
+            mapping = {
+                (r["filename"][:-3] if r["filename"].endswith(".md")
+                 else r["filename"]): (r["user_key"] or "")
+                for r in conn.execute(
+                    "SELECT filename, user_key FROM trajectories").fetchall()
+            }
+        _traj_user_cache[key] = (
+            time.monotonic() + _TRAJ_USER_TTL_SECONDS, mapping,
+        )
+        return mapping
 
 
 def skill_contributors(skill: str, *, min_weight: int = CONTRIBUTOR_MIN_WEIGHT,

src/xskill/pipeline/registry.py

@@ -185,6 +185,8 @@ def dashboard_visible_trajectory_sql(
     was_new     INTEGER       -- 1=首次加入 0=覆盖
 );
 CREATE INDEX IF NOT EXISTS idx_atom_adopt ON atom_adoption(atom_id);
+-- 血缘/主贡献人按 skill 过滤（skill_lineage、skill_main_producers 的 IN 查询）
+CREATE INDEX IF NOT EXISTS idx_atom_adopt_skill ON atom_adoption(skill);
 
 CREATE TABLE IF NOT EXISTS canary_decision (
     id              INTEGER PRIMARY KEY AUTOINCREMENT,

tests/test_dashboard_my_scan_cache.py

@@ -0,0 +1,119 @@
+"""「我的」页全表扫收敛回归：三处 5s 单飞缓存 + atom_adoption(skill) 索引。
+
+a4 线上事故：我的页端点每请求各自全表扫 trajectories / atom_adoption，
+首屏叠出十余次全扫，SQLite gate 串行化拖慢全部面板接口。修复后同一波
+请求只扫一次（TTL 内共享只读结果），adoption 按 skill 过滤走索引。
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+from xskill import events
+from xskill.dashboard import console, explore
+from xskill.pipeline.registry import get_connection
+
+
+def _seed_db(db: Path) -> None:
+    conn = get_connection(db)
+    conn.execute(
+        "INSERT INTO watch_dirs(id,path,label,ecosystem)"
+        " VALUES(1,'/cc','alice','claude_code')",
+    )
+    conn.execute(
+        "INSERT INTO trajectories(watch_dir_id,filename,status,user_key)"
+        " VALUES(1,'traj_1.md','done','alice')",
+    )
+    conn.execute(
+        "INSERT INTO trajectories(watch_dir_id,filename,status,user_key)"
+        " VALUES(1,'traj_2.md','done','bob')",
+    )
+    conn.execute(
+        "INSERT INTO atom_adoption(atom_id,skill,weightscore,ts)"
+        " VALUES('atom_traj_1_0001','demo',3,'2026-07-30 00:00:00')",
+    )
+    conn.commit()
+    conn.close()
+
+
+def test_traj_user_map_cached_within_ttl(tmp_path):
+    db = tmp_path / "r.db"
+    _seed_db(db)
+    events._traj_user_cache.clear()
+    m1 = events._traj_user_map(db)
+    assert m1["traj_1"] == "alice"
+
+    conn = get_connection(db)
+    conn.execute(
+        "INSERT INTO trajectories(watch_dir_id,filename,status,user_key)"
+        " VALUES(1,'traj_3.md','done','carol')",
+    )
+    conn.commit()
+    conn.close()
+
+    m2 = events._traj_user_map(db)
+    assert m2 is m1            # TTL 内同一份共享对象，不重扫
+    assert "traj_3" not in m2  # 短窗内允许略陈旧
+
+    events._traj_user_cache.clear()
+    m3 = events._traj_user_map(db)
+    assert m3["traj_3"] == "carol"  # 清缓存后立刻可见
+
+
+def test_visible_traj_info_cached_and_correct(tmp_path):
+    db = tmp_path / "r.db"
+    _seed_db(db)
+    explore._visible_traj_cache.clear()
+    t1 = explore._visible_traj_info(db)
+    t2 = explore._visible_traj_info(db)
+    assert t1 is t2
+    assert t1["traj_1"]["user"] == "alice"
+    assert t1["traj_2"]["user"] == "alice"  # 同 watch_dir 的 label
+
+
+def test_adoption_rows_cached_and_fresh_after_clear(tmp_path):
+    db = tmp_path / "r.db"
+    _seed_db(db)
+    console._adoption_rows_cache.clear()
+    r1 = console._adoption_rows(db)
+    r2 = console._adoption_rows(db)
+    assert r1 is r2
+    assert r1[0]["skill"] == "demo"
+
+    conn = get_connection(db)
+    conn.execute(
+        "INSERT INTO atom_adoption(atom_id,skill,weightscore,ts)"
+        " VALUES('atom_traj_2_0001','demo2',1,'2026-07-30 00:00:01')",
+    )
+    conn.commit()
+    conn.close()
+    assert len(console._adoption_rows(db)) == 1  # 仍走缓存
+    console._adoption_rows_cache.clear()
+    assert len(console._adoption_rows(db)) == 2
+
+
+def test_atom_adoption_skill_index_exists(tmp_path):
+    db = tmp_path / "r.db"
+    _seed_db(db)
+    conn = get_connection(db)
+    names = {
+        r["name"] for r in conn.execute(
+            "SELECT name FROM sqlite_master"
+            " WHERE type='index' AND tbl_name='atom_adoption'",
+        )
+    }
+    conn.close()
+    assert "idx_atom_adopt_skill" in names
+
+
+def test_skill_lineage_correct_with_shared_traj_cache(tmp_path):
+    db = tmp_path / "r.db"
+    _seed_db(db)
+    explore._visible_traj_cache.clear()
+    skill_dir = tmp_path / "skills"
+    (skill_dir / "demo").mkdir(parents=True)
+
+    lin1 = explore.skill_lineage(skill_dir, "demo", db_path=db)
+    lin2 = explore.skill_lineage(skill_dir, "demo", db_path=db)
+    assert lin1["by_user"] == [{"user": "alice", "atoms": 1}]
+    assert lin1["atoms"][0]["atom_id"] == "atom_traj_1_0001"
+    assert lin2["by_user"] == lin1["by_user"]  # 缓存路径结果一致