211b55815e
Two bugs in sync_skills(): 1. Failed copytree poisons manifest: when shutil.copytree fails (disk full, permission error), the skill is still recorded in the manifest. On the next sync, the skill appears as "in manifest but not on disk" which is interpreted as "user deliberately deleted it" — the skill is never retried. Fix: only write to manifest on successful copy. 2. Failed update destroys user copy: rmtree deletes the existing skill directory before copytree runs. If copytree then fails, the user's skill is gone with no way to recover. Fix: move to .bak before copying, restore from backup if copytree fails. Both bugs are proven by new regression tests that fail on the old code and pass on the fix.
262 lines
9.6 KiB
Python
262 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Skills Sync -- Manifest-based seeding and updating of bundled skills.
|
|
|
|
Copies bundled skills from the repo's skills/ directory into ~/.hermes/skills/
|
|
and uses a manifest to track which skills have been synced and their origin hash.
|
|
|
|
Manifest format (v2): each line is "skill_name:origin_hash" where origin_hash
|
|
is the MD5 of the bundled skill at the time it was last synced to the user dir.
|
|
Old v1 manifests (plain names without hashes) are auto-migrated.
|
|
|
|
Update logic:
|
|
- NEW skills (not in manifest): copied to user dir, origin hash recorded.
|
|
- EXISTING skills (in manifest, present in user dir):
|
|
* If user copy matches origin hash: user hasn't modified it → safe to
|
|
update from bundled if bundled changed. New origin hash recorded.
|
|
* If user copy differs from origin hash: user customized it → SKIP.
|
|
- DELETED by user (in manifest, absent from user dir): respected, not re-added.
|
|
- REMOVED from bundled (in manifest, gone from repo): cleaned from manifest.
|
|
|
|
The manifest lives at ~/.hermes/skills/.bundled_manifest.
|
|
"""
|
|
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import Dict, List, Tuple
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
HERMES_HOME = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
|
|
SKILLS_DIR = HERMES_HOME / "skills"
|
|
MANIFEST_FILE = SKILLS_DIR / ".bundled_manifest"
|
|
|
|
|
|
def _get_bundled_dir() -> Path:
|
|
"""Locate the bundled skills/ directory in the repo."""
|
|
return Path(__file__).parent.parent / "skills"
|
|
|
|
|
|
def _read_manifest() -> Dict[str, str]:
|
|
"""
|
|
Read the manifest as a dict of {skill_name: origin_hash}.
|
|
|
|
Handles both v1 (plain names) and v2 (name:hash) formats.
|
|
v1 entries get an empty hash string which triggers migration on next sync.
|
|
"""
|
|
if not MANIFEST_FILE.exists():
|
|
return {}
|
|
try:
|
|
result = {}
|
|
for line in MANIFEST_FILE.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if ":" in line:
|
|
# v2 format: name:hash
|
|
name, _, hash_val = line.partition(":")
|
|
result[name.strip()] = hash_val.strip()
|
|
else:
|
|
# v1 format: plain name — empty hash triggers migration
|
|
result[line] = ""
|
|
return result
|
|
except (OSError, IOError):
|
|
return {}
|
|
|
|
|
|
def _write_manifest(entries: Dict[str, str]):
|
|
"""Write the manifest file in v2 format (name:hash)."""
|
|
MANIFEST_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
lines = [f"{name}:{hash_val}" for name, hash_val in sorted(entries.items())]
|
|
MANIFEST_FILE.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def _discover_bundled_skills(bundled_dir: Path) -> List[Tuple[str, Path]]:
|
|
"""
|
|
Find all SKILL.md files in the bundled directory.
|
|
Returns list of (skill_name, skill_directory_path) tuples.
|
|
"""
|
|
skills = []
|
|
if not bundled_dir.exists():
|
|
return skills
|
|
|
|
for skill_md in bundled_dir.rglob("SKILL.md"):
|
|
path_str = str(skill_md)
|
|
if "/.git/" in path_str or "/.github/" in path_str or "/.hub/" in path_str:
|
|
continue
|
|
skill_dir = skill_md.parent
|
|
skill_name = skill_dir.name
|
|
skills.append((skill_name, skill_dir))
|
|
|
|
return skills
|
|
|
|
|
|
def _compute_relative_dest(skill_dir: Path, bundled_dir: Path) -> Path:
|
|
"""
|
|
Compute the destination path in SKILLS_DIR preserving the category structure.
|
|
e.g., bundled/skills/mlops/axolotl -> ~/.hermes/skills/mlops/axolotl
|
|
"""
|
|
rel = skill_dir.relative_to(bundled_dir)
|
|
return SKILLS_DIR / rel
|
|
|
|
|
|
def _dir_hash(directory: Path) -> str:
|
|
"""Compute a hash of all file contents in a directory for change detection."""
|
|
hasher = hashlib.md5()
|
|
try:
|
|
for fpath in sorted(directory.rglob("*")):
|
|
if fpath.is_file():
|
|
rel = fpath.relative_to(directory)
|
|
hasher.update(str(rel).encode("utf-8"))
|
|
hasher.update(fpath.read_bytes())
|
|
except (OSError, IOError):
|
|
pass
|
|
return hasher.hexdigest()
|
|
|
|
|
|
def sync_skills(quiet: bool = False) -> dict:
|
|
"""
|
|
Sync bundled skills into ~/.hermes/skills/ using the manifest.
|
|
|
|
Returns:
|
|
dict with keys: copied (list), updated (list), skipped (int),
|
|
user_modified (list), cleaned (list), total_bundled (int)
|
|
"""
|
|
bundled_dir = _get_bundled_dir()
|
|
if not bundled_dir.exists():
|
|
return {
|
|
"copied": [], "updated": [], "skipped": 0,
|
|
"user_modified": [], "cleaned": [], "total_bundled": 0,
|
|
}
|
|
|
|
SKILLS_DIR.mkdir(parents=True, exist_ok=True)
|
|
manifest = _read_manifest()
|
|
bundled_skills = _discover_bundled_skills(bundled_dir)
|
|
bundled_names = {name for name, _ in bundled_skills}
|
|
|
|
copied = []
|
|
updated = []
|
|
user_modified = []
|
|
skipped = 0
|
|
|
|
for skill_name, skill_src in bundled_skills:
|
|
dest = _compute_relative_dest(skill_src, bundled_dir)
|
|
bundled_hash = _dir_hash(skill_src)
|
|
|
|
if skill_name not in manifest:
|
|
# ── New skill — never offered before ──
|
|
try:
|
|
if dest.exists():
|
|
# User already has a skill with the same name — don't overwrite
|
|
skipped += 1
|
|
manifest[skill_name] = bundled_hash
|
|
else:
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copytree(skill_src, dest)
|
|
copied.append(skill_name)
|
|
manifest[skill_name] = bundled_hash
|
|
if not quiet:
|
|
print(f" + {skill_name}")
|
|
except (OSError, IOError) as e:
|
|
if not quiet:
|
|
print(f" ! Failed to copy {skill_name}: {e}")
|
|
# Do NOT add to manifest — next sync should retry
|
|
|
|
elif dest.exists():
|
|
# ── Existing skill — in manifest AND on disk ──
|
|
origin_hash = manifest.get(skill_name, "")
|
|
user_hash = _dir_hash(dest)
|
|
|
|
if not origin_hash:
|
|
# v1 migration: no origin hash recorded. Set baseline from
|
|
# user's current copy so future syncs can detect modifications.
|
|
manifest[skill_name] = user_hash
|
|
if user_hash == bundled_hash:
|
|
skipped += 1 # already in sync
|
|
else:
|
|
# Can't tell if user modified or bundled changed — be safe
|
|
skipped += 1
|
|
continue
|
|
|
|
if user_hash != origin_hash:
|
|
# User modified this skill — don't overwrite their changes
|
|
user_modified.append(skill_name)
|
|
if not quiet:
|
|
print(f" ~ {skill_name} (user-modified, skipping)")
|
|
continue
|
|
|
|
# User copy matches origin — check if bundled has a newer version
|
|
if bundled_hash != origin_hash:
|
|
try:
|
|
# Move old copy to a backup so we can restore on failure
|
|
backup = dest.with_suffix(".bak")
|
|
shutil.move(str(dest), str(backup))
|
|
try:
|
|
shutil.copytree(skill_src, dest)
|
|
manifest[skill_name] = bundled_hash
|
|
updated.append(skill_name)
|
|
if not quiet:
|
|
print(f" ↑ {skill_name} (updated)")
|
|
# Remove backup after successful copy
|
|
shutil.rmtree(backup, ignore_errors=True)
|
|
except (OSError, IOError):
|
|
# Restore from backup
|
|
if backup.exists() and not dest.exists():
|
|
shutil.move(str(backup), str(dest))
|
|
raise
|
|
except (OSError, IOError) as e:
|
|
if not quiet:
|
|
print(f" ! Failed to update {skill_name}: {e}")
|
|
else:
|
|
skipped += 1 # bundled unchanged, user unchanged
|
|
|
|
else:
|
|
# ── In manifest but not on disk — user deleted it ──
|
|
skipped += 1
|
|
|
|
# Clean stale manifest entries (skills removed from bundled dir)
|
|
cleaned = sorted(set(manifest.keys()) - bundled_names)
|
|
for name in cleaned:
|
|
del manifest[name]
|
|
|
|
# Also copy DESCRIPTION.md files for categories (if not already present)
|
|
for desc_md in bundled_dir.rglob("DESCRIPTION.md"):
|
|
rel = desc_md.relative_to(bundled_dir)
|
|
dest_desc = SKILLS_DIR / rel
|
|
if not dest_desc.exists():
|
|
try:
|
|
dest_desc.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(desc_md, dest_desc)
|
|
except (OSError, IOError) as e:
|
|
logger.debug("Could not copy %s: %s", desc_md, e)
|
|
|
|
_write_manifest(manifest)
|
|
|
|
return {
|
|
"copied": copied,
|
|
"updated": updated,
|
|
"skipped": skipped,
|
|
"user_modified": user_modified,
|
|
"cleaned": cleaned,
|
|
"total_bundled": len(bundled_skills),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Syncing bundled skills into ~/.hermes/skills/ ...")
|
|
result = sync_skills(quiet=False)
|
|
parts = [
|
|
f"{len(result['copied'])} new",
|
|
f"{len(result['updated'])} updated",
|
|
f"{result['skipped']} unchanged",
|
|
]
|
|
if result["user_modified"]:
|
|
parts.append(f"{len(result['user_modified'])} user-modified (kept)")
|
|
if result["cleaned"]:
|
|
parts.append(f"{len(result['cleaned'])} cleaned from manifest")
|
|
print(f"\nDone: {', '.join(parts)}. {result['total_bundled']} total bundled.")
|