diff --git a/README.md b/README.md index 823b63d..c4a251a 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ cd ~/.skills ./install.sh # 安装 CLI，并将 skiff 项目 skill 安装到所有 Agent skiff install declarative-openspec-loop +skiff select # 交互式选择并批量安装 skiff list skiff status ``` diff --git a/registry.yaml b/registry.yaml index a4a9006..49edc8a 100644 --- a/registry.yaml +++ b/registry.yaml @@ -4,6 +4,9 @@ # repo: # ref: (default: main) # path: (default: .) +# description: (optional, shown by `skiff select`) +# tags: (optional) +# - # # Example: # example-skills: diff --git a/skiff/README.md b/skiff/README.md index 4b12264..bb31bf2 100644 --- a/skiff/README.md +++ b/skiff/README.md @@ -70,6 +70,7 @@ skiff bootstrap | 命令 | 说明 | |------|------| | `skiff add [--global] [-a AGENT...] [-y]` | 安装到 Agent 目录（软链） | +| `skiff select [--global] [-a AGENT...]` | 打开终端多选界面，批量安装 skill | | `skiff remove [--global] [-a AGENT...] [-y]` | 移除软链（`rm` / `r` 别名） | | `skiff add --list` | 列出可用自研 skill | | `skiff publish [paths] -m MSG [--push]` | 在 ~/.skills 内 git add/commit/push | @@ -89,9 +90,28 @@ skiff bootstrap | 命令 | 说明 | |------|------| | `skiff registry add [--ref main] [--path .]` | 写入 `registry.yaml` | -| `skiff fetch ` | 克隆/更新到 `~/.local/share/skills/externals//` | +| `skiff fetch ` | 克隆或更新外部仓库缓存 | | `skiff add [-g] [-a AGENT...]` | 安装 registry 中的外部 skill（缺失时自动 fetch） | +`registry.yaml` 条目可额外提供 `description` 和 `tags`。`description` +会显示在 `skiff select` 的候选列表中。同一 `repo` 与 `ref` 下的多个 skill +共享一份 Git checkout，再通过各自的 `path` 定位目录。 + +### 交互式批量安装 + +```bash +skiff select # 当前项目，全部 Agent +skiff select -a codex # 当前项目，仅 Codex +skiff select -g # 全局安装 +skiff select --project ~/code/app # 指定项目 +``` + +使用方向键移动、空格勾选、`/` 搜索、Enter 安装，按 `q` 或 Esc +取消。已经安装到目标范围的 skill 默认勾选；取消勾选不会卸载已有 skill。 +项目模式会把成功选择的项目写入 `.skills.yaml`。非交互环境请使用 +`skiff add ...`。使用 `-a` 限定 Agent 时，该范围会记录在对应的 +skill 条目中，后续 `skiff sync` 不会扩散到其他 Agent。 + ### 项目级 | 命令 | 说明 | @@ -170,7 +190,7 @@ skiff/ | `SKILLS_HOME` | `~/.skills` | skills 仓库（软链） | | `SKILLS_DIR` | `~/.skills/skills/` | 自研 skill 目录 | | `REGISTRY_FILE` | `~/.skills/registry.yaml` | 外部 skill 注册表 | -| `EXTERNALS_DIR` | `~/.local/share/skills/externals/` | 已 fetch 的外部仓库 | +| `EXTERNALS_DIR` | `~/.local/share/skills/externals/` | 已 fetch 的外部仓库；新条目按 repo/ref 共享缓存 | ## 注意事项 diff --git a/skiff/cli.py b/skiff/cli.py index 98a744e..5024aa6 100644 --- a/skiff/cli.py +++ b/skiff/cli.py @@ -30,7 +30,13 @@ from skiff.project import ( resolve_manifest_skill, save_manifest, ) -from skiff.registry import external_skill_path, load_registry, save_registry +from skiff.registry import ( + external_checkout_path, + external_skill_path, + load_registry, + save_registry, +) +from skiff.selector import SkillChoice, select_skills from skiff.skills import ( list_owned_skills, owned_skill_path, @@ -77,13 +83,17 @@ def _ensure_external_fetched(name: str) -> None: entry = registry[name] path = external_skill_path(name, entry) - if path.exists(): + if (path / "SKILL.md").is_file(): return repo = entry["repo"] ref = entry.get("ref", "main") - dest = EXTERNALS_DIR / name - EXTERNALS_DIR.mkdir(parents=True, exist_ok=True) + dest = external_checkout_path(name, entry) + dest.parent.mkdir(parents=True, exist_ok=True) + if dest.exists(): + raise SystemExit( + f"外部仓库已存在但 skill 路径无效: {external_skill_path(name, entry)}" + ) _print(f"拉取外部 skill: {name}") subprocess.run( ["git", "clone", "--branch", ref, "--", repo, str(dest)], @@ -94,9 +104,25 @@ 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) - for target in targets: - link = agent_skill_dir(target, project_root=project_root) / name - create_link(link, skill_path) + links = [ + (target, agent_skill_dir(target, project_root=project_root) / name) + for target in targets + ] + original: list[tuple[Path, str | None]] = [] + try: + for _, link in links: + previous = str(link.readlink()) if link.is_symlink() else None + original.append((link, previous)) + create_link(link, skill_path) + except Exception: + for link, previous in reversed(original): + if link.is_symlink(): + link.unlink() + if previous is not None: + link.symlink_to(previous) + raise + + for target, _ in links: scope = "全局" if project_root is None else "项目" _print(f"已安装 ({scope}/{target}): {name} -> {skill_path}") @@ -115,6 +141,19 @@ def _list_installed_names(project_root: Path | None, targets: list[str]) -> list return sorted(names) +def _list_fully_installed_names( + project_root: Path | None, + targets: list[str], +) -> list[str]: + installed_by_target = [ + set(_list_installed_names(project_root, [target])) + for target in targets + ] + if not installed_by_target: + return [] + return sorted(set.intersection(*installed_by_target)) + + def _remove_skill( name: str, targets: list[str], @@ -237,6 +276,96 @@ def cmd_add(args: argparse.Namespace) -> None: _install_skill(name, targets, project_root=project_root) +def cmd_select(args: argparse.Namespace) -> None: + ensure_skills_home() + if not sys.stdin.isatty() or not sys.stdout.isatty(): + raise SystemExit( + "`skiff select` 需要交互式终端；非交互环境请使用 `skiff add ...`" + ) + + targets = resolve_agent_args(flatten_agent_args(args.agents)) + project_root = None if args.global_scope else _project_root(args.project) + installed = set(_list_fully_installed_names(project_root, targets)) + registry = load_registry() + owned_names = list_owned_skills() + for name in [*owned_names, *registry]: + validate_skill_name(name) + collisions = set(owned_names) & set(registry) + for name in sorted(collisions): + _err(f"警告: registry 条目与 owned skill 同名，已忽略 external: {name}") + + choices = [ + SkillChoice( + name=name, + kind="owned", + description=skill_description(name) or "", + installed=name in installed, + ) + for name in owned_names + ] + choices.extend( + SkillChoice( + name=name, + kind="external", + description=str(entry.get("description", "")), + installed=name in installed, + ) + for name, entry in registry.items() + if name not in collisions + ) + + try: + selected = select_skills(choices) + except (RuntimeError, OSError) as exc: + raise SystemExit(str(exc)) from exc + if selected is None: + _print("已取消，未修改环境") + return + + names = sorted(selected - installed) + + failures: list[tuple[str, str]] = [] + manifest_path = project_root / ".skills.yaml" if project_root else None + successful = set(selected & installed) + for name in names: + try: + _install_skill(name, targets, project_root=project_root) + successful.add(name) + except (OSError, subprocess.CalledProcessError, SystemExit) as exc: + failures.append((name, str(exc))) + _err(f"✗ {name}: {exc}") + + if manifest_path is not None: + entry_targets = targets if args.agents else None + for name in sorted(successful): + entry = registry.get(name) + if name in owned_names: + add_skill_to_manifest( + manifest_path, + name, + source="owned", + extra={"targets": entry_targets} if entry_targets else None, + ) + else: + extra = {"ref": entry.get("ref", "main")} + if entry_targets: + extra["targets"] = entry_targets + add_skill_to_manifest( + manifest_path, + name, + source="registry", + extra=extra, + ) + + installed_count = len(names) - len(failures) + if not names: + _print("没有需要安装的新 skill") + else: + _print(f"安装完成: {installed_count} 成功，{len(failures)} 失败") + if failures: + raise SystemExit(1) + + def cmd_remove(args: argparse.Namespace) -> None: ensure_skills_home() project_root = None if args.global_scope else _project_root(args.project) @@ -294,7 +423,7 @@ def cmd_fetch(args: argparse.Namespace) -> None: entry = registry[args.name] repo = entry["repo"] ref = entry.get("ref", "main") - dest = EXTERNALS_DIR / args.name + dest = external_checkout_path(args.name, entry) EXTERNALS_DIR.mkdir(parents=True, exist_ok=True) @@ -364,7 +493,10 @@ def cmd_sync(args: argparse.Namespace) -> None: for entry in iter_manifest_skills(data): name = entry["name"] skill_path, _ = resolve_manifest_skill(entry) - for target in targets: + skill_targets = targets + if entry.get("targets"): + skill_targets = [target for target in targets if target in entry["targets"]] + for target in skill_targets: link = agent_skill_dir(target, project_root=root) / name create_link(link, skill_path) _print(f"已同步: {link} -> {skill_path}") @@ -555,6 +687,13 @@ def build_parser() -> argparse.ArgumentParser: _add_common_flags(p_add) p_add.set_defaults(func=cmd_add) + p_select = sub.add_parser( + "select", + help="交互式选择并批量安装 skill", + ) + _add_common_flags(p_select) + p_select.set_defaults(func=cmd_select) + p_remove = sub.add_parser( "remove", aliases=["rm", "r"], diff --git a/skiff/project.py b/skiff/project.py index 5070157..0fa33f6 100644 --- a/skiff/project.py +++ b/skiff/project.py @@ -65,13 +65,15 @@ def add_skill_to_manifest( file_path = path entries = [normalize_skill_entry(e) for e in data.get("skills", [])] - if any(e["name"] == name for e in entries): - return - item: dict[str, Any] = {"name": name, "source": source} if extra: item.update(extra) - entries.append(item) + for index, entry in enumerate(entries): + if entry["name"] == name: + entries[index] = item + break + else: + entries.append(item) data["skills"] = [_entry_to_yaml(e) for e in entries] save_manifest(file_path, data) diff --git a/skiff/registry.py b/skiff/registry.py index 50595b3..9af2857 100644 --- a/skiff/registry.py +++ b/skiff/registry.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib from pathlib import Path from typing import Any @@ -25,9 +26,33 @@ def save_registry(data: dict[str, dict[str, Any]], path: Path | None = None) -> path.write_text(yaml_io.safe_dump(data, allow_unicode=True, sort_keys=False), encoding="utf-8") -def external_skill_path(name: str, entry: dict[str, Any] | None = None) -> Path: +def external_repo_path(entry: dict[str, Any]) -> Path: + """Return the shared checkout path for a repo/ref pair.""" from skiff.paths import EXTERNALS_DIR + repo = str(entry.get("repo", "")) + ref = str(entry.get("ref", "main")) + digest = hashlib.sha256(f"{repo}\0{ref}".encode()).hexdigest()[:16] + return EXTERNALS_DIR / "_repos" / digest + + +def external_checkout_path(name: str, entry: dict[str, Any]) -> Path: + """Prefer an existing pre-shared-cache checkout for compatibility.""" + from skiff.paths import EXTERNALS_DIR + + legacy = EXTERNALS_DIR / name + return legacy if legacy.is_dir() else external_repo_path(entry) + + +def external_skill_path(name: str, entry: dict[str, Any] | None = None) -> Path: entry = entry or load_registry().get(name, {}) subpath = entry.get("path", ".") or "." - return (EXTERNALS_DIR / name / subpath).resolve() + checkout = external_checkout_path(name, entry).resolve() + skill_path = (checkout / subpath).resolve() + try: + skill_path.relative_to(checkout) + except ValueError as exc: + raise SystemExit( + f"registry 条目 {name!r} 的 path 超出外部仓库: {subpath!r}" + ) from exc + return skill_path diff --git a/skiff/selector.py b/skiff/selector.py new file mode 100644 index 0000000..6b5b211 --- /dev/null +++ b/skiff/selector.py @@ -0,0 +1,113 @@ +"""Terminal multi-select UI for skills.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable, Iterable + + +@dataclass(frozen=True) +class SkillChoice: + name: str + kind: str + description: str = "" + installed: bool = False + + +def filter_choices(choices: Iterable[SkillChoice], query: str) -> list[SkillChoice]: + needle = query.casefold().strip() + if not needle: + return list(choices) + return [ + choice + for choice in choices + if needle in choice.name.casefold() + or needle in choice.kind.casefold() + or needle in choice.description.casefold() + ] + + +def select_skills( + choices: list[SkillChoice], + *, + wrapper: Callable[..., set[str] | None] | None = None, +) -> set[str] | None: + """Open the selector. Return names, or None when cancelled.""" + if not choices: + return set() + + try: + import curses + except ImportError as exc: + raise RuntimeError("当前 Python 环境不支持 curses，无法打开交互界面") from exc + + selected = {choice.name for choice in choices if choice.installed} + + def draw(stdscr: object) -> set[str] | None: + curses.curs_set(0) + stdscr.keypad(True) + current = 0 + query = "" + + while True: + visible = filter_choices(choices, query) + current = min(current, max(0, len(visible) - 1)) + stdscr.erase() + height, width = stdscr.getmaxyx() + header = "↑/↓ 移动 Space 勾选 / 搜索 Enter 安装 q 取消" + stdscr.addnstr(0, 0, header, max(0, width - 1)) + if query: + stdscr.addnstr(1, 0, f"搜索: {query}", max(0, width - 1)) + + rows = max(1, height - 4) + start = max(0, current - rows + 1) + for row, choice in enumerate(visible[start : start + rows], start=2): + index = start + row - 2 + mark = "x" if choice.name in selected else " " + suffix = f" · {choice.description}" if choice.description else "" + line = f"[{mark}] {choice.name} {choice.kind}{suffix}" + attr = curses.A_REVERSE if index == current else curses.A_NORMAL + stdscr.addnstr(row, 0, line, max(0, width - 1), attr) + + footer = f"已选择 {len(selected)} 项" + stdscr.addnstr(height - 1, 0, footer, max(0, width - 1)) + stdscr.refresh() + key = stdscr.get_wch() + + if key in ("q", "Q", "\x1b"): + return None + if key in ("\n", "\r") or key == curses.KEY_ENTER: + return set(selected) + if key == curses.KEY_UP and visible: + current = (current - 1) % len(visible) + elif key == curses.KEY_DOWN and visible: + current = (current + 1) % len(visible) + elif key == " " and visible: + name = visible[current].name + selected.symmetric_difference_update({name}) + elif key == "/": + curses.curs_set(1) + query = _read_query(stdscr, curses, width) + curses.curs_set(0) + current = 0 + + runner = wrapper or curses.wrapper + return runner(draw) + + +def _read_query(stdscr: object, curses: object, width: int) -> str: + query = "" + while True: + stdscr.move(1, 0) + stdscr.clrtoeol() + stdscr.addnstr(1, 0, f"搜索: {query}", max(0, width - 1)) + stdscr.refresh() + key = stdscr.get_wch() + if key in ("\n", "\r") or key == curses.KEY_ENTER: + return query + if key == "\x1b": + return "" + if key in ("\b", "\x7f") or key == curses.KEY_BACKSPACE: + query = query[:-1] + elif isinstance(key, str) and key.isprintable(): + query += key diff --git a/skiff/skills.py b/skiff/skills.py index df506fd..24cbb1d 100644 --- a/skiff/skills.py +++ b/skiff/skills.py @@ -42,9 +42,10 @@ def resolve_skill_source(name: str, *, source: str | None = None) -> tuple[Path, registry = load_registry() if source in (None, "registry") and name in registry: path = external_skill_path(name, registry[name]) - if not path.exists(): + if not (path / "SKILL.md").is_file(): raise SystemExit( - f"外部 skill {name!r} 尚未 fetch。请先运行: skiff fetch {name}" + f"外部 skill {name!r} 尚未 fetch 或 path 中缺少 SKILL.md。" + f"请运行: skiff fetch {name}" ) return path, "external" diff --git a/skiff/yaml_io.py b/skiff/yaml_io.py index 6eb7ab4..4f75097 100644 --- a/skiff/yaml_io.py +++ b/skiff/yaml_io.py @@ -236,11 +236,12 @@ def _dump_list_dict_item(item: dict[str, Any], indent: int) -> list[str]: lines: list[str] = [] first = True for key, value in item.items(): - prefix = f"{pad}- " if first else f"{pad} " + is_first = first + prefix = f"{pad}- " if is_first else f"{pad} " first = False if isinstance(value, (dict, list)): lines.append(f"{prefix}{key}:") - nested = _dump(value, indent + 4 if first else indent + 2) + nested = _dump(value, indent + 2 if is_first else indent + 4) lines.append(nested.rstrip()) else: lines.append(f"{prefix}{key}: {_scalar(value)}") diff --git a/tests/test_select.py b/tests/test_select.py new file mode 100644 index 0000000..0be534a --- /dev/null +++ b/tests/test_select.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +import argparse +import curses +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +from skiff import cli +from skiff import yaml_io +from skiff.registry import external_repo_path, external_skill_path +from skiff.selector import SkillChoice, filter_choices, select_skills + + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +class SelectorTests(unittest.TestCase): + def test_manifest_entry_targets_round_trip(self) -> None: + data = { + "skills": [ + { + "name": "external-one", + "source": "registry", + "targets": ["codex"], + } + ] + } + + self.assertEqual(yaml_io.safe_load(yaml_io.safe_dump(data)), data) + + def test_filter_matches_name_kind_and_description(self) -> None: + choices = [ + SkillChoice("frontend-design", "external", "创建界面"), + SkillChoice("discussion-notes", "owned", "维护讨论笔记"), + ] + + self.assertEqual([c.name for c in filter_choices(choices, "front")], ["frontend-design"]) + self.assertEqual([c.name for c in filter_choices(choices, "owned")], ["discussion-notes"]) + self.assertEqual([c.name for c in filter_choices(choices, "界面")], ["frontend-design"]) + + def test_selector_preserves_preselected_items(self) -> None: + choices = [ + SkillChoice("already-there", "owned", installed=True), + SkillChoice("new-skill", "external"), + ] + screen = Mock() + screen.getmaxyx.return_value = (24, 100) + screen.get_wch.return_value = "\n" + + with patch.object(curses, "curs_set"): + selected = select_skills( + choices, + wrapper=lambda draw: draw(screen), + ) + + self.assertEqual(selected, {"already-there"}) + + def test_shared_repo_path_is_same_for_different_skill_entries(self) -> None: + first = {"repo": "https://example.test/skills.git", "ref": "main", "path": "a"} + second = {"repo": "https://example.test/skills.git", "ref": "main", "path": "b"} + + self.assertEqual(external_repo_path(first), external_repo_path(second)) + + def test_external_skill_path_rejects_checkout_escape(self) -> None: + entry = { + "repo": "https://example.test/skills.git", + "ref": "main", + "path": "../../outside", + } + + with self.assertRaisesRegex(SystemExit, "超出外部仓库"): + external_skill_path("unsafe-skill", entry) + + +class SelectCommandTests(unittest.TestCase): + def test_non_tty_exits_with_add_guidance(self) -> None: + with tempfile.TemporaryDirectory() as temp: + home = Path(temp) + (home / ".skills" / "skills").mkdir(parents=True) + env = os.environ.copy() + env["HOME"] = str(home) + env["PYTHONPATH"] = str(REPO_ROOT) + + result = subprocess.run( + [sys.executable, "-m", "skiff", "select"], + cwd=REPO_ROOT, + env=env, + text=True, + capture_output=True, + check=False, + ) + + self.assertNotEqual(result.returncode, 0) + self.assertIn("skiff add ", result.stderr) + + def test_project_selection_installs_new_and_records_all_selected(self) -> None: + with tempfile.TemporaryDirectory() as temp: + project = Path(temp) + args = argparse.Namespace( + agents=[["codex"]], + global_scope=False, + project=str(project), + yes=False, + ) + stdin = Mock() + stdout = Mock() + stdin.isatty.return_value = True + stdout.isatty.return_value = True + installed: list[str] = [] + + with ( + patch.object(cli.sys, "stdin", stdin), + patch.object(cli.sys, "stdout", stdout), + patch.object(cli, "ensure_skills_home"), + patch.object(cli, "list_owned_skills", return_value=["owned-one"]), + patch.object(cli, "skill_description", return_value="owned"), + patch.object( + cli, + "load_registry", + return_value={ + "external-one": { + "repo": "https://example.test/skills.git", + "ref": "main", + "path": "external-one", + } + }, + ), + patch.object(cli, "_list_fully_installed_names", return_value=["owned-one"]), + patch.object( + cli, + "select_skills", + return_value={"owned-one", "external-one"}, + ), + patch.object( + cli, + "_install_skill", + side_effect=lambda name, targets, project_root: installed.append(name), + ), + ): + cli.cmd_select(args) + + manifest = (project / ".skills.yaml").read_text(encoding="utf-8") + + self.assertEqual(installed, ["external-one"]) + self.assertIn("owned-one", manifest) + self.assertIn('name: "external-one"', manifest) + self.assertIn("source: registry", manifest) + self.assertIn("targets:", manifest) + self.assertIn("codex", manifest) + + def test_select_rejects_invalid_registry_name_before_rendering(self) -> None: + args = argparse.Namespace( + agents=None, + global_scope=True, + project=None, + yes=False, + ) + stdin = Mock() + stdout = Mock() + stdin.isatty.return_value = True + stdout.isatty.return_value = True + + with ( + patch.object(cli.sys, "stdin", stdin), + patch.object(cli.sys, "stdout", stdout), + patch.object(cli, "ensure_skills_home"), + patch.object(cli, "list_owned_skills", return_value=[]), + patch.object( + cli, + "load_registry", + return_value={"../../victim": {"repo": "https://example.test/repo.git"}}, + ), + patch.object(cli, "select_skills") as selector, + ): + with self.assertRaisesRegex(SystemExit, "skill 名称无效"): + cli.cmd_select(args) + + selector.assert_not_called() + + def test_install_rolls_back_earlier_target_when_later_target_fails(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + skill = root / "source" + skill.mkdir() + skill.joinpath("SKILL.md").write_text("---\\n", encoding="utf-8") + links = { + "cursor": root / "cursor" / "demo", + "claude": root / "claude" / "demo", + } + links["claude"].mkdir(parents=True) + + with ( + patch.object(cli, "_ensure_external_fetched"), + patch.object(cli, "resolve_skill_source", return_value=(skill, "owned")), + patch.object( + cli, + "agent_skill_dir", + side_effect=lambda target, project_root=None: links[target].parent, + ), + ): + with self.assertRaises(FileExistsError): + cli._install_skill("demo", ["cursor", "claude"]) + + self.assertFalse(links["cursor"].exists()) + self.assertTrue(links["claude"].is_dir()) + + +if __name__ == "__main__": + unittest.main()