feat: auto-update pouch and catalog sources during normal use

Add taste-skill to catalog.yaml, and refresh pouch plus cloned catalog
or custom Git sources on a 24h interval so users do not need a
separate update command.
This commit is contained in:
2026-10-13 16:41:55 +08:00
parent d3b6732d40
commit 8390706df8
14 changed files with 836 additions and 40 deletions
+4 -1
View File
@@ -90,11 +90,13 @@ builder 与 deployer 的项目接入走各自 skill 的「初始化」模式(
| Source | 来源 |
| --- | --- |
| waza | [https://github.com/tw93/Waza](https://github.com/tw93/Waza) |
| taste-skill | [https://github.com/Leonxlnx/taste-skill](https://github.com/Leonxlnx/taste-skill) |
```bash
pouch fetch waza
pouch add waza/think -g
pouch add taste-skill/taste-skill -g
```
### 自定义仓库(Custom Sources)
@@ -237,6 +239,7 @@ pouch sync
| `pouch add <name> [-g]` | 项目或全局安装(symlink) |
| `pouch remove <name> [-g]` | 移除 symlink |
| `pouch catalog add` / `pouch fetch` | 管理和拉取 catalog source |
| `pouch update` | 更新 pouch 自身和已拉取的 catalog / custom source |
### 草稿与健康检查
@@ -346,7 +349,7 @@ Claude Code 对 symlink 支持不稳定:可能无法发现 skill,或写入
| 项目启用 | `pouch add <name>` 或 `pouch enable <name>` | 项目目录 |
| 看状态 | `pouch status` | 任意 |
| 装社区 skill | `npx skills add owner/repo -g -y` | 任意 |
| 更新外部 skill | `pouch fetch <name>` | 任意 |
| 更新外部 skill | `pouch fetch <name>` 或日常命令自动更新 | 任意 |
---
+2
View File
@@ -166,6 +166,8 @@ pouch remove ack
```bash
pouch fetch waza
pouch add waza/think -g
pouch add taste-skill/taste-skill -g # 只要核心 frontend skill
pouch add taste-skill -g # 或装整个 Taste Skill collection
```
接入团队自己的本地目录或 Git 仓库:
+10
View File
@@ -29,3 +29,13 @@ waza:
- engineering
- workflow
- skill-collection
taste-skill:
repo: https://github.com/Leonxlnx/taste-skill.git
ref: main
path: skills
description: Taste Skill collection for anti-slop frontend, redesign, image generation, brand kits, and output enforcement.
tags:
- frontend
- design
- skill-collection
+1 -1
View File
@@ -49,7 +49,7 @@ pouch status
pouch doctor
```
软链异常就 `pouch doctor --fix`。更新 pouch 自身:`pouch update`(在 `~/.pouch` 里 `git pull`)。
软链异常就 `pouch doctor --fix`。日常用 pouch 命令时会按间隔自动更新自身和已拉取的外部 source;要立刻更新就 `pouch update`。
## 把要用的 skill 装上
+32 -2
View File
@@ -65,7 +65,7 @@ pouch bootstrap
| 命令 | 说明 |
|------|------|
| `pouch bootstrap` | 将本项目的 `pouch` skill 全局安装到所有 Agent |
| `pouch update` | 在 `~/.pouch` 执行 `git pull`,更新 pouch 自身 |
| `pouch update` | 更新 pouch 自身,以及已拉取的 catalog / custom source |
| `pouch init <name> [--project DIR]` | 使用 builtin skill 自带模板初始化项目状态(目前用于 ack) |
### Skill 安装
@@ -105,6 +105,33 @@ collection 会自动发现下一层所有含 `SKILL.md` 的目录;`pouch add <
安装全部,`pouch select` 则展开为 `<name>/<skill>` 供分别勾选。同一
`repo` 与 `ref` 共享一份 Git checkout。
预置 collection 示例:
```bash
pouch add waza/think -g
pouch add taste-skill/taste-skill -g
pouch add taste-skill -g # 安装 Taste Skill 全部子 skill
```
### 自动更新
日常执行 `pouch list`、`pouch add`、`pouch status` 等命令时,pouch 会按间隔
顺带更新自身和已经 clone 过的 catalog / custom Git source,不必单独跑
`pouch fetch` 或 `pouch update`。默认每 24 小时最多自动更新一次;网络失败或
工作区不干净时跳过,不打断当前命令。`pouch add <catalog>` 在安装前会再刷新
该来源,以便拿到最新子 skill。
在 `~/.config/pouch/config.yaml` 里关闭或改间隔:
```yaml
auto_update:
enabled: true
interval: 24h # 也可用 1d、60m、3600
```
本次跳过:`pouch --no-auto-update list`,或设置环境变量 `POUCH_NO_AUTO_UPDATE=1`。
立刻更新全部:`pouch update`。
### 交互式批量安装
```bash
@@ -222,6 +249,8 @@ pouch/
├── paths.py # 路径常量与 Agent 目标
├── skills.py # builtin/catalog/custom 统一解析
├── catalog.py # catalog.yaml 读写与 Skill 发现
├── config.py # 用户配置与自动更新设定
├── auto_update.py # 命令执行时顺带更新 pouch 与已拉取 source
├── sources.py # custom source 配置、发现与 Git 管理
├── project.py # .pouch.yaml 管理
├── symlinks.py # 软链创建/检查/修复
@@ -238,7 +267,8 @@ pouch/
| `SKILLS_DIR` | `~/.pouch/skills/` | builtin skill 目录 |
| `CATALOG_FILE` | `~/.pouch/catalog.yaml` | 预置 Skill 来源目录 |
| `CATALOG_CACHE_DIR` | `~/.local/share/pouch/externals/` | catalog checkout 缓存;按 repo/ref 共享 |
| `CONFIG_FILE` | `~/.config/pouch/config.yaml` | custom source 配置 |
| `CONFIG_FILE` | `~/.config/pouch/config.yaml` | custom source 与 auto_update 配置 |
| `STATE_FILE` | `~/.local/share/pouch/state.yaml` | 自动更新时间戳 |
| `SOURCES_DIR` | `~/.local/share/pouch/sources/` | custom Git source 默认 checkout |
## 注意事项
+308
View File
@@ -0,0 +1,308 @@
"""在日常命令里顺带更新 pouch 自身和已拉取的外部 source。"""
from __future__ import annotations
import os
import subprocess
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from pouch import yaml_io
from pouch.catalog import catalog_checkout_path, load_catalog
from pouch.config import load_auto_update_settings
from pouch.paths import POUCH_HOME, STATE_FILE
from pouch.sources import load_sources, source_checkout
GIT_TIMEOUT_SECONDS = 20
_SKIP_COMMANDS = {"update", "publish"}
_TRUTHY = {"1", "true", "yes", "on"}
_refreshed_paths: set[str] = set()
@dataclass(frozen=True)
class UpdateResult:
name: str
status: str
detail: str = ""
@property
def changed(self) -> bool:
return self.status == "updated"
def _print_err(msg: str) -> None:
print(msg, file=sys.stderr)
def _run_git(
repo: Path,
*args: str,
check: bool = False,
timeout: int = GIT_TIMEOUT_SECONDS,
) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["git", "-C", str(repo), *args],
check=check,
text=True,
capture_output=True,
timeout=timeout,
)
def _is_git_repo(path: Path) -> bool:
if not path.is_dir():
return False
try:
result = _run_git(path, "rev-parse", "--is-inside-work-tree")
except (OSError, subprocess.TimeoutExpired):
return False
return result.returncode == 0 and result.stdout.strip() == "true"
def _working_tree_dirty(path: Path) -> bool:
result = _run_git(path, "status", "--porcelain")
return result.returncode != 0 or bool(result.stdout.strip())
def _head_sha(path: Path) -> str:
result = _run_git(path, "rev-parse", "HEAD")
return result.stdout.strip() if result.returncode == 0 else ""
def refresh_git_checkout(
path: Path,
ref: str | None = None,
*,
name: str,
strict: bool = False,
allow_dirty: bool = False,
) -> UpdateResult:
key = str(path.resolve()) if path.exists() else str(path)
if key in _refreshed_paths:
return UpdateResult(name, "current")
if not _is_git_repo(path):
_refreshed_paths.add(key)
result = UpdateResult(name, "skipped", "不是 git 仓库")
if strict:
raise SystemExit(f"{name} 不是 git 仓库: {path}")
return result
if not allow_dirty and _working_tree_dirty(path):
_refreshed_paths.add(key)
result = UpdateResult(name, "skipped", "工作区不干净")
if strict:
raise SystemExit(f"{name} 工作区不干净,拒绝更新: {path}")
return result
before = _head_sha(path)
commands: list[tuple[str, ...]] = [("fetch", "--all", "--tags")]
if ref:
commands.append(("checkout", ref))
commands.append(("pull", "--ff-only"))
try:
for command in commands:
ran = _run_git(path, *command, check=strict)
if ran.returncode != 0:
detail = (ran.stderr or ran.stdout or "git 失败").strip()
return UpdateResult(name, "failed", detail)
except subprocess.CalledProcessError as exc:
detail = ((exc.stderr or exc.stdout or str(exc))).strip()
if strict:
raise SystemExit(f"更新 {name} 失败: {detail}") from exc
return UpdateResult(name, "failed", detail)
except subprocess.TimeoutExpired:
result = UpdateResult(name, "failed", "git 超时")
if strict:
raise SystemExit(f"更新 {name} 超时: {path}") from None
return result
except OSError as exc:
result = UpdateResult(name, "failed", str(exc))
if strict:
raise SystemExit(f"更新 {name} 失败: {exc}") from exc
return result
after = _head_sha(path)
_refreshed_paths.add(key)
if before and after and before != after:
return UpdateResult(name, "updated", after[:8])
return UpdateResult(name, "current")
def update_pouch_home(*, strict: bool = False) -> UpdateResult:
return refresh_git_checkout(
POUCH_HOME,
name="pouch",
strict=strict,
allow_dirty=strict,
)
def update_cloned_catalogs(*, strict: bool = False) -> list[UpdateResult]:
results: list[UpdateResult] = []
seen: set[Path] = set()
for name, entry in load_catalog().items():
repo = str(entry.get("repo", ""))
if repo.startswith("~"):
continue
dest = catalog_checkout_path(name, entry)
resolved = dest.resolve() if dest.exists() else dest
if resolved in seen or not dest.is_dir():
continue
seen.add(resolved)
results.append(
refresh_git_checkout(
dest,
str(entry.get("ref", "main") or "main"),
name=f"catalog:{name}",
strict=strict,
)
)
return results
def update_custom_sources(*, strict: bool = False) -> list[UpdateResult]:
results: list[UpdateResult] = []
for name, entry in load_sources().items():
if entry.get("local_path"):
continue
checkout = source_checkout(name, entry)
if not checkout.is_dir():
continue
results.append(
refresh_git_checkout(
checkout,
str(entry.get("ref", "main") or "main"),
name=f"custom:{name}",
strict=strict,
)
)
return results
def run_updates(*, strict_pouch: bool = False) -> list[UpdateResult]:
results = [update_pouch_home(strict=strict_pouch)]
results.extend(update_cloned_catalogs(strict=False))
results.extend(update_custom_sources(strict=False))
return results
def load_state(path: Path | None = None) -> dict[str, Any]:
path = path or STATE_FILE
if not path.is_file():
return {}
data = yaml_io.safe_load(path.read_text(encoding="utf-8")) or {}
return data if isinstance(data, dict) else {}
def save_state(data: dict[str, Any], path: Path | None = None) -> None:
path = path or STATE_FILE
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
yaml_io.safe_dump(data, allow_unicode=True, sort_keys=False),
encoding="utf-8",
)
def record_auto_update(path: Path | None = None) -> None:
state = load_state(path)
state["last_auto_update"] = datetime.now(timezone.utc).isoformat()
save_state(state, path)
def _interval_elapsed(state: dict[str, Any], interval_seconds: int) -> bool:
if interval_seconds <= 0:
return True
raw = state.get("last_auto_update")
if not raw:
return True
try:
last = datetime.fromisoformat(str(raw))
except ValueError:
return True
if last.tzinfo is None:
last = last.replace(tzinfo=timezone.utc)
elapsed = (datetime.now(timezone.utc) - last).total_seconds()
return elapsed >= interval_seconds
def _env_disabled() -> bool:
return os.environ.get("POUCH_NO_AUTO_UPDATE", "").strip().lower() in _TRUTHY
def _acquire_lock(path: Path) -> Any | None:
path.parent.mkdir(parents=True, exist_ok=True)
try:
lock_file = path.open("a+")
except OSError:
return None
try:
import fcntl
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
lock_file.close()
return None
return lock_file
def _release_lock(lock_file: Any | None) -> None:
if lock_file is None:
return
try:
import fcntl
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
except OSError:
pass
try:
lock_file.close()
except OSError:
pass
def reset_refresh_cache() -> None:
_refreshed_paths.clear()
def report_results(results: Iterable[UpdateResult], *, prefix: str = "[pouch]") -> None:
changed = [item.name for item in results if item.changed]
failed = [item for item in results if item.status == "failed"]
if changed:
_print_err(f"{prefix} 已自动更新: {', '.join(changed)}")
for item in failed:
detail = f": {item.detail}" if item.detail else ""
_print_err(f"{prefix} 更新 {item.name} 失败,已跳过{detail}")
def maybe_auto_update(*, command: str | None = None, disabled: bool = False) -> list[UpdateResult]:
if disabled or command in _SKIP_COMMANDS or _env_disabled():
return []
if not POUCH_HOME.is_dir():
return []
settings = load_auto_update_settings()
if not settings.enabled:
return []
state = load_state()
if not _interval_elapsed(state, settings.interval_seconds):
return []
lock_file = _acquire_lock(STATE_FILE.with_name("auto-update.lock"))
if lock_file is None:
return []
try:
# 拿到锁后再读一次,避免并发重复拉取。
state = load_state()
if not _interval_elapsed(state, settings.interval_seconds):
return []
results = run_updates(strict_pouch=False)
record_auto_update()
report_results(results)
return results
except Exception as exc: # noqa: BLE001 — 自动更新不得打断用户命令
_print_err(f"[pouch] 自动更新失败,已跳过: {exc}")
return []
finally:
_release_lock(lock_file)
+85 -16
View File
@@ -58,6 +58,7 @@ from pouch.skills import (
validate_skill_dir,
validate_skill_name,
)
from pouch.auto_update import maybe_auto_update, record_auto_update, refresh_git_checkout, run_updates
from pouch.sources import (
discover_source_skills,
fetch_source,
@@ -250,15 +251,33 @@ def _collect_skill_names(positional: list[str] | None, flagged: list[str] | None
return names
def _ensure_source_fetched(name: str, source: str | None = None) -> None:
def _ensure_source_fetched(
name: str,
source: str | None = None,
*,
refresh: bool = False,
) -> 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])
entry = sources[source]
root = source_skills_root(source, entry)
if not root.is_dir():
_print(f"拉取 source: {source}")
fetch_source(source, sources[source])
fetch_source(source, entry)
elif refresh and not entry.get("local_path"):
result = refresh_git_checkout(
source_checkout(source, entry),
str(entry.get("ref", "main") or "main"),
name=f"custom:{source}",
)
if result.status == "updated":
_err(f"[pouch] 已更新 custom:{source}")
elif result.status == "failed":
_err(
f"[pouch] 更新 custom:{source} 失败,使用本地缓存: {result.detail}"
)
return
catalog_name = (
@@ -274,7 +293,11 @@ def _ensure_source_fetched(name: str, source: str | None = None) -> None:
source in (None, "catalog", catalog_name)
or source == f"catalog:{catalog_name}"
):
_ensure_catalog_fetched(catalog_name, catalog[catalog_name])
_ensure_catalog_fetched(
catalog_name,
catalog[catalog_name],
refresh=refresh,
)
return
if source not in (None, "catalog"):
@@ -283,17 +306,31 @@ def _ensure_source_fetched(name: str, source: str | None = None) -> None:
if name not in catalog:
return
_ensure_catalog_fetched(name, catalog[name])
_ensure_catalog_fetched(name, catalog[name], refresh=refresh)
def _ensure_catalog_fetched(name: str, entry: dict[str, object]) -> None:
def _ensure_catalog_fetched(
name: str,
entry: dict[str, object],
*,
refresh: bool = False,
) -> None:
path = catalog_skill_path(name, entry)
repo = catalog_repo(entry)
ref = str(entry.get("ref", "main") or "main")
dest = catalog_checkout_path(name, entry)
if dest.is_dir() and (dest / ".git").exists() and refresh and not str(entry.get("repo", "")).startswith("~"):
result = refresh_git_checkout(dest, ref, name=f"catalog:{name}")
if result.status == "updated":
_err(f"[pouch] 已更新 catalog:{name}")
elif result.status == "failed":
_err(f"[pouch] 更新 catalog:{name} 失败,使用本地缓存: {result.detail}")
if (path / "SKILL.md").is_file() or discover_catalog_skills(name, entry):
return
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(
@@ -334,14 +371,17 @@ def _expand_install_request(
catalog = load_catalog()
sources = load_sources()
if source in sources:
_ensure_source_fetched(name, source, refresh=True)
return [(name, source)]
if source == "catalog" and name in catalog:
_ensure_catalog_fetched(name, catalog[name], refresh=True)
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:
_ensure_catalog_fetched(source, catalog[source], refresh=True)
available = _catalog_skill_names(source, catalog[source])
if name not in available:
raise SystemExit(
@@ -349,13 +389,14 @@ def _expand_install_request(
)
return [(name, f"catalog:{source}")]
if source is None and name in catalog:
_ensure_catalog_fetched(name, catalog[name], refresh=True)
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)
_ensure_source_fetched(name, name, refresh=True)
available = discover_source_skills(name, sources[name])
if not available:
raise SystemExit(f"custom source {name!r} 中没有可安装的 skill")
@@ -375,7 +416,7 @@ def _install_skill(
source: str | None = None,
) -> str:
name, source = split_skill_spec(name, source)
_ensure_source_fetched(name, source)
_ensure_source_fetched(name, source, refresh=True)
skill_path, resolved_source = resolve_skill_source(name, source=source)
links = [
(target, agent_skill_dir(target, project_root=project_root) / name)
@@ -540,7 +581,23 @@ def cmd_update(args: argparse.Namespace) -> None:
del args
ensure_pouch_home()
_print(f"更新 pouch: {POUCH_HOME}")
subprocess.run(["git", "-C", str(POUCH_HOME), "pull"], check=True)
results = run_updates()
record_auto_update()
labels = {
"updated": "已更新",
"current": "已是最新",
"skipped": "跳过",
"failed": "失败",
}
for item in results:
extra = f"({item.detail})" if item.detail else ""
line = f"{labels.get(item.status, item.status)} {item.name}{extra}"
if item.status == "failed":
_err(line)
else:
_print(line)
if any(item.status == "failed" for item in results):
raise SystemExit("部分来源更新失败")
def _installed_links(
@@ -640,7 +697,7 @@ def cmd_add(args: argparse.Namespace) -> None:
elif source == "catalog":
names = list(load_catalog())
else:
_ensure_source_fetched("", source)
_ensure_source_fetched("", source, refresh=True)
names = list_custom_skills(source)[source]
else:
names = list_builtin_skills()
@@ -662,7 +719,7 @@ def cmd_add(args: argparse.Namespace) -> None:
original_links: list[tuple[Path, str | None]] = []
for skill_name, source in requests:
validate_skill_name(skill_name)
_ensure_source_fetched(skill_name, source)
_ensure_source_fetched(skill_name, source, refresh=True)
resolve_skill_source(skill_name, source=source)
for target in targets:
link = agent_skill_dir(target, project_root=project_root) / skill_name
@@ -1122,7 +1179,7 @@ def cmd_sync(args: argparse.Namespace) -> None:
for entry in iter_manifest_skills(data):
name = entry["name"]
source = entry.get("source")
_ensure_source_fetched(name, source)
_ensure_source_fetched(name, source, refresh=True)
skill_path, _ = resolve_manifest_skill(entry)
skill_targets = targets
if entry.get("targets"):
@@ -1702,6 +1759,11 @@ def build_parser() -> argparse.ArgumentParser:
description="自研 Agent Skills 安装与管理 CLI(接口对齐 Vercel skills)",
)
parser.add_argument("--version", action="version", version=f"pouch {__version__}")
parser.add_argument(
"--no-auto-update",
action="store_true",
help="本次跳过自动更新 pouch 与已拉取的 source",
)
sub = parser.add_subparsers(dest="command", required=True)
@@ -1711,7 +1773,10 @@ def build_parser() -> argparse.ArgumentParser:
)
p_bootstrap.set_defaults(func=cmd_bootstrap)
p_update = sub.add_parser("update", help="通过 git pull 更新 pouch 自身")
p_update = sub.add_parser(
"update",
help="更新 pouch 自身以及已拉取的 catalog / custom source",
)
p_update.set_defaults(func=cmd_update)
p_list = sub.add_parser("list", help="列出所有 source 中的 skill")
@@ -1845,6 +1910,10 @@ def build_parser() -> argparse.ArgumentParser:
def main(argv: list[str] | None = None) -> None:
parser = build_parser()
args = parser.parse_args(argv)
maybe_auto_update(
command=getattr(args, "command", None),
disabled=bool(getattr(args, "no_auto_update", False)),
)
args.func(args)
+87
View File
@@ -0,0 +1,87 @@
"""用户级 pouch 配置(~/.config/pouch/config.yaml)。"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from pouch import yaml_io
from pouch.paths import CONFIG_FILE
DEFAULT_AUTO_UPDATE_INTERVAL = "24h"
_INTERVAL_UNITS = {
"s": 1,
"m": 60,
"h": 3600,
"d": 86400,
}
@dataclass(frozen=True)
class AutoUpdateSettings:
enabled: bool = True
interval_seconds: int = 24 * 3600
def load_user_config(path: Path | None = None) -> dict[str, Any]:
path = path or CONFIG_FILE
if not path.is_file():
return {}
data = yaml_io.safe_load(path.read_text(encoding="utf-8")) or {}
if not isinstance(data, dict):
raise SystemExit(f"pouch 配置格式错误: {path}")
return data
def save_user_config(data: dict[str, Any], path: Path | None = None) -> None:
path = path or CONFIG_FILE
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
yaml_io.safe_dump(data, allow_unicode=True, sort_keys=False),
encoding="utf-8",
)
def parse_interval(value: Any) -> int:
if value is None or value is True:
return AutoUpdateSettings().interval_seconds
if isinstance(value, bool):
raise ValueError(f"无效的自动更新间隔: {value!r}")
if isinstance(value, int):
if value < 0:
raise ValueError(f"无效的自动更新间隔: {value!r}")
return value
text = str(value).strip().lower()
if not text:
return AutoUpdateSettings().interval_seconds
if text.isdigit():
return int(text)
unit = text[-1]
amount = text[:-1]
if unit in _INTERVAL_UNITS and amount.isdigit():
return int(amount) * _INTERVAL_UNITS[unit]
raise ValueError(f"无效的自动更新间隔: {value!r}(例如 24h、1d、60m、3600)")
def load_auto_update_settings(
config: dict[str, Any] | None = None,
path: Path | None = None,
) -> AutoUpdateSettings:
if config is None:
config = load_user_config(path)
raw = config.get("auto_update", True)
if raw is False:
return AutoUpdateSettings(enabled=False)
if raw is True or raw is None:
return AutoUpdateSettings()
if not isinstance(raw, dict):
return AutoUpdateSettings()
enabled = raw.get("enabled", True)
if enabled is False:
return AutoUpdateSettings(enabled=False)
try:
interval = parse_interval(raw.get("interval", DEFAULT_AUTO_UPDATE_INTERVAL))
except ValueError as exc:
raise SystemExit(str(exc)) from exc
return AutoUpdateSettings(enabled=bool(enabled), interval_seconds=interval)
+7
View File
@@ -45,11 +45,18 @@ CONFIG_FILE = _first_existing_file(
HOME / ".config" / "skiff" / "config.yaml",
default=HOME / ".config" / "pouch" / "config.yaml",
)
DATA_DIR = _first_existing_dir(
HOME / ".local" / "share" / "pouch",
HOME / ".local" / "share" / "skiff",
default=HOME / ".local" / "share" / "pouch",
)
SOURCES_DIR = _first_existing_dir(
DATA_DIR / "sources",
HOME / ".local" / "share" / "pouch" / "sources",
HOME / ".local" / "share" / "skiff" / "sources",
default=HOME / ".local" / "share" / "pouch" / "sources",
)
STATE_FILE = DATA_DIR / "state.yaml"
AGENT_GLOBAL: dict[str, Path] = {
"cursor": HOME / ".cursor" / "skills",
+6 -13
View File
@@ -6,7 +6,7 @@ import subprocess
from pathlib import Path
from typing import Any
from pouch import yaml_io
from pouch.config import load_user_config, save_user_config
from pouch.paths import CONFIG_FILE, SOURCES_DIR
RESERVED_SOURCES = {"builtin", "catalog", "owned", "registry"}
@@ -21,27 +21,20 @@ def validate_source_name(name: str) -> None:
def load_sources(path: Path | None = None) -> dict[str, dict[str, Any]]:
path = path or CONFIG_FILE
if not path.is_file():
return {}
data = yaml_io.safe_load(path.read_text(encoding="utf-8")) or {}
if not isinstance(data, dict):
raise SystemExit(f"pouch 配置格式错误: {path}")
data = load_user_config(path)
raw = data.get("sources", {})
if raw is None:
return {}
if not isinstance(raw, dict):
raise SystemExit(f"pouch 配置 sources 格式错误: {path}")
raise SystemExit(f"pouch 配置 sources 格式错误: {path or CONFIG_FILE}")
return {str(name): entry for name, entry in raw.items() if isinstance(entry, dict)}
def save_sources(sources: dict[str, dict[str, Any]], path: Path | None = None) -> None:
path = path or CONFIG_FILE
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
yaml_io.safe_dump({"sources": sources}, allow_unicode=True, sort_keys=False),
encoding="utf-8",
)
data = load_user_config(path)
data["sources"] = sources
save_user_config(data, path)
def source_checkout(name: str, entry: dict[str, Any]) -> Path:
+4 -2
View File
@@ -157,6 +157,8 @@ collection。安装 collection 全部内容或其中一个:
```bash
pouch add waza -a codex -g -y
pouch add waza/think -a codex -g -y
pouch add taste-skill/taste-skill -a codex -g -y
pouch add taste-skill -a codex -g -y
```
`pouch select` 会把 collection 显示为两级菜单:选择 `waza` 仓库会选中其
@@ -207,7 +209,7 @@ ACK 配置、检查接入状态或
| pouch | 说明 |
|-------|------|
| `bootstrap` | 将本项目的 `pouch` skill 全局安装到所有 Agent |
| `update` | 在 `~/.pouch` 执行 `git pull`,更新 pouch 自身 |
| `update` | 更新 pouch 自身以及已拉取的 catalog / custom source |
| `add <name> [-g] [-a AGENT...] [-y]` | 安装 |
| `remove <name> [-g] [-a AGENT...] [-y]` | 卸载(`rm` / `r` 别名) |
| `add --list` | 列出可用自研 skill |
@@ -226,4 +228,4 @@ ACK 配置、检查接入状态或
- 不要在 `project/.agents/skills/` 里直接改文件;应改 `~/.pouch/skills/` 再 `publish`
- 未完成的内容保留在 `~/.pouch/.drafts/`,不要直接放进正式 `skills/`
- symlink 正确时,**不需要 reinstall**;保存 SSOT 后各项目自动读到新内容
- 社区 skill 用 `npx skills add`,不要用 pouch `catalog add` 除非团队要 pin 版本
- 未写入 `catalog.yaml` 的社区 skill 用 `npx skills add`。waza、taste-skill 等预置来源用 `pouch add`
+248
View File
@@ -0,0 +1,248 @@
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from pouch.config import load_auto_update_settings, parse_interval
from pouch.sources import save_sources
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
def git_init(path: Path) -> None:
subprocess.run(
["git", "init", "-b", "main", str(path)],
check=True,
capture_output=True,
)
subprocess.run(
["git", "-C", str(path), "config", "user.name", "Test"],
check=True,
capture_output=True,
)
subprocess.run(
["git", "-C", str(path), "config", "user.email", "test@example.com"],
check=True,
capture_output=True,
)
subprocess.run(
["git", "-C", str(path), "config", "commit.gpgsign", "false"],
check=True,
capture_output=True,
)
def git_commit(path: Path, message: str) -> None:
subprocess.run(["git", "-C", str(path), "add", "."], check=True, capture_output=True)
subprocess.run(
["git", "-C", str(path), "commit", "-m", message],
check=True,
capture_output=True,
)
class IntervalTests(unittest.TestCase):
def test_parse_interval_units(self) -> None:
self.assertEqual(parse_interval("24h"), 24 * 3600)
self.assertEqual(parse_interval("1d"), 86400)
self.assertEqual(parse_interval("30m"), 1800)
self.assertEqual(parse_interval(0), 0)
self.assertEqual(parse_interval("90"), 90)
def test_auto_update_false_disables(self) -> None:
self.assertFalse(load_auto_update_settings({"auto_update": False}).enabled)
self.assertFalse(
load_auto_update_settings({"auto_update": {"enabled": False}}).enabled
)
def test_auto_update_interval_from_mapping(self) -> None:
settings = load_auto_update_settings({"auto_update": {"interval": "1h"}})
self.assertTrue(settings.enabled)
self.assertEqual(settings.interval_seconds, 3600)
class SaveSourcesPreservesConfigTests(unittest.TestCase):
def test_save_sources_keeps_auto_update_settings(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "config.yaml"
path.write_text(
"auto_update:\n enabled: false\n interval: 1h\nsources: {}\n",
encoding="utf-8",
)
save_sources(
{
"company": {
"local_path": str(Path(tmp) / "skills"),
"skills_path": ".",
}
},
path,
)
text = path.read_text(encoding="utf-8")
self.assertIn("enabled: false", text)
self.assertIn("interval: 1h", text)
self.assertIn("company:", text)
class AutoUpdateCliTests(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.home = Path(self.temp_dir.name)
self.catalog_upstream = self.home / "catalog-upstream"
write_skill(self.catalog_upstream / "skills", "alpha")
git_init(self.catalog_upstream)
git_commit(self.catalog_upstream, "initial catalog")
self.upstream = self.home / "pouch-upstream"
self.upstream.mkdir()
(self.upstream / "skills").mkdir()
(self.upstream / "catalog.yaml").write_text(
"demo:\n"
f" repo: {self.catalog_upstream}\n"
" ref: main\n"
" path: skills\n",
encoding="utf-8",
)
git_init(self.upstream)
git_commit(self.upstream, "initial pouch")
subprocess.run(
["git", "clone", str(self.upstream), str(self.home / ".pouch")],
check=True,
capture_output=True,
)
subprocess.run(
["git", "-C", str(self.home / ".pouch"), "config", "pull.ff", "only"],
check=True,
capture_output=True,
)
def tearDown(self) -> None:
self.temp_dir.cleanup()
def run_pouch(self, *args: str) -> subprocess.CompletedProcess[str]:
env = os.environ.copy()
env["HOME"] = str(self.home)
env["PYTHONPATH"] = str(REPO_ROOT)
env.pop("POUCH_NO_AUTO_UPDATE", None)
return subprocess.run(
[sys.executable, "-m", "pouch", *args],
cwd=REPO_ROOT,
env=env,
text=True,
capture_output=True,
check=False,
)
def write_config(self, text: str) -> None:
path = self.home / ".config" / "pouch" / "config.yaml"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
def test_auto_update_pulls_pouch_on_first_command(self) -> None:
marker = self.upstream / "AUTO_UPDATE_MARKER"
marker.write_text("yes\n", encoding="utf-8")
git_commit(self.upstream, "marker")
result = self.run_pouch("list")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertTrue((self.home / ".pouch" / "AUTO_UPDATE_MARKER").is_file())
self.assertIn("已自动更新: pouch", result.stderr)
def test_no_auto_update_flag_skips_pull(self) -> None:
marker = self.upstream / "AUTO_UPDATE_MARKER"
marker.write_text("yes\n", encoding="utf-8")
git_commit(self.upstream, "marker")
result = self.run_pouch("--no-auto-update", "list")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse((self.home / ".pouch" / "AUTO_UPDATE_MARKER").is_file())
self.assertNotIn("已自动更新", result.stderr)
def test_config_can_disable_auto_update(self) -> None:
self.write_config("auto_update:\n enabled: false\n")
marker = self.upstream / "AUTO_UPDATE_MARKER"
marker.write_text("yes\n", encoding="utf-8")
git_commit(self.upstream, "marker")
result = self.run_pouch("list")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse((self.home / ".pouch" / "AUTO_UPDATE_MARKER").is_file())
def test_dirty_pouch_is_skipped(self) -> None:
marker = self.upstream / "AUTO_UPDATE_MARKER"
marker.write_text("yes\n", encoding="utf-8")
git_commit(self.upstream, "marker")
(self.home / ".pouch" / "dirty.txt").write_text("local\n", encoding="utf-8")
result = self.run_pouch("list")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse((self.home / ".pouch" / "AUTO_UPDATE_MARKER").is_file())
def test_add_refreshes_cloned_catalog_source(self) -> None:
first = self.run_pouch("add", "demo/alpha", "-g", "-a", "codex")
self.assertEqual(first.returncode, 0, first.stderr)
write_skill(self.catalog_upstream / "skills", "beta")
git_commit(self.catalog_upstream, "add beta")
second = self.run_pouch("add", "demo", "-g", "-a", "codex")
self.assertEqual(second.returncode, 0, second.stderr)
self.assertTrue((self.home / ".codex" / "skills" / "beta").is_symlink())
def test_update_refreshes_cloned_catalog(self) -> None:
added = self.run_pouch("--no-auto-update", "add", "demo/alpha", "-g", "-a", "codex")
self.assertEqual(added.returncode, 0, added.stderr)
write_skill(self.catalog_upstream / "skills", "beta")
git_commit(self.catalog_upstream, "add beta")
updated = self.run_pouch("--no-auto-update", "update")
self.assertEqual(updated.returncode, 0, updated.stderr)
checkout = None
externals = self.home / ".local" / "share" / "pouch" / "externals"
for candidate in externals.rglob("SKILL.md"):
if candidate.parent.name == "alpha":
checkout = candidate.parents[1]
break
self.assertIsNotNone(checkout)
self.assertTrue((checkout / "beta" / "SKILL.md").is_file())
self.assertIn("catalog:demo", updated.stdout)
class ShippedCatalogTests(unittest.TestCase):
def test_catalog_includes_taste_skill_like_waza(self) -> None:
from pouch import yaml_io
data = yaml_io.safe_load(
(REPO_ROOT / "catalog.yaml").read_text(encoding="utf-8")
)
self.assertIn("waza", data)
self.assertIn("taste-skill", data)
entry = data["taste-skill"]
self.assertIn("Leonxlnx/taste-skill", entry["repo"])
self.assertEqual(entry["path"], "skills")
self.assertEqual(entry["ref"], "main")
if __name__ == "__main__":
unittest.main()
+30
View File
@@ -204,6 +204,36 @@ class CatalogCollectionTests(unittest.TestCase):
self.assertFalse((skill_dir / "first-skill").exists())
self.assertFalse((skill_dir / "second-skill").exists())
def test_add_refreshes_collection_before_installing_new_children(self) -> None:
first = self.run_pouch("add", "test-pack/first-skill", "-g", "-a", "codex")
self.assertEqual(first.returncode, 0, first.stderr)
write_skill(self.upstream / "skills", "third-skill")
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",
"add third",
],
check=True,
capture_output=True,
)
second = self.run_pouch("add", "test-pack", "-g", "-a", "codex")
skill_dir = self.home / ".codex" / "skills"
self.assertEqual(second.returncode, 0, second.stderr)
self.assertTrue((skill_dir / "third-skill").is_symlink())
self.assertTrue((skill_dir / "third-skill" / "SKILL.md").is_file())
def test_status_expands_collection_children(self) -> None:
installed = self.run_pouch("add", "test-pack", "-g", "-a", "codex")
status = self.run_pouch("status", "-a", "codex")
+12 -5
View File
@@ -10,6 +10,7 @@ from pathlib import Path
from unittest.mock import patch
from pouch import cli
from pouch.auto_update import UpdateResult
REPO_ROOT = Path(__file__).resolve().parents[1]
@@ -95,14 +96,20 @@ class CreateWorkflowTests(unittest.TestCase):
with (
patch.object(cli, "POUCH_HOME", self.skills_home),
patch.object(cli, "ensure_pouch_home"),
patch.object(cli.subprocess, "run") as run,
patch.object(
cli,
"run_updates",
return_value=[
UpdateResult("pouch", "updated", "abc1234"),
UpdateResult("catalog:waza", "current"),
],
) as run_updates,
patch.object(cli, "record_auto_update") as record,
):
cli.cmd_update(argparse.Namespace())
run.assert_called_once_with(
["git", "-C", str(self.skills_home), "pull"],
check=True,
)
run_updates.assert_called_once_with()
record.assert_called_once()
def test_create_writes_draft_and_brief_without_registering_owned_skill(self) -> None:
project = self.home / "project"