From 357ae9001bdc9fdbd329d9a47ced06e151711ec0 Mon Sep 17 00:00:00 2001
From: 370025263 <370025263@qq.com>
Date: Wed, 1 Jul 2026 21:52:25 +0800
Subject: [PATCH 1/7] feat: introduce typed xskill config

---
 src/xskill/__init__.py     |   3 +-
 src/xskill/config.py       | 126 ++++++++++++++++++++++++++++++-------
 src/xskill/core.py         |  38 +++++++----
 src/xskill/utils/llm.py    |  80 ++++++++++++++---------
 tests/test_config_class.py | 101 +++++++++++++++++++++++++++++
 5 files changed, 281 insertions(+), 67 deletions(-)
 create mode 100644 tests/test_config_class.py

diff --git a/src/xskill/__init__.py b/src/xskill/__init__.py
index 5c5f9ab8..c1762662 100644
--- a/src/xskill/__init__.py
+++ b/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",
 ]
diff --git a/src/xskill/config.py b/src/xskill/config.py
index 11c1d75c..5ea8cd12 100644
--- a/src/xskill/config.py
+++ b/src/xskill/config.py
@@ -7,9 +7,12 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterator, Mapping
+from copy import deepcopy
+from dataclasses import dataclass, field
 import logging
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 import yaml
 
@@ -22,10 +25,102 @@
 CHAT_DB = XSKILL_HOME / "chat_sessions.db"
 LOGS_DIR = XSKILL_HOME / "logs"
 
-_config: dict = {}
+_config: "XSkillConfig | None" = None
 _overrides: dict = {}
 
 
+@dataclass
+class XSkillConfig(Mapping[str, Any]):
+    """Dict-compatible xskill configuration loaded from YAML."""
+
+    data: dict[str, Any] = field(default_factory=dict)
+    source_path: Path | None = None
+
+    @classmethod
+    def from_yaml(
+        class_type,
+        path: Optional[Path] = None,
+        *,
+        validate_required: bool = True,
+    ) -> "XSkillConfig":
+        config_path = Path(path) if path else CONFIG_PATH
+        if not config_path.exists():
+            raise FileNotFoundError(
+                f"xskill config not found: {config_path}\n"
+                f"Run `xskill serve` once to auto-create a template, "
+                f"or call config.ensure_config_exists()."
+            )
+        with open(config_path, encoding="utf-8") as config_file:
+            loaded_data = yaml.safe_load(config_file) or {}
+        return class_type.from_dict(
+            loaded_data,
+            source_path=config_path,
+            validate_required=validate_required,
+        )
+
+    @classmethod
+    def from_dict(
+        class_type,
+        values: Mapping[str, Any] | None,
+        *,
+        source_path: Optional[Path] = None,
+        validate_required: bool = True,
+    ) -> "XSkillConfig":
+        config_data = deepcopy(dict(values or {}))
+        config_object = class_type(
+            data=config_data,
+            source_path=Path(source_path) if source_path else None,
+        )
+        if validate_required:
+            config_object.validate_required()
+        return config_object
+
+    def __getitem__(self, key: str) -> Any:
+        return self.data[key]
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self.data)
+
+    def __len__(self) -> int:
+        return len(self.data)
+
+    def as_dict(self) -> dict[str, Any]:
+        return deepcopy(self.data)
+
+    def section(self, section_name: str) -> dict[str, Any]:
+        section_value = self.data.get(section_name) or {}
+        if not isinstance(section_value, Mapping):
+            raise TypeError(
+                f"{section_name} config section must be a mapping, "
+                f"got {type(section_value).__name__}"
+            )
+        return dict(section_value)
+
+    @property
+    def llm_config(self) -> dict[str, Any]:
+        return self.section("llm")
+
+    @property
+    def llm_skill_config(self) -> dict[str, Any]:
+        return self.section("llm_skill")
+
+    @property
+    def embedding_config(self) -> dict[str, Any]:
+        return self.section("embedding")
+
+    @property
+    def skill_dir(self) -> Path:
+        configured_path = self.data.get("skill_dir") or str(XSKILL_HOME / "skill")
+        return Path(str(configured_path)).expanduser()
+
+    def validate_required(self) -> None:
+        location = self.source_path or "provided config"
+        if not self.llm_config.get("api_key"):
+            raise KeyError(f"llm.api_key missing in {location}")
+        if not self.embedding_config.get("api_key"):
+            raise KeyError(f"embedding.api_key missing in {location}")
+
+
 def set_overrides(**kwargs):
     """CLI flag 覆盖。仅 debug / quiet 两个保留。"""
     for k, v in kwargs.items():
@@ -214,32 +309,20 @@ def ensure_config_exists(path: Optional[Path] = None) -> bool:
     return False
 
 
-def load_config(path: Optional[Path] = None) -> dict:
+def load_config(path: Optional[Path] = None) -> XSkillConfig:
     """加载 ~/.xskill/config.yaml；不存在直接抛 FileNotFoundError。
 
     正常路径下 CLI 会先调 ``ensure_config_exists`` auto-init，不会走到这个
     FileNotFoundError；保留它作为 SDK 直接调用时的 fail-loud 兜底。
     """
     global _config
-    cfg_path = Path(path) if path else CONFIG_PATH
-    if not cfg_path.exists():
-        raise FileNotFoundError(
-            f"xskill config not found: {cfg_path}\n"
-            f"Run `xskill serve` once to auto-create a template, "
-            f"or call config.ensure_config_exists()."
-        )
-    with open(cfg_path, encoding="utf-8") as f:
-        _config = yaml.safe_load(f) or {}
-    if not _config.get("llm", {}).get("api_key"):
-        raise KeyError(f"llm.api_key missing in {cfg_path}")
-    if not _config.get("embedding", {}).get("api_key"):
-        raise KeyError(f"embedding.api_key missing in {cfg_path}")
+    _config = XSkillConfig.from_yaml(path)
     return _config
 
 
-def get_config() -> dict:
-    if not _config:
-        load_config()
+def get_config() -> XSkillConfig:
+    if _config is None:
+        return load_config()
     return _config
 
 
@@ -334,9 +417,8 @@ def ingest_config(path: Optional[Path] = None) -> dict:
 
 def get_skill_dir() -> Path:
     """skill_dir: config.yaml 字段；默认 ~/.xskill/skill/"""
-    cfg = get_config()
-    raw = cfg.get("skill_dir") or str(XSKILL_HOME / "skill")
-    return Path(raw).expanduser()
+    config_object = get_config()
+    return config_object.skill_dir
 
 
 def get_logs_dir() -> Path:
diff --git a/src/xskill/core.py b/src/xskill/core.py
index 09046c74..f60f7cb5 100644
--- a/src/xskill/core.py
+++ b/src/xskill/core.py
@@ -7,10 +7,11 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
-from xskill.config import load_config, get_skill_dir
+from xskill.config import XSkillConfig, load_config
 from xskill.pipeline.registry import Registry
 from xskill.skill.repo import SkillRepo
 from xskill.pipeline.trajectory import Trajectory
@@ -40,27 +41,38 @@ class XSkill:
         xskill.skill_repo["fix-foo"]
     """
 
-    def __init__(self, config_path: Optional[Path] = None):
-        self.config = load_config(config_path)
+    def __init__(
+        self,
+        config_path: Optional[Path] = None,
+        config: XSkillConfig | Mapping[str, Any] | None = None,
+    ):
+        if config_path is not None and config is not None:
+            raise ValueError("config_path and config cannot both be provided")
+        if config is None:
+            self.config = load_config(config_path)
+        elif isinstance(config, XSkillConfig):
+            self.config = config
+        else:
+            self.config = XSkillConfig.from_dict(config)
         self.registry = Registry()
-        self.skill_repo = SkillRepo(get_skill_dir(), registry=self.registry)
-        self._llm = None
-        self._embed = None
+        self.skill_repo = SkillRepo(self.config.skill_dir, registry=self.registry)
+        self._llm_client = None
+        self._embed_client = None
 
     # ─── lazy LLM / embed clients ──────────────────────────────
     @property
     def llm(self):
-        if self._llm is None:
+        if self._llm_client is None:
             from xskill.utils.llm import create_llm_client
-            self._llm = create_llm_client(self.config)
-        return self._llm
+            self._llm_client = create_llm_client(self.config)
+        return self._llm_client
 
     @property
     def embed(self):
-        if self._embed is None:
+        if self._embed_client is None:
             from xskill.utils.llm import create_embed_client
-            self._embed = create_embed_client(self.config)
-        return self._embed
+            self._embed_client = create_embed_client(self.config)
+        return self._embed_client
 
     # ─── 检索（跨所有 registry）─────────────────────────────────
     def search_trajectories(self, query: str, top_k: int = 5,
diff --git a/src/xskill/utils/llm.py b/src/xskill/utils/llm.py
index 6eb68fc6..4b56a7d0 100644
--- a/src/xskill/utils/llm.py
+++ b/src/xskill/utils/llm.py
@@ -18,12 +18,15 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 import os, json, logging, time
 from dataclasses import dataclass, field
-from typing import Literal, Optional
+from typing import Any, Literal, Optional
 
 import numpy as np
 
+from xskill.config import XSkillConfig
+
 EmbedApiStyle = Literal["multimodal", "openai"]
 
 logger = logging.getLogger(__name__)
@@ -61,11 +64,11 @@ class LLMClient:
     _client: object = field(default=None, repr=False)
 
     @classmethod
-    def from_config(cls, cfg: dict) -> "LLMClient":
-        base_url = cfg.get("base_url", "").rstrip("/")
-        model = cfg.get("model", "")
+    def from_config(class_type, config_section: Mapping[str, Any]) -> "LLMClient":
+        base_url = config_section.get("base_url", "").rstrip("/")
+        model = config_section.get("model", "")
         api_key = (
-            cfg.get("api_key", "")
+            config_section.get("api_key", "")
             or os.environ.get("LLM_API_KEY", "")
             or os.environ.get("ANTHROPIC_API_KEY", "")
             or os.environ.get("OPENAI_API_KEY", "")
@@ -75,13 +78,13 @@ def from_config(cls, cfg: dict) -> "LLMClient":
         kwargs = dict(base_url=base_url, model=model, api_key=api_key)
         # 允许 config 覆盖 max_tokens / temperature（缺省则用 dataclass 默认）。
         # 之前这里不读 max_tokens，导致 yaml 配了也不生效。
-        if "max_tokens" in cfg:
-            kwargs["max_tokens"] = int(cfg["max_tokens"])
-        if "temperature" in cfg:
-            kwargs["temperature"] = float(cfg["temperature"])
-        if "rate_limit" in cfg:
-            kwargs["rate_limit_cfg"] = cfg["rate_limit"]
-        return cls(**kwargs)
+        if "max_tokens" in config_section:
+            kwargs["max_tokens"] = int(config_section["max_tokens"])
+        if "temperature" in config_section:
+            kwargs["temperature"] = float(config_section["temperature"])
+        if "rate_limit" in config_section:
+            kwargs["rate_limit_cfg"] = config_section["rate_limit"]
+        return class_type(**kwargs)
 
     def _get_client(self):
         if self._client is None:
@@ -203,22 +206,22 @@ class EmbedClient:
     _client: object = field(default=None, repr=False)
 
     @classmethod
-    def from_config(cls, cfg: dict) -> "EmbedClient":
-        base_url = cfg.get("base_url", "").rstrip("/")
-        model = cfg.get("model", "")
+    def from_config(class_type, config_section: Mapping[str, Any]) -> "EmbedClient":
+        base_url = config_section.get("base_url", "").rstrip("/")
+        model = config_section.get("model", "")
         api_key = (
-            cfg.get("api_key", "")
+            config_section.get("api_key", "")
             or os.environ.get("EMBED_API_KEY", "")
             or os.environ.get("OPENAI_API_KEY", "")
         )
-        dim = cfg.get("dim", 0)
+        dim = config_section.get("dim", 0)
         if not base_url or not model:
             raise ValueError("embedding.base_url 和 embedding.model 必须配置")
-        api_style = _resolve_embed_api_style(cfg, model)
-        inst = cls(
+        api_style = _resolve_embed_api_style(config_section, model)
+        instance = class_type(
             base_url=base_url, model=model, api_key=api_key, dim=dim, api_style=api_style,
         )
-        return inst
+        return instance
 
     def _get_session(self):
         if self._client is None:
@@ -320,20 +323,26 @@ def __repr__(self):
 # 工厂函数
 # ═══════════════════════════════════════════════════════════════════
 
-def create_embed_client(config: dict) -> "EmbedClient":
+def create_embed_client(config: Mapping[str, Any] | XSkillConfig) -> "EmbedClient":
     """根据配置创建 embedding 客户端，未配置或不可用时直接报错"""
-    embed_cfg = config.get("embedding", {})
+    if isinstance(config, XSkillConfig):
+        embed_config = config.embedding_config
+    else:
+        embed_config = config.get("embedding", {}) or {}
 
-    if not embed_cfg.get("base_url") or not embed_cfg.get("model"):
+    if not embed_config.get("base_url") or not embed_config.get("model"):
         raise ValueError("embedding.base_url 和 embedding.model 必须配置")
 
-    client = EmbedClient.from_config(embed_cfg)
+    client = EmbedClient.from_config(embed_config)
     client.probe_dim()
     logger.info(f"Embedding: {client}")
     return client
 
 
-def create_llm_client(config: dict, role: str = "default") -> "LLMClient | None":
+def create_llm_client(
+    config: Mapping[str, Any] | XSkillConfig,
+    role: str = "default",
+) -> "LLMClient | None":
     """根据配置创建 LLM 客户端，未配置返回 None。
 
     role:
@@ -350,16 +359,25 @@ def create_llm_client(config: dict, role: str = "default") -> "LLMClient | None"
         model: "doubao-seed-2-0-pro-260215"    # agent + eval 换大模型
         # base_url / api_key 缺省 → 继承 llm.*
     """
-    base_cfg = config.get("llm", {}) or {}
+    if isinstance(config, XSkillConfig):
+        base_config = config.llm_config
+    else:
+        base_config = config.get("llm", {}) or {}
     if role in ("skill", "eval"):
-        override_cfg = config.get("llm_skill", {}) or {}
-        merged = {**base_cfg, **{k: v for k, v in override_cfg.items() if v}}
+        if isinstance(config, XSkillConfig):
+            override_config = config.llm_skill_config
+        else:
+            override_config = config.get("llm_skill", {}) or {}
+        merged_config = {
+            **base_config,
+            **{key: value for key, value in override_config.items() if value},
+        }
     else:
-        merged = base_cfg
+        merged_config = base_config
 
-    if merged.get("base_url") and merged.get("model"):
+    if merged_config.get("base_url") and merged_config.get("model"):
         try:
-            client = LLMClient.from_config(merged)
+            client = LLMClient.from_config(merged_config)
             logger.info(f"LLM[{role}]: {client}")
             return client
         except Exception as e:
diff --git a/tests/test_config_class.py b/tests/test_config_class.py
new file mode 100644
index 00000000..9adf5dfa
--- /dev/null
+++ b/tests/test_config_class.py
@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from xskill import XSkillConfig
+from xskill.core import XSkill
+from xskill.utils.llm import EmbedClient, create_embed_client, create_llm_client
+
+
+def _sample_config(skill_directory: Path) -> dict:
+    return {
+        "skill_dir": str(skill_directory),
+        "llm": {
+            "base_url": "http://llm.example/v1",
+            "model": "chat-model",
+            "api_key": "llm-key",
+        },
+        "embedding": {
+            "base_url": "http://embedding.example/v1",
+            "model": "embedding-model",
+            "api_key": "embedding-key",
+            "dim": 3,
+        },
+    }
+
+
+def test_config_class_loads_yaml_and_exposes_sections(tmp_path):
+    skill_directory = tmp_path / "skill"
+    config_path = tmp_path / "config.yaml"
+    config_path.write_text(
+        "\n".join(
+            [
+                f"skill_dir: {skill_directory}",
+                "llm:",
+                "  base_url: http://llm.example/v1",
+                "  model: chat-model",
+                "  api_key: llm-key",
+                "embedding:",
+                "  base_url: http://embedding.example/v1",
+                "  model: embedding-model",
+                "  api_key: embedding-key",
+                "  dim: 3",
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    config_object = XSkillConfig.from_yaml(config_path)
+
+    assert config_object["llm"]["model"] == "chat-model"
+    assert config_object.llm_config["base_url"] == "http://llm.example/v1"
+    assert config_object.embedding_config["model"] == "embedding-model"
+    assert config_object.skill_dir == skill_directory
+
+
+def test_config_class_rejects_missing_required_keys(tmp_path):
+    config_values = _sample_config(tmp_path / "skill")
+    config_values["llm"]["api_key"] = ""
+
+    with pytest.raises(KeyError, match="llm.api_key missing"):
+        XSkillConfig.from_dict(config_values)
+
+
+def test_xskill_accepts_config_object(tmp_path, monkeypatch):
+    config_object = XSkillConfig.from_dict(_sample_config(tmp_path / "skill"))
+    registry_path = tmp_path / "registry.db"
+
+    monkeypatch.setattr("xskill.pipeline.registry.REGISTRY_DB", registry_path)
+
+    xskill_object = XSkill(config=config_object)
+
+    assert xskill_object.config is config_object
+    assert xskill_object.skill_repo.root == tmp_path / "skill"
+
+
+def test_xskill_rejects_config_and_path_together(tmp_path):
+    config_object = XSkillConfig.from_dict(_sample_config(tmp_path / "skill"))
+
+    with pytest.raises(ValueError, match="cannot both be provided"):
+        XSkill(config_path=tmp_path / "config.yaml", config=config_object)
+
+
+def test_llm_factory_accepts_config_class(tmp_path):
+    config_object = XSkillConfig.from_dict(_sample_config(tmp_path / "skill"))
+
+    llm_client = create_llm_client(config_object)
+
+    assert llm_client is not None
+    assert llm_client.model == "chat-model"
+
+
+def test_embed_factory_accepts_config_class(tmp_path, monkeypatch):
+    config_object = XSkillConfig.from_dict(_sample_config(tmp_path / "skill"))
+    monkeypatch.setattr(EmbedClient, "probe_dim", lambda self: self.dim)
+
+    embed_client = create_embed_client(config_object)
+
+    assert embed_client.model == "embedding-model"
+    assert embed_client.dim == 3

From 0eaf4ffec0fcef5aecdd91a3519800cd331fbca3 Mon Sep 17 00:00:00 2001
From: 370025263 <370025263@qq.com>
Date: Wed, 1 Jul 2026 21:55:02 +0800
Subject: [PATCH 2/7] fix: narrow llm config exception handling

---
 src/xskill/utils/llm.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/xskill/utils/llm.py b/src/xskill/utils/llm.py
index 4b56a7d0..91d8f394 100644
--- a/src/xskill/utils/llm.py
+++ b/src/xskill/utils/llm.py
@@ -19,7 +19,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-import os, json, logging, time
+import os, logging, time
 from dataclasses import dataclass, field
 from typing import Any, Literal, Optional
 
@@ -380,7 +380,7 @@ def create_llm_client(
             client = LLMClient.from_config(merged_config)
             logger.info(f"LLM[{role}]: {client}")
             return client
-        except Exception as e:
-            logger.warning(f"LLM[{role}] 初始化失败: {e}")
+        except (KeyError, TypeError, ValueError) as error:
+            logger.warning(f"LLM[{role}] 初始化失败: {error}")
             return None
     return None

From eced2e6b27d52495c722520f3365357bfad14cfc Mon Sep 17 00:00:00 2001
From: 370025263 <370025263@qq.com>
Date: Wed, 1 Jul 2026 22:20:39 +0800
Subject: [PATCH 3/7] thread typed config through runtime chain

---
 src/xskill/agents/agent_tools.py     | 22 ++++---
 src/xskill/agents/agno_factory.py    | 30 +++++++---
 src/xskill/api/app.py                | 34 +++++++++--
 src/xskill/core.py                   |  6 +-
 src/xskill/pipeline/runner.py        | 21 ++++++-
 tests/test_config_class.py           | 85 +++++++++++++++++++++++++++-
 tests/test_server_watcher_startup.py | 69 ++++++++++++++++++++++
 7 files changed, 241 insertions(+), 26 deletions(-)

diff --git a/src/xskill/agents/agent_tools.py b/src/xskill/agents/agent_tools.py
index 0303ea9d..97e89940 100644
--- a/src/xskill/agents/agent_tools.py
+++ b/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:
@@ -508,7 +513,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:
@@ -867,8 +873,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
diff --git a/src/xskill/agents/agno_factory.py b/src/xskill/agents/agno_factory.py
index 52cff757..ad61219b 100644
--- a/src/xskill/agents/agno_factory.py
+++ b/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)
diff --git a/src/xskill/api/app.py b/src/xskill/api/app.py
index ce35d960..fe5b6771 100644
--- a/src/xskill/api/app.py
+++ b/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",
diff --git a/src/xskill/core.py b/src/xskill/core.py
index f60f7cb5..e0eee128 100644
--- a/src/xskill/core.py
+++ b/src/xskill/core.py
@@ -186,7 +186,11 @@ def serve(self, host: str = "0.0.0.0", port: int = 8000,
         """
         import uvicorn
         from xskill.api import create_app
-        app = create_app(home_root=home_root, team_server=server_mode)
+        app = create_app(
+            home_root=home_root,
+            config=self.config,
+            team_server=server_mode,
+        )
         if server_mode:
             from xskill.team.server.state import ensure_join_token
             from xskill.config import get_team_server_state_path
diff --git a/src/xskill/pipeline/runner.py b/src/xskill/pipeline/runner.py
index 6f5f0c9d..83fa27bf 100644
--- a/src/xskill/pipeline/runner.py
+++ b/src/xskill/pipeline/runner.py
@@ -23,13 +23,18 @@
 
 from __future__ import annotations
 
+# ruff: noqa: BLE001
+
 import asyncio
 import logging
 import threading
 import time
+from collections.abc import Mapping
 from concurrent.futures import ThreadPoolExecutor, Future, as_completed
 from pathlib import Path
+from typing import Any
 
+from xskill.config import XSkillConfig
 from xskill.pipeline.registry import (
     list_watch_dirs,
     discover_trajectories,
@@ -80,7 +85,8 @@ class DirectoryWatcher:
       skill 的 ``.candidates.yml`` 时才 done（文件系统即队列，天然去重+断点续传）。
     """
 
-    def __init__(self, *, llm=None, embed_client=None, config=None,
+    def __init__(self, *, llm=None, embed_client=None,
+                 config: XSkillConfig | Mapping[str, Any] | None = None,
                  skill_dir=None, poll_interval=30.0, max_concurrent=30,
                  max_retries=3, db_path=None,
                  store=None, agno_agent_factory=None, home_root=None,
@@ -88,8 +94,17 @@ def __init__(self, *, llm=None, embed_client=None, config=None,
                  on_poll_hook=None, cluster_batch_size=8):
         self.llm = llm
         self.embed_client = embed_client
-        self.config = config or {}
-        self.skill_dir = Path(skill_dir) if skill_dir else None
+        self.config: XSkillConfig | Mapping[str, Any] = (
+            config if config is not None else {}
+        )
+        if skill_dir is not None:
+            self.skill_dir = Path(skill_dir)
+        elif isinstance(self.config, XSkillConfig):
+            self.skill_dir = self.config.skill_dir
+        elif self.config.get("skill_dir"):
+            self.skill_dir = Path(str(self.config["skill_dir"])).expanduser()
+        else:
+            self.skill_dir = None
         # home_root：install_to_claude_code 的 target root。生产 daemon 不
         # 传（None）→ 落到 server._home_root() (默认 Path.home())。测试
         # 必须显式传 tmp_path 防止污染真实 ~/.claude/skills/。
diff --git a/tests/test_config_class.py b/tests/test_config_class.py
index 9adf5dfa..dbbeb4fe 100644
--- a/tests/test_config_class.py
+++ b/tests/test_config_class.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+from unittest.mock import MagicMock
 
 import pytest
 
@@ -67,7 +68,11 @@ def test_xskill_accepts_config_object(tmp_path, monkeypatch):
     config_object = XSkillConfig.from_dict(_sample_config(tmp_path / "skill"))
     registry_path = tmp_path / "registry.db"
 
-    monkeypatch.setattr("xskill.pipeline.registry.REGISTRY_DB", registry_path)
+    monkeypatch.setattr(
+        "xskill.pipeline.registry.REGISTRY_DB",
+        registry_path,
+        raising=False,
+    )
 
     xskill_object = XSkill(config=config_object)
 
@@ -99,3 +104,81 @@ def test_embed_factory_accepts_config_class(tmp_path, monkeypatch):
 
     assert embed_client.model == "embedding-model"
     assert embed_client.dim == 3
+
+
+def test_xskill_serve_passes_same_config_to_create_app(tmp_path, monkeypatch):
+    config_object = XSkillConfig.from_dict(_sample_config(tmp_path / "skill"))
+    captured_kwargs = {}
+
+    def fake_create_app(**kwargs):
+        captured_kwargs.update(kwargs)
+        return MagicMock(name="fastapi-app")
+
+    def fake_uvicorn_run(application, **kwargs):
+        assert application is not None
+        assert kwargs["host"] == "127.0.0.1"
+        assert kwargs["port"] == 8765
+
+    monkeypatch.setattr("xskill.api.create_app", fake_create_app)
+    monkeypatch.setattr("uvicorn.run", fake_uvicorn_run)
+
+    XSkill(config=config_object).serve(host="127.0.0.1", port=8765)
+
+    assert captured_kwargs["config"] is config_object
+
+
+def test_agent_tools_description_opt_uses_injected_config(tmp_path, monkeypatch):
+    from xskill.agents import agent_tools
+
+    config_values = _sample_config(tmp_path / "skill")
+    config_values["skill_opt"] = {"enabled": True}
+    config_object = XSkillConfig.from_dict(config_values)
+    skill_dir = tmp_path / "skill"
+    target_dir = skill_dir / "demo"
+    target_dir.mkdir(parents=True)
+    captured_configs = []
+    saved_context = agent_tools.agent_tool_config.snapshot()
+
+    def fake_get_config():
+        raise AssertionError("agent_tools should not read global config")
+
+    def fake_create_llm_client(runtime_config):
+        captured_configs.append(runtime_config)
+        return object()
+
+    def fake_create_embed_client(runtime_config):
+        captured_configs.append(runtime_config)
+        return object()
+
+    def fake_make_default_factory(runtime_config):
+        captured_configs.append(runtime_config)
+        return object()
+
+    def fake_optimize_description(target, **kwargs):
+        assert target == target_dir
+        captured_configs.append(kwargs["config"])
+
+    try:
+        agent_tools.init_skill_authoring_tool_context(
+            skill_dir=skill_dir,
+            data_dir=skill_dir,
+            config=config_object,
+        )
+        monkeypatch.setattr("xskill.config.get_config", fake_get_config)
+        monkeypatch.setattr("xskill.utils.llm.create_llm_client", fake_create_llm_client)
+        monkeypatch.setattr("xskill.utils.llm.create_embed_client", fake_create_embed_client)
+        monkeypatch.setattr(
+            "xskill.agents.agno_factory.make_default_factory",
+            fake_make_default_factory,
+        )
+        monkeypatch.setattr(
+            "xskill.skill.description_opt.optimize_description",
+            fake_optimize_description,
+        )
+
+        agent_tools._run_description_optimization(target_dir, "demo")
+
+        assert captured_configs
+        assert all(runtime_config is config_object for runtime_config in captured_configs)
+    finally:
+        agent_tools.agent_tool_config.restore(saved_context)
diff --git a/tests/test_server_watcher_startup.py b/tests/test_server_watcher_startup.py
index 93993f31..98b3f441 100644
--- a/tests/test_server_watcher_startup.py
+++ b/tests/test_server_watcher_startup.py
@@ -10,6 +10,8 @@
 
 from unittest.mock import MagicMock, patch
 
+from xskill import XSkillConfig
+
 
 def test_watcher_starts_even_with_empty_registry(tmp_path):
     from xskill.api import app as srv
@@ -45,3 +47,70 @@ def test_watcher_starts_even_with_empty_registry(tmp_path):
         srv._watcher_ref.clear()
         srv._config = None
         srv._skill_dir = None
+
+
+def test_create_app_startup_passes_same_config_to_watcher(tmp_path, monkeypatch):
+    from starlette.testclient import TestClient
+    from xskill.api import app as srv
+
+    skill_dir = tmp_path / "skill"
+    skill_dir.mkdir()
+    config_object = XSkillConfig.from_dict({
+        "skill_dir": str(skill_dir),
+        "llm": {"base_url": "x", "model": "y", "api_key": "z"},
+        "embedding": {
+            "base_url": "embed",
+            "model": "vector",
+            "api_key": "key",
+        },
+        "watcher": {"poll_interval": 30},
+    })
+    captured_kwargs = {}
+
+    class FakeDirectoryWatcher:
+        def __init__(self, **kwargs):
+            captured_kwargs.update(kwargs)
+            self.stopped = False
+
+        @property
+        def stats(self):
+            return {"running": True}
+
+        def start(self):
+            captured_kwargs["started"] = True
+
+        def stop(self):
+            self.stopped = True
+
+    srv._watcher_ref.clear()
+    try:
+        def fake_create_llm_client(runtime_config):
+            assert runtime_config is config_object
+            return object()
+
+        def fake_create_embed_client(runtime_config):
+            assert runtime_config is config_object
+            return object()
+
+        def fake_init_skill_authoring_tool_context(**kwargs):
+            assert kwargs["config"] is config_object
+
+        def fake_detect_known_ecosystems(home_root):
+            assert home_root == tmp_path.resolve()
+            return []
+
+        monkeypatch.setattr(srv, "create_llm_client", fake_create_llm_client)
+        monkeypatch.setattr(srv, "create_embed_client", fake_create_embed_client)
+        monkeypatch.setattr(srv, "init_skill_authoring_tool_context", fake_init_skill_authoring_tool_context)
+        monkeypatch.setattr("xskill.ecosystems.detect_known_ecosystems", fake_detect_known_ecosystems)
+        monkeypatch.setattr("xskill.pipeline.runner.DirectoryWatcher", FakeDirectoryWatcher)
+
+        app = srv.create_app(home_root=tmp_path, config=config_object)
+        with TestClient(app):
+            assert captured_kwargs["config"] is config_object
+            assert captured_kwargs["skill_dir"] == skill_dir
+            assert captured_kwargs["started"] is True
+    finally:
+        srv._watcher_ref.clear()
+        srv._config = None
+        srv._skill_dir = None

From 1b2bb309f069b053d17e408ec80cde19be2f11b9 Mon Sep 17 00:00:00 2001
From: 370025263 <370025263@qq.com>
Date: Wed, 1 Jul 2026 22:25:21 +0800
Subject: [PATCH 4/7] avoid global config in team startup

---
 src/xskill/api/app.py                | 10 +++-
 tests/test_server_watcher_startup.py | 83 ++++++++++++++++++++++++++++
 2 files changed, 91 insertions(+), 2 deletions(-)

diff --git a/src/xskill/api/app.py b/src/xskill/api/app.py
index fe5b6771..b3be181f 100644
--- a/src/xskill/api/app.py
+++ b/src/xskill/api/app.py
@@ -1210,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):
diff --git a/tests/test_server_watcher_startup.py b/tests/test_server_watcher_startup.py
index 98b3f441..e673310c 100644
--- a/tests/test_server_watcher_startup.py
+++ b/tests/test_server_watcher_startup.py
@@ -114,3 +114,86 @@ def fake_detect_known_ecosystems(home_root):
         srv._watcher_ref.clear()
         srv._config = None
         srv._skill_dir = None
+
+
+def test_team_server_startup_uses_runtime_config_for_traj_root(tmp_path, monkeypatch):
+    from starlette.testclient import TestClient
+    from xskill.api import app as srv
+
+    skill_dir = tmp_path / "skill"
+    skill_dir.mkdir()
+    team_traj_root = tmp_path / "runtime-team-trajs"
+    config_object = XSkillConfig.from_dict({
+        "skill_dir": str(skill_dir),
+        "llm": {"base_url": "x", "model": "y", "api_key": "z"},
+        "embedding": {
+            "base_url": "embed",
+            "model": "vector",
+            "api_key": "key",
+        },
+        "team": {"server": {"traj_root": str(team_traj_root)}},
+        "watcher": {"poll_interval": 30},
+    })
+    captured_context = {}
+
+    class FakeDirectoryWatcher:
+        def __init__(self, **kwargs):
+            captured_context["watcher_config"] = kwargs["config"]
+
+        def start(self):
+            captured_context["watcher_started"] = True
+
+        def stop(self):
+            captured_context["watcher_stopped"] = True
+
+    class FakeClientRegistry:
+        def __init__(self, path):
+            captured_context["clients_db_path"] = path
+
+    def fail_get_config():
+        raise AssertionError("team startup should not read global config")
+
+    def fake_init_team_context(**kwargs):
+        captured_context.update(kwargs)
+
+    def fake_register_dir(path, label, ecosystem=None):
+        captured_context["registered_path"] = path
+        captured_context["registered_label"] = label
+        captured_context["registered_ecosystem"] = ecosystem
+
+    def fake_create_llm_client(runtime_config):
+        assert runtime_config is config_object
+        return object()
+
+    def fake_create_embed_client(runtime_config):
+        assert runtime_config is config_object
+        return object()
+
+    def fake_init_skill_authoring_tool_context(**kwargs):
+        assert kwargs["config"] is config_object
+
+    def fake_ensure_join_token(path):
+        captured_context["state_path"] = path
+        return "token"
+
+    srv._watcher_ref.clear()
+    try:
+        monkeypatch.setattr("xskill.config.get_config", fail_get_config)
+        monkeypatch.setattr(srv, "create_llm_client", fake_create_llm_client)
+        monkeypatch.setattr(srv, "create_embed_client", fake_create_embed_client)
+        monkeypatch.setattr(srv, "init_skill_authoring_tool_context", fake_init_skill_authoring_tool_context)
+        monkeypatch.setattr("xskill.team.server.state.ensure_join_token", fake_ensure_join_token)
+        monkeypatch.setattr("xskill.team.server.client_registry.ClientRegistry", FakeClientRegistry)
+        monkeypatch.setattr("xskill.team.server.api.init_team_context", fake_init_team_context)
+        monkeypatch.setattr("xskill.pipeline.registry.register_dir", fake_register_dir)
+        monkeypatch.setattr("xskill.pipeline.runner.DirectoryWatcher", FakeDirectoryWatcher)
+
+        app = srv.create_app(home_root=tmp_path, config=config_object, team_server=True)
+        with TestClient(app):
+            assert captured_context["traj_root"] == team_traj_root
+            assert captured_context["watcher_config"] is config_object
+            assert team_traj_root.is_dir()
+    finally:
+        srv._watcher_ref.clear()
+        srv._config = None
+        srv._skill_dir = None

From c82e67d5ee845ab3b5d90c8f0d56b414c7de85df Mon Sep 17 00:00:00 2001
From: 370025263 <370025263@qq.com>
Date: Wed, 1 Jul 2026 23:55:52 +0800
Subject: [PATCH 5/7] fix: use cls for classmethod receivers

---
 src/xskill/config.py    | 8 ++++----
 src/xskill/utils/llm.py | 8 ++++----
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/src/xskill/config.py b/src/xskill/config.py
index 5ea8cd12..ea2c609e 100644
--- a/src/xskill/config.py
+++ b/src/xskill/config.py
@@ -38,7 +38,7 @@ class XSkillConfig(Mapping[str, Any]):
 
     @classmethod
     def from_yaml(
-        class_type,
+        cls,
         path: Optional[Path] = None,
         *,
         validate_required: bool = True,
@@ -52,7 +52,7 @@ def from_yaml(
             )
         with open(config_path, encoding="utf-8") as config_file:
             loaded_data = yaml.safe_load(config_file) or {}
-        return class_type.from_dict(
+        return cls.from_dict(
             loaded_data,
             source_path=config_path,
             validate_required=validate_required,
@@ -60,14 +60,14 @@ def from_yaml(
 
     @classmethod
     def from_dict(
-        class_type,
+        cls,
         values: Mapping[str, Any] | None,
         *,
         source_path: Optional[Path] = None,
         validate_required: bool = True,
     ) -> "XSkillConfig":
         config_data = deepcopy(dict(values or {}))
-        config_object = class_type(
+        config_object = cls(
             data=config_data,
             source_path=Path(source_path) if source_path else None,
         )
diff --git a/src/xskill/utils/llm.py b/src/xskill/utils/llm.py
index 91d8f394..69533184 100644
--- a/src/xskill/utils/llm.py
+++ b/src/xskill/utils/llm.py
@@ -64,7 +64,7 @@ class LLMClient:
     _client: object = field(default=None, repr=False)
 
     @classmethod
-    def from_config(class_type, config_section: Mapping[str, Any]) -> "LLMClient":
+    def from_config(cls, config_section: Mapping[str, Any]) -> "LLMClient":
         base_url = config_section.get("base_url", "").rstrip("/")
         model = config_section.get("model", "")
         api_key = (
@@ -84,7 +84,7 @@ def from_config(class_type, config_section: Mapping[str, Any]) -> "LLMClient":
             kwargs["temperature"] = float(config_section["temperature"])
         if "rate_limit" in config_section:
             kwargs["rate_limit_cfg"] = config_section["rate_limit"]
-        return class_type(**kwargs)
+        return cls(**kwargs)
 
     def _get_client(self):
         if self._client is None:
@@ -206,7 +206,7 @@ class EmbedClient:
     _client: object = field(default=None, repr=False)
 
     @classmethod
-    def from_config(class_type, config_section: Mapping[str, Any]) -> "EmbedClient":
+    def from_config(cls, config_section: Mapping[str, Any]) -> "EmbedClient":
         base_url = config_section.get("base_url", "").rstrip("/")
         model = config_section.get("model", "")
         api_key = (
@@ -218,7 +218,7 @@ def from_config(class_type, config_section: Mapping[str, Any]) -> "EmbedClient":
         if not base_url or not model:
             raise ValueError("embedding.base_url 和 embedding.model 必须配置")
         api_style = _resolve_embed_api_style(config_section, model)
-        instance = class_type(
+        instance = cls(
             base_url=base_url, model=model, api_key=api_key, dim=dim, api_style=api_style,
         )
         return instance

From 1307c35d8417378725522357bc8f091f1b90467f Mon Sep 17 00:00:00 2001
From: 370025263 <370025263@qq.com>
Date: Thu, 2 Jul 2026 15:17:12 +0800
Subject: [PATCH 6/7] fix: address typed config review feedback

---
 src/xskill/agents/agent_tools.py |  42 +++++--------
 src/xskill/core.py               |  24 ++++---
 src/xskill/pipeline/runner.py    | 104 ++++++++++++++++---------------
 3 files changed, 85 insertions(+), 85 deletions(-)

diff --git a/src/xskill/agents/agent_tools.py b/src/xskill/agents/agent_tools.py
index 97e89940..09921834 100644
--- a/src/xskill/agents/agent_tools.py
+++ b/src/xskill/agents/agent_tools.py
@@ -162,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
 # ═══════════════════════════════════════════════════════════════════
@@ -273,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)
@@ -307,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.")
 
 
 # ═══════════════════════════════════════════════════════════════════
@@ -409,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)
@@ -492,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
@@ -551,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")
@@ -612,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 中。后续：
@@ -654,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 看到"还差多少到阈值"。
@@ -1056,7 +1048,7 @@ def absorb_user_edit_to_main(skill_name: str, message: str) -> str:
 
 
 # ═══════════════════════════════════════════════════════════════════
-# v2.2 渐进收敛工具（ClusterAgent 用，处理近义 slug 整合）
+# 渐进收敛工具（ClusterAgent 用，处理近义 slug 整合）
 # ═══════════════════════════════════════════════════════════════════
 
 @tool(name="rename_skill")
diff --git a/src/xskill/core.py b/src/xskill/core.py
index e0eee128..4213a804 100644
--- a/src/xskill/core.py
+++ b/src/xskill/core.py
@@ -2,7 +2,7 @@
 xskill.py — XSkill 顶层门面
 ═══════════════════════════════════════════════════════
 唯一对外入口。持 config + registry + skill_repo，
-提供 search / serve / score_trajectory_ux 三个动作方法。
+提供 search / serve / score_atoms_for_trajectory 三类动作方法。
 """
 
 from __future__ import annotations
@@ -33,8 +33,8 @@ class XSkill:
         # daemon
         xskill.serve(host="0.0.0.0", port=8000)
 
-        # 主动 UX 打分（维护性，watcher 会自动跑）
-        xskill.score_trajectory_ux(traj)
+        # 主动给轨迹内的 atom 补 UX 分（维护性，watcher 会自动跑）
+        xskill.score_atoms_for_trajectory(traj)
 
         # 子系统访问
         xskill.registry.list()
@@ -121,13 +121,13 @@ def search_skills(self, query: str, top_k: int = 5) -> list[SkillHit]:
             ))
         return out[:top_k]
 
-    # ─── UX 打分（主动；v2 atom 粒度）──────────────────────────
-    def score_trajectory_ux(self, traj: Trajectory) -> UxScoreResult:
-        """主动给一条 traj 的所有 atom 补 UX 分（幂等：已落盘的跳过）。
+    # ─── UX 打分（主动；atom 粒度）──────────────────────────
+    def score_atoms_for_trajectory(self, traj: Trajectory) -> UxScoreResult:
+        """主动给一条 traj 内的所有 atom 补 UX 分（幂等：已落盘的跳过）。
 
-        v2: 打分对象是 AtomTask，不是整条 traj。前置条件——该 traj 已被
-        watcher 走完 split 阶段（atoms 落在 ``<traj_dir>/<traj_id>/tasks/``）。
-        没拆过的 traj 调本方法 ``scored=0``，因为 store 里没东西。
+        打分对象是 AtomTask，不是整条 traj。前置条件：该 traj 已被
+        watcher 拆成 atoms，落在 ``<traj_dir>/<traj_id>/tasks/``。
+        没拆过的 traj 调本方法会返回 ``scored=False``，因为 store 里没 atom。
 
         watcher 自动跑；本方法用于 watcher 漏打 / 手动重打。
         """
@@ -164,7 +164,7 @@ def score_trajectory_ux(self, traj: Trajectory) -> UxScoreResult:
             commit_sha=header.get("sha", ""),
             canary_config=canary_cfg,
         )
-        # v2 返回 {scored: int, skipped: int, decision}；UxScoreResult.scored 是 bool
+        # 返回 {scored: int, skipped: int, decision}；UxScoreResult.scored 是 bool。
         return UxScoreResult(
             scored=bool(d.get("scored", 0) > 0),
             score=None,  # 多 atom 没有单一分数；调用方需读 .ux_scores.jsonl 细看
@@ -172,6 +172,10 @@ def score_trajectory_ux(self, traj: Trajectory) -> UxScoreResult:
             decision=d.get("decision", {}),
         )
 
+    def score_trajectory_ux(self, traj: Trajectory) -> UxScoreResult:
+        """兼容旧 SDK 名称；实际仍按 atom 粒度打分。"""
+        return self.score_atoms_for_trajectory(traj)
+
     # ─── daemon ────────────────────────────────────────────────
     def serve(self, host: str = "0.0.0.0", port: int = 8000,
               *, home_root: Path | str | None = None,
diff --git a/src/xskill/pipeline/runner.py b/src/xskill/pipeline/runner.py
index 83fa27bf..746f684b 100644
--- a/src/xskill/pipeline/runner.py
+++ b/src/xskill/pipeline/runner.py
@@ -4,21 +4,21 @@
 
 每条轨迹独立流转，不分批不阻塞：
 
-  discovered → meta_extracting → meta_done → indexed → processing → done
+  discovered → splitting → split_done → indexed → done
 
 每次扫描：
   1. 发现新文件
-  2. 对每条 discovered 提交 meta 提取任务（不等待）
-  3. 对每条 meta_done 提交 embedding 任务（不等待）
-  4. 对每条 indexed 提交 process_traj 任务（不等待）
+  2. 对每条 discovered 提交 atom 拆分任务（不等待）
+  3. 对每条 split_done 提交 atom 索引任务（不等待）
+  4. 对每条 indexed 提交 atom cluster 任务（不等待）
   5. 收割已完成的 futures，更新状态
-  6. 解析 xskill header → ux_score
+  6. cluster 完成后按 atom 写 ux_score
 
 所有耗时操作都在 ThreadPoolExecutor 中异步执行，扫描本身秒完。
 
-本模块还含 AtomTask 流水线核心入口 ``process_atom_task``（原 process.py）：
-v2 (AtomTask) 流水线下，对一个 atom 的"cluster → 触发 SkillEdit"是单一原子
-操作。``api/sse.py`` 与本模块的 ``DirectoryWatcher`` 都调它。
+本模块还含 AtomTask 流水线核心入口 ``process_atom_task``。每个 atom 的
+cluster 结果先进入 skill 的 candidates buffer，SkillEdit 由 watcher 每轮
+独立扫描触发。``api/sse.py`` 与本模块的 ``DirectoryWatcher`` 都调它。
 """
 
 from __future__ import annotations
@@ -31,6 +31,7 @@
 import time
 from collections.abc import Mapping
 from concurrent.futures import ThreadPoolExecutor, Future, as_completed
+from enum import Enum
 from pathlib import Path
 from typing import Any
 
@@ -49,33 +50,17 @@
 
 logger = logging.getLogger("xskill.watcher")
 
-# v2 (AtomTask 流水线) 的 action → status 映射
-# splitting → split_done → indexed → clustering → done
-_ACTION_STATUS = {
-    "clustered": "done",
-    "skip": "indexed",
-    "error": "error",
-}
-
-def _install_thread_event_loop() -> None:
-    """给工作线程装一个事件循环（Python 3.9 兼容）。
-
-    Python 3.9 上，在没有事件循环的非主线程里构造 asyncio 对象（如
-    ``asyncio.Lock()``）会 ``raise RuntimeError``。``agno`` 在模块导入期就
-    构造了一个 ``asyncio.Lock()``，而 watcher 线程 / pool 工作线程会懒加载
-    agno —— 不显式给线程装循环,导入即崩。3.10+ 的 ``asyncio.Lock()`` 不在
-    构造期抓 loop,本函数对其无影响。
-    """
-    asyncio.set_event_loop(asyncio.new_event_loop())
+
+class AtomProcessAction(str, Enum):
+    CLUSTERED = "clustered"
 
 
 class DirectoryWatcher:
     """流水线式目录监听器。每条 traj 独立流转，不分批不阻塞。
 
-    v2 状态机：
+    状态机：
       discovered → splitting → split_done → indexed → done
 
-    与 v1 (meta-level) 的差异：
     - splitting 阶段调 TaskAgent 拆 AtomTask，落盘到 ``<traj_root>/<traj_id>/tasks/``
     - indexed 阶段以 AtomTask 为单位整批重建 ``<traj_root>/index.pkl``
     - cluster 阶段**跨轨迹池化**：把所有 indexed 轨迹里尚未落地的 atom 汇成一池，
@@ -85,6 +70,18 @@ class DirectoryWatcher:
       skill 的 ``.candidates.yml`` 时才 done（文件系统即队列，天然去重+断点续传）。
     """
 
+    @staticmethod
+    def _install_thread_event_loop() -> None:
+        """给 watcher 线程和工作线程装一个事件循环（Python 3.9 兼容）。
+
+        Python 3.9 上，在没有事件循环的非主线程里构造 asyncio 对象（如
+        ``asyncio.Lock()``）会 ``raise RuntimeError``。``agno`` 在模块导入期就
+        构造了一个 ``asyncio.Lock()``，而 watcher 线程 / pool 工作线程会懒加载
+        agno —— 不显式给线程装循环，导入即崩。3.10+ 的 ``asyncio.Lock()`` 不在
+        构造期抓 loop，本方法对其无影响。
+        """
+        asyncio.set_event_loop(asyncio.new_event_loop())
+
     def __init__(self, *, llm=None, embed_client=None,
                  config: XSkillConfig | Mapping[str, Any] | None = None,
                  skill_dir=None, poll_interval=30.0, max_concurrent=30,
@@ -105,9 +102,9 @@ def __init__(self, *, llm=None, embed_client=None,
             self.skill_dir = Path(str(self.config["skill_dir"])).expanduser()
         else:
             self.skill_dir = None
-        # home_root：install_to_claude_code 的 target root。生产 daemon 不
-        # 传（None）→ 落到 server._home_root() (默认 Path.home())。测试
-        # 必须显式传 tmp_path 防止污染真实 ~/.claude/skills/。
+        # home_root 是被扫描和安装的用户家目录，例如 /home/alice。
+        # 生产 daemon 留 None，运行时从 server._home_root() 取 Path.home()；
+        # 测试传 tmp_path，避免写入真实 ~/.claude/skills/、~/.codex/ 等目录。
         self.home_root = Path(home_root) if home_root else None
         # server_mode：team server 模式。server 是纯 server——不装 skill 到
         # 本机生态、不做单机灰度轮转、不做本地手改回流（手改走 client
@@ -141,7 +138,7 @@ def __init__(self, *, llm=None, embed_client=None,
         # future）。1 = 退回逐 atom 一次往返的旧行为。
         self.cluster_batch_size = max(1, int(cluster_batch_size))
 
-        # v2 注入：AtomTaskStore + agno agent 工厂
+        # AtomTaskStore + agno agent 工厂
         # store None 时本 watcher 不能跑 splitting/clustering（仅 ux_score 还能跑）
         self.store = store
         self.agno_agent_factory = agno_agent_factory
@@ -150,7 +147,9 @@ def __init__(self, *, llm=None, embed_client=None,
         self._pause = threading.Event()
         self._thread: threading.Thread | None = None
         self._pool = ThreadPoolExecutor(
-            max_workers=max_concurrent, initializer=_install_thread_event_loop)
+            max_workers=max_concurrent,
+            initializer=self._install_thread_event_loop,
+        )
         self._futures: dict[Future, dict] = {}
         self._last_poll: float | None = None
         # 单机 canary 轮转节流：上次真跑 _reconcile_skill_sides 的时间戳。
@@ -158,10 +157,10 @@ def __init__(self, *, llm=None, embed_client=None,
         self._last_rotate_ts: float | None = None
         self._stats = {
             "polls": 0, "new_trajs": 0,
-            "atoms_extracted": 0,    # v2: 累计 atom 数（替代 meta_extracted）
+            "atoms_extracted": 0,    # 累计 atom 数
             "indexed": 0,            # 仍记录索引重建次数
-            "atoms_clustered": 0,    # v2: 累计 cluster 调用次数
-            "skills_edited": 0,      # v2: 触发的 SkillEdit 次数
+            "atoms_clustered": 0,    # 累计 cluster 调用次数
+            "skills_edited": 0,      # 触发的 SkillEdit 次数
             "scores": 0, "errors": 0, "retries": 0,
         }
 
@@ -216,7 +215,7 @@ def _db_kw(self):
     def _loop(self):
         # watcher 线程内会懒加载 agno（导入期即构造 asyncio.Lock()）。
         # Python 3.9 非主线程无事件循环时构造会崩 —— 先给本线程装一个。
-        _install_thread_event_loop()
+        self._install_thread_event_loop()
         while not self._stop.is_set():
             if not self._pause.is_set():
                 if self.on_poll_hook is not None:
@@ -876,14 +875,14 @@ def _factory(self):
     # 任务执行函数（在线程池中运行）
     # ───────────────────────────────────────────────────────────
 
-    # v2 流水线任务：split / atom_index / cluster
+    # 流水线任务：split / atom_index / cluster
 
     def _do_split(self, dir_path, fname):
         """跑 TaskAgent 拆 AtomTask。返回 (fname, num_atoms_added, last_offset, last_atom_id, err)。
 
-        v2.3: TaskAgent 走 agentic 工具调用（submit_atom/readfile/grep），用
-        和 cluster/edit 同一个 agno 工厂。``updated`` 状态的续写轨迹和首次
-        ``discovered`` 走同一条路径——TaskAgent 内部用 last_offset 续接点只拆
+        TaskAgent 走 agentic 工具调用（submit_atom/readfile/grep），并和
+        cluster/edit 使用同一个 agno 工厂。``updated`` 状态的续写轨迹和首次
+        ``discovered`` 走同一条路径，TaskAgent 内部用 last_offset 续接点只拆
         新增内容。
         """
         import time
@@ -1021,7 +1020,10 @@ def _on_cluster_batch_done(self, results):
         in_skills = [r for r in results if r.get("skill_name")]
         dropped = [
             r for r in results
-            if r.get("action") == "clustered" and not r.get("skill_name")
+            if (
+                r.get("action") == AtomProcessAction.CLUSTERED.value
+                and not r.get("skill_name")
+            )
         ]
 
         _emit = logger.info if n_total > 0 else logger.debug
@@ -1065,7 +1067,9 @@ def _sweep_done_trajs(self, wd_id, dir_path, **kw):
             if any(not self._atom_consumed(a) for a in atoms):
                 continue  # 还有未消费 atom → 等后续 batch 消费
             update_traj_status(
-                wd_id, fname, "done", process_action="clustered", **kw,
+                wd_id, fname, "done",
+                process_action=AtomProcessAction.CLUSTERED.value,
+                **kw,
             )
             # 该轨迹所有 atom 已落盘——ux_score 应当跑的时机。
             if self.server_mode:
@@ -1078,7 +1082,7 @@ def _sweep_done_trajs(self, wd_id, dir_path, **kw):
     # ───────────────────────────────────────────────────────────
 
     def _score_new(self, _watch_dir_id, _dir_path, _filenames, **_kwargs):
-        """v2: 不在发现新 traj 时打分（那时 atom 还没拆）。
+        """不在发现新 traj 时打分（那时 atom 还没拆）。
 
         实际打分在 ``_sweep_done_trajs`` → ``_score_atoms_for_traj`` 触发。
         此方法保留 hook 兼容 ``_scan_dir`` 末尾的调用；只在 traj 没有
@@ -1223,11 +1227,11 @@ def _score_atoms_for_traj_server(self, wd_id, fname, **kw):
 
 
 # ═══════════════════════════════════════════════════════════════════
-# AtomTask 流水线核心入口（原 process.py）
+# AtomTask 流水线核心入口
 # ═══════════════════════════════════════════════════════════════════
-# v2 (AtomTask) 流水线下，对一个 atom 的"cluster → 触发 SkillEdit"是单一原子
-# 操作；不存在"轨迹整篇喂 LLM"概念。api/sse.py / runner 的 DirectoryWatcher
-# 都调本函数，传入已 split + indexed 完毕的 atom_id。
+# 对一个 atom 的 cluster 操作只负责把候选落到 skill buffer；SkillEdit 由
+# watcher 独立扫描触发。api/sse.py / runner 的 DirectoryWatcher 都调本函数，
+# 传入已 split + indexed 完毕的 atom_id。
 
 _process_logger = logging.getLogger("xskill.process")
 
@@ -1308,7 +1312,7 @@ def process_atom_task(*, atom_id: str, config: dict, skill_dir: Path,
             logger.debug("atom adoption telemetry skipped", exc_info=True)
 
     return {
-        "action": "clustered",
+        "action": AtomProcessAction.CLUSTERED.value,
         "atom_id": atom_id,
         "skill_name": skill_name,
         "weightscore": weightscore,
@@ -1379,7 +1383,7 @@ def process_atom_batch(*, atom_ids: list[str], config: dict, skill_dir: Path,
             except Exception:  # pylint: disable=broad-exception-caught
                 logger.debug("atom adoption telemetry skipped", exc_info=True)
         results.append({
-            "action": "clustered",
+            "action": AtomProcessAction.CLUSTERED.value,
             "atom_id": aid,
             "skill_name": skill_name,
             "weightscore": weightscore,

From ee28855cf6c85007909b62f519090b253b68ee8c Mon Sep 17 00:00:00 2001
From: 370025263 <370025263@qq.com>
Date: Thu, 2 Jul 2026 15:26:22 +0800
Subject: [PATCH 7/7] fix: make skill edit tests encoding explicit

---
 tests/test_skill_edit_agent.py | 23 ++++++++++++++++-------
 1 file changed, 16 insertions(+), 7 deletions(-)

diff --git a/tests/test_skill_edit_agent.py b/tests/test_skill_edit_agent.py
index 12406ba1..6b01a581 100644
--- a/tests/test_skill_edit_agent.py
+++ b/tests/test_skill_edit_agent.py
@@ -474,7 +474,7 @@ def test_pipeline_contract_parts_kept(self):
 class _JamMergeStubAgno(_BabyStubAgno):
     """模拟 jam-merge：读 scenario 里的 skill_name + 目标路径，写合并正文，
     调 commit_update_main（而非 commit_to_staging）。"""
-    def run(self, user_msg, **kw):
+    def run(self, user_msg, **_kw):
         type(self).invoked = True
         type(self).user_msg = user_msg
         import re
@@ -497,7 +497,7 @@ class _R: pass
 
 class _JamNoCommitStubAgno(_BabyStubAgno):
     """模拟 agent 写了 SKILL.md 但没调用 commit_update_main。"""
-    def run(self, user_msg, **kw):
+    def run(self, user_msg, **_kw):
         type(self).invoked = True
         type(self).user_msg = user_msg
         import re
@@ -521,7 +521,10 @@ def _seed_candidates(skill_dir, total_ws):
 def test_jam_merge_fires_above_threshold_and_discards_staging(tmp_path):
     sd = _make_main_skill(tmp_path / "skill", "jam-skill")
     # 写点东西并开 staging（灰度中）
-    (sd / "SKILL.md").write_text((sd / "SKILL.md").read_text() + "\n<!-- staging draft -->\n", encoding="utf-8")
+    (sd / "SKILL.md").write_text(
+        (sd / "SKILL.md").read_text(encoding="utf-8") + "\n<!-- staging draft -->\n",
+        encoding="utf-8",
+    )
     assert commit_to_staging_branch(str(sd), "stub staging candidate") is True
     assert (sd.parent / ".canary" / "jam-skill" / "SKILL.md").is_file()
     # 候选攒到 60 ≥ jam_threshold(50)
@@ -550,7 +553,10 @@ def test_jam_merge_fires_above_threshold_and_discards_staging(tmp_path):
 
 def test_no_jam_below_threshold_keeps_staging(tmp_path):
     sd = _make_main_skill(tmp_path / "skill", "calm-skill")
-    (sd / "SKILL.md").write_text((sd / "SKILL.md").read_text() + "\n<!-- s -->\n", encoding="utf-8")
+    (sd / "SKILL.md").write_text(
+        (sd / "SKILL.md").read_text(encoding="utf-8") + "\n<!-- s -->\n",
+        encoding="utf-8",
+    )
     assert commit_to_staging_branch(str(sd), "stub staging") is True
     _seed_candidates(sd, 40)  # < 50
     _JamMergeStubAgno.invoked = False
@@ -567,7 +573,7 @@ def test_no_jam_below_threshold_keeps_staging(tmp_path):
 def test_jam_merge_without_main_commit_keeps_candidates_and_staging(tmp_path):
     sd = _make_main_skill(tmp_path / "skill", "jam-no-commit")
     (sd / "SKILL.md").write_text(
-        (sd / "SKILL.md").read_text() + "\n<!-- staging draft -->\n",
+        (sd / "SKILL.md").read_text(encoding="utf-8") + "\n<!-- staging draft -->\n",
         encoding="utf-8",
     )
     assert commit_to_staging_branch(str(sd), "stub staging") is True
@@ -594,7 +600,7 @@ class _JamMergeRematerializeStubAgno(_BabyStubAgno):
     """
     staging_body_content_seen: str = ""
 
-    def run(self, user_msg, **kw):
+    def run(self, user_msg, **_kw):
         type(self).invoked = True
         type(self).user_msg = user_msg
         import re
@@ -630,7 +636,10 @@ def test_jam_merge_rematerializes_missing_staging_body(tmp_path):
 
     sd = _make_main_skill(tmp_path / "skill", "rematerialize-skill")
     # 写 staging 分支（包含可识别内容）
-    staging_content = (sd / "SKILL.md").read_text() + "\n<!-- unique staging marker -->\n"
+    staging_content = (
+        (sd / "SKILL.md").read_text(encoding="utf-8")
+        + "\n<!-- unique staging marker -->\n"
+    )
     (sd / "SKILL.md").write_text(staging_content, encoding="utf-8")
     assert commit_to_staging_branch(str(sd), "stub staging candidate") is True
     # 确认 .canary 已物化