1825 lines
63 KiB
Python
1825 lines
63 KiB
Python
"""skiff CLI 入口。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import ctypes
|
||
import errno
|
||
import os
|
||
import re
|
||
import secrets
|
||
import stat
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from skiff import __version__
|
||
from skiff.agents import flatten_agent_args, resolve_agent_args
|
||
from skiff.gitops import publish as git_publish
|
||
from skiff.paths import (
|
||
ALL_TARGETS,
|
||
DRAFTS_DIR,
|
||
CATALOG_CACHE_DIR,
|
||
CONFIG_FILE,
|
||
SKILLS_DIR,
|
||
SKILLS_HOME,
|
||
TEMPLATE_DIR,
|
||
agent_skill_dir,
|
||
ensure_skills_home,
|
||
)
|
||
from skiff.project import (
|
||
add_skill_to_manifest,
|
||
iter_manifest_skills,
|
||
load_manifest,
|
||
remove_skill_from_manifest,
|
||
resolve_manifest_skill,
|
||
)
|
||
from skiff.catalog import (
|
||
discover_catalog_skills,
|
||
catalog_checkout_path,
|
||
catalog_skill_path,
|
||
load_catalog,
|
||
catalog_repo,
|
||
save_catalog,
|
||
)
|
||
from skiff.selector import SkillChoice, select_skills
|
||
from skiff.skills import (
|
||
list_custom_skills,
|
||
list_builtin_skills,
|
||
builtin_skill_path,
|
||
read_skill_meta,
|
||
normalize_source,
|
||
resolve_skill_source,
|
||
split_skill_spec,
|
||
skill_description,
|
||
validate_skill_dir,
|
||
validate_skill_name,
|
||
)
|
||
from skiff.sources import (
|
||
discover_source_skills,
|
||
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
|
||
|
||
_RENAME_NOREPLACE = 1
|
||
|
||
|
||
def _encode_single_path_component(value: str) -> bytes:
|
||
encoded = os.fsencode(value)
|
||
if (
|
||
not encoded
|
||
or encoded in {b".", b".."}
|
||
or b"/" in encoded
|
||
or b"\0" in encoded
|
||
):
|
||
raise ValueError(f"必须是单一路径组件: {value!r}")
|
||
return encoded
|
||
|
||
|
||
def _print(msg: str = "") -> None:
|
||
print(msg, file=sys.stdout)
|
||
|
||
|
||
def _err(msg: str) -> None:
|
||
print(msg, file=sys.stderr)
|
||
|
||
|
||
def _project_root(explicit: str | None = None) -> Path:
|
||
if explicit:
|
||
return Path(explicit).resolve()
|
||
return find_repo_root() or Path.cwd()
|
||
|
||
|
||
def _render_template(
|
||
source: Path,
|
||
destination: Path,
|
||
values: dict[str, str],
|
||
) -> str:
|
||
content = source.read_text(encoding="utf-8")
|
||
for placeholder, value in values.items():
|
||
content = content.replace(placeholder, value)
|
||
destination.write_text(content, encoding="utf-8")
|
||
return content
|
||
|
||
|
||
def _open_or_create_directory_at(
|
||
parent_fd: int,
|
||
name: str,
|
||
) -> tuple[int, bool]:
|
||
if (
|
||
not name
|
||
or name in {".", ".."}
|
||
or Path(name).name != name
|
||
or "/" in name
|
||
or "\\" in name
|
||
or (os.altsep is not None and os.altsep in name)
|
||
):
|
||
raise SystemExit(f"初始化目录名必须是单个安全路径段: {name!r}")
|
||
created = False
|
||
try:
|
||
os.mkdir(name, dir_fd=parent_fd)
|
||
created = True
|
||
except FileExistsError:
|
||
pass
|
||
try:
|
||
directory_fd = os.open(
|
||
name,
|
||
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
|
||
dir_fd=parent_fd,
|
||
)
|
||
except OSError as exc:
|
||
raise SystemExit(
|
||
f"初始化路径必须是普通目录且不能是软链接: {name}: {exc}"
|
||
) from exc
|
||
return directory_fd, created
|
||
|
||
|
||
def _rename_directory_noreplace(
|
||
source_parent_fd: int,
|
||
source_name: str,
|
||
destination_parent_fd: int,
|
||
destination_name: str,
|
||
) -> None:
|
||
"""Atomically publish a directory without replacing an existing path."""
|
||
source = _encode_single_path_component(source_name)
|
||
destination = _encode_single_path_component(destination_name)
|
||
if source == destination:
|
||
raise ValueError("暂存目录名与目标目录名不能相同")
|
||
try:
|
||
renameat2 = ctypes.CDLL(None, use_errno=True).renameat2
|
||
except (AttributeError, OSError) as exc:
|
||
raise SystemExit(
|
||
"当前平台缺少原子 no-replace 目录发布能力，拒绝执行初始化"
|
||
) from exc
|
||
|
||
renameat2.argtypes = [
|
||
ctypes.c_int,
|
||
ctypes.c_char_p,
|
||
ctypes.c_int,
|
||
ctypes.c_char_p,
|
||
ctypes.c_uint,
|
||
]
|
||
renameat2.restype = ctypes.c_int
|
||
ctypes.set_errno(0)
|
||
result = renameat2(
|
||
source_parent_fd,
|
||
source,
|
||
destination_parent_fd,
|
||
destination,
|
||
_RENAME_NOREPLACE,
|
||
)
|
||
if result == 0:
|
||
return
|
||
|
||
error_number = ctypes.get_errno()
|
||
if error_number in {errno.EEXIST, errno.ENOTEMPTY}:
|
||
raise FileExistsError(
|
||
error_number,
|
||
os.strerror(error_number),
|
||
destination_name,
|
||
)
|
||
if error_number in {
|
||
errno.ENOSYS,
|
||
errno.EINVAL,
|
||
getattr(errno, "ENOTSUP", errno.EOPNOTSUPP),
|
||
errno.EOPNOTSUPP,
|
||
}:
|
||
raise SystemExit(
|
||
"当前文件系统不支持原子 no-replace 目录发布，拒绝执行初始化"
|
||
)
|
||
if error_number == 0:
|
||
raise RuntimeError("renameat2 失败但未设置 errno")
|
||
raise OSError(
|
||
error_number,
|
||
os.strerror(error_number),
|
||
f"{source_name} -> {destination_name}",
|
||
)
|
||
|
||
|
||
def _assert_open_directory_path(
|
||
directory_fd: int,
|
||
path: Path,
|
||
*,
|
||
phase: str,
|
||
label: str = "项目目录",
|
||
) -> None:
|
||
"""Fail if a named directory no longer resolves to the opened inode."""
|
||
opened = os.fstat(directory_fd)
|
||
try:
|
||
current = os.stat(path, follow_symlinks=False)
|
||
except OSError as exc:
|
||
raise SystemExit(f"{phase}时{label}已移动或不可访问: {path}") from exc
|
||
if (
|
||
not stat.S_ISDIR(current.st_mode)
|
||
or (opened.st_dev, opened.st_ino) != (current.st_dev, current.st_ino)
|
||
):
|
||
raise SystemExit(f"{phase}时{label}已被替换: {path}")
|
||
|
||
|
||
def _directory_entry_matches_open_fd(
|
||
parent_fd: int,
|
||
name: str,
|
||
opened_fd: int,
|
||
) -> bool:
|
||
try:
|
||
current = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
|
||
except OSError:
|
||
return False
|
||
opened = os.fstat(opened_fd)
|
||
return (
|
||
stat.S_ISDIR(current.st_mode)
|
||
and (current.st_dev, current.st_ino) == (opened.st_dev, opened.st_ino)
|
||
)
|
||
|
||
|
||
def _collect_skill_names(positional: list[str] | None, flagged: list[str] | None) -> list[str]:
|
||
names = list(positional or [])
|
||
if flagged:
|
||
names.extend(flagged)
|
||
if "*" in names:
|
||
return list_builtin_skills()
|
||
return names
|
||
|
||
|
||
def _ensure_source_fetched(name: str, source: str | None = None) -> None:
|
||
name, source = split_skill_spec(name, source)
|
||
catalog = load_catalog()
|
||
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
|
||
|
||
catalog_name = (
|
||
source.split(":", 1)[1]
|
||
if source and source.startswith("catalog:")
|
||
else source
|
||
if source in catalog
|
||
else name
|
||
if name in catalog
|
||
else None
|
||
)
|
||
if catalog_name and (
|
||
source in (None, "catalog", catalog_name)
|
||
or source == f"catalog:{catalog_name}"
|
||
):
|
||
_ensure_catalog_fetched(catalog_name, catalog[catalog_name])
|
||
return
|
||
|
||
if source not in (None, "catalog"):
|
||
return
|
||
|
||
if name not in catalog:
|
||
return
|
||
|
||
_ensure_catalog_fetched(name, catalog[name])
|
||
|
||
|
||
def _ensure_catalog_fetched(name: str, entry: dict[str, object]) -> None:
|
||
path = catalog_skill_path(name, entry)
|
||
if (path / "SKILL.md").is_file() or discover_catalog_skills(name, entry):
|
||
return
|
||
|
||
repo = catalog_repo(entry)
|
||
ref = entry.get("ref", "main")
|
||
dest = catalog_checkout_path(name, entry)
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
if dest.exists():
|
||
raise SystemExit(
|
||
f"外部仓库已存在但 skill 路径无效: {catalog_skill_path(name, entry)}"
|
||
)
|
||
_print(f"拉取外部 skill: {name}")
|
||
subprocess.run(
|
||
["git", "clone", "--depth", "1", "--branch", ref, "--", repo, str(dest)],
|
||
check=True,
|
||
)
|
||
if not discover_catalog_skills(name, entry):
|
||
raise SystemExit(
|
||
f"catalog 条目 {name!r} 的 path 中没有可安装的 SKILL.md: {path}"
|
||
)
|
||
|
||
|
||
def _catalog_skill_names(
|
||
name: str,
|
||
entry: dict[str, object] | None = None,
|
||
) -> list[str]:
|
||
entry = entry or load_catalog().get(name)
|
||
if not entry:
|
||
raise SystemExit(f"catalog 中不存在: {name}")
|
||
_ensure_catalog_fetched(name, entry)
|
||
names = list(discover_catalog_skills(name, entry))
|
||
for skill_name in names:
|
||
validate_skill_name(skill_name)
|
||
if not names:
|
||
raise SystemExit(f"catalog 条目 {name!r} 中没有可安装的 skill")
|
||
return names
|
||
|
||
|
||
def _expand_install_request(
|
||
spec: str,
|
||
explicit_source: str | None = None,
|
||
) -> list[tuple[str, str | None]]:
|
||
name, source = split_skill_spec(spec, explicit_source)
|
||
catalog = load_catalog()
|
||
sources = load_sources()
|
||
if source in sources:
|
||
return [(name, source)]
|
||
if source == "catalog" and name in catalog:
|
||
available = _catalog_skill_names(name, catalog[name])
|
||
root = catalog_skill_path(name, catalog[name])
|
||
if (root / "SKILL.md").is_file():
|
||
return [(name, "catalog")]
|
||
return [(skill_name, f"catalog:{name}") for skill_name in available]
|
||
if source in catalog:
|
||
available = _catalog_skill_names(source, catalog[source])
|
||
if name not in available:
|
||
raise SystemExit(
|
||
f"catalog collection {source!r} 中找不到 skill {name!r}"
|
||
)
|
||
return [(name, f"catalog:{source}")]
|
||
if source is None and name in catalog:
|
||
available = _catalog_skill_names(name, catalog[name])
|
||
root = catalog_skill_path(name, catalog[name])
|
||
if (root / "SKILL.md").is_file():
|
||
return [(name, f"catalog:{name}")]
|
||
return [(skill_name, f"catalog:{name}") for skill_name in available]
|
||
if source is None and name in sources:
|
||
_ensure_source_fetched(name, name)
|
||
available = discover_source_skills(name, sources[name])
|
||
if not available:
|
||
raise SystemExit(f"custom source {name!r} 中没有可安装的 skill")
|
||
return [(skill_name, name) for skill_name in available]
|
||
return [(name, source)]
|
||
|
||
|
||
def _manifest_source_details(resolved_source: str) -> tuple[str, dict[str, object]]:
|
||
return resolved_source, {}
|
||
|
||
|
||
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
|
||
]
|
||
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}): {resolved_source}/{name} -> {skill_path}")
|
||
return resolved_source
|
||
|
||
|
||
def _list_installed_names(project_root: Path | None, targets: list[str]) -> list[str]:
|
||
names: set[str] = set()
|
||
for target in targets:
|
||
skill_dir = agent_skill_dir(target, project_root=project_root)
|
||
if not skill_dir.is_dir():
|
||
continue
|
||
for entry in skill_dir.iterdir():
|
||
if entry.name.startswith("."):
|
||
continue
|
||
if entry.is_symlink() or entry.is_dir():
|
||
names.add(entry.name)
|
||
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 _is_fully_installed(
|
||
name: str,
|
||
expected: Path,
|
||
project_root: Path | None,
|
||
targets: list[str],
|
||
) -> bool:
|
||
return all(
|
||
check_link(
|
||
agent_skill_dir(target, project_root=project_root) / name,
|
||
expected,
|
||
).ok
|
||
for target in targets
|
||
)
|
||
|
||
|
||
def _global_installation_note(
|
||
name: str,
|
||
expected: Path,
|
||
targets: list[str],
|
||
) -> str:
|
||
installed: list[str] = []
|
||
conflicts: list[str] = []
|
||
for target in targets:
|
||
link = agent_skill_dir(target) / name
|
||
status = check_link(link, expected)
|
||
if status.ok:
|
||
installed.append(target)
|
||
elif link.exists() or link.is_symlink():
|
||
conflicts.append(target)
|
||
|
||
parts: list[str] = []
|
||
if installed:
|
||
parts.append(f"全局: {','.join(installed)}")
|
||
if conflicts:
|
||
parts.append(f"全局同名冲突: {','.join(conflicts)}")
|
||
return "；".join(parts)
|
||
|
||
|
||
def _remove_skill(
|
||
name: str,
|
||
targets: list[str],
|
||
*,
|
||
project_root: Path | None,
|
||
) -> int:
|
||
removed = 0
|
||
for target in targets:
|
||
link = agent_skill_dir(target, project_root=project_root) / name
|
||
try:
|
||
if remove_link(link):
|
||
scope = "全局" if project_root is None else "项目"
|
||
_print(f"已移除 ({scope}/{target}): {name}")
|
||
removed += 1
|
||
except FileExistsError as exc:
|
||
_err(f"跳过 {link}: {exc}")
|
||
return removed
|
||
|
||
|
||
def cmd_list(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
source_filter = normalize_source(args.source)
|
||
builtin = list_builtin_skills()
|
||
catalog = load_catalog()
|
||
custom = (
|
||
{}
|
||
if source_filter in ("builtin", "catalog")
|
||
or (source_filter and source_filter.startswith("catalog:"))
|
||
else list_custom_skills(source_filter)
|
||
)
|
||
|
||
if source_filter in (None, "builtin"):
|
||
_print("内置 (builtin):")
|
||
for name in builtin:
|
||
_print(f" {name}")
|
||
|
||
if source_filter in (None, "catalog"):
|
||
_print("\n目录 (catalog):")
|
||
if not catalog:
|
||
_print(" (无)")
|
||
else:
|
||
for name, entry in catalog.items():
|
||
repo = entry.get("repo", "?")
|
||
_print(f" {name} ({repo})")
|
||
elif source_filter and source_filter.startswith("catalog:"):
|
||
provider = source_filter.split(":", 1)[1]
|
||
if provider not in catalog:
|
||
raise SystemExit(f"catalog 中不存在: {provider}")
|
||
_ensure_catalog_fetched(provider, catalog[provider])
|
||
_print(f"目录 (catalog:{provider}):")
|
||
for name in discover_catalog_skills(provider, catalog[provider]):
|
||
_print(f" {name}")
|
||
|
||
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:
|
||
del args
|
||
ensure_skills_home()
|
||
project_skill = "skiff"
|
||
builtin_skill_path(project_skill)
|
||
_install_skill(project_skill, list(ALL_TARGETS), project_root=None)
|
||
_print("已全局安装 builtin skiff skill 到所有 agent")
|
||
|
||
|
||
def cmd_update(args: argparse.Namespace) -> None:
|
||
del args
|
||
ensure_skills_home()
|
||
_print(f"更新 skiff: {SKILLS_HOME}")
|
||
subprocess.run(["git", "-C", str(SKILLS_HOME), "pull"], check=True)
|
||
|
||
|
||
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
|
||
rows.append((target, link, skill_path))
|
||
return rows
|
||
|
||
|
||
def cmd_status(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
targets = resolve_agent_args(flatten_agent_args(args.agents))
|
||
builtin = list_builtin_skills()
|
||
catalog = load_catalog()
|
||
custom = list_custom_skills()
|
||
entries = [("builtin", name) for name in builtin]
|
||
unfetched_catalog: list[str] = []
|
||
for package, entry in catalog.items():
|
||
discovered = discover_catalog_skills(package, entry)
|
||
if not discovered:
|
||
unfetched_catalog.append(package)
|
||
elif (catalog_skill_path(package, entry) / "SKILL.md").is_file():
|
||
entries.append((f"catalog:{package}", package))
|
||
else:
|
||
entries.extend(
|
||
(f"catalog:{package}", skill_name)
|
||
for skill_name in discovered
|
||
)
|
||
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 package in unfetched_catalog:
|
||
_print(f"[catalog] {package}\n (未 fetch)\n")
|
||
|
||
for source, name in entries:
|
||
_print(f"[{source}] {name}")
|
||
try:
|
||
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
|
||
for target, link, expected in rows:
|
||
status = check_link(link, expected)
|
||
mark = "✓" if status.ok else "✗"
|
||
detail = "" if status.ok else f" — {status.issue}"
|
||
_print(f" {mark} {target}: {link}{detail}")
|
||
_print("")
|
||
|
||
|
||
def _print_available_skills() -> None:
|
||
ensure_skills_home()
|
||
builtin = list_builtin_skills()
|
||
if not builtin:
|
||
_print("~/.skills/skills/ 中没有自研 skill")
|
||
return
|
||
|
||
_print(f"来源: {SKILLS_DIR}\n")
|
||
for name in builtin:
|
||
desc = skill_description(name)
|
||
_print(f" {name}")
|
||
if desc:
|
||
for line in desc.splitlines():
|
||
_print(f" {line}")
|
||
_print("")
|
||
|
||
|
||
def cmd_add(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
|
||
if args.list_available:
|
||
if args.source:
|
||
cmd_list(argparse.Namespace(source=args.source))
|
||
else:
|
||
_print_available_skills()
|
||
_print("使用 skiff add <name> 安装，或 skiff add <name> -g 全局安装")
|
||
return
|
||
|
||
if args.all:
|
||
if args.source:
|
||
source = normalize_source(args.source)
|
||
if source == "builtin":
|
||
names = list_builtin_skills()
|
||
elif source == "catalog":
|
||
names = list(load_catalog())
|
||
else:
|
||
_ensure_source_fetched("", source)
|
||
names = list_custom_skills(source)[source]
|
||
else:
|
||
names = list_builtin_skills()
|
||
targets = resolve_agent_args(["*"])
|
||
else:
|
||
names = _collect_skill_names(args.skills, args.skills_flag)
|
||
if not names:
|
||
raise SystemExit("请指定 skill 名称，或使用 skiff add --list 查看可用 skill")
|
||
targets = resolve_agent_args(flatten_agent_args(args.agents))
|
||
|
||
project_root = None if args.global_scope else _project_root(args.project)
|
||
|
||
requests = [
|
||
request
|
||
for name in names
|
||
for request in _expand_install_request(name, args.source)
|
||
]
|
||
requests = list(dict.fromkeys(requests))
|
||
original_links: list[tuple[Path, str | None]] = []
|
||
for skill_name, source in requests:
|
||
validate_skill_name(skill_name)
|
||
_ensure_source_fetched(skill_name, source)
|
||
resolve_skill_source(skill_name, source=source)
|
||
for target in targets:
|
||
link = agent_skill_dir(target, project_root=project_root) / skill_name
|
||
if link.exists() and not link.is_symlink():
|
||
raise FileExistsError(f"已存在非软链路径: {link}")
|
||
original_links.append(
|
||
(link, str(link.readlink()) if link.is_symlink() else None)
|
||
)
|
||
|
||
manifest_path = project_root / ".skills.yaml" if project_root else None
|
||
manifest_before = (
|
||
manifest_path.read_bytes()
|
||
if manifest_path and manifest_path.is_file()
|
||
else None
|
||
)
|
||
try:
|
||
for skill_name, source in requests:
|
||
resolved_source = _install_skill(
|
||
skill_name,
|
||
targets,
|
||
project_root=project_root,
|
||
source=source,
|
||
)
|
||
if manifest_path is not None:
|
||
manifest_source, extra = _manifest_source_details(resolved_source)
|
||
add_skill_to_manifest(
|
||
manifest_path,
|
||
skill_name,
|
||
source=manifest_source,
|
||
extra=extra or None,
|
||
)
|
||
except Exception:
|
||
for link, previous in reversed(original_links):
|
||
if link.is_symlink():
|
||
link.unlink()
|
||
if previous is not None:
|
||
link.symlink_to(previous)
|
||
if manifest_path is not None:
|
||
if manifest_before is None:
|
||
if manifest_path.is_file():
|
||
manifest_path.unlink()
|
||
else:
|
||
manifest_path.write_bytes(manifest_before)
|
||
raise
|
||
|
||
|
||
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)
|
||
catalog = load_catalog()
|
||
builtin_names = list_builtin_skills()
|
||
for name in [*builtin_names, *catalog]:
|
||
validate_skill_name(name)
|
||
|
||
def make_choice(
|
||
*,
|
||
name: str,
|
||
installed_name: str,
|
||
expected: Path,
|
||
kind: str,
|
||
description: str,
|
||
indent: int = 0,
|
||
) -> SkillChoice:
|
||
return SkillChoice(
|
||
name=name,
|
||
kind=kind,
|
||
description=description,
|
||
installed=_is_fully_installed(
|
||
installed_name,
|
||
expected,
|
||
project_root,
|
||
targets,
|
||
),
|
||
readonly_status=(
|
||
_global_installation_note(installed_name, expected, targets)
|
||
if project_root is not None
|
||
else ""
|
||
),
|
||
indent=indent,
|
||
)
|
||
|
||
choices = [
|
||
make_choice(
|
||
name=name,
|
||
installed_name=name,
|
||
expected=SKILLS_DIR / name,
|
||
kind="builtin",
|
||
description=skill_description(name) or "",
|
||
)
|
||
for name in builtin_names
|
||
]
|
||
choice_requests: dict[str, tuple[str, str]] = {
|
||
name: (name, "builtin") for name in builtin_names
|
||
}
|
||
|
||
def add_provider_choices(
|
||
provider: str,
|
||
*,
|
||
registration: str,
|
||
discovered: dict[str, Path],
|
||
description: str,
|
||
) -> None:
|
||
source = provider if registration == "custom" else f"catalog:{provider}"
|
||
if len(discovered) == 1 and provider in discovered:
|
||
expected = discovered[provider]
|
||
if provider in choice_requests:
|
||
_err(
|
||
f"警告: {registration} source 与已有 skill 同名，"
|
||
f"已忽略: {provider}"
|
||
)
|
||
return
|
||
choices.append(
|
||
make_choice(
|
||
name=provider,
|
||
installed_name=provider,
|
||
expected=expected,
|
||
kind=f"{registration}:{provider}",
|
||
description=description,
|
||
)
|
||
)
|
||
choice_requests[provider] = (provider, source)
|
||
return
|
||
|
||
child_names: list[str] = []
|
||
first_child = len(choices)
|
||
for skill_name, expected in discovered.items():
|
||
if (
|
||
skill_name in builtin_names
|
||
and expected.resolve() == (SKILLS_DIR / skill_name).resolve()
|
||
):
|
||
continue
|
||
choice_name = f"{provider}/{skill_name}"
|
||
choices.append(
|
||
make_choice(
|
||
name=choice_name,
|
||
installed_name=skill_name,
|
||
expected=expected,
|
||
kind=f"{registration}:{provider}",
|
||
description=read_skill_meta(expected).get("description", ""),
|
||
indent=1,
|
||
)
|
||
)
|
||
choice_requests[choice_name] = (skill_name, source)
|
||
child_names.append(choice_name)
|
||
if child_names:
|
||
choices.insert(
|
||
first_child,
|
||
SkillChoice(
|
||
name=provider,
|
||
kind=f"{registration} source",
|
||
description=description,
|
||
children=tuple(child_names),
|
||
),
|
||
)
|
||
|
||
for package, entry in catalog.items():
|
||
skill_names = _catalog_skill_names(package, entry)
|
||
root = catalog_skill_path(package, entry)
|
||
discovered = (
|
||
{package: root}
|
||
if (root / "SKILL.md").is_file()
|
||
else {skill_name: root / skill_name for skill_name in skill_names}
|
||
)
|
||
add_provider_choices(
|
||
package,
|
||
registration="catalog",
|
||
discovered=discovered,
|
||
description=str(entry.get("description", "")),
|
||
)
|
||
|
||
custom_sources = load_sources()
|
||
for provider, entry in custom_sources.items():
|
||
validate_skill_name(provider)
|
||
_ensure_source_fetched("", provider)
|
||
add_provider_choices(
|
||
provider,
|
||
registration="custom",
|
||
discovered=discover_source_skills(provider, entry),
|
||
description=str(entry.get("description", "")),
|
||
)
|
||
|
||
try:
|
||
scope_label = (
|
||
"全局"
|
||
if project_root is None
|
||
else f"项目 {project_root}（全局状态只读）"
|
||
)
|
||
selected = select_skills(choices, scope_label=scope_label)
|
||
except (RuntimeError, OSError) as exc:
|
||
raise SystemExit(str(exc)) from exc
|
||
if selected is None:
|
||
_print("已取消，未修改环境")
|
||
return
|
||
|
||
selected_installed_keys = {
|
||
choice.name for choice in choices if choice.installed
|
||
}
|
||
selected_outputs: dict[str, str] = {}
|
||
for key in selected:
|
||
skill_name, _ = choice_requests[key]
|
||
previous = selected_outputs.get(skill_name)
|
||
if previous and previous != key:
|
||
raise SystemExit(
|
||
f"选择冲突: {previous} 和 {key} 都会安装为 {skill_name!r}"
|
||
)
|
||
selected_outputs[skill_name] = key
|
||
names = sorted(selected - selected_installed_keys)
|
||
|
||
failures: list[tuple[str, str]] = []
|
||
manifest_path = project_root / ".skills.yaml" if project_root else None
|
||
successful = set(selected & selected_installed_keys)
|
||
for name in names:
|
||
skill_name, source = choice_requests[name]
|
||
try:
|
||
_install_skill(
|
||
skill_name,
|
||
targets,
|
||
project_root=project_root,
|
||
source=source,
|
||
)
|
||
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):
|
||
skill_name, source = choice_requests[name]
|
||
extra = {"targets": entry_targets} if entry_targets else None
|
||
add_skill_to_manifest(
|
||
manifest_path,
|
||
skill_name,
|
||
source=source,
|
||
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)
|
||
targets = resolve_agent_args(flatten_agent_args(args.agents))
|
||
|
||
if args.all:
|
||
names = _list_installed_names(project_root, targets)
|
||
else:
|
||
names = _collect_skill_names(args.skills, args.skills_flag)
|
||
|
||
if not names:
|
||
raise SystemExit("请指定 skill 名称，或使用 skiff remove --all")
|
||
|
||
catalog = load_catalog()
|
||
sources = load_sources()
|
||
expanded: list[str] = []
|
||
for spec in names:
|
||
name, source = split_skill_spec(spec)
|
||
if source and source.startswith("catalog:"):
|
||
provider = source.split(":", 1)[1]
|
||
if provider not in catalog:
|
||
raise SystemExit(f"catalog 中不存在: {provider}")
|
||
if name not in discover_catalog_skills(provider, catalog[provider]):
|
||
raise SystemExit(
|
||
f"catalog source {provider!r} 中找不到 skill {name!r}"
|
||
)
|
||
expanded.append(name)
|
||
elif source is None and name in catalog:
|
||
discovered = discover_catalog_skills(name, catalog[name])
|
||
root = catalog_skill_path(name, catalog[name])
|
||
expanded.extend(
|
||
[name]
|
||
if (root / "SKILL.md").is_file()
|
||
else list(discovered)
|
||
)
|
||
elif source is None and name in sources:
|
||
expanded.extend(discover_source_skills(name, sources[name]))
|
||
else:
|
||
expanded.append(name)
|
||
|
||
total = 0
|
||
for name in dict.fromkeys(expanded):
|
||
total += _remove_skill(name, targets, project_root=project_root)
|
||
if project_root is not None:
|
||
remove_skill_from_manifest(project_root / ".skills.yaml", name)
|
||
|
||
if total == 0:
|
||
_print("没有移除任何 skill")
|
||
|
||
|
||
def cmd_publish(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
paths = args.paths or ["."]
|
||
git_publish(
|
||
paths=paths,
|
||
message=args.message,
|
||
push=args.push,
|
||
no_commit=args.no_commit,
|
||
)
|
||
|
||
|
||
def cmd_catalog_add(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
validate_skill_name(args.name)
|
||
catalog = load_catalog()
|
||
if args.name in catalog:
|
||
raise SystemExit(f"catalog 中已存在: {args.name}")
|
||
|
||
catalog[args.name] = {
|
||
"repo": args.repo,
|
||
"ref": args.ref,
|
||
"path": args.path,
|
||
}
|
||
save_catalog(catalog)
|
||
_print(f"已添加 catalog 条目: {args.name}")
|
||
|
||
|
||
def cmd_fetch(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
catalog = load_catalog()
|
||
if args.name not in catalog:
|
||
raise SystemExit(f"catalog 中不存在: {args.name}")
|
||
|
||
entry = catalog[args.name]
|
||
repo = catalog_repo(entry)
|
||
ref = entry.get("ref", "main")
|
||
dest = catalog_checkout_path(args.name, entry)
|
||
|
||
CATALOG_CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
if dest.exists():
|
||
_print(f"更新: {dest}")
|
||
subprocess.run(["git", "-C", str(dest), "fetch", "--all", "--tags"], check=True)
|
||
subprocess.run(["git", "-C", str(dest), "checkout", ref], check=True)
|
||
subprocess.run(["git", "-C", str(dest), "pull", "--ff-only"], check=True)
|
||
else:
|
||
_print(f"克隆: {repo} -> {dest}")
|
||
subprocess.run(
|
||
["git", "clone", "--depth", "1", "--branch", ref, "--", repo, str(dest)],
|
||
check=True,
|
||
)
|
||
|
||
|
||
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_sync(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
root = _project_root(args.project)
|
||
manifest_path = root / ".skills.yaml"
|
||
if not manifest_path.is_file():
|
||
raise SystemExit(f"未找到 {manifest_path}")
|
||
|
||
_, data = load_manifest(manifest_path)
|
||
targets = resolve_agent_args(flatten_agent_args(args.agents))
|
||
manifest_targets = data.get("targets")
|
||
if manifest_targets:
|
||
targets = [t for t in targets if t in manifest_targets]
|
||
|
||
for entry in iter_manifest_skills(data):
|
||
name = entry["name"]
|
||
source = entry.get("source")
|
||
_ensure_source_fetched(name, source)
|
||
skill_path, _ = resolve_manifest_skill(entry)
|
||
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}")
|
||
|
||
|
||
def cmd_create(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
validate_skill_name(args.name)
|
||
if not TEMPLATE_DIR.is_dir():
|
||
raise SystemExit(f"模板目录不存在: {TEMPLATE_DIR}")
|
||
|
||
dst = DRAFTS_DIR / args.name
|
||
copy_template(TEMPLATE_DIR, dst)
|
||
|
||
skill_md = dst / "SKILL.md"
|
||
content = skill_md.read_text(encoding="utf-8")
|
||
content = re.sub(
|
||
r"(^name:\s*)skill-name\s*$",
|
||
rf"\g<1>{args.name}",
|
||
content,
|
||
count=1,
|
||
flags=re.MULTILINE,
|
||
)
|
||
skill_md.write_text(content, encoding="utf-8")
|
||
readme = dst / "README.md"
|
||
if readme.is_file():
|
||
readme_content = re.sub(
|
||
r"^#\s+skill-name\s*$",
|
||
f"# {args.name}",
|
||
readme.read_text(encoding="utf-8"),
|
||
count=1,
|
||
flags=re.MULTILINE,
|
||
)
|
||
readme.write_text(readme_content, encoding="utf-8")
|
||
brief = {
|
||
"name": args.name,
|
||
"idea": args.idea or "",
|
||
"source_project": str(Path(args.from_project).resolve()) if args.from_project else "",
|
||
"status": "draft",
|
||
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
||
}
|
||
(dst / "brief.yaml").write_text(safe_dump(brief), encoding="utf-8")
|
||
_print(f"草稿已创建: {dst}")
|
||
_print(f"下一步: 请完善 skiff 草稿 {args.name}")
|
||
_print(f"完成后运行: skiff check {args.name} && skiff finalize {args.name}")
|
||
|
||
|
||
def _draft_or_builtin_path(name: str) -> tuple[Path, str]:
|
||
draft = DRAFTS_DIR / name
|
||
if draft.is_dir():
|
||
return draft, "草稿"
|
||
builtin = SKILLS_DIR / name
|
||
if builtin.is_dir():
|
||
return builtin, "正式 skill"
|
||
raise SystemExit(f"找不到草稿或正式 skill: {name}")
|
||
|
||
|
||
def cmd_check(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
validate_skill_name(args.name)
|
||
path, kind = _draft_or_builtin_path(args.name)
|
||
issues = validate_skill_dir(path, args.name)
|
||
if kind == "草稿":
|
||
issues = [issue for issue in issues if "草稿文件: brief.yaml" not in issue]
|
||
if issues:
|
||
for issue in issues:
|
||
_err(f"✗ {issue}")
|
||
raise SystemExit(1)
|
||
_print(f"✓ 校验通过 ({kind}): {path}")
|
||
|
||
|
||
def cmd_finalize(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
validate_skill_name(args.name)
|
||
draft = DRAFTS_DIR / args.name
|
||
if not draft.is_dir():
|
||
raise SystemExit(f"草稿不存在: {args.name}")
|
||
final = SKILLS_DIR / args.name
|
||
if final.exists():
|
||
raise SystemExit(f"正式 skill 已存在: {final}")
|
||
|
||
issues = validate_skill_dir(draft, args.name)
|
||
issues = [issue for issue in issues if "草稿文件: brief.yaml" not in issue]
|
||
if issues:
|
||
for issue in issues:
|
||
_err(f"✗ {issue}")
|
||
raise SystemExit(1)
|
||
|
||
final.parent.mkdir(parents=True, exist_ok=True)
|
||
draft.replace(final)
|
||
brief = final / "brief.yaml"
|
||
if brief.exists():
|
||
brief.unlink()
|
||
_print(f"已完成 skill: {final}")
|
||
_print(f"下一步: skiff publish skills/{args.name} -m \"add {args.name}\" --push")
|
||
|
||
|
||
def cmd_doctor(args: argparse.Namespace) -> None:
|
||
ensure_skills_home()
|
||
targets = resolve_agent_args(flatten_agent_args(args.agents))
|
||
issues = 0
|
||
|
||
_print(f"skills 仓库: {SKILLS_HOME.resolve()}")
|
||
if SKILLS_HOME.is_symlink():
|
||
if not SKILLS_HOME.resolve().is_dir():
|
||
_err(f"✗ ~/.skills 指向无效路径: {SKILLS_HOME.resolve()}")
|
||
issues += 1
|
||
else:
|
||
_print("✓ ~/.skills 软链正常")
|
||
elif SKILLS_HOME.is_dir() and (SKILLS_HOME / "skills").is_dir():
|
||
_print("✓ ~/.skills 为本地仓库目录")
|
||
else:
|
||
_err("✗ ~/.skills 未正确配置")
|
||
issues += 1
|
||
|
||
for name in list_builtin_skills():
|
||
for target, link, expected in _installed_links(name, targets, source="builtin"):
|
||
status = check_link(link, expected)
|
||
if status.ok:
|
||
continue
|
||
_err(f"✗ [{name}/{target}] {status.issue}: {link}")
|
||
issues += 1
|
||
if args.fix:
|
||
try:
|
||
create_link(link, expected)
|
||
_print(f" 已修复: {link}")
|
||
except Exception as exc: # noqa: BLE001
|
||
_err(f" 修复失败: {exc}")
|
||
|
||
catalog = load_catalog()
|
||
for name in catalog:
|
||
ext = catalog_skill_path(name, catalog[name])
|
||
if not ext.exists():
|
||
_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:
|
||
_print(f"\n发现 {issues} 个问题")
|
||
if not args.fix:
|
||
_print("提示: 使用 skiff doctor --fix 尝试自动修复软链")
|
||
sys.exit(1)
|
||
|
||
|
||
def cmd_init(args: argparse.Namespace) -> None:
|
||
"""使用 builtin skill 自带的模板初始化目标项目状态。"""
|
||
ensure_skills_home()
|
||
validate_skill_name(args.name)
|
||
skills_root = SKILLS_DIR.resolve()
|
||
skill_source = (SKILLS_DIR / args.name).resolve()
|
||
try:
|
||
skill_source.relative_to(skills_root)
|
||
except ValueError as exc:
|
||
raise SystemExit(f"builtin skill 路径逃逸仓库边界: {args.name}") from exc
|
||
if not (skill_source / "SKILL.md").is_file():
|
||
raise SystemExit(f"builtin skill 不存在: {args.name}")
|
||
|
||
project = _project_root(args.project)
|
||
try:
|
||
initial_project_stat = os.stat(project, follow_symlinks=False)
|
||
except OSError as exc:
|
||
raise SystemExit(f"项目目录不存在: {project}") from exc
|
||
if not stat.S_ISDIR(initial_project_stat.st_mode):
|
||
raise SystemExit(f"项目目录不存在: {project}")
|
||
initial_project_identity = (
|
||
initial_project_stat.st_dev,
|
||
initial_project_stat.st_ino,
|
||
stat.S_IFMT(initial_project_stat.st_mode),
|
||
)
|
||
destination = project / "docs" / args.name
|
||
project_file = destination / "project.md"
|
||
tasks_file = destination / "tasks.yaml"
|
||
knowledge_file = destination / "knowledge.yaml"
|
||
delivery_file = destination / "delivery.yaml"
|
||
managed_targets = [project_file, tasks_file]
|
||
if args.name == "ack":
|
||
managed_targets.extend((knowledge_file, delivery_file))
|
||
existing = [path for path in managed_targets if path.exists() or path.is_symlink()]
|
||
if existing:
|
||
paths = ", ".join(str(path.relative_to(project)) for path in existing)
|
||
raise SystemExit(f"拒绝覆盖已有路径: {paths}")
|
||
|
||
project_template = skill_source / "templates" / "project.template.md"
|
||
tasks_template = skill_source / "templates" / "tasks.template.yaml"
|
||
template_targets = [
|
||
(project_template, project_file),
|
||
(tasks_template, tasks_file),
|
||
]
|
||
if args.name == "ack":
|
||
template_targets.extend(
|
||
(
|
||
(skill_source / "templates" / "knowledge.template.yaml", knowledge_file),
|
||
(skill_source / "templates" / "delivery.template.yaml", delivery_file),
|
||
)
|
||
)
|
||
missing = [path for path, _ in template_targets if not path.is_file()]
|
||
if missing:
|
||
paths = ", ".join(str(path.relative_to(SKILLS_HOME)) for path in missing)
|
||
raise SystemExit(f"skill 缺少初始化模板: {paths}")
|
||
validator = skill_source / "scripts" / "validate_tasks.py"
|
||
knowledge_validator = skill_source / "scripts" / "validate_knowledge.py"
|
||
delivery_validator = skill_source / "scripts" / "validate_delivery.py"
|
||
if args.name == "ack":
|
||
missing_validators = [
|
||
path
|
||
for path in (validator, knowledge_validator, delivery_validator)
|
||
if not path.is_file()
|
||
]
|
||
if missing_validators:
|
||
paths = ", ".join(path.name for path in missing_validators)
|
||
raise SystemExit(f"ACK skill 缺少初始化校验器: {paths}")
|
||
|
||
version_file = skill_source / "VERSION"
|
||
ack_version = version_file.read_text(encoding="utf-8").strip() if version_file.is_file() else "unknown"
|
||
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
||
values = {
|
||
"<project_name>": project.name,
|
||
"<repo_path>": str(project),
|
||
"<dev_worktree>": str(project),
|
||
"<overlay_file_path>": f"docs/{args.name}/project.md",
|
||
"<ack_version>": ack_version,
|
||
"<接入时的 ack skill 版本>": ack_version,
|
||
"<YYYY-MM-DDTHH:mm:ss+TZ>": now,
|
||
}
|
||
|
||
with tempfile.TemporaryDirectory(prefix=f"skiff-{args.name}-init-") as temp_dir:
|
||
staging = Path(temp_dir)
|
||
staged_files: dict[Path, Path] = {}
|
||
rendered_files: dict[Path, str] = {}
|
||
for template, target in template_targets:
|
||
staged = staging / target.relative_to(project)
|
||
staged.parent.mkdir(parents=True, exist_ok=True)
|
||
rendered_files[target] = _render_template(template, staged, values)
|
||
staged_files[target] = staged
|
||
|
||
if validator.is_file():
|
||
completed = subprocess.run(
|
||
[sys.executable, str(validator), str(staged_files[tasks_file])],
|
||
check=False,
|
||
)
|
||
if completed.returncode != 0:
|
||
raise SystemExit(
|
||
f"初始化任务板校验失败（exit {completed.returncode}）"
|
||
)
|
||
if args.name == "ack" and knowledge_validator.is_file():
|
||
completed = subprocess.run(
|
||
[
|
||
sys.executable,
|
||
str(knowledge_validator),
|
||
str(staged_files[knowledge_file]),
|
||
"--tasks",
|
||
str(staged_files[tasks_file]),
|
||
"--project-root",
|
||
str(staging),
|
||
],
|
||
check=False,
|
||
)
|
||
if completed.returncode != 0:
|
||
raise SystemExit(
|
||
f"初始化知识库校验失败（exit {completed.returncode}）"
|
||
)
|
||
if args.name == "ack" and delivery_validator.is_file():
|
||
completed = subprocess.run(
|
||
[
|
||
sys.executable,
|
||
str(delivery_validator),
|
||
str(staged_files[delivery_file]),
|
||
"--tasks",
|
||
str(staged_files[tasks_file]),
|
||
"--project-root",
|
||
str(staging),
|
||
],
|
||
check=False,
|
||
)
|
||
if completed.returncode != 0:
|
||
raise SystemExit(
|
||
f"初始化交付契约校验失败（exit {completed.returncode}）"
|
||
)
|
||
|
||
for target, staged in staged_files.items():
|
||
if staged.read_text(encoding="utf-8") != rendered_files[target]:
|
||
raise SystemExit(
|
||
f"初始化临时文件在校验期间发生变化: {target.name}"
|
||
)
|
||
|
||
# Validation may take time, so guard against a concurrent initializer before writing.
|
||
existing = [path for path in managed_targets if path.exists() or path.is_symlink()]
|
||
if existing:
|
||
paths = ", ".join(str(path.relative_to(project)) for path in existing)
|
||
raise SystemExit(f"拒绝覆盖已有路径: {paths}")
|
||
|
||
project_fd: int | None = None
|
||
docs_fd: int | None = None
|
||
transaction_fd: int | None = None
|
||
staging_fd: int | None = None
|
||
docs_created = False
|
||
transaction_name: str | None = None
|
||
staged_names: list[str] = []
|
||
published = False
|
||
committed = False
|
||
try:
|
||
try:
|
||
project_fd = os.open(
|
||
project,
|
||
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
|
||
)
|
||
except OSError as exc:
|
||
raise SystemExit(
|
||
f"校验期间项目目录已移动或不可访问: {project}"
|
||
) from exc
|
||
opened_project_stat = os.fstat(project_fd)
|
||
opened_project_identity = (
|
||
opened_project_stat.st_dev,
|
||
opened_project_stat.st_ino,
|
||
stat.S_IFMT(opened_project_stat.st_mode),
|
||
)
|
||
if opened_project_identity != initial_project_identity:
|
||
raise SystemExit(f"校验期间项目目录已被替换: {project}")
|
||
_assert_open_directory_path(
|
||
project_fd,
|
||
project,
|
||
phase="初始化",
|
||
)
|
||
docs_fd, docs_created = _open_or_create_directory_at(project_fd, "docs")
|
||
if docs_created:
|
||
os.fsync(project_fd)
|
||
_assert_open_directory_path(
|
||
project_fd,
|
||
project,
|
||
phase="初始化",
|
||
)
|
||
_assert_open_directory_path(
|
||
docs_fd,
|
||
project / "docs",
|
||
phase="初始化",
|
||
label="docs 目录",
|
||
)
|
||
try:
|
||
destination_stat = os.stat(
|
||
args.name,
|
||
dir_fd=docs_fd,
|
||
follow_symlinks=False,
|
||
)
|
||
except FileNotFoundError:
|
||
pass
|
||
else:
|
||
if stat.S_ISLNK(destination_stat.st_mode):
|
||
raise SystemExit(
|
||
"初始化路径必须是普通目录且不能是软链接: "
|
||
f"docs/{args.name}"
|
||
)
|
||
raise SystemExit(f"拒绝覆盖已有路径: docs/{args.name}")
|
||
|
||
for _ in range(32):
|
||
candidate = f".{args.name}-init-{secrets.token_hex(8)}"
|
||
try:
|
||
os.mkdir(candidate, mode=0o700, dir_fd=docs_fd)
|
||
except FileExistsError:
|
||
continue
|
||
transaction_name = candidate
|
||
break
|
||
if transaction_name is None:
|
||
raise SystemExit("无法创建唯一的初始化暂存目录")
|
||
|
||
transaction_fd = os.open(
|
||
transaction_name,
|
||
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
|
||
dir_fd=docs_fd,
|
||
)
|
||
os.mkdir("payload", mode=0o755, dir_fd=transaction_fd)
|
||
staging_fd = os.open(
|
||
"payload",
|
||
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
|
||
dir_fd=transaction_fd,
|
||
)
|
||
for target in staged_files:
|
||
file_fd = os.open(
|
||
target.name,
|
||
os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
|
||
0o644,
|
||
dir_fd=staging_fd,
|
||
)
|
||
staged_names.append(target.name)
|
||
with os.fdopen(file_fd, "w", encoding="utf-8") as destination_file:
|
||
destination_file.write(rendered_files[target])
|
||
destination_file.flush()
|
||
os.fsync(destination_file.fileno())
|
||
|
||
os.fsync(staging_fd)
|
||
os.fsync(transaction_fd)
|
||
_assert_open_directory_path(
|
||
project_fd,
|
||
project,
|
||
phase="发布",
|
||
)
|
||
_assert_open_directory_path(
|
||
docs_fd,
|
||
project / "docs",
|
||
phase="发布",
|
||
label="docs 目录",
|
||
)
|
||
try:
|
||
_rename_directory_noreplace(
|
||
transaction_fd,
|
||
"payload",
|
||
docs_fd,
|
||
args.name,
|
||
)
|
||
except FileExistsError as exc:
|
||
raise SystemExit(
|
||
f"拒绝覆盖已有路径: docs/{args.name}"
|
||
) from exc
|
||
published = True
|
||
_assert_open_directory_path(
|
||
staging_fd,
|
||
destination,
|
||
phase="发布",
|
||
label="ACK 目录",
|
||
)
|
||
if (
|
||
transaction_name is not None
|
||
and _directory_entry_matches_open_fd(
|
||
docs_fd,
|
||
transaction_name,
|
||
transaction_fd,
|
||
)
|
||
):
|
||
try:
|
||
os.rmdir(transaction_name, dir_fd=docs_fd)
|
||
except OSError:
|
||
pass
|
||
else:
|
||
transaction_name = None
|
||
try:
|
||
os.fsync(docs_fd)
|
||
except OSError as exc:
|
||
raise SystemExit(
|
||
"初始化目录已完整发布，但无法确认目录项持久化；"
|
||
f"请检查 docs/{args.name} 后再重试"
|
||
) from exc
|
||
_assert_open_directory_path(
|
||
project_fd,
|
||
project,
|
||
phase="完成初始化",
|
||
)
|
||
_assert_open_directory_path(
|
||
docs_fd,
|
||
project / "docs",
|
||
phase="完成初始化",
|
||
label="docs 目录",
|
||
)
|
||
_assert_open_directory_path(
|
||
staging_fd,
|
||
destination,
|
||
phase="完成初始化",
|
||
label="ACK 目录",
|
||
)
|
||
committed = True
|
||
except BaseException:
|
||
if not published:
|
||
for name in reversed(staged_names):
|
||
try:
|
||
if staging_fd is not None:
|
||
os.unlink(name, dir_fd=staging_fd)
|
||
except FileNotFoundError:
|
||
pass
|
||
if transaction_fd is not None:
|
||
try:
|
||
os.rmdir("payload", dir_fd=transaction_fd)
|
||
except OSError:
|
||
pass
|
||
if (
|
||
transaction_name is not None
|
||
and transaction_fd is not None
|
||
and docs_fd is not None
|
||
and _directory_entry_matches_open_fd(
|
||
docs_fd,
|
||
transaction_name,
|
||
transaction_fd,
|
||
)
|
||
):
|
||
try:
|
||
os.rmdir(transaction_name, dir_fd=docs_fd)
|
||
except OSError:
|
||
pass
|
||
raise
|
||
finally:
|
||
for directory_fd in (
|
||
staging_fd,
|
||
transaction_fd,
|
||
docs_fd,
|
||
project_fd,
|
||
):
|
||
if directory_fd is not None:
|
||
os.close(directory_fd)
|
||
|
||
if not committed:
|
||
raise SystemExit("初始化事务未提交")
|
||
|
||
_print(f"✓ skill 项目状态初始化完成: {args.name}")
|
||
_print(f" 项目: {project}")
|
||
_print(f" 覆盖层: {project_file}")
|
||
_print(f" 任务板: {tasks_file}")
|
||
if args.name == "ack":
|
||
_print(f" 知识库: {knowledge_file}")
|
||
_print(f" 交付契约: {delivery_file}（默认关闭）")
|
||
_print("下一步: 填写 project.md 中的项目命令、路径权限和 Base URL")
|
||
|
||
|
||
def _add_common_flags(parser: argparse.ArgumentParser) -> None:
|
||
parser.add_argument(
|
||
"-a",
|
||
"--agent",
|
||
dest="agents",
|
||
nargs="+",
|
||
action="append",
|
||
metavar="AGENT",
|
||
help="目标 agent（cursor、claude、claude-code、codex、opencode、*）",
|
||
)
|
||
parser.add_argument(
|
||
"-g",
|
||
"--global",
|
||
dest="global_scope",
|
||
action="store_true",
|
||
help="全局安装/卸载（默认当前项目）",
|
||
)
|
||
parser.add_argument("-y", "--yes", action="store_true", help="跳过确认（兼容 Vercel skills）")
|
||
parser.add_argument("--project", help="项目根目录（默认自动检测或当前目录）")
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(
|
||
prog="skiff",
|
||
description="自研 Agent Skills 安装与管理 CLI（接口对齐 Vercel skills）",
|
||
)
|
||
parser.add_argument("--version", action="version", version=f"skiff {__version__}")
|
||
|
||
sub = parser.add_subparsers(dest="command", required=True)
|
||
|
||
p_bootstrap = sub.add_parser(
|
||
"bootstrap",
|
||
help="将本项目 skiff skill 全局安装到所有 agent",
|
||
)
|
||
p_bootstrap.set_defaults(func=cmd_bootstrap)
|
||
|
||
p_update = sub.add_parser("update", help="通过 git pull 更新 skiff 自身")
|
||
p_update.set_defaults(func=cmd_update)
|
||
|
||
p_list = sub.add_parser("list", help="列出所有 source 中的 skill")
|
||
p_list.add_argument("--source", help="只列出指定来源（builtin、catalog 或 custom source）")
|
||
p_list.set_defaults(func=cmd_list)
|
||
|
||
p_status = sub.add_parser("status", help="安装状态总览")
|
||
p_status.add_argument("-a", "--agent", dest="agents", nargs="+", action="append", metavar="AGENT")
|
||
p_status.set_defaults(func=cmd_status)
|
||
|
||
p_add = sub.add_parser(
|
||
"add",
|
||
help="安装 builtin、catalog 或 custom source 中的 skill",
|
||
description="安装 builtin、catalog 或 custom source 中的 skill",
|
||
)
|
||
p_add.add_argument("skills", nargs="*", metavar="skill", help="skill 名称（可多个）")
|
||
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="列出可用 builtin 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)
|
||
|
||
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"],
|
||
help="从 agent 移除 skill",
|
||
)
|
||
p_remove.add_argument("skills", nargs="*", metavar="skill", help="skill 名称（可多个）")
|
||
p_remove.add_argument("-s", "--skill", dest="skills_flag", action="append", metavar="SKILL")
|
||
p_remove.add_argument("--all", action="store_true", help="移除当前范围内全部 skill")
|
||
_add_common_flags(p_remove)
|
||
p_remove.set_defaults(func=cmd_remove)
|
||
|
||
p_publish = sub.add_parser(
|
||
"publish",
|
||
help="在 ~/.skills 内 git add / commit / push",
|
||
)
|
||
p_publish.add_argument("paths", nargs="*", help="要提交的路径（默认 .）")
|
||
p_publish.add_argument("-m", "--message", help="commit 说明")
|
||
p_publish.add_argument("--push", action="store_true", help="commit 后 push")
|
||
p_publish.add_argument("--no-commit", action="store_true", help="只 git add，不 commit")
|
||
p_publish.set_defaults(func=cmd_publish)
|
||
|
||
p_catalog = sub.add_parser("catalog", help="管理 catalog.yaml 中的预置来源")
|
||
catalog_sub = p_catalog.add_subparsers(dest="catalog_command", required=True)
|
||
p_catalog_add = catalog_sub.add_parser("add", help="添加预置 Git source")
|
||
p_catalog_add.add_argument("name", help="catalog source 名称")
|
||
p_catalog_add.add_argument("repo", help="Git 仓库 URL")
|
||
p_catalog_add.add_argument("--ref", default="main", help="分支或 tag（默认 main）")
|
||
p_catalog_add.add_argument("--path", default=".", help="仓库内子路径（默认 .）")
|
||
p_catalog_add.set_defaults(func=cmd_catalog_add)
|
||
|
||
p_fetch = sub.add_parser("fetch", help="拉取/更新 catalog source")
|
||
p_fetch.add_argument("name", help="catalog 名称")
|
||
p_fetch.set_defaults(func=cmd_fetch)
|
||
|
||
p_source = sub.add_parser("source", help="管理自定义 Skill source")
|
||
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_sync = sub.add_parser("sync", help="按 .skills.yaml 重建项目软链")
|
||
p_sync.add_argument("-a", "--agent", dest="agents", nargs="+", action="append")
|
||
p_sync.add_argument("--project")
|
||
p_sync.set_defaults(func=cmd_sync)
|
||
|
||
p_create = sub.add_parser("create", help="从 _template 创建自研 skill 草稿")
|
||
p_create.add_argument("name", help="skill 名称")
|
||
p_create.add_argument("--idea", help="创建 skill 的原始想法")
|
||
p_create.add_argument("--from-project", help="想法来源项目（仅记录上下文）")
|
||
p_create.set_defaults(func=cmd_create)
|
||
|
||
p_check = sub.add_parser("check", help="校验草稿或正式 skill")
|
||
p_check.add_argument("name", help="skill 名称")
|
||
p_check.set_defaults(func=cmd_check)
|
||
|
||
p_finalize = sub.add_parser("finalize", help="校验草稿并转为正式 skill")
|
||
p_finalize.add_argument("name", help="skill 名称")
|
||
p_finalize.set_defaults(func=cmd_finalize)
|
||
|
||
p_init = sub.add_parser("init", help="使用 skill 模板初始化项目状态")
|
||
p_init.add_argument("name", help="skill 名称")
|
||
p_init.add_argument("--project", help="项目根目录（默认自动检测或当前目录）")
|
||
p_init.set_defaults(func=cmd_init)
|
||
|
||
p_doctor = sub.add_parser("doctor", help="软链健康检查")
|
||
p_doctor.add_argument("-a", "--agent", dest="agents", nargs="+", action="append")
|
||
p_doctor.add_argument("--fix", action="store_true", help="自动修复可修复的软链")
|
||
p_doctor.set_defaults(func=cmd_doctor)
|
||
|
||
return parser
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> None:
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
args.func(args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|