diff --git a/AGENTS.md b/AGENTS.md index 61cda54..9f53082 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -97,6 +97,28 @@ skiff fetch superpowers skiff install-external superpowers ``` +### 自定义仓库(Custom Sources) + +公司或团队维护、且一个仓库中包含多个 skill 时,使用命名 custom source: + +```bash +skiff source add company \ + git@git.company.com:platform/agent-skills.git \ + --skills-path skills +skiff list --source company +skiff add company/internal-review -g +``` + +也可以接入已有本地 checkout: + +```bash +skiff source add company --local ~/code/company-skills --skills-path skills +``` + +配置保存在 `~/.config/skiff/config.yaml`,Git source 默认 clone 到 +`~/.local/share/skiff/sources//`。项目 `.skills.yaml` 只记录逻辑 +source 名称,每台机器独立配置实际仓库地址。 + ### 社区(External NPM / GitHub) 推荐使用 Vercel CLI 安装第三方 skill: @@ -143,6 +165,7 @@ registry.yaml ←── skiff add / fetch | ---------------- | ---------------------------------- | ------------------------------- | | **Owned** | `skills//` | 本仓库 commit | | **External Git** | `~/.local/share/skills/externals/` | `skiff fetch` | +| **Custom Source** | `~/.local/share/skiff/sources/` 或本地路径 | `skiff source add/fetch` | | **External NPM** | `node_modules/` | `npx skills add` / `skills-npm` | ### 非 Skill 资料分类 @@ -176,6 +199,8 @@ skills: - name: superpowers source: registry ref: main + - name: internal-review + source: company targets: # 可选,默认 all - cursor diff --git a/skiff/README.md b/skiff/README.md index bb31bf2..c8a44c6 100644 --- a/skiff/README.md +++ b/skiff/README.md @@ -13,7 +13,8 @@ cd /path/to/skills # 本仓库根目录 ## 命令风格 -接口对齐 [Vercel skills CLI](https://github.com/vercel-labs/skills) 的 `add` / `remove`,专用于 **~/.skills 自研 skill**。 +接口对齐 [Vercel skills CLI](https://github.com/vercel-labs/skills) 的 `add` / `remove`, +统一管理 **~/.skills 自研 skill**、命名 custom source 和 registry 外部 skill。 ```bash # 浏览可用自研 skill @@ -56,7 +57,7 @@ skiff bootstrap | 命令 | 说明 | |------|------| -| `skiff list` | 列出自研 skill 与 registry 中的外部 skill | +| `skiff list [--source NAME]` | 列出所有来源或指定 source 中的 skill | | `skiff status [--target all\|cursor\|claude\|codex]` | 安装状态总览 | ### 项目初始化 @@ -112,6 +113,40 @@ skiff select --project ~/code/app # 指定项目 `skiff add ...`。使用 `-a` 限定 Agent 时,该范围会记录在对应的 skill 条目中,后续 `skiff sync` 不会扩散到其他 Agent。 +### Custom source(多-skill 仓库) + +公司或团队维护的仓库通常包含多个 skill。使用命名 source 接入: + +```bash +# Git 仓库,默认 clone 到 ~/.local/share/skiff/sources/company +skiff source add company \ + git@git.company.com:platform/agent-skills.git \ + --ref main \ + --skills-path internal/skills + +# 或接入已有本地仓库 +skiff source add company \ + --local ~/code/company-agent-skills \ + --skills-path skills + +skiff source list +skiff source fetch company +skiff list --source company +skiff add company/code-review -g -a codex +``` + +| 命令 | 说明 | +|------|------| +| `skiff source add [--ref REF] [--checkout PATH] [--skills-path PATH]` | 注册并克隆 Git source | +| `skiff source add --local PATH [--skills-path PATH]` | 接入已有本地仓库 | +| `skiff source list` / `show ` | 查看 source | +| `skiff source fetch ` / `fetch --all` | clone 或 fast-forward 更新 | +| `skiff source remove ` | 移除配置并保留 checkout | + +配置保存在 `~/.config/skiff/config.yaml`。Git/SSH 认证复用本机 Git 配置, +skiff 不保存 token。可以使用 `company/code-review`,也可以使用 +`skiff add code-review --source company`。多个来源包含同名 skill 时,必须明确来源。 + ### 项目级 | 命令 | 说明 | @@ -176,6 +211,7 @@ skiff/ ├── paths.py # 路径常量与 Agent 目标 ├── skills.py # 自研 skill 发现 ├── registry.py # registry.yaml 读写 +├── sources.py # custom source 配置、发现与 Git 管理 ├── project.py # .skills.yaml 管理 ├── symlinks.py # 软链创建/检查/修复 └── yaml_io.py # 轻量 YAML 解析(无第三方依赖) @@ -191,6 +227,8 @@ skiff/ | `SKILLS_DIR` | `~/.skills/skills/` | 自研 skill 目录 | | `REGISTRY_FILE` | `~/.skills/registry.yaml` | 外部 skill 注册表 | | `EXTERNALS_DIR` | `~/.local/share/skills/externals/` | 已 fetch 的外部仓库;新条目按 repo/ref 共享缓存 | +| `CONFIG_FILE` | `~/.config/skiff/config.yaml` | custom source 配置 | +| `SOURCES_DIR` | `~/.local/share/skiff/sources/` | custom Git source 默认 checkout | ## 注意事项 diff --git a/skiff/__init__.py b/skiff/__init__.py index 3e711ae..8a71d91 100644 --- a/skiff/__init__.py +++ b/skiff/__init__.py @@ -1,3 +1,3 @@ """skiff — Agent Skills 安装与管理 CLI。""" -__version__ = "0.4.0" +__version__ = "0.5.0" diff --git a/skiff/cli.py b/skiff/cli.py index 5024aa6..d9c7888 100644 --- a/skiff/cli.py +++ b/skiff/cli.py @@ -16,6 +16,7 @@ from skiff.paths import ( ALL_TARGETS, DRAFTS_DIR, EXTERNALS_DIR, + CONFIG_FILE, SKILLS_DIR, SKILLS_HOME, TEMPLATE_DIR, @@ -38,13 +39,23 @@ from skiff.registry import ( ) from skiff.selector import SkillChoice, select_skills from skiff.skills import ( + list_custom_skills, list_owned_skills, owned_skill_path, resolve_skill_source, + split_skill_spec, skill_description, validate_skill_dir, validate_skill_name, ) +from skiff.sources import ( + fetch_source, + load_sources, + save_sources, + source_checkout, + source_skills_root, + validate_source_name, +) from skiff.yaml_io import safe_dump from skiff.symlinks import check_link, copy_template, create_link, find_repo_root, remove_link @@ -76,7 +87,18 @@ def _collect_skill_names(positional: list[str] | None, flagged: list[str] | None return names -def _ensure_external_fetched(name: str) -> None: +def _ensure_source_fetched(name: str, source: str | None = None) -> None: + name, source = split_skill_spec(name, source) + sources = load_sources() + if source in sources: + root = source_skills_root(source, sources[source]) + if not root.is_dir(): + _print(f"拉取 source: {source}") + fetch_source(source, sources[source]) + return + if source not in (None, "registry"): + return + registry = load_registry() if name not in registry: return @@ -101,9 +123,16 @@ def _ensure_external_fetched(name: str) -> None: ) -def _install_skill(name: str, targets: list[str], project_root: Path | None = None) -> None: - _ensure_external_fetched(name) - skill_path, _ = resolve_skill_source(name) +def _install_skill( + name: str, + targets: list[str], + project_root: Path | None = None, + *, + source: str | None = None, +) -> str: + name, source = split_skill_spec(name, source) + _ensure_source_fetched(name, source) + skill_path, resolved_source = resolve_skill_source(name, source=source) links = [ (target, agent_skill_dir(target, project_root=project_root) / name) for target in targets @@ -124,7 +153,8 @@ def _install_skill(name: str, targets: list[str], project_root: Path | None = No for target, _ in links: scope = "全局" if project_root is None else "项目" - _print(f"已安装 ({scope}/{target}): {name} -> {skill_path}") + _print(f"已安装 ({scope}/{target}): {resolved_source}/{name} -> {skill_path}") + return resolved_source def _list_installed_names(project_root: Path | None, targets: list[str]) -> list[str]: @@ -177,18 +207,32 @@ def cmd_list(args: argparse.Namespace) -> None: ensure_skills_home() owned = list_owned_skills() registry = load_registry() + custom = ( + {} + if args.source in ("owned", "registry") + else list_custom_skills(args.source) + ) - _print("自研 (owned):") - for name in owned: - _print(f" {name}") + if args.source in (None, "owned"): + _print("自研 (owned):") + for name in owned: + _print(f" {name}") - _print("\n外部 (registry):") - if not registry: - _print(" (无)") - else: - for name, entry in registry.items(): - repo = entry.get("repo", "?") - _print(f" {name} ({repo})") + if args.source in (None, "registry"): + _print("\n外部 (registry):") + if not registry: + _print(" (无)") + else: + for name, entry in registry.items(): + repo = entry.get("repo", "?") + _print(f" {name} ({repo})") + + for source, names in custom.items(): + _print(f"\n自定义 ({source}):") + if not names: + _print(" (无,或 source 尚未 fetch)") + for name in names: + _print(f" {name}") def cmd_bootstrap(args: argparse.Namespace) -> None: @@ -200,8 +244,14 @@ def cmd_bootstrap(args: argparse.Namespace) -> None: _print("已安装项目 skill 到所有 agent") -def _installed_links(name: str, targets: list[str], project_root: Path | None = None) -> list[tuple[str, Path, Path]]: - skill_path, _ = resolve_skill_source(name) +def _installed_links( + name: str, + targets: list[str], + project_root: Path | None = None, + *, + source: str | None = None, +) -> list[tuple[str, Path, Path]]: + skill_path, _ = resolve_skill_source(name, source=source) rows: list[tuple[str, Path, Path]] = [] for target in targets: link = agent_skill_dir(target, project_root=project_root) / name @@ -214,16 +264,22 @@ def cmd_status(args: argparse.Namespace) -> None: targets = resolve_agent_args(flatten_agent_args(args.agents)) owned = list_owned_skills() registry = load_registry() - all_names = owned + [n for n in registry if n not in owned] + custom = list_custom_skills() + entries = [("owned", name) for name in owned] + entries.extend(("registry", name) for name in registry) + entries.extend((source, name) for source, names in custom.items() for name in names) _print(f"skills 仓库: {SKILLS_HOME.resolve()}") _print(f"agents: {', '.join(targets)}\n") - for name in all_names: - kind = "owned" if name in owned else "external" - _print(f"[{kind}] {name}") + for source, name in entries: + _print(f"[{source}] {name}") try: - rows = _installed_links(name, targets) + skill_path, _ = resolve_skill_source(name, source=source) + rows = [ + (target, agent_skill_dir(target) / name, skill_path) + for target in targets + ] except SystemExit: _print(" (未 fetch)") continue @@ -256,12 +312,24 @@ def cmd_add(args: argparse.Namespace) -> None: ensure_skills_home() if args.list_available: - _print_available_skills() + if args.source: + cmd_list(argparse.Namespace(source=args.source)) + else: + _print_available_skills() _print("使用 skiff add 安装,或 skiff add -g 全局安装") return if args.all: - names = list_owned_skills() + if args.source: + if args.source == "owned": + names = list_owned_skills() + elif args.source == "registry": + names = list(load_registry()) + else: + _ensure_source_fetched("", args.source) + names = list_custom_skills(args.source)[args.source] + else: + names = list_owned_skills() targets = resolve_agent_args(["*"]) else: names = _collect_skill_names(args.skills, args.skills_flag) @@ -272,8 +340,20 @@ def cmd_add(args: argparse.Namespace) -> None: project_root = None if args.global_scope else _project_root(args.project) for name in names: - validate_skill_name(name) - _install_skill(name, targets, project_root=project_root) + skill_name, source = split_skill_spec(name, args.source) + validate_skill_name(skill_name) + resolved_source = _install_skill( + skill_name, + targets, + project_root=project_root, + source=source, + ) + if project_root is not None: + add_skill_to_manifest( + project_root / ".skills.yaml", + skill_name, + source=resolved_source, + ) def cmd_select(args: argparse.Namespace) -> None: @@ -313,6 +393,10 @@ def cmd_select(args: argparse.Namespace) -> None: for name, entry in registry.items() if name not in collisions ) + choice_sources = { + **{name: "owned" for name in owned_names}, + **{name: "registry" for name in registry if name not in collisions}, + } try: selected = select_skills(choices) @@ -329,7 +413,12 @@ def cmd_select(args: argparse.Namespace) -> None: successful = set(selected & installed) for name in names: try: - _install_skill(name, targets, project_root=project_root) + _install_skill( + name, + targets, + project_root=project_root, + source=choice_sources[name], + ) successful.add(name) except (OSError, subprocess.CalledProcessError, SystemExit) as exc: failures.append((name, str(exc))) @@ -440,6 +529,94 @@ def cmd_fetch(args: argparse.Namespace) -> None: ) +def cmd_source_add(args: argparse.Namespace) -> None: + validate_source_name(args.name) + sources = load_sources() + if args.name in sources: + raise SystemExit(f"source 已存在: {args.name}") + if bool(args.repo) == bool(args.local): + raise SystemExit("必须且只能指定 Git repo 或 --local") + entry: dict[str, str] = {"skills_path": args.skills_path} + if args.local: + local = Path(args.local).expanduser().resolve() + if not local.is_dir(): + raise SystemExit(f"本地仓库不存在: {local}") + entry["local_path"] = str(local) + else: + entry.update({"repo": args.repo, "ref": args.ref}) + if args.checkout: + entry["checkout"] = str(Path(args.checkout).expanduser().resolve()) + # 注册前先校验路径不能逃出仓库。 + source_skills_root(args.name, entry) + if not args.local and not args.no_fetch: + _print(f"克隆 source: {args.name}") + fetch_source(args.name, entry) + sources[args.name] = entry + save_sources(sources) + _print(f"已添加 source: {args.name} ({CONFIG_FILE})") + + +def cmd_source_list(args: argparse.Namespace) -> None: + del args + sources = load_sources() + if not sources: + _print("未配置 custom source") + return + for name, entry in sources.items(): + kind = "local" if entry.get("local_path") else "git" + _print( + f"{name} [{kind}] checkout={source_checkout(name, entry)} " + f"skills={source_skills_root(name, entry)}" + ) + + +def cmd_source_show(args: argparse.Namespace) -> None: + sources = load_sources() + if args.name not in sources: + raise SystemExit(f"未配置 source: {args.name}") + entry = sources[args.name] + _print(f"name: {args.name}") + for key, value in entry.items(): + _print(f"{key}: {value}") + _print(f"checkout: {source_checkout(args.name, entry)}") + _print(f"skills_root: {source_skills_root(args.name, entry)}") + + +def cmd_source_fetch(args: argparse.Namespace) -> None: + sources = load_sources() + names = list(sources) if args.all else [args.name] + if not names or names == [None]: + raise SystemExit("请指定 source 名称,或使用 --all") + for name in names: + if name not in sources: + raise SystemExit(f"未配置 source: {name}") + _print(f"更新 source: {name}") + fetch_source(name, sources[name]) + + +def cmd_source_remove(args: argparse.Namespace) -> None: + sources = load_sources() + if args.name not in sources: + raise SystemExit(f"未配置 source: {args.name}") + entry = sources[args.name] + if args.delete_checkout and entry.get("local_path"): + raise SystemExit("不会删除 --local 指定的仓库") + checkout = source_checkout(args.name, entry) + if args.delete_checkout and ( + checkout in (Path("/"), Path.home().resolve()) or not (checkout / ".git").is_dir() + ): + raise SystemExit(f"拒绝删除不安全或非 Git checkout: {checkout}") + sources.pop(args.name) + save_sources(sources) + _print(f"已移除 source 配置: {args.name}") + if args.delete_checkout: + import shutil + + if checkout.is_dir(): + shutil.rmtree(checkout) + _print(f"已删除 checkout(不可恢复): {checkout}") + + def cmd_enable(args: argparse.Namespace) -> None: _warn_deprecated("skiff enable", "skiff add ") ensure_skills_home() @@ -447,13 +624,10 @@ def cmd_enable(args: argparse.Namespace) -> None: root = _project_root(args.project) manifest_path = root / ".skills.yaml" - registry = load_registry() - if args.name in registry: - extra = {"source": "registry", "ref": registry[args.name].get("ref", "main")} - add_skill_to_manifest(manifest_path, args.name, source="registry", extra=extra) - else: - owned_skill_path(args.name) - add_skill_to_manifest(manifest_path, args.name, source="owned") + name, source = split_skill_spec(args.name) + _ensure_source_fetched(name, source) + _, resolved_source = resolve_skill_source(name, source=source) + add_skill_to_manifest(manifest_path, name, source=resolved_source) targets = resolve_agent_args(flatten_agent_args(args.agents)) _, data = load_manifest(manifest_path) @@ -461,8 +635,8 @@ def cmd_enable(args: argparse.Namespace) -> None: if manifest_targets: targets = [t for t in targets if t in manifest_targets] - _install_skill(args.name, targets, project_root=root) - _print(f"已启用项目 skill: {args.name} @ {root}") + _install_skill(name, targets, project_root=root, source=resolved_source) + _print(f"已启用项目 skill: {resolved_source}/{name} @ {root}") def cmd_disable(args: argparse.Namespace) -> None: @@ -492,6 +666,7 @@ def cmd_sync(args: argparse.Namespace) -> None: for entry in iter_manifest_skills(data): name = entry["name"] + _ensure_source_fetched(name, entry.get("source")) skill_path, _ = resolve_manifest_skill(entry) skill_targets = targets if entry.get("targets"): @@ -603,7 +778,7 @@ def cmd_doctor(args: argparse.Namespace) -> None: issues += 1 for name in list_owned_skills(): - for target, link, expected in _installed_links(name, targets): + for target, link, expected in _installed_links(name, targets, source="owned"): status = check_link(link, expected) if status.ok: continue @@ -623,6 +798,16 @@ def cmd_doctor(args: argparse.Namespace) -> None: _err(f"✗ 外部 skill 未 fetch: {name}") issues += 1 + for source, entry in load_sources().items(): + checkout = source_checkout(source, entry) + root = source_skills_root(source, entry) + if not checkout.is_dir(): + _err(f"✗ source checkout 不存在: {source} -> {checkout}") + issues += 1 + elif not root.is_dir(): + _err(f"✗ source skills_path 不存在: {source} -> {root}") + issues += 1 + if issues == 0: _print("\n全部正常") else: @@ -668,7 +853,8 @@ def build_parser() -> argparse.ArgumentParser: ) p_bootstrap.set_defaults(func=cmd_bootstrap) - p_list = sub.add_parser("list", help="列出 ~/.skills 中的 skill 目录") + p_list = sub.add_parser("list", help="列出所有 source 中的 skill") + p_list.add_argument("--source", help="只列出指定来源(owned、registry 或 custom source)") p_list.set_defaults(func=cmd_list) p_status = sub.add_parser("status", help="安装状态总览") @@ -684,6 +870,7 @@ def build_parser() -> argparse.ArgumentParser: p_add.add_argument("-s", "--skill", dest="skills_flag", action="append", metavar="SKILL") p_add.add_argument("--list", dest="list_available", action="store_true", help="列出可用自研 skill,不安装") p_add.add_argument("--all", action="store_true", help="安装全部自研 skill 到全部 agent") + p_add.add_argument("--source", help="指定 skill 来源(也可使用 source/name)") _add_common_flags(p_add) p_add.set_defaults(func=cmd_add) @@ -728,6 +915,40 @@ def build_parser() -> argparse.ArgumentParser: p_fetch.add_argument("name", help="registry 名称") p_fetch.set_defaults(func=cmd_fetch) + p_source = sub.add_parser("source", help="管理包含多个 skills 的自定义仓库") + source_sub = p_source.add_subparsers(dest="source_command", required=True) + + p_source_add = source_sub.add_parser("add", help="注册 Git 或本地 skill source") + p_source_add.add_argument("name", help="source 名称") + p_source_add.add_argument("repo", nargs="?", help="Git 仓库 URL") + p_source_add.add_argument("--local", help="已有本地仓库路径") + p_source_add.add_argument("--ref", default="main", help="Git 分支或 tag(默认 main)") + p_source_add.add_argument("--checkout", help="Git checkout 路径") + p_source_add.add_argument("--skills-path", default="skills", help="仓库内 skills 父目录") + p_source_add.add_argument("--no-fetch", action="store_true", help="仅写配置,不立即 clone") + p_source_add.set_defaults(func=cmd_source_add) + + p_source_list = source_sub.add_parser("list", help="列出已配置 source") + p_source_list.set_defaults(func=cmd_source_list) + + p_source_show = source_sub.add_parser("show", help="显示 source 详情") + p_source_show.add_argument("name") + p_source_show.set_defaults(func=cmd_source_show) + + p_source_fetch = source_sub.add_parser("fetch", help="克隆或更新 source") + p_source_fetch.add_argument("name", nargs="?") + p_source_fetch.add_argument("--all", action="store_true") + p_source_fetch.set_defaults(func=cmd_source_fetch) + + p_source_remove = source_sub.add_parser("remove", help="移除 source 配置") + p_source_remove.add_argument("name") + p_source_remove.add_argument( + "--delete-checkout", + action="store_true", + help="同时永久删除 skiff 管理的 checkout", + ) + p_source_remove.set_defaults(func=cmd_source_remove) + p_enable = sub.add_parser("enable", help=argparse.SUPPRESS) p_enable.add_argument("name") p_enable.add_argument("-a", "--agent", dest="agents", nargs="+", action="append") diff --git a/skiff/paths.py b/skiff/paths.py index a40af1e..11089b9 100644 --- a/skiff/paths.py +++ b/skiff/paths.py @@ -11,6 +11,8 @@ TEMPLATE_DIR = SKILLS_DIR / "_template" DRAFTS_DIR = SKILLS_HOME / ".drafts" REGISTRY_FILE = SKILLS_HOME / "registry.yaml" EXTERNALS_DIR = HOME / ".local" / "share" / "skills" / "externals" +CONFIG_FILE = HOME / ".config" / "skiff" / "config.yaml" +SOURCES_DIR = HOME / ".local" / "share" / "skiff" / "sources" PROJECT_MANIFEST = ".skills.yaml" AGENT_GLOBAL: dict[str, Path] = { diff --git a/skiff/project.py b/skiff/project.py index 0fa33f6..9fe11e5 100644 --- a/skiff/project.py +++ b/skiff/project.py @@ -68,12 +68,13 @@ def add_skill_to_manifest( item: dict[str, Any] = {"name": name, "source": source} if extra: item.update(extra) - for index, entry in enumerate(entries): - if entry["name"] == name: - entries[index] = item - break - else: + existing_index = next((i for i, entry in enumerate(entries) if entry["name"] == name), None) + if existing_index is None: entries.append(item) + elif entries[existing_index] == item: + return + else: + entries[existing_index] = item data["skills"] = [_entry_to_yaml(e) for e in entries] save_manifest(file_path, data) diff --git a/skiff/skills.py b/skiff/skills.py index 24cbb1d..27cf381 100644 --- a/skiff/skills.py +++ b/skiff/skills.py @@ -7,6 +7,7 @@ from pathlib import Path from skiff.paths import SKILLS_DIR, TEMPLATE_DIR, ensure_skills_home from skiff.registry import external_skill_path, load_registry +from skiff.sources import list_source_skills, load_sources, source_skills_root def list_owned_skills() -> list[str]: @@ -31,26 +32,81 @@ def owned_skill_path(name: str) -> Path: return path +def split_skill_spec(spec: str, source: str | None = None) -> tuple[str, str | None]: + if "/" not in spec: + return spec, source + qualified_source, name = spec.split("/", 1) + if not qualified_source or not name or "/" in name: + raise SystemExit(f"skill 限定名称无效: {spec!r}(应为 source/name)") + if source and source != qualified_source: + raise SystemExit( + f"skill 来源冲突: {spec!r} 与 --source {source!r} 不一致" + ) + return name, qualified_source + + +def list_custom_skills(source: str | None = None) -> dict[str, list[str]]: + sources = load_sources() + if source: + if source not in sources: + raise SystemExit(f"未配置 source: {source}") + return {source: list_source_skills(source, sources[source])} + return {name: list_source_skills(name, entry) for name, entry in sources.items()} + + def resolve_skill_source(name: str, *, source: str | None = None) -> tuple[Path, str]: - """返回 (skill_path, kind),kind 为 owned 或 external。""" + """返回 (skill_path, source),未指定来源时拒绝同名歧义。""" ensure_skills_home() + name, source = split_skill_spec(name, source) owned = SKILLS_DIR / name - if source in (None, "owned") and (owned / "SKILL.md").is_file(): + if source == "owned": + if not (owned / "SKILL.md").is_file(): + raise SystemExit(f"owned source 中找不到 skill: {name}") return owned, "owned" registry = load_registry() - if source in (None, "registry") and name in registry: + if source == "registry": + if name not in registry: + raise SystemExit(f"registry 中找不到 skill: {name}") path = external_skill_path(name, registry[name]) if not (path / "SKILL.md").is_file(): raise SystemExit( f"外部 skill {name!r} 尚未 fetch 或 path 中缺少 SKILL.md。" f"请运行: skiff fetch {name}" ) - return path, "external" + return path, "registry" + sources = load_sources() + if source: + if source not in sources: + raise SystemExit( + f"项目依赖 source {source!r},但本机尚未配置。" + f"请运行: skiff source add {source} " + ) + path = source_skills_root(source, sources[source]) / name + if not (path / "SKILL.md").is_file(): + raise SystemExit(f"source {source!r} 中找不到 skill: {name}") + return path, source + + candidates: list[tuple[Path, str]] = [] if (owned / "SKILL.md").is_file(): - return owned, "owned" + candidates.append((owned, "owned")) + if name in registry: + candidates.append((external_skill_path(name, registry[name]), "registry")) + for source_name, entry in sources.items(): + path = source_skills_root(source_name, entry) / name + if (path / "SKILL.md").is_file(): + candidates.append((path, source_name)) + + if len(candidates) > 1: + choices = ", ".join(f"{candidate_source}/{name}" for _, candidate_source in candidates) + raise SystemExit(f"skill 名称存在多个来源,请明确指定: {choices}") + if candidates: + path, resolved_source = candidates[0] + if resolved_source == "registry" and not path.exists(): + raise SystemExit(f"外部 skill {name!r} 尚未 fetch。请先运行: skiff fetch {name}") + return path, resolved_source raise SystemExit(f"找不到 skill: {name}") diff --git a/skiff/sources.py b/skiff/sources.py new file mode 100644 index 0000000..bbc9811 --- /dev/null +++ b/skiff/sources.py @@ -0,0 +1,106 @@ +"""命名 custom skill source 的配置、发现与 Git 管理。""" + +from __future__ import annotations + +import subprocess +from pathlib import Path +from typing import Any + +from skiff import yaml_io +from skiff.paths import CONFIG_FILE, SOURCES_DIR + +RESERVED_SOURCES = {"owned", "registry"} + + +def validate_source_name(name: str) -> None: + from skiff.skills import validate_skill_name + + validate_skill_name(name) + if name in RESERVED_SOURCES: + raise SystemExit(f"source 名称为保留字: {name}") + + +def load_sources(path: Path | None = None) -> dict[str, dict[str, Any]]: + path = path or CONFIG_FILE + if not path.is_file(): + return {} + data = yaml_io.safe_load(path.read_text(encoding="utf-8")) or {} + if not isinstance(data, dict): + raise SystemExit(f"skiff 配置格式错误: {path}") + raw = data.get("sources", {}) + if raw is None: + return {} + if not isinstance(raw, dict): + raise SystemExit(f"skiff 配置 sources 格式错误: {path}") + return {str(name): entry for name, entry in raw.items() if isinstance(entry, dict)} + + +def save_sources(sources: dict[str, dict[str, Any]], path: Path | None = None) -> None: + path = path or CONFIG_FILE + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + yaml_io.safe_dump({"sources": sources}, allow_unicode=True, sort_keys=False), + encoding="utf-8", + ) + + +def source_checkout(name: str, entry: dict[str, Any]) -> Path: + local_path = entry.get("local_path") + if local_path: + return Path(str(local_path)).expanduser().resolve() + checkout = entry.get("checkout") + if checkout: + return Path(str(checkout)).expanduser().resolve() + return (SOURCES_DIR / name).resolve() + + +def source_skills_root(name: str, entry: dict[str, Any]) -> Path: + checkout = source_checkout(name, entry) + raw_subpath = str(entry.get("skills_path", "skills") or "skills") + subpath = Path(raw_subpath) + if subpath.is_absolute(): + raise SystemExit(f"source {name!r} 的 skills_path 必须是仓库内相对路径") + root = (checkout / subpath).resolve() + try: + root.relative_to(checkout) + except ValueError: + raise SystemExit(f"source {name!r} 的 skills_path 不能超出仓库目录") from None + return root + + +def list_source_skills(name: str, entry: dict[str, Any]) -> list[str]: + root = source_skills_root(name, entry) + if not root.is_dir(): + return [] + return [ + item.name + for item in sorted(root.iterdir()) + if item.is_dir() and not item.name.startswith("_") and (item / "SKILL.md").is_file() + ] + + +def fetch_source(name: str, entry: dict[str, Any]) -> Path: + if entry.get("local_path"): + checkout = source_checkout(name, entry) + if not checkout.is_dir(): + raise SystemExit(f"source {name!r} 的本地仓库不存在: {checkout}") + return checkout + + repo = entry.get("repo") + if not repo: + raise SystemExit(f"source {name!r} 缺少 repo 或 local_path") + checkout = source_checkout(name, entry) + ref = str(entry.get("ref", "main") or "main") + if checkout.exists(): + if not (checkout / ".git").exists(): + raise SystemExit(f"source checkout 已存在但不是 Git 仓库: {checkout}") + subprocess.run(["git", "-C", str(checkout), "fetch", "--all", "--tags"], check=True) + subprocess.run(["git", "-C", str(checkout), "checkout", ref], check=True) + subprocess.run(["git", "-C", str(checkout), "pull", "--ff-only"], check=True) + else: + checkout.parent.mkdir(parents=True, exist_ok=True) + subprocess.run( + ["git", "clone", "--branch", ref, "--", str(repo), str(checkout)], + check=True, + ) + return checkout diff --git a/tests/test_custom_sources.py b/tests/test_custom_sources.py new file mode 100644 index 0000000..1befa54 --- /dev/null +++ b/tests/test_custom_sources.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def write_skill(root: Path, name: str) -> Path: + skill = root / name + skill.mkdir(parents=True) + skill.joinpath("SKILL.md").write_text( + f"---\nname: {name}\ndescription: >-\n" + f" 测试 {name} skill。\n---\n\n# {name}\n", + encoding="utf-8", + ) + return skill + + +class CustomSourceTests(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + self.home = Path(self.temp_dir.name) + self.skills_home = self.home / ".skills" + (self.skills_home / "skills").mkdir(parents=True) + (self.skills_home / "registry.yaml").write_text("", encoding="utf-8") + + def tearDown(self) -> None: + self.temp_dir.cleanup() + + def run_skiff(self, *args: str) -> subprocess.CompletedProcess[str]: + env = os.environ.copy() + env["HOME"] = str(self.home) + env["PYTHONPATH"] = str(REPO_ROOT) + return subprocess.run( + [sys.executable, "-m", "skiff", *args], + cwd=REPO_ROOT, + env=env, + text=True, + capture_output=True, + check=False, + ) + + def test_local_source_lists_and_installs_qualified_skill(self) -> None: + company = self.home / "company" + expected = write_skill(company / "internal" / "skills", "code-review") + + added = self.run_skiff( + "source", + "add", + "company", + "--local", + str(company), + "--skills-path", + "internal/skills", + ) + listed = self.run_skiff("list", "--source", "company") + installed = self.run_skiff("add", "company/code-review", "-g", "-a", "codex") + + self.assertEqual(added.returncode, 0, added.stderr) + self.assertEqual(listed.returncode, 0, listed.stderr) + self.assertIn("code-review", listed.stdout) + self.assertEqual(installed.returncode, 0, installed.stderr) + link = self.home / ".codex" / "skills" / "code-review" + self.assertTrue(link.is_symlink()) + self.assertEqual(link.resolve(), expected.resolve()) + config = self.home / ".config" / "skiff" / "config.yaml" + self.assertIn("company:", config.read_text(encoding="utf-8")) + + def test_project_add_persists_resolved_source_in_manifest(self) -> None: + company = self.home / "company" + expected = write_skill(company / "skills", "code-review") + project = self.home / "project" + project.mkdir() + added = self.run_skiff("source", "add", "company", "--local", str(company)) + self.assertEqual(added.returncode, 0, added.stderr) + + result = self.run_skiff( + "add", + "company/code-review", + "--project", + str(project), + "-a", + "codex", + ) + + self.assertEqual(result.returncode, 0, result.stderr) + manifest = project.joinpath(".skills.yaml").read_text(encoding="utf-8") + self.assertIn('name: "code-review"', manifest) + self.assertIn("source: company", manifest) + self.assertEqual( + (project / ".agents" / "skills" / "code-review").resolve(), + expected.resolve(), + ) + + def test_project_add_updates_manifest_when_source_changes(self) -> None: + write_skill(self.skills_home / "skills", "code-review") + company = self.home / "company" + write_skill(company / "skills", "code-review") + project = self.home / "project" + project.mkdir() + project.joinpath(".skills.yaml").write_text( + "skills:\n - code-review\n", + encoding="utf-8", + ) + added = self.run_skiff("source", "add", "company", "--local", str(company)) + self.assertEqual(added.returncode, 0, added.stderr) + + result = self.run_skiff( + "add", + "company/code-review", + "--project", + str(project), + "-a", + "codex", + ) + + self.assertEqual(result.returncode, 0, result.stderr) + manifest = project.joinpath(".skills.yaml").read_text(encoding="utf-8") + self.assertIn("source: company", manifest) + + def test_unqualified_duplicate_requires_explicit_source(self) -> None: + write_skill(self.skills_home / "skills", "code-review") + company = self.home / "company" + write_skill(company / "skills", "code-review") + added = self.run_skiff("source", "add", "company", "--local", str(company)) + self.assertEqual(added.returncode, 0, added.stderr) + + result = self.run_skiff("add", "code-review", "-g", "-a", "codex") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("owned/code-review", result.stderr) + self.assertIn("company/code-review", result.stderr) + + def test_sync_uses_manifest_source(self) -> None: + company = self.home / "company" + expected = write_skill(company / "skills", "code-review") + project = self.home / "project" + project.mkdir() + project.joinpath(".skills.yaml").write_text( + "skills:\n - name: code-review\n source: company\n" + "targets:\n - codex\n", + encoding="utf-8", + ) + added = self.run_skiff("source", "add", "company", "--local", str(company)) + self.assertEqual(added.returncode, 0, added.stderr) + + result = self.run_skiff("sync", "--project", str(project), "-a", "codex") + + self.assertEqual(result.returncode, 0, result.stderr) + link = project / ".agents" / "skills" / "code-review" + self.assertEqual(link.resolve(), expected.resolve()) + + def test_sync_reports_missing_machine_source(self) -> None: + project = self.home / "project" + project.mkdir() + project.joinpath(".skills.yaml").write_text( + "skills:\n - name: code-review\n source: company\n", + encoding="utf-8", + ) + + result = self.run_skiff("sync", "--project", str(project), "-a", "codex") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("本机尚未配置", result.stderr) + self.assertIn("skiff source add company", result.stderr) + + def test_git_source_is_cloned_to_default_checkout(self) -> None: + upstream = self.home / "upstream" + write_skill(upstream / "skills", "release-check") + subprocess.run(["git", "init", "-b", "main", str(upstream)], check=True, capture_output=True) + subprocess.run(["git", "-C", str(upstream), "add", "."], check=True) + subprocess.run( + [ + "git", + "-C", + str(upstream), + "-c", + "user.name=Test", + "-c", + "user.email=test@example.com", + "commit", + "-m", + "initial", + ], + check=True, + capture_output=True, + ) + + result = self.run_skiff("source", "add", "company", str(upstream)) + + self.assertEqual(result.returncode, 0, result.stderr) + checkout = self.home / ".local" / "share" / "skiff" / "sources" / "company" + self.assertTrue((checkout / ".git").is_dir()) + self.assertTrue((checkout / "skills" / "release-check" / "SKILL.md").is_file()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_select.py b/tests/test_select.py index 0be534a..923a84b 100644 --- a/tests/test_select.py +++ b/tests/test_select.py @@ -139,7 +139,7 @@ class SelectCommandTests(unittest.TestCase): patch.object( cli, "_install_skill", - side_effect=lambda name, targets, project_root: installed.append(name), + side_effect=lambda name, targets, project_root, **kwargs: installed.append(name), ), ): cli.cmd_select(args) @@ -195,7 +195,7 @@ class SelectCommandTests(unittest.TestCase): links["claude"].mkdir(parents=True) with ( - patch.object(cli, "_ensure_external_fetched"), + patch.object(cli, "_ensure_source_fetched"), patch.object(cli, "resolve_skill_source", return_value=(skill, "owned")), patch.object( cli,