4 Commits

19 changed files with 961 additions and 61 deletions
+3 -2
View File
@@ -60,6 +60,7 @@ AGENTS.md # 本文档
| Skill | 说明 | | Skill | 说明 |
| ---------------------------------------------------------------------- | ------------------------------------------------- | | ---------------------------------------------------------------------- | ------------------------------------------------- |
| [ack](skills/ack/SKILL.md) | ACK 入口:显式初始化、检查并运行项目三角色协作闭环 |
| [skiff](skills/skiff/SKILL.md) | 本项目工作流:创建、使用、反馈与更新 owned skill | | [skiff](skills/skiff/SKILL.md) | 本项目工作流:创建、使用、反馈与更新 owned skill |
| [declarative-openspec-loop](skills/declarative-openspec-loop/SKILL.md) | 声明式编程循环:用户提供校验方式,Agent 自动 propose/apply/校验并迭代直到通过 | | [declarative-openspec-loop](skills/declarative-openspec-loop/SKILL.md) | 声明式编程循环:用户提供校验方式,Agent 自动 propose/apply/校验并迭代直到通过 |
| [discussion-notes](skills/discussion-notes/SKILL.md) | 讨论沉淀:边讨论边维护 Markdown 笔记,无 .raw.md | | [discussion-notes](skills/discussion-notes/SKILL.md) | 讨论沉淀:边讨论边维护 Markdown 笔记,无 .raw.md |
@@ -78,9 +79,9 @@ AGENTS.md # 本文档
| Kit | 说明 | 推荐落地位置 | | Kit | 说明 | 推荐落地位置 |
| --- | --- | --- | | --- | --- | --- |
| [agent-collaboration-kit](kits/agent-collaboration-kit/README.md) | 多 Agent 协作闭环规范包 | `docs/agent-collaboration-kit/` | | [ack](kits/ack/README.md) | 多 Agent 协作闭环规范包 | `docs/ack/` |
新建规范包:创建 `kits/<name>/README.md`,写清适用场景、复制到项目后的推荐目录、需要项目填充的占位符,并在本仓库 commit。 新建规范包:创建 `kits/<name>/README.md`,写清适用场景、复制到项目后的推荐目录、需要项目填充的占位符,并在本仓库 commit。已有 kit 可通过 `skiff kit init <name>` 初始化到项目。
### 外部(External Git) ### 外部(External Git)
+8 -1
View File
@@ -38,6 +38,7 @@ AGENTS.md # 详细规范与架构说明
| Skill | 说明 | | Skill | 说明 |
|-------|------| |-------|------|
| [ack](skills/ack/SKILL.md) | 显式初始化、检查并运行 ACK 三角色协作闭环 |
| [skiff](skills/skiff/SKILL.md) | 在项目中创建、安装、反馈和维护 owned skill | | [skiff](skills/skiff/SKILL.md) | 在项目中创建、安装、反馈和维护 owned skill |
| [declarative-openspec-loop](skills/declarative-openspec-loop/SKILL.md) | 声明式编程循环:用户提供校验方式,Agent 自动迭代直到通过 | | [declarative-openspec-loop](skills/declarative-openspec-loop/SKILL.md) | 声明式编程循环:用户提供校验方式,Agent 自动迭代直到通过 |
| [discussion-notes](skills/discussion-notes/SKILL.md) | 讨论沉淀:边讨论边维护 Markdown 笔记 | | [discussion-notes](skills/discussion-notes/SKILL.md) | 讨论沉淀:边讨论边维护 Markdown 笔记 |
@@ -48,7 +49,13 @@ AGENTS.md # 详细规范与架构说明
| Kit | 说明 | | Kit | 说明 |
|-----|------| |-----|------|
| [agent-collaboration-kit](kits/agent-collaboration-kit/README.md) | 多 Agent 协作闭环规范包,复制到项目 `docs/agent-collaboration-kit/` 后按项目情况填写 | | [ack](kits/ack/README.md) | 多 Agent 协作闭环规范包,通过 `skiff kit init ack` 初始化到项目 `docs/ack/` |
```bash
skiff kit init ack # 当前项目,软链接到 SSOT(推荐)
skiff kit init ack --project ~/app # 指定项目
skiff kit init ack --copy # 整份复制,后续需手动升级
```
新建 skill: 新建 skill:
+9
View File
@@ -20,6 +20,15 @@
# ref: main # ref: main
# path: skills # path: skills
skills:
repo: ~/.skills
ref: main
path: skills
description: 本地自研 Agent Skills 集合,由 skiff 统一创建、维护和安装。
tags:
- skill-management
- owned
waza: waza:
repo: https://github.com/tw93/Waza.git repo: https://github.com/tw93/Waza.git
ref: main ref: main
+6 -1
View File
@@ -65,6 +65,8 @@ skiff bootstrap
| 命令 | 说明 | | 命令 | 说明 |
|------|------| |------|------|
| `skiff bootstrap` | 将本项目的 `skiff` skill 全局安装到所有 Agent | | `skiff bootstrap` | 将本项目的 `skiff` skill 全局安装到所有 Agent |
| `skiff update` | 在 `~/.skills` 执行 `git pull`,更新 skiff 自身 |
| `skiff kit init <name> [--project DIR] [--copy]` | 在项目的 `docs/<name>/` 初始化规范包;默认软链接到 SSOT |
### 全局安装(自研 skill) ### 全局安装(自研 skill)
@@ -112,7 +114,10 @@ skiff select --project ~/code/app # 指定项目
``` ```
使用方向键移动、空格勾选、`/` 搜索、Enter 安装,按 `q` 或 Esc 使用方向键移动、空格勾选、`/` 搜索、Enter 安装,按 `q` 或 Esc
取消。已经安装到目标范围的 skill 默认勾选;取消勾选不会卸载已有 skill。 取消。普通 `skiff select` 只安装到项目,并在每一项旁只读显示各 Agent 的
全局安装状态;`skiff select -g` 只安装到全局。已经安装到目标范围的 skill
默认勾选;取消勾选不会卸载已有 skill,卸载请使用 `skiff remove`。如果全局
存在同名但指向其它来源的 skill,项目选择器会显示“全局同名冲突”。
项目模式会把成功选择的项目写入 `.skills.yaml`。非交互环境请使用 项目模式会把成功选择的项目写入 `.skills.yaml`。非交互环境请使用
`skiff add <name>...`。使用 `-a` 限定 Agent 时,该范围会记录在对应的 `skiff add <name>...`。使用 `-a` 限定 Agent 时,该范围会记录在对应的
skill 条目中,后续 `skiff sync` 不会扩散到其他 Agent。 skill 条目中,后续 `skiff sync` 不会扩散到其他 Agent。
+189 -26
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import argparse import argparse
import re import re
import shutil
import subprocess import subprocess
import sys import sys
from datetime import datetime from datetime import datetime
@@ -17,6 +18,7 @@ from skiff.paths import (
DRAFTS_DIR, DRAFTS_DIR,
EXTERNALS_DIR, EXTERNALS_DIR,
CONFIG_FILE, CONFIG_FILE,
KITS_DIR,
SKILLS_DIR, SKILLS_DIR,
SKILLS_HOME, SKILLS_HOME,
TEMPLATE_DIR, TEMPLATE_DIR,
@@ -36,6 +38,7 @@ from skiff.registry import (
external_checkout_path, external_checkout_path,
external_skill_path, external_skill_path,
load_registry, load_registry,
registry_repo,
save_registry, save_registry,
) )
from skiff.selector import SkillChoice, select_skills from skiff.selector import SkillChoice, select_skills
@@ -80,6 +83,13 @@ def _project_root(explicit: str | None = None) -> Path:
return find_repo_root() or Path.cwd() return find_repo_root() or Path.cwd()
def _render_kit_template(source: Path, destination: Path, values: dict[str, str]) -> None:
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")
def _collect_skill_names(positional: list[str] | None, flagged: list[str] | None) -> list[str]: def _collect_skill_names(positional: list[str] | None, flagged: list[str] | None) -> list[str]:
names = list(positional or []) names = list(positional or [])
if flagged: if flagged:
@@ -130,7 +140,7 @@ def _ensure_registry_fetched(name: str, entry: dict[str, object]) -> None:
if (path / "SKILL.md").is_file() or discover_external_skills(name, entry): if (path / "SKILL.md").is_file() or discover_external_skills(name, entry):
return return
repo = str(entry["repo"]) repo = registry_repo(entry)
ref = entry.get("ref", "main") ref = entry.get("ref", "main")
dest = external_checkout_path(name, entry) dest = external_checkout_path(name, entry)
dest.parent.mkdir(parents=True, exist_ok=True) dest.parent.mkdir(parents=True, exist_ok=True)
@@ -140,7 +150,7 @@ def _ensure_registry_fetched(name: str, entry: dict[str, object]) -> None:
) )
_print(f"拉取外部 skill: {name}") _print(f"拉取外部 skill: {name}")
subprocess.run( subprocess.run(
["git", "clone", "--branch", ref, "--", repo, str(dest)], ["git", "clone", "--depth", "1", "--branch", ref, "--", repo, str(dest)],
check=True, check=True,
) )
if not discover_external_skills(name, entry): if not discover_external_skills(name, entry):
@@ -277,6 +287,29 @@ def _is_fully_installed(
) )
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( def _remove_skill(
name: str, name: str,
targets: list[str], targets: list[str],
@@ -337,6 +370,13 @@ def cmd_bootstrap(args: argparse.Namespace) -> None:
_print("已安装项目 skill 到所有 agent") _print("已安装项目 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( def _installed_links(
name: str, name: str,
targets: list[str], targets: list[str],
@@ -516,17 +556,40 @@ def cmd_select(args: argparse.Namespace) -> None:
for name in [*owned_names, *registry]: for name in [*owned_names, *registry]:
validate_skill_name(name) validate_skill_name(name)
choices = [ def make_choice(
SkillChoice( *,
name: str,
installed_name: str,
expected: Path,
kind: str,
description: str,
indent: int = 0,
) -> SkillChoice:
return SkillChoice(
name=name, name=name,
kind="owned", kind=kind,
description=skill_description(name) or "", description=description,
installed=_is_fully_installed( installed=_is_fully_installed(
name, installed_name,
SKILLS_DIR / name, expected,
project_root, project_root,
targets, 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="owned",
description=skill_description(name) or "",
) )
for name in owned_names for name in owned_names
] ]
@@ -541,40 +604,60 @@ def cmd_select(args: argparse.Namespace) -> None:
_err(f"警告: registry 条目与 owned skill 同名,已忽略 external: {package}") _err(f"警告: registry 条目与 owned skill 同名,已忽略 external: {package}")
continue continue
choices.append( choices.append(
SkillChoice( make_choice(
name=package, name=package,
installed_name=package,
expected=root,
kind="external", kind="external",
description=str(entry.get("description", "")), description=str(entry.get("description", "")),
installed=_is_fully_installed(
package,
root,
project_root,
targets,
),
) )
) )
choice_requests[package] = (package, "registry") choice_requests[package] = (package, "registry")
continue continue
for skill_name in skill_names: for skill_name in skill_names:
expected = root / skill_name
if (
skill_name in owned_names
and expected.resolve() == (SKILLS_DIR / skill_name).resolve()
):
continue
choice_name = f"{package}/{skill_name}" choice_name = f"{package}/{skill_name}"
description = read_skill_meta(root / skill_name).get("description", "") description = read_skill_meta(expected).get("description", "")
choices.append( choices.append(
SkillChoice( make_choice(
name=choice_name, name=choice_name,
installed_name=skill_name,
expected=expected,
kind=f"external:{package}", kind=f"external:{package}",
description=description, description=description,
installed=_is_fully_installed( indent=1,
skill_name,
root / skill_name,
project_root,
targets,
),
) )
) )
choice_requests[choice_name] = (skill_name, f"registry:{package}") choice_requests[choice_name] = (skill_name, f"registry:{package}")
child_names = tuple(
f"{package}/{skill_name}"
for skill_name in skill_names
if f"{package}/{skill_name}" in choice_requests
)
if child_names:
first_child = len(choices) - len(child_names)
choices.insert(
first_child,
SkillChoice(
name=package,
kind="repository",
description=str(entry.get("description", "")),
children=child_names,
),
)
try: try:
selected = select_skills(choices) scope_label = (
"全局"
if project_root is None
else f"项目 {project_root}(全局状态只读)"
)
selected = select_skills(choices, scope_label=scope_label)
except (RuntimeError, OSError) as exc: except (RuntimeError, OSError) as exc:
raise SystemExit(str(exc)) from exc raise SystemExit(str(exc)) from exc
if selected is None: if selected is None:
@@ -729,7 +812,7 @@ def cmd_fetch(args: argparse.Namespace) -> None:
raise SystemExit(f"registry 中不存在: {args.name}") raise SystemExit(f"registry 中不存在: {args.name}")
entry = registry[args.name] entry = registry[args.name]
repo = entry["repo"] repo = registry_repo(entry)
ref = entry.get("ref", "main") ref = entry.get("ref", "main")
dest = external_checkout_path(args.name, entry) dest = external_checkout_path(args.name, entry)
@@ -743,7 +826,7 @@ def cmd_fetch(args: argparse.Namespace) -> None:
else: else:
_print(f"克隆: {repo} -> {dest}") _print(f"克隆: {repo} -> {dest}")
subprocess.run( subprocess.run(
["git", "clone", "--branch", ref, "--", repo, str(dest)], ["git", "clone", "--depth", "1", "--branch", ref, "--", repo, str(dest)],
check=True, check=True,
) )
@@ -1039,6 +1122,71 @@ def cmd_doctor(args: argparse.Namespace) -> None:
sys.exit(1) sys.exit(1)
def cmd_kit_init(args: argparse.Namespace) -> None:
"""将 owned kit 初始化到目标项目。"""
ensure_skills_home()
kit_source = KITS_DIR / args.name
if not kit_source.is_dir():
available = sorted(path.name for path in KITS_DIR.iterdir() if path.is_dir()) if KITS_DIR.is_dir() else []
suffix = f";可用 kit: {', '.join(available)}" if available else ""
raise SystemExit(f"kit 不存在: {args.name}{suffix}")
project = _project_root(args.project)
if not project.is_dir():
raise SystemExit(f"项目目录不存在: {project}")
destination = project / "docs" / args.name
kit_target = destination / "kit"
project_file = destination / "project.md"
tasks_file = destination / "tasks.yaml"
managed_targets = (kit_target, project_file, tasks_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 = kit_source / "templates" / "project.template.md"
tasks_template = kit_source / "templates" / "tasks.template.yaml"
missing = [path for path in (project_template, tasks_template) if not path.is_file()]
if missing:
paths = ", ".join(str(path.relative_to(SKILLS_HOME)) for path in missing)
raise SystemExit(f"kit 缺少初始化模板: {paths}")
destination.mkdir(parents=True, exist_ok=True)
if args.copy:
shutil.copytree(kit_source, kit_target)
mode = "copy"
else:
kit_target.symlink_to(kit_source.resolve(), target_is_directory=True)
mode = "symlink"
version_file = kit_source / "VERSION"
kit_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",
"<kit_version>": kit_version,
"<接入时的 ack 版本,见 kit 根 VERSION>": kit_version,
"<YYYY-MM-DDTHH:mm:ss+TZ>": now,
}
_render_kit_template(project_template, project_file, values)
_render_kit_template(tasks_template, tasks_file, values)
validator = kit_target / "scripts" / "validate_tasks.py"
if validator.is_file():
subprocess.run([sys.executable, str(validator), str(tasks_file)], check=True)
_print(f"✓ kit 初始化完成: {args.name}")
_print(f" 项目: {project}")
_print(f" 模式: {mode}")
_print(f" kit: {kit_target}")
_print(f" 覆盖层: {project_file}")
_print(f" 任务板: {tasks_file}")
_print("下一步: 填写 project.md 中的项目命令、路径权限和 Base URL")
def _add_common_flags(parser: argparse.ArgumentParser) -> None: def _add_common_flags(parser: argparse.ArgumentParser) -> None:
parser.add_argument( parser.add_argument(
"-a", "-a",
@@ -1075,6 +1223,9 @@ def build_parser() -> argparse.ArgumentParser:
) )
p_bootstrap.set_defaults(func=cmd_bootstrap) 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 = sub.add_parser("list", help="列出所有 source 中的 skill")
p_list.add_argument("--source", help="只列出指定来源(owned、registry 或 custom source)") p_list.add_argument("--source", help="只列出指定来源(owned、registry 或 custom source)")
p_list.set_defaults(func=cmd_list) p_list.set_defaults(func=cmd_list)
@@ -1207,6 +1358,18 @@ def build_parser() -> argparse.ArgumentParser:
p_doctor.add_argument("--fix", action="store_true", help="自动修复可修复的软链") p_doctor.add_argument("--fix", action="store_true", help="自动修复可修复的软链")
p_doctor.set_defaults(func=cmd_doctor) p_doctor.set_defaults(func=cmd_doctor)
p_kit = sub.add_parser("kit", help="管理项目规范包")
kit_sub = p_kit.add_subparsers(dest="kit_command", required=True)
p_kit_init = kit_sub.add_parser("init", help="在项目中初始化 kit")
p_kit_init.add_argument("name", help="kit 名称")
p_kit_init.add_argument("--project", help="项目根目录(默认自动检测或当前目录)")
p_kit_init.add_argument(
"--copy",
action="store_true",
help="复制 kit,而不是创建指向 SSOT 的软链接",
)
p_kit_init.set_defaults(func=cmd_kit_init)
return parser return parser
+1
View File
@@ -7,6 +7,7 @@ from pathlib import Path
HOME = Path.home() HOME = Path.home()
SKILLS_HOME = HOME / ".skills" SKILLS_HOME = HOME / ".skills"
SKILLS_DIR = SKILLS_HOME / "skills" SKILLS_DIR = SKILLS_HOME / "skills"
KITS_DIR = SKILLS_HOME / "kits"
TEMPLATE_DIR = SKILLS_DIR / "_template" TEMPLATE_DIR = SKILLS_DIR / "_template"
DRAFTS_DIR = SKILLS_HOME / ".drafts" DRAFTS_DIR = SKILLS_HOME / ".drafts"
REGISTRY_FILE = SKILLS_HOME / "registry.yaml" REGISTRY_FILE = SKILLS_HOME / "registry.yaml"
+12 -1
View File
@@ -26,6 +26,12 @@ 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") path.write_text(yaml_io.safe_dump(data, allow_unicode=True, sort_keys=False), encoding="utf-8")
def registry_repo(entry: dict[str, Any]) -> str:
"""Expand a local home-relative repo while leaving remote URLs unchanged."""
repo = str(entry.get("repo", ""))
return str(Path(repo).expanduser()) if repo.startswith("~") else repo
def external_repo_path(entry: dict[str, Any]) -> Path: def external_repo_path(entry: dict[str, Any]) -> Path:
"""Return the shared checkout path for a repo/ref pair.""" """Return the shared checkout path for a repo/ref pair."""
from skiff.paths import EXTERNALS_DIR from skiff.paths import EXTERNALS_DIR
@@ -37,9 +43,14 @@ def external_repo_path(entry: dict[str, Any]) -> Path:
def external_checkout_path(name: str, entry: dict[str, Any]) -> Path: def external_checkout_path(name: str, entry: dict[str, Any]) -> Path:
"""Prefer an existing pre-shared-cache checkout for compatibility.""" """Use a local repo directly, otherwise return its external checkout."""
from skiff.paths import EXTERNALS_DIR from skiff.paths import EXTERNALS_DIR
configured_repo = str(entry.get("repo", ""))
local_repo = Path(registry_repo(entry))
if configured_repo.startswith("~") and local_repo.is_dir():
return local_repo.resolve()
legacy = EXTERNALS_DIR / name legacy = EXTERNALS_DIR / name
return legacy if legacy.is_dir() else external_repo_path(entry) return legacy if legacy.is_dir() else external_repo_path(entry)
+136 -21
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import unicodedata
from dataclasses import dataclass from dataclasses import dataclass
from typing import Callable, Iterable from typing import Callable, Iterable
@@ -12,6 +13,38 @@ class SkillChoice:
kind: str kind: str
description: str = "" description: str = ""
installed: bool = False installed: bool = False
readonly_status: str = ""
children: tuple[str, ...] = ()
indent: int = 0
def fit_to_width(text: str, width: int) -> str:
"""Trim text to terminal display width without splitting wide characters."""
if width <= 0:
return ""
result: list[str] = []
used = 0
for char in text:
if unicodedata.combining(char):
char_width = 0
else:
char_width = 2 if unicodedata.east_asian_width(char) in {"W", "F"} else 1
if used + char_width > width:
break
result.append(char)
used += char_width
return "".join(result)
def display_width(text: str) -> int:
return sum(
0
if unicodedata.combining(char)
else 2
if unicodedata.east_asian_width(char) in {"W", "F"}
else 1
for char in text
)
def filter_choices(choices: Iterable[SkillChoice], query: str) -> list[SkillChoice]: def filter_choices(choices: Iterable[SkillChoice], query: str) -> list[SkillChoice]:
@@ -24,12 +57,14 @@ def filter_choices(choices: Iterable[SkillChoice], query: str) -> list[SkillChoi
if needle in choice.name.casefold() if needle in choice.name.casefold()
or needle in choice.kind.casefold() or needle in choice.kind.casefold()
or needle in choice.description.casefold() or needle in choice.description.casefold()
or needle in choice.readonly_status.casefold()
] ]
def select_skills( def select_skills(
choices: list[SkillChoice], choices: list[SkillChoice],
*, *,
scope_label: str = "当前作用域",
wrapper: Callable[..., set[str] | None] | None = None, wrapper: Callable[..., set[str] | None] | None = None,
) -> set[str] | None: ) -> set[str] | None:
"""Open the selector. Return names, or None when cancelled.""" """Open the selector. Return names, or None when cancelled."""
@@ -41,36 +76,103 @@ def select_skills(
except ImportError as exc: except ImportError as exc:
raise RuntimeError("当前 Python 环境不支持 curses,无法打开交互界面") from exc raise RuntimeError("当前 Python 环境不支持 curses,无法打开交互界面") from exc
selected = {choice.name for choice in choices if choice.installed} selected = {
choice.name
for choice in choices
if choice.installed and not choice.children
}
def mark_for(choice: SkillChoice) -> str:
if not choice.children:
return "x" if choice.name in selected else " "
selected_children = selected.intersection(choice.children)
if len(selected_children) == len(choice.children):
return "x"
return "-" if selected_children else " "
def draw(stdscr: object) -> set[str] | None: def draw(stdscr: object) -> set[str] | None:
curses.curs_set(0) curses.curs_set(0)
stdscr.keypad(True) stdscr.keypad(True)
current = 0 current = 0
query = "" query = ""
source_attr = curses.A_NORMAL
status_attr = curses.A_NORMAL
description_attr = curses.A_DIM
try:
if curses.has_colors():
curses.start_color()
curses.use_default_colors()
curses.init_pair(1, curses.COLOR_CYAN, -1)
curses.init_pair(2, curses.COLOR_GREEN, -1)
source_attr = curses.color_pair(1)
status_attr = curses.color_pair(2)
except curses.error:
pass
while True: while True:
visible = filter_choices(choices, query) visible = filter_choices(choices, query)
current = min(current, max(0, len(visible) - 1)) current = min(current, max(0, len(visible) - 1))
stdscr.erase() stdscr.erase()
height, width = stdscr.getmaxyx() height, width = stdscr.getmaxyx()
header = "↑/↓ 移动 Space 勾选 / 搜索 Enter 安装 q 取消" text_width = max(0, width - 1)
stdscr.addnstr(0, 0, header, max(0, width - 1)) stdscr.addnstr(
0,
0,
fit_to_width(f"安装目标: {scope_label}", text_width),
text_width,
)
header = "↑↓ 移动 Space 选择 / 搜索 Enter 安装 q 退出"
stdscr.addnstr(1, 0, fit_to_width(header, text_width), text_width)
if query: if query:
stdscr.addnstr(1, 0, f"搜索: {query}", max(0, width - 1)) stdscr.addnstr(
2,
0,
fit_to_width(f"搜索: {query}", text_width),
text_width,
)
rows = max(1, height - 4) item_capacity = max(1, (height - 4) // 3)
start = max(0, current - rows + 1) start = max(0, current - item_capacity + 1)
for row, choice in enumerate(visible[start : start + rows], start=2): for offset, choice in enumerate(visible[start : start + item_capacity]):
index = start + row - 2 index = start + offset
mark = "x" if choice.name in selected else " " row = 3 + offset * 3
suffix = f" · {choice.description}" if choice.description else "" mark = mark_for(choice)
line = f"[{mark}] {choice.name} {choice.kind}{suffix}" description = f" {choice.description}" if choice.description else ""
attr = curses.A_REVERSE if index == current else curses.A_NORMAL focus_attr = curses.A_REVERSE if index == current else curses.A_NORMAL
stdscr.addnstr(row, 0, line, max(0, width - 1), attr) parts = [
(f"{' ' * choice.indent}[{mark}] ", focus_attr),
(choice.name, focus_attr | curses.A_BOLD),
(f" {choice.kind}", focus_attr | source_attr),
]
if choice.readonly_status:
parts.append(
(f" {choice.readonly_status}", focus_attr | status_attr)
)
column = 0
for text, attr in parts:
available = text_width - column
if available <= 0:
break
fitted = fit_to_width(text, available)
stdscr.addnstr(row, column, fitted, available, attr)
column += display_width(fitted)
stdscr.addnstr(
row + 1,
0,
fit_to_width(description, text_width),
text_width,
focus_attr | description_attr,
)
footer = f"已选择 {len(selected)} 项" footer = (
stdscr.addnstr(height - 1, 0, footer, max(0, width - 1)) f"已选择 {len(selected)} 项 · 取消选择不卸载,卸载用 skiff remove"
)
stdscr.addnstr(
height - 1,
0,
fit_to_width(footer, text_width),
text_width,
)
stdscr.refresh() stdscr.refresh()
key = stdscr.get_wch() key = stdscr.get_wch()
@@ -83,11 +185,18 @@ def select_skills(
elif key == curses.KEY_DOWN and visible: elif key == curses.KEY_DOWN and visible:
current = (current + 1) % len(visible) current = (current + 1) % len(visible)
elif key == " " and visible: elif key == " " and visible:
name = visible[current].name choice = visible[current]
selected.symmetric_difference_update({name}) if choice.children:
children = set(choice.children)
if children.issubset(selected):
selected.difference_update(children)
else:
selected.update(children)
else:
selected.symmetric_difference_update({choice.name})
elif key == "/": elif key == "/":
curses.curs_set(1) curses.curs_set(1)
query = _read_query(stdscr, curses, width) query = _read_query(stdscr, curses, width, row=2)
curses.curs_set(0) curses.curs_set(0)
current = 0 current = 0
@@ -95,12 +204,18 @@ def select_skills(
return runner(draw) return runner(draw)
def _read_query(stdscr: object, curses: object, width: int) -> str: def _read_query(stdscr: object, curses: object, width: int, *, row: int = 1) -> str:
query = "" query = ""
while True: while True:
stdscr.move(1, 0) stdscr.move(row, 0)
stdscr.clrtoeol() stdscr.clrtoeol()
stdscr.addnstr(1, 0, f"搜索: {query}", max(0, width - 1)) text_width = max(0, width - 1)
stdscr.addnstr(
row,
0,
fit_to_width(f"搜索: {query}", text_width),
text_width,
)
stdscr.refresh() stdscr.refresh()
key = stdscr.get_wch() key = stdscr.get_wch()
if key in ("\n", "\r") or key == curses.KEY_ENTER: if key in ("\n", "\r") or key == curses.KEY_ENTER:
+11 -1
View File
@@ -100,7 +100,17 @@ def fetch_source(name: str, entry: dict[str, Any]) -> Path:
else: else:
checkout.parent.mkdir(parents=True, exist_ok=True) checkout.parent.mkdir(parents=True, exist_ok=True)
subprocess.run( subprocess.run(
["git", "clone", "--branch", ref, "--", str(repo), str(checkout)], [
"git",
"clone",
"--depth",
"1",
"--branch",
ref,
"--",
str(repo),
str(checkout),
],
check=True, check=True,
) )
return checkout return checkout
+93
View File
@@ -0,0 +1,93 @@
---
name: ack
description: >-
初始化、检查并运行 ACK(Agent Collaboration Kit)三角色协作闭环。仅在用户显式调用
/ack 或 $ack,并要求初始化 ACK、检查 docs/ack 配置、按 ACK 规划需求或指挥
Coordinator/Developer/Test 工作时使用。
---
# ACK 项目协作入口
把全局 skill 作为入口,把项目状态留在 `docs/ack/`,把通用规范留在
`~/.skills/kits/ack/`。不要在本 skill 复制或改写 kit 的核心规范。
## 选择模式
- 用户要求初始化、接入或安装 ACK:执行“初始化”。
- 用户要求检查 ACK 是否可用、配置是否完整:执行“检查”。
- 用户要求用 ACK 做需求、修复问题或继续任务:执行“工作”。
始终先解析真实项目根目录。优先使用 `git rev-parse --show-toplevel`;不是 Git
项目时使用用户指定目录或当前目录。不要修改项目的 `AGENTS.md`、`CLAUDE.md`
或其它 Agent 指令文件。
## 初始化
1. 确认 `skiff` 可执行,并检查 `<project>/docs/ack` 是否存在。
2. 不存在时执行:
```bash
skiff kit init ack --project <project-root>
```
默认使用软链接模式。只有用户明确要求项目自带完整副本时才加 `--copy`。
3. 如果 `docs/ack` 已存在,不重复初始化、不覆盖文件;转入“检查”,报告缺失项并
只补用户授权且能安全确定的内容。
4. 读取项目的公开配置和文档,例如 README、语言清单、包管理清单、测试配置与
CI,确定项目名、技术栈、源码/规格/测试路径及真实可执行命令。
5. 完善 `docs/ack/project.md`:
- 用实际项目值替换全部占位符。
- 无服务地址时把 Base URL 写为 `n/a`,不要虚构端口。
- 无法从项目证据确定的命令写为 `n/a`,并在结果中列为待配置项。
- 保留 `docs/ack/kit/core/` 引用,不复制 core 内容。
6. 完善 `docs/ack/tasks.yaml` 的项目信息。纯初始化且用户没有提供真实任务时,
删除模板示例任务并保留 `tasks: []`;不要虚构需求或缺陷。
7. 更新 `updatedAt`,并运行:
```bash
python3 docs/ack/kit/scripts/validate_tasks.py docs/ack/tasks.yaml
```
8. 检查 `project.md` 与 `tasks.yaml` 是否仍有 `<...>` 占位符。结构校验通过且
必填项目事实完整时才称“初始化完成”;否则称“部分完成”并列出缺失值。
9. 报告创建的路径、软链接或复制模式、检测到的命令、校验结果和下一步。除非用户
明确要求,不提交、不推送。
## 检查
1. 检查以下路径:
- `docs/ack/kit`
- `docs/ack/project.md`
- `docs/ack/tasks.yaml`
2. 确认 `kit` 可访问 `VERSION`、`core/`、`templates/` 和 `scripts/`。软链接模式下
同时报告其真实目标。
3. 对比 `tasks.yaml` 的 `kitVersion` 与 `kit/VERSION`。
4. 查找未替换占位符,并核对项目路径、覆盖层路径、Developer 白盒命令、Test
黑盒命令和 Base URL。
5. 运行任务板校验器。只报告证据明确的问题,不因可选字段缺失而宣称失败。
6. 检查不会自动修复或覆盖现有配置;用户明确要求修复后再修改。
## 工作
1. 若 `docs/ack` 不存在,停止并建议先用 `/ack` 初始化;不要静默初始化。
2. 依次读取:
- `docs/ack/project.md`
- `docs/ack/tasks.yaml`
- `docs/ack/kit/core/kickoff.md`
- kickoff 指定且与当前任务相关的 core 文件
3. 当前会话担任 Coordinator,遵守项目覆盖层中的命令、路径权限、模型路由和
worker 复用规则。项目覆盖层优先于通用示例命令。
4. 新需求先写产品文档、任务拆分与可观测验收信号,更新 `tasks.yaml` 并校验,
然后交给用户确认;确认前不派发实现。
5. 用户已确认的任务按 ACK 闭环执行:Developer 实现与白盒验证,Test 独立黑盒
复测,Coordinator 读取证据终检并唯一写入 `tasks.yaml`。
6. 不把 `worker_done` 或 Test 自报成功直接当作完成。每项最多三轮,仍失败则记录
`leftover` 并继续其它任务。
## 边界
- 不修改或追加任何项目 Agent 指令文件,包括 `AGENTS.md`。
- 不在 skill 中维护第二份 ACK core、模板或任务 schema。
- 不猜测项目命令、服务地址、worker handle 或模型名称。
- 不覆盖已有 `docs/ack` 文件,不擅自提交、推送、创建终端或新 worktree。
- kit 初始化的机械操作始终委托给 `skiff kit init ack`。
+6
View File
@@ -0,0 +1,6 @@
interface:
display_name: "ACK"
short_description: "初始化、检查并运行项目里的 ACK 三角色协作闭环"
default_prompt: "Use $ack to initialize ACK for this project or coordinate work from docs/ack."
policy:
allow_implicit_invocation: false
+21 -3
View File
@@ -2,8 +2,9 @@
name: skiff name: skiff
description: >- description: >-
创建和维护 ~/.skills 自研 skill:把项目开发中产生的想法提炼为草稿,完善并校验后发布, 创建和维护 ~/.skills 自研 skill:把项目开发中产生的想法提炼为草稿,完善并校验后发布,
或用 skiff add/remove 在项目及全局挂卸 skill。触发词:skiff、自研 skill、创建 skill、 用 skiff add/remove 在项目及全局挂卸 skill,或用 skiff kit init 初始化项目规范包。
想做一个 skill、publish skill、安装自研 skill、更新 skill 到项目。 触发词:skiff、自研 skill、创建 skill、想做一个 skill、publish skill、安装自研 skill、
更新 skill 到项目、初始化 kit。
--- ---
# skiff 自研 Skill 工作流 # skiff 自研 Skill 工作流
@@ -114,7 +115,11 @@ skiff add waza -a codex -g -y
skiff add waza/think -a codex -g -y skiff add waza/think -a codex -g -y
``` ```
`skiff select` 会把 collection 展开为 `waza/think`、`waza/ui` 等候选项。 `skiff select` 会把 collection 显示为两级菜单:选择 `waza` 仓库会选中其
全部子 skill,也可以只选择 `waza/think`、`waza/ui` 中的若干项。
普通 `skiff select` 只向项目安装,并只读显示每个 Agent 的全局安装状态;
`skiff select -g` 只向全局安装。取消已勾选项不会卸载,卸载继续使用
`skiff remove`。
卸载: 卸载:
@@ -130,6 +135,17 @@ skiff rm discussion-notes -g -y # rm 别名
skiff add --list skiff add --list
``` ```
初始化项目 kit:
```bash
skiff kit init ack # 当前项目,默认软链接到 SSOT
skiff kit init ack --project ~/app # 指定项目
skiff kit init ack --copy # 用户明确要求时整份复制
```
`skiff` 只负责可靠地创建 kit 文件。需要分析项目并完善 ACK 配置、检查接入状态或
运行三角色闭环时,显式调用全局 `/ack` skill。
--- ---
## 与 Vercel `npx skills` 的分工 ## 与 Vercel `npx skills` 的分工
@@ -146,6 +162,7 @@ skiff add --list
| skiff | 说明 | | skiff | 说明 |
|-------|------| |-------|------|
| `bootstrap` | 将本项目的 `skiff` skill 全局安装到所有 Agent | | `bootstrap` | 将本项目的 `skiff` skill 全局安装到所有 Agent |
| `update` | 在 `~/.skills` 执行 `git pull`,更新 skiff 自身 |
| `add <name> [-g] [-a AGENT...] [-y]` | 安装 | | `add <name> [-g] [-a AGENT...] [-y]` | 安装 |
| `remove <name> [-g] [-a AGENT...] [-y]` | 卸载(`rm` / `r` 别名) | | `remove <name> [-g] [-a AGENT...] [-y]` | 卸载(`rm` / `r` 别名) |
| `add --list` | 列出可用自研 skill | | `add --list` | 列出可用自研 skill |
@@ -155,6 +172,7 @@ skiff add --list
| `create <name> --idea TEXT [--from-project PATH]` | 创建自研 skill 草稿 | | `create <name> --idea TEXT [--from-project PATH]` | 创建自研 skill 草稿 |
| `check <name>` | 校验草稿或正式 skill | | `check <name>` | 校验草稿或正式 skill |
| `finalize <name>` | 校验草稿并转为正式 skill | | `finalize <name>` | 校验草稿并转为正式 skill |
| `kit init <name> [--project DIR] [--copy]` | 在项目中初始化 kit |
--- ---
+35
View File
@@ -0,0 +1,35 @@
from __future__ import annotations
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
class AckSkillContentTests(unittest.TestCase):
def test_ack_skill_routes_to_kit_without_modifying_agent_instructions(self) -> None:
content = (REPO_ROOT / "skills" / "ack" / "SKILL.md").read_text(encoding="utf-8")
for expected in (
"skiff kit init ack --project <project-root>",
"docs/ack/project.md",
"docs/ack/tasks.yaml",
"tasks: []",
"validate_tasks.py",
"不要修改项目的 `AGENTS.md`",
"当前会话担任 Coordinator",
):
self.assertIn(expected, content)
def test_ack_skill_is_explicit_only(self) -> None:
metadata = (REPO_ROOT / "skills" / "ack" / "agents" / "openai.yaml").read_text(
encoding="utf-8"
)
self.assertIn('display_name: "ACK"', metadata)
self.assertIn("allow_implicit_invocation: false", metadata)
if __name__ == "__main__":
unittest.main()
+17
View File
@@ -1,11 +1,15 @@
from __future__ import annotations from __future__ import annotations
import argparse
import os import os
import subprocess import subprocess
import sys import sys
import tempfile import tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import patch
from skiff import cli
REPO_ROOT = Path(__file__).resolve().parents[1] REPO_ROOT = Path(__file__).resolve().parents[1]
@@ -76,6 +80,19 @@ class CreateWorkflowTests(unittest.TestCase):
self.assertIn("claude", first.stdout) self.assertIn("claude", first.stdout)
self.assertIn("codex", first.stdout) self.assertIn("codex", first.stdout)
def test_update_pulls_skills_home(self) -> None:
with (
patch.object(cli, "SKILLS_HOME", self.skills_home),
patch.object(cli, "ensure_skills_home"),
patch.object(cli.subprocess, "run") as run,
):
cli.cmd_update(argparse.Namespace())
run.assert_called_once_with(
["git", "-C", str(self.skills_home), "pull"],
check=True,
)
def test_create_writes_draft_and_brief_without_registering_owned_skill(self) -> None: def test_create_writes_draft_and_brief_without_registering_owned_skill(self) -> None:
project = self.home / "project" project = self.home / "project"
project.mkdir() project.mkdir()
+28
View File
@@ -6,7 +6,9 @@ import sys
import tempfile import tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import patch
from skiff.sources import fetch_source
REPO_ROOT = Path(__file__).resolve().parents[1] REPO_ROOT = Path(__file__).resolve().parents[1]
@@ -228,6 +230,32 @@ class CustomSourceTests(unittest.TestCase):
self.assertTrue((checkout / ".git").is_dir()) self.assertTrue((checkout / ".git").is_dir())
self.assertTrue((checkout / "skills" / "release-check" / "SKILL.md").is_file()) self.assertTrue((checkout / "skills" / "release-check" / "SKILL.md").is_file())
def test_git_source_clone_is_shallow(self) -> None:
checkout = self.home / "checkout"
entry = {
"repo": "https://example.test/company-skills.git",
"ref": "main",
"checkout": str(checkout),
}
with patch("skiff.sources.subprocess.run") as run:
fetch_source("company", entry)
run.assert_called_once_with(
[
"git",
"clone",
"--depth",
"1",
"--branch",
"main",
"--",
"https://example.test/company-skills.git",
str(checkout),
],
check=True,
)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+107
View File
@@ -0,0 +1,107 @@
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
class KitInitTests(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.home = Path(self.temp_dir.name)
self.skills_home = self.home / ".skills"
kit = self.skills_home / "kits" / "ack"
(kit / "templates").mkdir(parents=True)
(kit / "scripts").mkdir()
(kit / "VERSION").write_text("1.2.3\n", encoding="utf-8")
(kit / "templates" / "project.template.md").write_text(
"# <project_name>\nversion=<kit_version>\npath=<overlay_file_path>\n",
encoding="utf-8",
)
(kit / "templates" / "tasks.template.yaml").write_text(
'updatedAt: "<YYYY-MM-DDTHH:mm:ss+TZ>"\n'
'kitVersion: "<接入时的 ack 版本,见 kit 根 VERSION>"\n'
'project:\n'
' name: "<project_name>"\n'
' repoPath: "<repo_path>"\n'
' devWorktree: "<dev_worktree>"\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_init_creates_symlink_and_rendered_project_files(self) -> None:
project = self.home / "sample-app"
project.mkdir()
result = self.run_skiff("kit", "init", "ack", "--project", str(project))
self.assertEqual(result.returncode, 0, result.stderr)
target = project / "docs" / "ack"
self.assertTrue((target / "kit").is_symlink())
self.assertEqual((target / "kit").resolve(), self.skills_home / "kits" / "ack")
project_content = (target / "project.md").read_text(encoding="utf-8")
tasks_content = (target / "tasks.yaml").read_text(encoding="utf-8")
self.assertIn("# sample-app", project_content)
self.assertIn("version=1.2.3", project_content)
self.assertIn(f'repoPath: "{project}"', tasks_content)
self.assertNotIn("<project_name>", tasks_content)
def test_init_copy_mode_copies_kit(self) -> None:
project = self.home / "copied-app"
project.mkdir()
result = self.run_skiff("kit", "init", "ack", "--copy", "--project", str(project))
self.assertEqual(result.returncode, 0, result.stderr)
target = project / "docs" / "ack" / "kit"
self.assertTrue(target.is_dir())
self.assertFalse(target.is_symlink())
self.assertEqual((target / "VERSION").read_text(encoding="utf-8"), "1.2.3\n")
def test_init_refuses_to_overwrite_existing_files(self) -> None:
project = self.home / "existing-app"
target = project / "docs" / "ack"
target.mkdir(parents=True)
existing = target / "project.md"
existing.write_text("keep me", encoding="utf-8")
result = self.run_skiff("kit", "init", "ack", "--project", str(project))
self.assertNotEqual(result.returncode, 0)
self.assertIn("拒绝覆盖已有路径", result.stderr)
self.assertEqual(existing.read_text(encoding="utf-8"), "keep me")
self.assertFalse((target / "kit").exists())
def test_init_rejects_missing_project_directory(self) -> None:
project = self.home / "missing-app"
result = self.run_skiff("kit", "init", "ack", "--project", str(project))
self.assertNotEqual(result.returncode, 0)
self.assertIn("项目目录不存在", result.stderr)
self.assertFalse(project.exists())
if __name__ == "__main__":
unittest.main()
+2
View File
@@ -19,6 +19,8 @@ class ProjectSkillContentTests(unittest.TestCase):
"skiff check <name>", "skiff check <name>",
"~/.skills/skills/<name>/", "~/.skills/skills/<name>/",
"第三方 skill", "第三方 skill",
"skiff kit init ack",
"显式调用全局 `/ack` skill",
): ):
self.assertIn(expected, content) self.assertIn(expected, content)
+14
View File
@@ -7,6 +7,9 @@ import sys
import tempfile import tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import patch
from skiff.registry import external_checkout_path, registry_repo
REPO_ROOT = Path(__file__).resolve().parents[1] REPO_ROOT = Path(__file__).resolve().parents[1]
@@ -79,6 +82,17 @@ class RegistryCollectionTests(unittest.TestCase):
check=False, check=False,
) )
def test_registry_repo_expands_home_relative_local_path(self) -> None:
with patch.dict(os.environ, {"HOME": str(self.home)}):
self.assertEqual(
registry_repo({"repo": "~/.skills"}),
str(self.skills_home),
)
self.assertEqual(
external_checkout_path("skills", {"repo": "~/.skills"}),
self.skills_home.resolve(),
)
def test_add_collection_installs_every_discovered_skill(self) -> None: def test_add_collection_installs_every_discovered_skill(self) -> None:
result = self.run_skiff("add", "test-pack", "-g", "-a", "codex") result = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
+263 -5
View File
@@ -13,13 +13,17 @@ from unittest.mock import Mock, patch
from skiff import cli from skiff import cli
from skiff import yaml_io from skiff import yaml_io
from skiff.registry import external_repo_path, external_skill_path from skiff.registry import external_repo_path, external_skill_path
from skiff.selector import SkillChoice, filter_choices, select_skills from skiff.selector import SkillChoice, filter_choices, fit_to_width, select_skills
REPO_ROOT = Path(__file__).resolve().parents[1] REPO_ROOT = Path(__file__).resolve().parents[1]
class SelectorTests(unittest.TestCase): class SelectorTests(unittest.TestCase):
def test_fit_to_width_counts_cjk_as_two_columns(self) -> None:
self.assertEqual(fit_to_width("a中文b", 4), "a中")
self.assertEqual(fit_to_width("初始化 ACK", 8), "初始化 A")
def test_manifest_entry_targets_round_trip(self) -> None: def test_manifest_entry_targets_round_trip(self) -> None:
data = { data = {
"skills": [ "skills": [
@@ -36,12 +40,18 @@ class SelectorTests(unittest.TestCase):
def test_filter_matches_name_kind_and_description(self) -> None: def test_filter_matches_name_kind_and_description(self) -> None:
choices = [ choices = [
SkillChoice("frontend-design", "external", "创建界面"), SkillChoice("frontend-design", "external", "创建界面"),
SkillChoice("discussion-notes", "owned", "维护讨论笔记"), SkillChoice(
"discussion-notes",
"owned",
"维护讨论笔记",
readonly_status="全局: codex",
),
] ]
self.assertEqual([c.name for c in filter_choices(choices, "front")], ["frontend-design"]) 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, "owned")], ["discussion-notes"])
self.assertEqual([c.name for c in filter_choices(choices, "界面")], ["frontend-design"]) self.assertEqual([c.name for c in filter_choices(choices, "界面")], ["frontend-design"])
self.assertEqual([c.name for c in filter_choices(choices, "codex")], ["discussion-notes"])
def test_selector_preserves_preselected_items(self) -> None: def test_selector_preserves_preselected_items(self) -> None:
choices = [ choices = [
@@ -59,6 +69,114 @@ class SelectorTests(unittest.TestCase):
) )
self.assertEqual(selected, {"already-there"}) self.assertEqual(selected, {"already-there"})
rendered = [call.args[2] for call in screen.addnstr.call_args_list]
self.assertTrue(any("取消选择不卸载" in line for line in rendered))
def test_selector_renders_scope_and_readonly_global_status(self) -> None:
choices = [
SkillChoice(
"ack",
"owned",
installed=False,
readonly_status="全局: cursor,claude,codex",
)
]
screen = Mock()
screen.getmaxyx.return_value = (24, 120)
screen.get_wch.return_value = "\n"
with patch.object(curses, "curs_set"):
selected = select_skills(
choices,
scope_label="项目 /code/app(全局状态只读)",
wrapper=lambda draw: draw(screen),
)
rendered = [call.args[2] for call in screen.addnstr.call_args_list]
self.assertEqual(selected, set())
self.assertTrue(any("安装目标: 项目 /code/app" in line for line in rendered))
title = "".join(
call.args[2]
for call in screen.addnstr.call_args_list
if call.args[0] == 3
)
self.assertIn("[ ] ack", title)
self.assertIn("全局: cursor,claude,codex", title)
self.assertNotIn("初始化", title)
def test_selector_renders_description_on_indented_second_line(self) -> None:
choices = [
SkillChoice(
"ack",
"owned",
description="初始化、检查并运行 ACK",
readonly_status="全局: codex",
)
]
screen = Mock()
screen.getmaxyx.return_value = (24, 100)
screen.get_wch.return_value = "\n"
with patch.object(curses, "curs_set"):
select_skills(choices, wrapper=lambda draw: draw(screen))
title_calls = [
call for call in screen.addnstr.call_args_list if call.args[0] == 3
]
description_call = next(
call for call in screen.addnstr.call_args_list if call.args[0] == 4
)
self.assertEqual(
"".join(call.args[2] for call in title_calls),
"[ ] ack owned 全局: codex",
)
self.assertEqual(description_call.args[2], " 初始化、检查并运行 ACK")
name_call = next(call for call in title_calls if call.args[2] == "ack")
self.assertTrue(name_call.args[4] & curses.A_BOLD)
self.assertTrue(description_call.args[4] & curses.A_DIM)
def test_repository_choice_toggles_all_child_skills(self) -> None:
choices = [
SkillChoice(
"waza",
"repository",
children=("waza/think", "waza/ui"),
),
SkillChoice("waza/think", "external:waza", indent=1),
SkillChoice("waza/ui", "external:waza", indent=1),
]
screen = Mock()
screen.getmaxyx.return_value = (24, 100)
screen.get_wch.side_effect = [" ", "\n"]
with patch.object(curses, "curs_set"):
selected = select_skills(choices, wrapper=lambda draw: draw(screen))
self.assertEqual(selected, {"waza/think", "waza/ui"})
def test_repository_choice_renders_partial_selection(self) -> None:
choices = [
SkillChoice(
"waza",
"repository",
children=("waza/think", "waza/ui"),
),
SkillChoice("waza/think", "external:waza", installed=True, indent=1),
SkillChoice("waza/ui", "external:waza", indent=1),
]
screen = Mock()
screen.getmaxyx.return_value = (24, 100)
screen.get_wch.return_value = "\n"
with patch.object(curses, "curs_set"):
select_skills(choices, wrapper=lambda draw: draw(screen))
first_row = "".join(
call.args[2]
for call in screen.addnstr.call_args_list
if call.args[0] == 3
)
self.assertIn("[-] waza", first_row)
def test_shared_repo_path_is_same_for_different_skill_entries(self) -> None: def test_shared_repo_path_is_same_for_different_skill_entries(self) -> None:
first = {"repo": "https://example.test/skills.git", "ref": "main", "path": "a"} first = {"repo": "https://example.test/skills.git", "ref": "main", "path": "a"}
@@ -97,6 +215,43 @@ class SelectorTests(unittest.TestCase):
self.assertEqual(discover_external_skills("unsafe", entry), {}) self.assertEqual(discover_external_skills("unsafe", entry), {})
def test_registry_clone_is_shallow(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
checkout = root / "checkout"
skill_root = checkout / "skills"
entry = {
"repo": "https://example.test/skills.git",
"ref": "main",
"path": "skills",
}
with (
patch.object(cli, "external_skill_path", return_value=skill_root),
patch.object(cli, "external_checkout_path", return_value=checkout),
patch.object(
cli,
"discover_external_skills",
side_effect=[{}, {"demo": skill_root / "demo"}],
),
patch.object(cli.subprocess, "run") as run,
):
cli._ensure_registry_fetched("demo-pack", entry)
run.assert_called_once_with(
[
"git",
"clone",
"--depth",
"1",
"--branch",
"main",
"--",
"https://example.test/skills.git",
str(checkout),
],
check=True,
)
class SelectCommandTests(unittest.TestCase): class SelectCommandTests(unittest.TestCase):
def test_non_tty_exits_with_add_guidance(self) -> None: def test_non_tty_exits_with_add_guidance(self) -> None:
@@ -136,6 +291,17 @@ class SelectCommandTests(unittest.TestCase):
stdin.isatty.return_value = True stdin.isatty.return_value = True
stdout.isatty.return_value = True stdout.isatty.return_value = True
installed: list[str] = [] installed: list[str] = []
selected_choices: list[SkillChoice] = []
selected_scope: list[str] = []
def choose(
choices: list[SkillChoice],
*,
scope_label: str,
) -> set[str]:
selected_choices.extend(choices)
selected_scope.append(scope_label)
return {"owned-one", "external-one"}
with ( with (
patch.object(cli.sys, "stdin", stdin), patch.object(cli.sys, "stdin", stdin),
@@ -157,6 +323,11 @@ class SelectCommandTests(unittest.TestCase):
patch.object(cli, "_registry_skill_names", return_value=["external-one"]), patch.object(cli, "_registry_skill_names", return_value=["external-one"]),
patch.object(cli, "external_skill_path", return_value=external), patch.object(cli, "external_skill_path", return_value=external),
patch.object(cli, "_list_fully_installed_names", return_value=["owned-one"]), patch.object(cli, "_list_fully_installed_names", return_value=["owned-one"]),
patch.object(
cli,
"_global_installation_note",
return_value="全局: codex",
),
patch.object( patch.object(
cli, cli,
"_is_fully_installed", "_is_fully_installed",
@@ -165,7 +336,7 @@ class SelectCommandTests(unittest.TestCase):
patch.object( patch.object(
cli, cli,
"select_skills", "select_skills",
return_value={"owned-one", "external-one"}, side_effect=choose,
), ),
patch.object( patch.object(
cli, cli,
@@ -183,6 +354,40 @@ class SelectCommandTests(unittest.TestCase):
self.assertIn("source: registry", manifest) self.assertIn("source: registry", manifest)
self.assertIn("targets:", manifest) self.assertIn("targets:", manifest)
self.assertIn("codex", manifest) self.assertIn("codex", manifest)
self.assertEqual(
[choice.readonly_status for choice in selected_choices],
["全局: codex", "全局: codex"],
)
self.assertIn("全局状态只读", selected_scope[0])
def test_global_installation_note_reports_agents_and_source_conflicts(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
expected = root / "expected"
other = root / "other"
expected.mkdir()
other.mkdir()
global_dirs = {
target: root / target / "skills"
for target in ("cursor", "claude", "codex")
}
for directory in global_dirs.values():
directory.mkdir(parents=True)
(global_dirs["cursor"] / "ack").symlink_to(expected, target_is_directory=True)
(global_dirs["claude"] / "ack").symlink_to(other, target_is_directory=True)
with patch.object(
cli,
"agent_skill_dir",
side_effect=lambda target, project_root=None: global_dirs[target],
):
note = cli._global_installation_note(
"ack",
expected,
["cursor", "claude", "codex"],
)
self.assertEqual(note, "全局: cursor;全局同名冲突: claude")
def test_select_rejects_invalid_registry_name_before_rendering(self) -> None: def test_select_rejects_invalid_registry_name_before_rendering(self) -> None:
args = argparse.Namespace( args = argparse.Namespace(
@@ -227,7 +432,7 @@ class SelectCommandTests(unittest.TestCase):
selected_choices: list[SkillChoice] = [] selected_choices: list[SkillChoice] = []
installed: list[tuple[str, str | None]] = [] installed: list[tuple[str, str | None]] = []
def choose(choices: list[SkillChoice]) -> set[str]: def choose(choices: list[SkillChoice], **_: object) -> set[str]:
selected_choices.extend(choices) selected_choices.extend(choices)
return {"waza/think"} return {"waza/think"}
@@ -265,9 +470,62 @@ class SelectCommandTests(unittest.TestCase):
): ):
cli.cmd_select(args) cli.cmd_select(args)
self.assertEqual([choice.name for choice in selected_choices], ["waza/think", "waza/ui"]) self.assertEqual(
[choice.name for choice in selected_choices],
["waza", "waza/think", "waza/ui"],
)
self.assertEqual(selected_choices[0].children, ("waza/think", "waza/ui"))
self.assertEqual([choice.indent for choice in selected_choices[1:]], [1, 1])
self.assertEqual(installed, [("think", "registry:waza")]) self.assertEqual(installed, [("think", "registry:waza")])
def test_select_deduplicates_local_registry_collection_from_owned(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
with tempfile.TemporaryDirectory() as temp:
skills_root = Path(temp) / "skills"
(skills_root / "ack").mkdir(parents=True)
selected_choices: list[SkillChoice] = []
def choose(choices: list[SkillChoice], **_: object) -> set[str]:
selected_choices.extend(choices)
return set()
with (
patch.object(cli.sys, "stdin", stdin),
patch.object(cli.sys, "stdout", stdout),
patch.object(cli, "SKILLS_DIR", skills_root),
patch.object(cli, "ensure_skills_home"),
patch.object(cli, "list_owned_skills", return_value=["ack"]),
patch.object(cli, "skill_description", return_value="owned ack"),
patch.object(
cli,
"load_registry",
return_value={
"skills": {
"repo": "~/.skills",
"ref": "main",
"path": "skills",
}
},
),
patch.object(cli, "_registry_skill_names", return_value=["ack"]),
patch.object(cli, "external_skill_path", return_value=skills_root),
patch.object(cli, "_is_fully_installed", return_value=False),
patch.object(cli, "select_skills", side_effect=choose),
):
cli.cmd_select(args)
self.assertEqual([choice.name for choice in selected_choices], ["ack"])
def test_install_rolls_back_earlier_target_when_later_target_fails(self) -> None: def test_install_rolls_back_earlier_target_when_later_target_fails(self) -> None:
with tempfile.TemporaryDirectory() as temp: with tempfile.TemporaryDirectory() as temp:
root = Path(temp) root = Path(temp)