zcbot/tools/skill_authoring.py

"""用户 skill 创作工具: save_skill / fork_skill。

为什么是 host-side typed tool(而非让 agent 用 fs/shell 写):
  - fs/shell 的 base_dir 锚在进程 cwd(host)或容器 workdir(docker),**都不指向**
    `user_root/.skills` —— 用相对路径写 `.skills` 只在 docker 下碰巧成立,host 下会
    写错地方,跨 backend 不可靠。
  - host-side 工具直接知道 `user_root/.skills`,一个落点两种 backend 通吃(与
    seedream / DocumentDownload 直接 host 侧写 working_dir 完全一致的范式)。
  - docker 下 user_root 整个 bind 到 /workspace,host 侧写进 .skills 的文件在容器内
    `/workspace/.skills/...` 自动可见(fork 带过来的脚本随之可跑)。

这两个工具不在 `executor_docker.CONTAINER_TOOLS` 里,故恒在 host 侧执行。
"""
from __future__ import annotations

import re
import shutil
from pathlib import Path
from typing import Optional

from core.skills import SkillRegistry, parse_frontmatter

from .base import Tool

# skill 名:小写字母 / 数字开头,可含 _ -,≤64。挡住路径穿越(`../`、`/`)、空格、大写。
_SKILL_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
_FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---\n?", re.DOTALL)


def _validate_skill_name(name: str) -> Optional[str]:
    """返回错误串(不合法)或 None(合法)。"""
    if not name or not _SKILL_NAME_RE.match(name):
        return (
            f"skill 名 '{name}' 不合法:须小写字母/数字开头,只含小写字母、数字、_ 、-,"
            "长度 ≤64(用于目录名,故挡空格 / 大写 / 路径分隔符)"
        )
    return None


def _set_frontmatter_name(text: str, new_name: str) -> str:
    """把 markdown frontmatter 里的 name 改成 new_name(只动 name 行,不重排其余字段)。

    fork 后复制来的 SKILL.md 仍带原 skill 的 `name:`,不改的话注册时会用旧名 →
    与被 fork 的内置同名、反而覆盖了内置。故 fork 落盘后必须把 name 对齐到 new_name。
    """
    m = _FRONTMATTER_RE.match(text)
    if not m:
        return f"---\nname: {new_name}\n---\n\n" + text
    fm = m.group(1)
    if re.search(r"(?m)^name:", fm):
        fm = re.sub(r"(?m)^name:.*$", f"name: {new_name}", fm, count=1)
    else:
        fm = f"name: {new_name}\n" + fm
    return f"---\n{fm}\n---\n" + text[m.end():]


class SaveSkillTool(Tool):
    name = "save_skill"
    description = (
        "Create or overwrite one of the USER's own skills at .skills/<name>/SKILL.md. "
        "Use to author a skill from scratch or save an edited copy. The content must be a "
        "full SKILL.md with YAML frontmatter containing `name` and `description` "
        "(description is the routing blurb shown in the skill list — make it specific: "
        "trigger words + when NOT to use). Takes effect from the user's NEXT message. "
        "To copy a built-in skill that bundles scripts (e.g. ppt), use fork_skill instead "
        "so the scripts come along."
    )
    parameters = {
        "type": "object",
        "properties": {
            "name": {
                "type": "string",
                "description": "Skill name = directory name (lowercase, [a-z0-9_-], <=64). "
                "Same name as a built-in => overrides it for this user; pick a new name to keep both.",
            },
            "content": {
                "type": "string",
                "description": "Full SKILL.md text including --- frontmatter --- (name + description).",
            },
        },
        "required": ["name", "content"],
    }

    def __init__(
        self,
        user_skills_dir: Path,
        registry: SkillRegistry,
        base_dir: Optional[Path] = None,
        user_root: Optional[Path] = None,
    ) -> None:
        super().__init__(base_dir, user_root=user_root)
        self.user_skills_dir = Path(user_skills_dir)
        self.registry = registry

    def execute(self, name: str, content: str) -> str:
        err = _validate_skill_name(name)
        if err is not None:
            return f"[Error] {err}"
        # frontmatter 必须合法且有 description —— 写时就挡住"加载失败"黑洞
        try:
            meta, _ = parse_frontmatter(content)
        except Exception as e:  # yaml.YAMLError 等
            return f"[Error] frontmatter YAML 非法,无法保存:{e}"
        if not (meta.get("description") or "").strip():
            return "[Error] frontmatter 缺 description —— 这是 skill 列表里的路由说明,必填(写清触发词 + 何时别用)"
        fm_name = (meta.get("name") or "").strip()
        if fm_name and fm_name != name:
            return (
                f"[Error] frontmatter 里 name='{fm_name}' 与目标名 '{name}' 不一致 —— "
                "请改成一致(或省略 frontmatter 的 name,默认用目录名)"
            )
        skill_dir = self.user_skills_dir / name
        skill_dir.mkdir(parents=True, exist_ok=True)
        (skill_dir / "SKILL.md").write_text(content, encoding="utf-8")
        builtin = self.registry.get(name)
        note = ""
        if builtin is not None and builtin.source == "builtin":
            note = f" (注意:与内置 skill '{name}' 同名,你这条将覆盖内置;想并存请改名)"
        return f"[saved skill '{name}' to .skills/{name}/SKILL.md,下条消息生效]{note}"


class ForkSkillTool(Tool):
    name = "fork_skill"
    description = (
        "Copy an existing skill (built-in or one of the user's own) — INCLUDING its bundled "
        "scripts/assets — into the user's .skills/<new_name>/, so they can customize it. "
        "This is the right way to 'copy zcbot's ppt skill and tweak it': fork first, then "
        "edit the copied SKILL.md (with the edit tool or save_skill). The new copy's "
        "frontmatter name is auto-set to new_name. Takes effect from the user's NEXT message."
    )
    parameters = {
        "type": "object",
        "properties": {
            "src": {"type": "string", "description": "Name of the skill to copy (as listed in the skill discovery block)."},
            "new_name": {
                "type": "string",
                "description": "New skill name (lowercase, [a-z0-9_-], <=64). Must not already exist under .skills/.",
            },
        },
        "required": ["src", "new_name"],
    }

    def __init__(
        self,
        user_skills_dir: Path,
        registry: SkillRegistry,
        base_dir: Optional[Path] = None,
        user_root: Optional[Path] = None,
    ) -> None:
        super().__init__(base_dir, user_root=user_root)
        self.user_skills_dir = Path(user_skills_dir)
        self.registry = registry

    def execute(self, src: str, new_name: str) -> str:
        err = _validate_skill_name(new_name)
        if err is not None:
            return f"[Error] {err}"
        skill = self.registry.get(src)
        if skill is None:
            available = ", ".join(self.registry.skills.keys()) or "(none)"
            return f"[Error] 源 skill '{src}' 不存在。可选:{available}"
        dest = self.user_skills_dir / new_name
        if dest.exists():
            return f"[Error] .skills/{new_name}/ 已存在 —— 换个名字,或先删旧的"
        self.user_skills_dir.mkdir(parents=True, exist_ok=True)
        # copytree 整目录(SKILL.md + scripts/ + references/ + assets/ 一并带过来)
        shutil.copytree(skill.skill_dir, dest)
        md = dest / "SKILL.md"
        n_files = sum(1 for _ in dest.rglob("*") if _.is_file())
        if md.exists():
            md.write_text(
                _set_frontmatter_name(md.read_text(encoding="utf-8"), new_name),
                encoding="utf-8",
            )
        return (
            f"[forked '{src}' → .skills/{new_name}/ ({n_files} 个文件,frontmatter name 已设为 "
            f"'{new_name}'),下条消息生效。现在可以编辑 .skills/{new_name}/SKILL.md 改造它]"
        )