laily/.pouch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add interactive skill selector

Browse Source
This commit is contained in:
laily
2026-07-27 18:07:39 +08:00
parent 86cd1fa36d
commit b9963b66de
10 changed files with 540 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -10,6 +10,7 @@ cd ~/.skills
./install.sh # 安装 CLI，并将 skiff 项目 skill 安装到所有 Agent
skiff install declarative-openspec-loop
skiff select # 交互式选择并批量安装
skiff list
skiff status
```
registry.yaml
+3
View File
@@ -4,6 +4,9 @@
# repo: <git-url>
# ref: <branch|tag> (default: main)
# path: <subpath> (default: .)
# description: <text> (optional, shown by `skiff select`)
# tags: (optional)
# - <tag>
#
# Example:
# example-skills:
skiff/README.md
+22 -2
View File
@@ -70,6 +70,7 @@ skiff bootstrap
| 命令 | 说明 |
|------|------|
| `skiff add <name> [--global] [-a AGENT...] [-y]` | 安装到 Agent 目录（软链） |
| `skiff select [--global] [-a AGENT...]` | 打开终端多选界面，批量安装 skill |
| `skiff remove <name> [--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 <name> <repo-url> [--ref main] [--path .]` | 写入 `registry.yaml` |
| `skiff fetch <name>` | 克隆/更新到 `~/.local/share/skills/externals/<name>/` |
| `skiff fetch <name>` | 克隆或更新外部仓库缓存 |
| `skiff add <name> [-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 <name>...`。使用 `-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 共享缓存 |
## 注意事项
skiff/cli.py
+148 -9
View File
@@ -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 <name>...`"
)
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"],
skiff/project.py
+6 -4
View File
@@ -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)
skiff/registry.py
+27 -2
View File
@@ -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
skiff/selector.py
+113
View File
@@ -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
skiff/skills.py
+3 -2
View File
@@ -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"
skiff/yaml_io.py
+3 -2
View File
@@ -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)}")
tests/test_select.py
+214
View File
@@ -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 <name>", 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()