Skill Loading：按需加载领域知识

import os
import re
import yaml
from pathlib import Path

class SkillLoader:
    def __init__(self, skills_dirs: list[Path] = None):
        """多目录技能加载器，模拟真实的优先级解析"""
        self.skills_dirs = skills_dirs or [Path("skills")]
        self.skills: dict[str, dict] = {}
        self.conditional_skills: dict[str, dict] = {}
        self._scan()

    def _parse_frontmatter(self, content: str) -> tuple[dict, str]:
        """解析 YAML frontmatter"""
        if content.startswith('---'):
            end = content.find('---', 3)
            if end != -1:
                fm_text = content[3:end].strip()
                body = content[end + 3:].strip()
                try:
                    fm = yaml.safe_load(fm_text) or {}
                    return fm, body
                except yaml.YAMLError:
                    pass
        return {}, content

    def _scan(self):
        """扫描所有技能目录，按优先级加载（先加载的优先）"""
        seen = set()
        for skills_dir in self.skills_dirs:
            if not skills_dir.exists():
                continue
            for skill_dir in sorted(skills_dir.iterdir()):
                if not skill_dir.is_dir():
                    continue
                skill_md = skill_dir / "SKILL.md"
                if not skill_md.exists():
                    continue

                # 去重：已存在则跳过（模拟 realpath 去重）
                resolved = skill_md.resolve()
                if resolved in seen:
                    continue
                seen.add(resolved)

                content = skill_md.read_text()
                frontmatter, body = self._parse_frontmatter(content)

                name = frontmatter.get("name", skill_dir.name)
                desc = frontmatter.get("description", "")
                if not desc:
                    # fallback：取 markdown 第一行
                    lines = body.strip().splitlines()
                    desc = lines[0].lstrip("# ").strip() if lines else ""

                skill_info = {
                    "name": name,
                    "description": desc,
                    "when_to_use": frontmatter.get("when-to-use", ""),
                    "allowed_tools": frontmatter.get("allowed-tools", []),
                    "paths": frontmatter.get("paths"),
                    "context": frontmatter.get("context"),  # fork / inline
                    "model": frontmatter.get("model"),
                    "path": skill_md,
                    "base_dir": str(skill_dir),
                    "body": body,
                }

                slug = skill_dir.name

                # 条件技能：有 paths 字段的延迟激活
                if skill_info["paths"]:
                    self.conditional_skills[slug] = skill_info
                else:
                    self.skills[slug] = skill_info

    def activate_for_paths(self, file_paths: list[str], cwd: str) -> list[str]:
        """当文件被操作时，激活匹配的条件技能（简化版 gitignore 匹配）"""
        import fnmatch
        activated = []
        to_remove = []
        for slug, skill in self.conditional_skills.items():
            patterns = skill["paths"]
            if isinstance(patterns, str):
                patterns = [patterns]
            for fp in file_paths:
                rel = os.path.relpath(fp, cwd)
                if any(fnmatch.fnmatch(rel, p) for p in patterns):
                    self.skills[slug] = skill
                    to_remove.append(slug)
                    activated.append(slug)
                    break
        for slug in to_remove:
            del self.conditional_skills[slug]
        return activated

    def list_summary(self, char_budget: int = 8000) -> str:
        """生成技能列表摘要，带预算控制"""
        if not self.skills:
            return "No skills available."

        lines = ["## Available Skills\n"]
        total_chars = 0
        for slug, info in self.skills.items():
            desc = info["description"]
            if info["when_to_use"]:
                desc = f"{desc} - {info['when_to_use']}"
            # 单条描述硬上限（模拟 MAX_LISTING_DESC_CHARS = 250）
            if len(desc) > 250:
                desc = desc[:249] + "..."
            entry = f"- **{slug}**: {desc}"
            if total_chars + len(entry) > char_budget:
                entry = f"- **{slug}**"  # 超预算只显示名称
            lines.append(entry)
            total_chars += len(entry)

        lines.append("\n使用 Skill 工具加载完整技能文档。")
        return "\n".join(lines)

    def load(self, skill_name: str) -> dict:
        """加载完整技能，返回内容和元数据"""
        if skill_name not in self.skills:
            available = ", ".join(self.skills.keys())
            raise ValueError(f"Skill '{skill_name}' not found. Available: {available}")

        skill = self.skills[skill_name]
        content = skill["body"]

        # 变量替换（模拟 ${CLAUDE_SKILL_DIR}）
        content = content.replace("${SKILL_DIR}", skill["base_dir"])

        return {
            "content": f"Base directory for this skill: {skill['base_dir']}\n\n{content}",
            "allowed_tools": skill["allowed_tools"],
            "model": skill["model"],
            "context": skill["context"],
        }


# 实例化：模拟真实的多目录优先级
skill_loader = SkillLoader([
    Path.home() / ".claude" / "managed" / "skills",  # 1. 策略级
    Path.home() / ".claude" / "skills",               # 2. 用户级
    Path(".claude") / "skills",                        # 3. 项目级
])

SYSTEM = f"""你是一个 Coding Agent，可以执行工具完成编程任务。

{skill_loader.list_summary()}
"""

TOOLS.append({
    "name": "Skill",
    "description": "Execute a skill within the main conversation. "
        "When users reference a slash command (e.g. /commit), use this tool.",
    "input_schema": {
        "type": "object",
        "properties": {
            "skill": {
                "type": "string",
                "description": "技能名称（目录名），如 'code-review', 'write-tests'"
            },
            "args": {
                "type": "string",
                "description": "可选参数"
            }
        },
        "required": ["skill"]
    }
})

def handle_skill(skill: str, args: str = "") -> dict:
    """处理技能调用，区分 inline 和 fork 模式"""
    result = skill_loader.load(skill)

    if result["context"] == "fork":
        # Fork 模式：启动子 Agent
        sub_result = run_sub_agent(
            prompt=result["content"],
            allowed_tools=result["allowed_tools"],
            model=result.get("model"),
        )
        return {"status": "forked", "result": sub_result}

    # Inline 模式：注入到当前会话
    return {
        "status": "inline",
        "content": result["content"],
        "allowed_tools": result["allowed_tools"],
        "model": result.get("model"),
    }

TOOL_HANDLERS["Skill"] = lambda **kw: handle_skill(kw["skill"], kw.get("args", ""))