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

feat: support registry collections and docker publishing

Browse Source
This commit is contained in:
laily
2026-07-27 23:25:46 +08:00
parent b78126830b
commit 983183d5b3
13 changed files with 891 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
registry.yaml
+17 -1
View File
@@ -8,8 +8,24 @@
# tags: (optional)
# - <tag>
#
# Example:
# Single-skill example (`path` contains SKILL.md):
# example-skills:
# repo: https://github.com/example/skills
# ref: main
# path: .
#
# Collection example (`path` contains skill directories):
# example-collection:
# repo: https://github.com/example/skill-collection
# ref: main
# path: skills
waza:
repo: https://github.com/tw93/Waza.git
ref: main
path: skills
description: Waza engineering workflow skills for planning, UI, review, debugging, writing, research, reading, and agent health.
tags:
- engineering
- workflow
- skill-collection
skiff/README.md
+7 -3
View File
@@ -92,11 +92,15 @@ skiff bootstrap
|------|------|
| `skiff registry add <name> <repo-url> [--ref main] [--path .]` | 写入 `registry.yaml` |
| `skiff fetch <name>` | 克隆或更新外部仓库缓存 |
| `skiff add <name> [-g] [-a AGENT...]` | 安装 registry 中的外部 skill（缺失时自动 fetch） |
| `skiff add <name> [-g] [-a AGENT...]` | 安装 registry 中的单个 skill 或完整 collection（缺失时自动 fetch） |
| `skiff add <collection>/<skill> [...]` | 只安装 collection 中指定的 skill |
`registry.yaml` 条目可额外提供 `description` 和 `tags`。`description`
会显示在 `skiff select` 的候选列表中。同一 `repo` 与 `ref` 下的多个 skill
共享一份 Git checkout，再通过各自的 `path` 定位目录。
会显示在 `skiff select` 的候选列表中。`path` 可以直接指向含
`SKILL.md` 的单个 skill，也可以指向由多个 skill 目录组成的 collection。
collection 会自动发现下一层所有含 `SKILL.md` 的目录；`skiff add <name>`
安装全部，`skiff select` 则展开为 `<name>/<skill>` 供分别勾选。同一
`repo` 与 `ref` 共享一份 Git checkout。
### 交互式批量安装
skiff/cli.py
+257 -35
View File
@@ -32,6 +32,7 @@ from skiff.project import (
save_manifest,
)
from skiff.registry import (
discover_external_skills,
external_checkout_path,
external_skill_path,
load_registry,
@@ -42,6 +43,7 @@ from skiff.skills import (
list_custom_skills,
list_owned_skills,
owned_skill_path,
read_skill_meta,
resolve_skill_source,
split_skill_spec,
skill_description,
@@ -89,6 +91,7 @@ def _collect_skill_names(positional: list[str] | None, flagged: list[str] | None
def _ensure_source_fetched(name: str, source: str | None = None) -> None:
name, source = split_skill_spec(name, source)
registry = load_registry()
sources = load_sources()
if source in sources:
root = source_skills_root(source, sources[source])
@@ -96,19 +99,38 @@ def _ensure_source_fetched(name: str, source: str | None = None) -> None:
_print(f"拉取 source: {source}")
fetch_source(source, sources[source])
return
registry_name = (
source.split(":", 1)[1]
if source and source.startswith("registry:")
else source
if source in registry
else name
if name in registry
else None
)
if registry_name and (
source in (None, "registry", registry_name)
or source == f"registry:{registry_name}"
):
_ensure_registry_fetched(registry_name, registry[registry_name])
return
if source not in (None, "registry"):
return
registry = load_registry()
if name not in registry:
return
entry = registry[name]
_ensure_registry_fetched(name, registry[name])
def _ensure_registry_fetched(name: str, entry: dict[str, object]) -> None:
path = external_skill_path(name, entry)
if (path / "SKILL.md").is_file():
if (path / "SKILL.md").is_file() or discover_external_skills(name, entry):
return
repo = entry["repo"]
repo = str(entry["repo"])
ref = entry.get("ref", "main")
dest = external_checkout_path(name, entry)
dest.parent.mkdir(parents=True, exist_ok=True)
@@ -121,6 +143,62 @@ def _ensure_source_fetched(name: str, source: str | None = None) -> None:
["git", "clone", "--branch", ref, "--", repo, str(dest)],
check=True,
)
if not discover_external_skills(name, entry):
raise SystemExit(
f"registry 条目 {name!r} 的 path 中没有可安装的 SKILL.md: {path}"
)
def _registry_skill_names(
name: str,
entry: dict[str, object] | None = None,
) -> list[str]:
entry = entry or load_registry().get(name)
if not entry:
raise SystemExit(f"registry 中不存在: {name}")
_ensure_registry_fetched(name, entry)
names = list(discover_external_skills(name, entry))
for skill_name in names:
validate_skill_name(skill_name)
if not names:
raise SystemExit(f"registry 条目 {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)
registry = load_registry()
if source in load_sources():
return [(name, source)]
if source == "registry" and name in registry:
available = _registry_skill_names(name, registry[name])
root = external_skill_path(name, registry[name])
if (root / "SKILL.md").is_file():
return [(name, "registry")]
return [(skill_name, f"registry:{name}") for skill_name in available]
if source in registry:
available = _registry_skill_names(source, registry[source])
if name not in available:
raise SystemExit(
f"registry collection {source!r} 中找不到 skill {name!r}"
)
return [(name, f"registry:{source}")]
if source is None and name in registry:
available = _registry_skill_names(name, registry[name])
root = external_skill_path(name, registry[name])
if (root / "SKILL.md").is_file():
return [(name, "registry")]
return [(skill_name, f"registry:{name}") for skill_name in available]
return [(name, source)]
def _manifest_source_details(resolved_source: str) -> tuple[str, dict[str, object]]:
if resolved_source.startswith("registry:"):
return "registry", {"registry": resolved_source.split(":", 1)[1]}
return resolved_source, {}
def _install_skill(
@@ -184,6 +262,21 @@ def _list_fully_installed_names(
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 _remove_skill(
name: str,
targets: list[str],
@@ -266,12 +359,26 @@ def cmd_status(args: argparse.Namespace) -> None:
registry = load_registry()
custom = list_custom_skills()
entries = [("owned", name) for name in owned]
entries.extend(("registry", name) for name in registry)
unfetched_registry: list[str] = []
for package, entry in registry.items():
discovered = discover_external_skills(package, entry)
if not discovered:
unfetched_registry.append(package)
elif (external_skill_path(package, entry) / "SKILL.md").is_file():
entries.append(("registry", package))
else:
entries.extend(
(f"registry:{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_registry:
_print(f"[registry] {package}\n (未 fetch)\n")
for source, name in entries:
_print(f"[{source}] {name}")
try:
@@ -339,21 +446,60 @@ def cmd_add(args: argparse.Namespace) -> None:
project_root = None if args.global_scope else _project_root(args.project)
for name in names:
skill_name, source = split_skill_spec(name, args.source)
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 project_root is not None:
if manifest_path is not None:
manifest_source, extra = _manifest_source_details(resolved_source)
add_skill_to_manifest(
project_root / ".skills.yaml",
manifest_path,
skill_name,
source=resolved_source,
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:
@@ -365,38 +511,67 @@ def cmd_select(args: argparse.Namespace) -> None:
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,
installed=_is_fully_installed(
name,
SKILLS_DIR / name,
project_root,
targets,
),
)
for name in owned_names
]
choices.extend(
choice_requests: dict[str, tuple[str, str]] = {
name: (name, "owned") for name in owned_names
}
for package, entry in registry.items():
skill_names = _registry_skill_names(package, entry)
root = external_skill_path(package, entry)
if (root / "SKILL.md").is_file():
if package in choice_requests:
_err(f"警告: registry 条目与 owned skill 同名，已忽略 external: {package}")
continue
choices.append(
SkillChoice(
name=name,
name=package,
kind="external",
description=str(entry.get("description", "")),
installed=name in installed,
installed=_is_fully_installed(
package,
root,
project_root,
targets,
),
)
for name, entry in registry.items()
if name not in collisions
)
choice_sources = {
**{name: "owned" for name in owned_names},
**{name: "registry" for name in registry if name not in collisions},
}
choice_requests[package] = (package, "registry")
continue
for skill_name in skill_names:
choice_name = f"{package}/{skill_name}"
description = read_skill_meta(root / skill_name).get("description", "")
choices.append(
SkillChoice(
name=choice_name,
kind=f"external:{package}",
description=description,
installed=_is_fully_installed(
skill_name,
root / skill_name,
project_root,
targets,
),
)
)
choice_requests[choice_name] = (skill_name, f"registry:{package}")
try:
selected = select_skills(choices)
@@ -406,18 +581,31 @@ def cmd_select(args: argparse.Namespace) -> None:
_print("已取消，未修改环境")
return
names = sorted(selected - installed)
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 & installed)
successful = set(selected & selected_installed_keys)
for name in names:
skill_name, source = choice_requests[name]
try:
_install_skill(
name,
skill_name,
targets,
project_root=project_root,
source=choice_sources[name],
source=source,
)
successful.add(name)
except (OSError, subprocess.CalledProcessError, SystemExit) as exc:
@@ -427,21 +615,29 @@ def cmd_select(args: argparse.Namespace) -> None:
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:
skill_name, source = choice_requests[name]
if source == "owned":
add_skill_to_manifest(
manifest_path,
name,
skill_name,
source="owned",
extra={"targets": entry_targets} if entry_targets else None,
)
else:
extra = {"ref": entry.get("ref", "main")}
package = (
skill_name
if source == "registry"
else source.split(":", 1)[1]
)
entry = registry[package]
extra: dict[str, object] = {"ref": entry.get("ref", "main")}
if source != "registry":
extra["registry"] = package
if entry_targets:
extra["targets"] = entry_targets
add_skill_to_manifest(
manifest_path,
name,
skill_name,
source="registry",
extra=extra,
)
@@ -468,9 +664,32 @@ def cmd_remove(args: argparse.Namespace) -> None:
if not names:
raise SystemExit("请指定 skill 名称，或使用 skiff remove --all")
registry = load_registry()
expanded: list[str] = []
for spec in names:
name, source = split_skill_spec(spec)
if source in registry and source not in load_sources():
if name not in discover_external_skills(source, registry[source]):
raise SystemExit(
f"registry collection {source!r} 中找不到 skill {name!r}"
)
expanded.append(name)
elif source is None and name in registry:
discovered = discover_external_skills(name, registry[name])
root = external_skill_path(name, registry[name])
expanded.extend(
[name]
if (root / "SKILL.md").is_file()
else list(discovered)
)
else:
expanded.append(name)
total = 0
for name in names:
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")
@@ -666,7 +885,10 @@ def cmd_sync(args: argparse.Namespace) -> None:
for entry in iter_manifest_skills(data):
name = entry["name"]
_ensure_source_fetched(name, entry.get("source"))
source = entry.get("source")
if source == "registry" and entry.get("registry"):
source = f"registry:{entry['registry']}"
_ensure_source_fetched(name, source)
skill_path, _ = resolve_manifest_skill(entry)
skill_targets = targets
if entry.get("targets"):
skiff/project.py
+2
View File
@@ -101,4 +101,6 @@ def iter_manifest_skills(data: dict[str, Any]) -> list[dict[str, Any]]:
def resolve_manifest_skill(entry: dict[str, Any]) -> tuple[Path, str]:
name = entry["name"]
source = entry.get("source", "owned")
if source == "registry" and entry.get("registry"):
source = f"registry:{entry['registry']}"
return resolve_skill_source(name, source=source)
skiff/registry.py
+49
View File
@@ -56,3 +56,52 @@ def external_skill_path(name: str, entry: dict[str, Any] | None = None) -> Path:
f"registry 条目 {name!r} 的 path 超出外部仓库: {subpath!r}"
) from exc
return skill_path
def discover_external_skills(
name: str,
entry: dict[str, Any] | None = None,
) -> dict[str, Path]:
"""Discover a single registry skill or a collection of sibling skills."""
entry = entry or load_registry().get(name, {})
root = external_skill_path(name, entry)
if (root / "SKILL.md").is_file():
return {name: root}
if not root.is_dir():
return {}
resolved_root = root.resolve()
skills: dict[str, Path] = {}
for item in sorted(root.iterdir()):
skill_md = item / "SKILL.md"
if (
item.name.startswith("_")
or item.is_symlink()
or not item.is_dir()
or skill_md.is_symlink()
or not skill_md.is_file()
):
continue
resolved_item = item.resolve()
resolved_skill_md = skill_md.resolve()
try:
resolved_item.relative_to(resolved_root)
resolved_skill_md.relative_to(resolved_item)
except ValueError:
continue
skills[item.name] = item
return skills
def external_collection_skill_path(
collection: str,
skill_name: str,
entry: dict[str, Any] | None = None,
) -> Path:
skills = discover_external_skills(collection, entry)
if skill_name not in skills:
available = ", ".join(skills) or "(无)"
raise SystemExit(
f"registry collection {collection!r} 中找不到 skill {skill_name!r}。"
f"可用: {available}"
)
return skills[skill_name]
skiff/skills.py
+24 -2
View File
@@ -6,7 +6,11 @@ import re
from pathlib import Path
from skiff.paths import SKILLS_DIR, TEMPLATE_DIR, ensure_skills_home
from skiff.registry import external_skill_path, load_registry
from skiff.registry import (
external_collection_skill_path,
external_skill_path,
load_registry,
)
from skiff.sources import list_source_skills, load_sources, source_skills_root
@@ -66,6 +70,25 @@ def resolve_skill_source(name: str, *, source: str | None = None) -> tuple[Path,
return owned, "owned"
registry = load_registry()
sources = load_sources()
registry_collection = None
if source and source.startswith("registry:"):
registry_collection = source.split(":", 1)[1]
elif (
source
and source not in ("owned", "registry")
and source in registry
and source not in sources
):
registry_collection = source
if registry_collection:
path = external_collection_skill_path(
registry_collection,
name,
registry[registry_collection],
)
return path, f"registry:{registry_collection}"
if source == "registry":
if name not in registry:
raise SystemExit(f"registry 中找不到 skill: {name}")
@@ -77,7 +100,6 @@ def resolve_skill_source(name: str, *, source: str | None = None) -> tuple[Path,
)
return path, "registry"
sources = load_sources()
if source:
if source not in sources:
raise SystemExit(
skills/publish-docker-image/SKILL.md
+64
View File
@@ -0,0 +1,64 @@
---
name: publish-docker-image
description: >-
构建当前项目的 Docker 镜像，并将其上传到用户指定的镜像仓库。仅当用户显式指定
$publish-docker-image 或明确说“使用 publish-docker-image skill”时使用；
不要因普通编码、编辑 Dockerfile、本地构建、测试或一般 Docker 问题而自动触发。
---
# 发布 Docker 镜像
安全、可复现地构建当前提交对应的 Docker 镜像，并按用户指定的目标上传。
## 执行流程
1. 读取项目的 `AGENTS.md`、Dockerfile、构建脚本和相关发布文档。
2. 收集目标 registry、repository、tag、platform、构建上下文和 Dockerfile。优先使用用户已明确提供的值；缺少会改变发布结果的值时，停止并询问。
3. 检查 Git 工作区与当前提交。若存在未提交修改，明确说明镜像将包含哪些修改。
4. 按 [registry.md](references/registry.md) 检查仓库规则和认证状态。
5. 在执行外部写操作前，向用户展示完整镜像引用、platform、Dockerfile、构建上下文和源 commit。只有用户已明确要求上传到该目标时才继续。
6. 使用 [publish.sh](scripts/publish.sh) 构建并上传。不要自行拼接包含凭据的命令。
7. 检查命令退出状态，并尽可能获取远端 digest。
8. 汇报完整镜像引用、digest、platform、源 commit，以及是否包含未提交修改。
## 命令
默认构建并上传：
```bash
scripts/publish.sh \
--registry REGISTRY \
--repository NAMESPACE/IMAGE \
--tag TAG \
--platform PLATFORM
```
先验证而不构建或上传：
```bash
scripts/publish.sh \
--registry REGISTRY \
--repository NAMESPACE/IMAGE \
--tag TAG \
--platform PLATFORM \
--dry-run
```
仅当用户明确要求本地构建时使用 `--load`。多平台镜像不能使用 `--load`。
## 安全边界
- 不把密码、访问令牌或 Docker 配置写入 skill、项目文件、命令参数或输出。
- 不主动执行 `docker login`；认证缺失时让用户通过交互式登录或其凭据管理器完成。
- 不覆盖已存在的 release tag，除非用户明确授权。无法可靠检查远端 tag 时说明这一限制。
- 不把 `latest` 作为隐含默认 tag。
- 不上传用户未指定的附加 tag。
- 不擅自修改 Dockerfile、发布配置、仓库权限或镜像保留策略。
- 若仓库、tag、platform 或目标环境存在歧义，在上传前询问用户。
## 验证
- 确认 `docker buildx build` 成功且启用了 `--push`。
- 优先用 `docker buildx imagetools inspect FULL_IMAGE_REF` 验证远端引用及平台。
- 记录远端 digest；若仓库不允许检查，明确报告只验证了 push 命令成功。
- 将发布所用的 Git commit 与工作区状态一并报告。
skills/publish-docker-image/references/registry.md
+39
View File
@@ -0,0 +1,39 @@
# 镜像仓库规则
执行发布前，从用户输入和当前项目文档中确定以下信息：
| 字段 | 要求 |
| --- | --- |
| Registry | 必须显式确定，例如 `registry.example.com` |
| Repository | 必须包含项目约定的 namespace，例如 `team/service` |
| Tag | 必须显式确定；优先使用版本号或 Git SHA |
| Platform | 必须显式确定，例如 `linux/amd64` 或 `linux/amd64,linux/arm64` |
| Dockerfile | 默认 `Dockerfile`，不存在或项目另有约定时明确指定 |
| Context | 默认当前项目根目录 |
## 信息来源优先级
1. 用户本次请求中明确给出的值。
2. 当前项目的 `AGENTS.md` 和发布文档。
3. `Makefile`、CI 配置、Compose 文件或现有构建脚本中一致且无歧义的配置。
4. 询问用户。
不要从其他项目、shell history 或无关的本地配置中猜测发布目标。
## 认证
使用 Docker 当前配置的 credential helper 或已有登录状态。可用不泄露凭据的只读操作检查目标是否可访问。认证缺失或过期时，停止并让用户自行完成登录。
不要读取、打印或复制以下内容：
- registry 密码或访问令牌
- `~/.docker/config.json` 中的认证字段
- CI secret 的值
- 包含凭据的环境变量值
## Tag 策略
- release tag（如 `v1.2.3`）默认视为不可变。
- Git SHA tag 应对应当前源 commit。
- `latest`、`stable` 等浮动 tag 只有在用户明确要求时才发布。
- 用户未给 tag 且项目没有唯一明确规则时，必须询问，不要自行选择。
skills/publish-docker-image/scripts/publish.sh
Executable
+116
View File
@@ -0,0 +1,116 @@
#!/usr/bin/env bash
set -euo pipefail
usage() {
sed -n '2,22p' "$0"
}
# Build and publish a Docker image with buildx.
#
# Usage:
# publish.sh --registry HOST --repository PATH --tag TAG --platform PLATFORMS [options]
#
# Options:
# --registry HOST Registry host, without a URL scheme
# --repository PATH Repository path, such as team/service
# --tag TAG Image tag
# --platform LIST Comma-separated platforms
# --file PATH Dockerfile path (default: Dockerfile)
# --context PATH Build context (default: .)
# --builder NAME Existing buildx builder
# --load Load a single-platform image instead of pushing
# --dry-run Print the resolved build without executing it
# --help Show this help
registry=
repository=
tag=
platform=
dockerfile=Dockerfile
build_context=.
builder=
mode=push
dry_run=false
while (($#)); do
case "$1" in
--registry) registry=${2-}; shift 2 ;;
--repository) repository=${2-}; shift 2 ;;
--tag) tag=${2-}; shift 2 ;;
--platform) platform=${2-}; shift 2 ;;
--file) dockerfile=${2-}; shift 2 ;;
--context) build_context=${2-}; shift 2 ;;
--builder) builder=${2-}; shift 2 ;;
--load) mode=load; shift ;;
--dry-run) dry_run=true; shift ;;
--help|-h) usage; exit 0 ;;
*) printf 'Unknown argument: %s\n' "$1" >&2; usage >&2; exit 2 ;;
esac
done
for required_name in registry repository tag platform; do
if [[ -z ${!required_name} ]]; then
printf 'Missing required option: --%s\n' "$required_name" >&2
exit 2
fi
done
if [[ $registry == *://* || $registry == */* ]]; then
printf '%s\n' 'Registry must be a host without a URL scheme or path.' >&2
exit 2
fi
if [[ $repository == /* || $repository == */ || $repository != */* ]]; then
printf '%s\n' 'Repository must look like namespace/image.' >&2
exit 2
fi
if [[ $tag == *:* || $tag == */* ]]; then
printf '%s\n' 'Tag must not contain ":" or "/".' >&2
exit 2
fi
if [[ $mode == load && $platform == *,* ]]; then
printf '%s\n' '--load supports only one platform.' >&2
exit 2
fi
if [[ ! -f $dockerfile ]]; then
printf 'Dockerfile not found: %s\n' "$dockerfile" >&2
exit 2
fi
if [[ ! -d $build_context ]]; then
printf 'Build context not found: %s\n' "$build_context" >&2
exit 2
fi
if [[ $dry_run == false ]] && ! command -v docker >/dev/null 2>&1; then
printf '%s\n' 'docker is not installed or not available in PATH.' >&2
exit 127
fi
image_ref="${registry}/${repository}:${tag}"
build_cmd=(docker buildx build --file "$dockerfile" --platform "$platform" --tag "$image_ref")
if [[ -n $builder ]]; then
build_cmd+=(--builder "$builder")
fi
if [[ $mode == push ]]; then
build_cmd+=(--push)
else
build_cmd+=(--load)
fi
build_cmd+=("$build_context")
printf 'Image: %s\n' "$image_ref"
printf 'Platform: %s\n' "$platform"
printf 'Dockerfile: %s\n' "$dockerfile"
printf 'Context: %s\n' "$build_context"
printf 'Mode: %s\n' "$mode"
if [[ $dry_run == true ]]; then
printf 'Command:'
printf ' %q' "${build_cmd[@]}"
printf '\n'
exit 0
fi
"${build_cmd[@]}"
if [[ $mode == push ]]; then
docker buildx imagetools inspect "$image_ref"
fi
skills/skiff/SKILL.md
+10
View File
@@ -106,6 +106,16 @@ skiff add discussion-notes -a cursor -g -y
skiff add discussion-notes -a cursor -a codex -g -y
```
registry 条目既可以指向单个 skill，也可以指向包含多个 skill 目录的
collection。安装 collection 全部内容或其中一个：
```bash
skiff add waza -a codex -g -y
skiff add waza/think -a codex -g -y
```
`skiff select` 会把 collection 展开为 `waza/think`、`waza/ui` 等候选项。
卸载：
```bash
tests/test_custom_sources.py
+29
View File
@@ -137,6 +137,35 @@ class CustomSourceTests(unittest.TestCase):
self.assertIn("owned/code-review", result.stderr)
self.assertIn("company/code-review", result.stderr)
def test_custom_source_namespace_is_not_shadowed_by_registry_collection(self) -> None:
company = self.home / "company"
expected = write_skill(company / "skills", "code-review")
registry_repo = self.home / "registry-repo"
write_skill(registry_repo / "skills", "other-skill")
self.skills_home.joinpath("registry.yaml").write_text(
"company:\n"
f" repo: {registry_repo}\n"
" ref: main\n"
" path: skills\n",
encoding="utf-8",
)
added = self.run_skiff("source", "add", "company", "--local", str(company))
result = self.run_skiff(
"add",
"company/code-review",
"-g",
"-a",
"codex",
)
self.assertEqual(added.returncode, 0, added.stderr)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(
(self.home / ".codex" / "skills" / "code-review").resolve(),
expected.resolve(),
)
def test_sync_uses_manifest_source(self) -> None:
company = self.home / "company"
expected = write_skill(company / "skills", "code-review")
tests/test_registry_collections.py
+179
View File
@@ -0,0 +1,179 @@
from __future__ import annotations
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
def write_skill(root: Path, name: str) -> Path:
skill = root / name
skill.mkdir(parents=True)
skill.joinpath("SKILL.md").write_text(
f"---\nname: {name}\ndescription: >-\n"
f" Test {name} skill.\n---\n\n# {name}\n",
encoding="utf-8",
)
return skill
class RegistryCollectionTests(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.home = Path(self.temp_dir.name)
self.skills_home = self.home / ".skills"
(self.skills_home / "skills").mkdir(parents=True)
self.upstream = self.home / "upstream"
self.first = write_skill(self.upstream / "skills", "first-skill")
self.second = write_skill(self.upstream / "skills", "second-skill")
subprocess.run(
["git", "init", "-b", "main", str(self.upstream)],
check=True,
capture_output=True,
)
subprocess.run(["git", "-C", str(self.upstream), "add", "."], check=True)
subprocess.run(
[
"git",
"-C",
str(self.upstream),
"-c",
"user.name=Test",
"-c",
"user.email=test@example.com",
"commit",
"-m",
"initial",
],
check=True,
capture_output=True,
)
self.skills_home.joinpath("registry.yaml").write_text(
"test-pack:\n"
f" repo: {self.upstream}\n"
" ref: main\n"
" path: skills\n",
encoding="utf-8",
)
def tearDown(self) -> None:
self.temp_dir.cleanup()
def run_skiff(self, *args: str) -> subprocess.CompletedProcess[str]:
env = os.environ.copy()
env["HOME"] = str(self.home)
env["PYTHONPATH"] = str(REPO_ROOT)
return subprocess.run(
[sys.executable, "-m", "skiff", *args],
cwd=REPO_ROOT,
env=env,
text=True,
capture_output=True,
check=False,
)
def test_add_collection_installs_every_discovered_skill(self) -> None:
result = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
self.assertEqual(result.returncode, 0, result.stderr)
skill_dir = self.home / ".codex" / "skills"
self.assertTrue((skill_dir / "first-skill").is_symlink())
self.assertTrue((skill_dir / "first-skill" / "SKILL.md").is_file())
self.assertTrue((skill_dir / "second-skill").is_symlink())
self.assertTrue((skill_dir / "second-skill" / "SKILL.md").is_file())
self.assertFalse((skill_dir / "test-pack").exists())
def test_add_qualified_collection_skill_installs_only_that_skill(self) -> None:
result = self.run_skiff("add", "test-pack/second-skill", "-g", "-a", "codex")
self.assertEqual(result.returncode, 0, result.stderr)
skill_dir = self.home / ".codex" / "skills"
self.assertFalse((skill_dir / "first-skill").exists())
self.assertTrue((skill_dir / "second-skill").is_symlink())
self.assertTrue((skill_dir / "second-skill" / "SKILL.md").is_file())
def test_add_collection_with_registry_source_installs_all(self) -> None:
result = self.run_skiff(
"add",
"test-pack",
"--source",
"registry",
"-g",
"-a",
"codex",
)
self.assertEqual(result.returncode, 0, result.stderr)
skill_dir = self.home / ".codex" / "skills"
self.assertTrue((skill_dir / "first-skill").is_symlink())
self.assertTrue((skill_dir / "second-skill").is_symlink())
def test_project_add_records_collection_for_sync(self) -> None:
project = self.home / "project"
project.mkdir()
added = self.run_skiff(
"add",
"test-pack/first-skill",
"--project",
str(project),
"-a",
"codex",
)
link = project / ".agents" / "skills" / "first-skill"
link.unlink()
shutil.rmtree(self.home / ".local" / "share" / "skills" / "externals")
synced = self.run_skiff(
"sync",
"--project",
str(project),
"-a",
"codex",
)
self.assertEqual(added.returncode, 0, added.stderr)
self.assertIn('registry: "test-pack"', project.joinpath(".skills.yaml").read_text())
self.assertEqual(synced.returncode, 0, synced.stderr)
self.assertTrue(link.is_symlink())
self.assertTrue((link / "SKILL.md").is_file())
def test_collection_preflight_prevents_partial_install(self) -> None:
blocked = self.home / ".codex" / "skills" / "second-skill"
blocked.mkdir(parents=True)
result = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
self.assertNotEqual(result.returncode, 0)
self.assertFalse((self.home / ".codex" / "skills" / "first-skill").exists())
self.assertTrue(blocked.is_dir())
def test_remove_collection_removes_all_child_links(self) -> None:
installed = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
removed = self.run_skiff("remove", "test-pack", "-g", "-a", "codex")
self.assertEqual(installed.returncode, 0, installed.stderr)
self.assertEqual(removed.returncode, 0, removed.stderr)
skill_dir = self.home / ".codex" / "skills"
self.assertFalse((skill_dir / "first-skill").exists())
self.assertFalse((skill_dir / "second-skill").exists())
def test_status_expands_collection_children(self) -> None:
installed = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
status = self.run_skiff("status", "-a", "codex")
self.assertEqual(installed.returncode, 0, installed.stderr)
self.assertEqual(status.returncode, 0, status.stderr)
self.assertIn("[registry:test-pack] first-skill", status.stdout)
self.assertIn("[registry:test-pack] second-skill", status.stdout)
self.assertNotIn("[registry] test-pack\n (未 fetch)", status.stdout)
if __name__ == "__main__":
unittest.main()
tests/test_select.py
+86
View File
@@ -76,6 +76,27 @@ class SelectorTests(unittest.TestCase):
with self.assertRaisesRegex(SystemExit, "超出外部仓库"):
external_skill_path("unsafe-skill", entry)
def test_collection_discovery_ignores_symlinked_skill(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
checkout = root / "checkout"
collection = checkout / "skills"
outside = root / "outside"
collection.mkdir(parents=True)
outside.mkdir()
outside.joinpath("SKILL.md").write_text("---\n", encoding="utf-8")
collection.joinpath("escaped").symlink_to(outside, target_is_directory=True)
entry = {
"repo": "https://example.test/skills.git",
"ref": "main",
"path": "skills",
}
with patch("skiff.registry.external_checkout_path", return_value=checkout):
from skiff.registry import discover_external_skills
self.assertEqual(discover_external_skills("unsafe", entry), {})
class SelectCommandTests(unittest.TestCase):
def test_non_tty_exits_with_add_guidance(self) -> None:
@@ -101,6 +122,9 @@ class SelectCommandTests(unittest.TestCase):
def test_project_selection_installs_new_and_records_all_selected(self) -> None:
with tempfile.TemporaryDirectory() as temp:
project = Path(temp)
external = project / "external-one"
external.mkdir()
external.joinpath("SKILL.md").write_text("---\n", encoding="utf-8")
args = argparse.Namespace(
agents=[["codex"]],
global_scope=False,
@@ -130,7 +154,14 @@ class SelectCommandTests(unittest.TestCase):
}
},
),
patch.object(cli, "_registry_skill_names", return_value=["external-one"]),
patch.object(cli, "external_skill_path", return_value=external),
patch.object(cli, "_list_fully_installed_names", return_value=["owned-one"]),
patch.object(
cli,
"_is_fully_installed",
side_effect=lambda name, expected, project_root, targets: name == "owned-one",
),
patch.object(
cli,
"select_skills",
@@ -182,6 +213,61 @@ class SelectCommandTests(unittest.TestCase):
selector.assert_not_called()
def test_select_expands_registry_collection_choices(self) -> None:
args = argparse.Namespace(
agents=[["codex"]],
global_scope=True,
project=None,
yes=False,
)
stdin = Mock()
stdout = Mock()
stdin.isatty.return_value = True
stdout.isatty.return_value = True
selected_choices: list[SkillChoice] = []
installed: list[tuple[str, str | None]] = []
def choose(choices: list[SkillChoice]) -> set[str]:
selected_choices.extend(choices)
return {"waza/think"}
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={
"waza": {
"repo": "https://example.test/waza.git",
"ref": "main",
"path": "skills",
}
},
),
patch.object(
cli,
"_registry_skill_names",
return_value=["think", "ui"],
),
patch.object(cli, "_list_fully_installed_names", return_value=[]),
patch.object(cli, "_is_fully_installed", return_value=False),
patch.object(cli, "select_skills", side_effect=choose),
patch.object(
cli,
"_install_skill",
side_effect=lambda name, targets, project_root, **kwargs: installed.append(
(name, kwargs.get("source"))
),
),
):
cli.cmd_select(args)
self.assertEqual([choice.name for choice in selected_choices], ["waza/think", "waza/ui"])
self.assertEqual(installed, [("think", "registry:waza")])
def test_install_rolls_back_earlier_target_when_later_target_fails(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)