Introduce typed XSkill config

src/xskill/__init__.py

@@ -22,6 +22,7 @@
 
 # 顶级公开面：3 个核心类
 from xskill.core import XSkill
+from xskill.config import XSkillConfig
 from xskill.skill.skill import Skill
 from xskill.pipeline.trajectory import Trajectory
 
@@ -30,6 +31,6 @@
 from xskill.skill.repo import SkillRepo
 
 __all__ = [
-    "XSkill", "Skill", "Trajectory",
+    "XSkill", "XSkillConfig", "Skill", "Trajectory",
     "Registry", "SkillRepo",
 ]

src/xskill/agents/agent_tools.py

@@ -11,12 +11,17 @@
 
 from __future__ import annotations
 
+# ruff: noqa: BLE001,S110
+
 import json, logging
+from collections.abc import Mapping
 from datetime import date, datetime
 from pathlib import Path
+from typing import Any
 
 from agno.tools import tool
 
+from xskill.config import XSkillConfig
 from xskill.skill.frontmatter import (
     parse as fm_parse,
     parse_strict as fm_parse_strict,
@@ -33,7 +38,7 @@ class AgentToolConfig:
     def __init__(self):
         self._skill_dir: Path | None = None
         self._data_dir: Path | None = None
-        self._config: dict = {}
+        self._config: XSkillConfig | Mapping[str, Any] | None = None
         self._atom_skill_dir: Path | None = None
         self._atom_store = None
         self._default_traj_root: Path | None = None
@@ -65,7 +70,7 @@ def snapshot(self) -> dict:
     def restore(self, snapshot: dict) -> None:
         self._skill_dir = snapshot.get("skill_dir")
         self._data_dir = snapshot.get("data_dir")
-        self._config = snapshot.get("config") or {}
+        self._config = snapshot.get("config")
         self._atom_skill_dir = snapshot.get("atom_skill_dir")
         self._atom_store = snapshot.get("atom_store")
         self._default_traj_root = snapshot.get("default_traj_root")
@@ -84,8 +89,8 @@ def data_dir(self) -> Path | None:
         return self._data_dir
 
     @property
-    def config(self) -> dict:
-        return self._config or {}
+    def config(self) -> XSkillConfig | Mapping[str, Any] | None:
+        return self._config
 
     @property
     def atom_skill_dir(self) -> Path | None:
@@ -157,21 +162,6 @@ def _sanitize_frontmatter_dates(fm: dict) -> dict:
     return fm
 
 
-def _read_skill_md(skill_path: Path) -> tuple[dict, str, Path]:
-    """Return (frontmatter_dict, body, path_of_SKILL.md). Supports legacy
-    lowercase `skill.md` as a fallback read path (writes always go to
-    SKILL.md)."""
-    upper = skill_path / "SKILL.md"
-    lower = skill_path / "skill.md"
-    if upper.exists():
-        fm, body = fm_parse(upper.read_text(encoding="utf-8"))
-        return fm, body, upper
-    if lower.exists():
-        fm, body = fm_parse(lower.read_text(encoding="utf-8"))
-        return fm, body, lower
-    return {}, "", upper
-
-
 # ═══════════════════════════════════════════════════════════════════
 # Read tools
 # ═══════════════════════════════════════════════════════════════════
@@ -268,7 +258,7 @@ def read_file(path: str) -> str:
 @tool(name="create_skill")
 def create_skill(skill_name: str) -> str:
     """
-    Scaffold a new skill directory in the v2 layout.
+    Scaffold a new skill directory.
 
     Creates:
         ./skill/<name>/SKILL.md          (stub frontmatter + placeholder body)
@@ -302,7 +292,7 @@ def create_skill(skill_name: str) -> str:
     logger.info(f"📁 created skill scaffold: {target}")
     return (f"created: {target}\n"
             f"files: SKILL.md (stub), scripts/.gitkeep, references/.gitkeep\n"
-            f"Next: overwrite {target}/SKILL.md with your full v2 content via write_file.")
+            f"Next: overwrite {target}/SKILL.md with full content via write_file.")
 
 
 # ═══════════════════════════════════════════════════════════════════
@@ -404,9 +394,8 @@ def list_candidates(skill_name: str) -> str:
 def write_file(path: str, content: str) -> str:
     """Write or overwrite a file under ./skill/ only.
 
-    v2 行为：只做路径安全 + frontmatter 日期消毒。旧 v1 ``source_trajs ≥ 3``
-    gate 和 ``N/M 条轨迹`` warning 消毒已删——v2 用 ``source_atoms`` 引用 atom
-    而非 traj，且质量保障靠 candidates buffer 累计 weightscore ≥ 10 的硬门槛，
+    只做路径安全 + frontmatter 日期消毒。SKILL.md 用 ``source_atoms`` 引用
+    atom 而非 traj，且质量保障靠 candidates buffer 累计 weightscore ≥ 10 的硬门槛，
     不需要 SKILL.md 写入端再卡一道。
     """
     p = Path(path)
@@ -487,7 +476,15 @@ def update_frontmatter_metadata(skill_name: str, source_trajs: list[str] | None
     if not target.exists():
         return f"error: skill directory not found ({skill_name})"
 
-    fm, body, path = _read_skill_md(target)
+    path = target / "SKILL.md"
+    lower_path = target / "skill.md"
+    if path.exists():
+        fm, body = fm_parse(path.read_text(encoding="utf-8"))
+    elif lower_path.exists():
+        path = lower_path
+        fm, body = fm_parse(path.read_text(encoding="utf-8"))
+    else:
+        fm, body = {}, ""
     meta = fm.setdefault("metadata", {})
 
     # source_trajs union
@@ -508,7 +505,8 @@ def update_frontmatter_metadata(skill_name: str, source_trajs: list[str] | None
 
     # LLM-generated 2-sentence summary (for embeddings)
     from xskill.utils.llm import create_llm_client
-    llm_client = create_llm_client(agent_tool_config.config)
+    runtime_config = agent_tool_config.config
+    llm_client = create_llm_client(runtime_config) if runtime_config is not None else None
     if llm_client:
         skill_text = (fm.get("description", "") + "\n\n" + body)[:4000]
         try:
@@ -545,7 +543,7 @@ def update_frontmatter_metadata(skill_name: str, source_trajs: list[str] | None
 
 
 # ═══════════════════════════════════════════════════════════════════
-# AtomTask-era tools (v2) — consumed by TaskClusterAgent / SkillEditAgent
+# AtomTask tools — consumed by TaskClusterAgent / SkillEditAgent
 # ═══════════════════════════════════════════════════════════════════
 
 @tool(name="atom_task_read")
@@ -606,7 +604,7 @@ def read_traj(traj_id: str, offset_start: int, offset_end: int) -> str:
 
 @tool(name="new_skill_folder")
 def new_skill_folder(skill_name: str, description: str) -> str:
-    """v2: 创建 skill 目录 → git init → checkout baby 分支 → 首次 commit
+    """创建 skill 目录 → git init → checkout baby 分支 → 首次 commit
     （含 stub SKILL.md + .gitignore）。
 
     description 必填，落到 stub SKILL.md 的 frontmatter 中。后续：
@@ -648,7 +646,7 @@ def skill_read(skill_name: str) -> str:
 
 @tool(name="add_task_to_skill")
 def add_task_to_skill(skill_name: str, atom_id: str, weightscore: int) -> str:
-    """v2.1: 把 atom 加进 skill 的 candidates buffer。
+    """把 atom 加进 skill 的 candidates buffer。
 
     同 atom 重复 add 时**覆盖**（不累加，cluster 可改主意）。返回末尾附该
     atom 的 weightscore + buffer 总分 / 10，让 agent 看到"还差多少到阈值"。
@@ -867,8 +865,10 @@ def _run_description_optimization(target: Path, slug: str) -> None:
     （退回 agent 写的 description 继续提交）。LLM/embed 客户端在这个确定性
     workflow 内从 config 创建，不从 agent tool context 借对象。
     """
-    from xskill.config import get_config
-    config = agent_tool_config.config or get_config()
+    config = agent_tool_config.config
+    if config is None:
+        logger.warning("skip description_opt: agent tool config not initialized")
+        return
     if not (config.get("skill_opt", {}) or {}).get("enabled", True):
         return
     from xskill.utils.llm import create_embed_client, create_llm_client
@@ -1076,7 +1076,7 @@ def absorb_user_edit_to_main(skill_name: str, message: str) -> str:
 
 
 # ═══════════════════════════════════════════════════════════════════
-# v2.2 渐进收敛工具（ClusterAgent 用，处理近义 slug 整合）
+# 渐进收敛工具（ClusterAgent 用，处理近义 slug 整合）
 # ═══════════════════════════════════════════════════════════════════
 
 @tool(name="rename_skill")

src/xskill/agents/agno_factory.py

@@ -14,11 +14,15 @@
 """
 from __future__ import annotations
 
+# ruff: noqa: BLE001,S110
+
 import inspect
 import logging
 import os
+from collections.abc import Mapping
 from typing import Any, Callable
 
+from xskill.config import XSkillConfig
 from xskill.utils.logging import StreamLog
 from xskill.utils.llm import _ssl_verify
 
@@ -52,12 +56,13 @@ def _inject_verify_off_if_requested(model_cls, model_kwargs: dict,
             model_kwargs[name] = async_client
             injected.append(name)
             break
-    msg_log = log or (lambda *a, **kw: None)
     if injected:
-        msg_log(f"T2S_SSL_VERIFY=false → {model_cls.__name__} 注入 "
+        if log:
+            log(f"T2S_SSL_VERIFY=false → {model_cls.__name__} 注入 "
                 f"{'+'.join(injected)} (verify=False)", "step")
     else:
-        msg_log(f"T2S_SSL_VERIFY=false 但 {model_cls.__name__} 不接受 http_client "
+        if log:
+            log(f"T2S_SSL_VERIFY=false 但 {model_cls.__name__} 不接受 http_client "
                 f"kwarg，改用 SSL_CERT_FILE=/path/to/ca.pem", "error")
 
 
@@ -135,7 +140,7 @@ def rate_limited_invoke(messages, **kwargs):
     return model
 
 
-def build_chat_model(llm_cfg: dict, log: StreamLog | None = None):
+def build_chat_model(llm_cfg: Mapping[str, Any], log: StreamLog | None = None):
     """根据 ``llm_cfg.base_url`` 路由到合适的 agno model 类。
 
     为什么不一律用 ``OpenAIChat``：DeepSeek 直连（``api.deepseek.com``）的
@@ -280,7 +285,9 @@ def traced_invoke(messages, **kwargs):
     return model
 
 
-def make_default_factory(config: dict) -> Callable[..., Any]:
+def make_default_factory(
+    config: XSkillConfig | Mapping[str, Any],
+) -> Callable[..., Any]:
     """生产环境的 agno Agent 工厂。
 
     返回的 callable 签名 ``(*, instructions, tools) -> agno.agent.Agent``，
@@ -292,9 +299,16 @@ def make_default_factory(config: dict) -> Callable[..., Any]:
     """
     from agno.agent import Agent
 
-    base_cfg = config.get("llm", {}) or {}
-    override_cfg = config.get("llm_skill", {}) or {}
-    llm_cfg = {**base_cfg, **{k: v for k, v in override_cfg.items() if v}}
+    if isinstance(config, XSkillConfig):
+        base_cfg = config.llm_config
+        override_cfg = config.llm_skill_config
+    else:
+        base_cfg = config.get("llm", {}) or {}
+        override_cfg = config.get("llm_skill", {}) or {}
+    llm_cfg = {
+        **base_cfg,
+        **{config_key: value for config_key, value in override_cfg.items() if value},
+    }
 
     def factory(*, instructions, tools, **kwargs):
         model = build_chat_model(llm_cfg)

src/xskill/api/app.py

@@ -10,6 +10,8 @@
 
 from __future__ import annotations
 
+# ruff: noqa: BLE001
+
 # Upgrade sqlite3 to support RETURNING clause (needed by Agno session DB)
 import sys as _sys
 try:
@@ -20,15 +22,16 @@
 
 import logging
 import tempfile
+from collections.abc import Mapping
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 from fastapi import APIRouter, FastAPI, HTTPException
 from fastapi.responses import FileResponse
 from pydantic import BaseModel, Field
 
 from xskill import __version__
-from xskill.config import load_config, get_skill_dir
+from xskill.config import XSkillConfig, load_config
 from xskill.utils.search import search as search_trajs, search_all as search_trajs_all
 from xskill.skill.repo import (
     import_skill,
@@ -65,7 +68,7 @@
 # server 启动路径首次调用时填充。endpoints 在 startup hook 之后才被 hit，
 # 拿到的就是非 None；测试如果只 import ``_exec_tool`` / 常量，模块加载阶段
 # 完全不读 config。
-_config: dict | None = None
+_config: XSkillConfig | Mapping[str, Any] | None = None
 _skill_dir: Path | None = None
 _watcher_ref: dict = {}  # {"instance": DirectoryWatcher} — set in create_app startup
 
@@ -83,17 +86,35 @@ def _home_root() -> Path:
     return _home_root_override if _home_root_override is not None else Path.home()
 
 
-def _ensure_loaded() -> None:
+def _ensure_loaded(
+    config: XSkillConfig | Mapping[str, Any] | None = None,
+) -> None:
     """幂等：第一次调用时载入配置 + 解析关键目录，之后是 no-op。
 
     server 内部的 endpoint / startup / chat 等代码路径都通过模块级
     ``_config`` / ``_skill_dir`` 等访问，这里只负责把 None 占位填上。
     """
     global _config, _skill_dir
+    if config is not None:
+        _config = config
+        if isinstance(config, XSkillConfig):
+            _skill_dir = config.skill_dir
+        elif config.get("skill_dir"):
+            _skill_dir = Path(str(config["skill_dir"])).expanduser()
+        elif _skill_dir is None:
+            _skill_dir = Path("~/.xskill/skill").expanduser()
+        return
     if _config is not None:
+        if _skill_dir is None:
+            if isinstance(_config, XSkillConfig):
+                _skill_dir = _config.skill_dir
+            elif _config.get("skill_dir"):
+                _skill_dir = Path(str(_config["skill_dir"])).expanduser()
+            else:
+                _skill_dir = Path("~/.xskill/skill").expanduser()
         return
     _config = load_config()
-    _skill_dir = get_skill_dir()
+    _skill_dir = _config.skill_dir
 
 # ---------------------------------------------------------------------------
 # Pydantic request / response models
@@ -774,6 +795,7 @@ async def api_reindex():
 # ---------------------------------------------------------------------------
 
 def create_app(home_root: Path | str | None = None,
+               config: XSkillConfig | Mapping[str, Any] | None = None,
                *, team_server: bool = False) -> FastAPI:
     """Build the FastAPI app. Calls ``_ensure_loaded`` first so all module-level
     config globals (``_config``/``_skill_dir``/...) are populated before any
@@ -790,7 +812,7 @@ def create_app(home_root: Path | str | None = None,
     global _home_root_override
     if home_root is not None:
         _home_root_override = Path(home_root).expanduser().resolve()
-    _ensure_loaded()
+    _ensure_loaded(config)
     """Create and configure the FastAPI application."""
     app = FastAPI(
         title="xskill",
@@ -1188,15 +1210,21 @@ def _ensure_ingesters_for_detected_ecosystems():
                 from xskill.team.server.state import ensure_join_token
                 from xskill.config import (
                     get_team_clients_db_path, get_team_server_state_path,
-                    get_team_trajectories_dir,
+                    XSKILL_HOME,
                 )
                 from xskill.pipeline.registry import register_dir as _register_dir
                 from xskill.canary import CanaryConfig
 
                 join_token = ensure_join_token(get_team_server_state_path())
                 client_registry = ClientRegistry(get_team_clients_db_path())
-                traj_root = get_team_trajectories_dir()
                 team_cfg = _config.get("team", {}).get("server", {})
+                traj_root = Path(
+                    str(
+                        team_cfg.get("traj_root")
+                        or XSKILL_HOME / "team_trajectories"
+                    )
+                ).expanduser()
+                traj_root.mkdir(parents=True, exist_ok=True)
                 canary_cfg = CanaryConfig.from_dict(_config.get("canary", {}))
 
                 def _team_register_dir(path, label):