From 8390706df859fc2930fb6d035b9bb6ade7f9fc5d Mon Sep 17 00:00:00 2001 From: laily Date: Tue, 13 Oct 2026 16:41:55 +0800 Subject: [PATCH] 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. --- AGENTS.md | 5 +- README.md | 2 + catalog.yaml | 10 + docs/getting-started.md | 2 +- pouch/README.md | 34 +++- pouch/auto_update.py | 308 ++++++++++++++++++++++++++++++ pouch/cli.py | 101 ++++++++-- pouch/config.py | 87 +++++++++ pouch/paths.py | 7 + pouch/sources.py | 19 +- skills/pouch/SKILL.md | 6 +- tests/test_auto_update.py | 248 ++++++++++++++++++++++++ tests/test_catalog_collections.py | 30 +++ tests/test_create_workflow.py | 17 +- 14 files changed, 836 insertions(+), 40 deletions(-) create mode 100644 pouch/auto_update.py create mode 100644 pouch/config.py create mode 100644 tests/test_auto_update.py diff --git a/AGENTS.md b/AGENTS.md index c9f00d0..fc054c0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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 [-g]` | 项目或全局安装（symlink） | | `pouch remove [-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 ` 或 `pouch enable ` | 项目目录 | | 看状态 | `pouch status` | 任意 | | 装社区 skill | `npx skills add owner/repo -g -y` | 任意 | -| 更新外部 skill | `pouch fetch ` | 任意 | +| 更新外部 skill | `pouch fetch ` 或日常命令自动更新 | 任意 | --- diff --git a/README.md b/README.md index 12325a0..5c9569d 100644 --- a/README.md +++ b/README.md @@ -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 仓库： diff --git a/catalog.yaml b/catalog.yaml index e4f9abf..67ea670 100644 --- a/catalog.yaml +++ b/catalog.yaml @@ -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 diff --git a/docs/getting-started.md b/docs/getting-started.md index 60bc734..337b956 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -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 装上 diff --git a/pouch/README.md b/pouch/README.md index fa248c4..4574f1f 100644 --- a/pouch/README.md +++ b/pouch/README.md @@ -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 [--project DIR]` | 使用 builtin skill 自带模板初始化项目状态（目前用于 ack） | ### Skill 安装 @@ -105,6 +105,33 @@ collection 会自动发现下一层所有含 `SKILL.md` 的目录；`pouch add < 安装全部，`pouch select` 则展开为 `/` 供分别勾选。同一 `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 ` 在安装前会再刷新 +该来源，以便拿到最新子 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 | ## 注意事项 diff --git a/pouch/auto_update.py b/pouch/auto_update.py new file mode 100644 index 0000000..e2a273c --- /dev/null +++ b/pouch/auto_update.py @@ -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) diff --git a/pouch/cli.py b/pouch/cli.py index 95e7e87..307ff68 100644 --- a/pouch/cli.py +++ b/pouch/cli.py @@ -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) diff --git a/pouch/config.py b/pouch/config.py new file mode 100644 index 0000000..c6868b1 --- /dev/null +++ b/pouch/config.py @@ -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) diff --git a/pouch/paths.py b/pouch/paths.py index ea53e70..3c195c6 100644 --- a/pouch/paths.py +++ b/pouch/paths.py @@ -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", diff --git a/pouch/sources.py b/pouch/sources.py index 3a6f4e5..f7ffa27 100644 --- a/pouch/sources.py +++ b/pouch/sources.py @@ -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: diff --git a/skills/pouch/SKILL.md b/skills/pouch/SKILL.md index 60f8c95..3a1a216 100644 --- a/skills/pouch/SKILL.md +++ b/skills/pouch/SKILL.md @@ -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 [-g] [-a AGENT...] [-y]` | 安装 | | `remove [-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` diff --git a/tests/test_auto_update.py b/tests/test_auto_update.py new file mode 100644 index 0000000..bfe05e1 --- /dev/null +++ b/tests/test_auto_update.py @@ -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() diff --git a/tests/test_catalog_collections.py b/tests/test_catalog_collections.py index 4c12743..3878f94 100644 --- a/tests/test_catalog_collections.py +++ b/tests/test_catalog_collections.py @@ -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") diff --git a/tests/test_create_workflow.py b/tests/test_create_workflow.py index 769bd66..4626d3a 100644 --- a/tests/test_create_workflow.py +++ b/tests/test_create_workflow.py @@ -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"